repo: archive legacy pico dashboard

This commit is contained in:
2026-06-09 00:52:00 -06:00
parent 9cb06b86ff
commit f5e2c5fd24
38 changed files with 107 additions and 20 deletions
+38
View File
@@ -0,0 +1,38 @@
# Legacy Pico Dashboard
This directory contains the archived Raspberry Pi Pico dashboard implementation.
It is no longer active project firmware.
## Why this was archived
The current dashboard direction is the Waveshare ESP32-S3 touchscreen dashboard communicating with the Cargo ESP32 over WiFi/HTTP REST.
The Pico dashboard path was based around:
- Raspberry Pi Pico / Pico 2 W
- UART communication
- Pico-specific display/touch/buzzer hardware
- Python/MicroPython-style dashboard structure
That architecture is no longer the current plan.
## What may still be useful
Some concepts may still be useful as reference material when building the ESP32-S3 dashboard:
- HTTP client flow
- State model ideas
- Alarm logic concepts
- Screen/view-model structure
- API payload expectations
Treat this as reference only. Do not add new active dashboard code here.
## Current dashboard target
Active dashboard work should go under:
firmware/esp32-s3-dashboard/
or another explicitly approved ESP32-S3 dashboard firmware location.
@@ -0,0 +1,13 @@
BATTERY_SOC_LOW = "battery_soc_low"
BATTERY_VOLTAGE_LOW = "battery_voltage_low"
FRIDGE_ZONE_1_HIGH = "fridge_zone_1_high"
FRIDGE_ZONE_2_HIGH = "fridge_zone_2_high"
SENSOR_FAILURE = "sensor_failure"
COMMUNICATION_LOST = "communication_lost"
DEFAULT_THRESHOLDS = {
"battery_soc_low": 20,
"battery_voltage_low": 12.0,
"fridge_temp_high": 40.0,
}
@@ -0,0 +1,44 @@
from .alarm_definitions import (
BATTERY_SOC_LOW,
BATTERY_VOLTAGE_LOW,
FRIDGE_ZONE_1_HIGH,
FRIDGE_ZONE_2_HIGH,
SENSOR_FAILURE,
COMMUNICATION_LOST,
DEFAULT_THRESHOLDS,
)
class AlarmManager:
def __init__(self, thresholds=None):
self.thresholds = dict(DEFAULT_THRESHOLDS)
if thresholds:
self.thresholds.update(thresholds)
def evaluate(self, state):
alarms = []
soc = state.battery.get("soc")
if soc is not None and soc < self.thresholds["battery_soc_low"]:
alarms.append(BATTERY_SOC_LOW)
voltage = state.battery.get("voltage")
if voltage is not None and voltage < self.thresholds["battery_voltage_low"]:
alarms.append(BATTERY_VOLTAGE_LOW)
z1 = state.temps.get("fridge_zone_1")
if z1 is not None and z1 > self.thresholds["fridge_temp_high"]:
alarms.append(FRIDGE_ZONE_1_HIGH)
z2 = state.temps.get("fridge_zone_2")
if z2 is not None and z2 > self.thresholds["fridge_temp_high"]:
alarms.append(FRIDGE_ZONE_2_HIGH)
if not state.network.get("uart_connected", False):
alarms.append(COMMUNICATION_LOST)
for sensor_name, online in state.sensor_health.items():
if not online:
alarms.append(f"{SENSOR_FAILURE}:{sensor_name}")
return alarms
+46
View File
@@ -0,0 +1,46 @@
from alarms.alarm_manager import AlarmManager
from ui.alarm_view_model import AlarmViewModel
from ui.dashboard_view_model import DashboardViewModel
class PicoDashboardApp:
def __init__(
self,
app_state,
communication_service,
screen_manager,
touch_router,
display,
dashboard_renderer,
buzzer,
alarm_manager=None,
):
self.state = app_state
self.comms = communication_service
self.screen_manager = screen_manager
self.touch_router = touch_router
self.display = display
self.dashboard_renderer = dashboard_renderer
self.buzzer = buzzer
self.alarm_manager = alarm_manager or AlarmManager()
self.alarms = []
def tick(self, touch_event=None):
self.comms.poll()
self.comms.auto_select_transport()
if touch_event:
self.touch_router.handle_touch(touch_event)
self.alarms = self.alarm_manager.evaluate(self.state)
alarm_view = AlarmViewModel(self.alarms)
self.buzzer.update_from_alarm_view(alarm_view)
self.render()
def render(self):
if self.screen_manager.current_screen == "dashboard":
vm = DashboardViewModel(self.state, self.alarms)
self.dashboard_renderer.render(vm)
self.display.flush()
@@ -0,0 +1,110 @@
from .protocol import (
make_status_request,
make_set_relay,
is_status_response,
is_relay_response,
is_error,
)
class CommunicationService:
def __init__(
self,
uart_client,
app_state,
http_client=None,
clock=None,
timeout_seconds=5,
):
self.uart_client = uart_client
self.http_client = http_client
self.app_state = app_state
self.clock = clock
self.timeout_seconds = timeout_seconds
self.last_messages = []
self.use_http_fallback = False
self.last_status_received_at = None
def now(self):
if self.clock:
return self.clock()
try:
import time
return time.time()
except ImportError:
return 0
def request_status(self):
if self.use_http_fallback and self.http_client:
message = self.http_client.get_status()
self.handle_message(message)
return message
self.uart_client.send_message(make_status_request())
return None
def set_relay(self, relay, enabled):
if self.use_http_fallback and self.http_client:
message = self.http_client.set_relay(relay, enabled)
self.handle_message(message)
return message
self.uart_client.send_message(make_set_relay(relay, enabled))
return None
def poll(self):
messages = self.uart_client.read_available_messages()
self.last_messages = messages
for message in messages:
self.handle_message(message)
self.update_connection_state()
return messages
def update_connection_state(self):
if self.last_status_received_at is None:
self.app_state.network["uart_connected"] = False
return
age = self.now() - self.last_status_received_at
self.app_state.network["uart_connected"] = age <= self.timeout_seconds
def should_use_http_fallback(self):
return (
self.http_client is not None
and not self.app_state.network.get("uart_connected", False)
)
def auto_select_transport(self):
self.use_http_fallback = self.should_use_http_fallback()
return self.use_http_fallback
def enable_http_fallback(self):
self.use_http_fallback = True
def disable_http_fallback(self):
self.use_http_fallback = False
def handle_message(self, message):
if is_status_response(message):
self.last_status_received_at = self.now()
self.app_state.update_from_status(message)
self.app_state.network["uart_connected"] = True
return
if is_relay_response(message):
self.app_state.update_from_relay_response(message)
return
if is_error(message):
self.app_state.set_error(message)
return
self.app_state.set_error({
"type": "error",
"message": "unknown_message_type",
"raw": message,
})
@@ -0,0 +1,17 @@
class HttpClient:
def __init__(self, requests_module, base_url="http://192.168.4.1"):
self.requests = requests_module
self.base_url = base_url.rstrip("/")
if not self.base_url.endswith("/api/v1"):
self.base_url = f"{self.base_url}/api/v1"
def get_status(self):
response = self.requests.get(f"{self.base_url}/status")
return response.json()
def set_relay(self, relay, enabled):
response = self.requests.post(
f"{self.base_url}/relay/set",
json={"id": relay, "state": enabled},
)
return response.json()
@@ -0,0 +1,35 @@
STATUS_REQUEST = "status_request"
STATUS_RESPONSE = "status_response"
SET_RELAY = "set_relay"
RELAY_RESPONSE = "relay_response"
ERROR = "error"
VALID_RELAYS = {"starlink", "fridge"}
def make_status_request():
return {"type": STATUS_REQUEST}
def make_set_relay(relay, enabled):
if relay not in VALID_RELAYS:
raise ValueError(f"Invalid relay: {relay}")
return {
"type": SET_RELAY,
"relay": relay,
"enabled": bool(enabled),
}
def is_status_response(message):
return message.get("type") == STATUS_RESPONSE
def is_relay_response(message):
return message.get("type") == RELAY_RESPONSE
def is_error(message):
return message.get("type") == ERROR
@@ -0,0 +1,40 @@
import json
class UartClient:
def __init__(self, uart):
self.uart = uart
self.buffer = ""
def send_message(self, message):
payload = json.dumps(message, separators=(",", ":")) + "\n"
self.uart.write(payload.encode("utf-8"))
def read_available_messages(self):
messages = []
while self.uart.any():
chunk = self.uart.read()
if not chunk:
break
self.buffer += chunk.decode("utf-8")
while "\n" in self.buffer:
line, self.buffer = self.buffer.split("\n", 1)
line = line.strip()
if not line:
continue
try:
messages.append(json.loads(line))
except ValueError:
messages.append({
"type": "error",
"message": "invalid_json",
"raw": line,
})
return messages
@@ -0,0 +1,26 @@
class Buzzer:
def __init__(self, pin=None):
self.pin = pin
self.enabled = False
self.pattern = None
def on(self):
self.enabled = True
if self.pin:
self.pin.value(1)
def off(self):
self.enabled = False
self.pattern = None
if self.pin:
self.pin.value(0)
def set_pattern(self, pattern):
self.pattern = pattern
self.enabled = pattern is not None
def update_from_alarm_view(self, alarm_view):
if alarm_view.should_buzz():
self.set_pattern("critical")
else:
self.off()
@@ -0,0 +1,31 @@
class Display:
def __init__(self, driver=None):
self.driver = driver
self.commands = []
def clear(self):
self.commands.append(("clear",))
def text(self, x, y, value, size=1):
self.commands.append(("text", x, y, str(value), size))
def rect(self, x, y, w, h, filled=False):
self.commands.append(("rect", x, y, w, h, filled))
def flush(self):
if not self.driver:
return
for command in self.commands:
name = command[0]
if name == "clear":
self.driver.clear()
elif name == "text":
_, x, y, value, size = command
self.driver.text(x, y, value, size)
elif name == "rect":
_, x, y, w, h, filled = command
self.driver.rect(x, y, w, h, filled)
self.commands = []
@@ -0,0 +1,21 @@
class TouchEvent:
def __init__(self, x, y, pressed=True):
self.x = x
self.y = y
self.pressed = pressed
class TouchController:
def __init__(self, touch_device=None):
self.touch_device = touch_device
def read_event(self):
if not self.touch_device:
return None
point = self.touch_device.read()
if not point:
return None
return TouchEvent(point["x"], point["y"], point.get("pressed", True))
+52
View File
@@ -0,0 +1,52 @@
from app import PicoDashboardApp
from state.app_state import AppState
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
from hardware.display import Display
from hardware.buzzer import Buzzer
from ui.screen_manager import ScreenManager
from ui.touch_router import TouchRouter
from ui.renderers import DashboardRenderer
class NullUart:
def write(self, data):
pass
def any(self):
return False
def read(self):
return b""
def build_app():
state = AppState()
uart_client = UartClient(NullUart())
comms = CommunicationService(uart_client, state)
screens = ScreenManager()
display = Display()
buzzer = Buzzer()
touch_router = TouchRouter(screens)
dashboard_renderer = DashboardRenderer(display)
return PicoDashboardApp(
app_state=state,
communication_service=comms,
screen_manager=screens,
touch_router=touch_router,
display=display,
dashboard_renderer=dashboard_renderer,
buzzer=buzzer,
)
def main():
app = build_app()
app.tick()
return app
if __name__ == "__main__":
main()
@@ -0,0 +1,63 @@
class AppState:
def __init__(self):
self.battery = {
"soc": None,
"voltage": None,
"current": None,
"remaining_ah": None,
"runtime_hours": None,
"temperature_f": None,
}
self.temps = {
"fridge_zone_1": None,
"fridge_zone_2": None,
"rear_seat": None,
"outside": None,
}
self.sensor_health = {
"fridge_zone_1": False,
"fridge_zone_2": False,
"rear_seat": False,
"outside": False,
}
self.relays = {
"relay_1": False,
"relay_2": False,
"starlink": False,
"fridge": False,
}
self.vehicle = {
"ignition_on": False,
}
self.network = {
"uart_connected": False,
"wifi_enabled": False,
}
self.last_error = None
self.last_status_timestamp = None
def update_from_status(self, message):
self.last_status_timestamp = message.get("timestamp")
self.battery.update(message.get("battery", {}))
self.temps.update(message.get("temps", {}))
self.sensor_health.update(message.get("sensor_health", {}))
self.relays.update(message.get("relays", {}))
self.vehicle.update(message.get("vehicle", {}))
self.network.update(message.get("network", {}))
def update_from_relay_response(self, message):
relay = message.get("relay", message.get("id"))
enabled = message.get("enabled", message.get("state"))
if relay in self.relays and enabled is not None:
self.relays[relay] = bool(enabled)
def set_error(self, message):
self.last_error = message
@@ -0,0 +1,859 @@
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
PICO = ROOT / "pico-dashboard"
sys.path.insert(0, str(PICO))
from comms.protocol import make_status_request, make_set_relay
from state.app_state import AppState
from alarms.alarm_manager import AlarmManager
from ui.screen_manager import ScreenManager
def test_protocol_messages():
assert make_status_request() == {"type": "status_request"}
assert make_set_relay("starlink", True) == {
"type": "set_relay",
"relay": "starlink",
"enabled": True,
}
def test_app_state_status_update():
state = AppState()
state.update_from_status({
"timestamp": 123,
"battery": {"soc": 82, "voltage": 13.2},
"temps": {"fridge_zone_1": 34.5},
"sensor_health": {"fridge_zone_1": True},
"relays": {"starlink": True},
"vehicle": {"ignition_on": False},
"network": {"uart_connected": True},
})
assert state.last_status_timestamp == 123
assert state.battery["soc"] == 82
assert state.temps["fridge_zone_1"] == 34.5
assert state.sensor_health["fridge_zone_1"] is True
assert state.relays["starlink"] is True
assert state.network["uart_connected"] is True
def test_alarm_manager():
state = AppState()
state.update_from_status({
"battery": {"soc": 10, "voltage": 11.8},
"temps": {"fridge_zone_1": 45.0, "fridge_zone_2": 35.0},
"sensor_health": {
"fridge_zone_1": True,
"fridge_zone_2": True,
"rear_seat": False,
"outside": True,
},
"network": {"uart_connected": False},
})
alarms = AlarmManager().evaluate(state)
assert "battery_soc_low" in alarms
assert "battery_voltage_low" in alarms
assert "fridge_zone_1_high" in alarms
assert "communication_lost" in alarms
assert "sensor_failure:rear_seat" in alarms
def test_screen_manager():
screens = ScreenManager()
assert screens.current_screen == "dashboard"
screens.go_to("battery")
assert screens.current_screen == "battery"
screens.next_screen()
assert screens.current_screen == "temps"
screens.previous_screen()
assert screens.current_screen == "battery"
class FakeUart:
def __init__(self):
self.writes = []
self.read_chunks = []
def write(self, data):
self.writes.append(data)
def any(self):
return len(self.read_chunks) > 0
def read(self):
if self.read_chunks:
return self.read_chunks.pop(0)
return b""
def test_uart_client_send_message():
from comms.uart_client import UartClient
fake = FakeUart()
client = UartClient(fake)
client.send_message({"type": "status_request"})
assert fake.writes == [b'{"type":"status_request"}\n']
def test_uart_client_reads_newline_delimited_json():
from comms.uart_client import UartClient
fake = FakeUart()
fake.read_chunks = [
b'{"type":"status_response","timestamp":1}\n',
b'{"type":"relay_response","relay":"fridge","enabled":true}\n',
]
client = UartClient(fake)
messages = client.read_available_messages()
assert messages == [
{"type": "status_response", "timestamp": 1},
{"type": "relay_response", "relay": "fridge", "enabled": True},
]
def test_uart_client_handles_partial_messages():
from comms.uart_client import UartClient
fake = FakeUart()
client = UartClient(fake)
fake.read_chunks = [b'{"type":"status_', b'response"}\n']
messages = client.read_available_messages()
assert messages == [{"type": "status_response"}]
def test_uart_client_handles_invalid_json():
from comms.uart_client import UartClient
fake = FakeUart()
fake.read_chunks = [b'not json\n']
client = UartClient(fake)
messages = client.read_available_messages()
assert messages[0]["type"] == "error"
assert messages[0]["message"] == "invalid_json"
def test_communication_service_requests_status():
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
fake = FakeUart()
state = AppState()
service = CommunicationService(UartClient(fake), state)
service.request_status()
assert fake.writes == [b'{"type":"status_request"}\n']
def test_communication_service_sends_relay_command():
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
fake = FakeUart()
state = AppState()
service = CommunicationService(UartClient(fake), state)
service.set_relay("fridge", False)
assert fake.writes == [b'{"type":"set_relay","relay":"fridge","enabled":false}\n']
def test_communication_service_updates_state_from_status():
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
fake = FakeUart()
fake.read_chunks = [
b'{"type":"status_response","timestamp":99,"battery":{"soc":75},"network":{"uart_connected":true}}\n'
]
state = AppState()
service = CommunicationService(UartClient(fake), state)
messages = service.poll()
assert messages[0]["type"] == "status_response"
assert state.last_status_timestamp == 99
assert state.battery["soc"] == 75
assert state.network["uart_connected"] is True
def test_communication_service_updates_state_from_relay_response():
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
fake = FakeUart()
fake.read_chunks = [
b'{"type":"relay_response","relay":"starlink","enabled":true,"ok":true}\n'
]
state = AppState()
service = CommunicationService(UartClient(fake), state)
service.poll()
assert state.relays["starlink"] is True
def test_communication_service_records_errors():
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
fake = FakeUart()
fake.read_chunks = [b'{"type":"error","message":"invalid_json"}\n']
state = AppState()
service = CommunicationService(UartClient(fake), state)
service.poll()
assert state.last_error["message"] == "invalid_json"
class FakeResponse:
def __init__(self, payload):
self.payload = payload
def json(self):
return self.payload
class FakeRequests:
def __init__(self):
self.urls = []
self.posts = []
self.responses = []
def get(self, url):
self.urls.append(url)
if self.responses:
return FakeResponse(self.responses.pop(0))
return FakeResponse({"ok": True})
def post(self, url, json=None):
self.urls.append(url)
self.posts.append({"url": url, "json": json})
if self.responses:
return FakeResponse(self.responses.pop(0))
return FakeResponse({"ok": True})
def test_http_client_get_status():
from comms.http_client import HttpClient
fake_requests = FakeRequests()
fake_requests.responses = [
{"type": "status_response", "battery": {"soc": 82}}
]
client = HttpClient(fake_requests)
payload = client.get_status()
assert fake_requests.urls == ["http://192.168.4.1/api/v1/status"]
assert payload["type"] == "status_response"
assert payload["battery"]["soc"] == 82
def test_http_client_set_relay_on():
from comms.http_client import HttpClient
fake_requests = FakeRequests()
client = HttpClient(fake_requests)
payload = client.set_relay("relay_1", True)
assert fake_requests.urls == ["http://192.168.4.1/api/v1/relay/set"]
assert fake_requests.posts == [{
"url": "http://192.168.4.1/api/v1/relay/set",
"json": {"id": "relay_1", "state": True},
}]
assert payload == {"ok": True}
def test_http_client_set_relay_off():
from comms.http_client import HttpClient
fake_requests = FakeRequests()
client = HttpClient(fake_requests)
payload = client.set_relay("relay_2", False)
assert fake_requests.urls == ["http://192.168.4.1/api/v1/relay/set"]
assert fake_requests.posts == [{
"url": "http://192.168.4.1/api/v1/relay/set",
"json": {"id": "relay_2", "state": False},
}]
assert payload == {"ok": True}
def test_communication_service_http_fallback_status():
from comms.uart_client import UartClient
from comms.http_client import HttpClient
from comms.communication_service import CommunicationService
fake_uart = FakeUart()
fake_requests = FakeRequests()
fake_requests.responses = [
{"type": "status_response", "battery": {"soc": 66}}
]
state = AppState()
service = CommunicationService(
UartClient(fake_uart),
state,
HttpClient(fake_requests),
)
service.enable_http_fallback()
response = service.request_status()
assert response["type"] == "status_response"
assert state.battery["soc"] == 66
assert fake_uart.writes == []
assert fake_requests.urls == ["http://192.168.4.1/api/v1/status"]
def test_communication_service_http_fallback_relay():
from comms.uart_client import UartClient
from comms.http_client import HttpClient
from comms.communication_service import CommunicationService
fake_uart = FakeUart()
fake_requests = FakeRequests()
fake_requests.responses = [
{
"type": "relay_response",
"id": "relay_1",
"state": True,
"ok": True,
}
]
state = AppState()
service = CommunicationService(
UartClient(fake_uart),
state,
HttpClient(fake_requests),
)
service.enable_http_fallback()
response = service.set_relay("relay_1", True)
assert response["type"] == "relay_response"
assert state.relays["relay_1"] is True
assert fake_uart.writes == []
assert fake_requests.urls == ["http://192.168.4.1/api/v1/relay/set"]
assert fake_requests.posts == [{
"url": "http://192.168.4.1/api/v1/relay/set",
"json": {"id": "relay_1", "state": True},
}]
def test_communication_service_marks_uart_connected_on_status():
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
fake = FakeUart()
fake.read_chunks = [
b'{"type":"status_response","timestamp":1,"network":{"uart_connected":true}}\n'
]
state = AppState()
current_time = [100.0]
service = CommunicationService(
UartClient(fake),
state,
clock=lambda: current_time[0],
timeout_seconds=5,
)
service.poll()
assert state.network["uart_connected"] is True
assert service.last_status_received_at == 100.0
def test_communication_service_marks_uart_disconnected_after_timeout():
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
fake = FakeUart()
fake.read_chunks = [
b'{"type":"status_response","timestamp":1,"network":{"uart_connected":true}}\n'
]
state = AppState()
current_time = [100.0]
service = CommunicationService(
UartClient(fake),
state,
clock=lambda: current_time[0],
timeout_seconds=5,
)
service.poll()
assert state.network["uart_connected"] is True
current_time[0] = 106.0
service.poll()
assert state.network["uart_connected"] is False
def test_communication_service_auto_selects_http_fallback_when_uart_down():
from comms.uart_client import UartClient
from comms.http_client import HttpClient
from comms.communication_service import CommunicationService
fake_uart = FakeUart()
fake_requests = FakeRequests()
state = AppState()
service = CommunicationService(
UartClient(fake_uart),
state,
HttpClient(fake_requests),
clock=lambda: 100.0,
timeout_seconds=5,
)
service.poll()
assert service.auto_select_transport() is True
assert service.use_http_fallback is True
def test_dashboard_view_model_formats_dashboard_summary():
from ui.dashboard_view_model import DashboardViewModel
state = AppState()
state.update_from_status({
"battery": {
"soc": 82,
"voltage": 13.2,
"current": -6.4,
"runtime_hours": 12.0,
},
"temps": {
"fridge_zone_1": 34.5,
"fridge_zone_2": 36.0,
},
"relays": {
"starlink": False,
"fridge": True,
},
"network": {
"uart_connected": True,
},
})
vm = DashboardViewModel(state, alarms=[]).as_dict()
assert vm["top_bar"]["soc"] == "82%"
assert vm["top_bar"]["comms"] == "UART OK"
assert vm["top_bar"]["alarms"] == 0
assert vm["battery"]["voltage"] == "13.2V"
assert vm["battery"]["current"] == "-6.4A"
assert vm["battery"]["runtime"] == "12.0 hr"
assert vm["fridge"]["zone_1"] == "34.5°F"
assert vm["fridge"]["zone_2"] == "36.0°F"
assert vm["power"]["starlink"] == "OFF"
assert vm["power"]["fridge"] == "ON"
def test_dashboard_view_model_handles_missing_values():
from ui.dashboard_view_model import DashboardViewModel
state = AppState()
vm = DashboardViewModel(state, alarms=["communication_lost"]).as_dict()
assert vm["top_bar"]["soc"] == "--"
assert vm["top_bar"]["comms"] == "UART LOST"
assert vm["top_bar"]["alarms"] == 1
assert vm["battery"]["voltage"] == "--"
assert vm["fridge"]["zone_1"] == "--"
def test_battery_view_model():
from ui.detail_view_models import BatteryViewModel
state = AppState()
state.update_from_status({
"battery": {
"soc": 82,
"voltage": 13.2,
"current": -6.4,
"remaining_ah": 82.0,
"runtime_hours": 12.0,
"temperature_f": 76.0,
}
})
vm = BatteryViewModel(state).as_dict()
assert vm["soc"] == "82%"
assert vm["voltage"] == "13.2V"
assert vm["current"] == "-6.4A"
assert vm["remaining_ah"] == "82.0Ah"
assert vm["runtime_hours"] == "12.0 hr"
assert vm["temperature_f"] == "76.0°F"
def test_temps_view_model():
from ui.detail_view_models import TempsViewModel
state = AppState()
state.update_from_status({
"temps": {
"fridge_zone_1": 34.5,
"fridge_zone_2": 36.0,
"rear_seat": 71.2,
"outside": None,
},
"sensor_health": {
"fridge_zone_1": True,
"fridge_zone_2": True,
"rear_seat": True,
"outside": False,
}
})
rows = TempsViewModel(state).as_list()
assert rows[0]["label"] == "Fridge Zone 1"
assert rows[0]["temperature"] == "34.5°F"
assert rows[0]["status"] == "OK"
assert rows[3]["label"] == "Outside Air"
assert rows[3]["temperature"] == "--"
assert rows[3]["status"] == "FAULT"
def test_power_view_model():
from ui.detail_view_models import PowerViewModel
state = AppState()
state.update_from_status({
"relays": {
"starlink": False,
"fridge": True,
}
})
vm = PowerViewModel(state).as_dict()
assert vm["starlink"] == "OFF"
assert vm["fridge"] == "ON"
def test_system_view_model():
from ui.detail_view_models import SystemViewModel
state = AppState()
state.update_from_status({
"sensor_health": {
"fridge_zone_1": True,
"fridge_zone_2": True,
"rear_seat": False,
"outside": True,
},
"vehicle": {
"ignition_on": False,
},
"network": {
"uart_connected": True,
"wifi_enabled": True,
}
})
vm = SystemViewModel(state).as_dict()
assert vm["esp32"] == "Online"
assert vm["uart"] == "Connected"
assert vm["wifi_api"] == "Available"
assert vm["sensors"] == "3 / 4 OK"
assert vm["ignition"] == "Off"
def test_alarm_view_model_no_alarms():
from ui.alarm_view_model import AlarmViewModel
vm = AlarmViewModel([])
assert vm.count() == 0
assert vm.has_alarms() is False
assert vm.has_critical() is False
assert vm.should_buzz() is False
assert vm.primary_alarm() is None
assert vm.as_list() == []
def test_alarm_view_model_critical_alarm():
from ui.alarm_view_model import AlarmViewModel
vm = AlarmViewModel(["fridge_zone_1_high"])
assert vm.count() == 1
assert vm.has_alarms() is True
assert vm.has_critical() is True
assert vm.should_buzz() is True
primary = vm.primary_alarm()
assert primary["severity"] == "critical"
assert primary["title"] == "Fridge Zone 1 High Temp"
def test_alarm_view_model_sensor_failure():
from ui.alarm_view_model import AlarmViewModel
vm = AlarmViewModel(["sensor_failure:outside"])
assert vm.has_alarms() is True
assert vm.has_critical() is False
assert vm.should_buzz() is False
alarm = vm.primary_alarm()
assert alarm["severity"] == "warning"
assert alarm["title"] == "Sensor Failure"
assert alarm["detail"] == "Outside"
class FakePin:
def __init__(self):
self.values = []
def value(self, value):
self.values.append(value)
def test_buzzer_on_off():
from hardware.buzzer import Buzzer
pin = FakePin()
buzzer = Buzzer(pin)
buzzer.on()
assert buzzer.enabled is True
assert pin.values[-1] == 1
buzzer.off()
assert buzzer.enabled is False
assert pin.values[-1] == 0
def test_buzzer_updates_from_alarm_view():
from hardware.buzzer import Buzzer
from ui.alarm_view_model import AlarmViewModel
buzzer = Buzzer()
buzzer.update_from_alarm_view(AlarmViewModel(["fridge_zone_1_high"]))
assert buzzer.enabled is True
assert buzzer.pattern == "critical"
buzzer.update_from_alarm_view(AlarmViewModel([]))
assert buzzer.enabled is False
assert buzzer.pattern is None
def test_touch_controller_reads_event():
from hardware.touch import TouchController
class FakeTouchDevice:
def read(self):
return {"x": 100, "y": 200, "pressed": True}
controller = TouchController(FakeTouchDevice())
event = controller.read_event()
assert event.x == 100
assert event.y == 200
assert event.pressed is True
def test_touch_router_bottom_nav_changes_screen():
from hardware.touch import TouchEvent
from ui.touch_router import TouchRouter
screens = ScreenManager()
router = TouchRouter(screens)
handled = router.handle_touch(TouchEvent(200, 460, True))
assert handled is True
assert screens.current_screen == "power"
def test_touch_router_ignores_non_nav_touch():
from hardware.touch import TouchEvent
from ui.touch_router import TouchRouter
screens = ScreenManager()
router = TouchRouter(screens)
handled = router.handle_touch(TouchEvent(200, 200, True))
assert handled is False
assert screens.current_screen == "dashboard"
def test_touch_router_ignores_unpressed_event():
from hardware.touch import TouchEvent
from ui.touch_router import TouchRouter
screens = ScreenManager()
router = TouchRouter(screens)
handled = router.handle_touch(TouchEvent(200, 460, False))
assert handled is False
assert screens.current_screen == "dashboard"
def test_display_records_commands():
from hardware.display import Display
display = Display()
display.clear()
display.text(1, 2, "hello", size=2)
display.rect(3, 4, 5, 6, filled=True)
assert display.commands == [
("clear",),
("text", 1, 2, "hello", 2),
("rect", 3, 4, 5, 6, True),
]
def test_dashboard_renderer_creates_draw_commands():
from hardware.display import Display
from ui.dashboard_view_model import DashboardViewModel
from ui.renderers import DashboardRenderer
state = AppState()
state.update_from_status({
"battery": {
"soc": 82,
"voltage": 13.2,
"current": -6.4,
"runtime_hours": 12.0,
},
"temps": {
"fridge_zone_1": 34.5,
"fridge_zone_2": 36.0,
},
"relays": {
"starlink": False,
"fridge": True,
},
"network": {
"uart_connected": True,
},
})
display = Display()
renderer = DashboardRenderer(display)
renderer.render(DashboardViewModel(state))
assert display.commands[0] == ("clear",)
assert ("text", 0, 0, "SOC 82%", 1) in display.commands
assert ("text", 120, 0, "UART OK", 1) in display.commands
assert ("text", 0, 48, "Battery", 1) in display.commands
assert ("text", 0, 128, "Fridge", 1) in display.commands
assert ("text", 0, 232, "Power", 1) in display.commands
assert ("text", 0, 440, "[Dash]", 1) in display.commands
def test_pico_dashboard_app_tick_updates_alarms_and_buzzer():
from app import PicoDashboardApp
from comms.uart_client import UartClient
from comms.communication_service import CommunicationService
from hardware.display import Display
from hardware.buzzer import Buzzer
from ui.renderers import DashboardRenderer
from ui.touch_router import TouchRouter
fake_uart = FakeUart()
fake_uart.read_chunks = [
b'{"type":"status_response","battery":{"soc":10,"voltage":11.8},"temps":{"fridge_zone_1":45.0},"sensor_health":{"fridge_zone_1":true,"fridge_zone_2":true,"rear_seat":true,"outside":true},"network":{"uart_connected":true}}\n'
]
state = AppState()
comms = CommunicationService(UartClient(fake_uart), state)
screens = ScreenManager()
display = Display()
buzzer = Buzzer()
app = PicoDashboardApp(
app_state=state,
communication_service=comms,
screen_manager=screens,
touch_router=TouchRouter(screens),
display=display,
dashboard_renderer=DashboardRenderer(display),
buzzer=buzzer,
)
app.tick()
assert "battery_soc_low" in app.alarms
assert "battery_voltage_low" in app.alarms
assert "fridge_zone_1_high" in app.alarms
assert buzzer.enabled is True
assert display.commands
def test_pico_main_builds_app():
import main
app = main.build_app()
assert app.state is not None
assert app.comms is not None
assert app.screen_manager.current_screen == "dashboard"
def test_pico_main_runs_one_tick():
import main
app = main.main()
assert app is not None
@@ -0,0 +1,74 @@
from alarms.alarm_definitions import (
BATTERY_SOC_LOW,
BATTERY_VOLTAGE_LOW,
FRIDGE_ZONE_1_HIGH,
FRIDGE_ZONE_2_HIGH,
SENSOR_FAILURE,
COMMUNICATION_LOST,
)
ALARM_LABELS = {
BATTERY_SOC_LOW: "Battery SOC Low",
BATTERY_VOLTAGE_LOW: "Battery Voltage Low",
FRIDGE_ZONE_1_HIGH: "Fridge Zone 1 High Temp",
FRIDGE_ZONE_2_HIGH: "Fridge Zone 2 High Temp",
COMMUNICATION_LOST: "ESP32 Communication Lost",
}
CRITICAL_ALARMS = {
BATTERY_SOC_LOW,
BATTERY_VOLTAGE_LOW,
FRIDGE_ZONE_1_HIGH,
FRIDGE_ZONE_2_HIGH,
COMMUNICATION_LOST,
}
class AlarmViewModel:
def __init__(self, alarms):
self.alarms = alarms or []
def count(self):
return len(self.alarms)
def has_alarms(self):
return self.count() > 0
def has_critical(self):
return any(self._base_alarm(alarm) in CRITICAL_ALARMS for alarm in self.alarms)
def should_buzz(self):
return self.has_critical()
def primary_alarm(self):
if not self.alarms:
return None
return self.format_alarm(self.alarms[0])
def as_list(self):
return [self.format_alarm(alarm) for alarm in self.alarms]
def format_alarm(self, alarm):
base = self._base_alarm(alarm)
if base == SENSOR_FAILURE:
sensor = alarm.split(":", 1)[1] if ":" in alarm else "unknown"
return {
"code": alarm,
"severity": "warning",
"title": "Sensor Failure",
"detail": sensor.replace("_", " ").title(),
}
return {
"code": alarm,
"severity": "critical" if base in CRITICAL_ALARMS else "info",
"title": ALARM_LABELS.get(base, alarm),
"detail": None,
}
def _base_alarm(self, alarm):
return alarm.split(":", 1)[0]
@@ -0,0 +1,59 @@
def format_value(value, suffix="", decimals=1, missing="--"):
if value is None:
return missing
if isinstance(value, int):
return f"{value}{suffix}"
if isinstance(value, float):
return f"{value:.{decimals}f}{suffix}"
return f"{value}{suffix}"
def format_bool(value, true_text="ON", false_text="OFF"):
return true_text if value else false_text
class DashboardViewModel:
def __init__(self, state, alarms=None):
self.state = state
self.alarms = alarms or []
def top_bar(self):
soc = format_value(self.state.battery.get("soc"), "%", decimals=0)
uart = "UART OK" if self.state.network.get("uart_connected") else "UART LOST"
alarm_count = len(self.alarms)
return {
"soc": soc,
"comms": uart,
"alarms": alarm_count,
}
def battery_summary(self):
return {
"voltage": format_value(self.state.battery.get("voltage"), "V", decimals=1),
"current": format_value(self.state.battery.get("current"), "A", decimals=1),
"runtime": format_value(self.state.battery.get("runtime_hours"), " hr", decimals=1),
}
def fridge_summary(self):
return {
"zone_1": format_value(self.state.temps.get("fridge_zone_1"), "°F", decimals=1),
"zone_2": format_value(self.state.temps.get("fridge_zone_2"), "°F", decimals=1),
}
def power_summary(self):
return {
"starlink": format_bool(self.state.relays.get("starlink")),
"fridge": format_bool(self.state.relays.get("fridge")),
}
def as_dict(self):
return {
"top_bar": self.top_bar(),
"battery": self.battery_summary(),
"fridge": self.fridge_summary(),
"power": self.power_summary(),
}
@@ -0,0 +1,74 @@
from .dashboard_view_model import format_value, format_bool
class BatteryViewModel:
def __init__(self, state):
self.state = state
def as_dict(self):
battery = self.state.battery
return {
"soc": format_value(battery.get("soc"), "%", decimals=0),
"voltage": format_value(battery.get("voltage"), "V", decimals=1),
"current": format_value(battery.get("current"), "A", decimals=1),
"remaining_ah": format_value(battery.get("remaining_ah"), "Ah", decimals=1),
"runtime_hours": format_value(battery.get("runtime_hours"), " hr", decimals=1),
"temperature_f": format_value(battery.get("temperature_f"), "°F", decimals=1),
}
class TempsViewModel:
SENSOR_LABELS = {
"fridge_zone_1": "Fridge Zone 1",
"fridge_zone_2": "Fridge Zone 2",
"rear_seat": "Rear Seat",
"outside": "Outside Air",
}
def __init__(self, state):
self.state = state
def as_list(self):
rows = []
for key, label in self.SENSOR_LABELS.items():
online = self.state.sensor_health.get(key, False)
temp = self.state.temps.get(key)
rows.append({
"key": key,
"label": label,
"temperature": format_value(temp, "°F", decimals=1),
"status": "OK" if online else "FAULT",
})
return rows
class PowerViewModel:
def __init__(self, state):
self.state = state
def as_dict(self):
return {
"starlink": format_bool(self.state.relays.get("starlink")),
"fridge": format_bool(self.state.relays.get("fridge")),
}
class SystemViewModel:
def __init__(self, state):
self.state = state
def as_dict(self):
sensor_count = sum(1 for online in self.state.sensor_health.values() if online)
total_sensors = len(self.state.sensor_health)
return {
"esp32": "Online" if self.state.network.get("uart_connected") else "Unknown",
"uart": "Connected" if self.state.network.get("uart_connected") else "Disconnected",
"wifi_api": "Available" if self.state.network.get("wifi_enabled") else "Unavailable",
"sensors": f"{sensor_count} / {total_sensors} OK",
"ignition": "On" if self.state.vehicle.get("ignition_on") else "Off",
}
@@ -0,0 +1,51 @@
class DashboardRenderer:
def __init__(self, display):
self.display = display
def render(self, view_model):
vm = view_model.as_dict()
self.display.clear()
top = vm["top_bar"]
self.display.text(0, 0, f'SOC {top["soc"]}')
self.display.text(120, 0, top["comms"])
self.display.text(260, 0, f'ALM {top["alarms"]}')
battery = vm["battery"]
self.display.text(0, 48, "Battery")
self.display.text(
0,
72,
f'{battery["voltage"]} {battery["current"]} {battery["runtime"]}',
)
fridge = vm["fridge"]
self.display.text(0, 128, "Fridge")
self.display.text(0, 152, f'Zone 1: {fridge["zone_1"]}')
self.display.text(0, 176, f'Zone 2: {fridge["zone_2"]}')
power = vm["power"]
self.display.text(0, 232, "Power")
self.display.text(
0,
256,
f'Starlink {power["starlink"]} Fridge {power["fridge"]}',
)
self.render_nav("dashboard")
def render_nav(self, active_screen):
labels = [
("dashboard", "Dash"),
("battery", "Bat"),
("temps", "Temps"),
("power", "Power"),
("system", "System"),
]
x = 0
for screen, label in labels:
text = f"[{label}]" if screen == active_screen else label
self.display.text(x, 440, text)
x += 64
@@ -0,0 +1,21 @@
class ScreenManager:
VALID_SCREENS = ("dashboard", "battery", "temps", "power", "system")
def __init__(self):
self.current_screen = "dashboard"
def go_to(self, screen_name):
if screen_name not in self.VALID_SCREENS:
raise ValueError(f"Invalid screen: {screen_name}")
self.current_screen = screen_name
def next_screen(self):
current_index = self.VALID_SCREENS.index(self.current_screen)
next_index = (current_index + 1) % len(self.VALID_SCREENS)
self.current_screen = self.VALID_SCREENS[next_index]
def previous_screen(self):
current_index = self.VALID_SCREENS.index(self.current_screen)
previous_index = (current_index - 1) % len(self.VALID_SCREENS)
self.current_screen = self.VALID_SCREENS[previous_index]
@@ -0,0 +1,38 @@
NAV_ITEMS = [
("dashboard", 0, 64),
("battery", 64, 128),
("temps", 128, 192),
("power", 192, 256),
("system", 256, 320),
]
class TouchRouter:
def __init__(self, screen_manager, width=320, height=480, nav_height=56):
self.screen_manager = screen_manager
self.width = width
self.height = height
self.nav_height = nav_height
def handle_touch(self, event):
if not event or not event.pressed:
return False
if self._is_bottom_nav(event.y):
screen = self._screen_for_nav_x(event.x)
if screen:
self.screen_manager.go_to(screen)
return True
return False
def _is_bottom_nav(self, y):
return y >= self.height - self.nav_height
def _screen_for_nav_x(self, x):
for screen, start_x, end_x in NAV_ITEMS:
if start_x <= x < end_x:
return screen
return None