diff --git a/carwatch/elm327.py b/carwatch/elm327.py index 91884b0..a362288 100644 --- a/carwatch/elm327.py +++ b/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 = {} diff --git a/tests/test_elm327_traces.py b/tests/test_elm327_traces.py new file mode 100644 index 0000000..e5c4159 --- /dev/null +++ b/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()