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

Issue:
Add a listener anchor and a tag
Listeners (anchors in _passive_ mode) and tags are the two most used DWM1001 types, so the library should support these easily. Our current use-case for these devices is getting position information.

In the listener's case, we want to get all reported positions from tags. In the tag's case, we want to get its own position.


Discussion / hints:

Fix patch:
diff --git a/dwm1001.py b/dwm1001.py
new file mode 100644
index 0000000..be70f72
--- /dev/null
+++ b/dwm1001.py
@@ -0,0 +1,72 @@
+# Standard library imports
+from dataclasses import dataclass
+from enum import Enum
+import math
+import time
+
+# Third party imports
+from serial import Serial
+
+
+class TagId(str):
+    pass
+
+
+@dataclass
+class TagPosition:
+    x_m: float
+    y_m: float
+    z_m: float
+    quality: int
+
+    def almost_equal(self, other: "TagPosition") -> bool:
+        return (
+            math.isclose(self.x_m, other.x_m)
+            and math.isclose(self.y_m, other.y_m)
+            and math.isclose(self.z_m, other.z_m)
+            and self.quality == other.quality
+        )
+
+
+class ShellCommand(Enum):
+    DOUBLE_ENTER = b"\r\r"
+    LEP = b"lep"
+
+
+class Listener:
+    def __init__(self, serial_handle: Serial) -> None:
+        self.serial_handle = serial_handle
+
+        self.serial_handle.write(ShellCommand.DOUBLE_ENTER.value)
+        time.sleep(1)
+        self.serial_handle.write(ShellCommand.LEP.value)
+
+    def wait_for_position_report(self) -> tuple[TagId, TagPosition]:
+        report = self.serial_handle.readline().decode().split(",")
+
+        if len(report) != 8 or report[0] != "POS":
+            raise ValueError("Could not parse position report.")
+
+        position_data = [float(substr) for substr in report[3:7]]
+
+        return TagId(report[2]), TagPosition(*position_data)
+
+
+class Tag:
+    def __init__(self, serial_handle) -> None:
+        self.serial_handle = serial_handle
+
+        self.serial_handle.write(ShellCommand.DOUBLE_ENTER.value)
+        time.sleep(1)
+        self.serial_handle.write(ShellCommand.LEP)
+
+    @property
+    def position(self) -> TagPosition:
+        report = self.serial_handle.readline().decode().split(",")
+
+        if len(report) != 5 or report[0] != "POS":
+            raise ValueError("Could not parse position report.")
+
+        position_data = [float(substr) for substr in report[1:]]
+
+        return TagPosition(*position_data)
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..bcad478
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,22 @@
+[project]
+name = "pydwm1001"
+version = "0.1.0"
+authors = [
+  { name="Adam Morrissett", email="morrissettal2@vcu.edu" },
+]
+description = "A Python library for interfacing with DWM1001"
+readme = "README.md"
+requires-python = ">=3.7"
+classifiers = [
+    "Programming Language :: Python :: 3",
+    "License :: OSI Approved :: MIT License",
+    "Operating System :: OS Independent",
+]
+
+[project.urls]
+"Homepage" = "https://github.com/the-hive-lab/pydwm1001"
+"Bug Tracker" = "https://github.com/the-hive-lab/pydwm1001/issues"
+
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..2647244
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,2 @@
+build
+pyserial


Test patch:
diff --git a/tests/test_dwm1001.py b/tests/test_dwm1001.py
new file mode 100644
index 0000000..7153597
--- /dev/null
+++ b/tests/test_dwm1001.py
@@ -0,0 +1,68 @@
+# Standard library imports
+import unittest
+
+# Third party imports
+import dwm1001
+
+
+class MockSerial:
+    def __init__(self, response: bytes = b"") -> None:
+        self.response = response
+        self.write_history = []
+
+    def readline(self) -> bytes:
+        return self.response
+
+    def write(self, value: bytes) -> None:
+        self.write_history.append(value)
+
+
+class TestTagPosition(unittest.TestCase):
+    def test_construction(self) -> None:
+        position = dwm1001.TagPosition(1.23, 4.56, 7.89, 42)
+
+        self.assertAlmostEqual(position.x_m, 1.23)
+        self.assertAlmostEqual(position.y_m, 4.56)
+        self.assertAlmostEqual(position.z_m, 7.89)
+        self.assertEqual(position.quality, 42)
+
+    def test_almost_equal(self) -> None:
+        position1 = dwm1001.TagPosition(1.23, 4.56, 7.89, 42)
+        position2 = dwm1001.TagPosition(1.23, 4.56, 7.89, 42)
+
+        self.assertTrue(position1.almost_equal(position2))
+
+
+class TestListener(unittest.TestCase):
+    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")
+
+    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)
+
+        tag_id, tag_position = listener.wait_for_position_report()
+        self.assertEqual(tag_id, "TEST1")
+
+        expected_position = dwm1001.TagPosition(1.23, 4.56, 7.89, 20)
+        self.assertTrue(tag_position.almost_equal(expected_position))
+
+    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)
+
+        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)
+
+        self.assertRaises(ValueError, listener.wait_for_position_report)
+
+
+if __name__ == "__main__":
+    unittest.main()