Include debug plugin

Author: madduck

README.md

@@ -62,6 +62,9 @@ DEBUG:root:Finished.
 
 I hope you get the idea.
 
+There's also a "debug" plugin included, which allows interaction with the CLI as
+its running via key presses. Hit `?` to get an overview of commands available.
+
 Looking forward to your feedback.
 
 Oh, and if someone wanted to turn this into a proper package with tests and everything, I think it could be published to pip/pypy. I need to stop shaving this yak now, though.

click_async_plugins/__init__.py

@@ -1,18 +1,26 @@
 from .command import plugin
 from .core import cli_core, runner
 from .group import plugin_group
-from .itc import ITC, pass_itc
+from .itc import ITC
 from .typedefs import PluginFactory, PluginLifespan
-from .util import create_plugin_task, run_plugins, run_tasks, setup_plugins
+from .util import (
+    CliContext,
+    create_plugin_task,
+    pass_clictx,
+    run_plugins,
+    run_tasks,
+    setup_plugins,
+)
 
 __all__ = [
+    "CliContext",
     "cli_core",
     "create_plugin_task",
     "ITC",
-    "pass_itc",
+    "pass_clictx",
     "plugin",
-    "plugin_group",
     "PluginFactory",
+    "plugin_group",
     "PluginLifespan",
     "runner",
     "run_plugins",

click_async_plugins/core.py

@@ -2,16 +2,17 @@
 
 import click
 
+from click_async_plugins.itc import ITC
+
 from .group import plugin_group
-from .itc import ITC
 from .typedefs import PluginFactory
-from .util import run_plugins
+from .util import CliContext, run_plugins
 
 
 @plugin_group
 @click.pass_context
 def cli_core(ctx: click.Context) -> None:
-    ctx.ensure_object(ITC)
+    ctx.obj = CliContext(itc=ITC())
 
 
 @cli_core.result_callback()

click_async_plugins/debug.py

@@ -0,0 +1,167 @@
+# needed < 3.14 so that annotations aren't evaluated
+from __future__ import annotations
+
+import asyncio
+import datetime
+import logging
+import os
+import sys
+from collections.abc import Callable
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from functools import partial
+
+from click_async_plugins.util import CliContext
+
+from . import PluginLifespan, pass_clictx, plugin
+
+logger = logging.getLogger(__name__)
+
+
+def puts(s: str) -> None:
+    print(s, file=sys.stderr)
+
+
+def simulate_reload_tpdata(clictx: CliContext) -> None:
+    """Simulate event that TPData was reloaded"""
+    clictx.itc.fire("tpdata")
+
+
+def echo_newline(_: CliContext) -> None:
+    """Outputs a new line"""
+    puts("")
+
+
+def terminal_block(_: CliContext) -> None:
+    """Outputs a couple of newlines and the current time"""
+    puts(f"{'\n' * 8}The time is now: {datetime.datetime.now().isoformat(sep=' ')}\n")
+
+
+def debug_info(clictx: CliContext) -> None:
+    """Prints debugging information on tasks and CliContext"""
+    puts("*** BEGIN DEBUG INFO: ***")
+    puts("Tasks:")
+    for i, task in enumerate(asyncio.all_tasks(asyncio.get_event_loop()), 1):
+        coro = task.get_coro()
+        puts(
+            f"  {i:02n}  {task.get_name():32s}  "
+            f"state={task._state.lower():8s}  "
+            f"coro={None if coro is None else coro.__qualname__}"
+        )
+    puts("CliContext:")
+    maxlen = max([len(k) for k in clictx.__dict__.keys()])
+    for attr, value in clictx.__dict__.items():
+        puts(f"  {attr:>{maxlen}s} = {value!r}")
+    puts("*** END DEBUG INFO: ***")
+
+
+_LOGLEVELS = {
+    logging.DEBUG: "DEBUG",
+    logging.INFO: "INFO",
+    logging.WARN: "WARN",
+    logging.ERROR: "ERROR",
+    logging.CRITICAL: "CRITICAL",
+}
+
+
+def adjust_loglevel(_: CliContext, change: int) -> None:
+    """Adjusts the log level"""
+    rootlogger = logging.getLogger()
+    newlevel = rootlogger.getEffectiveLevel() + change
+    if newlevel < logging.DEBUG or newlevel > logging.CRITICAL:
+        return
+
+    rootlogger.setLevel(newlevel)
+    puts(f"Log level now at {_LOGLEVELS[logger.getEffectiveLevel()]}")
+
+
+@dataclass
+class KeyAndFunc:
+    key: str
+    func: Callable[[CliContext], None]
+
+
+type KeyCmdMapType = dict[int, KeyAndFunc]
+
+
+def print_help(_: CliContext, key_to_cmd: KeyCmdMapType) -> None:
+    puts("Keys I know about for debugging:")
+    for keyfunc in key_to_cmd.values():
+        puts(f"  {keyfunc.key:5s} {keyfunc.func.__doc__}")
+    puts("  ?     Print this message")
+
+
+try:
+    import fcntl
+    import termios
+    import tty
+
+    async def _monitor_stdin(clictx: CliContext, key_to_cmd: KeyCmdMapType) -> None:
+        fd = sys.stdin.fileno()
+        termios_saved = termios.tcgetattr(fd)
+        fnctl_flags = fcntl.fcntl(sys.stdin, fcntl.F_GETFL)
+
+        try:
+            logger.debug("Configuring stdin for raw input")
+            tty.setcbreak(fd)
+            fcntl.fcntl(sys.stdin, fcntl.F_SETFL, fnctl_flags | os.O_NONBLOCK)
+
+            while True:
+                ch = sys.stdin.read(1)
+
+                if len(ch) == 0:
+                    await asyncio.sleep(0.1)
+                    continue
+
+                if (key := ord(ch)) == 0x3F:
+                    print_help(clictx, key_to_cmd)
+
+                elif (keyfunc := key_to_cmd.get(key)) is not None and callable(
+                    keyfunc.func
+                ):
+                    keyfunc.func(clictx)
+
+                else:
+                    logger.debug(f"Ignoring character 0x{key:02x} on stdin")
+
+        finally:
+            logger.debug("Restoring stdin")
+            termios.tcsetattr(fd, termios.TCSADRAIN, termios_saved)
+            fcntl.fcntl(sys.stdin, fcntl.F_SETFL, fnctl_flags)
+
+except ImportError:
+
+    async def _monitor_stdin(clictx: CliContext, key_to_cmd: KeyCmdMapType) -> None:
+        _ = clictx, key_to_cmd
+        logger.warning("The 'debug' plugin does not work on this platform")
+        return None
+
+
+@asynccontextmanager
+async def monitor_stdin_for_debug_commands(clictx: CliContext) -> PluginLifespan:
+    increase_loglevel = partial(adjust_loglevel, change=-10)
+    increase_loglevel.__doc__ = "Increase the logging level"
+    decrease_loglevel = partial(adjust_loglevel, change=10)
+    decrease_loglevel.__doc__ = "Decrease the logging level"
+
+    key_to_cmd = {
+        0xA: KeyAndFunc(r"\n", echo_newline),
+        0x12: KeyAndFunc("^R", simulate_reload_tpdata),
+        0x1B: KeyAndFunc("<Esc>", terminal_block),
+        0x4: KeyAndFunc("^D", debug_info),
+        0x2B: KeyAndFunc("+", increase_loglevel),
+        0x2D: KeyAndFunc("-", decrease_loglevel),
+    }
+    yield _monitor_stdin(clictx, key_to_cmd)
+
+
+@plugin
+@pass_clictx
+async def debug(clictx: CliContext) -> PluginLifespan:
+    """Monitor stdin for keypresses to trigger debugging functions
+
+    Press '?' to get a list of possible keys.
+    """
+
+    async with monitor_stdin_for_debug_commands(clictx) as task:
+        yield task

click_async_plugins/itc.py

@@ -4,8 +4,6 @@
 from collections.abc import AsyncGenerator
 from typing import Any
 
-import click
-
 logger = logging.getLogger(__name__)
 
 
@@ -47,6 +45,3 @@ def has_subscribers(self, key: str) -> bool:
 
     def knows_about(self, key: str) -> bool:
         return key in self._objects
-
-
-pass_itc = click.make_pass_decorator(ITC)

click_async_plugins/util.py

@@ -2,14 +2,26 @@
 import logging
 from collections.abc import Callable
 from contextlib import AsyncExitStack
+from dataclasses import dataclass
 from functools import partial, update_wrapper
 from typing import Any, Never
 
+import click
+
+from .itc import ITC
 from .typedefs import PluginFactory, PluginTask
 
 logger = logging.getLogger(__name__)
 
 
+@dataclass
+class CliContext:
+    itc: ITC
+
+
+pass_clictx = click.make_pass_decorator(CliContext)
+
+
 async def sleep_forever(sleep: float = 1, *, forever: bool = True) -> Never | None:
     while await asyncio.sleep(sleep, forever):
         pass

demo.py

@@ -3,7 +3,14 @@
 
 import click
 
-from click_async_plugins import ITC, PluginLifespan, cli_core, pass_itc, plugin
+from click_async_plugins import (
+    CliContext,
+    PluginLifespan,
+    cli_core,
+    pass_clictx,
+    plugin,
+)
+from click_async_plugins.debug import debug
 
 logging.basicConfig(level=logging.DEBUG)
 logger = logging.getLogger()
@@ -22,13 +29,15 @@
 @click.option(
     "--sleep", "-s", type=float, default=1, help="Sleep this long between counts"
 )
-@pass_itc
-async def countdown(itc: ITC, start: int = 3, sleep: float = 1) -> PluginLifespan:
+@pass_clictx
+async def countdown(
+    clictx: CliContext, start: int = 3, sleep: float = 1
+) -> PluginLifespan:
     async def counter(start: int, sleep: float) -> None:
         cur = start
         while cur > 0:
             logger.info(f"Counting down… {cur}")
-            itc.set("countdown", cur)
+            clictx.itc.set("countdown", cur)
             cur = await asyncio.sleep(sleep, cur - 1)
 
         logger.info("Finished counting down")
@@ -45,10 +54,10 @@ async def counter(start: int, sleep: float) -> None:
     is_flag=True,
     help="Don't wait for first update but echo right upon start",
 )
-@pass_itc
-async def echo(itc: ITC, immediately: bool) -> PluginLifespan:
+@pass_clictx
+async def echo(clictx: CliContext, immediately: bool) -> PluginLifespan:
     async def reactor() -> None:
-        async for cur in itc.updates("countdown", yield_immediately=immediately):
+        async for cur in clictx.itc.updates("countdown", yield_immediately=immediately):
             logger.info(f"Countdown currently at {cur}")
 
     yield reactor()
@@ -57,6 +66,7 @@ async def reactor() -> None:
 
 
 cli_core.add_command(echo)
+cli_core.add_command(debug)
 
 if __name__ == "__main__":
     import sys