| """Apply ELM327-emulator PID overrides to simulate automotive faults.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| import re |
| import threading |
| from dataclasses import dataclass |
| from typing import Any |
|
|
| from elm_server import _emulator_lock, ensure_elm_emulator, get_emulator |
| from obd_connection import obd_session_lock |
|
|
| logger = logging.getLogger("car-diagnostic-agent.faults") |
|
|
| _active_fault_id: str | None = None |
| _state_lock = threading.Lock() |
|
|
| _DTC_PATTERN = re.compile(r"[PCBU]\d{4}") |
|
|
|
|
| @dataclass(frozen=True) |
| class FaultProfile: |
| id: str |
| name: str |
| category: str |
| description: str |
| signature: str |
| typical_dtc: str |
| llm_hint: str |
| overrides: dict[str, str] |
|
|
|
|
| |
|
|
|
|
| def trim_b(pct: float) -> int: |
| return int(128 + pct * 128 / 100) |
|
|
|
|
| def maf_hex(gps: float) -> str: |
| return f"{int(gps * 100):04X}" |
|
|
|
|
| def clt_hex(deg_c: float) -> str: |
| return f"{int(deg_c + 40):02X}" |
|
|
|
|
| def rpm_hex(rpm: float) -> str: |
| return f"{int(4 * rpm):04X}" |
|
|
|
|
| def volt_hex(volts: float) -> str: |
| return f"{int(volts * 1000):04X}" |
|
|
|
|
| def pct_byte(pct: float) -> str: |
| return f"{int(pct * 255 / 100):02X}" |
|
|
|
|
| def ax(expr: str) -> str: |
| """Build an emulator answer XML string with one <exec> block.""" |
| return f"<exec>{expr}</exec><writeln />" |
|
|
|
|
| def _randint_expr(low: int, high: int) -> str: |
| return f'__import__("random").randint({low}, {high})' |
|
|
|
|
| def parse_dtc_codes(typical_dtc: str) -> list[str]: |
| """Extract OBD-II codes like P0171 from a fault's typical_dtc field.""" |
| return _DTC_PATTERN.findall(typical_dtc or "") |
|
|
|
|
| def dtc_to_bytes(code: str) -> tuple[int, int]: |
| kind = {"P": 0, "C": 1, "B": 2, "U": 3}[code[0]] |
| b1 = (kind << 6) | (int(code[1]) << 4) | int(code[2], 16) |
| b2 = int(code[3:5], 16) |
| return b1, b2 |
|
|
|
|
| def build_get_dtc_answer(typical_dtc: str) -> str | None: |
| """Build emulator GET_DTC override XML for Mode 03 stored codes.""" |
| codes = parse_dtc_codes(typical_dtc) |
| if not codes: |
| return None |
|
|
| body = ["43", f"{len(codes):02X}"] |
| for code in codes: |
| b1, b2 = dtc_to_bytes(code) |
| body.extend([f"{b1:02X}", f"{b2:02X}"]) |
| payload = " ".join(body) |
| return ax(f'ECU_R_ADDR_E + " {len(body):02X} {payload}"') |
|
|
|
|
| def get_active_dtc_codes() -> list[str]: |
| """Return DTC codes for the active fault profile, if any.""" |
| fault = get_active_fault() |
| if fault is None: |
| return [] |
| return parse_dtc_codes(fault.typical_dtc) |
|
|
|
|
| def _effective_overrides(fault: FaultProfile) -> dict[str, str]: |
| overrides = dict(fault.overrides) |
| dtc_answer = build_get_dtc_answer(fault.typical_dtc) |
| if dtc_answer is not None: |
| overrides["GET_DTC"] = dtc_answer |
| return overrides |
|
|
|
|
| |
|
|
|
|
| def _build_faults() -> dict[str, FaultProfile]: |
| faults: dict[str, FaultProfile] = {} |
|
|
| def add(profile: FaultProfile) -> None: |
| faults[profile.id] = profile |
|
|
| add( |
| FaultProfile( |
| id="vacuum_leak", |
| name="Vacuum leak (idle)", |
| category="Air system", |
| description="Unmetered air — cracked hose, PCV leak, intake gasket.", |
| signature="STFT ↑↑, LTFT ↑↑, MAF low, idle RPM slightly high", |
| typical_dtc="P0171", |
| llm_hint="Likely vacuum leak; positive trims and low MAF at idle.", |
| overrides={ |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(20):02X}"' |
| ), |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(15):02X}"' |
| ), |
| "MAF": ax(f'ECU_R_ADDR_E + " 04 41 10 {maf_hex(12)}"'), |
| "INTAKE_PRESSURE": ax('ECU_R_ADDR_E + " 03 41 0B 37"'), |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(950)}"'), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 00"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="dirty_maf", |
| name="Dirty MAF sensor (cruise)", |
| category="Air system", |
| description="MAF underestimates airflow at cruise.", |
| signature="MAF low vs RPM/load, LTFT +, STFT +", |
| typical_dtc="P0101", |
| llm_hint="MAF reading low for RPM; dirty or failed MAF suspected.", |
| overrides={ |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(2500)}"'), |
| "ENGINE_LOAD": ax('ECU_R_ADDR_E + " 03 41 04 80"'), |
| "MAF": ax(f'ECU_R_ADDR_E + " 04 41 10 {maf_hex(16)}"'), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(12):02X}"' |
| ), |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(10):02X}"' |
| ), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 50"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="air_filter_restriction", |
| name="Air filter restriction", |
| category="Air system", |
| description="Blocked airflow — needs more throttle for same power.", |
| signature="Throttle ↑, MAF ↓, engine load ↓ at same RPM", |
| typical_dtc="—", |
| llm_hint="High throttle with low MAF and load suggests intake restriction.", |
| overrides={ |
| "THROTTLE_POS": ax('ECU_R_ADDR_E + " 03 41 11 A0"'), |
| "MAF": ax(f'ECU_R_ADDR_E + " 04 41 10 {maf_hex(35)}"'), |
| "ENGINE_LOAD": ax('ECU_R_ADDR_E + " 03 41 04 45"'), |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(2200)}"'), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 35"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="egr_stuck_open", |
| name="EGR stuck open (unstable idle)", |
| category="Air system", |
| description="Exhaust gas enters intake continuously.", |
| signature="Rough idle, RPM unstable, possible STFT +", |
| typical_dtc="P0401 / misfire codes", |
| llm_hint="Unstable idle with commanded EGR fully open.", |
| overrides={ |
| "COMMANDED_EGR": ax('ECU_R_ADDR_E + " 03 41 2C FF"'), |
| "RPM": ax( |
| f'ECU_R_ADDR_E + " 04 41 0C %.4X" % int(4 * {_randint_expr(600, 1100)})' |
| ), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(8):02X}"' |
| ), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 00"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="fuel_pump_weak_load", |
| name="Fuel pump weakening (under load)", |
| category="Fuel system", |
| description="Low fuel pressure under heavy acceleration.", |
| signature="Normal idle trims; under load STFT +20%, LTFT +15%", |
| typical_dtc="P0171", |
| llm_hint="Lean under load with high engine load and RPM.", |
| overrides={ |
| "ENGINE_LOAD": ax('ECU_R_ADDR_E + " 03 41 04 FF"'), |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(4500)}"'), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 90"'), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(20):02X}"' |
| ), |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(15):02X}"' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="dirty_injectors", |
| name="Dirty fuel injectors", |
| category="Fuel system", |
| description="Reduced spray — uneven combustion.", |
| signature="Positive trims, mild RPM instability", |
| typical_dtc="P0300", |
| llm_hint="Positive trims with unstable RPM at idle.", |
| overrides={ |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(14):02X}"' |
| ), |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(11):02X}"' |
| ), |
| "RPM": ax( |
| f'ECU_R_ADDR_E + " 04 41 0C %.4X" % int(4 * {_randint_expr(1100, 1350)})' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="injector_leak", |
| name="Leaking injector (rich)", |
| category="Fuel system", |
| description="Continuous fuel drip — rich condition.", |
| signature="STFT −, LTFT − (negative trims)", |
| typical_dtc="P0172", |
| llm_hint="Negative fuel trims indicate rich running.", |
| overrides={ |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(-18):02X}"' |
| ), |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(-12):02X}"' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="injector_stuck_closed", |
| name="Injector stuck closed (lean misfire)", |
| category="Fuel system", |
| description="One cylinder lean → misfire tendency.", |
| signature="Positive trims, RPM instability at idle", |
| typical_dtc="P0301–P0304", |
| llm_hint="Lean trims with unstable idle — possible cylinder misfire.", |
| overrides={ |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(16):02X}"' |
| ), |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(8):02X}"' |
| ), |
| "RPM": ax( |
| f'ECU_R_ADDR_E + " 04 41 0C %.4X" % int(4 * {_randint_expr(750, 1250)})' |
| ), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 00"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="ignition_coil_failure", |
| name="Ignition coil failure", |
| category="Ignition", |
| description="Weak spark on one cylinder.", |
| signature="RPM instability, O2 swings, trims may go positive", |
| typical_dtc="P0301", |
| llm_hint="High RPM std at idle with erratic O2 — misfire pattern.", |
| overrides={ |
| "RPM": ax( |
| f'ECU_R_ADDR_E + " 04 41 0C %.4X" % int(4 * {_randint_expr(500, 1500)})' |
| ), |
| "O2_B1S2": ax( |
| f'ECU_R_ADDR_E + " 03 41 14 %.2X" % {_randint_expr(10, 200)}' |
| ), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(6):02X}"' |
| ), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 00"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="spark_plug_wear", |
| name="Spark plug wear (mid stage)", |
| category="Ignition", |
| description="Gradual ignition efficiency loss.", |
| signature="Slowly increasing fuel trims, reduced performance", |
| typical_dtc="—", |
| llm_hint="Elevated LTFT/STFT without obvious air or fuel mechanical fault.", |
| overrides={ |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(9):02X}"' |
| ), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(6):02X}"' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="thermostat_stuck_open", |
| name="Thermostat stuck open", |
| category="Cooling", |
| description="Engine never reaches operating temperature.", |
| signature="Coolant temp low after long run (~65°C vs ~90°C expected)", |
| typical_dtc="P0128", |
| llm_hint="Engine not reaching operating temperature; stuck-open thermostat.", |
| overrides={ |
| "COOLANT_TEMP": ax(f'ECU_R_ADDR_E + " 03 41 05 {clt_hex(65)}"'), |
| "RUN_TIME": ax('ECU_R_ADDR_E + " 04 41 1F 0E 10"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="thermostat_stuck_closed", |
| name="Thermostat stuck closed (overheating)", |
| category="Cooling", |
| description="Coolant cannot circulate — overheating.", |
| signature="Coolant temp rapidly rising → 105–112°C", |
| typical_dtc="P0217", |
| llm_hint="High coolant temperature under moderate load.", |
| overrides={ |
| "COOLANT_TEMP": ax(f'ECU_R_ADDR_E + " 03 41 05 {clt_hex(112)}"'), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 25"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="cooling_fan_failure_idle", |
| name="Cooling fan failure (idle/traffic)", |
| category="Cooling", |
| description="Overheats at idle but normal while moving.", |
| signature="High coolant only when speed ≈ 0", |
| typical_dtc="P0217", |
| llm_hint="Overheating at idle with normal cruise temps in comparison sessions.", |
| overrides={ |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 00"'), |
| "COOLANT_TEMP": ax(f'ECU_R_ADDR_E + " 03 41 05 {clt_hex(108)}"'), |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(750)}"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="water_pump_degradation", |
| name="Water pump degradation (high load)", |
| category="Cooling", |
| description="Reduced circulation — gradual overheating under load.", |
| signature="Temp rising under high RPM/load", |
| typical_dtc="—", |
| llm_hint="Elevated coolant at high load vs healthy baseline at same RPM.", |
| overrides={ |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(3800)}"'), |
| "ENGINE_LOAD": ax('ECU_R_ADDR_E + " 03 41 04 C0"'), |
| "COOLANT_TEMP": ax(f'ECU_R_ADDR_E + " 03 41 05 {clt_hex(102)}"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="low_coolant", |
| name="Low coolant (unstable temp)", |
| category="Cooling", |
| description="Unstable temperature, overheating spikes.", |
| signature="High std on coolant temperature", |
| typical_dtc="—", |
| llm_hint="Erratic coolant readings — possible low coolant level.", |
| overrides={ |
| "COOLANT_TEMP": ax( |
| f'ECU_R_ADDR_E + " 03 41 05 %.2X" % int(85 + {_randint_expr(-8, 25)})' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="catalyst_degradation", |
| name="Catalytic converter degradation", |
| category="Emissions", |
| description="Downstream O2 oscillates like upstream (unhealthy).", |
| signature="Upstream and downstream O2 both oscillating", |
| typical_dtc="P0420", |
| llm_hint="Downstream O2 oscillation suggests catalyst efficiency loss.", |
| overrides={ |
| "O2_B1S2": ax( |
| f'ECU_R_ADDR_E + " 03 41 14 %.2X" % {_randint_expr(30, 170)}' |
| ), |
| "O2_S1_WR_VOLTAGE": ax( |
| f'ECU_R_ADDR_E + " 04 41 24 %.4X" % {_randint_expr(200, 3500)}' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="catalyst_restriction", |
| name="Catalytic converter restriction", |
| category="Emissions", |
| description="Clogged catalyst — low power, high exhaust backpressure.", |
| signature="High load, low MAF, poor acceleration, cat temp ↑", |
| typical_dtc="P0420", |
| llm_hint="High load with low MAF and elevated catalyst temperature.", |
| overrides={ |
| "ENGINE_LOAD": ax('ECU_R_ADDR_E + " 03 41 04 F0"'), |
| "MAF": ax(f'ECU_R_ADDR_E + " 04 41 10 {maf_hex(28)}"'), |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(2800)}"'), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 40"'), |
| "CATALYST_TEMP_B1S1": ax('ECU_R_ADDR_E + " 04 41 3C 23 28"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="evap_purge_stuck_open", |
| name="EVAP purge valve stuck open", |
| category="Emissions", |
| description="Fuel vapors enter intake at wrong time.", |
| signature="At idle: fuel trims positive, worse after refueling", |
| typical_dtc="P0441", |
| llm_hint="Positive trims at idle with purge valve fully open.", |
| overrides={ |
| "EVAPORATIVE_PURGE": ax('ECU_R_ADDR_E + " 03 41 2E FF"'), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(14):02X}"' |
| ), |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(6):02X}"' |
| ), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 00"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="o2_sensor_drift", |
| name="O2 sensor drift", |
| category="Sensors", |
| description="Sensor aging — trims drift without obvious mechanical fault.", |
| signature="LTFT slowly drifting, O2 less responsive", |
| typical_dtc="P0136", |
| llm_hint="Elevated LTFT with sluggish downstream O2.", |
| overrides={ |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(10):02X}"' |
| ), |
| "O2_B1S2": ax( |
| f'ECU_R_ADDR_E + " 03 41 14 %.2X" % (128 + {_randint_expr(-5, 5)})' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="o2_sensor_failure_low", |
| name="O2 sensor failure (stuck low)", |
| category="Sensors", |
| description="Sensor dead — flatlined low voltage.", |
| signature="O2 stuck at 0V continuously", |
| typical_dtc="P0136", |
| llm_hint="Flatlined O2 voltage — failed downstream sensor.", |
| overrides={"O2_B1S2": ax('ECU_R_ADDR_E + " 03 41 14 00"')}, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="o2_sensor_failure_high", |
| name="O2 sensor failure (stuck high)", |
| category="Sensors", |
| description="Sensor dead — flatlined high voltage.", |
| signature="O2 stuck high continuously", |
| typical_dtc="P0136", |
| llm_hint="O2 pinned high — failed downstream sensor.", |
| overrides={"O2_B1S2": ax('ECU_R_ADDR_E + " 03 41 14 C8"')}, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="iat_sensor_failure_cold", |
| name="IAT sensor failure (stuck cold)", |
| category="Sensors", |
| description="Implausible intake air temperature (-40°C).", |
| signature="IAT = -40°C", |
| typical_dtc="P0112", |
| llm_hint="Implausible IAT reading — sensor circuit low.", |
| overrides={"INTAKE_TEMP": ax('ECU_R_ADDR_E + " 03 41 0F 00"')}, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="ect_sensor_failure", |
| name="Coolant temp sensor failure", |
| category="Sensors", |
| description="Coolant reads -40°C when engine is warm → ECU runs rich.", |
| signature="COOLANT_TEMP implausible, rich trims", |
| typical_dtc="P0117", |
| llm_hint="Impossible ECT with rich compensation trims.", |
| overrides={ |
| "COOLANT_TEMP": ax('ECU_R_ADDR_E + " 03 41 05 00"'), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(-10):02X}"' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="map_sensor_failure", |
| name="MAP sensor failure", |
| category="Sensors", |
| description="Intake pressure inconsistent with RPM/throttle/MAF.", |
| signature="High RPM but vacuum reads like idle", |
| typical_dtc="P0106", |
| llm_hint="MAP does not match RPM and throttle — sensor performance fault.", |
| overrides={ |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(3000)}"'), |
| "THROTTLE_POS": ax('ECU_R_ADDR_E + " 03 41 11 70"'), |
| "INTAKE_PRESSURE": ax('ECU_R_ADDR_E + " 03 41 0B 95"'), |
| "MAF": ax(f'ECU_R_ADDR_E + " 04 41 10 {maf_hex(45)}"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="tps_failure", |
| name="Throttle position sensor failure", |
| category="Sensors", |
| description="RPM changes but throttle reading stuck.", |
| signature="TPS stuck while RPM varies", |
| typical_dtc="P0122 / P0123", |
| llm_hint="Throttle position does not track RPM changes.", |
| overrides={ |
| "THROTTLE_POS": ax('ECU_R_ADDR_E + " 03 41 11 20"'), |
| "RPM": ax( |
| f'ECU_R_ADDR_E + " 04 41 0C %.4X" % int(4 * {_randint_expr(900, 2500)})' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="alternator_failure", |
| name="Alternator failure", |
| category="Electrical", |
| description="Charging system cannot maintain voltage.", |
| signature="Module voltage low while running (~12.7 V)", |
| typical_dtc="P0562", |
| llm_hint="Low control-module voltage while engine running.", |
| overrides={ |
| "CONTROL_MODULE_VOLTAGE": ax( |
| f'ECU_R_ADDR_E + " 04 41 42 {volt_hex(12.7)}"' |
| ), |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(1100)}"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="voltage_regulator_failure", |
| name="Voltage regulator failure", |
| category="Electrical", |
| description="Overcharging while running.", |
| signature="15 V+ while running", |
| typical_dtc="P0563", |
| llm_hint="System voltage dangerously high — regulator fault.", |
| overrides={ |
| "CONTROL_MODULE_VOLTAGE": ax( |
| f'ECU_R_ADDR_E + " 04 41 42 {volt_hex(15.8)}"' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="compression_loss", |
| name="Engine compression loss", |
| category="Mechanical", |
| description="Ring/valve wear — indirect OBD signs only.", |
| signature="Abnormal trims, misfires, poor power", |
| typical_dtc="P0300", |
| llm_hint="High throttle with low MAF/load and unstable RPM.", |
| overrides={ |
| "RPM": ax( |
| f'ECU_R_ADDR_E + " 04 41 0C %.4X" % int(4 * {_randint_expr(700, 1300)})' |
| ), |
| "ENGINE_LOAD": ax('ECU_R_ADDR_E + " 03 41 04 30"'), |
| "THROTTLE_POS": ax('ECU_R_ADDR_E + " 03 41 11 C0"'), |
| "MAF": ax(f'ECU_R_ADDR_E + " 04 41 10 {maf_hex(22)}"'), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(10):02X}"' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="head_gasket_leak", |
| name="Head gasket leak", |
| category="Mechanical", |
| description="Coolant unstable, misfires, overheating tendency.", |
| signature="Unstable coolant, misfires, overheating", |
| typical_dtc="P0300", |
| llm_hint="Combined overheating and misfire pattern.", |
| overrides={ |
| "COOLANT_TEMP": ax( |
| f'ECU_R_ADDR_E + " 03 41 05 %.2X" % int(95 + {_randint_expr(0, 20)})' |
| ), |
| "RPM": ax( |
| f'ECU_R_ADDR_E + " 04 41 0C %.4X" % int(4 * {_randint_expr(650, 1200)})' |
| ), |
| "SHORT_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 06 {trim_b(8):02X}"' |
| ), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="timing_chain_stretch", |
| name="Timing chain stretch", |
| category="Mechanical", |
| description="Subtle performance and timing anomalies.", |
| signature="Reduced performance, timing advance low, LTFT +", |
| typical_dtc="—", |
| llm_hint="Subtle timing and trim anomalies at cruise — compare to baseline.", |
| overrides={ |
| "TIMING_ADVANCE": ax('ECU_R_ADDR_E + " 03 41 0E 18"'), |
| "LONG_FUEL_TRIM_1": ax( |
| f'ECU_R_ADDR_E + " 03 41 07 {trim_b(7):02X}"' |
| ), |
| "ENGINE_LOAD": ax('ECU_R_ADDR_E + " 03 41 04 55"'), |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(2400)}"'), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 55"'), |
| }, |
| ) |
| ) |
|
|
| add( |
| FaultProfile( |
| id="transmission_slip", |
| name="Transmission slip (proxy)", |
| category="Drivetrain", |
| description="High RPM with low speed increase — limited OBD-II proxy.", |
| signature="High RPM, low speed, moderate load", |
| typical_dtc="—", |
| llm_hint="RPM high relative to speed — possible slip (proxy signal only).", |
| overrides={ |
| "RPM": ax(f'ECU_R_ADDR_E + " 04 41 0C {rpm_hex(4200)}"'), |
| "SPEED": ax('ECU_R_ADDR_E + " 03 41 0D 25"'), |
| "ENGINE_LOAD": ax('ECU_R_ADDR_E + " 03 41 04 90"'), |
| }, |
| ) |
| ) |
|
|
| return faults |
|
|
|
|
| FAULTS: dict[str, FaultProfile] = _build_faults() |
|
|
|
|
| def list_fault_choices() -> list[tuple[str, str]]: |
| """Return Gradio dropdown choices grouped by category.""" |
| by_category: dict[str, list[tuple[str, str]]] = {} |
| for fault in FAULTS.values(): |
| label = f"{fault.category}: {fault.name}" |
| by_category.setdefault(fault.category, []).append((label, fault.id)) |
|
|
| choices: list[tuple[str, str]] = [("Healthy (no fault)", "none")] |
| for category in sorted(by_category): |
| for label, fault_id in sorted(by_category[category], key=lambda x: x[0]): |
| choices.append((label, fault_id)) |
| return choices |
|
|
|
|
| def get_fault(fault_id: str | None) -> FaultProfile | None: |
| if not fault_id or fault_id == "none": |
| return None |
| return FAULTS.get(fault_id) |
|
|
|
|
| def get_active_fault_id() -> str | None: |
| with _state_lock: |
| return _active_fault_id |
|
|
|
|
| def get_active_fault() -> FaultProfile | None: |
| return get_fault(get_active_fault_id()) |
|
|
|
|
| def describe_fault(fault_id: str | None) -> str: |
| fault = get_fault(fault_id) |
| if fault is None: |
| return ( |
| "**Healthy baseline** — no PID overrides. " |
| "The emulator uses the default `car` scenario values." |
| ) |
|
|
| return ( |
| f"### {fault.name}\n\n" |
| f"{fault.description}\n\n" |
| f"**OBD signature:** {fault.signature}\n\n" |
| f"**Typical DTC:** `{fault.typical_dtc}`\n\n" |
| f"**Hint:** {fault.llm_hint}\n\n" |
| f"**Overrides:** {len(fault.overrides)} PID(s)" |
| + (f" + GET_DTC" if parse_dtc_codes(fault.typical_dtc) else "") |
| + " — " |
| + ", ".join(f"`{k}`" for k in fault.overrides) |
| + (", `GET_DTC`" if parse_dtc_codes(fault.typical_dtc) else "") |
| ) |
|
|
|
|
| FAULT_INVESTIGATION_PROMPT = ( |
| "Something seems wrong with my vehicle. " |
| "Please check live OBD vitals, recent stored snapshots, and any stored trouble codes " |
| "to figure out what's going on and recommend what I should do next." |
| ) |
|
|
|
|
| def fault_investigation_prompt() -> str: |
| """Generic driver message — agent must discover the fault via OBD tools.""" |
| return FAULT_INVESTIGATION_PROMPT |
|
|
|
|
| def _clear_emulator_overrides(emulator: Any) -> None: |
| """Drop PID/DTC overrides; scenario defaults apply without a full reset.""" |
| emulator.answer.clear() |
|
|
|
|
| def _apply_overrides(emulator: Any, overrides: dict[str, str]) -> None: |
| _clear_emulator_overrides(emulator) |
| for pid, answer_xml in overrides.items(): |
| emulator.answer[pid] = answer_xml |
|
|
|
|
| def _require_emulator() -> Any: |
| """Return the in-process emulator, starting it only if missing.""" |
| emulator = get_emulator() |
| if emulator is not None: |
| return emulator |
| ensure_elm_emulator() |
| emulator = get_emulator() |
| if emulator is None: |
| raise RuntimeError("ELM327 emulator is not ready") |
| return emulator |
|
|
|
|
| def apply_fault(fault_id: str) -> str: |
| """Apply a fault profile to the running emulator.""" |
| global _active_fault_id |
|
|
| if fault_id == "none": |
| return clear_faults() |
|
|
| fault = FAULTS.get(fault_id) |
| if fault is None: |
| raise ValueError(f"Unknown fault profile: {fault_id}") |
|
|
| emulator = _require_emulator() |
|
|
| with obd_session_lock: |
| with _emulator_lock: |
| _apply_overrides(emulator, _effective_overrides(fault)) |
|
|
| with _state_lock: |
| _active_fault_id = fault_id |
|
|
| logger.info("Applied fault simulation: %s (%s)", fault.id, fault.name) |
| return ( |
| f"**Active fault:** {fault.name} (`{fault.id}`)\n\n" |
| f"Typical DTC: `{fault.typical_dtc}`. Vitals refreshed and stored." |
| ) |
|
|
|
|
| def clear_faults() -> str: |
| """Reset emulator overrides and return to healthy car scenario.""" |
| global _active_fault_id |
|
|
| emulator = _require_emulator() |
|
|
| with obd_session_lock: |
| with _emulator_lock: |
| _clear_emulator_overrides(emulator) |
|
|
| with _state_lock: |
| _active_fault_id = None |
|
|
| logger.info("Cleared fault simulation overrides") |
| return "**Healthy baseline** — all PID overrides cleared." |
|
|
