Allow custom monitoring scripts

client/cli.py

@@ -11,7 +11,7 @@
 from typing import List, Optional, Type
 from result import Err
 import backend_domain
-from monitor_serial import MonitorSerial
+from monitor_serial import MonitorSerial, MonitorSerialHelper
 from rich import print as richprint
 from rich_util import print_error, print_success, print_json
 from agent import AgentConfig
@@ -542,7 +542,9 @@ def hardware_serial_monitor(
         sys.exit(1)
 
     # Monitor hardware
-    monitor_result = asyncio.run(backend_config.hardware_serial_monitor(hardware_id, monitor_class()))
+    monitor_result = asyncio.run(backend_config.hardware_serial_monitor(
+        hardware_id,
+        monitor_class(MonitorSerialHelper)))
     if isinstance(monitor_result, Err):
         print()
         print_error(f"Failed to monitor hardware: {monitor_result.value}")

client/monitor_serial.py

@@ -1,12 +1,72 @@
 """Module for functionality related to serial socket monitoring"""
 
 from typing import Any
-from protocol import MonitorListenerIncomingMessage, MonitorListenerOutgoingMessage
+from protocol import \
+    MonitorListenerIncomingMessage, \
+    MonitorListenerOutgoingMessage, \
+    MonitorUnavailable, \
+    SerialMonitorMessageToClient, \
+    SerialMonitorMessageToAgent
 from ws import Socketlike
+from codec import CodecParseException
+
+
+class MonitorSerialHelperLike:
+    """Helper for managing monitor messages"""
+
+    @staticmethod
+    def isMonitorUnavailable(message: Any) -> bool:
+        """Check if message is of type 'MonitorUnavailable'"""
+        pass
+
+    @staticmethod
+    def isSerialMonitorMessageToClient(message: Any) -> bool:
+        """Check if message is of type 'SerialMonitorMessageToClient'"""
+        pass
+
+    @staticmethod
+    def isCodecParseException(instance: Any) -> bool:
+        """Check if class instance is of type 'CodecParseException'"""
+        pass
+
+    @staticmethod
+    def createSerialMonitorMessageToAgent(payload: bytes) -> Any:
+        """Create createSerialMonitorMessageToAgent from bytes"""
+        pass
+
+
+class MonitorSerialHelper(MonitorSerialHelperLike):
+    """Helper for managing monitor messages"""
+
+    @staticmethod
+    def isMonitorUnavailable(message: Any) -> bool:
+        """Check if message is of type 'MonitorUnavailable'"""
+        return isinstance(message, MonitorUnavailable)
+
+    @staticmethod
+    def isSerialMonitorMessageToClient(message: Any) -> bool:
+        """Check if message is of type 'SerialMonitorMessageToClient'"""
+        return isinstance(message, SerialMonitorMessageToClient)
+
+    @staticmethod
+    def isCodecParseException(instance: Any) -> bool:
+        """Check if class instance is of type 'CodecParseException'"""
+        return isinstance(instance, CodecParseException)
+
+    @staticmethod
+    def createSerialMonitorMessageToAgent(payload: bytes) -> Any:
+        """Create createSerialMonitorMessageToAgent from bytes"""
+        return SerialMonitorMessageToAgent.from_bytes(payload)
 
 
 class MonitorSerial:
     """Interface for serial socket monitors"""
+
+    helper: MonitorSerialHelperLike
+
+    def __init__(self, helper):
+        self.helper = helper
+
     async def run(
         self,
         socketlike: Socketlike[Any, MonitorListenerIncomingMessage, MonitorListenerOutgoingMessage]

client/monitor_serial_hexbytes.py

@@ -25,52 +25,6 @@
 
 LOGGER = log.timed_named_logger("monitor_serial")
 
-# Mocks for DIP client classes for python type hinting
-# N.B This whole section can be removed if you don't care about type checking
-#   and if you're just editing in Notepad and doing it trial-and-error style
-
-# class Socketlike(Generic[SERIALIZABLE, PI, PO]):
-#     """Interface for interactions w/ sockets"""
-#     async def connect(self) -> Optional[Exception]:
-#         pass
-#
-#     async def disconnect(self) -> Optional[Exception]:
-#         pass
-#
-#     async def rx(self) -> Result[PI, Exception]:
-#         pass
-#
-#     async def tx(self, data: PO) -> Optional[Exception]:
-#         pass
-#
-# @dataclass(frozen=True, eq=False)
-# class MonitorUnavailable:
-#     """Message regarding hardware monitor unavailability"""
-#     reason: str
-#
-# @dataclass(frozen=True, eq=False)
-# class SerialMonitorMessageToClient:
-#     """Message from hardware serial monitor to client"""
-#     base64Bytes: str
-#
-#     @staticmethod
-#     def from_bytes(content: bytes):
-#         """Construct message from bytes"""
-#         return SerialMonitorMessageToClient(base64.b64encode(content).decode("utf-8"))
-#
-#     def to_bytes(self) -> bytes:
-#         """Construct bytes from message"""
-#         return base64.b64decode(self.base64Bytes)
-#
-# class MonitorSerial:
-#     """"""
-#     async def run(
-#         self,
-#         socketike: Socketlike[Any, protocol.MonitorListenerIncomingMessage, protocol.MonitorListenerOutgoingMessage]
-#     ):
-#         """Receive serial monitor websocket messages & implement user interfacing"""
-#         pass
-
 
 # Actual monitor implementation
 class MonitorSerialHexbytes(MonitorSerial):
@@ -173,7 +127,7 @@ async def run(
         # Redirect stdin to serial monitor socket
         asyncio_loop = asyncio.get_event_loop()
         stdin_capture_task = asyncio_loop.create_task(
-            MonitorSerialHexbytes.keep_transmitting_to_agent(socketlike))
+            self.keep_transmitting_to_agent(socketlike))
 
         # Define end-of-monitor handler
         death = Death()
@@ -191,7 +145,7 @@ async def run(
         # Run monitoring loop
         while True:
             incoming_message_result = await socketlike.rx()
-            result = MonitorSerialHexbytes.render_message_data_or_finish(death, handle_finish, incoming_message_result)
+            result = self.render_message_data_or_finish(death, handle_finish, incoming_message_result)
             if result is not None:
                 return result
 

examples/monitor_serial_hexbytes/Pipfile

@@ -0,0 +1,12 @@
+[[source]]
+url = "https://pypi.org/simple"
+verify_ssl = true
+name = "pypi"
+
+[packages]
+websockets = "*"
+mypy = "*"
+result = "*"
+
+[requires]
+python_version = "3.9"