feat(obd_ai): add foundation layer and approved command catalog

obd_ai/__init__.py

@@ -0,0 +1,16 @@
+"""Safe, high-level interfaces for LLM/MCP-style OBD access."""
+
+from .catalog import (
+    ApprovedCommand,
+    ApprovedCommandCatalog,
+    DEFAULT_APPROVED_COMMAND_CATALOG,
+)
+from .session import OBDAISession, OBDAISessionManager
+
+__all__ = [
+    "ApprovedCommand",
+    "ApprovedCommandCatalog",
+    "DEFAULT_APPROVED_COMMAND_CATALOG",
+    "OBDAISession",
+    "OBDAISessionManager",
+]

obd_ai/catalog.py

@@ -0,0 +1,183 @@
+"""Approved read-only command catalog for the OBD AI layer."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Iterable, Mapping, Tuple
+
+import obd
+from obd.OBDCommand import OBDCommand
+
+
+@dataclass(frozen=True)
+class ApprovedCommand:
+    """Metadata for a command that is safe to expose to the OBD AI layer."""
+
+    key: str
+    public_name: str
+    category: str
+    description: str
+    obd_command_name: str
+    obd_command: OBDCommand
+
+
+@dataclass(frozen=True)
+class _CommandDefinition:
+    key: str
+    public_name: str
+    category: str
+    description: str
+    obd_command_name: str
+
+
+class ApprovedCommandCatalog:
+    """Read-only catalog that maps public command keys to OBD commands."""
+
+    def __init__(self, definitions: Iterable[_CommandDefinition]):
+        mutable_map = {}
+        ordered = []
+
+        for definition in definitions:
+            if definition.key in mutable_map:
+                raise ValueError(f"Duplicate approved command key: {definition.key}")
+
+            try:
+                obd_command = getattr(obd.commands, definition.obd_command_name)
+            except AttributeError as exc:
+                raise ValueError(
+                    f"Unknown obd command mapping: {definition.obd_command_name}"
+                ) from exc
+
+            approved = ApprovedCommand(
+                key=definition.key,
+                public_name=definition.public_name,
+                category=definition.category,
+                description=definition.description,
+                obd_command_name=definition.obd_command_name,
+                obd_command=obd_command,
+            )
+
+            mutable_map[approved.key] = approved
+            ordered.append(approved)
+
+        self._commands: Mapping[str, ApprovedCommand] = MappingProxyType(mutable_map)
+        self._ordered_commands: Tuple[ApprovedCommand, ...] = tuple(ordered)
+
+    def __contains__(self, key: str) -> bool:
+        return key in self._commands
+
+    def __len__(self) -> int:
+        return len(self._ordered_commands)
+
+    def get(self, key: str) -> ApprovedCommand:
+        try:
+            return self._commands[key]
+        except KeyError as exc:
+            raise KeyError(f"Unapproved command key: {key}") from exc
+
+    def keys(self) -> Tuple[str, ...]:
+        return tuple(command.key for command in self._ordered_commands)
+
+    def list(self) -> Tuple[ApprovedCommand, ...]:
+        return self._ordered_commands
+
+    def as_mapping(self) -> Mapping[str, ApprovedCommand]:
+        return self._commands
+
+
+_MVP_READ_ONLY_COMMANDS = (
+    _CommandDefinition(
+        key="status_since_dtc_clear",
+        public_name="Status Since DTC Clear",
+        category="diagnostics",
+        description="Read current monitor status and MIL flags.",
+        obd_command_name="STATUS",
+    ),
+    _CommandDefinition(
+        key="engine_rpm",
+        public_name="Engine RPM",
+        category="powertrain",
+        description="Read current engine speed in revolutions per minute.",
+        obd_command_name="RPM",
+    ),
+    _CommandDefinition(
+        key="vehicle_speed",
+        public_name="Vehicle Speed",
+        category="powertrain",
+        description="Read current vehicle speed.",
+        obd_command_name="SPEED",
+    ),
+    _CommandDefinition(
+        key="engine_coolant_temperature",
+        public_name="Engine Coolant Temperature",
+        category="powertrain",
+        description="Read current coolant temperature.",
+        obd_command_name="COOLANT_TEMP",
+    ),
+    _CommandDefinition(
+        key="intake_air_temperature",
+        public_name="Intake Air Temperature",
+        category="powertrain",
+        description="Read current intake air temperature.",
+        obd_command_name="INTAKE_TEMP",
+    ),
+    _CommandDefinition(
+        key="calculated_engine_load",
+        public_name="Calculated Engine Load",
+        category="powertrain",
+        description="Read current calculated engine load percentage.",
+        obd_command_name="ENGINE_LOAD",
+    ),
+    _CommandDefinition(
+        key="throttle_position",
+        public_name="Throttle Position",
+        category="powertrain",
+        description="Read current throttle position percentage.",
+        obd_command_name="THROTTLE_POS",
+    ),
+    _CommandDefinition(
+        key="fuel_level",
+        public_name="Fuel Level",
+        category="fuel",
+        description="Read current fuel level percentage.",
+        obd_command_name="FUEL_LEVEL",
+    ),
+    _CommandDefinition(
+        key="control_module_voltage",
+        public_name="Control Module Voltage",
+        category="electrical",
+        description="Read current ECU/control module voltage.",
+        obd_command_name="CONTROL_MODULE_VOLTAGE",
+    ),
+    _CommandDefinition(
+        key="engine_run_time",
+        public_name="Engine Run Time",
+        category="diagnostics",
+        description="Read total engine run time since startup.",
+        obd_command_name="RUN_TIME",
+    ),
+    _CommandDefinition(
+        key="stored_trouble_codes",
+        public_name="Stored Trouble Codes",
+        category="diagnostics",
+        description="Read currently stored diagnostic trouble codes.",
+        obd_command_name="GET_DTC",
+    ),
+    _CommandDefinition(
+        key="pending_trouble_codes",
+        public_name="Pending Trouble Codes",
+        category="diagnostics",
+        description="Read pending trouble codes from current drive cycle.",
+        obd_command_name="GET_CURRENT_DTC",
+    ),
+    _CommandDefinition(
+        key="vehicle_identification_number",
+        public_name="Vehicle Identification Number",
+        category="vehicle_info",
+        description="Read the vehicle identification number (VIN).",
+        obd_command_name="VIN",
+    ),
+)
+
+DEFAULT_APPROVED_COMMAND_CATALOG = ApprovedCommandCatalog(_MVP_READ_ONLY_COMMANDS)

obd_ai/session.py

@@ -0,0 +1,86 @@
+"""Session abstractions for the OBD AI layer."""
+
+from __future__ import annotations
+
+from typing import Callable, Optional
+
+import obd
+from obd.OBDResponse import OBDResponse
+from obd.utils import OBDStatus
+
+from .catalog import ApprovedCommandCatalog, DEFAULT_APPROVED_COMMAND_CATALOG
+
+OBDFactory = Callable[..., obd.OBD]
+
+
+class OBDAISession:
+    """Safe wrapper around an OBD connection constrained by an approved catalog."""
+
+    def __init__(self, connection: obd.OBD, catalog: ApprovedCommandCatalog):
+        self._connection = connection
+        self._catalog = catalog
+
+    def __enter__(self) -> "OBDAISession":
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> None:
+        self.close()
+
+    def list_approved_commands(self):
+        return self._catalog.list()
+
+    def query(self, approved_key: str) -> OBDResponse:
+        approved = self._catalog.get(approved_key)
+        return self._connection.query(approved.obd_command)
+
+    def supports(self, approved_key: str) -> bool:
+        approved = self._catalog.get(approved_key)
+        return self._connection.supports(approved.obd_command)
+
+    def status(self) -> OBDStatus:
+        return self._connection.status()
+
+    def is_connected(self) -> bool:
+        return self._connection.is_connected()
+
+    def port_name(self) -> str:
+        return self._connection.port_name()
+
+    def close(self) -> None:
+        self._connection.close()
+
+
+class OBDAISessionManager:
+    """Factory for creating OBD AI sessions over `obd.OBD`."""
+
+    def __init__(
+        self,
+        catalog: ApprovedCommandCatalog = DEFAULT_APPROVED_COMMAND_CATALOG,
+        obd_factory: Optional[OBDFactory] = None,
+    ):
+        self._catalog = catalog
+        self._obd_factory = obd_factory or obd.OBD
+
+    def open_session(
+        self,
+        portstr=None,
+        baudrate=None,
+        protocol=None,
+        fast: bool = True,
+        timeout: float = 0.1,
+        check_voltage: bool = True,
+        start_low_power: bool = False,
+    ) -> OBDAISession:
+        connection = self._obd_factory(
+            portstr=portstr,
+            baudrate=baudrate,
+            protocol=protocol,
+            fast=fast,
+            timeout=timeout,
+            check_voltage=check_voltage,
+            start_low_power=start_low_power,
+        )
+        return OBDAISession(connection=connection, catalog=self._catalog)
+
+    def wrap_existing(self, connection: obd.OBD) -> OBDAISession:
+        return OBDAISession(connection=connection, catalog=self._catalog)

tests/test_obd_ai_catalog.py

@@ -0,0 +1,35 @@
+import pytest
+
+import obd
+from obd_ai.catalog import DEFAULT_APPROVED_COMMAND_CATALOG
+
+
+def test_default_catalog_maps_to_real_obd_commands():
+    catalog = DEFAULT_APPROVED_COMMAND_CATALOG
+
+    rpm = catalog.get("engine_rpm")
+    assert rpm.public_name == "Engine RPM"
+    assert rpm.category == "powertrain"
+    assert rpm.obd_command_name == "RPM"
+    assert rpm.obd_command is obd.commands.RPM
+
+
+def test_default_catalog_excludes_write_commands_for_mvp_safety():
+    catalog = DEFAULT_APPROVED_COMMAND_CATALOG
+
+    assert "stored_trouble_codes" in catalog
+    assert "pending_trouble_codes" in catalog
+    assert "clear_trouble_codes" not in catalog
+    assert "CLEAR_DTC" not in catalog
+
+
+def test_catalog_unknown_key_raises_key_error():
+    with pytest.raises(KeyError, match="Unapproved command key"):
+        DEFAULT_APPROVED_COMMAND_CATALOG.get("clear_trouble_codes")
+
+
+def test_catalog_mapping_is_read_only():
+    mapping = DEFAULT_APPROVED_COMMAND_CATALOG.as_mapping()
+
+    with pytest.raises(TypeError):
+        mapping["new"] = "value"