Files
obdash/obdcore/registry.py
T
justin 6bee9c0d7f Scaffold obdcore (headless acquisition core) + ARCHITECTURE.md
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
2026-06-30 13:41:24 -04:00

165 lines
6.7 KiB
Python

"""PID + DTC registry for the Ford 6.0L Power Stroke (plus generic OBD-II).
Canonical home for the verified Mode-22 addresses, scaling, and the DTC
database. Decoders are plain callables on the raw byte list. Confidence:
verified -- multi-source AND confirmed on the truck's scan/crank
doc -- corroborated in sources, not (yet) read on the truck
tentative -- single-source or disputed scaling
PID numbers/scaling corrected 2026-06-29 by the ford-60-pid-hunt workflow;
see diagnostics/2026-06-29-no-start/pid-research.md. 09D0 (FICM Main) was
confirmed on-truck 2026-06-30 (read 48.0V during a crank, intermittent).
"""
from dataclasses import dataclass, field
from typing import Callable, Tuple
def _u16(b):
return (b[0] << 8) + b[1]
@dataclass
class Pid:
key: str
name: str
mode: str # "01" | "22" | "atrv" | "derived"
pid: str = "" # hex: "1446" (m22) or "0C" (m01)
nbytes: int = 2
decode: Callable = None # m01/m22: f(raw_bytes); derived: f(dep_values)
unit: str = ""
group: str = "misc" # fuel | ficm | air | engine | driveline | power
vmin: float = 0.0
vmax: float = 100.0
confidence: str = "verified"
deps: Tuple[str, ...] = () # for derived channels
notes: str = ""
def _build():
P = []
a = P.append
# ---- Ford-enhanced Mode 22 -- pressures / fuel ----
a(Pid("ICP", "Injection Control Pressure", "22", "1446", 2,
lambda b: round(_u16(b) * 0.57, 1), "psi", "fuel", 0, 3500,
"verified", notes="need ~500+ psi to fire"))
a(Pid("ICP_V", "ICP Sensor Voltage", "22", "16AD", 2,
lambda b: round(_u16(b) * 0.000072, 4), "V", "fuel", 0, 5,
"tentative", notes="single-source"))
a(Pid("IPR", "Injection Pressure Regulator", "22", "1434", 1,
lambda b: round(b[0] * 13.53 / 35, 1), "%", "fuel", 0, 100,
"tentative", notes="KOEO ~14-15%, cranking ~30-40%"))
a(Pid("MAP", "Manifold Absolute Pressure", "22", "1440", 2,
lambda b: round(_u16(b) * 0.03625, 2), "psia", "air", 0, 60,
"verified"))
a(Pid("BARO", "Barometric Pressure", "22", "1442", 2,
lambda b: round(_u16(b) * 0.03625, 2), "psia", "air", 0, 20,
"verified"))
a(Pid("EBP", "Exhaust Back Pressure", "22", "1445", 2,
lambda b: round(_u16(b) * 0.03625, 2), "psia", "air", 0, 60,
"verified", notes="minus BARO = gauge"))
a(Pid("EOT", "Engine Oil Temperature", "22", "1310", 2,
lambda b: round(_u16(b) / 100.0 - 40, 1), "C", "engine", -40, 160,
"verified"))
# ---- FICM ----
a(Pid("FICM_M", "FICM Main Power", "22", "09D0", 2,
lambda b: round(_u16(b) / 256.0, 1), "V", "ficm", 0, 55,
"verified", notes="~48V; <45 suspect; reads intermittently while cranking"))
a(Pid("FICM_L", "FICM Logic Power", "22", "09CF", 2,
lambda b: round(_u16(b) / 256.0, 1), "V", "ficm", 0, 16,
"doc"))
a(Pid("FICM_V", "FICM Vehicle Power", "22", "09CE", 2,
lambda b: round(_u16(b) / 256.0, 1), "V", "ficm", 0, 16,
"doc"))
a(Pid("FICM_SYNC", "FICM Sync", "22", "09CD", 1,
lambda b: (b[0] >> 1) & 1, "", "ficm", 0, 1,
"doc", notes="1=in sync, 0=no sync"))
# ---- Driveline ----
a(Pid("GEAR", "Current Gear", "22", "11B3", 1,
lambda b: b[0] // 2, "", "driveline", 0, 6, "verified"))
a(Pid("TSS", "Trans Input Shaft Speed", "22", "11B4", 2,
lambda b: round(_u16(b) / 4), "rpm", "driveline", 0, 4000, "verified"))
# ---- Generic Mode 01 ----
a(Pid("RPM", "Engine RPM", "01", "0C", 2,
lambda b: round(_u16(b) / 4), "rpm", "engine", 0, 4000, "verified"))
a(Pid("ECT", "Engine Coolant Temp", "01", "05", 1,
lambda b: b[0] - 40, "C", "engine", -40, 160, "verified"))
a(Pid("IAT", "Intake Air Temp", "01", "0F", 1,
lambda b: b[0] - 40, "C", "air", -40, 160, "verified"))
a(Pid("LOAD", "Engine Load", "01", "04", 1,
lambda b: round(b[0] * 100 / 255), "%", "engine", 0, 100, "verified"))
a(Pid("VPCM", "Module Voltage", "01", "42", 2,
lambda b: round(_u16(b) / 1000.0, 2), "V", "power", 0, 16, "verified"))
# ---- Pseudo / derived ----
a(Pid("BATT", "Battery (OBD port)", "atrv", "", 0,
None, "V", "power", 0, 16, "verified"))
a(Pid("BOOST", "Boost (MGP)", "derived", "", 0,
lambda vals: round(vals[0] - vals[1], 2), "psi", "air", -5, 40,
"verified", deps=("MAP", "BARO"), notes="MAP - BARO"))
return P
# Subscription presets per perspective (key -> default poll Hz set by scheduler)
PRESETS = {
"crank": ["ICP", "FICM_M", "BATT", "RPM"],
"driving": ["BOOST", "EOT", "ECT", "EBP", "LOAD", "RPM", "IPR", "FICM_M", "BATT"],
"vitals": ["ICP", "FICM_M", "FICM_L", "IPR", "BATT", "RPM", "ECT", "EOT",
"IAT", "VPCM"],
}
class PidRegistry:
def __init__(self):
self._by_key = {p.key: p for p in _build()}
def get(self, key):
return self._by_key.get(key)
def all(self):
return list(self._by_key.values())
def group(self, g):
return [p for p in self._by_key.values() if p.group == g]
def preset(self, name):
return [self._by_key[k] for k in PRESETS.get(name, []) if k in self._by_key]
# ---------------------------------------------------------------------------
# DTC database -- generic SAE + notable Ford 6.0 codes. The full Ford code
# DB is being built by a separate cross-verified workflow; this is the seed.
# ---------------------------------------------------------------------------
@dataclass
class Dtc:
code: str
desc: str
system: str = "powertrain"
no_start: bool = False
causes: str = ""
def _dtcs():
rows = [
Dtc("P0087", "Fuel rail/system pressure too LOW", "fuel", True),
Dtc("P0088", "Fuel rail/system pressure too HIGH", "fuel"),
Dtc("P0148", "Fuel delivery error (low pressure / HPOP / IPR)", "fuel", True),
Dtc("P0335", "Crankshaft position (CKP) sensor circuit", "engine", True),
Dtc("P0340", "Camshaft position (CMP) sensor circuit", "engine", True),
Dtc("P0611", "FICM performance", "ficm", True),
Dtc("P1316", "Injector circuit/FICM codes detected", "ficm", True),
Dtc("P0606", "PCM processor fault", "power", True),
Dtc("U0100", "Lost communication with PCM/ECM", "network", True),
Dtc("P0670", "Glow plug control module circuit", "engine"),
]
return {d.code: d for d in rows}
class DtcDatabase:
def __init__(self):
self._db = _dtcs()
def get(self, code):
return self._db.get(code) or Dtc(code, "(unknown - look up this code)")
def all(self):
return list(self._db.values())