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

rename and remove pages from flipper zero

Browse Source
This commit is contained in:
Alexander Poletaev
2026-01-30 17:06:00 +03:00
parent cfccf5e75c
commit f611c2a6c6
1 changed files with 93 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

235
obd2_client/src/flipper/pages.py
View File

@@ -50,49 +50,35 @@ class PageManager:
def _register_default_pages(self) -> None:
"""Register default page definitions."""
# Page 0: Live Vehicle Data
# Page 0: Car Data (OBD2 + UDS combined)
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="Live Data",
generator=self._generate_live_data_page,
title="Car Data",
generator=self._generate_car_data_page,
))
# Page 1: Statistics
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="Statistics",
generator=self._generate_stats_page,
))
# Page 2: System Info
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="System Info",
generator=self._generate_system_page,
))
# Page 3: App Status (handlers, storage, sync state)
# Page 1: App Status (handlers, storage, sync state, rate)
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="App Status",
generator=self._generate_app_status_page,
))
# Page 4: UDS Extended Data
# Page 2: System Info
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="UDS Data",
generator=self._generate_uds_page,
title="System",
generator=self._generate_system_page,
))
# Page 5: UPS Status
# Page 3: UPS Status
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="UPS Status",
title="UPS",
generator=self._generate_ups_page,
))
# Page 6: Actions Menu (last navigable page)
# Page 4: Actions Menu (last navigable page)
self._pages.append(PageDefinition(
page_type=PageType.MENU,
title="Actions",
@@ -105,7 +91,7 @@ class PageManager:
],
))
# Page 6: Confirm (dynamic, hidden from navigation)
# Page 5: Confirm (dynamic, hidden from navigation)
self._pages.append(PageDefinition(
page_type=PageType.CONFIRM,
title="Confirm",
@@ -350,14 +336,14 @@ class PageManager:
if self._pending_action:
action = self._pending_action
self._pending_action = None
self._current_index = 6 # Back to actions menu
self._current_index = self.total_pages - 1 # Back to actions menu
return self.execute_action(action)
return False, "No pending action"
def cancel_action(self) -> None:
"""Cancel pending action."""
self._pending_action = None
self._current_index = 6 # Back to actions menu
self._current_index = self.total_pages - 1 # Back to actions menu
def _get_action_success_message(self, action_id: ActionID) -> str:
"""Get success message for action."""
@@ -373,34 +359,37 @@ class PageManager:
# Page generators
def _generate_live_data_page(self, mgr: "PageManager") -> Page:
"""Generate live vehicle data page with all available PIDs."""
state = mgr.get_data("vehicle_state")
def _generate_car_data_page(self, mgr: "PageManager") -> Page:
"""Generate combined car data page with OBD2 and UDS data."""
lines = []
has_obd2 = False
has_uds = False
# OBD2 Data Section
state = mgr.get_data("vehicle_state")
if state:
all_values = state.get_all()
_logger.debug(f"Live data: got {len(all_values)} PIDs from state")
_logger.debug(f"Car data: got {len(all_values)} OBD2 PIDs")
if all_values:
# Format functions for different value types
has_obd2 = True
lines.append("--- OBD2 ---")
def fmt_int(val):
return f"{val.value:.0f} {val.unit}"
def fmt_float(val):
return f"{val.value:.1f} {val.unit}"
# Display order with custom formatting
display_order = [
(0x0C, "RPM", fmt_int), # Engine RPM
(0x0D, "Speed", fmt_int), # Vehicle speed
(0x05, "Coolant", fmt_int), # Coolant temp
(0x5C, "Oil temp", fmt_int), # Oil temp
(0x0F, "Intake temp", fmt_int), # Intake temp
(0x11, "Throttle", fmt_float), # Throttle pos
(0x04, "Engine load", fmt_float), # Engine load
(0x2F, "Fuel level", fmt_float), # Fuel level
(0x10, "MAF", fmt_float), # MAF rate
(0x0C, "RPM", fmt_int),
(0x0D, "Speed", fmt_int),
(0x05, "Coolant", fmt_int),
(0x5C, "Oil", fmt_int),
(0x0F, "Intake", fmt_int),
(0x11, "Throttle", fmt_float),
(0x04, "Load", fmt_float),
(0x2F, "Fuel", fmt_float),
]
shown_pids = set()
@@ -410,50 +399,73 @@ class PageManager:
lines.append(f"{label}: {formatter(val)}")
shown_pids.add(pid_code)
# Add any other PIDs not in the display order
for pid_code, val in sorted(all_values.items()):
if pid_code not in shown_pids:
lines.append(f"{val.name}: {val.value:.1f} {val.unit}")
if not lines:
lines = ["Waiting for data...", "", "Polling active"]
else:
lines = ["No OBD2 connection", "", "Use Actions menu", "to reconnect"]
# UDS Data Section
uds_values = mgr.get_data("uds_values", {})
if uds_values:
has_uds = True
lines.append("--- UDS ---")
_logger.debug(f"Live data page: {len(lines)} lines")
return Page(PageType.INFO, "Live Data", lines)
# Boost pressure
if 0x202A in uds_values:
v = uds_values[0x202A]
lines.append(f"Boost: {v.value:.0f} kPa")
def _generate_stats_page(self, mgr: "PageManager") -> Page:
"""Generate statistics page."""
# Torque
if 0x437C in uds_values:
v = uds_values[0x437C]
lines.append(f"Torque: {v.value:.0f} Nm")
# Lambda
if 0x10C0 in uds_values:
v = uds_values[0x10C0]
lines.append(f"Lambda: {v.value:.3f}")
# Ignition timing
if 0x2004 in uds_values:
v = uds_values[0x2004]
lines.append(f"Timing: {v.value:.1f} deg")
# Wastegate
if 0x39A2 in uds_values:
v = uds_values[0x39A2]
lines.append(f"Wastegate: {v.value:.0f}%")
# Gear
if 0x3816 in uds_values:
v = uds_values[0x3816]
gear = int(v.value)
gear_str = "R" if gear == -1 else ("N" if gear == 0 else str(gear))
lines.append(f"Gear: {gear_str}")
if not has_obd2 and not has_uds:
lines = ["No vehicle data", "", "Use Actions menu", "to reconnect"]
_logger.debug(f"Car data page: {len(lines)} lines")
return Page(PageType.INFO, "Car Data", lines)
def _generate_app_status_page(self, mgr: "PageManager") -> Page:
"""Generate application status page with config, handler states, and rate."""
lines = []
# Polling stats (rate, uptime)
stats = mgr.get_data("poller_stats", {})
uptime = mgr.get_data("uptime", 0)
queries = stats.get("queries", 0)
successes = stats.get("successes", 0)
failures = stats.get("failures", 0)
rate = (successes / queries * 100) if queries > 0 else 0
if stats or uptime:
queries = stats.get("queries", 0)
successes = stats.get("successes", 0)
rate = (successes / queries * 100) if queries > 0 else 0
qps = queries / uptime if uptime > 0 else 0
hours = int(uptime // 3600)
minutes = int((uptime % 3600) // 60)
seconds = int(uptime % 60)
hours = int(uptime // 3600)
minutes = int((uptime % 3600) // 60)
# Calculate queries per second
qps = queries / uptime if uptime > 0 else 0
lines = [
f"Queries: {queries}",
f"Success: {successes}",
f"Failures: {failures}",
f"Rate: {rate:.1f}%",
f"Q/sec: {qps:.1f}",
f"Uptime: {hours}h {minutes}m {seconds}s",
]
return Page(PageType.INFO, "Statistics", lines)
def _generate_app_status_page(self, mgr: "PageManager") -> Page:
"""Generate application status page with config and handler states."""
lines = []
lines.append(f"Rate: {rate:.1f}% ({qps:.1f}/s)")
lines.append(f"Uptime: {hours}h {minutes}m")
# Pipeline stats
pipeline_stats = mgr.get_data("pipeline_stats", {})
@@ -469,32 +481,27 @@ class PageManager:
saved = storage.get("saved_count", 0)
pending = storage.get("pending_in_batch", 0)
db_size = storage.get("db_size_mb", 0)
lines.append(f"SQLite: {saved} saved")
lines.append(f"Pending: {pending}, DB: {db_size}MB")
lines.append(f"SQLite: {saved}, {db_size}MB")
if pending > 0:
lines.append(f"Pending: {pending}")
# PostgreSQL handler status
pg = handlers.get("postgresql", {})
if pg:
connected = pg.get("connected", False)
synced = pg.get("synced_count", 0)
status = "OK" if connected else "Offline"
lines.append(f"PG: {status}, synced: {synced}")
else:
lines.append("Pipeline: Not active")
status = "OK" if connected else "Off"
lines.append(f"PG: {status}, sync: {synced}")
# Config info
config = mgr.get_data("config")
if config:
lines.append(f"CAN: {config.can.interface}")
if config.postgresql.enabled:
lines.append(f"PG: {config.postgresql.host}")
else:
lines.append("PG: Disabled")
# Session info
session = mgr.get_data("session_id")
if session:
lines.append(f"Session: {session}")
lines.append(f"Sess: {session[:8]}")
if not lines:
lines = ["No status data", "available"]
@@ -523,62 +530,6 @@ class PageManager:
return Page(PageType.INFO, "System", lines)
def _generate_uds_page(self, mgr: "PageManager") -> Page:
"""Generate UDS extended diagnostics page."""
uds_values = mgr.get_data("uds_values", {})
uds_stats = mgr.get_data("uds_stats", {})
if not uds_values and not uds_stats:
lines = ["UDS not enabled", "", "Enable in config.json:", '"uds": {"enabled": true}']
return Page(PageType.INFO, "UDS Data", lines)
lines = []
# Key UDS values
# Boost pressure (0x202A)
if 0x202A in uds_values:
v = uds_values[0x202A]
lines.append(f"Boost: {v.value:.0f} kPa")
# Torque (0x437C)
if 0x437C in uds_values:
v = uds_values[0x437C]
lines.append(f"Torque: {v.value:.0f} Nm")
# Lambda (0x10C0)
if 0x10C0 in uds_values:
v = uds_values[0x10C0]
lines.append(f"Lambda: {v.value:.3f}")
# Ignition timing (0x2004)
if 0x2004 in uds_values:
v = uds_values[0x2004]
lines.append(f"Timing: {v.value:.1f} deg")
# Wastegate (0x39A2)
if 0x39A2 in uds_values:
v = uds_values[0x39A2]
lines.append(f"Wastegate: {v.value:.0f}%")
# Gear (0x3816)
if 0x3816 in uds_values:
v = uds_values[0x3816]
gear = int(v.value)
gear_str = "R" if gear == -1 else ("N" if gear == 0 else str(gear))
lines.append(f"Gear: {gear_str}")
# Stats
if uds_stats:
queries = uds_stats.get("queries", 0)
successes = uds_stats.get("successes", 0)
rate = (successes / queries * 100) if queries > 0 else 0
lines.append(f"UDS: {rate:.0f}% success")
if not lines:
lines = ["Waiting for UDS data..."]
return Page(PageType.INFO, "UDS Data", lines)
def _generate_actions_page(self, mgr: "PageManager") -> Page:
"""Generate actions menu page."""
actions = [