import sys, os, io import asyncio, threading, time, textwrap, re import encodings.aliases import prompt_toolkit as pt from prompt_toolkit import Application from prompt_toolkit.buffer import Buffer from prompt_toolkit.layout.containers import HSplit, VSplit, Window, ConditionalContainer from prompt_toolkit.layout.controls import BufferControl from prompt_toolkit.layout.layout import Layout from prompt_toolkit.layout import FormattedTextControl, Container, is_container, FloatContainer, Float, WindowAlign from prompt_toolkit.widgets import Box, Button, Dialog, Label, TextArea, ValidationToolbar, RadioList from prompt_toolkit.layout.dimension import Dimension as D from prompt_toolkit.application.current import get_app from prompt_toolkit.validation import Validator from prompt_toolkit.filters import Condition, has_focus, is_true from prompt_toolkit.key_binding import KeyBindings, merge_key_bindings from prompt_toolkit.key_binding.defaults import load_key_bindings from prompt_toolkit.data_structures import Point from prompt_toolkit.key_binding.bindings.focus import focus_next, focus_previous from prompt_toolkit.output import ColorDepth, create_output parentdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(parentdir) from api import ServerConnection server = ServerConnection() 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) intset = {} # interface settings def getColorDepth(): try: return ColorDepth(f"DEPTH_{intset["color"]}_BIT") except ValueError: return ColorDepth.DEPTH_4_BIT def defaultAppSettings(): return {"full_screen":True, "color_depth":getColorDepth(), "output": createEncodedOutput(intset["encoding"])} 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 containIfNeeded(el): if not is_container(el): return Window(el) return el def message_dialog( # derived from prompt_toolkit's code 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() ) 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() floats = [] def spawnDialog(dlg): global floats app = get_app() f = Float(content=dlg) floats.append(f) prev = app.layout.current_window app.layout.focus(dlg) app.invalidate() def despawn(): global floats floats.remove(f) app.layout.focus(prev) app.invalidate() return despawn async def spawnSimplePopup(title,body): 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) 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) async def mainScreen(): kb = KeyBindings() @kb.add("c-q",eager=True) def exit_(event): event.app.exit() stat = await server.status() if stat["type"]=="error" or not stat["response"]["login"]: return name = stat["response"]["name"] # ------------------------------------------------------------ # Settings -> Accounts -> Log Out # ------------------------------------------------------------ def logout(): get_app().exit(result="login") # ------------------------------------------------------------ # Settings -> Accounts -> Change Name # ------------------------------------------------------------ def changename(): usernameValidator = Validator.from_callable(isValidUsername, error_message='Invalid username') def nameaccept(buf: Buffer): # get_app().layout.focus(okbtn) dlgclose(True) return True nntxtfield = TextArea( text=name, multiline=False, password=False, completer=None, validator=usernameValidator, accept_handler=nameaccept, ) okbtn = Button("OK",handler=lambda: dlgclose(True)) cancelbtn = Button("Cancel",handler=lambda: dlgclose(False)) async def _changename(): res = await server.changename(nntxtfield.text) if res["type"]=="error": await spawnSimplePopup("Error",errorTxt(res)) else: get_app().exit(result="restart") def dlgclose(res): despawn() if not res: return get_app().create_background_task(_changename()) dlg = Dialog( title="Change name", body=HSplit([ Label(text="Insert your account's new name here:", dont_extend_height=True), nntxtfield, Label(text="Keep in mind that you can only change your name every 7 days.", dont_extend_height=True), ValidationToolbar(), ]), buttons=[okbtn, cancelbtn] ) despawn = spawnDialog(dlg) # ------------------------------------------------------------ # Settings -> Accounts -> Change Password # ------------------------------------------------------------ def changepass(): def oldaccept(buf: Buffer): get_app().layout.focus(newtxtfield) return True def newaccept(buf: Buffer): # get_app().layout.focus(okbtn) dlgclose(True) return True oldtxtfield = TextArea( text="", multiline=False, password=True, completer=None, validator=None, accept_handler=oldaccept, ) newtxtfield = TextArea( text="", multiline=False, password=True, completer=None, validator=None, accept_handler=newaccept, ) okbtn = Button("OK",handler=lambda: dlgclose(True)) cancelbtn = Button("Cancel",handler=lambda: dlgclose(False)) async def _changepass(): res = await server.changepass(oldtxtfield.text,newtxtfield.text) if res["type"]=="error": await spawnSimplePopup("Error",errorTxt(res)) def dlgclose(res): despawn() if not res: return get_app().create_background_task(_changepass()) dlg = Dialog( title="Change password", body=HSplit([ Label(text="Old password:", dont_extend_height=True), oldtxtfield, Label(text="New password:", dont_extend_height=True), newtxtfield, ValidationToolbar(), ]), buttons=[okbtn, cancelbtn] ) despawn = spawnDialog(dlg) # ------------------------------------------------------------ # Settings -> Interface -> Change color depth # ------------------------------------------------------------ async def _updateisettings(): res = await server.setisettings("basic",intset) if res["type"]=="error": if res["server"]: await spawnSimplePopup("Error",f"{errorTxt(res)}\nYour changes might not have been sent to the internal server. Please try again.") else: await spawnSimplePopup("Error",errorTxt(res)) def changecolor(): def dlgclose(): despawn() intset["color"] = colorselect.current_value get_app().create_background_task(_updateisettings()) ok_button = Button(text="OK", handler=dlgclose, width=16) colorselect = RadioList([(24,"24 bits (True Color)"),(8,"8 bits (Paletted)"),(4,"4 bits (ANSI/RGBI)"),(1,"1 bit (Monochrome)")],default=intset["color"]) dlg = Dialog( title="Change color depth", body=HSplit( [ Label(text="Please select a color depth supported by this terminal:", dont_extend_height=True), colorselect, Label(text="Use 1 or 4 bits for compatibility with older terminals, otherwise 8 or 24 bits.", dont_extend_height=True), Label(text="The more bits in the color depth, the more colors there can be displayed on-screen."), ValidationToolbar(), ], padding=D(preferred=1, max=1), ), buttons=[ok_button], with_background=True, ) despawn = spawnDialog(dlg) def changeencother(): def dlgclose(): despawn() intset["encoding"] = encselect.current_value get_app().create_background_task(_updateisettings()) ok_button = Button(text="OK", handler=dlgclose, width=16) encodingList = sorted(set(encodings.aliases.aliases.values())) for e in encodingList.copy(): try: ("test").encode(e,errors="replace") except LookupError: encodingList.remove(e) encselect = RadioList([(e,e) for e in encodingList],default=intset["encoding"]) dlg = Dialog( title="Change text encoding", body=HSplit( [ Label(text="Please select the encoding supported by this terminal:", dont_extend_height=True), encselect, Label(text="For more information about every supported encoding, open this webpage:\nhttps://docs.python.org/3.14/library/codecs.html#standard-encodings",dont_extend_height=True), ValidationToolbar(), ], padding=D(preferred=1, max=1), ), buttons=[ok_button], with_background=True, ) despawn = spawnDialog(dlg) def changeenc(): def dlgclose(): despawn() v = encselect.current_value if v=="other": changeencother() else: intset["encoding"] = v get_app().create_background_task(_updateisettings()) ok_button = Button(text="OK", handler=dlgclose, width=16) default = intset["encoding"] if not default in ["utf8","ascii","latin_1","cp437"]: default = "other" encselect = RadioList([("utf8","UTF-8 (Unicode)"),("ascii","ASCII (7-bit)"),("latin_1","Latin-1/ISO-8859-1 (Ext. ASCII)"),("cp437","CP437 (Extended ASCII)"),("other","Other (Python-supported)")],default=default) dlg = Dialog( title="Change text encoding", body=HSplit( [ Label(text="Please select the encoding supported by this terminal:", dont_extend_height=True), encselect, Label(text="UTF-8 is generally reccomended for a normal modern terminal, and supports everything from Unicode.",dont_extend_height=True), Label(text="ASCII is very barebones but is supported by every terminal. Please use it if all else fails.", dont_extend_height=True), Label(text="You may see garbled text if you chose the wrong text encoding.",dont_extend_height=True), ValidationToolbar(), ], padding=D(preferred=1, max=1), ), buttons=[ok_button], with_background=True, ) despawn = spawnDialog(dlg) # ------------------------------------------------------------ # Settings -> Interface (Status) # ------------------------------------------------------------ def showISettings(): color = ({1: "1 bit (Monochrome)",4: "4 bits (ANSI/RGBI)",8: "8 bits",24: "24 bits (True Color)"})[intset["color"]] encoding = intset["encoding"] if encoding=="utf8": encoding="UTF-8 (Unicode)" if encoding=="ascii": encoding="ASCII (7-bit)" if encoding=="cp437": encoding="CP437 (Extended ASCII)" if encoding=="latin_1": encoding="Latin-1/ISO-8859-1 (Ext. ASCII)" return [ ("","Color depth: "),("bold",color),("","\n"), ("","Text encoding: "),("bold",encoding),("","\n") ] # ------------------------------------------------------------ # Settings Menu # ------------------------------------------------------------ settings = HSelectLayout(kb,[ HButtonSelect([("Log out",logout),("Change username",changename),("Change password",changepass)]), HSplit([ Window(FormattedTextControl(text=showISettings,focusable=False),dont_extend_height=True), Box(HButtonSelect([("Change color depth",changecolor),("Change text encoding",changeenc)]),padding=0), Label("\n\nYou will need to restart or relog to apply any new changes."), ]) ],[ None, None ],["Account","Interface"]) # ------------------------------------------------------------ # Tab Layout # ------------------------------------------------------------ tablayout = TabLayout(kb,[ Label("Not implemented yet 1"), Label("Not implemented yet 2"), settings.el ],[ None, None, settings.select ],["Forum Boards","E-Messages","Settings"],name) # ------------------------------------------------------------ # Application # ------------------------------------------------------------ layout = Layout(FloatContainer(content=tablayout.el,floats=floats),focused_element=tablayout.tabs[0]) app = Application(layout=layout, key_bindings=kb, **defaultAppSettings()) app.pre_run_callables.append(tablayout.focusTabWindow) return await app.run_async() async def firstSetupScreen(): def ok_handler() -> None: get_app().exit() ok_button = Button(text="OK", handler=ok_handler, width=16) colorselect = RadioList([(24,"24 bits (True Color)"),(8,"8 bits (Paletted)"),(4,"4 bits (ANSI/RGBI)"),(1,"1 bit (Monochrome)")],default=8) encselect = RadioList([("utf8","UTF-8 (Unicode)"),("ascii","ASCII (7-bit)"),("latin_1","ISO-8859-1 (Ext. ASCII)"),("cp437","CP437 (Extended ASCII)")],default="utf8") cdialog = Dialog( title="First Setup", body=HSplit( [ Label(text="Please select a color depth supported by this terminal:", dont_extend_height=True), colorselect, Label(text="Use 1 or 4 bits for compatibility with older terminals, otherwise 8 or 24 bits.", dont_extend_height=True), Label(text="You can always change this later in the interface settings.",dont_extend_height=True), ValidationToolbar(), ], padding=D(preferred=1, max=1), ), buttons=[ok_button], with_background=True, ) capp = Application(layout=Layout(cdialog), **defaultAppSettings()) await capp.run_async() edialog = Dialog( title="First Setup", body=HSplit( [ Label(text="Please insert the text encoding supported by this terminal:", dont_extend_height=True), encselect, Label(text="Use ASCII for compatibility with older terminals, otherwise UTF-8.", dont_extend_height=True), Label(text="You can always change this later in the interface settings.",dont_extend_height=True), ValidationToolbar(), ], padding=D(preferred=1, max=1), ), buttons=[ok_button], with_background=True, ) eapp = Application(layout=Layout(edialog), **defaultAppSettings()) await eapp.run_async() return {"color":colorselect.current_value,"encoding":encselect.current_value} async def createAccountScreen(): def nameaccept(buf: Buffer) -> bool: get_app().layout.focus(passtextfield) return True # Keep text. def pwrdaccept(buf: Buffer) -> bool: # get_app().layout.focus(login_button) get_app().layout.focus(passtextconffield) return True # Keep text. def pwrdcaccept(buf: Buffer) -> bool: # get_app().layout.focus(login_button) ok_handler() return True # Keep text. def _return_none() -> None: "Button handler that returns None." get_app().exit() def ok_handler() -> None: get_app().exit(result=(usertextfield.text,passtextfield.text)) def login_handler() -> None: get_app().exit(result="login") create_button = Button(text="Create Account", handler=ok_handler, width=16) login_button = Button(text="Log In", handler=login_handler, width=16) exit_button = Button(text="Exit", handler=_return_none, width=16) usernameValidator = Validator.from_callable(isValidUsername, error_message='Invalid username') samePassValidator = Validator.from_callable(lambda x: passtextfield.text==x, error_message='Passwords are not the same') usertextfield = TextArea( text="", multiline=False, password=False, completer=None, validator=usernameValidator, accept_handler=nameaccept, ) passtextfield = TextArea( text="", multiline=False, password=True, completer=None, validator=None, accept_handler=pwrdaccept, ) passtextconffield = TextArea( text="", multiline=False, password=True, completer=None, validator=samePassValidator, accept_handler=pwrdcaccept, ) dialog = Dialog( title="Create a StingNET account", body=HSplit( [ Label(text="Username:", dont_extend_height=True), usertextfield, Label(text="Password:", dont_extend_height=True), passtextfield, Label(text="Password (confirm):", dont_extend_height=True), passtextconffield, ValidationToolbar(), ], padding=D(preferred=1, max=1), ), buttons=[create_button, login_button, exit_button], with_background=True, ) app = Application(key_bindings=quitKB, layout=Layout(dialog), **defaultAppSettings()) while True: usertextfield.text = "" passtextfield.text = "" passtextconffield.text = "" appres = await app.run_async() if type(appres)==tuple: name,pwrd = appres elif type(appres)==str and appres=="login": return "login" else: await server.disconnect() exit() res = await server.createuser(name,pwrd) if res["type"]=="response": lres = await server.login(name,pwrd) if res["type"]=="error": await message_dialog(title="Error (create)",text=f"The account has not been created.\n{errorTxt(res)}").run_async() if lres["type"]=="error": await message_dialog(title="Error (login)",text=f"The account has been created, but has not been logged in.\n{errorTxt(lres)}").run_async() else: break async def loginCredsScreen(): def nameaccept(buf: Buffer) -> bool: get_app().layout.focus(passtextfield) return True # Keep text. def pwrdaccept(buf: Buffer) -> bool: # get_app().layout.focus(login_button) ok_handler() return True # Keep text. def _return_none() -> None: "Button handler that returns None." get_app().exit() def ok_handler() -> None: get_app().exit(result=(usertextfield.text,passtextfield.text)) def create_handler() -> None: get_app().exit(result="create") login_button = Button(text="Log In", handler=ok_handler, width=16) create_button = Button(text="Create Account", handler=create_handler, width=16) exit_button = Button(text="Exit", handler=_return_none, width=16) usernameValidator = Validator.from_callable(isValidUsername, error_message='Invalid username') usertextfield = TextArea( text="", multiline=False, password=False, completer=None, validator=usernameValidator, accept_handler=nameaccept, ) passtextfield = TextArea( text="", multiline=False, password=True, completer=None, validator=None, accept_handler=pwrdaccept, ) dialog = Dialog( title="Log in to StingNET", body=HSplit( [ Label(text="Username:", dont_extend_height=True), usertextfield, Label(text="Password:", dont_extend_height=True), passtextfield, ValidationToolbar(), ], padding=D(preferred=1, max=1), ), buttons=[login_button, create_button, exit_button], with_background=True, ) app = Application(key_bindings=quitKB, layout=Layout(dialog), **defaultAppSettings()) while True: usertextfield.text = "" passtextfield.text = "" appres = await app.run_async() if type(appres)==tuple: name,pwrd = appres elif type(appres)==str and appres=="create": return "create" else: await server.disconnect() exit() res = await server.login(name,pwrd) if res["type"]=="error": await message_dialog(title="Error",text=errorTxt(res)).run_async() else: break async def logoutScreen(): while True: res = await loginCredsScreen() if res != "create": return res = await createAccountScreen() if res != "login": return async def main(): global intset try: await asyncio.wait_for(server.connect(), timeout=10) except TimeoutError: print("The internal server is currently down. Please press Ctrl-C (you may see a KeyboardInterrupt error).") exit() try: while True: stat = await server.status() if stat["type"]=="error": continue if stat["response"]["login"]: ires = await server.getisettings("basic") if ires["type"]=="error": intset = await firstSetupScreen() ires = await server.setisettings("basic",intset) if ires["type"]=="error": await message_dialog(title="Error",text=f"{errorTxt(ires)}\n\nThese settings might not have been sent to the internal server.\nYou will be asked again the next time you log in to your account.").run_async() else: intset = ires["response"] res = await mainScreen() if res!="restart": if res!="login": break await server.logout() else: intset = {"color":4,"encoding":"ascii"} await logoutScreen() finally: await server.disconnect() asyncio.run(main())