bugfixes

software/python-package/shepherd_sheep/commons.py

@@ -33,6 +33,10 @@
 MSG_DBG_FN_TESTS = 0xAF
 MSG_DBG_VSRC_HRV_P_INP = 0xB1
 
+# NOTE: below messages were previously exclusive to kernel space
+
+MSG_STATUS_RESTARTING_ROUTINE = 0xC0
+
 # TODO: these 9 lines below are replaced by the following dict
 MSG_ERROR = 0xE0
 MSG_ERR_MEMCORRUPTION = 0xE1
@@ -44,6 +48,9 @@
 MSG_ERR_SYNC_STATE_NOT_IDLE = 0xE7
 MSG_ERR_VALUE = 0xE8
 
+MSG_TEST = 0xEA
+MSG_SYNC = 0xEB
+
 pru_errors: dict[int, str] = {
     0xE0: "General (unspecified) PRU-error [MSG_ERROR]",
     0xE1: "PRU received a faulty msg.id from kernel [MSG_ERR_MEMCORRUPTION]",

software/python-package/shepherd_sheep/shepherd_io.py

@@ -8,7 +8,6 @@
 import time
 from contextlib import suppress
 from types import TracebackType
-from typing import Union
 
 from pydantic import validate_call
 from shepherd_core import CalibrationEmulator
@@ -46,8 +45,8 @@
 class ShepherdIOError(Exception):
     ID_TIMEOUT = 9999
 
-    def __init__(self, message: str, id_num: int = 0, value: Union[int, list, None] = 0) -> None:
-        super().__init__(message + f" [id=0x{id_num:x}, val={value}]")
+    def __init__(self, message: str, id_num: int = 0, value: int | list | None = 0) -> None:
+        super().__init__(message + f" [id=0x{id_num:X}, val={value}]")
         self.id_num = id_num
         self.value = value
 
@@ -131,11 +130,11 @@ def __enter__(self) -> Self:
             log.debug("Switching to '%s'-mode", self.mode)
             sfs.write_mode(self.mode)
 
+            self.refresh_shared_mem()
+
             # clean up msg-channel provided by kernel module
             self._flush_msgs()
 
-            self.refresh_shared_mem()
-
         except Exception:
             log.exception("ShepherdIO.Init caught an exception -> exit now")
             self._power_down_shp()
@@ -539,6 +538,18 @@ def get_buffer(
                 log.info("Received cmd to print: %d", value)
                 continue
 
+            if msg_type == commons.MSG_TEST:
+                log.debug("Received test-message from PRU: %d", value)
+                continue
+
+            if msg_type == commons.MSG_SYNC:
+                log.debug("Received sync-message from PRU: %d", value)
+                continue
+
+            if msg_type == commons.MSG_STATUS_RESTARTING_ROUTINE:
+                log.debug("PRU is restarting its main routine, val=%d", value)
+                continue
+
             error_msg: str | None = commons.pru_errors.get(msg_type)
             if error_msg is not None:
                 raise ShepherdIOError(error_msg, msg_type, value)