commit 34cd247dca815e87cca540bf250d7f1e48d142b2 Author: Microscope Camera Date: Sat Oct 31 00:44:46 2026 -0400 Microscope camera kiosk 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. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..16c34dc --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +# Byte-compiled / cached files +__pycache__/ +*.py[cod] +*.egg-info/ + +# Local scratch +*.log +*.tmp +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..1c07b5a --- /dev/null +++ b/README.md @@ -0,0 +1,232 @@ +# Microscope Camera + +Touchscreen kiosk for the Raspberry Pi 4 + IMX477 (HQ) microscope camera. + +* live preview on the 1024x600 HDMI touchscreen from boot +* **REC / STOP** — raw 12-bit Bayer frame recording, 2028x1520, ~29-30 fps +* **PHOTO** — full-resolution 4056x3040 still as DNG + 16-bit TIFF + JPEG +* **⟳ rotate** (bottom-left) — turns the image 90° at a time, on screen *and* + in the files it writes +* **tap the status pill** (top-left) — opens sliders for exposure, gain, + sharpness, contrast, saturation and brightness; tap it again to close +* a red **SSD NOT MOUNTED** badge that appears only when the capture SSD is + missing, and the SSD auto-mounts whenever it is plugged in +* everything lands on the SSD as `date/time/format/` + +## On-screen controls + +| Control | Where | What it does | +|---|---|---| +| `● REC` / `■ STOP` | bottom-right | starts/stops raw frame capture into a new `date/time/raw` session | +| `◉ PHOTO` | bottom-right | full-resolution still: DNG + 16-bit TIFF + JPEG in one `date/time` folder | +| `⟳ n°` | bottom-left | rotates preview and saved output clockwise in 90° steps (remembered across reboots) | +| status pill | top-left | tap to show/hide the camera settings sliders | +| `⏏ MOUNT SSD` | above REC, only when needed | appears when the drive is present but not mounted | + +### Rotating + +The rotate button turns both the live preview and everything that gets saved: + +* **JPEG / TIFF** pixels are physically rotated before they are written. +* **DNG** keeps its sensor-native Bayer data (rotating a mosaic would change its + colour-filter pattern) and records the rotation in the standard TIFF + `Orientation` tag, which raw converters honour. +* **Raw recordings** store frames sensor-native too and record the angle in + `session.json` as `rotation`; `tools/unpack_raw.py` applies it on export, so + raw frames you unpack on a desktop come out the way the screen showed them. + +### Camera settings panel + +Tapping the top-left status pill reveals sliders for the controls the sensor +actually exposes (anything unsupported is greyed out): + +* **Exposure** — logarithmic, from the sensor minimum (1 µs) up to 66 ms, so the + useful low end is not crammed into the first pixel of travel. +* **Gain** — analogue gain, 1.0x to 16.0x. +* **Sharpness / Contrast / Saturation / Brightness** — ISP tuning. + +Touching Exposure or Gain switches **Auto exposure** off automatically (AE would +otherwise overwrite the slider every frame). The button at the top of the panel +hands control back to AE and restores the frame-duration pin. + +A manual exposure longer than `1/target_fps` temporarily widens +`FrameDurationLimits`, because the recording pin would otherwise clamp it; the +sensor cannot both expose for 50 ms and deliver 30 fps, so the frame rate drops +for as long as the long exposure is set. Auto exposure restores the pin. + +## Layout on the SSD + +``` +/mnt/microscope/ +├── 2026-10-30/ <- date +│ ├── 21-33-05/ <- time (start of the session) +│ │ ├── raw/ +│ │ │ ├── frame_000001.raw 4.67 MB packed 12-bit Bayer +│ │ │ └── ... +│ │ ├── index.jsonl one line per frame: timestamp, exposure, gain, drops +│ │ └── session.json totals: frames, bytes, fps, dropped, missed +│ └── 21-40-12/ <- a photo session +│ ├── raw/still_0001.dng 18 MB, untouched 12-bit Bayer +│ ├── tiff/still_0001.tiff 74 MB, 16-bit sRGB (demosaiced, WB+colour matrix) +│ ├── jpeg/still_0001.jpg ISP-processed 8-bit +│ └── photo.json exposure, gain, white balance, timings +└── _archive/ (old test recordings moved off the SD card) +``` + +## What the record button actually writes + +Frames are the sensor's raw stream, stored **verbatim**: MIPI-packed 12-bit +Bayer, 2028x1520 (2x2 binned, full field of view), one file per frame. No +debayer, no compression, no white balance — nothing is baked in. Per frame: +`stride=3072`, `frame_bytes=4669440`, format `SBGGR12_CSI2P`. + +Frame rate is governed by the SSD, not the sensor. Measured on this hardware: + +| Target | Rate needed | Result | +|---|---|---| +| 40 fps (sensor mode max) | 247 MB/s | not possible | +| **30 fps** | **140 MB/s** | **sustained: 29.0 fps, 0 dropped, 0 missed** | +| 25 fps | 117 MB/s | comfortable | + +The drive sustains ~156 MB/s (~145 MB/s after its 220 MB SLC burst), so 30 fps is +close to the limit. Frame integrity is never fudged: `session.json` reports +`dropped` (writer queue overflow) and `missed` (sensor frames that never reached +disk, computed from `SensorTimestamp` gaps). + +`record.limit_frame_duration` pins the sensor's frame duration to +`1/target_fps`. Without it auto exposure can pick e.g. 66 ms in dim light and +the recording silently runs at 15 fps. With it, AE compensates with gain +instead. Set it `false` if you would rather have a bright, slow image. + +## Reading raw frames on a desktop + +```bash +# whole session -> 16-bit .npy per frame (geometry read from session.json) +python3 tools/unpack_raw.py /media/ssd/2026-10-30/21-33-05 + +# quick look: 8-bit PNG, contrast stretched +python3 tools/unpack_raw.py --png frame_000001.raw + +# 16-bit greyscale Bayer TIFF +python3 tools/unpack_raw.py --tiff16 frame_000001.raw + +# 16-bit RGB TIFF (bilinear demosaic of the 12-bit data) +python3 tools/unpack_raw.py --rgb16 frame_000001.raw + +# session metadata +python3 tools/unpack_raw.py --info /media/ssd/2026-10-30/21-33-05 +``` + +Raw files are loadable without the tool as long as you know the geometry: + +```python +import numpy as np +buf = np.fromfile("frame_000001.raw", dtype=np.uint8).reshape(1520, 3072) +trip = buf[:, :2028 * 3 // 2].reshape(1520, 1014, 3).astype(np.uint16) +b0, b1, b2 = trip[:, :, 0], trip[:, :, 1], trip[:, :, 2] +even = (b0 << 4) | (b1 >> 4) # 12-bit unpacked, 0-4095 +odd = ((b1 & 0x0F) << 8) | b2 +image = np.empty((1520, 2028), np.uint16) +image[:, 0::2], image[:, 1::2] = even, odd +``` + +Note the byte order: in each 3-byte group byte0 holds bits 11..4 of the first +pixel, byte1's **high** nibble holds its bits 3..0, and byte1's **low** nibble +holds the top of the second pixel. Swapping those nibbles still yields values in +0-4095, so it fails silently as colour-fringed mush — if your Python is +producing that, this is why. + +## The 16-bit TIFF on the photo button + +The Pi's ISP only ever outputs 8-bit processed images, so a "16-bit TIFF of the +ISP output" would just be 8-bit data in a 16-bit container. Instead the TIFF is +built from the 12-bit Bayer frame through the same chain a raw converter uses: + + unpack 12-bit MIPI -> black-level subtract -> white balance + -> colour-correction matrix -> bilinear demosaic -> sRGB gamma -> 16-bit + +so it opens as a normal, viewable 16-bit image rather than raw linear data. The +untouched sensor data stays in the DNG beside it, so nothing is lost for +quantitative work. The settings are in `config.json` under `photo`: set +`tiff_apply_white_balance` / `tiff_apply_colour_matrix` / `tiff_apply_gamma` to +`false` for a linear, un-white-balanced export instead. The camera's gains and +matrix are recorded in `photo.json` either way. + +## Layout of the software + +``` +/opt/microscope-camera/ +├── app.py GTK3 kiosk UI: preview, buttons, badge, settings sliders +├── camera.py picamera2 wrapper: video (raw+preview) and still configs +├── recorder.py writer thread for raw frames + index.jsonl/session.json +├── photos.py still capture (DNG/JPEG) and the 16-bit TIFF worker +├── storage.py SSD mount state and the date/time/format session paths +├── config.py defaults; config.json holds overrides +├── config.json tunables (edit and restart the service to apply) +├── install.sh reinstalls the app + system integration onto a fresh OS +├── system/ copies of the installed systemd units, udev rule, helper +└── tools/unpack_raw.py desktop-side unpacker / demosaicer +``` + +The copies in `system/` are the source of truth for the machine-specific files; +`install.sh` puts them back in place (see below). + +System integration: + +| Piece | Purpose | +|---|---| +| `/etc/systemd/system/microscope-camera.service` | runs `cage -s -- python3 app.py` fullscreen on tty1, `Restart=always` | +| `/usr/local/sbin/microscope-ssd` | `mount` / `umount` / `status` helper (log: `/run/microscope-ssd.log`) | +| `/etc/udev/rules.d/99-microscope-ssd.rules` | on plug-in, starts the mount service | +| `/etc/systemd/system/microscope-ssd-mount.service` | oneshot mount unit triggered by udev | +| `/etc/systemd/system/microscope-boot-check.service` | writes a boot self-test to `~/boot-check.log` + a screenshot | +| `/etc/fstab` | `UUID=...` entry so the SSD also mounts at boot | +| `/etc/sudoers.d/microscope-ssd` | lets the kiosk UI mount without a password | + +Why the mount is split this way: udev workers run **without `CAP_SYS_ADMIN`**, so +`mount` inside a `RUN+=` rule always fails with `permission denied`. The mount +has to happen in a service. That service is deliberately oneshot *without* +`RemainAfterExit`, otherwise it stays "active" after the first mount and systemd +silently skips every later replug. + +The SSD is mounted by three independent mechanisms, so a missing mount is always +caught: the `fstab` entry mounts it at boot, the udev rule mounts it on hotplug, +and the running app re-tries every few seconds as a backstop. + +## Reinstalling + +```bash +sudo ./install.sh # packages, app, units, udev rule, fstab, enable +``` + +`install.sh` is idempotent: it replaces the app and units, leaves an existing +`config.json` alone, and only appends the fstab line if it is not already there. + +## Operating + +```bash +systemctl status microscope-camera.service # is the UI running +journalctl -u microscope-camera.service -f # live app log +sudo -n /usr/local/sbin/microscope-ssd status # mounted | stale | unmounted | absent +sudo systemctl restart microscope-camera # after editing config.json +``` + +Touch buttons work directly. With a keyboard the same actions are on +`r` (record), `p` (photo), `t` (rotate), `e` (settings panel) and `Esc`/`q` quit. + +Recording refuses to start when the SSD is not mounted, so nothing silently +fills the SD card. If the SSD is yanked mid-recording the writer notices within +~0.5 s, stops the recording, and keeps whatever was already flushed. +`stale` in the helper's status means a mount entry survived a yank; the next +mount clears it automatically. + +## Known limits + +* **Light.** With the lens capped, AE wants 66 ms and the preview is black + noise. Pin the frame duration (default) or add light for 30 fps. +* The SSD's ~156 MB/s ceiling is the whole reason 40 fps raw is not possible at + 2028x1520. A faster drive would raise the ceiling. +* Photos freeze the preview for ~2-3 s (sensor mode switch + settle + capture). + The 74 MB TIFF is written by a background thread afterwards. +* DNG comes from `pidng` (pure Python): ~0.6 s at 12 MP, fine for stills, far + too slow for frames. diff --git a/app.py b/app.py new file mode 100644 index 0000000..017df5c --- /dev/null +++ b/app.py @@ -0,0 +1,1003 @@ +#!/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( + "● REC %s %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("starting camera...") + else: + sensor = self.cfg["sensor"] + self.status_main.set_markup("READY %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("CAMERA ERROR") + 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()) diff --git a/camera.py b/camera.py new file mode 100644 index 0000000..d45ed5f --- /dev/null +++ b/camera.py @@ -0,0 +1,194 @@ +"""picamera2 wrapper: preview + raw recording pipeline, and the still pipeline. + +Two sensor modes are used: + +* **video** – 2028x1520 (IMX477 2x2 binned, full field of view, up to 40 fps). + The ISP produces a 1024x600 XRGB8888 preview while the raw stream gives + MIPI-packed 12-bit Bayer frames for recording. +* **still** – the full 4056x3040 mode, used only for the photo button. + +Mode switches are cheap (~0.1 s) but they do restart AE/AWB, so they only happen +for stills. +""" + +import logging + +from picamera2 import Picamera2 + +log = logging.getLogger("camera") + + +class MicroscopeCamera: + def __init__(self, cfg): + self.cfg = cfg + self.picam2 = Picamera2() + self._video_config = None + self._still_config = None + self._mode = None + self.build_configs() + + # ---------------------------------------------------------------- config + def build_configs(self): + cfg = self.cfg + sensor = cfg["sensor"] + preview = cfg["preview"] + record = cfg["record"] + photo = cfg["photo"] + + self._video_config = self.picam2.create_video_configuration( + main={ + "size": (int(preview["width"]), int(preview["height"])), + "format": str(preview["format"]), + }, + raw={"size": (int(sensor["width"]), int(sensor["height"]))}, + sensor={ + "output_size": (int(sensor["width"]), int(sensor["height"])), + "bit_depth": int(sensor["bit_depth"]), + }, + buffer_count=int(record["buffer_count"]), + ) + # Pin the frame duration so auto exposure cannot quietly drop the + # recording rate. In dim light AE would otherwise choose e.g. 66 ms and + # the sensor would only manage ~15 fps; here it must use gain instead. + self._video_config["controls"] = {} + if record.get("limit_frame_duration", True): + frame_us = int(1_000_000 / max(1, int(record.get("target_fps", 30)))) + self._video_config["controls"]["FrameDurationLimits"] = (frame_us, frame_us) + log.info("frame duration pinned to %d us (%d fps)", frame_us, + int(record.get("target_fps", 30))) + + self._still_config = self.picam2.create_still_configuration( + main={ + "size": (int(photo["width"]), int(photo["height"])), + "format": "YUV420", + }, + raw={"size": (int(photo["width"]), int(photo["height"]))}, + sensor={ + "output_size": (int(photo["width"]), int(photo["height"])), + "bit_depth": 12, + }, + buffer_count=int(photo["buffer_count"]), + ) + return self._video_config, self._still_config + + # ------------------------------------------------------------- lifecycle + def start(self, mode="video"): + config = self._video_config if mode == "video" else self._still_config + self.picam2.configure(config) + self.picam2.options["quality"] = int(self.cfg.get("photo.jpeg_quality", 95)) + self.picam2.start() + self._mode = mode + log.info("camera started in %s mode", mode) + return config + + def stop(self): + try: + self.picam2.stop() + except Exception: # already stopped + log.debug("camera stop raised", exc_info=True) + self._mode = None + + def close(self): + try: + self.picam2.close() + except Exception: + log.debug("camera close raised", exc_info=True) + + def switch_to_still(self): + if self._mode != "still": + self.picam2.switch_mode(self._still_config) + self._mode = "still" + + def switch_to_video(self): + if self._mode != "video": + self.picam2.switch_mode(self._video_config) + self._mode = "video" + + # --------------------------------------------------------------- capture + def capture_request(self): + return self.picam2.capture_request() + + def preview_array(self, request): + """Raster preview frame as a writable numpy array (h, w, 4) uint8.""" + return request.make_array("main") + + def capture_metadata(self): + return self.picam2.capture_metadata() + + def stream_config(self, name): + try: + return self.picam2.stream_configuration(name) + except Exception: + return None + + # --------------------------------------------------------------- controls + def control_ranges(self): + """Ranges for the controls the settings panel exposes. + + picamera2 reports ``camera_controls`` as {name: (min, max, default)}; + anything the sensor/IPA does not offer is simply left out so the panel + can hide that row instead of offering a slider that does nothing. + """ + wanted = ( + "ExposureTime", "AnalogueGain", "Sharpness", "Contrast", + "Saturation", "Brightness", "ColourGains", "FrameDurationLimits", + ) + ranges = {} + try: + available = self.picam2.camera_controls + except Exception: + return ranges + for name in wanted: + if name in available: + low, high, default = available[name] + ranges[name] = {"min": low, "max": high, "default": default} + return ranges + + def apply_controls(self, controls): + """Apply a control dict; called from the camera thread only.""" + if not controls: + return + try: + self.picam2.set_controls(controls) + log.info("controls applied: %s", controls) + except Exception: + log.exception("could not apply controls: %s", controls) + + def frame_duration_limits(self, fps): + """FrameDurationLimits tuple pinning the sensor to *fps*.""" + frame_us = int(1_000_000 / max(1, int(fps))) + return (frame_us, frame_us) + + def relaxed_frame_duration_limits(self, exposure_us): + """Frame-duration window that admits a manual exposure of *exposure_us*. + + The recording pipeline pins FrameDurationLimits to 1/target_fps; a + longer manual exposure would just be clamped by that pin, so the window + has to be widened. It is clamped to what the sensor actually offers + (IMX477 reports roughly 33 ms .. 120 ms here) because libcamera rejects + out-of-range control values outright. + """ + info = self.control_ranges().get("FrameDurationLimits") + if not info: + return None + low = int(info["min"]) + high = int(info["max"]) + wanted = int(max(low, exposure_us)) + return (min(wanted, high), high) + + def raw_geometry(self): + """Describe the raw stream for the session index.""" + cfg = self.stream_config("raw") + if not cfg: + return None + return { + "format": cfg.get("format"), + "width": cfg.get("size", (None, None))[0], + "height": cfg.get("size", (None, None))[1], + "stride": cfg.get("stride"), + "frame_bytes": cfg.get("framesize"), + } + + @property + def mode(self): + return self._mode diff --git a/config.json b/config.json new file mode 100644 index 0000000..a2a7b7b --- /dev/null +++ b/config.json @@ -0,0 +1,47 @@ +{ + "archive_min_free_mb": 500, + "device_label": "MICROSCOPE", + "fs_uuid": "caf4320b-0fa3-4674-bfa9-f17f6165f681", + "mount_point": "/mnt/microscope", + "photo": { + "black_level": 256, + "buffer_count": 2, + "height": 3040, + "jpeg_quality": 95, + "settle_frames": 3, + "tiff_apply_colour_matrix": true, + "tiff_apply_gamma": true, + "tiff_apply_white_balance": true, + "white_level": 4095, + "width": 4056, + "write_dng": true, + "write_jpeg": true, + "write_tiff": true + }, + "preview": { + "format": "XRGB8888", + "height": 600, + "width": 1024 + }, + "record": { + "buffer_count": 12, + "index_stride": 1, + "limit_frame_duration": true, + "target_fps": 30, + "write_queue": 8 + }, + "sensor": { + "bit_depth": 12, + "height": 1520, + "sensor_fps": 40, + "width": 2028 + }, + "ui": { + "fullscreen": true, + "hide_cursor": true, + "preview_refresh_ms": 33, + "rotation": 0, + "status_refresh_ms": 250, + "storage_poll_ms": 1000 + } +} diff --git a/config.py b/config.py new file mode 100644 index 0000000..cfdfa97 --- /dev/null +++ b/config.py @@ -0,0 +1,120 @@ +"""Configuration for the microscope camera app. + +Everything tunable lives in ``config.json`` next to this file; missing keys fall +back to ``DEFAULTS`` so an older config file never breaks a new release. +""" + +import copy +import json +import os + +HERE = os.path.dirname(os.path.abspath(__file__)) +CONFIG_PATH = os.path.join(HERE, "config.json") + +DEFAULTS = { + # ---- storage --------------------------------------------------------- + "mount_point": "/mnt/microscope", + "fs_uuid": "caf4320b-0fa3-4674-bfa9-f17f6165f681", + "device_label": "MICROSCOPE", + "archive_min_free_mb": 500, # refuse to keep recording below this + + # ---- sensor / video pipeline (live preview + raw recording) ---------- + # 2028x1520 is the IMX477's 2x2-binned mode: full field of view at up to 40fps. + "sensor": {"width": 2028, "height": 1520, "bit_depth": 12, "sensor_fps": 40}, + + # Preview is produced by the ISP and handed to cairo, so it must be a + # 4-byte format: XRGB8888 in memory is B,G,R,X == cairo FORMAT_RGB24. + "preview": {"width": 1024, "height": 600, "format": "XRGB8888"}, + + "record": { + "target_fps": 30, # frame pacing request (SSD sustains ~29 fps at 12-bit packed) + "buffer_count": 12, # camera buffers: absorb SSD write stalls (~300ms at 40fps) + "write_queue": 8, # requests queued for the writer thread + "index_stride": 1, # 1 = log every frame in index.jsonl + # Pin the sensor's frame duration to 1/target_fps. Without this, auto + # exposure is free to pick e.g. 66 ms in dim light and the recording + # silently runs at 15 fps. With it, AE compensates using gain instead. + # Set false if you would rather have a bright slow image than 30 fps. + "limit_frame_duration": True, + }, + + # ---- still photo pipeline (full 4056x3040) -------------------------- + "photo": { + "width": 4056, + "height": 3040, + "buffer_count": 2, + "jpeg_quality": 95, + "write_dng": True, + "write_tiff": True, + "write_jpeg": True, + "settle_frames": 3, # let AE/AWB settle before the frame we keep + # The TIFF is meant to be a viewable 16-bit sRGB image. The DNG remains + # the untouched linear Bayer frame for quantitative work, so applying + # white balance + CCM to the TIFF costs nothing that cannot be recovered. + "tiff_apply_white_balance": True, + "tiff_apply_colour_matrix": True, + "tiff_apply_gamma": True, + # IMX477 12-bit levels; used to normalise the Bayer data before demosaic. + "black_level": 256, + "white_level": 4095, + }, + + "ui": { + "fullscreen": True, + "hide_cursor": True, + "preview_refresh_ms": 33, + "status_refresh_ms": 250, + "storage_poll_ms": 1000, + # 0/90/180/270, changed by the rotate button and remembered across boots + "rotation": 0, + }, +} + + +def _deep_merge(base, override): + out = copy.deepcopy(base) + for key, value in (override or {}).items(): + if isinstance(value, dict) and isinstance(out.get(key), dict): + out[key] = _deep_merge(out[key], value) + else: + out[key] = value + return out + + +class Config: + """Dict-like config with attribute-free access and disk persistence.""" + + def __init__(self, path=CONFIG_PATH): + self.path = path + self.data = copy.deepcopy(DEFAULTS) + self.load() + + def load(self): + if os.path.exists(self.path): + try: + with open(self.path, "r", encoding="utf-8") as handle: + user = json.load(handle) + self.data = _deep_merge(DEFAULTS, user) + except (OSError, ValueError): + # A corrupt config must never stop the microscope from booting. + self.data = copy.deepcopy(DEFAULTS) + return self.data + + def save(self): + tmp = self.path + ".tmp" + with open(tmp, "w", encoding="utf-8") as handle: + json.dump(self.data, handle, indent=2, sort_keys=True) + handle.write("\n") + os.replace(tmp, self.path) + + def get(self, path, default=None): + """Fetch a nested value: ``cfg.get("record.target_fps")``.""" + node = self.data + for part in path.split("."): + if not isinstance(node, dict) or part not in node: + return default + node = node[part] + return node + + def __getitem__(self, key): + return self.data[key] diff --git a/install.sh b/install.sh new file mode 100755 index 0000000..0214616 --- /dev/null +++ b/install.sh @@ -0,0 +1,80 @@ +#!/bin/bash +# Install the microscope camera kiosk and its storage plumbing onto this machine. +# +# This is the scripted form of what was originally set up by hand; running it on +# a fresh Raspberry Pi OS install should reproduce the whole system: +# +# sudo ./install.sh +# +# Everything it writes is listed in the README under "System integration". +set -euo pipefail + +HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PREFIX="${PREFIX:-/opt/microscope-camera}" +MOUNT_POINT="${MOUNT_POINT:-/mnt/microscope}" + +if [[ $EUID -ne 0 ]]; then + echo "run me with sudo" >&2 + exit 1 +fi + +echo "==> packages" +apt-get update +apt-get install -y --no-install-recommends \ + cage python3-gi python3-gi-cairo gir1.2-gtk-3.0 python3-cairo \ + python3-picamera2 python3-pidng python3-simplejpeg python3-tifffile \ + python3-pil python3-numpy grim + +echo "==> application -> $PREFIX" +mkdir -p "$PREFIX" +install -m 0644 "$HERE/app.py" "$HERE/camera.py" "$HERE/config.py" \ + "$HERE/photos.py" "$HERE/recorder.py" "$HERE/storage.py" \ + "$HERE/README.md" "$PREFIX/" +if [[ ! -f "$PREFIX/config.json" ]]; then + install -m 0644 "$HERE/config.json" "$PREFIX/" +fi +mkdir -p "$PREFIX/tools" +install -m 0755 "$HERE/tools/unpack_raw.py" "$PREFIX/tools/" +chown -R tyler:tyler "$PREFIX" + +echo "==> SSD mount helper" +install -m 0755 "$HERE/system/microscope-ssd" /usr/local/sbin/microscope-ssd +install -m 0755 "$HERE/system/microscope-boot-check" /usr/local/sbin/microscope-boot-check + +echo "==> sudoers" +install -m 0440 -o root -g root "$HERE/system/sudoers-microscope-ssd" \ + /etc/sudoers.d/microscope-ssd + +echo "==> udev rule" +install -m 0644 "$HERE/system/99-microscope-ssd.rules" \ + /etc/udev/rules.d/99-microscope-ssd.rules + +echo "==> systemd units" +install -m 0644 "$HERE/system/microscope-camera.service" \ + "$HERE/system/microscope-ssd-mount.service" \ + "$HERE/system/microscope-boot-check.service" \ + /etc/systemd/system/ + +echo "==> fstab (mount the capture SSD at boot when present)" +if ! grep -q "$MOUNT_POINT" /etc/fstab; then + mkdir -p "$MOUNT_POINT" + UUID="$(blkid -s UUID -o value "$(lsblk -no PATH,LABEL | awk '$2=="MICROSCOPE"{print $1; exit}')" 2>/dev/null || true)" + if [[ -n "${UUID:-}" ]]; then + printf '\n# Microscope capture SSD - mounts automatically when plugged in\n' + printf 'UUID=%s %s ext4 nofail,noatime,errors=remount-ro,x-systemd.device-timeout=5 0 2\n' \ + "$UUID" "$MOUNT_POINT" >> /etc/fstab + else + echo " no MICROSCOPE-labelled partition found; add the fstab line by hand" >&2 + fi +fi + +echo "==> enable services" +systemctl daemon-reload +udevadm control --reload-rules +systemctl enable microscope-camera.service microscope-boot-check.service +systemctl restart microscope-camera.service + +echo +echo "done. The kiosk should now be on the display." +echo " systemctl status microscope-camera.service" +echo " journalctl -u microscope-camera.service -f" diff --git a/photos.py b/photos.py new file mode 100644 index 0000000..6e2e888 --- /dev/null +++ b/photos.py @@ -0,0 +1,443 @@ +"""Full-resolution still capture: DNG + 16-bit TIFF + JPEG. + +The IMX477 is a 12-bit sensor and the Pi's ISP only outputs 8-bit processed +images, so a genuinely 16-bit TIFF cannot come from the ISP. Instead the TIFF is +built from the raw Bayer frame through the same chain a raw converter would use: + + unpack 12-bit MIPI -> black-level subtract -> white balance + -> colour-correction matrix -> bilinear demosaic -> sRGB gamma -> 16-bit + +The result is a normal, viewable 16-bit image rather than linear sensor data. +The untouched sensor data stays in the DNG next to it, so nothing is lost for +quantitative work. + +The slow part (demosaic + 74 MB TIFF write) runs on a worker thread so the live +preview resumes as soon as the capture itself is done. +""" + +import datetime as _dt +import json +import logging +import os +import queue +import threading +import time + +import numpy as np + +log = logging.getLogger("photos") + +_PATTERN_OFFSETS = { + # pattern -> (row parity, col parity) of the red sample in the 2x2 tile + "RGGB": (0, 0), + "BGGR": (1, 1), + "GRBG": (0, 1), + "GBRG": (1, 0), +} + + +def pattern_from_format(fmt: str) -> str: + """'SRGGB12_CSI2P' -> 'RGGB'.""" + if not fmt: + return "RGGB" + text = fmt.upper() + if text.startswith("S"): + text = text[1:] + return text[:4] if text[:4] in _PATTERN_OFFSETS else "RGGB" + + +def unpack_mipi12(packed: np.ndarray, width: int, height: int, stride: int) -> np.ndarray: + """Unpack MIPI CSI-2 packed 12-bit Bayer into a uint16 (height, width) array. + + Two pixels live in three bytes with the *first* pixel in the high bits: + + byte0 byte1 byte2 + p0[11:4] p0[3:0] p1[11:8] p1[7:0] + + so even pixels are ``(b0 << 4) | (b1 >> 4)`` and odd pixels are + ``((b1 & 0x0F) << 8) | b2``. Getting this order wrong still yields values in + the 0-4095 range, but it destroys the image: the tell-tale is that the two + green sites of the Bayer tile stop matching each other. + """ + rows = packed.reshape(height, stride) + useful = rows[:, : (width * 3) // 2] + triplets = useful.reshape(height, width // 2, 3).astype(np.uint16) + b0 = triplets[:, :, 0] + b1 = triplets[:, :, 1] + b2 = triplets[:, :, 2] + even = (b0 << 4) | (b1 >> 4) + odd = ((b1 & 0x0F) << 8) | b2 + out = np.empty((height, width), dtype=np.uint16) + out[:, 0::2] = even + out[:, 1::2] = odd + return out + + +def demosaic(raw: np.ndarray, pattern: str = "BGGR") -> np.ndarray: + """Bilinear demosaic of a Bayer plane into a uint16 (h, w, 3) RGB image. + + Quantisation-free for 12-bit input (max 4095) and cheap enough (~1 s at + 12 MP) to run on the Pi. + """ + rp, cp = _PATTERN_OFFSETS.get(pattern, (0, 0)) + h, w = raw.shape + tiles = raw.reshape(h // 2, 2, w // 2, 2).transpose(0, 2, 1, 3) + red = tiles[..., rp, cp].astype(np.uint32) + blue = tiles[..., 1 - rp, 1 - cp].astype(np.uint32) + green_r = tiles[..., rp, 1 - cp].astype(np.uint32) + green_b = tiles[..., 1 - rp, cp].astype(np.uint32) + + def padded(plane): + return np.pad(plane, 1, mode="edge") + + def upsample(plane): + p = padded(plane) + center = p[1:-1, 1:-1] + left = p[1:-1, :-2] + right = p[1:-1, 2:] + up = p[:-2, 1:-1] + down = p[2:, 1:-1] + horiz = (center + (right if cp == 0 else left)) // 2 + vert = (center + (down if rp == 0 else up)) // 2 + col_n = right if cp == 0 else left + row_n = down if rp == 0 else up + if cp == 0 and rp == 0: + corner = p[2:, 2:] + elif cp == 0 and rp == 1: + corner = p[2:, :-2] + elif cp == 1 and rp == 0: + corner = p[:-2, 2:] + else: + corner = p[:-2, :-2] + diag = (center + col_n + row_n + corner) // 4 + full = np.empty((h, w), dtype=np.uint16) + full[rp::2, cp::2] = center + full[rp::2, (1 - cp)::2] = horiz + full[(1 - rp)::2, cp::2] = vert + full[(1 - rp)::2, (1 - cp)::2] = diag + return full + + red_full = upsample(red) + blue_full = upsample(blue) + + g1 = padded(green_r) + g2 = padded(green_b) + g1_left = (g1[1:-1, 1:-1] + g1[1:-1, :-2]) // 2 + g2_right = (g2[1:-1, 1:-1] + g2[1:-1, 2:]) // 2 + green_full = np.empty((h, w), dtype=np.uint16) + green_full[rp::2, 1 - cp::2] = green_r + green_full[1 - rp::2, cp::2] = green_b + green_full[rp::2, cp::2] = g1_left + green_full[1 - rp::2, 1 - cp::2] = g2_right + + rgb = np.empty((h, w, 3), dtype=np.uint16) + rgb[..., 0] = red_full + rgb[..., 1] = green_full + rgb[..., 2] = blue_full + return rgb + + +def srgb_encode(linear: np.ndarray) -> np.ndarray: + """Apply the sRGB transfer function (linear light -> display values).""" + out = np.empty_like(linear) + low = linear <= 0.0031308 + np.multiply(linear, 12.92, out=out, where=low) + high = ~low + np.power(np.clip(linear, 1e-8, None), 1.0 / 2.4, out=out, where=high) + np.multiply(out, 1.055, out=out, where=high) + np.subtract(out, 0.055, out=out, where=high) + return out + + +def render_still(raw, pattern, black=256.0, white=4095.0, gains=None, ccm=None, + gamma=True) -> np.ndarray: + """Turn a 12-bit Bayer frame into a viewable 16-bit RGB image. + + Mirrors what a raw converter does, in the order the ISP does it: + black-level subtract, white balance, colour matrix, demosaic, gamma. + + ``gains`` is picamera2's ``ColourGains`` (red, blue; green is 1.0) and + ``ccm`` is its ``ColourCorrectionMatrix`` (3x3, camera RGB -> sRGB). Either + may be None, in which case that step is skipped. + """ + rgb = demosaic(raw, pattern) + img = rgb.astype(np.float32) + del rgb + + scale = 1.0 / float(white - black) + img *= scale + img -= black * scale + np.clip(img, 0.0, None, out=img) + + if gains is not None: + img[..., 0] *= float(gains[0]) + img[..., 2] *= float(gains[1]) + + if ccm is not None: + matrix = np.asarray(ccm, dtype=np.float32).reshape(3, 3) + img = img @ matrix.T + np.clip(img, 0.0, None, out=img) + + # Highlight clipping happens here, as it does in the ISP. + np.clip(img, 0.0, 1.0, out=img) + if gamma: + img = srgb_encode(img) + + return np.clip(img * 65535.0 + 0.5, 0, 65535).astype(np.uint16) + + +def _rotate_array(rgb, rotation): + """Rotate an (h, w, 3) array clockwise by 0/90/180/270 degrees.""" + rotation = int(rotation) % 360 + if rotation == 0: + return rgb + if rotation == 90: + return np.ascontiguousarray(np.rot90(rgb, k=3)) + if rotation == 180: + return np.ascontiguousarray(np.rot90(rgb, k=2)) + if rotation == 270: + return np.ascontiguousarray(np.rot90(rgb, k=1)) + return rgb + + +def _pil_transpose(rotation): + """PIL constant that rotates an image clockwise by *rotation* degrees. + + PIL's ROTATE_* constants turn counter-clockwise, hence the inversion. + """ + from PIL import Image + + transpose = getattr(Image, "Transpose", Image) + return {90: transpose.ROTATE_270, 180: transpose.ROTATE_180, + 270: transpose.ROTATE_90}[int(rotation) % 360] + + +def rotate_jpeg(path, rotation): + """Rotate a JPEG on disk in place (used for the ISP-written still).""" + rotation = int(rotation) % 360 + if rotation == 0: + return False + from PIL import Image + + with Image.open(path) as image: + turned = image.transpose(_pil_transpose(rotation)) + turned.save(path, quality=95, subsampling=0) + return True + + +def set_dng_orientation(path, rotation): + """Write the DNG/TIFF Orientation tag (274) in place. + + Rotating the Bayer mosaic itself would change the sensor pattern, so the raw + DNG keeps its pixels and records the rotation the way every raw converter + expects: as metadata. Orientation 1/3/6/8 = 0/180/90/270 degrees. + """ + value = {0: 1, 90: 6, 180: 3, 270: 8}.get(int(rotation) % 360) + if value in (None, 1): + return False + try: + with open(path, "r+b") as handle: + header = handle.read(8) + if header[:2] not in (b"II", b"MM"): + return False + order = "little" if header[:2] == b"II" else "big" + ifd_offset = int.from_bytes(header[4:8], order) + handle.seek(ifd_offset) + count = int.from_bytes(handle.read(2), order) + entries = handle.read(12 * count) + for index in range(count): + entry = entries[index * 12:(index + 1) * 12] + tag = int.from_bytes(entry[0:2], order) + if tag == 274: + handle.seek(ifd_offset + 2 + index * 12 + 8) + handle.write(value.to_bytes(2, order) + b"\x00\x00") + return True + except OSError: + log.exception("could not set DNG orientation on %s", path) + return False + + +class TiffWorker(threading.Thread): + """Serial background worker for the expensive demosaic + TIFF write.""" + + def __init__(self, on_done=None): + super().__init__(name="tiff-writer", daemon=True) + self._jobs = queue.Queue() + self._on_done = on_done + + def submit(self, job): + self._jobs.put(job) + + def stop(self, timeout=90): + """Finish any queued job, then end the thread.""" + if self.is_alive(): + self._jobs.put(None) + self.join(timeout=timeout) + + def run(self): + while True: + job = self._jobs.get() + if job is None: + log.info("TIFF worker stopping") + return + try: + self._do(job) + except Exception: + log.exception("TIFF job failed") + if self._on_done: + self._on_done(job, None, "TIFF failed (see log)") + + def _do(self, job): + import tifffile + + started = time.monotonic() + raw = unpack_mipi12(job["packed"], job["width"], job["height"], job["stride"]) + rgb = render_still( + raw, + job["pattern"], + black=job.get("black_level", 256.0), + white=job.get("white_level", 4095.0), + gains=job.get("gains") if job.get("apply_white_balance", True) else None, + ccm=job.get("ccm") if job.get("apply_ccm", True) else None, + gamma=job.get("gamma", True), + ) + del raw + rotation = int(job.get("rotation", 0) or 0) + if rotation: + rgb = _rotate_array(rgb, rotation) + + description = ( + "16-bit RGB from the 12-bit Bayer frame; " + f"pattern={job['pattern']} black={job.get('black_level')} " + f"white={job.get('white_level')} " + f"white_balance={job.get('gains') if job.get('apply_white_balance', True) else None} " + f"colour_matrix={'applied' if job.get('apply_ccm', True) and job.get('ccm') else 'none'} " + f"transfer={'sRGB' if job.get('gamma', True) else 'linear'}; " + "raw sensor data preserved in the DNG alongside." + ) + tifffile.imwrite( + job["path"], rgb, photometric="rgb", + description=description, compression=None, + ) + elapsed = time.monotonic() - started + log.info("TIFF written in %.1fs: %s", elapsed, job["path"]) + if self._on_done: + self._on_done(job, elapsed, None) + + +def capture_still(camera, storage, cfg, tiff_worker, on_tiff_done=None, rotation=0): + """Capture one full-resolution still. Must run on the camera thread. + + *rotation* (clockwise degrees) is applied to the saved output: the JPEG is + turned on disk, the TIFF is turned before it is written, and the DNG carries + it as an Orientation tag (its Bayer data must stay sensor-native). + """ + photo_cfg = cfg["photo"] + rotation = int(rotation or 0) % 360 + session = storage.new_session("photo") + raw_dir = storage.format_dir(session, "raw") + serial = storage.next_serial(raw_dir, "still") + stem = "still_%04d" % serial + + t_switch = time.monotonic() + camera.switch_to_still() + settle = max(0, int(photo_cfg.get("settle_frames", 3))) + for _ in range(settle): + stale = camera.capture_request() + stale.release() + switch_seconds = time.monotonic() - t_switch + + t_capture = time.monotonic() + request = camera.capture_request() + metadata = request.get_metadata() + written = {} + + try: + if photo_cfg.get("write_dng", True): + path = raw_dir / (stem + ".dng") + request.save_dng(str(path), "raw") + if rotation: + set_dng_orientation(str(path), rotation) + written["dng"] = str(path) + + if photo_cfg.get("write_jpeg", True): + jpeg_dir = storage.format_dir(session, "jpeg") + path = jpeg_dir / (stem + ".jpg") + request.save("main", str(path)) + if rotation: + rotate_jpeg(str(path), rotation) + written["jpeg"] = str(path) + + packed = None + if photo_cfg.get("write_tiff", True): + raw_cfg = camera.stream_config("raw") or {} + packed = np.array(request.make_buffer("raw"), dtype=np.uint8) + width = int(raw_cfg.get("size", (photo_cfg["width"], photo_cfg["height"]))[0]) + height = int(raw_cfg.get("size", (photo_cfg["width"], photo_cfg["height"]))[1]) + stride = int(raw_cfg.get("stride") or (width * 3 // 2)) + pattern = pattern_from_format(raw_cfg.get("format", "SRGGB12")) + ccm = metadata.get("ColourCorrectionMatrix") + tiff_dir = storage.format_dir(session, "tiff") + job = { + "packed": packed, + "width": width, + "height": height, + "stride": stride, + "pattern": pattern, + "gains": metadata.get("ColourGains"), + "apply_white_balance": photo_cfg.get("tiff_apply_white_balance", True), + "ccm": list(ccm) if ccm is not None else None, + "apply_ccm": photo_cfg.get("tiff_apply_colour_matrix", True), + "gamma": photo_cfg.get("tiff_apply_gamma", True), + "black_level": photo_cfg.get("black_level", 256), + "white_level": photo_cfg.get("white_level", 4095), + "rotation": rotation, + "path": str(tiff_dir / (stem + ".tiff")), + "stem": stem, + "session": str(session), + } + tiff_worker.submit(job) + written["tiff"] = job["path"] + finally: + request.release() + camera.switch_to_video() + + summary = { + "kind": "photo", + "stem": stem, + "session": str(session), + "taken": _dt.datetime.now().astimezone().isoformat(timespec="milliseconds"), + "files": {k: storage.relative(v) for k, v in written.items()}, + "mode": "4056x3040", + "rotation": rotation, + "capture_ms": round((time.monotonic() - t_capture) * 1000), + "switch_ms": round(switch_seconds * 1000), + "exposure_us": metadata.get("ExposureTime"), + "gain": metadata.get("AnalogueGain"), + "awb": list(metadata.get("ColourGains")) if metadata.get("ColourGains") else None, + "colour_correction_matrix": list(metadata["ColourCorrectionMatrix"]) + if metadata.get("ColourCorrectionMatrix") is not None else None, + "black_level": photo_cfg.get("black_level", 256), + "white_level": photo_cfg.get("white_level", 4095), + "sensor_temperature": metadata.get("SensorTemperature"), + "notes": [ + "DNG holds the untouched 12-bit Bayer frame.", + "TIFF is 16-bit sRGB, demosaiced from that Bayer frame with white balance", + "and the colour-correction matrix applied.", + "JPEG is the ISP-processed 8-bit image.", + "rotation is the clockwise display/export rotation in degrees.", + ], + } + _write_json(session / "photo.json", summary) + return summary + + +def _write_json(path, payload): + tmp = str(path) + ".tmp" + try: + with open(tmp, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2, default=str) + handle.write("\n") + os.replace(tmp, path) + except OSError: + log.exception("could not write %s", path) diff --git a/recorder.py b/recorder.py new file mode 100644 index 0000000..469db07 --- /dev/null +++ b/recorder.py @@ -0,0 +1,314 @@ +"""Raw frame recorder. + +Frames come straight off the sensor's raw stream as MIPI-packed 12-bit Bayer +(``SBGGR12_CSI2P``) and are written to disk verbatim: no debayer, no +compression, no loss. One camera buffer per file. + +Writes happen on a dedicated thread so the capture loop never blocks on the +SSD. The bounded queue plus the camera's own buffer count provide backpressure; +if the writer falls behind (the SSD sustains ~156 MB/s, i.e. ~29 fps at +2028x1520x12-bit) frames are dropped and counted, never silently reordered. +""" + +import json +import logging +import os +import queue +import threading +import time + +log = logging.getLogger("recorder") + + +def _write_raw_buffer(request, path): + """Write the raw stream buffer to *path*; return (bytes_written, geometry).""" + config = request.config.get("raw", {}) + try: + # Zero-copy path: borrow the camera's mapped buffer and write it out. + from picamera2.request import _MappedBuffer + except ImportError: # pragma: no cover - older picamera2 + _MappedBuffer = None + + if _MappedBuffer is not None: + with _MappedBuffer(request, "raw", write=False) as buf: + view = memoryview(buf) + with open(path, "wb") as handle: + handle.write(view) + return view.nbytes, config + + data = request.make_buffer("raw") + with open(path, "wb") as handle: + handle.write(data) + return int(data.nbytes), config + + +class RawRecorder: + """Owns the writer thread and the session index for one recording.""" + + def __init__(self, storage, cfg, on_state_change=None): + self.storage = storage + self.cfg = cfg + self.on_state_change = on_state_change + self.sensor_fps = float(cfg.get("sensor.sensor_fps", 40)) + self.frame_period_ns = int(1e9 / self.sensor_fps) + self.target_fps = int(cfg.get("record.target_fps", 30)) + self.min_free_mb = int(cfg.get("archive_min_free_mb", 500)) + + self._queue = queue.Queue(maxsize=int(cfg.get("record.write_queue", 8))) + self._thread = None + self._active = False + self._stop_requested = threading.Event() + self._lock = threading.Lock() + + self.session_dir = None + self.rotation = 0 + self.stop_reason = None + self._raw_geometry = None + self._index_handle = None + self._start_monotonic = None + self._start_iso = None + self._stats = {} + + # ------------------------------------------------------------------ state + @property + def active(self) -> bool: + return self._active + + def stats(self): + with self._lock: + data = dict(self._stats) + if data and self._start_monotonic: + elapsed = max(1e-6, time.monotonic() - self._start_monotonic) + data["elapsed"] = elapsed + data["actual_fps"] = data.get("frames", 0) / elapsed + data["mb_per_s"] = data.get("bytes", 0) / elapsed / 1e6 + return data + + # ------------------------------------------------------------------ start + def start(self, session_dir, rotation=0): + if self._active: + return None + session_dir = os.fspath(session_dir) + os.makedirs(os.path.join(session_dir, "raw"), exist_ok=True) + self.session_dir = session_dir + # Clockwise display rotation in degrees. Raw Bayer frames are stored + # sensor-native, so the rotation is recorded here for whatever reads + # them later (tools/unpack_raw.py applies it and also bakes it in when + # writing .npy/.tiff/.png output). + self.rotation = int(rotation or 0) % 360 + self.stop_reason = None + self._stop_requested.clear() + self._start_monotonic = time.monotonic() + self._start_iso = _iso_now() + self._stats = { + "frames": 0, "bytes": 0, "dropped": 0, "missed": 0, + "first_sensor_ts": None, "last_sensor_ts": None, "started": self._start_iso, + } + self._index_handle = open(os.path.join(session_dir, "index.jsonl"), "w", encoding="utf-8") + self._write_session_json(final=False) + self._active = True + self._thread = threading.Thread(target=self._writer_loop, name="raw-writer", daemon=True) + self._thread.start() + log.info("recording started: %s (target %d fps)", session_dir, self.target_fps) + if self.on_state_change: + self.on_state_change() + return session_dir + + # ----------------------------------------------------------------- submit + def submit(self, request): + """Hand a completed request to the writer. Takes ownership of *request*.""" + if not self._active: + request.release() + return + try: + self._queue.put_nowait(request) + except queue.Full: + with self._lock: + self._stats["dropped"] += 1 + request.release() + + # ------------------------------------------------------------------- stop + def stop(self, reason=None): + if not self._active: + return self.stats() + self._active = False + self.stop_reason = reason or "stopped" + self._stop_requested.set() + thread, self._thread = self._thread, None + if thread is not None: + thread.join(timeout=30) + # Anything still queued was never written; count it honestly. + leftovers = 0 + while True: + try: + request = self._queue.get_nowait() + except queue.Empty: + break + leftovers += 1 + request.release() + if leftovers: + with self._lock: + self._stats["dropped"] += leftovers + if self._index_handle is not None: + self._index_handle.close() + self._index_handle = None + self._write_session_json(final=True) + summary = self.stats() + log.info("recording stopped (%s): %d frames, %.1f MB, %d dropped, %.1f fps", + self.stop_reason, summary.get("frames", 0), summary.get("bytes", 0) / 1e6, + summary.get("dropped", 0), summary.get("actual_fps", 0.0)) + if self.on_state_change: + self.on_state_change() + return summary + + # ------------------------------------------------------------ writer side + def _writer_loop(self): + free_check_countdown = 15 # ~0.5s at 30fps + while True: + try: + request = self._queue.get(timeout=0.5) + except queue.Empty: + if self._stop_requested.is_set(): + break + continue + try: + self._write_one(request) + except Exception: # a bad frame must not kill the recording + log.exception("failed writing frame") + with self._lock: + self._stats["dropped"] += 1 + finally: + request.release() + + free_check_countdown -= 1 + if free_check_countdown <= 0: + free_check_countdown = 15 + reason = self._check_environment() + if reason: + self.stop_reason = reason + self._active = False + self._stop_requested.set() + break + if self._stop_requested.is_set() and self._queue.empty(): + break + + def _check_environment(self): + if not self.storage.is_mounted(): + log.warning("SSD disappeared while recording") + return "ssd removed" + free_mb = self.storage.free_mb() + if free_mb and free_mb < self.min_free_mb: + log.warning("SSD below minimum free space (%d MB)", free_mb) + return "disk full" + return None + + def _write_one(self, request): + metadata = request.get_metadata() + with self._lock: + stats = self._stats + frame_no = stats["frames"] + 1 + missed = 0 + sensor_ts = metadata.get("SensorTimestamp") + # Judge gaps against the duration the sensor is actually running at, + # which auto exposure can change; assuming the mode's maximum rate + # would report a "missed" frame on every single frame in dim light. + period_ns = self.frame_period_ns + frame_duration_us = metadata.get("FrameDuration") + if frame_duration_us: + period_ns = max(1, int(float(frame_duration_us) * 1000)) + if sensor_ts is not None: + last = stats["last_sensor_ts"] + if last is not None and sensor_ts > last: + gap = sensor_ts - last + missed = max(0, int(round(gap / period_ns)) - 1) + stats["last_sensor_ts"] = sensor_ts + if stats["first_sensor_ts"] is None: + stats["first_sensor_ts"] = sensor_ts + stats["missed"] += missed + + filename = "frame_%06d.raw" % frame_no + path = os.path.join(self.session_dir, "raw", filename) + written, raw_cfg = _write_raw_buffer(request, path) + + elapsed = time.monotonic() - self._start_monotonic + record = { + "frame": frame_no, + "file": "raw/" + filename, + "t": round(elapsed, 6), + "sensor_ts": sensor_ts, + "exposure_us": metadata.get("ExposureTime"), + "gain": _round(metadata.get("AnalogueGain")), + "awb": _round_pair(metadata.get("ColourGains")), + "lux": _round(metadata.get("Lux")), + "bytes": written, + "missed_before": missed, + } + if frame_no % int(self.cfg.get("record.index_stride", 1)) == 0: + self._index_handle.write(json.dumps(record, separators=(",", ":")) + "\n") + + with self._lock: + self._stats["frames"] = frame_no + self._stats["bytes"] += written + return raw_cfg + + # --------------------------------------------------------------- metadata + def set_raw_geometry(self, geometry): + """Remember the raw stream geometry so session.json can describe it.""" + self._raw_geometry = geometry + + def _write_session_json(self, final: bool): + summary = self.stats() + payload = { + "kind": "record", + "started": self._start_iso, + "ended": _iso_now() if final else None, + "stop_reason": self.stop_reason if final else None, + "sensor": self.cfg.get("sensor", {}), + "raw": getattr(self, "_raw_geometry", None), + "target_fps": self.target_fps, + "rotation": getattr(self, "rotation", 0), + "frames": summary.get("frames", 0), + "bytes": summary.get("bytes", 0), + "dropped": summary.get("dropped", 0), + "missed": summary.get("missed", 0), + "actual_fps": round(summary.get("actual_fps", 0.0), 3), + "mb_per_s": round(summary.get("mb_per_s", 0.0), 2), + "elapsed_s": round(summary.get("elapsed", 0.0), 3), + "layout": "date/HH-MM-SS//", + "frame_orientation": "sensor-native; apply 'rotation' clockwise for display", + "notes": [ + "Frames are MIPI-packed 12-bit Bayer, one file each, no processing.", + "'missed' counts sensor frames that never made it to disk.", + "Use tools/unpack_raw.py on a desktop to unpack to 16-bit .npy/.tiff.", + ], + } + path = os.path.join(self.session_dir, "session.json") + tmp = path + ".tmp" + try: + with open(tmp, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2) + handle.write("\n") + os.replace(tmp, path) + except OSError: + log.exception("could not write %s", path) + + +def _iso_now(): + import datetime + return datetime.datetime.now().astimezone().isoformat(timespec="milliseconds") + + +def _round(value, digits=4): + try: + return round(float(value), digits) + except (TypeError, ValueError): + return None + + +def _round_pair(value): + if not value: + return None + try: + return [round(float(value[0]), 4), round(float(value[1]), 4)] + except (TypeError, ValueError, IndexError): + return None diff --git a/storage.py b/storage.py new file mode 100644 index 0000000..edc3ccb --- /dev/null +++ b/storage.py @@ -0,0 +1,173 @@ +"""SSD discovery, mount state and the on-disk session layout. + +Layout (date, then time, then file format): + + /mnt/microscope/2026-10-30/21-33-05/raw/frame_000001.raw + /mnt/microscope/2026-10-30/21-33-05/index.jsonl + /mnt/microscope/2026-10-30/21-33-05/session.json + /mnt/microscope/2026-10-30/21-40-12/raw/still_0001.dng + /mnt/microscope/2026-10-30/21-40-12/tiff/still_0001.tiff + /mnt/microscope/2026-10-30/21-40-12/jpeg/still_0001.jpg +""" + +import datetime as _dt +import logging +import os +import re +import shutil +import subprocess +import time +from pathlib import Path + +log = logging.getLogger("storage") + +UUID_BY_PATH = "/dev/disk/by-uuid" +MOUNT_HELPER = "/usr/local/sbin/microscope-ssd" + + +class Storage: + def __init__(self, cfg): + self.cfg = cfg + self.mount_point = Path(cfg.get("mount_point", "/mnt/microscope")) + self.uuid = cfg.get("fs_uuid", "") + self._last_session_key = None + self._last_session_dir = None + + # ------------------------------------------------------------------ state + @property + def device_node(self): + if not self.uuid: + return None + node = Path(UUID_BY_PATH) / self.uuid + return node if node.exists() else None + + def device_present(self) -> bool: + """True when the SSD is plugged in and the partition node exists.""" + return self.device_node is not None + + def mount_source(self): + """Device backing the mount point, or None when it is a plain directory.""" + target = os.fspath(self.mount_point) + try: + with open("/proc/self/mounts", "r", encoding="utf-8") as handle: + for line in handle: + parts = line.split() + if len(parts) >= 2 and parts[1] == target: + return parts[0] + except OSError: + log.debug("could not read /proc/self/mounts", exc_info=True) + return None + + def is_mounted(self) -> bool: + """True only when the SSD is really mounted *and* still present. + + Yanking a USB drive leaves a stale entry in /proc/mounts that still + reports as a mount point, so the backing device is checked too. + """ + source = self.mount_source() + if source is None: + return False + try: + os.statvfs(self.mount_point) + except OSError: + return False + if self.uuid: + return (Path(UUID_BY_PATH) / self.uuid).exists() + return os.path.exists(source) + + def free_mb(self) -> int: + if not self.is_mounted(): + return 0 + try: + usage = shutil.disk_usage(self.mount_point) + return int(usage.free // (1024 * 1024)) + except OSError: + return 0 + + def state(self) -> str: + """One of: 'ready', 'unmounted', 'absent'.""" + if self.is_mounted(): + return "ready" + if self.device_present(): + return "unmounted" + return "absent" + + # ------------------------------------------------------------- mount ops + def _run_helper(self, action: str, timeout: float = 20.0): + if not os.access(MOUNT_HELPER, os.X_OK): + return False, f"{MOUNT_HELPER} not installed" + try: + proc = subprocess.run( + ["sudo", "-n", MOUNT_HELPER, action], + capture_output=True, text=True, timeout=timeout, + ) + except subprocess.TimeoutExpired: + return False, "mount helper timed out" + except OSError as exc: + return False, str(exc) + out = (proc.stdout + proc.stderr).strip() + return proc.returncode == 0, out + + def mount(self): + if self.is_mounted(): + return True, "already mounted" + return self._run_helper("mount") + + def unmount(self): + if not self.is_mounted(): + return True, "not mounted" + return self._run_helper("umount") + + # --------------------------------------------------------- session paths + def new_session(self, kind: str) -> Path: + """Create (or reuse) the date/time session directory for a capture. + + Two captures in the same second share a directory and continue the + frame/still numbering, which keeps rapid photo bursts tidy. + """ + now = _dt.datetime.now() + date_part = now.strftime("%Y-%m-%d") + time_part = now.strftime("%H-%M-%S") + key = f"{date_part}/{time_part}" + session = self.mount_point / date_part / time_part + + if key == self._last_session_key and session.is_dir(): + return session + + session.mkdir(parents=True, exist_ok=True) + self._last_session_key = key + self._last_session_dir = session + log.info("new %s session: %s", kind, session) + return session + + @staticmethod + def format_dir(session: Path, fmt: str) -> Path: + path = session / fmt + path.mkdir(exist_ok=True) + return path + + @staticmethod + def next_serial(directory: Path, prefix: str) -> int: + """Highest existing ``prefix_NNNN`` number in *directory*, plus one.""" + highest = 0 + pattern = re.compile(rf"^{re.escape(prefix)}_(\d+)\.[A-Za-z0-9]+$") + if directory.is_dir(): + for entry in directory.iterdir(): + match = pattern.match(entry.name) + if match: + highest = max(highest, int(match.group(1))) + return highest + 1 + + def relative(self, path: Path) -> str: + try: + return str(Path(path).relative_to(self.mount_point)) + except ValueError: + return str(path) + + def wait_mounted(self, timeout: float = 15.0) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if self.is_mounted(): + return True + time.sleep(0.25) + return self.is_mounted() diff --git a/system/99-microscope-ssd.rules b/system/99-microscope-ssd.rules new file mode 100644 index 0000000..71a1118 --- /dev/null +++ b/system/99-microscope-ssd.rules @@ -0,0 +1,7 @@ +# Auto-mount the microscope capture SSD as soon as it is plugged in. +# +# The mount itself must happen in a systemd service, not in RUN: udev workers +# run without CAP_SYS_ADMIN, so mount(2) there fails with EPERM. The service is +# oneshot *without* RemainAfterExit so it goes inactive again after each run and +# a later replug can start it afresh. +ACTION=="add", SUBSYSTEM=="block", ENV{ID_FS_UUID}=="caf4320b-0fa3-4674-bfa9-f17f6165f681", ENV{SYSTEMD_WANTS}+="microscope-ssd-mount.service", TAG+="systemd" diff --git a/system/microscope-boot-check b/system/microscope-boot-check new file mode 100755 index 0000000..d9dfa13 --- /dev/null +++ b/system/microscope-boot-check @@ -0,0 +1,17 @@ +#!/bin/bash +# One-shot boot verification: proves the kiosk came up and the SSD mounted. +LOG=/home/tyler/boot-check.log +export XDG_RUNTIME_DIR=/run/user/1000 WAYLAND_DISPLAY=wayland-0 +sleep 20 +{ + echo "=== boot check $(date -Is) ===" + echo "camera service : $(systemctl is-active microscope-camera.service) (enabled: $(systemctl is-enabled microscope-camera.service))" + echo "ssd mount : $(/usr/local/sbin/microscope-ssd status)" + echo "findmnt : $(findmnt -no SOURCE,TARGET,FSTYPE /mnt/microscope || echo 'not mounted')" + echo "cage : $(pgrep -a cage | head -1)" + echo "app : $(pgrep -af 'python3 app.py' | head -1)" +} >> "$LOG" +for d in HDMI-A-1 HDMI-A-2 DSI-1 DPI-1 HDMI-1; do + if grim -o "$d" /home/tyler/boot-check.png 2>>"$LOG"; then echo "screenshot : $d" >> "$LOG"; break; fi +done +chown tyler:tyler "$LOG" /home/tyler/boot-check.png 2>/dev/null diff --git a/system/microscope-boot-check.service b/system/microscope-boot-check.service new file mode 100644 index 0000000..57bd0cd --- /dev/null +++ b/system/microscope-boot-check.service @@ -0,0 +1,10 @@ +[Unit] +Description=Microscope camera boot verification +After=microscope-camera.service + +[Service] +Type=oneshot +ExecStart=/usr/local/sbin/microscope-boot-check + +[Install] +WantedBy=multi-user.target diff --git a/system/microscope-camera.service b/system/microscope-camera.service new file mode 100644 index 0000000..2da5317 --- /dev/null +++ b/system/microscope-camera.service @@ -0,0 +1,35 @@ +[Unit] +Description=Microscope camera kiosk (cage + capture UI) +Documentation=file:/opt/microscope-camera/README.md +After=systemd-user-sessions.service systemd-logind.service +Wants=systemd-logind.service +Conflicts=getty@tty1.service +After=getty@tty1.service +ConditionPathExists=/opt/microscope-camera/app.py + +[Service] +Type=simple +User=tyler +Group=tyler +# A logind session on tty1 gives cage DRM master plus input device access. +PAMName=login +TTYPath=/dev/tty1 +TTYReset=yes +TTYVHangup=yes +TTYVTDisallocate=yes +StandardInput=tty +StandardOutput=journal +StandardError=journal +Environment=XDG_RUNTIME_DIR=/run/user/1000 +Environment=HOME=/home/tyler +Environment=PYTHONUNBUFFERED=1 +ExecStartPre=/bin/sh -c 'install -d -m 700 -o tyler -g tyler /run/user/1000' +ExecStart=/usr/bin/cage -s -- /usr/bin/python3 /opt/microscope-camera/app.py +Restart=always +RestartSec=2 +# Keep the recorder alive long enough to flush its last frames on restart. +KillSignal=SIGTERM +TimeoutStopSec=30 + +[Install] +WantedBy=multi-user.target diff --git a/system/microscope-ssd b/system/microscope-ssd new file mode 100755 index 0000000..049bc89 --- /dev/null +++ b/system/microscope-ssd @@ -0,0 +1,96 @@ +#!/bin/bash +# Manage the microscope capture SSD mount. +# +# microscope-ssd mount mount the SSD (waits briefly for it to appear) +# microscope-ssd umount flush and unmount safely +# microscope-ssd status print: mounted | stale | unmounted | absent +# +# Called by the udev-triggered microscope-ssd-mount.service when the drive is +# plugged in, by fstab at boot, and by the kiosk UI (through a NOPASSWD sudoers +# entry) when the operator taps MOUNT SSD. +# +# stdout stays machine-readable (status) while everything also lands in a log +# file, because udev and systemd capture stdout in ways that are awkward to read. +set -u + +MOUNT_POINT="${MICROSCOPE_MOUNT:-/mnt/microscope}" +UUID="${MICROSCOPE_UUID:-caf4320b-0fa3-4674-bfa9-f17f6165f681}" +OWNER="${MICROSCOPE_OWNER:-tyler:tyler}" +DEV="/dev/disk/by-uuid/${UUID}" +LOG_FILE="${MICROSCOPE_LOG:-/run/microscope-ssd.log}" + +log() { + local message="microscope-ssd: $*" + echo "$message" >>"$LOG_FILE" 2>/dev/null || true + echo "$message" +} + +in_mounts() { grep -qs " ${MOUNT_POINT} " /proc/self/mounts; } + +do_status() { + # 'stale' means a mount entry survived a yank: the device node is gone. + if in_mounts; then + if [ -e "$DEV" ]; then echo mounted; else echo stale; fi + elif [ -e "$DEV" ]; then + echo unmounted + else + echo absent + fi +} + +do_mount() { + mkdir -p "$MOUNT_POINT" + + if in_mounts; then + if [ -e "$DEV" ]; then + log "already mounted" + return 0 + fi + log "clearing stale mount entry" + umount -l "$MOUNT_POINT" 2>/dev/null || true + fi + + for _ in $(seq 1 40); do + [ -e "$DEV" ] && break + sleep 0.25 + done + if [ ! -e "$DEV" ]; then + log "SSD not present ($DEV)" + return 1 + fi + + if mount -t ext4 -o noatime,errors=remount-ro "$DEV" "$MOUNT_POINT"; then + chown "$OWNER" "$MOUNT_POINT" 2>/dev/null || true + chmod 755 "$MOUNT_POINT" 2>/dev/null || true + log "mounted $DEV at $MOUNT_POINT" + return 0 + fi + log "mount failed" + return 1 +} + +do_umount() { + if ! in_mounts; then + log "not mounted" + return 0 + fi + sync + if umount "$MOUNT_POINT"; then + log "unmounted (safe to unplug)" + return 0 + fi + log "unmount failed - something is still using $MOUNT_POINT" + return 1 +} + +ACTION="${1:-status}" +case "$ACTION" in + mount|umount) log "--- $(date -Is) action=$ACTION euid=$(id -u) ---" ;; +esac + +case "$ACTION" in + mount) do_mount ;; + umount) do_umount ;; + status) do_status ;; + *) echo "usage: $0 {mount|umount|status}" >&2; exit 2 ;; +esac diff --git a/system/microscope-ssd-mount.service b/system/microscope-ssd-mount.service new file mode 100644 index 0000000..776170b --- /dev/null +++ b/system/microscope-ssd-mount.service @@ -0,0 +1,13 @@ +[Unit] +Description=Mount the microscope capture SSD +Documentation=man:mount(8) +After=systemd-udevd.service +Before=local-fs.target + +[Service] +# No RemainAfterExit: the unit returns to inactive after each run so that every +# subsequent plug-in event can start it again. The helper is idempotent, so an +# already-mounted drive is a no-op. +Type=oneshot +ExecStart=/usr/local/sbin/microscope-ssd mount +TimeoutStartSec=30 diff --git a/system/sudoers-microscope-ssd b/system/sudoers-microscope-ssd new file mode 100644 index 0000000..ba6b28b --- /dev/null +++ b/system/sudoers-microscope-ssd @@ -0,0 +1 @@ +tyler ALL=(root) NOPASSWD: /usr/local/sbin/microscope-ssd diff --git a/tools/unpack_raw.py b/tools/unpack_raw.py new file mode 100644 index 0000000..6bd909c --- /dev/null +++ b/tools/unpack_raw.py @@ -0,0 +1,365 @@ +#!/usr/bin/env python3 +"""Unpack the microscope camera's raw frames on a desktop machine. + +Recording writes MIPI-packed 12-bit Bayer frames (``SBGGR12_CSI2P`` style: two +pixels in three bytes) straight off the sensor. This tool turns them into files +you can actually look at: + + unpack_raw.py frame_000001.raw -> frame_000001.npy (uint16) + unpack_raw.py --png frame_000001.raw -> frame_000001.png (quick look) + unpack_raw.py --tiff16 frame_000001.raw -> frame_000001.tiff (16-bit mono) + unpack_raw.py --debayer --rgb16 frame_000001.raw-> frame_000001_rgb.tiff (16-bit RGB) + unpack_raw.py /mnt/microscope/2026-10-30/21-33-05 (whole session) + unpack_raw.py --info /mnt/microscope/2026-10-30/21-33-05 + +Geometry (width, height, stride, Bayer pattern) is picked up automatically from +the session's ``session.json``; if that is missing the tool falls back to the +file size and the command line defaults. +""" + +import argparse +import json +import os +import sys + +import numpy as np + +PATTERNS = { + # pattern -> (row parity, col parity) of the red sample in the 2x2 tile + "RGGB": (0, 0), + "BGGR": (1, 1), + "GRBG": (0, 1), + "GBRG": (1, 0), +} + + +def pattern_from_format(fmt): + text = (fmt or "SRGGB12").upper() + if text.startswith("S"): + text = text[1:] + head = text[:4] + return head if head in PATTERNS else "RGGB" + + +def unpack_mipi12(packed, width, height, stride): + """Unpack MIPI CSI-2 RAW12 (3 bytes -> 2 pixels). + + Byte order matters: byte0 holds the low 8 bits of pixel 0, byte1 holds the + high 4 bits of pixel 0 in its HIGH nibble and the high 4 bits of pixel 1 in + its LOW nibble. Treating byte1's low nibble as pixel 0's high bits (and + its high nibble as pixel 1's) transposes the 12-bit words within each pair, + which scrambles the Bayer mosaic and produces colour-fringed output. + """ + rows = packed[: stride * height].reshape(height, stride) + useful = rows[:, : (width * 3) // 2] + triplets = useful.reshape(height, width // 2, 3).astype(np.uint16) + b0 = triplets[:, :, 0] + b1 = triplets[:, :, 1] + b2 = triplets[:, :, 2] + even = (b0 << 4) | (b1 >> 4) + odd = ((b1 & 0x0F) << 8) | b2 + out = np.empty((height, width), dtype=np.uint16) + out[:, 0::2] = even + out[:, 1::2] = odd + return out + + +def demosaic(raw, pattern="RGGB"): + rp, cp = PATTERNS.get(pattern, (0, 0)) + h, w = raw.shape + tiles = raw.reshape(h // 2, 2, w // 2, 2).transpose(0, 2, 1, 3) + red = tiles[..., rp, cp].astype(np.uint32) + blue = tiles[..., 1 - rp, 1 - cp].astype(np.uint32) + green_r = tiles[..., rp, 1 - cp] + green_b = tiles[..., 1 - rp, cp] + + def upsample(plane): + p = np.pad(plane, 1, mode="edge") + center = p[1:-1, 1:-1] + left, right = p[1:-1, :-2], p[1:-1, 2:] + up, down = p[:-2, 1:-1], p[2:, 1:-1] + col_n = right if cp == 0 else left + row_n = down if rp == 0 else up + corner = p[2:, 2:] if (cp == 0 and rp == 0) else \ + p[2:, :-2] if (cp == 0 and rp == 1) else \ + p[:-2, 2:] if (cp == 1 and rp == 0) else p[:-2, :-2] + full = np.empty((h, w), dtype=np.uint16) + full[rp::2, cp::2] = center + full[rp::2, (1 - cp)::2] = (center + col_n) // 2 + full[(1 - rp)::2, cp::2] = (center + row_n) // 2 + full[(1 - rp)::2, (1 - cp)::2] = (center + col_n + row_n + corner) // 4 + return full + + red_full, blue_full = upsample(red), upsample(blue) + g1 = np.pad(green_r.astype(np.uint32), 1, mode="edge") + g2 = np.pad(green_b.astype(np.uint32), 1, mode="edge") + green_full = np.empty((h, w), dtype=np.uint16) + green_full[rp::2, (1 - cp)::2] = green_r + green_full[(1 - rp)::2, cp::2] = green_b + green_full[rp::2, cp::2] = (g1[1:-1, 1:-1] + g1[1:-1, :-2]) // 2 + green_full[(1 - rp)::2, (1 - cp)::2] = (g2[1:-1, 1:-1] + g2[1:-1, 2:]) // 2 + + rgb = np.empty((h, w, 3), dtype=np.uint16) + rgb[..., 1] = green_full + return rgb, red_full, blue_full + + +def apply_rotation(image, rotation): + """Rotate an array clockwise by 0/90/180/270 degrees.""" + rotation = int(rotation or 0) % 360 + if rotation == 0: + return image + if rotation == 90: + return np.ascontiguousarray(np.rot90(image, k=3)) + if rotation == 180: + return np.ascontiguousarray(np.rot90(image, k=2)) + if rotation == 270: + return np.ascontiguousarray(np.rot90(image, k=1)) + return image + + +_PATTERN_TILE = { + "RGGB": (("R", "G"), ("G", "B")), + "BGGR": (("B", "G"), ("G", "R")), + "GRBG": (("G", "R"), ("B", "G")), + "GBRG": (("G", "B"), ("R", "G")), +} + + +def rotate_pattern(pattern, rotation): + """CFA pattern after rotating the Bayer plane clockwise. + + Rotating a mosaic without moving its pattern would put red samples where + blue belong, so a rotated .npy/.tiff must record the rotated pattern. + """ + steps = (int(rotation or 0) % 360) // 90 + if not steps: + return pattern + tile = [list(row) for row in _PATTERN_TILE.get(pattern, _PATTERN_TILE["RGGB"])] + for _ in range(steps): + tile = [[tile[1][0], tile[0][0]], [tile[1][1], tile[0][1]]] # 90 deg clockwise + text = "".join(tile[0] + tile[1]) + for name, known in _PATTERN_TILE.items(): + if "".join(known[0] + known[1]) == text: + return name + return pattern + + +def find_session(path): + """Walk up from *path* looking for session.json / photo.json.""" + current = os.path.abspath(path if os.path.isdir(path) else os.path.dirname(path)) + for _ in range(4): + for name in ("session.json", "photo.json"): + candidate = os.path.join(current, name) + if os.path.isfile(candidate): + return current, candidate + parent = os.path.dirname(current) + if parent == current: + break + current = parent + return None, None + + +def geometry_for(path, args): + width, height, stride, pattern = args.width, args.height, args.stride, args.pattern + session_dir, meta_file = find_session(path) + meta = {} + if meta_file: + try: + with open(meta_file, "r", encoding="utf-8") as handle: + meta = json.load(handle) + except (OSError, ValueError): + meta = {} + raw = meta.get("raw") or {} + if raw: + width = width or raw.get("width") + height = height or raw.get("height") + stride = stride or raw.get("stride") + pattern = args.pattern or pattern_from_format(raw.get("format")) + path_size = os.path.getsize(path) + height = height or 1520 + if not width: + # packed 12-bit: bytes per row = width * 3/2 + if stride: + width = int(stride * 2 / 3) + else: + width = 2028 + if not stride: + stride = path_size // height + expected = width * height if args.unpacked else stride * height + if path_size < expected: + raise SystemExit( + "%s is %d bytes but %dx%d needs %d — pass --width/--height/--stride" + % (os.path.basename(path), path_size, width, height, expected) + ) + rotation = args.rotate if args.rotate is not None else int(meta.get("rotation") or 0) + return width, height, stride, pattern, session_dir, meta, rotation + + +def contrast_stretch(image): + lo, hi = np.percentile(image, (0.5, 99.5)) + if hi <= lo: + hi = lo + 1 + scaled = np.clip((image.astype(np.float32) - lo) * (255.0 / (hi - lo)), 0, 255) + return scaled.astype(np.uint8) + + +def srgb_encode(rgb16): + """Linear 16-bit RGB -> display-ready 8-bit sRGB. + + Without this the frame looks flat and washed out, because linear sensor data + held in a 16-bit container has all its mid-tones crammed into the dark end. + """ + x = np.clip(rgb16.astype(np.float32) / 65535.0, 0.0, 1.0) + x = np.where(x <= 0.0031308, x * 12.92, + 1.055 * np.power(x, 1.0 / 2.4) - 0.055) + return np.clip(x * 255.0 + 0.5, 0, 255).astype(np.uint8) + + +def white_balance(rgb, red_full, blue_full, gains): + """Apply camera white balance. Only called when --white-balance is given: + gains above 1.0 clip at 65535, which loses highlight data.""" + if not gains: + gains = (1.0, 1.0) + out = rgb + out[..., 0] = np.clip(red_full.astype(np.float32) * float(gains[0]), 0, 65535).astype(np.uint16) + out[..., 2] = np.clip(blue_full.astype(np.float32) * float(gains[1]), 0, 65535).astype(np.uint16) + return out + + +def process(path, args): + width, height, stride, pattern, session_dir, meta, rotation = geometry_for(path, args) + stem = os.path.splitext(os.path.basename(path))[0] + outdir = args.outdir or os.path.dirname(os.path.abspath(path)) + os.makedirs(outdir, exist_ok=True) + payload = np.fromfile(path, dtype=np.uint8) + + if args.unpacked: + image = payload[: width * height * 2].view(np.uint16).reshape(height, width) + else: + image = unpack_mipi12(payload, width, height, stride) + + print("%-28s %dx%d %s stride=%d rotation=%d° max=%d mean=%.1f" + % (os.path.basename(path), width, height, pattern, stride, rotation, + int(image.max()), float(image.mean()))) + + # The session's rotation is the operator's display orientation, so honour it + # in everything this tool writes. Bayer exports are rotated *with* their CFA + # pattern; the RGB path is demosaiced first and then turned, which is + # equivalent and avoids re-deriving the pattern. + bayer = apply_rotation(image, rotation) + bayer_pattern = rotate_pattern(pattern, rotation) + + made = [] + if args.npy or not (args.png or args.tiff16 or args.rgb16): + out = os.path.join(outdir, stem + ".npy") + np.save(out, bayer) + made.append(out) + if args.png: + from PIL import Image + out = os.path.join(outdir, stem + ".png") + Image.fromarray(contrast_stretch(bayer)).save(out) + made.append(out) + if args.tiff16: + import tifffile + out = os.path.join(outdir, stem + ".tiff") + tifffile.imwrite(out, bayer, + description="16-bit unpacked Bayer %s (rotated %d deg clockwise)" + % (bayer_pattern, rotation)) + made.append(out) + if args.rgb16: + import tifffile + gains = meta.get("awb") + # Demosaic in sensor orientation, fill all three channels, then rotate: + # rotating an RGB triple is equivalent to rotating the mosaic and + # re-deriving its CFA pattern, without the pattern bookkeeping. + rgb, red_full, blue_full = demosaic(image, pattern) + rgb[..., 0] = np.clip(red_full.astype(np.float32) * 16.0, 0, 65535).astype(np.uint16) + rgb[..., 2] = np.clip(blue_full.astype(np.float32) * 16.0, 0, 65535).astype(np.uint16) + rgb[..., 1] = np.clip(rgb[..., 1].astype(np.float32) * 16.0, 0, 65535).astype(np.uint16) + del red_full, blue_full, image + if rotation: + rgb = apply_rotation(rgb, rotation) + applied = False + if args.white_balance: + rgb = white_balance(rgb, rgb[..., 0], rgb[..., 2], gains) + applied = True + out = os.path.join(outdir, stem + "_rgb.tiff") + tifffile.imwrite(out, rgb, photometric="rgb", + description="linear 16-bit RGB, %s, rotation=%d deg clockwise, " + "white_balance_applied=%s camera_white_balance=%s" + % (pattern, rotation, applied, gains)) + made.append(out) + for item in made: + print(" -> %s (%.1f MB)" % (item, os.path.getsize(item) / 1e6)) + return made + + +def show_info(target): + session_dir, meta_file = find_session(target) + if not meta_file: + raise SystemExit("no session.json/photo.json found near %s" % target) + with open(meta_file, "r", encoding="utf-8") as handle: + meta = json.load(handle) + print(json.dumps(meta, indent=2)) + + +def collect_inputs(paths, recursive): + files = [] + for path in paths: + if os.path.isdir(path): + raw_dir = os.path.join(path, "raw") + scan = raw_dir if os.path.isdir(raw_dir) else path + for name in sorted(os.listdir(scan)): + if name.endswith(".raw"): + files.append(os.path.join(scan, name)) + else: + files.append(path) + return files + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("paths", nargs="+", help="raw files, session dirs, or date dirs") + parser.add_argument("--outdir", help="where to write results (default: next to input)") + parser.add_argument("--npy", action="store_true", help="write 16-bit .npy (default)") + parser.add_argument("--png", action="store_true", help="write 8-bit contrast-stretched PNG") + parser.add_argument("--tiff16", action="store_true", help="write 16-bit Bayer TIFF") + parser.add_argument("--rgb16", action="store_true", help="write 16-bit RGB TIFF (demosaiced)") + parser.add_argument("--debayer", action="store_true", help="alias for --rgb16") + parser.add_argument("--rotate", type=int, choices=(0, 90, 180, 270), + help="override the session's display rotation (degrees clockwise)") + parser.add_argument("--no-rotate", dest="rotate", action="store_const", const=0, + help="ignore the rotation recorded in the session metadata") + parser.add_argument("--white-balance", action="store_true", + help="apply the camera's colour gains to the RGB output " + "(clips highlights; off by default, gains are recorded anyway)") + parser.add_argument("--info", action="store_true", help="print session metadata and exit") + parser.add_argument("--pattern", choices=sorted(PATTERNS), help="Bayer pattern override") + parser.add_argument("--width", type=int, help="frame width override") + parser.add_argument("--height", type=int, help="frame height override") + parser.add_argument("--stride", type=int, help="row stride in bytes override") + parser.add_argument("--unpacked", action="store_true", + help="input is already unpacked 16-bit little-endian") + args = parser.parse_args() + + if args.debayer: + args.rgb16 = True + if args.info: + for path in args.paths: + show_info(path) + return 0 + + files = collect_inputs(args.paths, recursive=True) + if not files: + raise SystemExit("no .raw files found") + total = 0 + for path in files: + total += len(process(path, args)) + print("done: %d file(s) from %d input(s)" % (total, len(files))) + return 0 + + +if __name__ == "__main__": + sys.exit(main())