From f1078542278b9bcc5cc7c3d73a8095947b6d2ad3 Mon Sep 17 00:00:00 2001 From: virtuscyber Date: Sat, 11 Apr 2026 11:22:31 -0400 Subject: [PATCH] feat(obd_ai): add read-only diagnostic tool surface --- obd_ai/__init__.py | 2 + obd_ai/catalog.py | 7 + obd_ai/tools.py | 394 +++++++++++++++++++++++++++++++++++++ tests/test_obd_ai_tools.py | 215 ++++++++++++++++++++ 4 files changed, 618 insertions(+) create mode 100644 obd_ai/tools.py create mode 100644 tests/test_obd_ai_tools.py diff --git a/obd_ai/__init__.py b/obd_ai/__init__.py index 0ea3a272..ddfd6013 100644 --- a/obd_ai/__init__.py +++ b/obd_ai/__init__.py @@ -14,6 +14,7 @@ serialize_value, ) from .session import OBDAISession, OBDAISessionManager +from .tools import OBDAIReadOnlyToolSurface __all__ = [ "ApprovedCommand", @@ -21,6 +22,7 @@ "DEFAULT_APPROVED_COMMAND_CATALOG", "OBDAISession", "OBDAISessionManager", + "OBDAIReadOnlyToolSurface", "serialize_approved_command", "serialize_approved_command_catalog", "serialize_connection_metadata", diff --git a/obd_ai/catalog.py b/obd_ai/catalog.py index b3ad7e72..2a79ed65 100644 --- a/obd_ai/catalog.py +++ b/obd_ai/catalog.py @@ -171,6 +171,13 @@ def as_mapping(self) -> Mapping[str, ApprovedCommand]: description="Read pending trouble codes from current drive cycle.", obd_command_name="GET_CURRENT_DTC", ), + _CommandDefinition( + key="freeze_frame_trouble_code", + public_name="Freeze Frame Trouble Code", + category="diagnostics", + description="Read the DTC that triggered freeze-frame capture.", + obd_command_name="FREEZE_DTC", + ), _CommandDefinition( key="vehicle_identification_number", public_name="Vehicle Identification Number", diff --git a/obd_ai/tools.py b/obd_ai/tools.py new file mode 100644 index 00000000..06f57d2e --- /dev/null +++ b/obd_ai/tools.py @@ -0,0 +1,394 @@ +"""Read-only tool surface for safe LLM/MCP-style OBD diagnostics.""" + +from __future__ import annotations + +from typing import Any, Dict, Mapping, Optional, Sequence, Tuple + +from .catalog import ApprovedCommandCatalog, DEFAULT_APPROVED_COMMAND_CATALOG +from .serializers import serialize_approved_command_catalog, serialize_obd_response +from .session import OBDAISession, OBDAISessionManager + +try: + from serial import SerialTimeoutException # type: ignore +except Exception: # pragma: no cover - pyserial should be present, but keep import safe + SerialTimeoutException = () # type: ignore[assignment] + + +class OBDAIReadOnlyToolSurface: + """Constrained, read-only tool interface over an ``OBDAISession``.""" + + def __init__( + self, + session_manager: Optional[OBDAISessionManager] = None, + catalog: Optional[ApprovedCommandCatalog] = None, + ): + resolved_catalog = catalog or DEFAULT_APPROVED_COMMAND_CATALOG + self._session_manager = session_manager or OBDAISessionManager( + catalog=resolved_catalog + ) + self._catalog = getattr(self._session_manager, "_catalog", resolved_catalog) + self._session: Optional[OBDAISession] = None + + def connect_vehicle(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + """Open a vehicle session and return connection metadata.""" + + payload = dict(input or {}) + allowed_fields = { + "portstr", + "baudrate", + "protocol", + "fast", + "timeout", + "check_voltage", + "start_low_power", + } + + unknown_fields = sorted(set(payload.keys()) - allowed_fields) + if unknown_fields: + return self._error( + tool="connect_vehicle", + code="invalid_input", + message="Unsupported connect_vehicle input fields.", + details={"unknown_fields": unknown_fields}, + ) + + self.close() + self._session = self._session_manager.open_session(**payload) + + metadata = self._session.connection_metadata() + if not metadata["is_connected"]: + return self._error( + tool="connect_vehicle", + code="disconnected_adapter", + message="OBD adapter is not connected to a vehicle.", + details={"connection": metadata}, + ) + + return self._success("connect_vehicle", {"connection": metadata}) + + def get_vehicle_status(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + """Return transport-level vehicle connection status.""" + + del input + session, error = self._require_connected(tool="get_vehicle_status") + if error is not None: + return error + + return self._success( + "get_vehicle_status", + {"connection": session.connection_metadata()}, + ) + + def list_supported_commands( + self, + input: Optional[Mapping[str, Any]] = None, + ) -> Dict[str, Any]: + """List approved commands, optionally including adapter support status.""" + + payload = dict(input or {}) + include_support = bool(payload.get("include_support", True)) + only_supported = bool(payload.get("only_supported", False)) + + if only_supported and not include_support: + return self._error( + tool="list_supported_commands", + code="invalid_input", + message="only_supported=true requires include_support=true.", + ) + + if include_support: + session, error = self._require_connected(tool="list_supported_commands") + if error is not None: + return error + commands = session.list_approved_command_descriptors(include_support=True) + else: + commands = serialize_approved_command_catalog(self._catalog) + + if only_supported: + commands = [command for command in commands if command.get("supported")] + + return self._success( + "list_supported_commands", + { + "commands": commands, + "count": len(commands), + }, + ) + + def read_sensor(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + """Read one approved sensor key.""" + + payload = dict(input or {}) + command_key = payload.get("command_key") + + if not isinstance(command_key, str) or not command_key: + return self._error( + tool="read_sensor", + code="invalid_input", + message="read_sensor requires a non-empty string command_key.", + ) + + response, error = self._query_approved_command( + tool="read_sensor", + command_key=command_key, + ) + if error is not None: + return error + + return self._success( + "read_sensor", + { + "command_key": command_key, + "response": response, + }, + ) + + def read_sensors(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + """Read multiple approved sensor keys with per-key status.""" + + payload = dict(input or {}) + command_keys = payload.get("command_keys") + + if not isinstance(command_keys, Sequence) or isinstance(command_keys, (str, bytes)): + return self._error( + tool="read_sensors", + code="invalid_input", + message="read_sensors requires command_keys as a list of strings.", + ) + + results = [] + has_errors = False + + for command_key in command_keys: + if not isinstance(command_key, str) or not command_key: + has_errors = True + results.append( + { + "command_key": command_key, + "ok": False, + "error": { + "code": "invalid_input", + "message": "Each command key must be a non-empty string.", + "details": {}, + }, + } + ) + continue + + response, error = self._query_approved_command( + tool="read_sensors", + command_key=command_key, + ) + if error is not None: + has_errors = True + results.append( + { + "command_key": command_key, + "ok": False, + "error": error["error"], + } + ) + continue + + results.append( + { + "command_key": command_key, + "ok": True, + "response": response, + } + ) + + return { + "ok": not has_errors, + "tool": "read_sensors", + "data": { + "results": results, + "count": len(results), + }, + } + + def get_dtc_codes(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + """Read stored and pending DTC code lists.""" + + del input + return self._multi_read( + tool="get_dtc_codes", + reads={ + "stored": "stored_trouble_codes", + "pending": "pending_trouble_codes", + }, + ) + + def get_freeze_frame(self, input: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: + """Read freeze-frame trigger code (DTC).""" + + del input + response, error = self._query_approved_command( + tool="get_freeze_frame", + command_key="freeze_frame_trouble_code", + ) + if error is not None: + return error + + return self._success( + "get_freeze_frame", + { + "command_key": "freeze_frame_trouble_code", + "response": response, + }, + ) + + def get_emissions_monitor_status( + self, + input: Optional[Mapping[str, Any]] = None, + ) -> Dict[str, Any]: + """Read emissions monitor status (mode 01 PID 01).""" + + del input + response, error = self._query_approved_command( + tool="get_emissions_monitor_status", + command_key="status_since_dtc_clear", + ) + if error is not None: + return error + + return self._success( + "get_emissions_monitor_status", + { + "command_key": "status_since_dtc_clear", + "response": response, + }, + ) + + def close(self) -> None: + """Close the active session if one exists.""" + + if self._session is not None: + self._session.close() + self._session = None + + def _multi_read(self, tool: str, reads: Mapping[str, str]) -> Dict[str, Any]: + results: Dict[str, Any] = {} + has_errors = False + + for label, command_key in reads.items(): + response, error = self._query_approved_command(tool=tool, command_key=command_key) + if error is not None: + has_errors = True + results[label] = { + "ok": False, + "command_key": command_key, + "error": error["error"], + } + else: + results[label] = { + "ok": True, + "command_key": command_key, + "response": response, + } + + return { + "ok": not has_errors, + "tool": tool, + "data": results, + } + + def _query_approved_command( + self, + tool: str, + command_key: str, + ) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]: + if command_key not in self._catalog: + return None, self._error( + tool=tool, + code="unsupported_pid", + message="Requested command is not in the approved read-only catalog.", + details={"command_key": command_key}, + ) + + session, error = self._require_connected(tool=tool) + if error is not None: + return None, error + + if not session.supports(command_key): + return None, self._error( + tool=tool, + code="unsupported_pid", + message="Requested PID is not supported by the connected vehicle.", + details={"command_key": command_key}, + ) + + try: + response = session.query(command_key) + except Exception as exc: # pragma: no branch + if self._is_timeout_error(exc): + return None, self._error( + tool=tool, + code="timeout", + message="Timed out waiting for a response from the adapter.", + details={"command_key": command_key}, + ) + raise + + if response.is_null(): + return None, self._error( + tool=tool, + code="null_response", + message="Adapter returned no data for the requested PID.", + details={"command_key": command_key}, + ) + + return serialize_obd_response(response), None + + def _require_connected( + self, + tool: str, + ) -> Tuple[Optional[OBDAISession], Optional[Dict[str, Any]]]: + if self._session is None: + return None, self._error( + tool=tool, + code="disconnected_adapter", + message="No active OBD session. Call connect_vehicle first.", + ) + + if not self._session.is_connected(): + return None, self._error( + tool=tool, + code="disconnected_adapter", + message="OBD adapter is disconnected from the vehicle.", + details={"connection": self._session.connection_metadata()}, + ) + + return self._session, None + + @staticmethod + def _success(tool: str, data: Mapping[str, Any]) -> Dict[str, Any]: + return { + "ok": True, + "tool": tool, + "data": dict(data), + } + + @staticmethod + def _error( + tool: str, + code: str, + message: str, + details: Optional[Mapping[str, Any]] = None, + ) -> Dict[str, Any]: + return { + "ok": False, + "tool": tool, + "error": { + "code": code, + "message": message, + "details": dict(details or {}), + }, + } + + @staticmethod + def _is_timeout_error(exc: Exception) -> bool: + timeout_types = (TimeoutError,) + if SerialTimeoutException: + timeout_types = timeout_types + (SerialTimeoutException,) + return isinstance(exc, timeout_types) diff --git a/tests/test_obd_ai_tools.py b/tests/test_obd_ai_tools.py new file mode 100644 index 00000000..e33a83cb --- /dev/null +++ b/tests/test_obd_ai_tools.py @@ -0,0 +1,215 @@ +import obd +from obd.OBDResponse import OBDResponse, Status +from obd.utils import OBDStatus + +from obd_ai.session import OBDAISessionManager +from obd_ai.tools import OBDAIReadOnlyToolSurface + + +class FakeConnection: + def __init__( + self, + *, + connected=True, + support_overrides=None, + response_overrides=None, + query_exceptions=None, + ): + self.connected = connected + self.query_calls = [] + self.closed = False + self.support_overrides = support_overrides or {} + self.response_overrides = response_overrides or {} + self.query_exceptions = query_exceptions or {} + + def query(self, command, force=False): + self.query_calls.append((command, force)) + + if command.name in self.query_exceptions: + raise self.query_exceptions[command.name] + + if command.name in self.response_overrides: + return self.response_overrides[command.name] + + response = OBDResponse(command=command, messages=[object()]) + response.value = obd.Unit.Quantity(900, obd.Unit.rpm) + return response + + def status(self): + return OBDStatus.CAR_CONNECTED if self.connected else OBDStatus.NOT_CONNECTED + + def is_connected(self): + return self.connected + + @staticmethod + def port_name(): + return "FAKEPORT" + + def supports(self, command): + return self.support_overrides.get(command.name, True) + + def close(self): + self.closed = True + + +class FakeFactory: + def __init__(self, connection): + self.connection = connection + + def __call__(self, **kwargs): + del kwargs + return self.connection + + +def _surface_for(connection): + manager = OBDAISessionManager(obd_factory=FakeFactory(connection)) + return OBDAIReadOnlyToolSurface(session_manager=manager) + + +def _null_response(command): + return OBDResponse(command=command, messages=[]) + + +def test_connect_vehicle_returns_success_payload_for_connected_adapter(): + surface = _surface_for(FakeConnection(connected=True)) + + payload = surface.connect_vehicle() + + assert payload["ok"] is True + assert payload["tool"] == "connect_vehicle" + assert payload["data"]["connection"]["is_connected"] is True + + +def test_connect_vehicle_returns_disconnected_adapter_error(): + surface = _surface_for(FakeConnection(connected=False)) + + payload = surface.connect_vehicle() + + assert payload["ok"] is False + assert payload["error"]["code"] == "disconnected_adapter" + + +def test_read_sensor_returns_structured_success_for_approved_supported_pid(): + surface = _surface_for(FakeConnection()) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "engine_rpm"}) + + assert payload["ok"] is True + assert payload["data"]["command_key"] == "engine_rpm" + assert payload["data"]["response"]["command"]["name"] == "RPM" + + +def test_read_sensor_blocks_unapproved_command_keys(): + surface = _surface_for(FakeConnection()) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "clear_trouble_codes"}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "unsupported_pid" + + +def test_read_sensor_returns_disconnected_adapter_error_without_active_session(): + surface = _surface_for(FakeConnection()) + + payload = surface.read_sensor({"command_key": "engine_rpm"}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "disconnected_adapter" + + +def test_read_sensor_returns_unsupported_pid_when_vehicle_lacks_support(): + connection = FakeConnection(support_overrides={"RPM": False}) + surface = _surface_for(connection) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "engine_rpm"}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "unsupported_pid" + + +def test_read_sensor_returns_null_response_error_when_adapter_returns_no_data(): + connection = FakeConnection(response_overrides={"RPM": _null_response(obd.commands.RPM)}) + surface = _surface_for(connection) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "engine_rpm"}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "null_response" + + +def test_read_sensor_returns_timeout_error_when_query_times_out(): + connection = FakeConnection(query_exceptions={"RPM": TimeoutError("query timed out")}) + surface = _surface_for(connection) + surface.connect_vehicle() + + payload = surface.read_sensor({"command_key": "engine_rpm"}) + + assert payload["ok"] is False + assert payload["error"]["code"] == "timeout" + + +def test_list_supported_commands_only_reports_approved_allowlist_keys(): + connection = FakeConnection(support_overrides={"RPM": True, "SPEED": False}) + surface = _surface_for(connection) + surface.connect_vehicle() + + payload = surface.list_supported_commands({"include_support": True}) + + assert payload["ok"] is True + keys = {item["key"] for item in payload["data"]["commands"]} + assert "engine_rpm" in keys + assert "vehicle_speed" in keys + assert "clear_trouble_codes" not in keys + + +def test_get_dtc_codes_reads_stored_and_pending_pids(): + stored = OBDResponse(command=obd.commands.GET_DTC, messages=[object()]) + stored.value = [("P0101", "Mass or Volume Air Flow Circuit Range/Performance")] + + pending = OBDResponse(command=obd.commands.GET_CURRENT_DTC, messages=[object()]) + pending.value = [] + + connection = FakeConnection( + response_overrides={ + "GET_DTC": stored, + "GET_CURRENT_DTC": pending, + } + ) + surface = _surface_for(connection) + surface.connect_vehicle() + + payload = surface.get_dtc_codes() + + assert payload["ok"] is True + assert payload["data"]["stored"]["response"]["value"]["type"] == "dtc_list" + assert payload["data"]["pending"]["response"]["value"]["count"] == 0 + + +def test_get_freeze_frame_and_emissions_monitor_status_use_approved_reads(): + freeze = OBDResponse(command=obd.commands.FREEZE_DTC, messages=[object()]) + freeze.value = ("P0101", "Mass or Volume Air Flow Circuit Range/Performance") + + status = OBDResponse(command=obd.commands.STATUS, messages=[object()]) + status.value = Status() + + connection = FakeConnection( + response_overrides={ + "FREEZE_DTC": freeze, + "STATUS": status, + } + ) + surface = _surface_for(connection) + surface.connect_vehicle() + + freeze_payload = surface.get_freeze_frame() + status_payload = surface.get_emissions_monitor_status() + + assert freeze_payload["ok"] is True + assert freeze_payload["data"]["response"]["command"]["name"] == "FREEZE_DTC" + + assert status_payload["ok"] is True + assert status_payload["data"]["response"]["value"]["type"] == "status"