update flipper zero pages and add app status

This commit is contained in:
2026-01-30 12:56:01 +03:00
parent 958279b1b7
commit 19d3885bb2
2 changed files with 79 additions and 11 deletions

View File

@@ -71,7 +71,21 @@ class PageManager:
generator=self._generate_system_page,
))
# Page 3: Actions Menu
# Page 3: App Status (handlers, storage, sync state)
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="App Status",
generator=self._generate_app_status_page,
))
# Page 4: UPS Status
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="UPS Status",
generator=self._generate_ups_page,
))
# Page 5: Actions Menu (last navigable page)
self._pages.append(PageDefinition(
page_type=PageType.MENU,
title="Actions",
@@ -84,14 +98,7 @@ class PageManager:
],
))
# Page 4: UPS Status (if available)
self._pages.append(PageDefinition(
page_type=PageType.INFO,
title="UPS Status",
generator=self._generate_ups_page,
))
# Page 5: Confirm (dynamic)
# Page 6: Confirm (dynamic, hidden from navigation)
self._pages.append(PageDefinition(
page_type=PageType.CONFIRM,
title="Confirm",
@@ -336,14 +343,14 @@ class PageManager:
if self._pending_action:
action = self._pending_action
self._pending_action = None
self._current_index = 3 # Back to actions menu
self._current_index = 5 # 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 = 3 # Back to actions menu
self._current_index = 5 # Back to actions menu
def _get_action_success_message(self, action_id: ActionID) -> str:
"""Get success message for action."""
@@ -437,6 +444,56 @@ class PageManager:
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 = []
# Pipeline stats
pipeline_stats = mgr.get_data("pipeline_stats", {})
if pipeline_stats:
total = pipeline_stats.get("total_readings", 0)
handlers = pipeline_stats.get("handlers", {})
lines.append(f"Readings: {total}")
# Storage handler status
storage = handlers.get("storage", {})
if storage:
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")
# 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")
# 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}")
if not lines:
lines = ["No status data", "available"]
return Page(PageType.INFO, "App Status", lines)
def _generate_system_page(self, mgr: "PageManager") -> Page:
"""Generate system info page."""
ip = self._get_ip_address()

View File

@@ -134,6 +134,8 @@ class OBD2Client:
pm.set_data_provider("uptime", lambda: time.time() - self._start_time)
pm.set_data_provider("can_interface", lambda: self.config.can.interface)
pm.set_data_provider("pipeline_stats", lambda: self.pipeline.get_stats())
pm.set_data_provider("config", lambda: self.config)
pm.set_data_provider("session_id", self._get_session_id)
# Add UPS status page if enabled
if self.config.flipper.show_ups:
@@ -149,6 +151,15 @@ class OBD2Client:
pm.set_action_handler(ActionID.RECONNECT_OBD, self._action_reconnect_obd)
pm.set_action_handler("force_sync", self._action_force_sync)
def _get_session_id(self) -> str:
"""Get current session ID from storage handler."""
storage_handler = self.pipeline.get_handler("storage")
if storage_handler and hasattr(storage_handler, "storage"):
storage = storage_handler.storage
if storage:
return storage.get_current_session_id() or "No session"
return "N/A"
def _action_force_sync(self) -> bool:
"""Force sync to PostgreSQL."""
pg_handler = self.pipeline.get_handler("postgresql")