#!/usr/bin/env python3 """e-Paper status display with camera preview. Two screens: - DATA: UTC time, recording status, BME280 reading, SD-card usage, and CPU temp -- sourced from /run/bubblecam/*.json state files plus cheap local reads (disk_usage, /sys thermal), never from the sensor or camera hardware directly. - CAMERA: a live frame from the stereo camera, both eyes side by side, dithered to 1-bit. Behavior depends on whether recording is active (camera services running, per their own state files): - RECORDING: data screen only, refreshed every epaper.recording_refresh_seconds. No camera preview -- the capture service owns the video device exclusively, and grabbing frames from a mid-mission recording is exactly the kind of interference this display must never cause. - NOT recording: alternates data <-> camera preview every epaper.idle_toggle_seconds (the device is free, so previewing is safe). Refreshes use the panel's flash-free partial mode, with a full (flashing) refresh every epaper.full_refresh_every updates to clear accumulated ghosting -- dithered photo content ghosts noticeably on e-paper. """ from __future__ import annotations import os import shutil import signal import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path from typing import List, Optional sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from bubblecam import config, state_lib # noqa: E402 from bubblecam.hardware.epaper import EPaperDisplay, HEIGHT, WIDTH # noqa: E402 FONT_PATH = "/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf" _shutdown_requested = False def _handle_sigterm(signum, frame): global _shutdown_requested _shutdown_requested = True def camera_service_names(cfg: dict) -> List[str]: mode = cfg["camera"]["mode"] if mode.startswith("dual_node"): return ["camera_left", "camera_right"] return ["camera_primary"] def recording_active(cfg: dict) -> bool: return all(state_lib.is_running(name) for name in camera_service_names(cfg)) def cpu_temp_c() -> Optional[float]: try: with open("/sys/class/thermal/thermal_zone0/temp") as f: return int(f.read().strip()) / 1000.0 except (OSError, ValueError): return None # dev machine / unexpected platform def sd_usage_line(cfg: dict) -> str: """e.g. "SD: 4.2/1700.0GB (0%)". Denominator is used+free, NOT total: on a big card the filesystem's reserved/overhead blocks are tens of GB, and what a diver checking this display needs is what's actually usable for recording.""" target = Path(cfg["paths"]["video_dir"]) if not target.exists(): target = Path("/") du = shutil.disk_usage(target) used_gb = (du.total - du.free) / 1e9 available_gb = du.free / 1e9 usable_gb = used_gb + available_gb pct = 100 * used_gb / usable_gb if usable_gb else 0.0 return f"SD: {used_gb:.1f}/{usable_gb:.1f}GB ({pct:.0f}%)" def _load_fonts(): from PIL import ImageFont try: return ImageFont.truetype(FONT_PATH, 15), ImageFont.truetype(FONT_PATH, 13) except OSError: font = ImageFont.load_default() return font, font def build_data_image(cfg: dict): from PIL import Image, ImageDraw image = Image.new("1", (WIDTH, HEIGHT), 255) draw = ImageDraw.Draw(image) font_header, font_body = _load_fonts() now_utc = datetime.now(timezone.utc) camera_label = "RECORDING" if recording_active(cfg) else "NOT RECORDING" sensors_state = state_lib.read_state("sensors") or {} if state_lib.is_running("sensors") and "temperature_c" in sensors_state: sensor_line = ( f"{sensors_state['temperature_c']:.1f}C " f"{sensors_state['humidity_pct']:.0f}%RH " f"{sensors_state['pressure_hpa']:.0f}hPa" ) else: sensor_line = "sensors: no data" cpu = cpu_temp_c() cpu_line = f"CPU: {cpu:.1f}C" if cpu is not None else "CPU: n/a" lines = [ (font_header, f"UTC {now_utc.strftime('%Y-%m-%d %H:%M:%S')}"), (font_body, f"Camera: {camera_label}"), (font_body, sensor_line), (font_body, sd_usage_line(cfg)), (font_body, cpu_line), ] y = 6 for font, line in lines: draw.text((4, y), line, font=font, fill=0) y += 23 return image def grab_camera_frame(cfg: dict): """One MJPEG frame via ffmpeg, written to tmpfs (no SD wear). Returns a PIL Image or None (device busy/missing, sim mode). Captures at epaper.preview_resolution -- deliberately the camera's SMALLEST native mode, never the recording resolution. This preview is destined for a 250x122 1-bit panel, so extra pixels buy nothing, while decoding a full-resolution frame costs hundreds of MB: on a 512MB Zero 2 W with swap disabled, grabbing 3200x1200 previews on a timer invokes the OOM killer, which takes out NetworkManager and wpa_supplicant and silently strands the Pi with no network. """ from PIL import Image if os.environ.get("BUBBLECAM_SIM") == "1": return None camera_cfg = cfg["camera"] tmp = Path(os.environ.get("BUBBLECAM_STATE_DIR", "/run/bubblecam")) / "epaper_preview.jpg" tmp.parent.mkdir(parents=True, exist_ok=True) cmd = [ "ffmpeg", "-nostdin", "-loglevel", "error", "-y", "-threads", "1", "-f", "v4l2", "-input_format", "mjpeg", "-video_size", cfg["epaper"]["preview_resolution"], "-i", camera_cfg["device_primary"], "-frames:v", "1", str(tmp), ] try: subprocess.run(cmd, check=True, timeout=10, capture_output=True) with Image.open(tmp) as img: return img.copy() except Exception: return None def build_camera_image(cfg: dict): """Both stereo eyes side by side (the camera's native combined frame), scaled to the panel and Floyd-Steinberg dithered to 1-bit.""" from PIL import Image, ImageDraw frame = grab_camera_frame(cfg) canvas = Image.new("1", (WIDTH, HEIGHT), 255) draw = ImageDraw.Draw(canvas) _, font_body = _load_fonts() if frame is None: draw.text((6, 50), "camera preview unavailable", font=font_body, fill=0) draw.rectangle((0, 0, WIDTH - 1, HEIGHT - 1), outline=0) return canvas frame.thumbnail((WIDTH, HEIGHT)) mono = frame.convert("L").convert("1") x0 = (WIDTH - mono.width) // 2 y0 = (HEIGHT - mono.height) // 2 canvas.paste(mono, (x0, y0)) mid = x0 + mono.width // 2 draw.line((mid, y0, mid, y0 + mono.height), fill=0) draw.text((x0 + 2, y0), "L", font=font_body, fill=0) draw.text((mid + 3, y0), "R", font=font_body, fill=0) return canvas def main() -> int: signal.signal(signal.SIGTERM, _handle_sigterm) cfg = config.load() epaper_cfg = cfg["epaper"] idle_toggle_seconds = epaper_cfg["idle_toggle_seconds"] recording_refresh_seconds = epaper_cfg["recording_refresh_seconds"] full_refresh_every = epaper_cfg["full_refresh_every"] state_lib.write_state("epaper", "starting") try: display = EPaperDisplay() except Exception as exc: state_lib.write_state("epaper", "error", detail=f"init failed: {exc}") return 1 showing_camera = False updates = 0 while not _shutdown_requested: try: recording = recording_active(cfg) if recording: showing_camera = False image = build_data_image(cfg) interval = recording_refresh_seconds else: showing_camera = not showing_camera image = build_camera_image(cfg) if showing_camera else build_data_image(cfg) interval = idle_toggle_seconds updates += 1 if updates % full_refresh_every == 0: display.render(image) else: display.render_partial(image) state_lib.write_state( "epaper", "running", screen="camera" if showing_camera else "data", recording=recording, ) except Exception as exc: state_lib.write_state("epaper", "error", detail=str(exc)) interval = idle_toggle_seconds slept = 0.0 while slept < interval and not _shutdown_requested: step = min(1.0, interval - slept) time.sleep(step) slept += step display.sleep() return 0 if __name__ == "__main__": sys.exit(main())