elm327: leave a trace when we swallow an exception (#20)

carwatch/elm327.py

@@ -22,6 +22,7 @@
 from __future__ import annotations
 
 import os
+import sys
 import time
 
 
@@ -39,6 +40,30 @@ def _first_present_port() -> str:
     return "/dev/ttyUSB0"
 
 
+# Swallowed-exception trace (#20). These handlers deliberately continue - an
+# OBD dongle vanishing mid-drive is normal, not an error - but with no record
+# at all a decoder bug and a missing dongle look identical in the field, and
+# a dropped reading just shows as a blank tile.
+#
+# Deduped on purpose: "as a rule" means a bare log would flood the journal on
+# exactly the drive you are trying to diagnose. First occurrence prints, then
+# only at 10 / 100 / 1000. stderr, never stdout: this module's CLI prints JSON
+# and a stray line would corrupt it.
+_SWALLOWED: dict = {}
+
+
+def _swallowed(site: str, exc: BaseException) -> None:
+    key = (site, type(exc).__name__)
+    n = _SWALLOWED.get(key, 0) + 1
+    _SWALLOWED[key] = n
+    if n == 1:
+        print(f"elm327: {site}: {type(exc).__name__}: {exc}",
+              file=sys.stderr, flush=True)
+    elif n in (10, 100, 1000):
+        print(f"elm327: {site}: {type(exc).__name__} x{n}",
+              file=sys.stderr, flush=True)
+
+
 DEFAULT_PORT = _first_present_port()
 DEFAULT_BAUD = 38400
 
@@ -100,8 +125,8 @@ def __init__(self, port: str = DEFAULT_PORT, baud: int = DEFAULT_BAUD):
     def close(self):
         try:
             os.close(self.fd)
-        except Exception:
-            pass
+        except Exception as e:
+            _swallowed("close", e)
 
     def _read_until_prompt(self, timeout: float = 5.0) -> str:
         """ELM327 ends every response with '>'. Read until it, or timeout."""
@@ -156,7 +181,10 @@ def _parse_pid_reply(text: str, pid: int):
                 return None
             try:
                 return name, dec(data)
-            except Exception:
+            except Exception as e:
+                # A decoder raising means a reading silently disappears from
+                # the dash. Worth a line even though we carry on.
+                _swallowed(f"decode pid 0x{pid:02X}", e)
                 return None
     return None
 
@@ -251,7 +279,8 @@ def _parse_ext_reply(text: str, pid: int):
                 return {"key": key, "label": label, "unit": unit,
                         "group": group, "pid": f"0x{pid:02X}",
                         "value": dec(data)}
-            except Exception:
+            except Exception as e:
+                _swallowed(f"decode ext pid 0x{pid:02X}", e)
                 return None
     return None
 
@@ -264,7 +293,10 @@ def read_all_extended(elm: Elm327, pids=None) -> dict:
     if pids is None:
         try:
             supported = set(scan_supported_quiet(elm))
-        except Exception:
+        except Exception as e:
+            # Falling back to the full PID list hides that the capability
+            # scan failed; the sweep then looks merely unlucky.
+            _swallowed("scan_supported_quiet", e)
             supported = set()
         pids = [p for p in EXT_PIDS if p in supported] or list(PIDS)
     groups: dict = {}

tests/test_elm327_traces.py

@@ -0,0 +1,77 @@
+"""#20: elm327.py swallowed exceptions with no record at all.
+
+hermes's review, 31 Aug. An OBD dongle vanishing mid-drive is normal, so the
+handlers are right to continue - but with nothing logged, a decoder bug and a
+missing dongle look identical in the field, and a dropped reading shows only
+as a blank tile on the dash.
+"""
+import io
+import os
+import sys
+import unittest
+from contextlib import redirect_stderr, redirect_stdout
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from carwatch import elm327  # noqa: E402
+
+
+class SwallowedTrace(unittest.TestCase):
+
+    def setUp(self):
+        elm327._SWALLOWED.clear()
+
+    def test_first_occurrence_is_reported(self):
+        err = io.StringIO()
+        with redirect_stderr(err):
+            elm327._swallowed("decode pid 0x0C", ValueError("bad byte"))
+        out = err.getvalue()
+        self.assertIn("decode pid 0x0C", out)
+        self.assertIn("ValueError", out)
+        self.assertIn("bad byte", out)
+
+    def test_repeats_do_not_flood_the_journal(self):
+        # A dongle that vanishes "as a rule" must not drown the drive it is
+        # being diagnosed on.
+        err = io.StringIO()
+        with redirect_stderr(err):
+            for _ in range(9):
+                elm327._swallowed("close", OSError("gone"))
+        self.assertEqual(err.getvalue().count("\n"), 1,
+                         "repeats should be summarised, not printed each time")
+
+    def test_milestones_still_surface(self):
+        err = io.StringIO()
+        with redirect_stderr(err):
+            for _ in range(10):
+                elm327._swallowed("close", OSError("gone"))
+        self.assertIn("x10", err.getvalue())
+
+    def test_distinct_sites_and_types_are_tracked_separately(self):
+        err = io.StringIO()
+        with redirect_stderr(err):
+            elm327._swallowed("close", OSError("a"))
+            elm327._swallowed("close", ValueError("b"))
+            elm327._swallowed("scan_supported_quiet", OSError("c"))
+        self.assertEqual(err.getvalue().count("\n"), 3)
+
+    def test_never_writes_to_stdout(self):
+        # This module's CLI prints JSON; a stray line would corrupt it.
+        out, err = io.StringIO(), io.StringIO()
+        with redirect_stdout(out), redirect_stderr(err):
+            elm327._swallowed("close", OSError("gone"))
+        self.assertEqual(out.getvalue(), "")
+        self.assertTrue(err.getvalue())
+
+    def test_no_bare_swallow_remains_at_the_four_sites(self):
+        src = open(os.path.join(os.path.dirname(os.path.dirname(
+            os.path.abspath(__file__))), "carwatch", "elm327.py")).read()
+        for marker in ("_swallowed(\"close\"",
+                       "_swallowed(f\"decode pid",
+                       "_swallowed(f\"decode ext pid",
+                       "_swallowed(\"scan_supported_quiet\""):
+            self.assertIn(marker, src, f"{marker} lost its trace")
+
+
+if __name__ == "__main__":
+    unittest.main()