357 rich and textual (#367)

* Setup Qodana, install Textual

* Reset to just rich

* Prep for using rich table

* First pass re-implement using live table

* Table is too laggy

* Use Logging handler

* Bracket time stamp

.github/workflows/qodana_code_quality.yml

@@ -0,0 +1,19 @@
+name: Qodana
+on:
+  workflow_dispatch:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  qodana:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+      - name: 'Qodana Scan'
+        uses: JetBrains/[email protected]
+        env:
+          QODANA_TOKEN: ${{ secrets.QODANA_TOKEN }}

pyproject.toml

@@ -38,6 +38,7 @@ dependencies = [
     "pythonnet==3.0.3",
     "requests==2.32.3",
     "sensapex==1.400.1",
+    "rich==13.7.1",
     "vbl-aquarium==0.0.19"
 ]
 
@@ -114,4 +115,7 @@ exclude_lines = [
     "no cov",
     "if __name__ == .__main__.:",
     "if TYPE_CHECKING:",
-]
+]
+
+[tool.ruff.lint]
+extend-ignore = ["DTZ005"]

qodana.yaml

@@ -0,0 +1,29 @@
+#-------------------------------------------------------------------------------#
+#               Qodana analysis is configured by qodana.yaml file               #
+#             https://www.jetbrains.com/help/qodana/qodana-yaml.html            #
+#-------------------------------------------------------------------------------#
+version: "1.0"
+
+#Specify inspection profile for code analysis
+profile:
+  name: qodana.starter
+
+#Enable inspections
+#include:
+#  - name: <SomeEnabledInspectionId>
+
+#Disable inspections
+#exclude:
+#  - name: <SomeDisabledInspectionId>
+#    paths:
+#      - <path/where/not/run/inspection>
+
+#Execute shell command before Qodana execution (Applied in CI/CD pipeline)
+#bootstrap: sh ./prepare-qodana.sh
+
+#Install IDE plugins before Qodana execution (Applied in CI/CD pipeline)
+#plugins:
+#  - id: <plugin.id> #(plugin id can be found at https://plugins.jetbrains.com)
+
+#Specify Qodana linter for analysis (Applied in CI/CD pipeline)
+linter: jetbrains/qodana-python:latest

scripts/logger_test.py

@@ -0,0 +1,17 @@
+import logging
+
+from rich.logging import RichHandler
+
+logging.basicConfig(level="NOTSET", format="%(message)s", datefmt="[%X]", handlers=[RichHandler(rich_tracebacks=True)])
+
+log = logging.getLogger("rich")
+log.debug("This message should go to the log file")
+log.info("So should this")
+log.warning("And this, too")
+log.error("And non-ASCII stuff, too, like Øresund and Malmö")
+log.error("[bold red blink]Server is shutting down!", extra={"markup": True})
+log.critical("Critical error! [b red]Server is shutting down!", extra={"markup": True})
+try:
+    print(1 / 0)
+except Exception:
+    log.exception("[b magenta]unable print![/] [i magenta]asdf", extra={"markup": True})

src/ephys_link/__about__.py

@@ -1 +1 @@
-__version__ = "2.0.0b0"
+__version__ = "2.0.0b2"

src/ephys_link/back_end/platform_handler.py

@@ -71,7 +71,7 @@ def _match_platform_type(self, platform_type: str) -> BaseBindings:
                 return FakeBindings()
             case _:
                 error_message = f'Platform type "{platform_type}" not recognized.'
-                self._console.labeled_error_print("PLATFORM", error_message)
+                self._console.critical_print(error_message)
                 raise ValueError(error_message)
 
     # Ephys Link metadata.
@@ -185,7 +185,7 @@ async def set_position(self, request: SetPositionRequest) -> PositionalResponse:
             # Disallow setting manipulator position while inside the brain.
             if request.manipulator_id in self._inside_brain:
                 error_message = 'Can not move manipulator while inside the brain. Set the depth ("set_depth") instead.'
-                self._console.error_print(error_message)
+                self._console.error_print("Set Position", error_message)
                 return PositionalResponse(error=error_message)
 
             # Move to the new position.
@@ -209,7 +209,7 @@ async def set_position(self, request: SetPositionRequest) -> PositionalResponse:
                         f" position on axis {list(Vector4.model_fields.keys())[index]}."
                         f"Requested: {request.position}, got: {final_unified_position}."
                     )
-                    self._console.error_print(error_message)
+                    self._console.error_print("Set Position", error_message)
                     return PositionalResponse(error=error_message)
         except Exception as e:
             self._console.exception_error_print("Set Position", e)

src/ephys_link/back_end/server.py

@@ -75,7 +75,7 @@ async def connect_proxy() -> None:
     # Helper functions.
     def _malformed_request_response(self, request: str, data: tuple[tuple[Any], ...]) -> str:
         """Return a response for a malformed request."""
-        self._console.labeled_error_print("MALFORMED REQUEST", f"{request}: {data}")
+        self._console.error_print("MALFORMED REQUEST", f"{request}: {data}")
         return dumps({"error": "Malformed request."})
 
     async def _run_if_data_available(
@@ -127,7 +127,9 @@ async def connect(self, sid: str, _: str) -> bool:
             self._console.info_print("CONNECTION GRANTED", sid)
             return True
 
-        self._console.error_print(f"CONNECTION REFUSED to {sid}. Client {self._client_sid} already connected.")
+        self._console.error_print(
+            "CONNECTION REFUSED", f"Cannot connect {sid} as {self._client_sid} is already connected."
+        )
         return False
 
     async def disconnect(self, sid: str) -> None:
@@ -142,7 +144,7 @@ async def disconnect(self, sid: str) -> None:
         if self._client_sid == sid:
             self._client_sid = ""
         else:
-            self._console.error_print(f"Client {sid} disconnected without being connected.")
+            self._console.error_print("DISCONNECTION", f"Client {sid} disconnected without being connected.")
 
     # noinspection PyTypeChecker
     async def platform_event_handler(self, event: str, *args: tuple[Any]) -> str:
@@ -196,5 +198,5 @@ async def platform_event_handler(self, event: str, *args: tuple[Any]) -> str:
             case "stop_all":
                 return await self._platform_handler.stop_all()
             case _:
-                self._console.error_print(f"Unknown event: {event}.")
+                self._console.error_print("EVENT", f"Unknown event: {event}.")
                 return dumps({"error": "Unknown event."})

src/ephys_link/bindings/ump_4_bindings.py

@@ -10,7 +10,6 @@
 
 from ephys_link.util.base_bindings import BaseBindings
 from ephys_link.util.common import RESOURCES_PATH, array_to_vector4, mm_to_um, mmps_to_umps, um_to_mm, vector4_to_array
-from ephys_link.util.console import Console
 
 
 class Ump4Bindings(BaseBindings):
@@ -24,7 +23,6 @@ def __init__(self) -> None:
         self._ump = UMP.get_ump()
         if self._ump is None:
             error_message = "Unable to connect to uMp"
-            Console.error_print(error_message)
             raise ValueError(error_message)
 
     async def get_manipulators(self) -> list[str]:

src/ephys_link/util/common.py

@@ -9,7 +9,6 @@
 from vbl_aquarium.models.unity import Vector4
 
 from ephys_link.__about__ import __version__
-from ephys_link.util.console import Console
 
 # Ephys Link ASCII.
 ASCII = r"""
@@ -47,8 +46,8 @@ def check_for_updates() -> None:
     response = get("https://api.github.com/repos/VirtualBrainLab/ephys-link/tags", timeout=10)
     latest_version = response.json()[0]["name"]
     if parse(latest_version) > parse(__version__):
-        Console.info_print("Update available", latest_version)
-        Console.info_print("", "Download at: https://github.com/VirtualBrainLab/ephys-link/releases/latest")
+        print(f"Update available: {latest_version} !")
+        print("Download at: https://github.com/VirtualBrainLab/ephys-link/releases/latest")
 
 
 # Unit conversions

src/ephys_link/util/console.py

@@ -6,12 +6,10 @@
 Usage: Create a Console object and call the appropriate method to print messages.
 """
 
-from traceback import print_exc
+from logging import DEBUG, ERROR, INFO, basicConfig, getLogger
 
-from colorama import Back, Fore, Style, init
-
-# Constants.
-TAB_BLOCK = "\t\t"
+from rich.logging import RichHandler
+from rich.traceback import install
 
 
 class Console:
@@ -21,34 +19,59 @@ def __init__(self, *, enable_debug: bool) -> None:
         :param enable_debug: Enable debug mode.
         :type enable_debug: bool
         """
-        self._enable_debug = enable_debug
-
         # Repeat message fields.
-        self._last_message = ""
-        self._repeat_counter = 1
+        self._last_message = (0, "", "")
+        self._repeat_counter = 0
 
-        # Initialize colorama.
-        init(autoreset=True)
+        # Config logger.
+        basicConfig(
+            level=DEBUG if enable_debug else INFO,
+            format="%(message)s",
+            datefmt="[%I:%M:%S %p]",
+            handlers=[RichHandler(rich_tracebacks=True)],
+        )
+        self._log = getLogger("rich")
 
-    @staticmethod
-    def error_print(msg: str) -> None:
-        """Print an error message to the console.
+        # Install Rich traceback.
+        install()
 
-        :param msg: Error message to print.
+    def debug_print(self, label: str, msg: str) -> None:
+        """Print a debug message to the console.
+
+        :param label: Label for the debug message.
+        :type label: str
+        :param msg: Debug message to print.
         :type msg: str
         """
-        print(f"\n{Back.RED}{Style.BRIGHT} ERROR {Style.RESET_ALL}{TAB_BLOCK}{Fore.RED}{msg}")
+        self._repeatable_log(DEBUG, f"[b green]{label}", f"[green]{msg}")
 
-    @staticmethod
-    def labeled_error_print(label: str, msg: str) -> None:
-        """Print an error message with a label to the console.
+    def info_print(self, label: str, msg: str) -> None:
+        """Print info to console.
+
+        :param label: Label for the message.
+        :type label: str
+        :param msg: Message to print.
+        :type msg: str
+        """
+        self._repeatable_log(INFO, f"[b blue]{label}", msg)
+
+    def error_print(self, label: str, msg: str) -> None:
+        """Print an error message to the console.
 
         :param label: Label for the error message.
         :type label: str
         :param msg: Error message to print.
         :type msg: str
         """
-        print(f"\n{Back.RED}{Style.BRIGHT} ERROR {label} {Style.RESET_ALL}{TAB_BLOCK}{Fore.RED}{msg}")
+        self._repeatable_log(ERROR, f"[b red]{label}", f"[red]{msg}")
+
+    def critical_print(self, msg: str) -> None:
+        """Print a critical message to the console.
+
+        :param msg: Critical message to print.
+        :type msg: str
+        """
+        self._log.critical(f"[b i red]{msg}", extra={"markup": True})
 
     @staticmethod
     def pretty_exception(exception: Exception) -> str:
@@ -61,52 +84,50 @@ def pretty_exception(exception: Exception) -> str:
         """
         return f"{type(exception).__name__}: {exception}"
 
-    @staticmethod
-    def exception_error_print(label: str, exception: Exception) -> None:
+    def exception_error_print(self, label: str, exception: Exception) -> None:
         """Print an error message with exception details to the console.
 
         :param label: Label for the error message.
         :type label: str
         :param exception: Exception to print.
         :type exception: Exception
         """
-        Console.labeled_error_print(label, Console.pretty_exception(exception))
-        print_exc()
-
-    def debug_print(self, label: str, msg: str) -> None:
-        """Print a debug message to the console.
-
-        :param label: Label for the debug message.
-        :type label: str
-        :param msg: Debug message to print.
-        :type msg: str
-        """
-        if self._enable_debug:
-            self._repeat_print(f"{Back.BLUE}{Style.BRIGHT} DEBUG {label} {Style.RESET_ALL}{TAB_BLOCK}{Fore.BLUE}{msg}")
+        self._log.exception(
+            f"[b magenta]{label}:[/] [magenta]{Console.pretty_exception(exception)}", extra={"markup": True}
+        )
 
-    @staticmethod
-    def info_print(label: str, msg: str) -> None:
-        """Print info to console.
+    # Helper methods.
+    def _repeatable_log(self, log_type: int, label: str, message: str) -> None:
+        """Add a row to the output table.
 
+        :param log_type: Type of log.
+        :type log_type: int
         :param label: Label for the message.
         :type label: str
-        :param msg: Message to print.
-        :type msg: str
+        :param message: Message.
+        :type message: str
         """
-        print(f"\n{Back.GREEN}{Style.BRIGHT} {label} {Style.RESET_ALL}{TAB_BLOCK}{Fore.GREEN}{msg}")
-
-    # Helper methods.
-    def _repeat_print(self, msg: str) -> None:
-        """Print a message to the console with repeat counter.
 
-        :param msg: Message to print.
-        :type msg: str
-        """
-        if msg == self._last_message:
+        # Compute if this is a repeated message.
+        message_set = (log_type, label, message)
+        if message_set == self._last_message:
+            # Handle repeat.
             self._repeat_counter += 1
-        else:
-            self._repeat_counter = 1
-            self._last_message = msg
-            print()
 
-        print(f"\r{msg}{f" (x{self._repeat_counter})" if self._repeat_counter > 1 else ""}", end="")
+            # Add an ellipsis row for first repeat.
+            if self._repeat_counter == 1:
+                self._log.log(log_type, "...")
+        else:
+            # Handle novel message.
+            if self._repeat_counter > 0:
+                # Complete previous repeat.
+                self._log.log(
+                    self._last_message[0],
+                    f"{self._last_message[1]}:[/] {self._last_message[2]}[/] x {self._repeat_counter}",
+                    extra={"markup": True},
+                )
+                self._repeat_counter = 0
+
+            # Log new message.
+            self._log.log(log_type, f"{label}:[/] {message}", extra={"markup": True})
+            self._last_message = message_set