34cd247dca
Touchscreen kiosk for the Raspberry Pi HQ camera (IMX477) used as a microscope: live preview from boot, raw frame recording, full-resolution stills, and automatic capture-SSD mounting. - app.py: GTK3 fullscreen UI over cage — preview, REC/STOP, PHOTO, 90-degree rotate button, tap-to-open exposure/gain/settings sliders, and an "SSD NOT MOUNTED" badge - camera.py: picamera2 wrapper with separate video (ISP preview + packed 12-bit raw) and still (4056x3040) configurations - recorder.py: writer thread turning the raw stream into one Bayer frame per file plus index.jsonl/session.json, with dropped/missed accounting - photos.py: still capture to DNG + JPEG, and a background worker that renders a viewable 16-bit sRGB TIFF from the 12-bit Bayer frame - storage.py: SSD mount state and the date/time/format session layout - tools/unpack_raw.py: desktop-side unpacker/demosaicer - system/ + install.sh: systemd units, udev rule, mount helper, sudoers Note: raw frames are written byte-for-byte as the sensor produced them, so they stay decodeable regardless of how the stills pipeline evolves.
1004 lines
38 KiB
Python
1004 lines
38 KiB
Python
#!/usr/bin/env python3
|
||
"""Microscope camera kiosk UI.
|
||
|
||
Full-screen live preview with two touch buttons:
|
||
|
||
REC / STOP – start/stop raw frame recording (packed 12-bit Bayer, 2028x1520)
|
||
PHOTO – full-resolution still (4056x3040) as DNG + 16-bit TIFF + JPEG
|
||
|
||
An SSD badge appears in the corner whenever the capture SSD is not mounted, and
|
||
recording refuses to start when it is missing so nothing lands on the SD card by
|
||
accident.
|
||
|
||
Run directly (``python3 app.py``) or through cage via the systemd unit.
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
import queue
|
||
import signal
|
||
import sys
|
||
import threading
|
||
import time
|
||
|
||
import numpy as np
|
||
|
||
import gi
|
||
|
||
gi.require_version("Gtk", "3.0")
|
||
from gi.repository import Gdk, GLib, Gtk # noqa: E402 (must follow gi.require_version)
|
||
|
||
import cairo # noqa: E402
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from camera import MicroscopeCamera # noqa: E402
|
||
from config import Config # noqa: E402
|
||
from photos import TiffWorker, capture_still # noqa: E402
|
||
from recorder import RawRecorder # noqa: E402
|
||
from storage import Storage # noqa: E402
|
||
|
||
log = logging.getLogger("app")
|
||
|
||
CSS = b"""
|
||
window, .root { background-color: #000000; }
|
||
drawingarea { background-color: #000000; }
|
||
.status {
|
||
background-color: rgba(0,0,0,0.55);
|
||
color: #e6e6e6;
|
||
padding: 10px 16px;
|
||
border-radius: 10px;
|
||
font-size: 17px;
|
||
}
|
||
.status-dim { color: #9aa0a6; font-size: 15px; }
|
||
.status-rec { color: #ff6b6b; }
|
||
.badge {
|
||
background-color: rgba(176,0,32,0.92);
|
||
color: #ffffff;
|
||
padding: 14px 20px;
|
||
border-radius: 12px;
|
||
font-size: 22px;
|
||
font-weight: bold;
|
||
}
|
||
.badge-hint { font-size: 15px; font-weight: normal; }
|
||
.toast {
|
||
background-color: rgba(20,20,20,0.94);
|
||
color: #ffffff;
|
||
padding: 16px 24px;
|
||
border-radius: 14px;
|
||
font-size: 22px;
|
||
}
|
||
.toast-error { background-color: rgba(150,20,20,0.95); }
|
||
button {
|
||
background-image: none;
|
||
background-color: rgba(28,28,28,0.88);
|
||
color: #f2f2f2;
|
||
border: 2px solid #5f6368;
|
||
border-radius: 12px;
|
||
font-size: 20px;
|
||
font-weight: bold;
|
||
padding: 8px 18px;
|
||
text-shadow: none;
|
||
}
|
||
button.rec { border-color: #d93025; color: #ff8a80; }
|
||
button.rec.recording { background-color: #c62828; color: #ffffff; border-color: #ff8a80; }
|
||
button.photo { border-color: #1a73e8; color: #8ab4f8; }
|
||
button.mount { border-color: #f9ab00; color: #fdd663; font-size: 16px; }
|
||
button.rotate { border-color: #34a853; color: #81c995; }
|
||
button:disabled { color: #5f6368; border-color: #303134; background-color: rgba(20,20,20,0.7); }
|
||
|
||
/* ---- exposure / gain settings panel ---- */
|
||
.settings {
|
||
background-color: rgba(10,10,10,0.95);
|
||
border: 1px solid #3c4043;
|
||
border-radius: 14px;
|
||
padding: 12px 16px;
|
||
}
|
||
.settings-title { font-size: 17px; font-weight: bold; color: #e6e6e6; }
|
||
.settings-hint { font-size: 13px; color: #9aa0a6; }
|
||
.setting-name { font-size: 14px; color: #c8c8c8; }
|
||
.setting-value { font-size: 14px; color: #8ab4f8; font-family: monospace; }
|
||
scale trough { min-height: 12px; min-width: 120px; border-radius: 6px; background-color: #303134; }
|
||
scale highlight { border-radius: 6px; background-color: #1a73e8; }
|
||
scale slider { min-width: 26px; min-height: 26px; border-radius: 13px;
|
||
background-color: #e8eaed; border: 1px solid #5f6368; }
|
||
.pill-hint { font-size: 13px; color: #8ab4f8; }
|
||
.pill-shut { font-size: 13px; color: #6b7075; }
|
||
"""
|
||
|
||
|
||
class PreviewArea(Gtk.DrawingArea):
|
||
"""Draws the newest camera frame, letterboxed into the window.
|
||
|
||
Rotation is applied to the raster here (not by the camera), so the button
|
||
responds instantly and recording keeps its full frame rate.
|
||
"""
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self._frame = None
|
||
self._lock = threading.Lock()
|
||
self._sequence = 0
|
||
self._rotation = 0
|
||
self.set_hexpand(True)
|
||
self.set_vexpand(True)
|
||
self.connect("draw", self._on_draw)
|
||
|
||
def set_rotation(self, degrees):
|
||
with self._lock:
|
||
self._rotation = int(degrees) % 360
|
||
|
||
def push_frame(self, array):
|
||
with self._lock:
|
||
self._frame = array
|
||
self._sequence += 1
|
||
|
||
def frame_state(self):
|
||
with self._lock:
|
||
rotation = self._rotation
|
||
frame = self._frame
|
||
if frame is None or rotation == 0:
|
||
return frame, rotation
|
||
# np.rot90 turns counter-clockwise, so invert the sign for clockwise UI degrees
|
||
return np.ascontiguousarray(np.rot90(frame, k=(-(rotation // 90)) % 4)), rotation
|
||
|
||
def _on_draw(self, _widget, cr):
|
||
cr.set_source_rgb(0, 0, 0)
|
||
cr.paint()
|
||
frame, _rotation = self.frame_state()
|
||
if frame is None:
|
||
cr.set_source_rgb(0.55, 0.58, 0.62)
|
||
cr.select_font_face("sans", cairo.FONT_SLANT_NORMAL, cairo.FONT_WEIGHT_NORMAL)
|
||
cr.set_font_size(28)
|
||
cr.move_to(320, 300)
|
||
cr.show_text("waiting for camera...")
|
||
return False
|
||
|
||
height, width = frame.shape[:2]
|
||
stride = frame.strides[0]
|
||
surface = cairo.ImageSurface.create_for_data(
|
||
frame, cairo.FORMAT_RGB24, width, height, stride
|
||
)
|
||
alloc_w = self.get_allocated_width()
|
||
alloc_h = self.get_allocated_height()
|
||
scale = min(alloc_w / width, alloc_h / height)
|
||
cr.save()
|
||
cr.translate((alloc_w - width * scale) / 2.0, (alloc_h - height * scale) / 2.0)
|
||
cr.scale(scale, scale)
|
||
cr.set_source_surface(surface, 0, 0)
|
||
cr.get_source().set_filter(cairo.FILTER_BILINEAR)
|
||
cr.paint()
|
||
cr.restore()
|
||
return False
|
||
|
||
|
||
class CameraWorker(threading.Thread):
|
||
"""Owns the camera, the preview feed and the raw recorder."""
|
||
|
||
def __init__(self, cfg, storage, ui):
|
||
super().__init__(name="camera", daemon=True)
|
||
self.cfg = cfg
|
||
self.storage = storage
|
||
self.ui = ui
|
||
self.camera = None
|
||
self.recorder = RawRecorder(storage, cfg, on_state_change=self._recorder_changed)
|
||
self.tiff_worker = TiffWorker(on_done=self._tiff_done)
|
||
self.commands = queue.Queue()
|
||
self._stop = threading.Event()
|
||
self._preview_lock = threading.Lock()
|
||
self._latest_frame = None
|
||
self._pending_controls = {}
|
||
self._pending_lock = threading.Lock()
|
||
self.auto_exposure = True
|
||
self.error = None
|
||
self.last_metadata = {}
|
||
|
||
# ------------------------------------------------------------- controls
|
||
def control_ranges(self):
|
||
if self.camera is None:
|
||
return {}
|
||
return self.camera.control_ranges()
|
||
|
||
def push_controls(self, changes):
|
||
"""Queue camera control changes; they are applied on the camera thread."""
|
||
with self._pending_lock:
|
||
self._pending_controls.update(changes)
|
||
|
||
def set_auto_exposure(self, enabled):
|
||
self.auto_exposure = bool(enabled)
|
||
|
||
def frame_duration_limits(self):
|
||
"""The pinned FrameDurationLimits used to hold the recording rate."""
|
||
if self.camera is not None:
|
||
return self.camera.frame_duration_limits(
|
||
int(self.cfg.get("record.target_fps", 30))
|
||
)
|
||
fps = max(1, int(self.cfg.get("record.target_fps", 30)))
|
||
return (int(1_000_000 / fps), int(1_000_000 / fps))
|
||
|
||
def relaxed_frame_duration_limits(self, exposure_us):
|
||
if self.camera is None:
|
||
return None
|
||
return self.camera.relaxed_frame_duration_limits(exposure_us)
|
||
|
||
def _apply_pending_controls(self):
|
||
with self._pending_lock:
|
||
pending = dict(self._pending_controls)
|
||
self._pending_controls.clear()
|
||
if pending and self.camera is not None:
|
||
self.camera.apply_controls(pending)
|
||
|
||
# ------------------------------------------------------------- plumbing
|
||
def _recorder_changed(self):
|
||
GLib.idle_add(self.ui.on_recorder_state_change)
|
||
|
||
def _tiff_done(self, job, elapsed, error):
|
||
GLib.idle_add(self.ui.on_tiff_done, job, elapsed, error)
|
||
|
||
def submit_command(self, command):
|
||
self.commands.put(command)
|
||
|
||
def stop(self):
|
||
self._stop.set()
|
||
self.commands.put("stop")
|
||
|
||
def preview_frame(self):
|
||
with self._preview_lock:
|
||
return self._latest_frame
|
||
|
||
@property
|
||
def recording(self):
|
||
return self.recorder.active
|
||
|
||
# ------------------------------------------------------------------ loop
|
||
def run(self):
|
||
# The TIFF worker is a threaded object, not a thread until it is started.
|
||
self.tiff_worker.start()
|
||
try:
|
||
self.camera = MicroscopeCamera(self.cfg)
|
||
self.camera.start("video")
|
||
geometry = self.camera.raw_geometry()
|
||
self.recorder.set_raw_geometry(geometry)
|
||
log.info("raw stream: %s", geometry)
|
||
except Exception as exc: # camera missing / in use / locked
|
||
self.error = str(exc)
|
||
log.exception("camera failed to start")
|
||
GLib.idle_add(self.ui.on_camera_error, str(exc))
|
||
return
|
||
|
||
GLib.idle_add(self.ui.on_camera_ready)
|
||
self._apply_pending_controls()
|
||
|
||
while not self._stop.is_set():
|
||
self._drain_commands()
|
||
self._apply_pending_controls()
|
||
if self._stop.is_set():
|
||
break
|
||
try:
|
||
request = self.camera.capture_request()
|
||
except Exception as exc:
|
||
log.exception("capture_request failed")
|
||
self.error = str(exc)
|
||
GLib.idle_add(self.ui.on_camera_error, str(exc))
|
||
break
|
||
|
||
try:
|
||
frame = self.camera.preview_array(request)
|
||
with self._preview_lock:
|
||
self._latest_frame = frame
|
||
try:
|
||
self.last_metadata = request.get_metadata()
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
log.debug("preview array failed", exc_info=True)
|
||
|
||
if self.recorder.active:
|
||
self.recorder.submit(request) # writer owns the request now
|
||
else:
|
||
request.release()
|
||
|
||
self.recorder.stop("app exiting")
|
||
try:
|
||
self.camera.stop()
|
||
self.camera.close()
|
||
except Exception:
|
||
log.debug("camera shutdown raised", exc_info=True)
|
||
# Flush a still's TIFF before the process goes away.
|
||
self.tiff_worker.stop()
|
||
log.info("camera thread finished")
|
||
|
||
def _drain_commands(self):
|
||
while True:
|
||
try:
|
||
command = self.commands.get_nowait()
|
||
except queue.Empty:
|
||
return
|
||
if command == "photo":
|
||
self._take_photo()
|
||
elif command == "stop":
|
||
return
|
||
|
||
def _take_photo(self):
|
||
if self.recorder.active:
|
||
GLib.idle_add(self.ui.toast, "Stop recording before taking a photo")
|
||
return
|
||
if self.storage.state() != "ready":
|
||
GLib.idle_add(self.ui.toast, "SSD not mounted — photo not saved", True)
|
||
return
|
||
GLib.idle_add(self.ui.on_photo_started)
|
||
try:
|
||
summary = capture_still(
|
||
self.camera, self.storage, self.cfg, self.tiff_worker,
|
||
on_tiff_done=self._tiff_done,
|
||
rotation=self.ui.rotation,
|
||
)
|
||
except Exception as exc:
|
||
log.exception("still capture failed")
|
||
GLib.idle_add(self.ui.on_photo_failed, str(exc))
|
||
return
|
||
GLib.idle_add(self.ui.on_photo_saved, summary)
|
||
|
||
|
||
class MicroscopeApp:
|
||
def __init__(self, cfg):
|
||
self.cfg = cfg
|
||
self.storage = Storage(cfg)
|
||
self.worker = CameraWorker(cfg, self.storage, self)
|
||
self._preview_sequence = -1
|
||
self._camera_ready = False
|
||
self._toast_timer = None
|
||
self._last_storage_state = None
|
||
self._last_mount_attempt = 0.0
|
||
self._session_dir = None
|
||
self.rotation = int(cfg.get("ui.rotation", 0) or 0) % 360
|
||
self._settings_visible = False
|
||
self._control_rows = {}
|
||
self._suspend_control_writes = False
|
||
|
||
self.window = Gtk.Window(title="Microscope Camera")
|
||
self.window.set_default_size(
|
||
int(cfg.get("preview.width", 1024)), int(cfg.get("preview.height", 600))
|
||
)
|
||
self.window.connect("destroy", self._on_destroy)
|
||
self.window.connect("key-press-event", self._on_key_press)
|
||
|
||
self.preview = PreviewArea()
|
||
self.preview.set_rotation(self.rotation)
|
||
|
||
self.overlay = Gtk.Overlay()
|
||
self.overlay.add(self.preview)
|
||
self.overlay.add_overlay(self._build_status_pill())
|
||
self.overlay.add_overlay(self._build_settings_panel())
|
||
self.overlay.add_overlay(self._build_ssd_badge())
|
||
self.overlay.add_overlay(self._build_button_bar())
|
||
self.overlay.add_overlay(self._build_rotate_button())
|
||
self.overlay.add_overlay(self._build_toast())
|
||
|
||
self.window.add(self.overlay)
|
||
self._apply_css()
|
||
|
||
# ------------------------------------------------------------------- UI
|
||
def _apply_css(self):
|
||
provider = Gtk.CssProvider()
|
||
provider.load_from_data(CSS)
|
||
screen = Gdk.Screen.get_default()
|
||
if screen is not None:
|
||
Gtk.StyleContext.add_provider_for_screen(
|
||
screen, provider, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION
|
||
)
|
||
|
||
def _build_status_pill(self):
|
||
# The pill doubles as the settings button: tapping it toggles the
|
||
# exposure/gain sliders (an EventBox is used so the whole pill is a
|
||
# target, which matters on a touchscreen).
|
||
event_box = Gtk.EventBox()
|
||
event_box.set_visible_window(False)
|
||
event_box.set_halign(Gtk.Align.START)
|
||
event_box.set_valign(Gtk.Align.START)
|
||
event_box.set_margin_top(14)
|
||
event_box.set_margin_start(14)
|
||
event_box.connect("button-press-event", self._on_status_clicked)
|
||
self.status_event_box = event_box
|
||
|
||
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
|
||
box.get_style_context().add_class("status")
|
||
|
||
self.status_main = Gtk.Label(label="starting...")
|
||
self.status_main.set_xalign(0)
|
||
self.status_main.get_style_context().add_class("status-rec")
|
||
self.status_info = Gtk.Label(label="")
|
||
self.status_info.set_xalign(0)
|
||
self.status_info.get_style_context().add_class("status-dim")
|
||
self.status_action = Gtk.Label()
|
||
self.status_action.set_xalign(0)
|
||
self.status_action.get_style_context().add_class("pill-hint")
|
||
self.status_action.set_markup("tap for exposure / gain")
|
||
box.pack_start(self.status_main, False, False, 0)
|
||
box.pack_start(self.status_info, False, False, 0)
|
||
box.pack_start(self.status_action, False, False, 0)
|
||
event_box.add(box)
|
||
return event_box
|
||
|
||
# ------------------------------------------------------- settings panel
|
||
def _build_settings_panel(self):
|
||
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
|
||
box.get_style_context().add_class("settings")
|
||
box.set_halign(Gtk.Align.START)
|
||
box.set_valign(Gtk.Align.START)
|
||
box.set_margin_start(14)
|
||
# Clear of the status pill, which grew a third line (the tap hint).
|
||
box.set_margin_top(122)
|
||
box.set_size_request(430, -1)
|
||
|
||
title = Gtk.Label(label="Camera settings")
|
||
title.set_xalign(0)
|
||
title.get_style_context().add_class("settings-title")
|
||
box.pack_start(title, False, False, 0)
|
||
|
||
self.auto_button = Gtk.Button(label="Auto exposure: ON")
|
||
self.auto_button.set_size_request(200, 44)
|
||
self.auto_button.connect("clicked", self._on_auto_toggled)
|
||
box.pack_start(self.auto_button, False, False, 0)
|
||
|
||
hint = Gtk.Label(label="tap the status pill again to hide")
|
||
hint.set_xalign(0)
|
||
hint.get_style_context().add_class("settings-hint")
|
||
box.pack_start(hint, False, False, 0)
|
||
|
||
# Exposure spans microseconds to seconds, so its slider is logarithmic:
|
||
# a linear one would waste 95% of its travel above 1 s/1000.
|
||
for name, label, kind in (
|
||
("ExposureTime", "Exposure", "log"),
|
||
("AnalogueGain", "Gain", "linear"),
|
||
("Sharpness", "Sharpness", "linear"),
|
||
("Contrast", "Contrast", "linear"),
|
||
("Saturation", "Saturation", "linear"),
|
||
("Brightness", "Brightness", "linear"),
|
||
):
|
||
row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||
row.set_halign(Gtk.Align.FILL)
|
||
name_label = Gtk.Label(label=label)
|
||
name_label.set_xalign(0)
|
||
name_label.set_size_request(96, -1)
|
||
name_label.get_style_context().add_class("setting-name")
|
||
scale = Gtk.Scale.new_with_range(Gtk.Orientation.HORIZONTAL, 0.0, 1.0, 0.001)
|
||
scale.set_draw_value(False)
|
||
scale.set_hexpand(True)
|
||
scale.set_size_request(200, -1)
|
||
scale.connect("value-changed", self._on_slider_changed, name)
|
||
value_label = Gtk.Label(label="-")
|
||
value_label.set_xalign(1)
|
||
value_label.set_size_request(104, -1)
|
||
value_label.get_style_context().add_class("setting-value")
|
||
row.pack_start(name_label, False, False, 0)
|
||
row.pack_start(scale, True, True, 0)
|
||
row.pack_start(value_label, False, False, 0)
|
||
box.pack_start(row, False, False, 0)
|
||
self._control_rows[name] = {"scale": scale, "value": value_label,
|
||
"kind": kind, "range": None}
|
||
|
||
self.settings_box = box
|
||
box.set_visible(False)
|
||
return box
|
||
|
||
def _build_rotate_button(self):
|
||
button = Gtk.Button(label="⟳ 0°")
|
||
button.get_style_context().add_class("rotate")
|
||
button.set_size_request(150, 62)
|
||
button.set_halign(Gtk.Align.START)
|
||
button.set_valign(Gtk.Align.END)
|
||
button.set_margin_start(20)
|
||
button.set_margin_bottom(20)
|
||
button.connect("clicked", self._on_rotate_clicked)
|
||
self.rotate_button = button
|
||
return button
|
||
|
||
def _build_ssd_badge(self):
|
||
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2)
|
||
box.get_style_context().add_class("badge")
|
||
box.set_halign(Gtk.Align.END)
|
||
box.set_valign(Gtk.Align.START)
|
||
box.set_margin_top(14)
|
||
box.set_margin_end(14)
|
||
|
||
self.ssd_label = Gtk.Label()
|
||
self.ssd_label.set_markup("⚠ SSD NOT MOUNTED")
|
||
self.ssd_hint = Gtk.Label()
|
||
self.ssd_hint.get_style_context().add_class("badge-hint")
|
||
self.ssd_hint.set_visible(False)
|
||
box.pack_start(self.ssd_label, False, False, 0)
|
||
box.pack_start(self.ssd_hint, False, False, 0)
|
||
self.ssd_badge = box
|
||
box.set_visible(False)
|
||
return box
|
||
|
||
def _build_button_bar(self):
|
||
# Vertical stack pinned to the bottom-right corner. The optional MOUNT
|
||
# button sits on top, so showing or hiding it never shifts REC/PHOTO.
|
||
bar = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12)
|
||
bar.set_halign(Gtk.Align.END)
|
||
bar.set_valign(Gtk.Align.END)
|
||
bar.set_margin_end(20)
|
||
bar.set_margin_bottom(20)
|
||
|
||
self.record_button = Gtk.Button(label="● REC")
|
||
self.record_button.get_style_context().add_class("rec")
|
||
self.record_button.set_size_request(150, 62)
|
||
self.record_button.connect("clicked", self._on_record_clicked)
|
||
|
||
self.photo_button = Gtk.Button(label="◉ PHOTO")
|
||
self.photo_button.get_style_context().add_class("photo")
|
||
self.photo_button.set_size_request(150, 62)
|
||
self.photo_button.connect("clicked", self._on_photo_clicked)
|
||
|
||
self.mount_button = Gtk.Button(label="⏏ MOUNT SSD")
|
||
self.mount_button.get_style_context().add_class("mount")
|
||
self.mount_button.set_size_request(150, 54)
|
||
self.mount_button.set_visible(False)
|
||
self.mount_button.connect("clicked", self._on_mount_clicked)
|
||
|
||
bar.pack_start(self.mount_button, False, False, 0)
|
||
bar.pack_start(self.record_button, False, False, 0)
|
||
bar.pack_start(self.photo_button, False, False, 0)
|
||
return bar
|
||
|
||
def _build_toast(self):
|
||
self.toast_label = Gtk.Label()
|
||
self.toast_label.get_style_context().add_class("toast")
|
||
self.toast_label.set_halign(Gtk.Align.CENTER)
|
||
self.toast_label.set_valign(Gtk.Align.CENTER)
|
||
self.toast_box = self.toast_label
|
||
self.toast_label.set_visible(False)
|
||
return self.toast_label
|
||
|
||
# -------------------------------------------------------------- lifecycle
|
||
def run(self):
|
||
if self.cfg.get("ui.fullscreen", True):
|
||
self.window.fullscreen()
|
||
self.window.show_all()
|
||
# show_all() would re-show things that must start hidden.
|
||
self.ssd_badge.set_visible(False)
|
||
self.toast_label.set_visible(False)
|
||
self.settings_box.set_visible(False)
|
||
self._settings_visible = False
|
||
self.rotate_button.set_label("⟳ %d°" % self.rotation)
|
||
if self.cfg.get("ui.hide_cursor", True):
|
||
self._hide_cursor()
|
||
|
||
self.worker.start()
|
||
interval = int(self.cfg.get("ui.preview_refresh_ms", 33))
|
||
GLib.timeout_add(interval, self._on_preview_tick)
|
||
GLib.timeout_add(int(self.cfg.get("ui.status_refresh_ms", 250)), self._on_status_tick)
|
||
GLib.timeout_add(int(self.cfg.get("ui.storage_poll_ms", 1000)), self._on_storage_tick)
|
||
# While the panel is open, keep the slider positions following the
|
||
# camera (AE keeps adjusting exposure/gain underneath).
|
||
GLib.timeout_add(500, self._on_controls_tick)
|
||
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGTERM, self._on_signal)
|
||
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, self._on_signal)
|
||
self._update_storage_ui()
|
||
Gtk.main()
|
||
|
||
def _hide_cursor(self):
|
||
display = Gdk.Display.get_default()
|
||
if display is None:
|
||
return
|
||
window = self.window.get_window()
|
||
if window is None:
|
||
return
|
||
try:
|
||
cursor = Gdk.Cursor.new_for_display(display, Gdk.CursorType.BLANK_CURSOR)
|
||
window.set_cursor(cursor)
|
||
except Exception:
|
||
log.debug("could not hide cursor", exc_info=True)
|
||
|
||
def shutdown(self):
|
||
log.info("shutting down")
|
||
try:
|
||
self.worker.stop()
|
||
self.worker.join(timeout=20)
|
||
except Exception:
|
||
log.debug("worker shutdown raised", exc_info=True)
|
||
Gtk.main_quit()
|
||
|
||
def _on_signal(self, *_args):
|
||
self.shutdown()
|
||
return False
|
||
|
||
def _on_destroy(self, *_args):
|
||
self.shutdown()
|
||
|
||
def _on_key_press(self, _widget, event):
|
||
key = event.keyval
|
||
if key in (Gdk.KEY_Escape, Gdk.KEY_q):
|
||
self.shutdown()
|
||
return True
|
||
if key in (Gdk.KEY_r, Gdk.KEY_R):
|
||
self._on_record_clicked(None)
|
||
return True
|
||
if key in (Gdk.KEY_p, Gdk.KEY_P):
|
||
self._on_photo_clicked(None)
|
||
return True
|
||
if key in (Gdk.KEY_t, Gdk.KEY_T):
|
||
self._cycle_rotation()
|
||
return True
|
||
if key in (Gdk.KEY_e, Gdk.KEY_E):
|
||
self._on_status_clicked(None, None)
|
||
return True
|
||
return False
|
||
|
||
# --------------------------------------------------------- rotate button
|
||
def _on_rotate_clicked(self, _button):
|
||
self._cycle_rotation()
|
||
|
||
def _cycle_rotation(self):
|
||
self.set_rotation((self.rotation + 90) % 360)
|
||
|
||
def set_rotation(self, degrees):
|
||
"""Rotate preview and all saved output by *degrees* clockwise."""
|
||
self.rotation = int(degrees) % 360
|
||
self.preview.set_rotation(self.rotation)
|
||
self.rotate_button.set_label("⟳ %d°" % self.rotation)
|
||
self.preview.queue_draw()
|
||
self.cfg.data.setdefault("ui", {})["rotation"] = self.rotation
|
||
try:
|
||
self.cfg.save()
|
||
except OSError:
|
||
log.debug("could not persist rotation", exc_info=True)
|
||
self.toast("Rotation %d°" % self.rotation)
|
||
|
||
# ------------------------------------------------------ settings panel
|
||
def _on_status_clicked(self, _widget, _event):
|
||
self._settings_visible = not self._settings_visible
|
||
if self._settings_visible:
|
||
self._refresh_control_ranges()
|
||
self.settings_box.set_visible(self._settings_visible)
|
||
self.status_action.set_markup(
|
||
"hide settings" if self._settings_visible else "tap for exposure / gain"
|
||
)
|
||
return False
|
||
|
||
def _on_controls_tick(self):
|
||
if self._settings_visible and self.worker.auto_exposure:
|
||
# Only track the camera automatically while AE is in charge; once the
|
||
# operator has taken manual control the sliders must not be yanked
|
||
# back by our own refresh.
|
||
self._refresh_control_values()
|
||
return True
|
||
|
||
def _refresh_control_values(self):
|
||
"""Update slider positions/labels from the latest camera metadata."""
|
||
metadata = self.worker.last_metadata
|
||
if not metadata:
|
||
return
|
||
self._suspend_control_writes = True
|
||
for name, row in self._control_rows.items():
|
||
info = row.get("range")
|
||
if not info:
|
||
continue
|
||
current = metadata.get(name)
|
||
if current is None:
|
||
continue
|
||
current = float(current)
|
||
if row["kind"] == "log":
|
||
low = max(1.0, float(info["min"]))
|
||
high = max(low + 1.0, float(info["max"]))
|
||
position = (np.log(max(low, current)) - np.log(low)) / (np.log(high) - np.log(low))
|
||
row["scale"].set_value(float(np.clip(position, 0.0, 1.0)))
|
||
else:
|
||
row["scale"].set_value(current)
|
||
row["value"].set_text(self._format_control_value(name, current))
|
||
self._suspend_control_writes = False
|
||
|
||
def _refresh_control_ranges(self):
|
||
"""Populate slider ranges from the camera and show current values."""
|
||
ranges = self.worker.control_ranges()
|
||
metadata = self.worker.last_metadata
|
||
self._suspend_control_writes = True
|
||
for name, row in self._control_rows.items():
|
||
info = ranges.get(name)
|
||
if not info or float(info["max"]) <= float(info["min"]):
|
||
row["scale"].set_sensitive(False)
|
||
row["value"].set_text("n/a")
|
||
continue
|
||
row["scale"].set_sensitive(True)
|
||
row["range"] = info
|
||
current = metadata.get(name)
|
||
if current is None:
|
||
current = info["default"]
|
||
current = float(current)
|
||
if row["kind"] == "log":
|
||
# Slider travels 0..1 in log space across the control's range.
|
||
low = max(1.0, float(info["min"]))
|
||
high = max(low + 1.0, float(info["max"]))
|
||
position = (np.log(max(low, current)) - np.log(low)) / (np.log(high) - np.log(low))
|
||
row["scale"].set_value(float(np.clip(position, 0.0, 1.0)))
|
||
else:
|
||
row["scale"].set_value(current)
|
||
row["value"].set_text(self._format_control_value(name, current))
|
||
self._suspend_control_writes = False
|
||
self.auto_button.set_label(
|
||
"Auto exposure: ON" if self.worker.auto_exposure else "Auto exposure: OFF"
|
||
)
|
||
|
||
@staticmethod
|
||
def _format_control_value(name, value):
|
||
if name == "ExposureTime":
|
||
if value >= 1_000_000:
|
||
return "%.2f s" % (value / 1_000_000.0)
|
||
return "%.2f ms" % (value / 1000.0)
|
||
if name == "AnalogueGain":
|
||
return "%.2fx" % value
|
||
return "%.2f" % value
|
||
|
||
@staticmethod
|
||
def _slider_to_value(row, position):
|
||
info = row["range"]
|
||
low, high = float(info["min"]), float(info["max"])
|
||
if row["kind"] == "log":
|
||
low = max(1.0, low)
|
||
high = max(low + 1.0, high)
|
||
return float(np.exp(np.log(low) + position * (np.log(high) - np.log(low))))
|
||
return low + position * (high - low)
|
||
|
||
def _on_slider_changed(self, scale, name):
|
||
if self._suspend_control_writes:
|
||
return
|
||
row = self._control_rows.get(name)
|
||
if not row or not row["range"]:
|
||
return
|
||
position = float(scale.get_value())
|
||
value = self._slider_to_value(row, position)
|
||
row["value"].set_text(self._format_control_value(name, value))
|
||
changes = {name: value}
|
||
# Touching exposure or gain means the operator wants manual control, so
|
||
# AE has to stand down or it would fight the slider every frame.
|
||
if name in ("ExposureTime", "AnalogueGain") and self.worker.auto_exposure:
|
||
self.worker.set_auto_exposure(False)
|
||
self.auto_button.set_label("Auto exposure: OFF")
|
||
if name == "ExposureTime":
|
||
changes["AeEnable"] = False
|
||
# The video config pins FrameDurationLimits to 1/target_fps to keep
|
||
# the recording rate honest. A manual exposure longer than that
|
||
# period would simply be clamped, so widen the window to match.
|
||
relaxed = self.worker.relaxed_frame_duration_limits(value)
|
||
if relaxed:
|
||
changes["FrameDurationLimits"] = relaxed
|
||
self.worker.push_controls(changes)
|
||
|
||
def _on_auto_toggled(self, _button):
|
||
enable = not self.worker.auto_exposure
|
||
self.worker.set_auto_exposure(enable)
|
||
self.auto_button.set_label(
|
||
"Auto exposure: ON" if enable else "Auto exposure: OFF"
|
||
)
|
||
changes = {"AeEnable": enable}
|
||
if enable:
|
||
# Hand exposure and gain back to the AE algorithm and restore the
|
||
# frame-duration pin that keeps recording at the target rate.
|
||
changes["FrameDurationLimits"] = self.worker.frame_duration_limits()
|
||
self._suspend_control_writes = True
|
||
for name in ("ExposureTime", "AnalogueGain"):
|
||
row = self._control_rows.get(name)
|
||
if row:
|
||
row["scale"].set_value(0.0)
|
||
self._suspend_control_writes = False
|
||
self.worker.push_controls(changes)
|
||
self._refresh_control_ranges()
|
||
else:
|
||
self.worker.push_controls(changes)
|
||
self.toast("Auto exposure %s" % ("enabled" if enable else "disabled"))
|
||
|
||
# ---------------------------------------------------------------- actions
|
||
def _on_record_clicked(self, _button):
|
||
if self.worker.recording:
|
||
summary = self.worker.recorder.stop("stopped by user")
|
||
self.toast("Saved %d frames (%.1f GB)" % (
|
||
summary.get("frames", 0), summary.get("bytes", 0) / 1e9))
|
||
return
|
||
|
||
state = self.storage.state()
|
||
if state != "ready" and state == "unmounted":
|
||
self.toast("Mounting SSD...")
|
||
self.storage.mount()
|
||
self._update_storage_ui()
|
||
state = self.storage.state()
|
||
if state != "ready":
|
||
self.toast("SSD not mounted — recording disabled", True)
|
||
return
|
||
if not self._camera_ready:
|
||
self.toast("Camera not ready", True)
|
||
return
|
||
|
||
session = self.storage.new_session("record")
|
||
self._session_dir = session
|
||
self.worker.recorder.start(session, rotation=self.rotation)
|
||
self.toast("Recording to %s" % self.storage.relative(session))
|
||
self._refresh_record_button()
|
||
|
||
def _on_photo_clicked(self, _button):
|
||
self.worker.submit_command("photo")
|
||
|
||
def _on_mount_clicked(self, _button):
|
||
ok, message = self.storage.mount()
|
||
if not ok:
|
||
self.toast("Mount failed: %s" % message, True)
|
||
self._update_storage_ui()
|
||
|
||
# ------------------------------------------------------------ UI updates
|
||
def _on_preview_tick(self):
|
||
frame = self.worker.preview_frame()
|
||
if frame is not None:
|
||
self.preview.push_frame(frame)
|
||
self.preview.queue_draw()
|
||
return True
|
||
|
||
def _on_status_tick(self):
|
||
stats = self.worker.recorder.stats()
|
||
metadata = self.worker.last_metadata
|
||
recording = self.worker.recording
|
||
|
||
if recording:
|
||
elapsed = stats.get("elapsed", 0.0)
|
||
self.status_main.set_markup(
|
||
"<b>● REC %s</b> %d frames %.1f fps %.0f MB/s%s" % (
|
||
_mmss(elapsed), stats.get("frames", 0), stats.get("actual_fps", 0.0),
|
||
stats.get("mb_per_s", 0.0),
|
||
(" dropped %d" % stats["dropped"]) if stats.get("dropped") else "",
|
||
)
|
||
)
|
||
elif not self._camera_ready:
|
||
self.status_main.set_markup("<b>starting camera...</b>")
|
||
else:
|
||
sensor = self.cfg["sensor"]
|
||
self.status_main.set_markup("<b>READY</b> %dx%d 12-bit RAW @ %d fps target" % (
|
||
sensor["width"], sensor["height"], self.cfg.get("record.target_fps", 30)))
|
||
|
||
free_mb = self.storage.free_mb()
|
||
free_text = ("%.0f GB free" % (free_mb / 1024)) if free_mb else "SSD not mounted"
|
||
exposure = metadata.get("ExposureTime")
|
||
gain = metadata.get("AnalogueGain")
|
||
temp = metadata.get("SensorTemperature")
|
||
parts = [free_text, "mode %s" % (self.worker.camera.mode if self.worker.camera else "?")]
|
||
if exposure:
|
||
parts.append("exp %.1f ms" % (exposure / 1000.0))
|
||
if gain:
|
||
parts.append("gain %.1f" % gain)
|
||
if temp:
|
||
parts.append("%.0f °C" % temp)
|
||
self.status_info.set_text(" ".join(parts))
|
||
self._refresh_record_button()
|
||
return True
|
||
|
||
def _on_storage_tick(self):
|
||
# Backstop for the udev rule: if the drive is present but nothing has
|
||
# mounted it (udev missed the event, boot-time race, manual umount) then
|
||
# mount it here. The helper returns immediately when the drive is ready.
|
||
state = self.storage.state()
|
||
if state == "unmounted":
|
||
now = time.monotonic()
|
||
if now - self._last_mount_attempt > 5.0:
|
||
self._last_mount_attempt = now
|
||
ok, message = self.storage.mount()
|
||
if ok:
|
||
log.info("auto-mounted the capture SSD")
|
||
self.toast("SSD mounted")
|
||
else:
|
||
log.warning("auto-mount failed: %s", message)
|
||
|
||
self._update_storage_ui()
|
||
if self.worker.recording and self.storage.state() != "ready":
|
||
summary = self.worker.recorder.stop("ssd removed")
|
||
self.toast("SSD removed — recording stopped (%d frames)" % summary.get("frames", 0), True)
|
||
return True
|
||
|
||
def _update_storage_ui(self):
|
||
state = self.storage.state()
|
||
self._last_storage_state = state
|
||
if state == "ready":
|
||
self.ssd_badge.set_visible(False)
|
||
self.mount_button.set_visible(False)
|
||
else:
|
||
self.ssd_badge.set_visible(True)
|
||
self.ssd_label.set_markup("⚠ SSD NOT MOUNTED")
|
||
if state == "unmounted":
|
||
self.ssd_hint.set_text("plugged in — tap MOUNT SSD")
|
||
self.ssd_hint.set_visible(True)
|
||
self.mount_button.set_visible(True)
|
||
else:
|
||
self.ssd_hint.set_text("plug the capture SSD into USB")
|
||
self.ssd_hint.set_visible(True)
|
||
self.mount_button.set_visible(False)
|
||
|
||
def _refresh_record_button(self):
|
||
recording = self.worker.recording
|
||
ctx = self.record_button.get_style_context()
|
||
if recording:
|
||
if not ctx.has_class("recording"):
|
||
ctx.add_class("recording")
|
||
self.record_button.set_label("■ STOP")
|
||
else:
|
||
if ctx.has_class("recording"):
|
||
ctx.remove_class("recording")
|
||
self.record_button.set_label("● REC")
|
||
self.photo_button.set_sensitive(not recording)
|
||
|
||
def toast(self, message, error=False):
|
||
ctx = self.toast_label.get_style_context()
|
||
if error and not ctx.has_class("toast-error"):
|
||
ctx.add_class("toast-error")
|
||
elif not error and ctx.has_class("toast-error"):
|
||
ctx.remove_class("toast-error")
|
||
self.toast_label.set_text(message)
|
||
self.toast_label.set_visible(True)
|
||
if self._toast_timer:
|
||
GLib.source_remove(self._toast_timer)
|
||
self._toast_timer = GLib.timeout_add(2600, self._hide_toast)
|
||
|
||
def _hide_toast(self):
|
||
self.toast_label.set_visible(False)
|
||
self._toast_timer = None
|
||
return False
|
||
|
||
# ------------------------------------------------------- worker callbacks
|
||
def on_camera_ready(self):
|
||
self._camera_ready = True
|
||
self._update_storage_ui()
|
||
return False
|
||
|
||
def on_camera_error(self, message):
|
||
self._camera_ready = False
|
||
self.status_main.set_markup("<b>CAMERA ERROR</b>")
|
||
self.status_info.set_text(message)
|
||
self.toast("Camera error: %s" % message, True)
|
||
return False
|
||
|
||
def on_recorder_state_change(self):
|
||
self._refresh_record_button()
|
||
return False
|
||
|
||
def on_photo_started(self):
|
||
self.toast("Capturing 4056x3040 still...")
|
||
return False
|
||
|
||
def on_photo_saved(self, summary):
|
||
files = ", ".join(sorted(k.upper() for k in summary.get("files", {})))
|
||
self.toast("%s saved (%s)" % (summary["stem"], files))
|
||
return False
|
||
|
||
def on_photo_failed(self, message):
|
||
self.toast("Photo failed: %s" % message, True)
|
||
return False
|
||
|
||
def on_tiff_done(self, job, elapsed, error):
|
||
if error:
|
||
self.toast(error, True)
|
||
else:
|
||
self.toast("16-bit TIFF written in %.1fs" % (elapsed or 0.0))
|
||
return False
|
||
|
||
|
||
def _mmss(seconds):
|
||
seconds = max(0.0, float(seconds))
|
||
minutes, secs = divmod(seconds, 60)
|
||
return "%02d:%04.1f" % (int(minutes), secs)
|
||
|
||
|
||
def main():
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
||
stream=sys.stdout,
|
||
)
|
||
cfg = Config()
|
||
app = MicroscopeApp(cfg)
|
||
try:
|
||
app.run()
|
||
except KeyboardInterrupt:
|
||
app.shutdown()
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|