justin
6bee9c0d7f
Foundation for the PySide6 + pyqtgraph Windows GUI, shared with the terminal
tool. Pure data/IO -- no Qt, no curses.
obdcore/
link.py ElmLink -- ELM327 serial (Mode-01/22, ATRV, DTC read/clear)
mock.py MockLink -- synthetic crank for tests + GUI dev (no truck)
registry.py PidRegistry (verified Ford 6.0 PIDs + confidence) + DtcDatabase
scheduler.py PollScheduler -- prioritized round-robin polling, dead-PID park,
derived channels; tick() is fake-clock test-drivable
store.py TimeSeriesStore (ring buffers + min/max) + CsvRecorder/replay
Design centers on the ELM327 bandwidth limit (~7-15 reads/sec): the active
view subscribes PIDs at chosen rates; acquisition runs off the UI thread;
the GUI only reads the store. FICM_M (09D0) promoted to verified after the
2026-06-30 on-truck crank read (48.0V, intermittent).
tests/test_obdcore.py: decoders vs real truck bytes, crank ramp + peak,
derived BOOST, dead-PID park/revive, record/replay roundtrip -- all pass.
ARCHITECTURE.md: layers, data model, GUI plan, 6.0 stock-PID limits
(no EGT/oil-PSI), feature backlog, P0-P5 roadmap.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016yT89n4zR4qbrySoSiEyZs
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""Hardware-free tests for the obdcore acquisition engine.
|
|
|
|
Drives the scheduler deterministically with a fake clock + MockLink, so the
|
|
prioritized polling, derived channels, dead-PID parking, store min/max, and
|
|
record/replay are all validated without a truck or serial port.
|
|
|
|
Run: python -m pytest tests/ -q (or) python tests/test_obdcore.py
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from obdcore import PidRegistry, TimeSeriesStore, PollScheduler, CsvRecorder, replay_csv
|
|
from obdcore.mock import MockLink
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self):
|
|
self.t = 0.0
|
|
|
|
def __call__(self):
|
|
return self.t
|
|
|
|
def advance(self, dt):
|
|
self.t += dt
|
|
|
|
|
|
def _setup(specs):
|
|
clk = FakeClock()
|
|
reg = PidRegistry()
|
|
store = TimeSeriesStore()
|
|
link = MockLink(clock=clk)
|
|
sch = PollScheduler(link, reg, store, clock=clk)
|
|
sch.set_subscriptions(specs)
|
|
return clk, reg, store, sch
|
|
|
|
|
|
def test_registry_decoders_match_truck_bytes():
|
|
reg = PidRegistry()
|
|
cases = {
|
|
"ICP": ([0x00, 0x16], 12.5), "EBP": ([0x01, 0x8F], 14.46),
|
|
"MAP": ([0x01, 0x89], 14.25), "BARO": ([0x01, 0x88], 14.21),
|
|
"EOT": ([0x1C, 0x92], 33.1), "GEAR": ([0x02], 1),
|
|
"FICM_M": ([0x30, 0x00], 48.0),
|
|
}
|
|
for key, (raw, expect) in cases.items():
|
|
got = reg.get(key).decode(raw)
|
|
assert abs(got - expect) < 0.05, f"{key}: {got} != {expect}"
|
|
print(" decoders match truck bytes: OK")
|
|
|
|
|
|
def test_crank_ramp_and_peak():
|
|
clk, reg, store, sch = _setup([("ICP", 5), ("FICM_M", 2), ("BATT", 2), ("RPM", 2)])
|
|
for _ in range(60): # ~3s at 50ms steps
|
|
sch.tick()
|
|
clk.advance(0.05)
|
|
icp = store.channel("ICP")
|
|
lo, hi = store.minmax("ICP")
|
|
assert hi >= 500, f"ICP should ramp past 500, got peak {hi}"
|
|
assert lo is not None and lo < 50, "ICP should start low"
|
|
assert store.latest("FICM_M") == 48.0
|
|
bat_lo, _ = store.minmax("BATT")
|
|
assert bat_lo <= 10.7, f"battery sag should be captured, got {bat_lo}"
|
|
print(f" crank ramp: ICP {lo} -> peak {hi}, FICM {store.latest('FICM_M')}V, "
|
|
f"batt min {bat_lo}V: OK")
|
|
|
|
|
|
def test_derived_boost_channel():
|
|
clk, reg, store, sch = _setup([("MAP", 5), ("BARO", 5), ("BOOST", 5)])
|
|
for _ in range(10):
|
|
sch.tick()
|
|
clk.advance(0.05)
|
|
boost = store.latest("BOOST")
|
|
assert boost is not None and abs(boost) < 0.5, f"atm boost ~0, got {boost}"
|
|
print(f" derived BOOST = MAP-BARO = {boost} psi: OK")
|
|
|
|
|
|
def test_dead_pid_parks_and_revives():
|
|
clk, reg, store, sch = _setup([("ICP", 5), ("IPR", 5)]) # IPR not in MockLink -> dead
|
|
for _ in range(20):
|
|
sch.tick()
|
|
clk.advance(0.05)
|
|
# IPR should have parked (4 fails) but scheduler still runs ICP
|
|
assert store.latest("ICP") is not None
|
|
assert store.latest("IPR") is None
|
|
# advance past revive window -> IPR re-attempted (still None, but tried)
|
|
clk.advance(6.0)
|
|
sch.tick()
|
|
print(" dead-PID parking + revive: OK")
|
|
|
|
|
|
def test_record_replay_roundtrip(tmp_path=None):
|
|
import tempfile
|
|
path = os.path.join(tempfile.gettempdir(), "obdcore_replay_test.csv")
|
|
clk, reg, store, sch = _setup([("ICP", 5), ("FICM_M", 5)])
|
|
store.recorder = CsvRecorder(path)
|
|
for _ in range(20):
|
|
sch.tick()
|
|
clk.advance(0.05)
|
|
store.recorder.close()
|
|
peak_orig = store.minmax("ICP")[1]
|
|
|
|
store2 = TimeSeriesStore()
|
|
replay_csv(path, store2)
|
|
peak_replay = store2.minmax("ICP")[1]
|
|
assert abs(peak_orig - peak_replay) < 0.01, "replay should reproduce peak ICP"
|
|
print(f" record/replay: peak ICP {peak_orig} == replay {peak_replay}: OK")
|
|
os.remove(path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
for fn in [test_registry_decoders_match_truck_bytes, test_crank_ramp_and_peak,
|
|
test_derived_boost_channel, test_dead_pid_parks_and_revives,
|
|
test_record_replay_roundtrip]:
|
|
fn()
|
|
print("\nALL obdcore TESTS PASS")
|