better logging

examples/serve_dictionary.py

@@ -1,4 +1,4 @@
 from textual_serve.server import Server
 
 server = Server("python dictionary.py")
-server.serve(debug=True)
+server.serve(debug=False)

src/textual_serve/server.py

@@ -17,12 +17,38 @@ import jinja2
 
 from rich import print
 from rich.logging import RichHandler
+from rich.highlighter import RegexHighlighter
 
 from .app_service import AppService
 
 log = logging.getLogger("textual-serve")
 
 
+class LogHighlighter(RegexHighlighter):
+    base_style = "repr."
+    highlights = [
+        r"(?P<number>(?<!\w)\-?[0-9]+\.?[0-9]*(e[-+]?\d+?)?\b|0x[0-9a-fA-F]*)",
+        r"(?P<path>\[.*?\])",
+        r"(?<![\\\w])(?P<str>b?'''.*?(?<!\\)'''|b?'.*?(?<!\\)'|b?\"\"\".*?(?<!\\)\"\"\"|b?\".*?(?<!\\)\")",
+    ]
+
+
+def to_int(value: str, default: int) -> int:
+    """Convert to an integer, or return a default if that's not possible.
+
+    Args:
+        number: A string possibly containing a decimal.
+        default: Default value if value can't be decoded.
+
+    Returns:
+        Integer.
+    """
+    try:
+        return int(value)
+    except ValueError:
+        return default
+
+
 class Server:
     """Serve a Textual app."""
 
@@ -63,8 +89,6 @@ class Server:
         self.statics_path = base_path / statics_path
         self.templates_path = base_path / templates_path
 
-        self.initialize_logging()
-
     def initialize_logging(self) -> None:
         FORMAT = "%(message)s"
         logging.basicConfig(
@@ -72,17 +96,18 @@ class Server:
             format=FORMAT,
             datefmt="[%X]",
             handlers=[
-                RichHandler(show_path=False, show_time=False, rich_tracebacks=True)
+                RichHandler(
+                    show_path=False,
+                    show_time=False,
+                    rich_tracebacks=True,
+                    highlighter=LogHighlighter(),
+                )
             ],
         )
 
-    def request_exit(self, reason: str | None = None) -> None:
-        """Gracefully exit the application, optionally supplying a reason.
-
-        Args:
-            reason: The reason for exiting which will be included in the Ganglion server log.
-        """
-        log.info(f"Exiting - {reason if reason else ''}")
+    def request_exit(self) -> None:
+        """Gracefully exit the app."""
+        log.info("Exit requested")
         raise GracefulExit()
 
     async def _make_app(self) -> web.Application:
@@ -113,6 +138,8 @@ class Server:
 
         """
         self.debug = debug
+        self.initialize_logging()
+
         loop = asyncio.get_event_loop()
         loop.add_signal_handler(signal.SIGINT, self.request_exit)
         loop.add_signal_handler(signal.SIGTERM, self.request_exit)
@@ -129,6 +156,7 @@ class Server:
     @aiohttp_jinja2.template("app_index.html")
     async def handle_index(self, request: web.Request) -> dict[str, Any]:
         router = request.app.router
+        font_size = to_int(request.query.get("fontsize", "16"), 16)
 
         def get_url(route: str, **args) -> str:
             """Get a URL from the aiohttp router."""
@@ -136,7 +164,7 @@ class Server:
             return f"{self.public_url}{path}"
 
         context = {
-            "font_size": 14,
+            "font_size": font_size,
             "app_websocket_url": get_url("websocket"),
         }
         context["config"] = {
@@ -149,6 +177,37 @@ class Server:
         }
         return context
 
+    async def _process_messages(
+        self, websocket: web.WebSocketResponse, app_service: AppService
+    ) -> None:
+        """Process messages from the websocket.
+
+        Args:
+            websocket: Websocket instance.
+            app_service: App service.
+        """
+        TEXT = WSMsgType.TEXT
+
+        async for message in websocket:
+            if message.type != TEXT:
+                continue
+            envelope = message.json()
+            assert isinstance(envelope, list)
+            type_ = envelope[0]
+            if type_ == "stdin":
+                data = envelope[1]
+                await app_service.send_bytes(data.encode("utf-8"))
+            elif type_ == "resize":
+                data = envelope[1]
+                await app_service.set_terminal_size(data["width"], data["height"])
+            elif type_ == "ping":
+                data = envelope[1]
+                await websocket.send_json(["pong", data])
+            elif type_ == "blur":
+                await app_service.blur()
+            elif type_ == "focus":
+                await app_service.focus()
+
     async def handle_websocket(self, request: web.Request) -> web.WebSocketResponse:
         """Handle the websocket that drives the remote process.
 
@@ -160,30 +219,11 @@ class Server:
         """
         websocket = web.WebSocketResponse(heartbeat=15)
 
-        def to_int(value: str, default: int) -> int:
-            """Convert to an integer, or return a default if that's not possible.
-
-            Args:
-                number: A string possibly containing a decimal.
-                default: Default value if value can't be decoded.
-
-            Returns:
-                Integer.
-            """
-            try:
-                return int(value)
-            except ValueError:
-                return default
-
         width = to_int(request.query.get("width", "80"), 80)
         height = to_int(request.query.get("height", "24"), 24)
 
-        TEXT = WSMsgType.TEXT
-        BINARY = WSMsgType.BINARY
-
         try:
             await websocket.prepare(request)
-
             app_service = AppService(
                 self.command,
                 write_bytes=websocket.send_bytes,
@@ -192,31 +232,8 @@ class Server:
                 debug=self.debug,
             )
             await app_service.start(width, height)
-
             try:
-                async for message in websocket:
-                    if message.type == TEXT:
-                        envelope = message.json()
-                        assert isinstance(envelope, list)
-                        type_ = envelope[0]
-                        if type_ == "stdin":
-                            data = envelope[1]
-                            await app_service.send_bytes(data.encode("utf-8"))
-                        elif type_ == "resize":
-                            data = envelope[1]
-                            await app_service.set_terminal_size(
-                                data["width"], data["height"]
-                            )
-                        elif type_ == "ping":
-                            data = envelope[1]
-                            await websocket.send_json(["pong", data])
-                        elif type_ == "blur":
-                            await app_service.blur()
-                        elif type_ == "focus":
-                            await app_service.focus()
-
-                    elif message.type == BINARY:
-                        pass
+                await self._process_messages(websocket, app_service)
             finally:
                 await app_service.stop()