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
127 lines
3.6 KiB
Python
127 lines
3.6 KiB
Python
"""Time-series store: per-channel ring buffers + min/max, with an optional
|
|
recorder for full-session capture/playback.
|
|
|
|
The acquisition thread pushes samples here; the GUI (or terminal) only reads.
|
|
This decoupling is what keeps the UI smooth while the ELM327 plods along at
|
|
~7-15 reads/sec. Nothing here touches serial or Qt -- pure data.
|
|
"""
|
|
import threading
|
|
from collections import deque
|
|
|
|
|
|
class Channel:
|
|
"""One PID's rolling history plus session min/max."""
|
|
|
|
def __init__(self, key, maxlen=3600):
|
|
self.key = key
|
|
self.buf = deque(maxlen=maxlen) # (t, value); value may be None
|
|
self.lo = None
|
|
self.hi = None
|
|
self.last_t = None
|
|
self.last_v = None
|
|
|
|
def push(self, t, v):
|
|
self.buf.append((t, v))
|
|
self.last_t, self.last_v = t, v
|
|
if v is not None:
|
|
self.lo = v if self.lo is None else min(self.lo, v)
|
|
self.hi = v if self.hi is None else max(self.hi, v)
|
|
|
|
def reset_minmax(self):
|
|
self.lo = self.hi = self.last_v
|
|
|
|
def series(self, since=None):
|
|
"""Return [(t, v), ...]; if since given, only samples with t >= since."""
|
|
if since is None:
|
|
return list(self.buf)
|
|
return [(t, v) for (t, v) in self.buf if t >= since]
|
|
|
|
|
|
class TimeSeriesStore:
|
|
"""Thread-safe collection of Channels keyed by PID key."""
|
|
|
|
def __init__(self, maxlen=3600):
|
|
self._ch = {}
|
|
self._maxlen = maxlen
|
|
self._lock = threading.Lock()
|
|
self.recorder = None # set to a recorder with .write(key, t, v)
|
|
|
|
def channel(self, key):
|
|
with self._lock:
|
|
c = self._ch.get(key)
|
|
if c is None:
|
|
c = Channel(key, self._maxlen)
|
|
self._ch[key] = c
|
|
return c
|
|
|
|
def push(self, key, t, v):
|
|
self.channel(key).push(t, v)
|
|
rec = self.recorder
|
|
if rec is not None:
|
|
rec.write(key, t, v)
|
|
|
|
def latest(self, key):
|
|
c = self._ch.get(key)
|
|
return None if c is None else c.last_v
|
|
|
|
def minmax(self, key):
|
|
c = self._ch.get(key)
|
|
return (None, None) if c is None else (c.lo, c.hi)
|
|
|
|
def reset_minmax(self, keys=None):
|
|
with self._lock:
|
|
for k, c in self._ch.items():
|
|
if keys is None or k in keys:
|
|
c.reset_minmax()
|
|
|
|
def keys(self):
|
|
with self._lock:
|
|
return list(self._ch.keys())
|
|
|
|
|
|
class CsvRecorder:
|
|
"""Long-format session recorder: one row per sample (t,key,value).
|
|
|
|
Long format (vs wide) tolerates per-PID poll rates and PIDs appearing
|
|
mid-session. Replay re-pushes rows into a fresh store in t order.
|
|
"""
|
|
|
|
def __init__(self, path):
|
|
self._f = open(path, "w")
|
|
self._f.write("t,key,value\n")
|
|
self._lock = threading.Lock()
|
|
|
|
def write(self, key, t, v):
|
|
with self._lock:
|
|
self._f.write(f"{t:.3f},{key},{'' if v is None else v}\n")
|
|
|
|
def close(self):
|
|
with self._lock:
|
|
self._f.close()
|
|
|
|
|
|
def replay_csv(path, store):
|
|
"""Load a CsvRecorder file back into a store (for playback)."""
|
|
with open(path) as f:
|
|
next(f, None) # header
|
|
for line in f:
|
|
parts = line.rstrip("\n").split(",", 2)
|
|
if len(parts) != 3:
|
|
continue
|
|
t, key, v = parts
|
|
try:
|
|
t = float(t)
|
|
except ValueError:
|
|
continue
|
|
val = None if v == "" else float(v) if _is_num(v) else v
|
|
store.push(key, t, val)
|
|
return store
|
|
|
|
|
|
def _is_num(s):
|
|
try:
|
|
float(s)
|
|
return True
|
|
except ValueError:
|
|
return False
|