34cd247dca
Touchscreen kiosk for the Raspberry Pi HQ camera (IMX477) used as a microscope: live preview from boot, raw frame recording, full-resolution stills, and automatic capture-SSD mounting. - app.py: GTK3 fullscreen UI over cage — preview, REC/STOP, PHOTO, 90-degree rotate button, tap-to-open exposure/gain/settings sliders, and an "SSD NOT MOUNTED" badge - camera.py: picamera2 wrapper with separate video (ISP preview + packed 12-bit raw) and still (4056x3040) configurations - recorder.py: writer thread turning the raw stream into one Bayer frame per file plus index.jsonl/session.json, with dropped/missed accounting - photos.py: still capture to DNG + JPEG, and a background worker that renders a viewable 16-bit sRGB TIFF from the 12-bit Bayer frame - storage.py: SSD mount state and the date/time/format session layout - tools/unpack_raw.py: desktop-side unpacker/demosaicer - system/ + install.sh: systemd units, udev rule, mount helper, sudoers Note: raw frames are written byte-for-byte as the sensor produced them, so they stay decodeable regardless of how the stills pipeline evolves.
195 lines
7.0 KiB
Python
195 lines
7.0 KiB
Python
"""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
|