354 lines
10 KiB
Python
354 lines
10 KiB
Python
# Helper functions and polyfills
|
|
|
|
import asyncio, re, sys, io
|
|
from typing import TextIO
|
|
|
|
from prompt_toolkit import Application
|
|
from prompt_toolkit.application.current import get_app
|
|
from prompt_toolkit.output import ColorDepth, create_output
|
|
from prompt_toolkit.layout import is_container, FormattedTextControl, Float
|
|
from prompt_toolkit.layout.layout import Layout
|
|
from prompt_toolkit.layout.containers import HSplit, VSplit, ConditionalContainer, Window
|
|
from prompt_toolkit.widgets import Button, Dialog, Label
|
|
from prompt_toolkit.key_binding import KeyBindings, merge_key_bindings
|
|
from prompt_toolkit.key_binding.defaults import load_key_bindings
|
|
from prompt_toolkit.filters import Condition, has_focus, is_true
|
|
from prompt_toolkit.key_binding.bindings.focus import focus_next, focus_previous
|
|
from prompt_toolkit.data_structures import Point
|
|
|
|
# ------------------------------------------------------------
|
|
# Interface
|
|
# ------------------------------------------------------------
|
|
|
|
class EncodingProxy(io.TextIOBase):
|
|
def __init__(self, raw, encoding: str):
|
|
self._raw = raw
|
|
self._encoding = encoding
|
|
# self.buffer = raw.buffer
|
|
self.pipeReplace = False
|
|
for ch in ["┌","┐","┘","└","─","│"]:
|
|
if ch.encode(encoding,errors="replace").decode(encoding)!=ch:
|
|
self.pipeReplace = True
|
|
break
|
|
|
|
@property
|
|
def encoding(self) -> str:
|
|
return self._encoding
|
|
|
|
def write(self, s: str) -> int:
|
|
if self.encoding=="ascii" or self.pipeReplace:
|
|
for corner in ["┌","┐","┘","└"]:
|
|
s=s.replace(corner,"+")
|
|
s=s.replace("─","-")
|
|
s=s.replace("│","|")
|
|
pass
|
|
|
|
byts = s.encode(self._encoding,errors="replace")
|
|
# print(s.encode("utf8"),byts)
|
|
self._raw.buffer.write(bytes(byts))
|
|
self._raw.flush()
|
|
return len(s)
|
|
|
|
def flush(self) -> None:
|
|
self._raw.flush()
|
|
|
|
def fileno(self):
|
|
return self._raw.fileno()
|
|
|
|
def isatty(self):
|
|
return True
|
|
|
|
def createEncodedOutput(enc):
|
|
proxy: TextIO = EncodingProxy(sys.stdout, encoding=enc)
|
|
return create_output(proxy)
|
|
|
|
def getColorDepth(intset):
|
|
try:
|
|
return ColorDepth(f"DEPTH_{intset["color"]}_BIT")
|
|
except ValueError:
|
|
return ColorDepth.DEPTH_4_BIT
|
|
|
|
def defaultAppSettings(intset):
|
|
return {"full_screen":True, "color_depth":getColorDepth(intset), "output": createEncodedOutput(intset["encoding"])}
|
|
|
|
def containIfNeeded(el):
|
|
if not is_container(el):
|
|
return Window(el)
|
|
return el
|
|
|
|
def message_dialog( # derived from prompt_toolkit's code
|
|
intset,
|
|
title = "",
|
|
text = "",
|
|
ok_text: str = "Ok",
|
|
style = None,
|
|
) -> Application[None]:
|
|
"""
|
|
Display a simple message box and wait until the user presses enter.
|
|
"""
|
|
def _return_none() -> None:
|
|
"Button handler that returns None."
|
|
get_app().exit()
|
|
|
|
dialog = Dialog(
|
|
title=title,
|
|
body=Label(text=text, dont_extend_height=True),
|
|
buttons=[Button(text=ok_text, handler=_return_none)],
|
|
with_background=True,
|
|
)
|
|
|
|
# Key bindings.
|
|
bindings = KeyBindings()
|
|
bindings.add("tab")(focus_next)
|
|
bindings.add("s-tab")(focus_previous)
|
|
|
|
return Application(
|
|
layout=Layout(dialog),
|
|
key_bindings=merge_key_bindings([load_key_bindings(), bindings]),
|
|
style=style,
|
|
**defaultAppSettings(intset)
|
|
)
|
|
|
|
class TabLayout():
|
|
def __init__(self,kb,tabs,tabFocus,tabNames,name):
|
|
self.tabs = tabs
|
|
self.tabFocus = tabFocus
|
|
self.tabNames = tabNames
|
|
|
|
self.current = 0
|
|
self.control = False
|
|
self.tabEls = [ConditionalContainer(content=tabs[i],filter=Condition(lambda i=i: self.current==i)) for i in range(len(tabs))]
|
|
|
|
# tab bar
|
|
self.tbarEls = [ConditionalContainer(Label("StingNET [WIP]"),Condition(lambda: get_app().output.get_size().columns >= 90)),ConditionalContainer(content=Label("[Ctrl-Up] ",dont_extend_width=True,style="fg:#bbbbbb"),filter=Condition(lambda: not self.control)),ConditionalContainer(content=Label("[Left/Right/Enter] ",dont_extend_width=True),filter=Condition(lambda: self.control))]
|
|
self.tbarBtns = []
|
|
for i in range(len(tabNames)):
|
|
def handle(i=i):
|
|
self.current=i
|
|
self.control=False
|
|
self.focusTabWindow()
|
|
btn = Button(text=tabNames[i],handler=handle,width=16)
|
|
self.tbarEls.append(btn)
|
|
self.tbarBtns.append(btn)
|
|
self.tbarEls=self.tbarEls+[Label(f" {name}",dont_extend_width=True,style="fg:#bbbbbb")]
|
|
|
|
self.tbar = VSplit(self.tbarEls,style="bg:#404040")
|
|
self.el = HSplit([self.tbar,VSplit(self.tabEls)])
|
|
|
|
# controls
|
|
@kb.add("c-up")
|
|
@kb.add("c-j")
|
|
def _(e):
|
|
self.control = True
|
|
self.focusCurrent(e)
|
|
|
|
@kb.add("left",filter=Condition(lambda: self.control))
|
|
def _(e):
|
|
self.current = (self.current+len(tabNames)-1)%len(tabNames)
|
|
self.focusCurrent(e)
|
|
|
|
@kb.add("right",filter=Condition(lambda: self.control))
|
|
def _(e):
|
|
self.current = (self.current+1)%len(tabNames)
|
|
self.focusCurrent(e)
|
|
|
|
def focusTabWindow(self):
|
|
try:
|
|
el = self.tabs[self.current]
|
|
if self.tabFocus[self.current]:
|
|
el = self.tabFocus[self.current]
|
|
get_app().layout.focus(el)
|
|
except Exception:
|
|
pass
|
|
|
|
def focusCurrent(self,e=None):
|
|
app = e.app if e else get_app()
|
|
app.layout.focus(self.tbarBtns[self.current])
|
|
|
|
class HSelectLayout():
|
|
def __init__(self,kb,opts,optFocus,optNames):
|
|
self.opts=opts
|
|
self.optFocus=optFocus
|
|
self.optNames=optNames
|
|
|
|
self.current = 0
|
|
self.select = Window(FormattedTextControl(text=self.show_sel,get_cursor_position=self.curpos,key_bindings=self.make_select_kb()),width=16,dont_extend_height=False,style="bg:#353535")
|
|
|
|
self.els = [self.select]+[ConditionalContainer(content=containIfNeeded(opts[i]),filter=Condition(lambda i=i: self.current==i)) for i in range(len(opts))]
|
|
self.el = VSplit(self.els)
|
|
|
|
self.returnFilter = has_focus(self.el) & ~has_focus(self.select)
|
|
@kb.add("c-left",filter=self.returnFilter)
|
|
@kb.add("c-h",filter=self.returnFilter)
|
|
@kb.add("escape",filter=self.returnFilter)
|
|
def _(e):
|
|
e.app.layout.focus(self.select)
|
|
|
|
def make_select_kb(self):
|
|
kb = KeyBindings()
|
|
|
|
l = len(self.optNames)
|
|
|
|
@kb.add("up")
|
|
def _(e):
|
|
self.current = (self.current+l-1)%l
|
|
|
|
@kb.add("down")
|
|
def _(e):
|
|
self.current = (self.current+1)%l
|
|
|
|
@kb.add("enter")
|
|
@kb.add("c-right")
|
|
@kb.add("c-l")
|
|
@kb.add(" ")
|
|
def _(e):
|
|
self.focusTabWindow()
|
|
|
|
return kb
|
|
|
|
def show_sel(self):
|
|
o = []
|
|
size = get_app().output.get_size()
|
|
height = size.rows
|
|
for i in range(len(self.opts)):
|
|
name = self.optNames[i]
|
|
o.append(("reverse",name) if i==self.current else ("",name))
|
|
o.append((""," \n"))
|
|
for i in range(height-len(self.opts)-2):
|
|
o.append(("","\n"))
|
|
if is_true(self.returnFilter):
|
|
o.append(("fg:#bbbbbb","[Ctrl-Left/ESC]"))
|
|
return o
|
|
|
|
def curpos(self):
|
|
return Point(x=len(self.optNames[self.current]),y=self.current)
|
|
|
|
def focusTabWindow(self):
|
|
if self.optFocus[self.current]:
|
|
try:
|
|
el = self.optFocus[self.current]
|
|
get_app().layout.focus(el)
|
|
return
|
|
except Exception:
|
|
pass
|
|
try:
|
|
el = self.opts[self.current]
|
|
get_app().layout.focus(el)
|
|
return
|
|
except Exception:
|
|
pass
|
|
try:
|
|
el = self.els[self.current+1]
|
|
get_app().layout.focus(el)
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
quitKB = KeyBindings()
|
|
|
|
@quitKB.add("c-q",eager=True)
|
|
def exit_(event):
|
|
event.app.exit()
|
|
|
|
def spawnDialog(dlg,floats):
|
|
app = get_app()
|
|
|
|
f = Float(content=dlg)
|
|
floats.append(f)
|
|
prev = app.layout.current_window
|
|
app.layout.focus(dlg)
|
|
app.invalidate()
|
|
|
|
def despawn():
|
|
floats.remove(f)
|
|
app.layout.focus(prev)
|
|
app.invalidate()
|
|
|
|
return despawn
|
|
|
|
async def spawnSimplePopup(title,body,floats):
|
|
title = str(title)
|
|
if type(body)==str:
|
|
body=Label(body)
|
|
|
|
res = asyncio.get_running_loop().create_future()
|
|
|
|
def handler():
|
|
if not res.done():
|
|
res.set_result(True)
|
|
despawn()
|
|
|
|
dlg = Dialog(
|
|
title=title,
|
|
body=body,
|
|
buttons=[
|
|
Button("OK",handler=handler)
|
|
]
|
|
)
|
|
|
|
despawn = spawnDialog(dlg,floats)
|
|
|
|
return await res
|
|
|
|
async def spawnYesNoPopup(title,body,floats):
|
|
title = str(title)
|
|
if type(body)==str:
|
|
body=Label(body)
|
|
|
|
res = asyncio.get_running_loop().create_future()
|
|
|
|
def handler(r):
|
|
if not res.done():
|
|
res.set_result(r)
|
|
despawn()
|
|
|
|
dlg = Dialog(
|
|
title=title,
|
|
body=body,
|
|
buttons=[
|
|
Button("Yes",handler=lambda: handler(True)),
|
|
Button("No",handler=lambda: handler(False))
|
|
]
|
|
)
|
|
|
|
despawn = spawnDialog(dlg,floats)
|
|
|
|
return await res
|
|
|
|
def HButtonSelect(buttonArgs):
|
|
kb = KeyBindings()
|
|
|
|
# layout = get_app().layout
|
|
|
|
btns = [Button(e[0],e[1],width=48) for e in buttonArgs]
|
|
|
|
first_selected = has_focus(btns[0])
|
|
last_selected = has_focus(btns[-1])
|
|
|
|
@kb.add("up",filter=~first_selected)
|
|
def _(e):
|
|
focus_previous(e)
|
|
|
|
@kb.add("down",filter=~last_selected)
|
|
def _(e):
|
|
focus_next(e)
|
|
|
|
return HSplit(btns,key_bindings=kb)
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
# Server/Other
|
|
# ------------------------------------------------------------
|
|
|
|
def errorTxt(obj):
|
|
if obj["server"]:
|
|
return f"Server error: {obj["error"]}"
|
|
return f"Client error: {obj["error"]}"
|
|
|
|
def isValidUsername(name):
|
|
if len(name)<3 or len(name)>16: return False
|
|
pattern = re.compile("^[a-z][a-z0-9-_]*")
|
|
match = pattern.match(name)
|
|
if match==None: return False
|
|
return match.start()==0 and match.end()==len(name)
|