Make app vehicle-agnostic: JSON vehicle profiles + menu bar
Vehicle data is now DATA, not code. PIDs/scaling/DTCs/presets live in profiles/*.json; the app loads them at runtime, so it works across vehicles and others can contribute profiles (open source). Core: - obdcore/formula.py: safe AST evaluator for scaling formulas (A/B/... byte vars, Torque/FORScan convention). Only arithmetic/bitwise + min/max/abs/ round/int/float; names/attrs/arbitrary calls rejected at load -> a community profile CANNOT execute code. - obdcore/profile.py: load/save/list profiles; compiles each formula into a decode callable. registry.py now profile-backed (PidRegistry/DtcDatabase take a Profile); hardcoded Ford table removed. - store.py: clear()/snapshot()/export_csv() for capture management. Profiles: - profiles/ford-6.0-powerstroke.json (27 PIDs, verified formulas, DTCs) - profiles/generic-obd2.json (standard SAE Mode-01 base, any vehicle) - profiles/README.md (schema + formula language + contributing) GUI: - Menu bar: File (new/record/export/replay capture, quit), Profile (switch/ load/import/reload/edit-JSON/export, live profile list), View (Graph/Table views, gauges P2, toggle PID dock, normalize, light/dark theme), Help (about/confidence legend/profile info). - PID browser + presets rebuild on profile switch; added Table view; raw-JSON profile editor dialog (validates schema+formulas before saving). Tests: profiles load+compile, formula sandbox rejects hostile input, decoders still match real truck bytes, crank/derived/dead-PID/replay -- all pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016yT89n4zR4qbrySoSiEyZs
This commit is contained in:
+21
-12
@@ -1,20 +1,29 @@
|
||||
"""obdcore -- headless OBD-II acquisition core for the ford-obd project.
|
||||
"""obdcore -- headless, vehicle-agnostic OBD-II acquisition core.
|
||||
|
||||
Layered, GUI-agnostic foundation shared by the terminal tool and the
|
||||
forthcoming PySide6 + pyqtgraph Windows app:
|
||||
Vehicle data (PIDs, scaling, DTCs, presets) lives in JSON profiles under
|
||||
profiles/ -- loaded at runtime, not hardcoded -- so the app works across
|
||||
vehicles and others can contribute profiles.
|
||||
|
||||
link.py ElmLink -- ELM327 serial transport (+ MockLink in mock.py)
|
||||
registry.py PidRegistry -- verified Ford 6.0 PID table + DTC database
|
||||
scheduler.py PollScheduler -- prioritized round-robin polling engine
|
||||
store.py TimeSeriesStore -- ring buffers, min/max, record/replay
|
||||
formula.py safe A/B/... scaling-formula evaluator (no code execution)
|
||||
profile.py load/save/list vehicle profiles (JSON)
|
||||
registry.py PidRegistry / DtcDatabase model + lookups
|
||||
link.py ElmLink ELM327 serial transport (+ MockLink in mock.py)
|
||||
scheduler.py PollScheduler prioritized polling engine
|
||||
store.py TimeSeriesStore ring buffers + record/replay
|
||||
|
||||
See ARCHITECTURE.md for the full design and roadmap.
|
||||
See ARCHITECTURE.md and profiles/README.md.
|
||||
"""
|
||||
from .registry import PidRegistry, DtcDatabase, Pid, Dtc, PRESETS
|
||||
from .store import TimeSeriesStore, CsvRecorder, replay_csv
|
||||
from .registry import PidRegistry, DtcDatabase, Pid, Dtc
|
||||
from .profile import (Profile, load_profile, save_profile, list_profiles,
|
||||
profiles_dir, default_profile_path, load_default)
|
||||
from .formula import compile_formula, FormulaError
|
||||
from .store import TimeSeriesStore, CsvRecorder, replay_csv, export_csv
|
||||
from .scheduler import PollScheduler
|
||||
|
||||
__all__ = [
|
||||
"PidRegistry", "DtcDatabase", "Pid", "Dtc", "PRESETS",
|
||||
"TimeSeriesStore", "CsvRecorder", "replay_csv", "PollScheduler",
|
||||
"PidRegistry", "DtcDatabase", "Pid", "Dtc",
|
||||
"Profile", "load_profile", "save_profile", "list_profiles",
|
||||
"profiles_dir", "default_profile_path", "load_default",
|
||||
"compile_formula", "FormulaError",
|
||||
"TimeSeriesStore", "CsvRecorder", "replay_csv", "export_csv", "PollScheduler",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
"""Safe formula evaluator for vehicle-profile PID scaling.
|
||||
|
||||
Profiles are community-contributed data, so decode formulas must NOT be able to
|
||||
execute arbitrary code. Formulas are arithmetic expressions over named
|
||||
variables -- the de-facto OBD convention used by Torque / FORScan / ScanGauge:
|
||||
|
||||
raw-mode PIDs: variables A, B, C, ... = response data bytes 0, 1, 2, ...
|
||||
e.g. "(A*256+B)*0.57" "A-40" "(A>>1)&1" "A//2"
|
||||
derived PIDs: variables are other PID keys
|
||||
e.g. "MAP - BARO"
|
||||
|
||||
Only numeric literals, the named variables, arithmetic/bitwise operators, and a
|
||||
small whitelist of functions are allowed. No names, attributes, subscripts,
|
||||
comprehensions, or calls outside the whitelist -- anything else raises
|
||||
FormulaError at compile time, so a bad/hostile profile fails loudly on load.
|
||||
"""
|
||||
import ast
|
||||
import operator
|
||||
|
||||
_BIN = {
|
||||
ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
|
||||
ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
|
||||
ast.Mod: operator.mod, ast.Pow: operator.pow,
|
||||
ast.BitAnd: operator.and_, ast.BitOr: operator.or_, ast.BitXor: operator.xor,
|
||||
ast.LShift: operator.lshift, ast.RShift: operator.rshift,
|
||||
}
|
||||
_UNARY = {ast.USub: operator.neg, ast.UAdd: operator.pos, ast.Invert: operator.invert}
|
||||
_FUNCS = {"min": min, "max": max, "abs": abs, "round": round,
|
||||
"int": int, "float": float}
|
||||
|
||||
|
||||
class FormulaError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _validate(node, allowed):
|
||||
if isinstance(node, ast.Expression):
|
||||
return _validate(node.body, allowed)
|
||||
if isinstance(node, ast.BinOp):
|
||||
if type(node.op) not in _BIN:
|
||||
raise FormulaError(f"operator not allowed: {type(node.op).__name__}")
|
||||
_validate(node.left, allowed)
|
||||
_validate(node.right, allowed)
|
||||
return
|
||||
if isinstance(node, ast.UnaryOp):
|
||||
if type(node.op) not in _UNARY:
|
||||
raise FormulaError(f"unary op not allowed: {type(node.op).__name__}")
|
||||
_validate(node.operand, allowed)
|
||||
return
|
||||
if isinstance(node, ast.Constant):
|
||||
if not isinstance(node.value, (int, float)) or isinstance(node.value, bool):
|
||||
raise FormulaError("only numeric constants allowed")
|
||||
return
|
||||
if isinstance(node, ast.Name):
|
||||
if node.id not in allowed:
|
||||
raise FormulaError(f"unknown variable {node.id!r} (allowed: {sorted(allowed)})")
|
||||
return
|
||||
if isinstance(node, ast.Call):
|
||||
if not isinstance(node.func, ast.Name) or node.func.id not in _FUNCS:
|
||||
raise FormulaError("only min/max/abs/round/int/float calls allowed")
|
||||
if node.keywords:
|
||||
raise FormulaError("keyword args not allowed")
|
||||
for a in node.args:
|
||||
_validate(a, allowed)
|
||||
return
|
||||
raise FormulaError(f"expression not allowed: {type(node).__name__}")
|
||||
|
||||
|
||||
def _eval(node, names):
|
||||
if isinstance(node, ast.Expression):
|
||||
return _eval(node.body, names)
|
||||
if isinstance(node, ast.BinOp):
|
||||
return _BIN[type(node.op)](_eval(node.left, names), _eval(node.right, names))
|
||||
if isinstance(node, ast.UnaryOp):
|
||||
return _UNARY[type(node.op)](_eval(node.operand, names))
|
||||
if isinstance(node, ast.Constant):
|
||||
return node.value
|
||||
if isinstance(node, ast.Name):
|
||||
return names[node.id]
|
||||
if isinstance(node, ast.Call):
|
||||
return _FUNCS[node.func.id](*[_eval(a, names) for a in node.args])
|
||||
raise FormulaError(f"expression not allowed: {type(node).__name__}")
|
||||
|
||||
|
||||
def compile_formula(expr, allowed_names):
|
||||
"""Return fn(names_dict) -> number. Raises FormulaError on disallowed input."""
|
||||
try:
|
||||
tree = ast.parse(expr, mode="eval")
|
||||
except SyntaxError as e:
|
||||
raise FormulaError(f"bad formula {expr!r}: {e}")
|
||||
allowed = set(allowed_names)
|
||||
_validate(tree, allowed)
|
||||
|
||||
def fn(names):
|
||||
return _eval(tree, names)
|
||||
|
||||
fn.expr = expr
|
||||
return fn
|
||||
@@ -0,0 +1,155 @@
|
||||
"""Vehicle profiles -- load/save/list the JSON files under profiles/.
|
||||
|
||||
A profile is pure data: vehicle metadata, PID definitions (with safe formula
|
||||
strings), DTC meanings, and named presets (perspectives). Loading a profile
|
||||
compiles each PID's formula into a decode callable; nothing in a profile can
|
||||
execute arbitrary code (see formula.py).
|
||||
|
||||
JSON schema (schema=1):
|
||||
{
|
||||
"schema": 1,
|
||||
"meta": {"name","make","model","years","engine","author","version",
|
||||
"protocol","notes"},
|
||||
"pids": [{"key","name","mode","pid","nbytes","formula","unit","group",
|
||||
"vmin","vmax","confidence","round","deps","notes"}, ...],
|
||||
"presets": {"crank":[keys...], ...},
|
||||
"dtcs": [{"code","desc","system","no_start","causes"}, ...]
|
||||
}
|
||||
"""
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .formula import compile_formula
|
||||
from .registry import Pid, Dtc
|
||||
|
||||
SCHEMA = 1
|
||||
BYTE_VARS = [chr(65 + i) for i in range(8)] # A..H
|
||||
|
||||
|
||||
@dataclass
|
||||
class Profile:
|
||||
meta: dict
|
||||
pids: list
|
||||
dtcs: list
|
||||
presets: dict
|
||||
path: str = None
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self.meta.get("name", "Unnamed profile")
|
||||
|
||||
|
||||
def profiles_dir():
|
||||
return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"profiles")
|
||||
|
||||
|
||||
def _round(v, rnd):
|
||||
if rnd is None:
|
||||
return v
|
||||
return int(round(v)) if rnd == 0 else round(v, rnd)
|
||||
|
||||
|
||||
def _build_decode(d):
|
||||
mode = d.get("mode", "22")
|
||||
rnd = d.get("round")
|
||||
if mode == "atrv":
|
||||
return None
|
||||
formula = d.get("formula", "")
|
||||
if mode == "derived":
|
||||
deps = tuple(d.get("deps", ()))
|
||||
fn = compile_formula(formula, deps)
|
||||
def dec(vals, fn=fn, deps=deps, rnd=rnd):
|
||||
return _round(fn(dict(zip(deps, vals))), rnd)
|
||||
return dec
|
||||
fn = compile_formula(formula, BYTE_VARS)
|
||||
def dec(raw, fn=fn, rnd=rnd):
|
||||
names = {BYTE_VARS[i]: raw[i] for i in range(min(len(raw), 8))}
|
||||
return _round(fn(names), rnd)
|
||||
return dec
|
||||
|
||||
|
||||
def _pid_from_dict(d):
|
||||
return Pid(
|
||||
key=d["key"], name=d.get("name", d["key"]), mode=d.get("mode", "22"),
|
||||
pid=d.get("pid", ""), nbytes=d.get("nbytes", 2),
|
||||
formula=d.get("formula", ""), decode=_build_decode(d),
|
||||
unit=d.get("unit", ""), group=d.get("group", "misc"),
|
||||
vmin=d.get("vmin", 0.0), vmax=d.get("vmax", 100.0),
|
||||
confidence=d.get("confidence", "verified"), round=d.get("round"),
|
||||
deps=tuple(d.get("deps", ())), notes=d.get("notes", ""),
|
||||
)
|
||||
|
||||
|
||||
def load_profile(path):
|
||||
with open(path) as f:
|
||||
raw = json.load(f)
|
||||
if raw.get("schema", 1) != SCHEMA:
|
||||
raise ValueError(f"unsupported profile schema {raw.get('schema')} in {path}")
|
||||
pids = [_pid_from_dict(d) for d in raw.get("pids", [])]
|
||||
dtcs = [Dtc(code=x["code"], desc=x.get("desc", ""), system=x.get("system", "powertrain"),
|
||||
no_start=x.get("no_start", False), causes=x.get("causes", ""))
|
||||
for x in raw.get("dtcs", [])]
|
||||
return Profile(meta=raw.get("meta", {}), pids=pids, dtcs=dtcs,
|
||||
presets=raw.get("presets", {}), path=path)
|
||||
|
||||
|
||||
def _pid_to_dict(p):
|
||||
d = {"key": p.key, "name": p.name, "mode": p.mode}
|
||||
if p.pid:
|
||||
d["pid"] = p.pid
|
||||
if p.mode in ("01", "22"):
|
||||
d["nbytes"] = p.nbytes
|
||||
if p.formula:
|
||||
d["formula"] = p.formula
|
||||
if p.deps:
|
||||
d["deps"] = list(p.deps)
|
||||
d.update({"unit": p.unit, "group": p.group, "vmin": p.vmin, "vmax": p.vmax,
|
||||
"confidence": p.confidence})
|
||||
if p.round is not None:
|
||||
d["round"] = p.round
|
||||
if p.notes:
|
||||
d["notes"] = p.notes
|
||||
return d
|
||||
|
||||
|
||||
def save_profile(profile, path=None):
|
||||
path = path or profile.path
|
||||
out = {
|
||||
"schema": SCHEMA,
|
||||
"meta": profile.meta,
|
||||
"pids": [_pid_to_dict(p) for p in profile.pids],
|
||||
"presets": profile.presets,
|
||||
"dtcs": [{"code": d.code, "desc": d.desc, "system": d.system,
|
||||
"no_start": d.no_start, "causes": d.causes} for d in profile.dtcs],
|
||||
}
|
||||
with open(path, "w") as f:
|
||||
json.dump(out, f, indent=2)
|
||||
return path
|
||||
|
||||
|
||||
def list_profiles(directory=None):
|
||||
"""Return [(path, meta_dict), ...] for every *.json profile in directory."""
|
||||
directory = directory or profiles_dir()
|
||||
out = []
|
||||
for p in sorted(glob.glob(os.path.join(directory, "*.json"))):
|
||||
try:
|
||||
with open(p) as f:
|
||||
meta = json.load(f).get("meta", {})
|
||||
out.append((p, meta))
|
||||
except Exception:
|
||||
continue
|
||||
return out
|
||||
|
||||
|
||||
DEFAULT_PROFILE = "ford-6.0-powerstroke.json"
|
||||
|
||||
|
||||
def default_profile_path():
|
||||
return os.path.join(profiles_dir(), DEFAULT_PROFILE)
|
||||
|
||||
|
||||
def load_default():
|
||||
return load_profile(default_profile_path())
|
||||
+30
-144
@@ -1,137 +1,49 @@
|
||||
"""PID + DTC registry for the Ford 6.0L Power Stroke (plus generic OBD-II).
|
||||
"""PID + DTC data model and registry, backed by a vehicle Profile.
|
||||
|
||||
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).
|
||||
The actual PID numbers, scaling formulas, and DTC meanings live in JSON
|
||||
vehicle profiles under profiles/ (data, not code) so the app is vehicle-
|
||||
agnostic and others can contribute profiles. This module is the in-memory
|
||||
model + lookups; profile.py loads/saves the JSON.
|
||||
"""
|
||||
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"
|
||||
mode: str = "22" # "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)
|
||||
formula: str = "" # scaling expr in A/B/... (raw) or dep keys (derived)
|
||||
decode: Callable = None # built from formula by profile loader
|
||||
unit: str = ""
|
||||
group: str = "misc" # fuel | ficm | air | engine | driveline | power
|
||||
group: str = "misc" # fuel | ficm | air | engine | driveline | power | misc
|
||||
vmin: float = 0.0
|
||||
vmax: float = 100.0
|
||||
confidence: str = "verified"
|
||||
deps: Tuple[str, ...] = () # for derived channels
|
||||
confidence: str = "verified" # verified | doc | tentative
|
||||
round: int = None # display rounding (None=raw float, 0=int)
|
||||
deps: Tuple[str, ...] = ()
|
||||
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"))
|
||||
# ---- More documented PIDs from the workflow (not yet truck-verified) ----
|
||||
a(Pid("VGT", "VGT Duty Cycle", "22", "096D", 2,
|
||||
lambda b: round(_u16(b) * 100 / 32767, 1), "%", "air", 0, 100,
|
||||
"doc", notes="turbo vane duty"))
|
||||
a(Pid("FAN", "Fan Speed", "22", "099F", 2,
|
||||
lambda b: round(_u16(b) / 4), "rpm", "engine", 0, 4000,
|
||||
"doc", notes="real ceiling ~3500"))
|
||||
a(Pid("INJ_TIMING", "Injection Timing", "22", "09CC", 2,
|
||||
lambda b: round(_u16(b) * 10 / 64, 1), "degBTDC", "fuel", -10, 30,
|
||||
"tentative", notes="scaling disputed; using *10/64 (ScanGauge), not /10"))
|
||||
a(Pid("VBAT", "Battery (PCM)", "22", "1172", 1,
|
||||
lambda b: round(b[0] / 16, 1), "V", "power", 0, 16,
|
||||
"tentative", notes="PCM-reported B+; distinct from ATRV port voltage"))
|
||||
a(Pid("FUEL_PUMP", "Fuel Pump Duty (HFCM)", "22", "1672", 1,
|
||||
lambda b: round(b[0] * 100 / 128, 1), "%", "fuel", 0, 100,
|
||||
"tentative", notes="sits ~100%, drops on high EOT"))
|
||||
a(Pid("FUEL_LVL", "Fuel Level", "22", "16C1", 2,
|
||||
lambda b: round(_u16(b) * 100 / 328, 1), "%", "misc", 0, 100,
|
||||
"tentative", notes="UNCALIBRATED -- needs per-truck full/empty cal"))
|
||||
a(Pid("MFDES", "Mass Fuel Desired", "22", "1411", 2,
|
||||
lambda b: _u16(b), "raw", "fuel", 0, 65535,
|
||||
"tentative", notes="~mg/stroke internal count; no verified GPH formula"))
|
||||
# ---- 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", "VGT", "EOT", "ECT", "EBP", "LOAD", "RPM", "IPR", "BATT"],
|
||||
"vitals": ["ICP", "FICM_M", "FICM_L", "IPR", "BATT", "RPM", "ECT", "EOT",
|
||||
"IAT", "VPCM"],
|
||||
}
|
||||
@dataclass
|
||||
class Dtc:
|
||||
code: str
|
||||
desc: str
|
||||
system: str = "powertrain"
|
||||
no_start: bool = False
|
||||
causes: str = ""
|
||||
|
||||
|
||||
class PidRegistry:
|
||||
def __init__(self):
|
||||
self._by_key = {p.key: p for p in _build()}
|
||||
"""In-memory PID set + presets for the active vehicle profile."""
|
||||
|
||||
def __init__(self, profile):
|
||||
self.profile = profile
|
||||
self._by_key = {p.key: p for p in profile.pids}
|
||||
self.presets = dict(profile.presets)
|
||||
|
||||
def get(self, key):
|
||||
return self._by_key.get(key)
|
||||
@@ -143,41 +55,15 @@ class PidRegistry:
|
||||
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]
|
||||
return [self._by_key[k] for k in self.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}
|
||||
def preset_names(self):
|
||||
return list(self.presets.keys())
|
||||
|
||||
|
||||
class DtcDatabase:
|
||||
def __init__(self):
|
||||
self._db = _dtcs()
|
||||
def __init__(self, profile):
|
||||
self._db = {d.code: d for d in profile.dtcs}
|
||||
|
||||
def get(self, code):
|
||||
return self._db.get(code) or Dtc(code, "(unknown - look up this code)")
|
||||
|
||||
@@ -87,6 +87,35 @@ class TimeSeriesStore:
|
||||
with self._lock:
|
||||
return list(self._ch.keys())
|
||||
|
||||
def clear(self):
|
||||
"""Empty every channel's history + min/max (start a fresh capture)."""
|
||||
with self._lock:
|
||||
chans = list(self._ch.values())
|
||||
for c in chans:
|
||||
with c._lock:
|
||||
c.buf.clear()
|
||||
c.lo = c.hi = c.last_v = c.last_t = None
|
||||
|
||||
def snapshot(self):
|
||||
"""Return {key: [(t, v), ...]} of all current channel history."""
|
||||
with self._lock:
|
||||
chans = dict(self._ch)
|
||||
return {k: c.series() for k, c in chans.items()}
|
||||
|
||||
|
||||
def export_csv(store, path):
|
||||
"""Write a store's current buffers to a long-format CSV (t,key,value)."""
|
||||
rows = []
|
||||
for key, series in store.snapshot().items():
|
||||
for t, v in series:
|
||||
rows.append((t, key, v))
|
||||
rows.sort(key=lambda r: r[0])
|
||||
with open(path, "w") as f:
|
||||
f.write("t,key,value\n")
|
||||
for t, key, v in rows:
|
||||
f.write(f"{t:.3f},{key},{'' if v is None else v}\n")
|
||||
return path
|
||||
|
||||
|
||||
class CsvRecorder:
|
||||
"""Long-format session recorder: one row per sample (t,key,value).
|
||||
|
||||
Reference in New Issue
Block a user