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

Create new module for flipper and update UI flipper zero

Browse Source
This commit is contained in:
Alexander Poletaev
2026-01-27 14:52:06 +03:00
parent 434feb2a1d
commit dab72ac1b1
18 changed files with 2659 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
can_sniffer/src/flipper/__init__.py Normal file
View File

@@ -0,0 +1,11 @@
"""
Flipper Zero Dynamic UI System.
This module provides a plugin-based page system for Flipper Zero display
with bidirectional communication via UART.
"""
from flipper.protocol import PageType, PageContent, Protocol
from flipper.page_manager import PageManager
__all__ = ["PageType", "PageContent", "Protocol", "PageManager"]

255
can_sniffer/src/flipper/page_manager.py Normal file
View File

@@ -0,0 +1,255 @@
"""
Page Manager for Flipper Zero UI.
Manages page navigation, state, and content generation.
"""
import threading
from typing import List, Optional, Dict, Any
from flipper.protocol import Protocol, PageContent, Command, CommandType
from flipper.pages.base import BasePage
class PageManager:
"""
Manages Flipper Zero pages and navigation.
Handles:
- Page registration and lifecycle
- Navigation between pages
- Command processing
- Content generation
"""
def __init__(self):
"""Initialize page manager."""
self._pages: List[BasePage] = []
self._current_index = 0
self._lock = threading.Lock()
self._last_result: Optional[str] = None
def register_page(self, page: BasePage) -> None:
"""
Register a page.
Args:
page: Page instance to register
"""
with self._lock:
# Check for duplicate names
for existing in self._pages:
if existing.name == page.name:
raise ValueError(f"Page '{page.name}' already registered")
self._pages.append(page)
def unregister_page(self, name: str) -> bool:
"""
Unregister a page by name.
Args:
name: Page name to remove
Returns:
True if page was found and removed
"""
with self._lock:
for i, page in enumerate(self._pages):
if page.name == name:
self._pages.pop(i)
# Adjust current index if needed
if self._current_index >= len(self._pages):
self._current_index = max(0, len(self._pages) - 1)
return True
return False
def get_enabled_pages(self) -> List[BasePage]:
"""Get list of enabled pages."""
with self._lock:
return [p for p in self._pages if p.is_enabled()]
def get_page_count(self) -> int:
"""Get number of enabled pages."""
return len(self.get_enabled_pages())
def get_current_page(self) -> Optional[BasePage]:
"""Get currently active page."""
pages = self.get_enabled_pages()
with self._lock:
if 0 <= self._current_index < len(pages):
return pages[self._current_index]
return None
def get_current_index(self) -> int:
"""Get current page index."""
with self._lock:
return self._current_index
def navigate_next(self) -> bool:
"""
Navigate to next page.
Returns:
True if navigation occurred
"""
pages = self.get_enabled_pages()
with self._lock:
# Don't navigate if current page has pending action
if self._current_index < len(pages):
current = pages[self._current_index]
if current.has_pending_action():
return False
current.on_leave()
if self._current_index < len(pages) - 1:
self._current_index += 1
pages[self._current_index].on_enter()
return True
return False
def navigate_prev(self) -> bool:
"""
Navigate to previous page.
Returns:
True if navigation occurred
"""
pages = self.get_enabled_pages()
with self._lock:
# Don't navigate if current page has pending action
if self._current_index < len(pages):
current = pages[self._current_index]
if current.has_pending_action():
return False
current.on_leave()
if self._current_index > 0:
self._current_index -= 1
pages[self._current_index].on_enter()
return True
return False
def navigate_to(self, index: int) -> bool:
"""
Navigate to specific page index.
Args:
index: Page index to navigate to
Returns:
True if navigation occurred
"""
pages = self.get_enabled_pages()
with self._lock:
if 0 <= index < len(pages):
if self._current_index < len(pages):
pages[self._current_index].on_leave()
self._current_index = index
pages[self._current_index].on_enter()
return True
return False
def get_current_content(self) -> Optional[str]:
"""
Get encoded content for current page.
Returns:
Protocol-encoded page content string or None
"""
page = self.get_current_page()
if page is None:
return None
content = page.get_content()
total = self.get_page_count()
index = self.get_current_index()
return Protocol.encode_page(index, total, content)
def process_command(self, command: Command) -> Optional[str]:
"""
Process command from Flipper Zero.
Args:
command: Parsed command
Returns:
Result message or None
"""
self._last_result = None
if command.cmd_type == CommandType.NAV_NEXT:
self.navigate_next()
return None
if command.cmd_type == CommandType.NAV_PREV:
self.navigate_prev()
return None
if command.cmd_type == CommandType.SELECT:
page = self.get_current_page()
if page:
result = page.handle_select(command.select_index)
self._last_result = result
return result
return None
if command.cmd_type == CommandType.CONFIRM:
page = self.get_current_page()
if page:
result = page.handle_confirm()
self._last_result = result
return result
return None
if command.cmd_type == CommandType.CANCEL:
page = self.get_current_page()
if page:
result = page.handle_cancel()
self._last_result = result
return result
return None
if command.cmd_type == CommandType.REFRESH:
# Just return None, content will be sent anyway
return None
return None
def get_last_result(self) -> Optional[str]:
"""Get last action result message."""
return self._last_result
def clear_last_result(self) -> None:
"""Clear last result message."""
self._last_result = None
def get_stats(self) -> Dict[str, Any]:
"""Get manager statistics."""
pages = self.get_enabled_pages()
return {
"total_pages": len(self._pages),
"enabled_pages": len(pages),
"current_index": self._current_index,
"current_page": pages[self._current_index].name if pages else None,
"pages": [p.name for p in pages]
}
def shutdown(self) -> None:
"""Shutdown all pages."""
with self._lock:
for page in self._pages:
page.on_leave()
self._pages.clear()
self._current_index = 0

20
can_sniffer/src/flipper/pages/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
"""
Flipper Zero Pages.
Each page is a plugin that provides content for display on Flipper Zero
and handles user actions.
"""
from flipper.pages.base import BasePage
from flipper.pages.can_stats import CANStatsPage
from flipper.pages.ups_status import UPSStatusPage
from flipper.pages.system_info import SystemInfoPage
from flipper.pages.actions import ActionsPage
__all__ = [
"BasePage",
"CANStatsPage",
"UPSStatusPage",
"SystemInfoPage",
"ActionsPage",
]

163
can_sniffer/src/flipper/pages/actions.py Normal file
View File

@@ -0,0 +1,163 @@
"""
Actions Page.
Menu page for system control actions (shutdown, reboot, etc.)
"""
from typing import Optional, Callable
from flipper.pages.base import BasePage
from flipper.protocol import PageContent, PageType
from flipper.providers.system_provider import SystemProvider
class ActionsPage(BasePage):
"""
Page with system control actions.
Provides menu items for:
- Shutdown
- Reboot
- Cancel pending shutdown
Dangerous actions require confirmation before execution.
"""
# Action identifiers
ACTION_SHUTDOWN = "shutdown"
ACTION_REBOOT = "reboot"
ACTION_CANCEL = "cancel_shutdown"
def __init__(self, on_result: Optional[Callable[[str], None]] = None):
"""
Initialize actions page.
Args:
on_result: Callback when action result is available
"""
super().__init__(
name="actions",
title="Actions",
icon="settings"
)
self._provider = SystemProvider()
self._on_result = on_result
self._confirm_mode = False
self._pending_action: Optional[str] = None
# Menu items: (label, action_id, requires_confirm)
self._menu_items = [
("Shutdown", self.ACTION_SHUTDOWN, True),
("Reboot", self.ACTION_REBOOT, True),
("Cancel Shutdown", self.ACTION_CANCEL, False),
]
def get_content(self) -> PageContent:
"""Get page content."""
if self._confirm_mode and self._pending_action:
# Show confirmation dialog
action_name = self._get_action_label(self._pending_action)
return PageContent(
page_type=PageType.CONFIRM,
title=f"Confirm {action_name}?",
lines=[f"Are you sure you want to {action_name.lower()}?"],
actions=["Yes", "No"],
selected=1, # Default to "No" for safety
icon=self.icon
)
# Normal menu mode
labels = [item[0] for item in self._menu_items]
return PageContent(
page_type=PageType.MENU,
title=self.title,
lines=[],
actions=labels,
selected=self._selected_index,
icon=self.icon
)
def handle_select(self, index: int) -> Optional[str]:
"""Handle menu item selection."""
self._selected_index = index
if 0 <= index < len(self._menu_items):
label, action_id, requires_confirm = self._menu_items[index]
if requires_confirm:
# Enter confirmation mode
self._confirm_mode = True
self._pending_action = action_id
return None # Will show confirm dialog
else:
# Execute immediately
return self._execute_action(action_id)
return None
def handle_confirm(self) -> Optional[str]:
"""Execute confirmed action."""
if self._pending_action:
action_id = self._pending_action
self._confirm_mode = False
self._pending_action = None
return self._execute_action(action_id)
return None
def handle_cancel(self) -> Optional[str]:
"""Cancel pending action."""
self._confirm_mode = False
self._pending_action = None
return "Cancelled"
def has_pending_action(self) -> bool:
"""Check if in confirmation mode."""
return self._confirm_mode
def _execute_action(self, action_id: str) -> str:
"""
Execute the specified action.
Args:
action_id: Action identifier
Returns:
Result message
"""
result = ""
if action_id == self.ACTION_SHUTDOWN:
if self._provider.shutdown(delay_minutes=1):
result = "Shutdown in 1 min"
else:
result = "Shutdown failed"
elif action_id == self.ACTION_REBOOT:
if self._provider.reboot():
result = "Rebooting..."
else:
result = "Reboot failed"
elif action_id == self.ACTION_CANCEL:
if self._provider.cancel_shutdown():
result = "Shutdown cancelled"
else:
result = "No pending shutdown"
# Notify callback if available
if self._on_result:
self._on_result(result)
return result
def _get_action_label(self, action_id: str) -> str:
"""Get human-readable label for action ID."""
for label, aid, _ in self._menu_items:
if aid == action_id:
return label
return action_id
def get_provider(self) -> SystemProvider:
"""Get the System provider instance."""
return self._provider

260
can_sniffer/src/flipper/pages/base.py Normal file
View File

@@ -0,0 +1,260 @@
"""
Base Page Interface for Flipper Zero UI.
All pages must inherit from BasePage and implement required methods.
"""
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any, Callable
from flipper.protocol import PageContent, PageType
class BasePage(ABC):
"""
Abstract base class for Flipper Zero pages.
Each page represents a screen on Flipper Zero display.
Pages can be informational (read-only), interactive menus,
or confirmation dialogs.
Attributes:
name: Unique identifier for the page
title: Display title (shown in header)
icon: Icon identifier for visual representation
enabled: Whether the page is active
"""
def __init__(
self,
name: str,
title: str,
icon: str = "",
enabled: bool = True
):
"""
Initialize base page.
Args:
name: Unique page identifier
title: Display title
icon: Icon identifier
enabled: Whether page is enabled
"""
self.name = name
self.title = title
self.icon = icon
self.enabled = enabled
self._selected_index = 0
self._pending_action: Optional[str] = None
@abstractmethod
def get_content(self) -> PageContent:
"""
Get current page content for display.
Returns:
PageContent object with type, title, lines, and actions
"""
pass
def handle_select(self, index: int) -> Optional[str]:
"""
Handle menu item selection.
Called when user presses OK on a menu item.
Args:
index: Selected item index
Returns:
Optional result message to display, or None
"""
self._selected_index = index
return None
def handle_confirm(self) -> Optional[str]:
"""
Handle confirmation action.
Called when user confirms a pending action.
Returns:
Optional result message, or None
"""
return None
def handle_cancel(self) -> Optional[str]:
"""
Handle cancel action.
Called when user cancels a pending action.
Returns:
Optional result message, or None
"""
self._pending_action = None
return None
def get_selected_index(self) -> int:
"""Get currently selected index for menu pages."""
return self._selected_index
def set_selected_index(self, index: int) -> None:
"""Set selected index for menu pages."""
self._selected_index = index
def has_pending_action(self) -> bool:
"""Check if there's a pending action requiring confirmation."""
return self._pending_action is not None
def get_pending_action(self) -> Optional[str]:
"""Get pending action name."""
return self._pending_action
def is_enabled(self) -> bool:
"""Check if page is enabled."""
return self.enabled
def on_enter(self) -> None:
"""Called when user navigates to this page."""
pass
def on_leave(self) -> None:
"""Called when user navigates away from this page."""
pass
class InfoPage(BasePage):
"""
Base class for information-only pages.
Info pages display read-only data that updates periodically.
They have no interactive elements.
"""
def __init__(
self,
name: str,
title: str,
icon: str = "",
enabled: bool = True
):
super().__init__(name, title, icon, enabled)
@abstractmethod
def get_lines(self) -> list[str]:
"""
Get content lines for display.
Returns:
List of strings to display (max 4-5 lines)
"""
pass
def get_content(self) -> PageContent:
"""Get page content as INFO type."""
return PageContent(
page_type=PageType.INFO,
title=self.title,
lines=self.get_lines(),
icon=self.icon
)
class MenuPage(BasePage):
"""
Base class for menu pages.
Menu pages display selectable items that can trigger actions.
"""
def __init__(
self,
name: str,
title: str,
icon: str = "",
enabled: bool = True
):
super().__init__(name, title, icon, enabled)
self._menu_items: list[tuple[str, Callable[[], Optional[str]]]] = []
def add_item(self, label: str, action: Callable[[], Optional[str]]) -> None:
"""
Add menu item.
Args:
label: Display label for the item
action: Callback function when item is selected
"""
self._menu_items.append((label, action))
def clear_items(self) -> None:
"""Clear all menu items."""
self._menu_items.clear()
self._selected_index = 0
def get_content(self) -> PageContent:
"""Get page content as MENU type."""
labels = [item[0] for item in self._menu_items]
return PageContent(
page_type=PageType.MENU,
title=self.title,
lines=[],
actions=labels,
selected=self._selected_index,
icon=self.icon
)
def handle_select(self, index: int) -> Optional[str]:
"""Execute action for selected menu item."""
self._selected_index = index
if 0 <= index < len(self._menu_items):
_, action = self._menu_items[index]
return action()
return None
class ConfirmPage(BasePage):
"""
Confirmation dialog page.
Used to confirm dangerous actions like shutdown or reboot.
"""
def __init__(
self,
name: str,
title: str,
message: str,
on_confirm: Callable[[], Optional[str]],
on_cancel: Optional[Callable[[], Optional[str]]] = None,
icon: str = "",
):
super().__init__(name, title, icon, enabled=True)
self.message = message
self._on_confirm = on_confirm
self._on_cancel = on_cancel
def get_content(self) -> PageContent:
"""Get page content as CONFIRM type."""
return PageContent(
page_type=PageType.CONFIRM,
title=self.title,
lines=[self.message],
actions=["Yes", "No"],
selected=1, # Default to "No" for safety
icon=self.icon
)
def handle_confirm(self) -> Optional[str]:
"""Execute confirm action."""
return self._on_confirm()
def handle_cancel(self) -> Optional[str]:
"""Execute cancel action."""
if self._on_cancel:
return self._on_cancel()
return None

66
can_sniffer/src/flipper/pages/can_stats.py Normal file
View File

@@ -0,0 +1,66 @@
"""
CAN Statistics Page.
Displays CAN sniffer statistics on Flipper Zero.
"""
from flipper.pages.base import InfoPage
from flipper.providers.can_provider import CANProvider
class CANStatsPage(InfoPage):
"""
Page displaying CAN bus statistics.
Shows:
- Total frames received
- Pending/processed frames
- Queue status
- Dropped frames (if any)
"""
def __init__(self):
super().__init__(
name="can_stats",
title="CAN Statistics",
icon="can"
)
self._provider = CANProvider()
def get_lines(self) -> list[str]:
"""Get statistics lines for display."""
data = self._provider.get_data()
lines = [
f"Total: {self._format_number(data.total_frames)}",
f"Processed: {self._format_number(data.processed_frames)}",
f"Queue: {data.queue_size}/{data.queue_capacity}",
]
# Add interface breakdown if available
if data.interfaces:
iface_str = ", ".join(
f"{k}: {self._format_number(v)}"
for k, v in data.interfaces.items()
)
if len(iface_str) <= 25:
lines.append(iface_str)
# Show dropped count if non-zero
if data.dropped_frames > 0:
lines.append(f"Dropped: {data.dropped_frames}")
return lines
def _format_number(self, num: int) -> str:
"""Format large numbers with K/M suffix."""
if num >= 1_000_000:
return f"{num / 1_000_000:.1f}M"
elif num >= 1_000:
return f"{num / 1_000:.1f}K"
else:
return str(num)
def get_provider(self) -> CANProvider:
"""Get the CAN provider instance."""
return self._provider

72
can_sniffer/src/flipper/pages/system_info.py Normal file
View File

@@ -0,0 +1,72 @@
"""
System Information Page.
Displays Raspberry Pi system metrics on Flipper Zero.
"""
from flipper.pages.base import InfoPage
from flipper.providers.system_provider import SystemProvider
class SystemInfoPage(InfoPage):
"""
Page displaying Raspberry Pi system metrics.
Shows:
- CPU temperature
- Power consumption
- Fan RPM
- Input voltage
"""
def __init__(self):
super().__init__(
name="system_info",
title="System Info",
icon="cpu"
)
self._provider = SystemProvider()
def get_lines(self) -> list[str]:
"""Get system info lines for display."""
data = self._provider.get_data()
lines = [
f"CPU Temp: {data.cpu_temp:.1f}C {self._get_temp_indicator(data.cpu_temp)}",
f"Power: {data.power_watts:.2f}W",
]
# Fan RPM (may not be available on all systems)
if data.fan_rpm > 0:
lines.append(f"Fan: {data.fan_rpm} RPM")
else:
lines.append("Fan: N/A")
# Input voltage
if data.input_voltage > 0:
lines.append(f"Input: {data.input_voltage:.2f}V")
return lines
def _get_temp_indicator(self, temp: float) -> str:
"""
Get temperature status indicator.
Args:
temp: Temperature in Celsius
Returns:
Status indicator string
"""
if temp >= 80:
return "(!)" # Critical
elif temp >= 70:
return "(W)" # Warning
elif temp >= 60:
return "(~)" # Warm
else:
return "" # OK
def get_provider(self) -> SystemProvider:
"""Get the System provider instance."""
return self._provider

85
can_sniffer/src/flipper/pages/ups_status.py Normal file
View File

@@ -0,0 +1,85 @@
"""
UPS Status Page.
Displays X120x UPS battery and power status on Flipper Zero.
"""
from flipper.pages.base import InfoPage
from flipper.providers.ups_provider import UPSProvider
class UPSStatusPage(InfoPage):
"""
Page displaying UPS (X120x) status.
Shows:
- Battery percentage and voltage
- Charging status
- Power loss indicator
- Input voltage
"""
def __init__(self):
super().__init__(
name="ups_status",
title="UPS Status",
icon="battery"
)
self._provider = UPSProvider()
def get_lines(self) -> list[str]:
"""Get UPS status lines for display."""
if not self._provider.is_available():
return [
"UPS not available",
self._provider.get_last_error() or "Check I2C connection"
]
data = self._provider.get_data()
# Battery bar visualization
battery_bar = self._get_battery_bar(data.capacity)
lines = [
f"Bat: {data.capacity:.1f}% {battery_bar}",
f"Voltage: {data.voltage:.2f}V",
]
# Charging/power status
if data.power_loss:
lines.append("Power: BATTERY MODE")
if data.capacity < 15:
lines.append("WARNING: Critical!")
elif data.capacity < 25:
lines.append("WARNING: Low battery")
else:
status = "Charging" if data.is_charging else "Full"
lines.append(f"Status: {status}")
lines.append(f"Input: {data.input_voltage:.2f}V")
return lines
def _get_battery_bar(self, percent: float) -> str:
"""
Create ASCII battery bar.
Args:
percent: Battery percentage (0-100)
Returns:
Battery visualization string like "[==== ]"
"""
filled = int(percent / 20) # 5 segments
empty = 5 - filled
bar = "[" + "=" * filled + " " * empty + "]"
return bar
def is_enabled(self) -> bool:
"""Check if page should be shown."""
return self.enabled and self._provider.is_available()
def get_provider(self) -> UPSProvider:
"""Get the UPS provider instance."""
return self._provider

194
can_sniffer/src/flipper/protocol.py Normal file
View File

@@ -0,0 +1,194 @@
"""
Flipper Zero Communication Protocol.
Defines message formats for RPi <-> Flipper Zero UART communication.
Protocol Messages:
RPi -> Flipper:
PAGE:<idx>/<total>|<type>|<title>|<lines>|<actions>|<selected>
ACK:<device>,ip=<ip>
RESULT:<status>|<message>
Flipper -> RPi:
INIT:<device>
STOP:<device>
CMD:NAV:<next|prev>
CMD:SELECT:<index>
CMD:CONFIRM
CMD:CANCEL
CMD:REFRESH
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional
class PageType(Enum):
"""Types of pages for different UI layouts."""
INFO = "info" # Read-only information display
MENU = "menu" # Selectable menu items
CONFIRM = "confirm" # Confirmation dialog (Yes/No)
class CommandType(Enum):
"""Types of commands from Flipper to RPi."""
INIT = "INIT"
STOP = "STOP"
NAV_NEXT = "NAV:next"
NAV_PREV = "NAV:prev"
SELECT = "SELECT"
CONFIRM = "CONFIRM"
CANCEL = "CANCEL"
REFRESH = "REFRESH"
@dataclass
class PageContent:
"""
Content structure for a page to be displayed on Flipper Zero.
Attributes:
page_type: Type of page (info, menu, confirm)
title: Page title (max ~20 chars for display)
lines: Content lines to display (max 4-5 lines)
actions: Available actions/menu items (for menu type)
selected: Currently selected item index (for menu type)
icon: Optional icon identifier for the page
"""
page_type: PageType
title: str
lines: List[str] = field(default_factory=list)
actions: List[str] = field(default_factory=list)
selected: int = 0
icon: str = ""
def __post_init__(self):
# Truncate title if too long
if len(self.title) > 20:
self.title = self.title[:17] + "..."
@dataclass
class Command:
"""Parsed command from Flipper Zero."""
cmd_type: CommandType
payload: str = ""
@property
def select_index(self) -> int:
"""Get selected index for SELECT command."""
if self.cmd_type == CommandType.SELECT:
try:
return int(self.payload)
except ValueError:
return 0
return 0
class Protocol:
"""
Protocol encoder/decoder for Flipper Zero communication.
Message format uses pipe (|) as field separator and semicolon (;)
as item separator within fields.
"""
FIELD_SEP = "|"
ITEM_SEP = ";"
@staticmethod
def encode_page(
page_index: int,
total_pages: int,
content: PageContent
) -> str:
"""
Encode page content to protocol message.
Args:
page_index: Current page index (0-based)
total_pages: Total number of pages
content: Page content to encode
Returns:
Encoded message string with newline
"""
# PAGE:<idx>/<total>|<type>|<title>|<lines>|<actions>|<selected>
lines_str = Protocol.ITEM_SEP.join(content.lines)
actions_str = Protocol.ITEM_SEP.join(content.actions)
parts = [
f"PAGE:{page_index}/{total_pages}",
content.page_type.value,
content.title,
lines_str,
actions_str,
str(content.selected)
]
return Protocol.FIELD_SEP.join(parts) + "\n"
@staticmethod
def encode_ack(device: str, ip: str) -> str:
"""Encode ACK message."""
return f"ACK:{device},ip={ip}\n"
@staticmethod
def encode_result(success: bool, message: str) -> str:
"""Encode result message after action execution."""
status = "OK" if success else "ERROR"
return f"RESULT:{status}|{message}\n"
@staticmethod
def decode_command(line: str) -> Optional[Command]:
"""
Decode command from Flipper Zero.
Args:
line: Raw command line (without newline)
Returns:
Parsed Command object or None if invalid
"""
line = line.strip()
if not line:
return None
# INIT:<device>
if line.startswith("INIT:"):
return Command(CommandType.INIT, line[5:])
# STOP:<device>
if line.startswith("STOP:"):
return Command(CommandType.STOP, line[5:])
# CMD:NAV:next / CMD:NAV:prev
if line == "CMD:NAV:next":
return Command(CommandType.NAV_NEXT)
if line == "CMD:NAV:prev":
return Command(CommandType.NAV_PREV)
# CMD:SELECT:<index>
if line.startswith("CMD:SELECT:"):
return Command(CommandType.SELECT, line[11:])
# CMD:CONFIRM
if line == "CMD:CONFIRM":
return Command(CommandType.CONFIRM)
# CMD:CANCEL
if line == "CMD:CANCEL":
return Command(CommandType.CANCEL)
# CMD:REFRESH
if line == "CMD:REFRESH":
return Command(CommandType.REFRESH)
return None

18
can_sniffer/src/flipper/providers/__init__.py Normal file
View File

@@ -0,0 +1,18 @@
"""
Data Providers for Flipper Zero Pages.
Providers abstract data sources (UPS, system metrics, CAN stats)
from the pages that display them.
"""
from flipper.providers.base import BaseProvider
from flipper.providers.ups_provider import UPSProvider
from flipper.providers.system_provider import SystemProvider
from flipper.providers.can_provider import CANProvider
__all__ = [
"BaseProvider",
"UPSProvider",
"SystemProvider",
"CANProvider",
]

119
can_sniffer/src/flipper/providers/base.py Normal file
View File

@@ -0,0 +1,119 @@
"""
Base Provider Interface.
Providers are singleton data sources that pages can query.
They handle caching and thread-safe data access.
"""
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import threading
import time
class BaseProvider(ABC):
"""
Abstract base class for data providers.
Providers:
- Are singletons (one instance per type)
- Cache data with configurable TTL
- Are thread-safe
- Handle errors gracefully
Attributes:
name: Provider identifier
cache_ttl: Cache time-to-live in seconds
"""
_instances: Dict[str, "BaseProvider"] = {}
_lock = threading.Lock()
def __new__(cls, *args, **kwargs):
"""Singleton pattern implementation."""
with cls._lock:
if cls.__name__ not in cls._instances:
instance = super().__new__(cls)
cls._instances[cls.__name__] = instance
return cls._instances[cls.__name__]
def __init__(self, name: str, cache_ttl: float = 1.0):
"""
Initialize provider.
Args:
name: Provider identifier
cache_ttl: Cache TTL in seconds
"""
# Prevent re-initialization for singleton
if hasattr(self, "_initialized") and self._initialized:
return
self.name = name
self.cache_ttl = cache_ttl
self._cache: Dict[str, Any] = {}
self._cache_timestamps: Dict[str, float] = {}
self._data_lock = threading.Lock()
self._initialized = True
self._available = True
self._last_error: Optional[str] = None
def _get_cached(self, key: str) -> Optional[Any]:
"""
Get cached value if not expired.
Args:
key: Cache key
Returns:
Cached value or None if expired/missing
"""
with self._data_lock:
if key not in self._cache:
return None
timestamp = self._cache_timestamps.get(key, 0)
if time.time() - timestamp > self.cache_ttl:
return None
return self._cache[key]
def _set_cached(self, key: str, value: Any) -> None:
"""
Set cached value.
Args:
key: Cache key
value: Value to cache
"""
with self._data_lock:
self._cache[key] = value
self._cache_timestamps[key] = time.time()
def _clear_cache(self) -> None:
"""Clear all cached values."""
with self._data_lock:
self._cache.clear()
self._cache_timestamps.clear()
@abstractmethod
def refresh(self) -> bool:
"""
Refresh provider data from source.
Returns:
True if refresh successful
"""
pass
def is_available(self) -> bool:
"""Check if provider is available."""
return self._available
def get_last_error(self) -> Optional[str]:
"""Get last error message."""
return self._last_error
def shutdown(self) -> None:
"""Cleanup provider resources."""
self._clear_cache()

135
can_sniffer/src/flipper/providers/can_provider.py Normal file
View File

@@ -0,0 +1,135 @@
"""
CAN Statistics Provider.
Provides CAN sniffer statistics from the message processor.
"""
from typing import Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from flipper.providers.base import BaseProvider
@dataclass
class CANData:
"""CAN statistics data."""
total_frames: int = 0
pending_frames: int = 0
processed_frames: int = 0
dropped_frames: int = 0
queue_size: int = 0
queue_capacity: int = 100000
interfaces: Dict[str, int] = field(default_factory=dict)
class CANProvider(BaseProvider):
"""
Provider for CAN sniffer statistics.
This provider is updated by the FlipperHandler when
CAN frames are processed.
"""
def __init__(self):
super().__init__(name="can", cache_ttl=0.5)
self._data = CANData()
self._stats_callback: Optional[Callable[[], Dict[str, Any]]] = None
def set_stats_callback(self, callback: Callable[[], Dict[str, Any]]) -> None:
"""
Set callback to retrieve stats from message processor.
Args:
callback: Function that returns stats dictionary
"""
self._stats_callback = callback
def update_stats(
self,
total: int = 0,
pending: int = 0,
processed: int = 0,
dropped: int = 0,
queue_size: int = 0
) -> None:
"""
Update CAN statistics directly.
Args:
total: Total frames received
pending: Pending frames in queue
processed: Processed frames
dropped: Dropped frames
queue_size: Current queue size
"""
self._data.total_frames = total
self._data.pending_frames = pending
self._data.processed_frames = processed
self._data.dropped_frames = dropped
self._data.queue_size = queue_size
self._set_cached("data", self._data)
def update_interface_stats(self, interface: str, count: int) -> None:
"""
Update per-interface statistics.
Args:
interface: Interface name (e.g., "can0")
count: Message count for this interface
"""
self._data.interfaces[interface] = count
def refresh(self) -> bool:
"""Refresh stats from callback if available."""
if self._stats_callback:
try:
stats = self._stats_callback()
self._data.total_frames = stats.get("total_frames", 0)
self._data.pending_frames = stats.get("pending_frames", 0)
self._data.processed_frames = stats.get("processed_frames", 0)
self._data.dropped_frames = stats.get("dropped_count", 0)
self._data.queue_size = stats.get("queue_size", 0)
self._set_cached("data", self._data)
return True
except Exception as e:
self._last_error = str(e)
return False
return True
def get_data(self) -> CANData:
"""Get current CAN statistics."""
cached = self._get_cached("data")
if cached is not None:
return cached
self.refresh()
return self._data
def get_total_frames(self) -> int:
"""Get total frames received."""
return self.get_data().total_frames
def get_pending_frames(self) -> int:
"""Get pending frames in queue."""
return self.get_data().pending_frames
def get_processed_frames(self) -> int:
"""Get processed frames."""
return self.get_data().processed_frames
def get_dropped_frames(self) -> int:
"""Get dropped frames."""
return self.get_data().dropped_frames
def get_queue_fill_percent(self) -> float:
"""Get queue fill percentage."""
data = self.get_data()
if data.queue_capacity == 0:
return 0.0
return (data.queue_size / data.queue_capacity) * 100
def get_interface_count(self, interface: str) -> int:
"""Get message count for specific interface."""
return self.get_data().interfaces.get(interface, 0)

240
can_sniffer/src/flipper/providers/system_provider.py Normal file
View File

@@ -0,0 +1,240 @@
"""
System Provider for Raspberry Pi metrics and control.
Provides CPU temperature, power consumption, fan RPM,
and system control actions (shutdown, reboot).
"""
import os
import subprocess
from typing import Optional
from dataclasses import dataclass
from pathlib import Path
from flipper.providers.base import BaseProvider
@dataclass
class SystemData:
"""System metrics data."""
cpu_temp: float = 0.0
cpu_volts: float = 0.0
cpu_amps: float = 0.0
power_watts: float = 0.0
fan_rpm: int = 0
input_voltage: float = 0.0
class SystemProvider(BaseProvider):
"""
Provider for Raspberry Pi system metrics.
Uses vcgencmd for hardware metrics and sysfs for fan speed.
Also provides system control actions.
"""
def __init__(self):
super().__init__(name="system", cache_ttl=2.0)
self._data = SystemData()
def _read_metric(self, args: list, strip_chars: str) -> Optional[float]:
"""
Read a hardware metric using vcgencmd.
Args:
args: Command arguments
strip_chars: Characters to strip from value
Returns:
Float value or None on error
"""
try:
output = subprocess.check_output(args).decode("utf-8")
value_str = output.split("=")[1].strip().rstrip(strip_chars)
return float(value_str)
except Exception:
return None
def _read_cpu_temp(self) -> float:
"""Read CPU temperature in Celsius."""
result = self._read_metric(["vcgencmd", "measure_temp"], "'C")
return result if result is not None else 0.0
def _read_cpu_volts(self) -> float:
"""Read CPU core voltage."""
result = self._read_metric(["vcgencmd", "pmic_read_adc", "VDD_CORE_V"], "V")
return result if result is not None else 0.0
def _read_cpu_amps(self) -> float:
"""Read CPU core current."""
result = self._read_metric(["vcgencmd", "pmic_read_adc", "VDD_CORE_A"], "A")
return result if result is not None else 0.0
def _read_input_voltage(self) -> float:
"""Read external 5V input voltage."""
result = self._read_metric(["vcgencmd", "pmic_read_adc", "EXT5V_V"], "V")
return result if result is not None else 0.0
def _read_power_watts(self) -> float:
"""
Calculate total system power consumption.
Reads all PMIC ADC values and calculates wattage.
"""
try:
output = subprocess.check_output(["vcgencmd", "pmic_read_adc"]).decode("utf-8")
lines = output.strip().split("\n")
amperages = {}
voltages = {}
for line in lines:
line = line.strip()
if not line:
continue
parts = line.split()
if len(parts) < 2:
continue
label = parts[0]
value_str = parts[-1]
if "=" not in value_str:
continue
val = float(value_str.split("=")[1][:-1])
short_label = label[:-2]
if label.endswith("A"):
amperages[short_label] = val
else:
voltages[short_label] = val
# Calculate total wattage
wattage = sum(
amperages[key] * voltages[key]
for key in amperages
if key in voltages
)
return wattage
except Exception:
return 0.0
def _read_fan_rpm(self) -> int:
"""
Read fan RPM from sysfs.
Returns:
Fan speed in RPM or 0 if not available
"""
try:
sys_path = Path("/sys/devices/platform/cooling_fan")
fan_files = list(sys_path.rglob("fan1_input"))
if not fan_files:
return 0
with open(fan_files[0], "r") as f:
rpm = f.read().strip()
return int(rpm)
except Exception:
return 0
def refresh(self) -> bool:
"""Refresh system metrics."""
try:
self._data = SystemData(
cpu_temp=self._read_cpu_temp(),
cpu_volts=self._read_cpu_volts(),
cpu_amps=self._read_cpu_amps(),
power_watts=self._read_power_watts(),
fan_rpm=self._read_fan_rpm(),
input_voltage=self._read_input_voltage()
)
self._set_cached("data", self._data)
return True
except Exception as e:
self._last_error = str(e)
return False
def get_data(self) -> SystemData:
"""Get current system data."""
cached = self._get_cached("data")
if cached is not None:
return cached
self.refresh()
return self._data
def get_cpu_temp(self) -> float:
"""Get CPU temperature in Celsius."""
return self.get_data().cpu_temp
def get_power_watts(self) -> float:
"""Get total power consumption in Watts."""
return self.get_data().power_watts
def get_fan_rpm(self) -> int:
"""Get fan speed in RPM."""
return self.get_data().fan_rpm
def shutdown(self, delay_minutes: int = 0) -> bool:
"""
Shutdown the system.
Args:
delay_minutes: Delay before shutdown (0 = immediate)
Returns:
True if command executed
"""
try:
if delay_minutes > 0:
cmd = f"sudo shutdown -P +{delay_minutes}"
else:
cmd = "sudo shutdown -P now"
subprocess.Popen(cmd, shell=True)
return True
except Exception as e:
self._last_error = str(e)
return False
def reboot(self) -> bool:
"""
Reboot the system.
Returns:
True if command executed
"""
try:
subprocess.Popen("sudo reboot", shell=True)
return True
except Exception as e:
self._last_error = str(e)
return False
def cancel_shutdown(self) -> bool:
"""
Cancel pending shutdown.
Returns:
True if command executed
"""
try:
subprocess.call("sudo shutdown -c", shell=True)
return True
except Exception as e:
self._last_error = str(e)
return False

228
can_sniffer/src/flipper/providers/ups_provider.py Normal file
View File

@@ -0,0 +1,228 @@
"""
UPS Provider for X120x UPS on Raspberry Pi.
Reads battery status, voltage, and charging state via I2C (SMBus).
Based on https://github.com/suptronics/x120x
"""
import struct
from typing import Optional, Tuple
from dataclasses import dataclass
from flipper.providers.base import BaseProvider
@dataclass
class UPSData:
"""UPS status data."""
voltage: float = 0.0
capacity: float = 0.0
is_charging: bool = False
power_loss: bool = False
input_voltage: float = 0.0
class UPSProvider(BaseProvider):
"""
Provider for X120x UPS data.
Reads from MAX17048 fuel gauge via I2C (address 0x36).
Also monitors power loss detection pin (GPIO 6).
Hardware requirements:
- X120x UPS HAT connected via I2C
- smbus2 library installed
- gpiozero for power loss detection
"""
# I2C address for MAX17048 fuel gauge
I2C_ADDRESS = 0x36
I2C_BUS = 1
# GPIO pins
CHG_ONOFF_PIN = 16
PLD_PIN = 6 # Power Loss Detection
def __init__(self):
super().__init__(name="ups", cache_ttl=2.0)
self._bus = None
self._pld_button = None
self._data = UPSData()
self._init_hardware()
def _init_hardware(self) -> None:
"""Initialize I2C and GPIO."""
try:
import smbus2
self._bus = smbus2.SMBus(self.I2C_BUS)
self._available = True
except ImportError:
self._last_error = "smbus2 not installed"
self._available = False
except Exception as e:
self._last_error = f"I2C init failed: {e}"
self._available = False
try:
from gpiozero import Button
self._pld_button = Button(self.PLD_PIN)
except ImportError:
pass # GPIO optional
except Exception:
pass # GPIO optional
def _read_voltage_and_capacity(self) -> Tuple[float, float]:
"""
Read voltage and capacity from MAX17048.
Returns:
Tuple of (voltage in V, capacity in %)
"""
if not self._bus:
return 0.0, 0.0
try:
# Read voltage register (0x02)
voltage_read = self._bus.read_word_data(self.I2C_ADDRESS, 0x02)
# Read capacity register (0x04)
capacity_read = self._bus.read_word_data(self.I2C_ADDRESS, 0x04)
# Swap bytes (little-endian to big-endian)
voltage_swapped = struct.unpack("<H", struct.pack(">H", voltage_read))[0]
capacity_swapped = struct.unpack("<H", struct.pack(">H", capacity_read))[0]
# Calculate actual values
voltage = voltage_swapped * 1.25 / 1000 / 16
capacity = capacity_swapped / 256
return voltage, capacity
except Exception as e:
self._last_error = f"Read error: {e}"
return 0.0, 0.0
def _get_power_loss_state(self) -> bool:
"""
Check power loss detection pin.
Returns:
True if power loss detected (running on battery)
"""
if self._pld_button is None:
return False
try:
# Button pressed = power OK, not pressed = power loss
return not self._pld_button.is_pressed
except Exception:
return False
def _read_input_voltage(self) -> float:
"""
Read input voltage using vcgencmd.
Returns:
Input voltage in V
"""
try:
from subprocess import check_output
output = check_output(["vcgencmd", "pmic_read_adc", "EXT5V_V"]).decode("utf-8")
value_str = output.split("=")[1].strip().rstrip("V")
return float(value_str)
except Exception:
return 0.0
def refresh(self) -> bool:
"""Refresh UPS data from hardware."""
if not self._available:
return False
try:
voltage, capacity = self._read_voltage_and_capacity()
power_loss = self._get_power_loss_state()
input_voltage = self._read_input_voltage()
# Determine charging state (charge disabled above 90%)
is_charging = capacity < 90 and not power_loss
self._data = UPSData(
voltage=voltage,
capacity=capacity,
is_charging=is_charging,
power_loss=power_loss,
input_voltage=input_voltage
)
self._set_cached("data", self._data)
return True
except Exception as e:
self._last_error = str(e)
return False
def get_data(self) -> UPSData:
"""
Get current UPS data.
Returns:
UPSData object with current status
"""
cached = self._get_cached("data")
if cached is not None:
return cached
self.refresh()
return self._data
def get_voltage(self) -> float:
"""Get battery voltage in V."""
return self.get_data().voltage
def get_capacity(self) -> float:
"""Get battery capacity in %."""
return self.get_data().capacity
def is_charging(self) -> bool:
"""Check if battery is charging."""
return self.get_data().is_charging
def has_power_loss(self) -> bool:
"""Check if external power is lost."""
return self.get_data().power_loss
def get_input_voltage(self) -> float:
"""Get input voltage in V."""
return self.get_data().input_voltage
def get_status_string(self) -> str:
"""
Get human-readable status string.
Returns:
Status string like "OK", "Battery", "Charging"
"""
data = self.get_data()
if data.power_loss:
if data.capacity < 15:
return "CRITICAL"
elif data.capacity < 25:
return "LOW"
else:
return "Battery"
elif data.is_charging:
return "Charging"
else:
return "OK"
def shutdown(self) -> None:
"""Cleanup resources."""
super().shutdown()
if self._bus:
try:
self._bus.close()
except Exception:
pass

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

@@ -1,15 +1,22 @@
"""
Flipper Zero UART Handler with Handshake Protocol.
Flipper Zero Dynamic UI Handler.
Waits for INIT command from Flipper Zero before sending statistics.
Supports handshake protocol for secure connection establishment.
Provides multi-page interface with bidirectional communication via UART.
Supports pluggable pages for CAN stats, UPS status, system info, and actions.
Protocol:
1. RPI5 waits in passive mode, listening for commands
2. Flipper sends: INIT:flipper\n
3. RPI5 responds: ACK:rpi5,ip=x.x.x.x\n
4. RPI5 starts sending: STATS:ip=...,total=...,pending=...,processed=...\n
5. Flipper sends: STOP:flipper\n to disconnect
RPi -> Flipper:
PAGE:<idx>/<total>|<type>|<title>|<lines>|<actions>|<selected>
ACK:<device>,ip=<ip>
RESULT:<status>|<message>
Flipper -> RPi:
INIT:<device>
STOP:<device>
CMD:NAV:<next|prev>
CMD:SELECT:<index>
CMD:CONFIRM / CMD:CANCEL
CMD:REFRESH
"""
import socket
@@ -22,6 +29,10 @@ from can_frame import CANFrame
from config import config
from logger import get_logger
from flipper.protocol import Protocol, Command, CommandType
from flipper.page_manager import PageManager
from flipper.pages import CANStatsPage, UPSStatusPage, SystemInfoPage, ActionsPage
logger = get_logger(__name__)
@@ -57,16 +68,14 @@ class FlipperHandler(BaseHandler):
"""
Handler that communicates with Flipper Zero via UART.
Implements handshake protocol:
- Waits for INIT:flipper command
- Responds with ACK:rpi5,ip=x.x.x.x
- Sends STATS periodically while connected
- Stops on STOP:flipper command
Provides dynamic multi-page interface:
- CAN Statistics
- UPS Status (if available)
- System Information
- Actions Menu
UART Configuration:
- Device: /dev/ttyAMA0 (or configured device)
- Baud: 115200
- Format: 8N1
Implements handshake protocol for connection management
and bidirectional command processing.
"""
def __init__(self, enabled: Optional[bool] = None):
@@ -83,6 +92,7 @@ class FlipperHandler(BaseHandler):
super().__init__(name="flipper_handler", enabled=enabled)
# Serial configuration
self.serial_port: Optional[Any] = None
self.device = "/dev/ttyAMA0"
self.baudrate = 115200
@@ -113,6 +123,39 @@ class FlipperHandler(BaseHandler):
# IP address
self._ip_address = "0.0.0.0"
# Page manager
self._page_manager = PageManager()
self._setup_pages()
def _setup_pages(self) -> None:
"""Setup default pages."""
# CAN Statistics (always available)
can_page = CANStatsPage()
self._page_manager.register_page(can_page)
# Keep reference to CAN provider for stats updates
self._can_provider = can_page.get_provider()
# UPS Status (if available)
ups_page = UPSStatusPage()
self._page_manager.register_page(ups_page)
# System Information
system_page = SystemInfoPage()
self._page_manager.register_page(system_page)
# Actions Menu
actions_page = ActionsPage(on_result=self._on_action_result)
self._page_manager.register_page(actions_page)
def _on_action_result(self, result: str) -> None:
"""Handle action result from actions page."""
self.logger.info(f"Action result: {result}")
# Send result to Flipper
if self._connected:
msg = Protocol.encode_result(True, result)
self._send_raw(msg)
def initialize(self) -> bool:
"""
Initialize UART connection.
@@ -162,13 +205,15 @@ class FlipperHandler(BaseHandler):
)
self._rx_thread.start()
# Start TX thread (sends stats when connected)
# Start TX thread (sends page content when connected)
self._tx_thread = threading.Thread(
target=self._tx_loop, name="FlipperTX", daemon=True
)
self._tx_thread.start()
self.logger.info("Flipper handler started, waiting for connection...")
self.logger.info(
f"Flipper handler started with {self._page_manager.get_page_count()} pages"
)
def _rx_loop(self) -> None:
"""Receive loop - listens for commands from Flipper."""
@@ -198,41 +243,61 @@ class FlipperHandler(BaseHandler):
self.logger.debug(f"RX error: {e}")
time.sleep(0.1)
def _process_command(self, command: str) -> None:
def _process_command(self, raw_command: str) -> None:
"""
Process received command from Flipper.
Args:
command: Received command string
raw_command: Raw command string
"""
self.logger.info(f"Received command: {command}")
self.logger.debug(f"RX: {raw_command}")
if command.startswith("INIT:"):
# Handshake initiation
client_id = command[5:].strip()
# Handle handshake commands directly
if raw_command.startswith("INIT:"):
client_id = raw_command[5:].strip()
self.logger.info(f"Handshake request from: {client_id}")
# Send ACK with IP address
# Update IP and send ACK
self._ip_address = get_ip_address()
ack_msg = f"ACK:rpi5,ip={self._ip_address}\n"
ack_msg = Protocol.encode_ack("rpi5", self._ip_address)
self._send_raw(ack_msg)
self._connected = True
self.logger.info(f"Connected to Flipper, IP: {self._ip_address}")
elif command.startswith("STOP:"):
# Disconnect request
client_id = command[5:].strip()
# Send initial page content
self._send_page_content()
return
if raw_command.startswith("STOP:"):
client_id = raw_command[5:].strip()
self.logger.info(f"Disconnect request from: {client_id}")
self._connected = False
self.logger.info("Disconnected from Flipper")
return
# Parse and process other commands
command = Protocol.decode_command(raw_command)
if command is None:
self.logger.debug(f"Unknown command: {raw_command}")
return
# Process command via page manager
result = self._page_manager.process_command(command)
# Send result if available
if result:
msg = Protocol.encode_result(True, result)
self._send_raw(msg)
# Always send updated page content after command
self._send_page_content()
def _tx_loop(self) -> None:
"""Transmit loop - sends stats when connected."""
"""Transmit loop - sends page content periodically when connected."""
while self._running:
try:
if self._connected:
self._send_stats()
self._send_page_content()
time.sleep(self.send_interval)
@@ -263,21 +328,16 @@ class FlipperHandler(BaseHandler):
self.logger.debug(f"Send error: {e}")
return False
def _send_stats(self) -> None:
"""Send current statistics to Flipper Zero."""
def _send_page_content(self) -> None:
"""Send current page content to Flipper Zero."""
if not self._connected:
return
with self._stats_lock:
total = self._total_frames
pending = self._pending_frames
processed = self._processed_frames
message = f"STATS:ip={self._ip_address},total={total},pending={pending},processed={processed}\n"
if self._send_raw(message):
with self._stats_lock:
self._sent_count += 1
content = self._page_manager.get_current_content()
if content:
if self._send_raw(content):
with self._stats_lock:
self._sent_count += 1
def handle(self, frame: CANFrame) -> bool:
"""
@@ -292,6 +352,14 @@ class FlipperHandler(BaseHandler):
with self._stats_lock:
self._total_frames += 1
self._pending_frames += 1
# Update CAN provider
self._can_provider.update_stats(
total=self._total_frames,
pending=self._pending_frames,
processed=self._processed_frames
)
return True
def handle_batch(self, frames: List[CANFrame]) -> int:
@@ -305,10 +373,19 @@ class FlipperHandler(BaseHandler):
Number of frames processed
"""
count = len(frames)
with self._stats_lock:
self._total_frames += count
self._processed_frames += count
self._pending_frames = max(0, self._pending_frames - count)
# Update CAN provider
self._can_provider.update_stats(
total=self._total_frames,
pending=self._pending_frames,
processed=self._processed_frames
)
return count
def update_pending(self, pending_count: int) -> None:
@@ -321,11 +398,17 @@ class FlipperHandler(BaseHandler):
with self._stats_lock:
self._pending_frames = pending_count
self._can_provider.update_stats(
total=self._total_frames,
pending=self._pending_frames,
processed=self._processed_frames
)
def flush(self) -> None:
"""Flush - send immediate stats if connected."""
"""Flush - send immediate page content if connected."""
if self._connected:
try:
self._send_stats()
self._send_page_content()
except Exception as e:
self.logger.debug(f"Flush error: {e}")
@@ -350,6 +433,9 @@ class FlipperHandler(BaseHandler):
except Exception as e:
self.logger.debug(f"Error closing serial port: {e}")
# Shutdown page manager
self._page_manager.shutdown()
self._initialized = False
self.logger.info("Flipper handler stopped")
@@ -361,7 +447,7 @@ class FlipperHandler(BaseHandler):
Dictionary with handler stats
"""
with self._stats_lock:
return {
stats = {
"total_frames": self._total_frames,
"pending_frames": self._pending_frames,
"processed_frames": self._processed_frames,
@@ -373,6 +459,15 @@ class FlipperHandler(BaseHandler):
"ip_address": self._ip_address,
}
# Add page manager stats
stats.update(self._page_manager.get_stats())
return stats
def is_connected(self) -> bool:
"""Check if Flipper is connected."""
return self._connected
def get_page_manager(self) -> PageManager:
"""Get page manager for external page registration."""
return self._page_manager