"""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 import sys from dataclasses import dataclass, field from .formula import compile_formula from .registry import Pid, Dtc from .actions import Action, ActionStep, validate_action 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 actions: list = None @property def name(self): return self.meta.get("name", "Unnamed profile") def profiles_dir(): # In a PyInstaller bundle, profiles/ is added as data and unpacked under # sys._MEIPASS; otherwise it's the repo's profiles/ next to obdcore/. if getattr(sys, "frozen", False): base = getattr(sys, "_MEIPASS", os.path.dirname(sys.executable)) return os.path.join(base, "profiles") 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", ""), warn_hi=d.get("warn_hi"), redline_hi=d.get("redline_hi"), warn_lo=d.get("warn_lo"), redline_lo=d.get("redline_lo"), ) 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", [])] actions = [] for a in raw.get("actions", []): act = Action( key=a["key"], name=a.get("name", a["key"]), kind=a.get("kind", "test"), risk=a.get("risk", "safe"), description=a.get("description", ""), warning=a.get("warning", ""), session=a.get("session"), security=a.get("security"), steps=[ActionStep(send=s["send"], expect=s.get("expect", "")) for s in a.get("steps", [])], success_msg=a.get("success_msg", "Done.")) validate_action(act) # rejects malformed hex actions.append(act) return Profile(meta=raw.get("meta", {}), pids=pids, dtcs=dtcs, presets=raw.get("presets", {}), path=path, actions=actions) 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 for z in ("warn_hi", "redline_hi", "warn_lo", "redline_lo"): if getattr(p, z) is not None: d[z] = getattr(p, z) 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], } if profile.actions: out["actions"] = [] for a in profile.actions: ad = {"key": a.key, "name": a.name, "kind": a.kind, "risk": a.risk} if a.description: ad["description"] = a.description if a.warning: ad["warning"] = a.warning if a.session: ad["session"] = a.session if a.security: ad["security"] = a.security ad["steps"] = [{"send": s.send, **({"expect": s.expect} if s.expect else {})} for s in a.steps] if a.success_msg != "Done.": ad["success_msg"] = a.success_msg out["actions"].append(ad) 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())