mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-09-01 05:33:49 +08:00
OBD
This commit is contained in:
@@ -3,13 +3,13 @@
|
||||
"id": "Chevrolet-Bolt-EV",
|
||||
"name": "Chevrolet Bolt EV / EUV",
|
||||
"provider": "OBDb",
|
||||
"revision": "v3.1.0",
|
||||
"revision": "v3.2.0",
|
||||
"protocol": "ISO 15765-4 (CAN 11/500)",
|
||||
"description": "OEM Mode 22 High-Voltage Battery & Inverter definitions for Chevrolet Bolt EV"
|
||||
},
|
||||
"commands": [
|
||||
{
|
||||
"id": "CMD_BOLT_BMS_MAIN",
|
||||
"id": "CMD_BOLT_SOC",
|
||||
"hdr": "7E4",
|
||||
"rax": "7EC",
|
||||
"service": 34,
|
||||
@@ -24,44 +24,74 @@
|
||||
"fmt": {
|
||||
"bix": 0,
|
||||
"len": 8,
|
||||
"mul": 0.5,
|
||||
"div": 1.0,
|
||||
"mul": 100.0,
|
||||
"div": 255.0,
|
||||
"unit": "%"
|
||||
}
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "CMD_BOLT_VOLTAGE",
|
||||
"hdr": "7E4",
|
||||
"rax": "7EC",
|
||||
"service": 34,
|
||||
"pid": "41A3",
|
||||
"freq": 2.0,
|
||||
"signals": [
|
||||
{
|
||||
"id": "BOLT_HVBAT_VOLTAGE",
|
||||
"name": "HV Battery Pack Voltage",
|
||||
"path": "Battery",
|
||||
"suggested_metric": "batteryVoltage",
|
||||
"fmt": {
|
||||
"bix": 16,
|
||||
"bix": 0,
|
||||
"len": 16,
|
||||
"mul": 0.05,
|
||||
"mul": 0.01,
|
||||
"div": 1.0,
|
||||
"unit": "V"
|
||||
}
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "CMD_BOLT_CURRENT",
|
||||
"hdr": "7E4",
|
||||
"rax": "7EC",
|
||||
"service": 34,
|
||||
"pid": "2409",
|
||||
"freq": 2.0,
|
||||
"signals": [
|
||||
{
|
||||
"id": "BOLT_HVBAT_CURRENT",
|
||||
"name": "HV Battery Pack Current",
|
||||
"path": "Battery",
|
||||
"suggested_metric": "batteryCurrent",
|
||||
"fmt": {
|
||||
"bix": 32,
|
||||
"bix": 0,
|
||||
"len": 16,
|
||||
"sign": true,
|
||||
"mul": 0.05,
|
||||
"div": 1.0,
|
||||
"unit": "A"
|
||||
}
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "CMD_BOLT_TEMP",
|
||||
"hdr": "7E4",
|
||||
"rax": "7EC",
|
||||
"service": 34,
|
||||
"pid": "41A6",
|
||||
"freq": 1.0,
|
||||
"signals": [
|
||||
{
|
||||
"id": "BOLT_HVBAT_TEMP",
|
||||
"name": "HV Battery Average Temperature",
|
||||
"path": "Battery",
|
||||
"fmt": {
|
||||
"bix": 48,
|
||||
"bix": 0,
|
||||
"len": 8,
|
||||
"mul": 1.0,
|
||||
"div": 1.0,
|
||||
@@ -72,11 +102,11 @@
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "CMD_BOLT_MOTOR",
|
||||
"id": "CMD_BOLT_MOTOR_SPEED",
|
||||
"hdr": "7E2",
|
||||
"rax": "7EA",
|
||||
"service": 34,
|
||||
"pid": "4080",
|
||||
"pid": "0038",
|
||||
"freq": 5.0,
|
||||
"signals": [
|
||||
{
|
||||
@@ -91,13 +121,23 @@
|
||||
"div": 1.0,
|
||||
"unit": "rpm"
|
||||
}
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "CMD_BOLT_MOTOR_TEMP",
|
||||
"hdr": "7E2",
|
||||
"rax": "7EA",
|
||||
"service": 34,
|
||||
"pid": "4084",
|
||||
"freq": 2.0,
|
||||
"signals": [
|
||||
{
|
||||
"id": "BOLT_MOTOR_TEMP",
|
||||
"name": "Motor Temperature",
|
||||
"path": "Motor",
|
||||
"fmt": {
|
||||
"bix": 16,
|
||||
"bix": 0,
|
||||
"len": 8,
|
||||
"mul": 1.0,
|
||||
"div": 1.0,
|
||||
|
||||
@@ -177,31 +177,41 @@ def parse_uds_dtcs(data: bytes, ecu: str | None = None) -> list[DiagnosticTroubl
|
||||
|
||||
# Standard OBD-II Functions
|
||||
|
||||
from openpilot.starpilot.system.obdyssey.elm327 import ElmContext, ElmNoDataError
|
||||
|
||||
DEFAULT_OBD_CONTEXT = ElmContext(tx_header=0x7DF, rx_filter=0x7E8)
|
||||
|
||||
|
||||
def read_current_data(elm: Elm327, pid: int, context: ElmContext | None = None) -> DiagnosticResponse:
|
||||
"""Mode 01: Read current powertrain diagnostic data."""
|
||||
return elm.request(bytes([0x01, pid & 0xFF]), context, retry=True)
|
||||
ctx = context if context is not None else DEFAULT_OBD_CONTEXT
|
||||
return elm.request(bytes([0x01, pid & 0xFF]), ctx, retry=True)
|
||||
|
||||
|
||||
def read_freeze_frame(elm: Elm327, pid: int, frame: int = 0, context: ElmContext | None = None) -> DiagnosticResponse:
|
||||
"""Mode 02: Read freeze frame data."""
|
||||
return elm.request(bytes([0x02, pid & 0xFF, frame & 0xFF]), context, retry=True)
|
||||
ctx = context if context is not None else DEFAULT_OBD_CONTEXT
|
||||
return elm.request(bytes([0x02, pid & 0xFF, frame & 0xFF]), ctx, retry=True)
|
||||
|
||||
|
||||
def read_stored_dtcs(elm: Elm327, context: ElmContext | None = None) -> list[DiagnosticTroubleCode]:
|
||||
"""Mode 03: Read confirmed/stored emission-related DTCs."""
|
||||
res = elm.request(bytes([0x03]), context, retry=True)
|
||||
ctx = context if context is not None else DEFAULT_OBD_CONTEXT
|
||||
res = elm.request(bytes([0x03]), ctx, retry=True)
|
||||
return parse_standard_dtcs(res.payload, source="OBD_STORED")
|
||||
|
||||
|
||||
def read_pending_dtcs(elm: Elm327, context: ElmContext | None = None) -> list[DiagnosticTroubleCode]:
|
||||
"""Mode 07: Read pending DTCs detected during current/last drive cycle."""
|
||||
res = elm.request(bytes([0x07]), context, retry=True)
|
||||
ctx = context if context is not None else DEFAULT_OBD_CONTEXT
|
||||
res = elm.request(bytes([0x07]), ctx, retry=True)
|
||||
return parse_standard_dtcs(res.payload, source="OBD_PENDING")
|
||||
|
||||
|
||||
def read_permanent_dtcs(elm: Elm327, context: ElmContext | None = None) -> list[DiagnosticTroubleCode]:
|
||||
"""Mode 0A: Read permanent DTCs."""
|
||||
res = elm.request(bytes([0x0A]), context, retry=True)
|
||||
ctx = context if context is not None else DEFAULT_OBD_CONTEXT
|
||||
res = elm.request(bytes([0x0A]), ctx, retry=True)
|
||||
return parse_standard_dtcs(res.payload, source="OBD_PERMANENT")
|
||||
|
||||
|
||||
@@ -218,12 +228,19 @@ def read_all_dtcs(elm: Elm327, context: ElmContext | None = None) -> list[Diagno
|
||||
|
||||
def clear_dtcs(elm: Elm327, context: ElmContext | None = None) -> DiagnosticResponse:
|
||||
"""Mode 04: Clear diagnostic trouble codes and reset MIL (Check Engine Light). Mutating!"""
|
||||
return elm.request(bytes([0x04]), context, retry=False)
|
||||
ctx = context if context is not None else DEFAULT_OBD_CONTEXT
|
||||
try:
|
||||
return elm.request(bytes([0x04]), ctx, retry=False)
|
||||
except ElmNoDataError:
|
||||
# Modern EV and hybrid architectures without standard emissions ECUs do not respond
|
||||
# to Mode 04 on 7DF. Broadcast UDS Service 0x14 (ClearDiagnosticInformation).
|
||||
return elm.request(bytes([SERVICE_TYPE.CLEAR_DIAGNOSTIC_INFORMATION, 0xFF, 0xFF, 0xFF]), ctx, retry=False)
|
||||
|
||||
|
||||
def read_vin(elm: Elm327, context: ElmContext | None = None) -> str:
|
||||
"""Mode 09 PID 02: Read Vehicle Identification Number (VIN)."""
|
||||
res = elm.request(bytes([0x09, 0x02]), context, retry=True)
|
||||
ctx = context if context is not None else DEFAULT_OBD_CONTEXT
|
||||
res = elm.request(bytes([0x09, 0x02]), ctx, retry=True)
|
||||
payload = res.payload
|
||||
|
||||
# Response format: 49 02 <data_items_or_line_nums> followed by ASCII VIN characters
|
||||
|
||||
@@ -49,18 +49,39 @@ def test_resolve_active_profile_from_params(tmp_path):
|
||||
|
||||
def test_group_signals_by_command(tmp_path):
|
||||
manager = ProfileManager(data_dir=tmp_path)
|
||||
bolt = manager.get_profile("Chevrolet-Bolt-EV")
|
||||
custom_data = {
|
||||
"metadata": {"name": "Multi-Signal Profile"},
|
||||
"commands": [
|
||||
{
|
||||
"id": "CMD_MULTI_1",
|
||||
"service": 1,
|
||||
"pid": "01",
|
||||
"signals": [
|
||||
{"id": "SIG_A", "name": "Signal A"},
|
||||
{"id": "SIG_B", "name": "Signal B"},
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "CMD_MULTI_2",
|
||||
"service": 1,
|
||||
"pid": "02",
|
||||
"signals": [
|
||||
{"id": "SIG_C", "name": "Signal C"},
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
profile = manager.install_profile_data("multi_test", custom_data)
|
||||
requested = ["SIG_A", "SIG_B", "SIG_C"]
|
||||
grouped = ProfileManager.group_signals_by_command(profile, requested)
|
||||
|
||||
# Request 3 signals that belong to the SAME BMS command (CMD_BOLT_BMS_MAIN)
|
||||
requested = ["BOLT_HVBAT_SOC", "BOLT_HVBAT_VOLTAGE", "BOLT_HVBAT_CURRENT"]
|
||||
grouped = ProfileManager.group_signals_by_command(bolt, requested)
|
||||
|
||||
# Must produce exactly ONE command group containing all 3 signals!
|
||||
assert len(grouped) == 1
|
||||
cmd, signals = grouped[0]
|
||||
assert cmd.id == "CMD_BOLT_BMS_MAIN"
|
||||
assert len(signals) == 3
|
||||
assert [s.id for s in signals] == requested
|
||||
assert len(grouped) == 2
|
||||
assert grouped[0][0].id == "CMD_MULTI_1"
|
||||
assert len(grouped[0][1]) == 2
|
||||
assert [s.id for s in grouped[0][1]] == ["SIG_A", "SIG_B"]
|
||||
assert grouped[1][0].id == "CMD_MULTI_2"
|
||||
assert len(grouped[1][1]) == 1
|
||||
assert [s.id for s in grouped[1][1]] == ["SIG_C"]
|
||||
|
||||
|
||||
def test_install_and_remove_custom_profile(tmp_path):
|
||||
|
||||
@@ -184,12 +184,13 @@ class OBDysseyScreen(Widget):
|
||||
threading.Thread(target=_task, daemon=True).start()
|
||||
|
||||
def _worker_loop(self):
|
||||
# 1. Fetch status and available signals once upon entry. DTC scans are
|
||||
# explicitly user-triggered so an empty, unscanned result is never shown
|
||||
# as a clean vehicle.
|
||||
# 1. Fetch status and available signals once upon entry. Auto-connect if adapter is idle.
|
||||
try:
|
||||
self._status = self._client.status()
|
||||
self._available_signals = self._client.list_signals()
|
||||
if not self._diagnostic_ready() and (not self._status or self._status.state in ("idle", "disabled")):
|
||||
self._client.connect()
|
||||
self._status = self._client.status()
|
||||
except Exception as err:
|
||||
self._last_error = str(err)
|
||||
|
||||
@@ -304,16 +305,29 @@ class OBDysseyScreen(Widget):
|
||||
self._dtc_button.render(rl.Rectangle(cur_x, btn_y, 240, BUTTON_HEIGHT))
|
||||
|
||||
def _render_content(self, rect: rl.Rectangle):
|
||||
# Total content height calculation
|
||||
telemetry_items = list(self._live_telemetry.items())
|
||||
num_cards = len(telemetry_items)
|
||||
# Build list of signals to display
|
||||
signals_to_render: list[tuple[str, Any, str, str]] = []
|
||||
if self._available_signals:
|
||||
for sig in self._available_signals:
|
||||
sig_id = sig["id"]
|
||||
val = self._live_telemetry.get(sig_id, None)
|
||||
common_info = COMMON_SIGNAL_NAMES.get(sig_id, (sig.get("name") or sig_id.replace("SAE_", "").replace("BOLT_", "").replace("_", " ").title(), sig.get("unit") or ""))
|
||||
name = sig.get("name") or common_info[0]
|
||||
unit = sig.get("unit") or common_info[1]
|
||||
signals_to_render.append((sig_id, val, name, unit))
|
||||
elif self._live_telemetry:
|
||||
for sig_id, val in self._live_telemetry.items():
|
||||
name, unit = COMMON_SIGNAL_NAMES.get(sig_id, (sig_id.replace("SAE_", "").replace("BOLT_", "").replace("_", " ").title(), ""))
|
||||
signals_to_render.append((sig_id, val, name, unit))
|
||||
|
||||
num_cards = len(signals_to_render)
|
||||
cols = 3
|
||||
card_margin = 25
|
||||
card_width = (rect.width - 2 * HEADER_PADDING - (cols - 1) * card_margin) / cols
|
||||
card_height = 150
|
||||
rows_count = (num_cards + cols - 1) // cols
|
||||
|
||||
telemetry_height = rows_count * (card_height + card_margin) + 70 if num_cards > 0 else 0
|
||||
telemetry_height = rows_count * (card_height + card_margin) + 70 if num_cards > 0 else 140
|
||||
dtc_height = 120 + max(1, len(self._dtcs)) * 140
|
||||
total_height = max(rect.height + 10, telemetry_height + dtc_height + 100)
|
||||
|
||||
@@ -324,12 +338,12 @@ class OBDysseyScreen(Widget):
|
||||
cur_y = rect.y + HEADER_PADDING + offset
|
||||
|
||||
# Section 1: Live Telemetry Grid
|
||||
if telemetry_items:
|
||||
gui_label(rl.Rectangle(rect.x + HEADER_PADDING, cur_y, rect.width - 2 * HEADER_PADDING, 50),
|
||||
tr("Live Sensor Telemetry"), font_size=46, font_weight=FontWeight.BOLD)
|
||||
cur_y += 65
|
||||
gui_label(rl.Rectangle(rect.x + HEADER_PADDING, cur_y, rect.width - 2 * HEADER_PADDING, 50),
|
||||
tr("Live Sensor Telemetry"), font_size=46, font_weight=FontWeight.BOLD)
|
||||
cur_y += 65
|
||||
|
||||
for idx, (sig_id, val) in enumerate(telemetry_items):
|
||||
if signals_to_render:
|
||||
for idx, (sig_id, val, name, unit) in enumerate(signals_to_render):
|
||||
col = idx % cols
|
||||
row = idx // cols
|
||||
card_x = rect.x + HEADER_PADDING + col * (card_width + card_margin)
|
||||
@@ -339,8 +353,7 @@ class OBDysseyScreen(Widget):
|
||||
if rl.check_collision_recs(card_rect, rect):
|
||||
rl.draw_rectangle_rounded(card_rect, 0.12, 16, CARD_BACKGROUND)
|
||||
|
||||
# Name & Unit
|
||||
name, default_unit = COMMON_SIGNAL_NAMES.get(sig_id, (sig_id.replace("SAE_", "").replace("BOLT_", "").replace("_", " ").title(), ""))
|
||||
# Name
|
||||
gui_label(rl.Rectangle(card_x + 20, card_y + 15, card_width - 40, 36), name, font_size=32, color=TEXT_SECONDARY)
|
||||
|
||||
# Value
|
||||
@@ -349,8 +362,8 @@ class OBDysseyScreen(Widget):
|
||||
val_color = TEXT_SECONDARY
|
||||
elif isinstance(val, (int, float)):
|
||||
val_str = f"{val:,.1f}" if isinstance(val, float) and not val.is_integer() else f"{int(val):,}"
|
||||
if default_unit:
|
||||
val_str += f" {default_unit}"
|
||||
if unit:
|
||||
val_str += f" {unit}"
|
||||
val_color = TEXT_CONNECTED
|
||||
else:
|
||||
val_str = str(val)
|
||||
@@ -359,6 +372,14 @@ class OBDysseyScreen(Widget):
|
||||
gui_label(rl.Rectangle(card_x + 20, card_y + 60, card_width - 40, 65), val_str, font_size=52, font_weight=FontWeight.BOLD, color=val_color)
|
||||
|
||||
cur_y += rows_count * (card_height + card_margin) + 30
|
||||
else:
|
||||
state_rect = rl.Rectangle(rect.x + HEADER_PADDING, cur_y, rect.width - 2 * HEADER_PADDING, 100)
|
||||
if rl.check_collision_recs(state_rect, rect):
|
||||
rl.draw_rectangle_rounded(state_rect, 0.12, 16, CARD_BACKGROUND)
|
||||
msg = tr("Connecting to adapter to load sensor telemetry...") if (self._status and self._status.state in ("connecting", "reconnecting", "initializing")) else tr("Connect to adapter to view live sensor telemetry.")
|
||||
gui_label(rl.Rectangle(state_rect.x + 30, state_rect.y + 30, state_rect.width - 60, 40),
|
||||
msg, font_size=36, color=TEXT_SECONDARY)
|
||||
cur_y += 130
|
||||
|
||||
# Section 2: DTC Fault Codes
|
||||
gui_label(rl.Rectangle(rect.x + HEADER_PADDING, cur_y, rect.width - 2 * HEADER_PADDING, 50),
|
||||
|
||||
Reference in New Issue
Block a user