128 lines
4.0 KiB
Python
128 lines
4.0 KiB
Python
"""Configuration management for OBD2 client."""
|
|
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
|
|
@dataclass
|
|
class CANConfig:
|
|
"""CAN bus configuration."""
|
|
|
|
interface: str = "can0"
|
|
bitrate: int = 500000
|
|
virtual: bool = False
|
|
|
|
|
|
@dataclass
|
|
class OBD2Config:
|
|
"""OBD2 protocol configuration."""
|
|
|
|
request_id: int = 0x7DF
|
|
response_id: int = 0x7E8
|
|
timeout: float = 0.1
|
|
|
|
|
|
@dataclass
|
|
class PollingConfig:
|
|
"""Polling configuration."""
|
|
|
|
interval_fast: float = 0.1
|
|
interval_slow: float = 1.0
|
|
fast_pids: List[int] = field(default_factory=lambda: [0x0C, 0x0D, 0x11])
|
|
slow_pids: List[int] = field(default_factory=lambda: [0x05, 0x2F, 0x5C])
|
|
|
|
|
|
@dataclass
|
|
class Config:
|
|
"""Main configuration container."""
|
|
|
|
can: CANConfig = field(default_factory=CANConfig)
|
|
obd2: OBD2Config = field(default_factory=OBD2Config)
|
|
polling: PollingConfig = field(default_factory=PollingConfig)
|
|
|
|
@classmethod
|
|
def load(cls, config_path: Optional[str] = None) -> "Config":
|
|
"""Load configuration from JSON file and environment variables.
|
|
|
|
Args:
|
|
config_path: Path to config.json file
|
|
|
|
Returns:
|
|
Config instance
|
|
"""
|
|
config = cls()
|
|
|
|
if config_path is None:
|
|
config_path = os.environ.get(
|
|
"OBD2_CONFIG_PATH",
|
|
str(Path(__file__).parent.parent / "config.json"),
|
|
)
|
|
|
|
config_file = Path(config_path)
|
|
if config_file.exists():
|
|
with open(config_file, "r") as f:
|
|
data = json.load(f)
|
|
config._load_from_dict(data)
|
|
|
|
config._load_from_env()
|
|
|
|
return config
|
|
|
|
def _load_from_dict(self, data: dict) -> None:
|
|
"""Load configuration from dictionary."""
|
|
if "can" in data:
|
|
can_data = data["can"]
|
|
self.can.interface = can_data.get("interface", self.can.interface)
|
|
self.can.bitrate = can_data.get("bitrate", self.can.bitrate)
|
|
self.can.virtual = can_data.get("virtual", self.can.virtual)
|
|
|
|
if "obd2" in data:
|
|
obd2_data = data["obd2"]
|
|
if "request_id" in obd2_data:
|
|
self.obd2.request_id = self._parse_hex(obd2_data["request_id"])
|
|
if "response_id" in obd2_data:
|
|
self.obd2.response_id = self._parse_hex(obd2_data["response_id"])
|
|
self.obd2.timeout = obd2_data.get("timeout", self.obd2.timeout)
|
|
|
|
if "polling" in data:
|
|
poll_data = data["polling"]
|
|
self.polling.interval_fast = poll_data.get(
|
|
"interval_fast", self.polling.interval_fast
|
|
)
|
|
self.polling.interval_slow = poll_data.get(
|
|
"interval_slow", self.polling.interval_slow
|
|
)
|
|
if "fast_pids" in poll_data:
|
|
self.polling.fast_pids = [
|
|
self._parse_hex(p) for p in poll_data["fast_pids"]
|
|
]
|
|
if "slow_pids" in poll_data:
|
|
self.polling.slow_pids = [
|
|
self._parse_hex(p) for p in poll_data["slow_pids"]
|
|
]
|
|
|
|
def _load_from_env(self) -> None:
|
|
"""Load configuration from environment variables."""
|
|
if "OBD2_CAN_INTERFACE" in os.environ:
|
|
self.can.interface = os.environ["OBD2_CAN_INTERFACE"]
|
|
if "OBD2_CAN_BITRATE" in os.environ:
|
|
self.can.bitrate = int(os.environ["OBD2_CAN_BITRATE"])
|
|
if "OBD2_CAN_VIRTUAL" in os.environ:
|
|
self.can.virtual = os.environ["OBD2_CAN_VIRTUAL"].lower() == "true"
|
|
if "OBD2_TIMEOUT" in os.environ:
|
|
self.obd2.timeout = float(os.environ["OBD2_TIMEOUT"])
|
|
|
|
@staticmethod
|
|
def _parse_hex(value) -> int:
|
|
"""Parse hex string or int to int."""
|
|
if isinstance(value, int):
|
|
return value
|
|
if isinstance(value, str):
|
|
if value.startswith("0x") or value.startswith("0X"):
|
|
return int(value, 16)
|
|
return int(value)
|
|
return value
|