Added basic console functionality

sbapp/main.py

@@ -2857,7 +2857,7 @@ class SidebandApp(MDApp):
             self.information_screen.ids.information_scrollview.effect_cls = ScrollEffect
             self.information_screen.ids.information_logo.icon = self.sideband.asset_dir+"/rns_256.png"
 
-            str_comps  =   " - [b]Reticulum[/b] (MIT License)\n - [b]LXMF[/b] (MIT License)\n - [b]KivyMD[/b] (MIT License)"
+            str_comps  =   " - [b]Reticulum[/b] (Reticulum License)\n - [b]LXMF[/b] (Reticulum License)\n - [b]KivyMD[/b] (MIT License)"
             str_comps += "\n - [b]Kivy[/b] (MIT License)\n - [b]Codec2[/b] (LGPL License)\n - [b]PyCodec2[/b] (BSD-3 License)"
             str_comps += "\n - [b]PyDub[/b] (MIT License)\n - [b]PyOgg[/b] (Public Domain)\n - [b]FFmpeg[/b] (GPL3 License)"
             str_comps += "\n - [b]GeoidHeight[/b] (LGPL License)\n - [b]Paho MQTT[/b] (EPL2 License)\n - [b]Python[/b] (PSF License)"
@@ -6442,6 +6442,7 @@ def run():
             config_path=args.config,
             is_client=False,
             verbose=(args.verbose or __debug_build__),
+            quiet=(args.interactive and not args.verbose),
             is_daemon=True
         )
 
@@ -6449,10 +6450,10 @@ def run():
         sideband.start()
         
         if args.interactive:
-            global sbcore; sbcore = sideband
+            while not sideband.getstate("core.started") == True: time.sleep(0.1)
-            while not sbcore.getstate("core.started") == True: time.sleep(0.1)
+            from .sideband import console
-            time.sleep(1)
+            console.attach(sideband)
-            import code; code.interact(local=globals())
+            
         else:
             while True: time.sleep(5)
     else:

sbapp/sideband/console.py

@@ -0,0 +1,103 @@
+import os
+import RNS
+import threading
+from prompt_toolkit.application import Application
+from prompt_toolkit.document import Document
+from prompt_toolkit.key_binding import KeyBindings
+from prompt_toolkit.layout.containers import HSplit, Window
+from prompt_toolkit.layout.layout import Layout
+from prompt_toolkit.styles import Style
+from prompt_toolkit.widgets import SearchToolbar, TextArea
+
+sideband = None
+application = None
+output_document = Document(text="", cursor_position=0)
+output_field = None
+
+def attach(target_core):
+    global sideband
+    sideband = target_core
+    RNS.logdest = RNS.LOG_CALLBACK
+    RNS.logcall = receive_output
+    console()
+
+def parse(uin):
+    args = uin.split(" ")
+    cmd = args[0]
+    if   cmd == "q" or cmd == "quit": quit_action()
+    elif cmd == "clear": cmd_clear(args)
+    elif cmd == "raw": cmd_raw(args, uin.replace("raw ", ""))
+    elif cmd == "log": cmd_log(args)
+    else: receive_output(f"Unknown command: {cmd}")
+
+def cmd_clear(args):
+    output_document = output_document = Document(text="", cursor_position=0)
+    output_field.buffer.document = output_document
+
+def cmd_raw(args, expr):
+    if expr != "" and expr != "raw":
+        try: receive_output(eval(expr))
+        except Exception as e: receive_output(str(e))
+
+def cmd_log(args):
+    try:
+        if len(args) == 1: receive_output(f"Current loglevel is {RNS.loglevel}")
+        else: RNS.loglevel = int(args[1]); receive_output(f"Loglevel set to {RNS.loglevel}")
+    except Exception as e:
+        receive_output("Invalid loglevel: {e}")
+
+def set_log(level=None):
+    if level: RNS.loglevel = level
+    if RNS.loglevel == 0: receive_output("Logging squelched. Use log command to print output to console.")
+
+def quit_action():
+    receive_output("Shutting down Sideband...")
+    sideband.should_persist_data()
+    application.exit()
+
+def receive_output(msg):
+    global output_document, output_field
+    content = f"{output_field.text}\n{msg}"
+    output_document = output_document = Document(text=content, cursor_position=len(content))
+    output_field.buffer.document = output_document
+
+def console():
+    global output_document, output_field, application
+    search_field = SearchToolbar()
+
+    output_field = TextArea(style="class:output-field", text="Sideband console ready")
+    input_field = TextArea(
+        height=1,
+        prompt="> ",
+        style="class:input-field",
+        multiline=False,
+        wrap_lines=False,
+        search_field=search_field)
+
+    container = HSplit([
+        output_field,
+        Window(height=1, char="-", style="class:line"),
+        input_field,
+        search_field])
+
+    def accept(buff): parse(input_field.text)
+    input_field.accept_handler = accept
+
+    kb = KeyBindings()
+    @kb.add("c-c")
+    @kb.add("c-q")
+    def _(event): quit_action()
+
+    style = Style([
+        ("line", "#004444"),
+    ])
+
+    application = Application(
+        layout=Layout(container, focused_element=input_field),
+        key_bindings=kb,
+        style=style,
+        mouse_support=True,
+        full_screen=False)
+
+    set_log()
+    application.run()

sbapp/sideband/core.py

@@ -148,7 +148,7 @@ class SidebandCore():
 
             self.log_announce(destination_hash, app_data, dest_type=SidebandCore.aspect_filter, stamp_cost=sc, link_stats=link_stats)
 
-    def __init__(self, owner_app, config_path = None, is_service=False, is_client=False, android_app_dir=None, verbose=False, owner_service=None, service_context=None, is_daemon=False, load_config_only=False):
+    def __init__(self, owner_app, config_path = None, is_service=False, is_client=False, android_app_dir=None, verbose=False, quiet=False, owner_service=None, service_context=None, is_daemon=False, load_config_only=False):
         self.is_service = is_service
         self.is_client = is_client
         self.is_daemon = is_daemon
@@ -164,7 +164,8 @@ class SidebandCore():
         else:
             self.is_standalone = False
 
-        self.log_verbose = verbose
+        self.log_verbose = (verbose and not quiet)
+        self.log_quiet = quiet
         self.log_deque = deque(maxlen=self.LOG_DEQUE_MAXLEN)
         self.owner_app = owner_app
         self.reticulum = None
@@ -4023,10 +4024,9 @@ class SidebandCore():
 
     def _reticulum_log_debug(self, debug=False):
         self.log_verbose = debug
-        if self.log_verbose:
+        if self.log_quiet: selected_level = 0
-            selected_level = 6
+        elif self.log_verbose: selected_level = 6
-        else:
+        else: selected_level = 2
-            selected_level = 2
 
         RNS.loglevel = selected_level
         if self.is_client:
@@ -4041,7 +4041,9 @@ class SidebandCore():
         return "\n".join(self.log_deque)
 
     def __start_jobs_immediate(self):
-        if self.log_verbose:
+        if self.log_quiet:
+            selected_level = 0
+        elif self.log_verbose:
             selected_level = 6
         else:
             selected_level = 2