Merge pull request #5 from michealroberts/feature/handlers/ReadTimeoutHandler

feat: add ReadTimeoutHandler to handlers in samps module

Author: michealroberts

src/samps/handlers.py

@@ -0,0 +1,95 @@
+# **************************************************************************************
+
+# @package        samps
+# @license        MIT License Copyright (c) 2025 Michael J. Roberts
+
+# **************************************************************************************
+
+
+from time import monotonic
+from typing import Optional
+
+# **************************************************************************************
+
+
+class ReadTimeoutHandler:
+    """
+    A handler for managing read timeouts in serial communication.
+
+    Tracks a timeout period in milliseconds, provides methods to check
+    whether the timeout has expired, retrieve the remaining time,
+    start or reset the timer, and a repr string.
+    """
+
+    def __init__(self, timeout: Optional[float]) -> None:
+        """
+        Initialize with a timeout value in milliseconds.
+
+        Args:
+            timeout: timeout duration in milliseconds, or None to disable.
+        """
+        self._timeout = timeout
+
+    def start(self) -> None:
+        """
+        Restart the timeout countdown from the current moment.
+        """
+        self._start = monotonic()
+
+    def has_expired(self) -> bool:
+        """
+        Return True if the elapsed time since start() has reached
+        or exceeded the timeout value.
+        """
+        # If no timeout is set, return None:
+        if self._timeout is None:
+            return False
+
+        # If the timer hasn't started, we can't calculate remaining time:
+        if self._start is None:
+            raise RuntimeError("Timeout not started. Call start() first.")
+
+        return ((monotonic() - self._start) * 1000) >= self._timeout
+
+    def remaining(self) -> Optional[float]:
+        """
+        Return the number of milliseconds left before expiration,
+        never negative.
+
+        Returns:
+            The remaining time in milliseconds, or None if timeouts are disabled.
+        """
+        # If no timeout is set, return None:
+        if self._timeout is None:
+            return None
+
+        # If the timer hasn't started, we can't calculate remaining time:
+        if self._start is None:
+            raise RuntimeError("Timeout not started. Call start() first.")
+
+        remaining = self._timeout - (monotonic() - self._start) * 1000
+        return remaining if remaining > 0 else 0.0
+
+    def reset(self) -> None:
+        """
+        Alias for start(): restart the timeout countdown.
+        """
+        self.start()
+
+    def __repr__(self) -> str:
+        # If the timeout is None, we can't calculate remaining time:
+        if self._timeout is None:
+            return "<ReadTimeoutHandler timeout=None>"
+
+        # If the timer hasn't started, we can't calculate remaining time:
+        if self._start is None:
+            return f"<ReadTimeoutHandler timeout={self._timeout:.0f}ms, remaining=NotStarted>"
+
+        # Otherwise, return back the remaining time:
+        return (
+            f"<ReadTimeoutHandler timeout={self._timeout:.0f}ms, "
+            f"remaining={self.remaining():.0f}ms>"
+        )
+
+
+# **************************************************************************************

test/test_handlers.py

@@ -0,0 +1,58 @@
+# **************************************************************************************
+
+# @package        samps
+# @license        MIT License Copyright (c) 2025 Michael J. Roberts
+
+# **************************************************************************************
+
+import unittest
+from time import sleep
+
+from samps.handlers import ReadTimeoutHandler
+
+# **************************************************************************************
+
+
+class TestReadTimeoutHandler(unittest.TestCase):
+    def test_no_timeout(self):
+        handler = ReadTimeoutHandler(None)
+        handler.start()
+        self.assertFalse(handler.has_expired())
+        self.assertIsNone(handler.remaining())
+        self.assertIn("timeout=None", repr(handler))
+
+    def test_timeout_not_expired(self):
+        handler = ReadTimeoutHandler(2.0)
+        handler.start()
+        self.assertFalse(handler.has_expired())
+        remaining = handler.remaining()
+        self.assertTrue(0.0 < remaining <= 2.0)
+
+    def test_timeout_expired(self):
+        handler = ReadTimeoutHandler(0.5)
+        handler.start()
+        sleep(0.6)
+        self.assertTrue(handler.has_expired())
+        self.assertEqual(handler.remaining(), 0.0)
+
+    def test_reset(self):
+        handler = ReadTimeoutHandler(0.1)
+        handler.start()
+        sleep(0.2)
+        self.assertTrue(handler.has_expired())
+        handler.reset()
+        self.assertFalse(handler.has_expired())
+        self.assertTrue(0.0 < handler.remaining() <= 0.1)
+
+    def test_start(self):
+        handler = ReadTimeoutHandler(0.1)
+        handler.start()
+        self.assertFalse(handler.has_expired())
+
+
+# **************************************************************************************
+
+if __name__ == "__main__":
+    unittest.main()
+
+# **************************************************************************************