Repository: the-hive-lab/pydwm1001 (language: python)

Issue:
Make Listener class more robust to actual device
The current `Listener` implementation works fine for a mocked DWM1001, but it does not work with a real one due to timing issues. This class should be made more robust.


Discussion / hints:

Fix patch:
diff --git a/dwm1001.py b/dwm1001.py
index be70f72..b44dc2b 100644
--- a/dwm1001.py
+++ b/dwm1001.py
@@ -29,17 +29,65 @@ def almost_equal(self, other: "TagPosition") -> bool:
 
 
 class ShellCommand(Enum):
+    ENTER = b"\r"
     DOUBLE_ENTER = b"\r\r"
     LEP = b"lep"
+    RESET = b"reset"
 
 
-class Listener:
+class UartDwm1001:
+    # These delay periods were experimentally determined
+    RESET_DELAY_PERIOD = 0.1
+    SHELL_STARTUP_DELAY_PERIOD = 1.0
+    SHELL_COMMAND_DELAY_PERIOD = 0.5
+
     def __init__(self, serial_handle: Serial) -> None:
         self.serial_handle = serial_handle
 
+    def send_shell_command(self, command: ShellCommand) -> None:
+        self.serial_handle.write(command.value)
+        self.serial_handle.write(ShellCommand.ENTER.value)
+        self.serial_handle.flush()
+
+        time.sleep(self.SHELL_COMMAND_DELAY_PERIOD)
+
+    def reset(self) -> None:
+        self.send_shell_command(ShellCommand.RESET)
+
+        time.sleep(self.RESET_DELAY_PERIOD)
+
+    def enter_shell_mode(self) -> None:
         self.serial_handle.write(ShellCommand.DOUBLE_ENTER.value)
-        time.sleep(1)
-        self.serial_handle.write(ShellCommand.LEP.value)
+        self.serial_handle.flush()
+
+        time.sleep(self.SHELL_STARTUP_DELAY_PERIOD)
+
+    def exit_shell_mode(self) -> None:
+        # If you quit shell mode (with `quit` command) without stopping
+        # a running command, the command will automatically continue
+        # when re-entering shell mode. This can be confusing, so we
+        # reset the device instead to ensure previously-running commands
+        # terminate.
+        self.reset()
+
+
+class Listener(UartDwm1001):
+    def __init__(self, serial_handle: Serial) -> None:
+        super().__init__(serial_handle)
+
+        # Device may not have shutdown properly previously
+        self.reset()
+        self.enter_shell_mode()
+
+    def start_position_reporting(self) -> None:
+        self.send_shell_command(ShellCommand.LEP)
+
+        # The first line after invoking the command will have part of
+        # the shell prompt mixed in, which would mess up parsing.
+        self.serial_handle.reset_input_buffer()
+
+    def stop_position_reporting(self) -> None:
+        self.send_shell_command(ShellCommand.ENTER)
 
     def wait_for_position_report(self) -> tuple[TagId, TagPosition]:
         report = self.serial_handle.readline().decode().split(",")


Test patch:
diff --git a/tests/test_dwm1001.py b/tests/test_dwm1001.py
index 7153597..9df6e80 100644
--- a/tests/test_dwm1001.py
+++ b/tests/test_dwm1001.py
@@ -9,13 +9,28 @@ class MockSerial:
     def __init__(self, response: bytes = b"") -> None:
         self.response = response
         self.write_history = []
+        self.should_return_response = False
 
     def readline(self) -> bytes:
-        return self.response
+        if self.should_return_response:
+            return self.response
+
+        return b""
 
     def write(self, value: bytes) -> None:
         self.write_history.append(value)
 
+        if value == b"lep":
+            self.should_return_response = True
+
+    @staticmethod
+    def flush() -> None:
+        pass
+
+    @staticmethod
+    def reset_input_buffer():
+        pass
+
 
 class TestTagPosition(unittest.TestCase):
     def test_construction(self) -> None:
@@ -38,12 +53,14 @@ def test_construction(self) -> None:
         serial_handle = MockSerial()
         dwm1001.Listener(serial_handle)
 
-        self.assertEqual(serial_handle.write_history[0], b"\r\r")
-        self.assertEqual(serial_handle.write_history[1], b"lep")
+        self.assertEqual(serial_handle.write_history[0], b"reset")
+        self.assertEqual(serial_handle.write_history[1], b"\r")
+        self.assertEqual(serial_handle.write_history[2], b"\r\r")
 
     def test_wait_for_position_report_good(self) -> None:
         serial_handle = MockSerial(b"POS,1,TEST1,1.23,4.56,7.89,20,86")
         listener = dwm1001.Listener(serial_handle)
+        listener.start_position_reporting()
 
         tag_id, tag_position = listener.wait_for_position_report()
         self.assertEqual(tag_id, "TEST1")
@@ -54,12 +71,14 @@ def test_wait_for_position_report_good(self) -> None:
     def test_wait_for_position_report_short(self) -> None:
         serial_handle = MockSerial(b"POS,1,TEST1,1.23,4.56,7.89")
         listener = dwm1001.Listener(serial_handle)
+        listener.start_position_reporting()
 
         self.assertRaises(ValueError, listener.wait_for_position_report)
 
     def test_wait_for_position_report_wrong_header(self) -> None:
         serial_handle = MockSerial(b"BAD,1,TEST1,1.23,4.56,7.89")
         listener = dwm1001.Listener(serial_handle)
+        listener.start_position_reporting()
 
         self.assertRaises(ValueError, listener.wait_for_position_report)