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

fix flipper_protocol

Browse Source
This commit is contained in:
Alexander Poletaev
2026-01-29 17:12:06 +03:00
parent 75b8810430
commit 5c64b01d5f
1 changed files with 107 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
can_sniffer/src/handlers/flipper_handler.py
View File

@@ -8,10 +8,15 @@ import threading
import time
from typing import Dict, Any, Optional, Callable
try:
import serial
except ImportError:
serial = None # type: ignore
from logger import get_logger
from config import config
from obd2.pids import OBD2Reading
from flipper.protocol import FlipperProtocol
from flipper.protocol import Protocol, Command
from flipper.page_manager import PageManager
from flipper.pages import OBD2StatsPage, OBD2CommPage, UPSStatusPage, SystemInfoPage, AppStatusPage
from flipper.providers.obd2_provider import OBD2Provider
@@ -49,12 +54,14 @@ class FlipperHandler(BaseHandler):
self._baudrate = baudrate
self._update_interval = update_interval
self._protocol: Optional[FlipperProtocol] = None
self._serial: Optional[serial.Serial] = None
self._page_manager: Optional[PageManager] = None
self._obd2_provider: Optional[OBD2Provider] = None
self._update_thread: Optional[threading.Thread] = None
self._rx_thread: Optional[threading.Thread] = None
self._running = False
self._connected = False
# Callbacks for data access
self._state_callback: Optional[Callable[[], Any]] = None
@@ -78,20 +85,22 @@ class FlipperHandler(BaseHandler):
logger.info("Flipper handler disabled")
return False
try:
# Initialize UART protocol
self._protocol = FlipperProtocol(
device=self._device,
baudrate=self._baudrate,
)
if serial is None:
logger.warning("pyserial not installed, Flipper handler disabled")
return False
if not self._protocol.connect():
logger.warning(f"Failed to connect to Flipper on {self._device}")
self._available = False
return False
try:
# Initialize UART connection
self._serial = serial.Serial(
port=self._device,
baudrate=self._baudrate,
timeout=0.1,
write_timeout=1.0,
)
self._connected = True
# Initialize page manager with OBD2 pages
self._page_manager = PageManager(self._protocol)
self._page_manager = PageManager()
# Create and register pages
obd2_page = OBD2StatsPage()
@@ -115,15 +124,23 @@ class FlipperHandler(BaseHandler):
if self._stats_callback:
self._obd2_provider.set_stats_callback(self._stats_callback)
# Start update thread
# Start threads
self._running = True
self._update_thread = threading.Thread(
target=self._update_loop,
name="Flipper-Update",
name="Flipper-TX",
daemon=True,
)
self._update_thread.start()
self._rx_thread = threading.Thread(
target=self._rx_loop,
name="Flipper-RX",
daemon=True,
)
self._rx_thread.start()
self._initialized = True
logger.info(
"Flipper handler initialized",
@@ -134,6 +151,10 @@ class FlipperHandler(BaseHandler):
)
return True
except serial.SerialException as e:
logger.warning(f"Failed to open Flipper UART: {e}")
self._connected = False
return False
except Exception as e:
logger.error(f"Failed to initialize Flipper handler: {e}")
return False
@@ -170,9 +191,19 @@ class FlipperHandler(BaseHandler):
if self._update_thread and self._update_thread.is_alive():
self._update_thread.join(timeout=2.0)
if self._protocol:
self._protocol.disconnect()
if self._rx_thread and self._rx_thread.is_alive():
self._rx_thread.join(timeout=2.0)
if self._page_manager:
self._page_manager.shutdown()
if self._serial and self._serial.is_open:
try:
self._serial.close()
except Exception as e:
logger.warning(f"Error closing serial: {e}")
self._connected = False
logger.info("Flipper handler shutdown")
self._initialized = False
@@ -183,30 +214,79 @@ class FlipperHandler(BaseHandler):
"enabled": self._enabled,
"initialized": self._initialized,
"device": self._device,
"connected": self._protocol.is_connected() if self._protocol else False,
"connected": self._connected,
}
if self._page_manager:
stats["current_page"] = self._page_manager.get_current_page_name()
stats["current_page"] = self._page_manager.get_stats().get("current_page")
return stats
def get_current_page_name(self) -> Optional[str]:
"""Get name of current page."""
if self._page_manager:
page = self._page_manager.get_current_page()
return page.name if page else None
return None
def _send(self, data: str) -> bool:
"""Send data to Flipper via UART."""
if not self._serial or not self._serial.is_open:
return False
try:
self._serial.write(data.encode('utf-8'))
return True
except Exception as e:
logger.error(f"Flipper send error: {e}")
self._connected = False
return False
def _update_loop(self) -> None:
"""Update loop for sending data to Flipper."""
logger.debug("Flipper update loop started")
logger.debug("Flipper TX loop started")
while self._running:
try:
if self._page_manager and self._protocol and self._protocol.is_connected():
# Update current page
self._page_manager.update()
# Handle input from Flipper
self._page_manager.handle_input()
if self._page_manager and self._connected:
content = self._page_manager.get_current_content()
if content:
self._send(content)
except Exception as e:
logger.error(f"Flipper update error: {e}")
time.sleep(self._update_interval)
logger.debug("Flipper update loop stopped")
logger.debug("Flipper TX loop stopped")
def _rx_loop(self) -> None:
"""Receive loop for commands from Flipper."""
logger.debug("Flipper RX loop started")
while self._running:
try:
if not self._serial or not self._serial.is_open:
time.sleep(0.1)
continue
# Read line from Flipper
line = self._serial.readline()
if not line:
continue
line_str = line.decode('utf-8', errors='ignore').strip()
if not line_str:
continue
# Parse command
command = Protocol.decode_command(line_str)
if command and self._page_manager:
result = self._page_manager.process_command(command)
if result:
self._send(result)
except Exception as e:
logger.error(f"Flipper RX error: {e}")
logger.debug("Flipper RX loop stopped")