qsethuk/carpibord
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix get PIDs and restart auto

Browse Source
This commit is contained in:
Alexander Poletaev
2026-01-30 00:05:27 +03:00
parent a96328d4e6
commit 9a3fe4457c
3 changed files with 106 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
obd2_client/src/config.py
View File

@@ -25,6 +25,15 @@ class OBD2Config:
timeout: float = 0.1
@dataclass
class ScanConfig:
"""PID scan configuration."""
retries: int = 5
retry_delay: float = 1.5
initial_delay: float = 1.0
@dataclass
class PollingConfig:
"""Polling configuration."""
@@ -41,6 +50,7 @@ class Config:
can: CANConfig = field(default_factory=CANConfig)
obd2: OBD2Config = field(default_factory=OBD2Config)
scan: ScanConfig = field(default_factory=ScanConfig)
polling: PollingConfig = field(default_factory=PollingConfig)
@classmethod
@@ -87,6 +97,12 @@ class Config:
self.obd2.response_id = self._parse_hex(obd2_data["response_id"])
self.obd2.timeout = obd2_data.get("timeout", self.obd2.timeout)
if "scan" in data:
scan_data = data["scan"]
self.scan.retries = scan_data.get("retries", self.scan.retries)
self.scan.retry_delay = scan_data.get("retry_delay", self.scan.retry_delay)
self.scan.initial_delay = scan_data.get("initial_delay", self.scan.initial_delay)
if "polling" in data:
poll_data = data["polling"]
self.polling.interval_fast = poll_data.get(

52
obd2_client/src/main.py
View File

@@ -74,13 +74,39 @@ class OBD2Client:
pm.set_action_handler(ActionID.RECONNECT_OBD, self._action_reconnect_obd)
def _action_reconnect_obd(self) -> bool:
"""Reconnect to OBD2."""
"""Reconnect to OBD2 and rescan PIDs."""
try:
self.poller.stop()
self.can_interface.disconnect()
time.sleep(0.5)
# Stop poller if running
if self.poller.is_running:
self.poller.stop()
# Disconnect CAN
if self.can_interface.is_connected:
self.can_interface.disconnect()
time.sleep(1.0) # Let CAN interface settle
# Reconnect
self.can_interface.connect()
self.poller.start()
# Rescan PIDs with retries
supported = self.scanner.scan(use_cache=False, retries=3, initial_delay=1.0)
if not supported:
self._logger.warning("No PIDs found after reconnect")
return False
# Configure and start poller
readable_pids = self.scanner.get_readable_pids()
if readable_pids:
fast_pids = [p for p in self.config.polling.fast_pids if p in readable_pids]
slow_pids = [p for p in self.config.polling.slow_pids if p in readable_pids]
if not fast_pids and not slow_pids:
fast_pids = readable_pids[:3]
slow_pids = readable_pids[3:]
self.poller.set_pids(fast_pids, slow_pids)
self.poller.start()
self._logger.info(f"OBD2 reconnected, {len(supported)} PIDs found")
return True
except Exception as e:
self._logger.error(f"Reconnect failed: {e}")
@@ -108,15 +134,19 @@ class OBD2Client:
self._logger.info(f"Connecting to {self.config.can.interface}...")
self.can_interface.connect()
self._logger.info("Scanning for supported PIDs...")
supported = self.scanner.scan()
self._logger.info("Scanning for supported PIDs (with retries)...")
supported = self.scanner.scan(
retries=self.config.scan.retries,
retry_delay=self.config.scan.retry_delay,
initial_delay=self.config.scan.initial_delay,
)
if not supported:
self._logger.error("No PIDs supported or no response from ECU")
if not self.flipper_server:
return
# Continue running for Flipper even without OBD2
self._logger.info("Running in Flipper-only mode...")
# Continue running for Flipper - user can reconnect OBD2 via Flipper menu
self._logger.info("Running in Flipper-only mode (use 'Reconnect OBD2' action)...")
while self._running:
time.sleep(1.0)
return
@@ -129,6 +159,10 @@ class OBD2Client:
readable_pids = self.scanner.get_readable_pids()
if not readable_pids:
self._logger.warning("No readable PIDs found")
if self.flipper_server:
self._logger.info("Running in Flipper-only mode...")
while self._running:
time.sleep(1.0)
return
fast_pids = [p for p in self.config.polling.fast_pids if p in readable_pids]

67
obd2_client/src/obd2/scanner.py
View File

@@ -46,43 +46,70 @@ class OBD2Scanner:
"""Check if scan has been performed."""
return self._scanned
def scan(self, use_cache: bool = True) -> List[int]:
"""Scan for supported PIDs.
def scan(
self,
use_cache: bool = True,
retries: int = 3,
retry_delay: float = 1.0,
initial_delay: float = 0.5,
) -> List[int]:
"""Scan for supported PIDs with retry logic.
Args:
use_cache: If True, try to load from cache first
retries: Number of retry attempts if scan fails
retry_delay: Delay between retries (seconds)
initial_delay: Delay before first scan attempt (seconds)
Returns:
List of supported PID codes
"""
import time
if use_cache and self._load_cache():
self._logger.info(f"Loaded {len(self._supported_pids)} PIDs from cache")
return self.supported_pids
self._logger.info("Starting PID scan...")
self._supported_pids.clear()
# Initial delay to let CAN interface and ECU stabilize
if initial_delay > 0:
self._logger.debug(f"Waiting {initial_delay}s for CAN/ECU to stabilize...")
time.sleep(initial_delay)
for base_pid in self.SUPPORTED_PID_QUERIES:
self._logger.debug(f"Querying supported PIDs from 0x{base_pid:02X}")
for attempt in range(retries):
if attempt > 0:
self._logger.info(f"Retry {attempt}/{retries} after {retry_delay}s...")
time.sleep(retry_delay)
supported = self.protocol.query_supported_pids(base_pid)
if not supported:
self._logger.debug(f"No response for base PID 0x{base_pid:02X}")
break
self._logger.info(f"Starting PID scan (attempt {attempt + 1}/{retries})...")
self._supported_pids.clear()
self._supported_pids.update(supported)
self._logger.debug(f"Found {len(supported)} supported PIDs")
for base_pid in self.SUPPORTED_PID_QUERIES:
self._logger.debug(f"Querying supported PIDs from 0x{base_pid:02X}")
next_base = base_pid + 0x20
if next_base not in supported:
break
supported = self.protocol.query_supported_pids(base_pid)
if not supported:
self._logger.debug(f"No response for base PID 0x{base_pid:02X}")
break
self._supported_pids.update(supported)
self._logger.debug(f"Found {len(supported)} supported PIDs")
next_base = base_pid + 0x20
if next_base not in supported:
break
# Success if we found any PIDs
if self._supported_pids:
self._scanned = True
self._logger.info(f"Scan complete. Found {len(self._supported_pids)} supported PIDs")
if self.cache_file:
self._save_cache()
return self.supported_pids
self._logger.warning(f"Scan failed after {retries} attempts")
self._scanned = True
self._logger.info(f"Scan complete. Found {len(self._supported_pids)} supported PIDs")
if self.cache_file:
self._save_cache()
return self.supported_pids
def get_readable_pids(self) -> List[int]: