"""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