nissaphotobooth/camera.py

"""Threaded webcam capture with a consistent 4:3 center crop.

The reader thread continuously grabs frames so the UI never blocks on the
camera. Every frame is center-cropped to 4:3 before anyone sees it, so the
live preview and the saved photos are guaranteed to show the same framing.
"""

import glob
import os
import threading
import time

import cv2
import numpy as np


def resolve_device(setting, wait_seconds=15):
    """Turn the config's camera device setting into something cv2 accepts.

    Accepts an integer index, an explicit "/dev/videoN" path, or
    "auto-<name>" which scans /dev/v4l/by-id/ for a device whose stable
    name contains <name> (case-insensitive), e.g. "auto-macally".

    A freshly plugged-in USB camera takes a few seconds to enumerate, so
    the auto- form keeps rescanning for up to wait_seconds before failing.
    """
    if isinstance(setting, int):
        return setting
    setting = str(setting)
    if setting.startswith("/dev/"):
        return setting
    if setting.startswith("auto-"):
        needle = setting[len("auto-"):].lower()
        deadline = time.monotonic() + wait_seconds
        waited = False
        while True:
            candidates = sorted(glob.glob("/dev/v4l/by-id/*-video-index0"))
            for path in candidates:
                if needle in os.path.basename(path).lower():
                    return os.path.realpath(path)
            if time.monotonic() > deadline:
                names = [os.path.basename(p) for p in candidates]
                raise RuntimeError(
                    f"No camera matching '{needle}' found. "
                    f"Available cameras: {names or 'none'}"
                )
            if not waited:
                waited = True
                print(f"Waiting for '{needle}' camera to show up "
                      f"(just plugged in? give it a moment)...")
            time.sleep(1)
    return int(setting)


def crop_4x3(frame):
    """Center-crop a BGR frame to 4:3."""
    h, w = frame.shape[:2]
    if w * 3 > h * 4:  # too wide: trim sides
        new_w = h * 4 // 3
        x = (w - new_w) // 2
        return frame[:, x:x + new_w]
    else:  # too tall: trim top/bottom
        new_h = w * 3 // 4
        y = (h - new_h) // 2
        return frame[y:y + new_h, :]


class Camera:
    def __init__(self, device_setting, width, height):
        device = resolve_device(device_setting)
        self.cap = cv2.VideoCapture(device, cv2.CAP_V4L2)
        if not self.cap.isOpened():
            raise RuntimeError(f"Could not open camera {device!r}")
        # MJPG lets most USB cams deliver full frame rates at higher resolutions
        self.cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*"MJPG"))
        self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
        self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)

        self._lock = threading.Lock()
        self._frame = None
        self._brightness = 0.0
        self._running = True
        self._thread = threading.Thread(target=self._reader, daemon=True)
        self._thread.start()

    def _reader(self):
        while self._running:
            ok, frame = self.cap.read()
            if not ok:
                continue
            frame = crop_4x3(frame)
            with self._lock:
                self._frame = frame
                self._brightness = float(frame.mean())

    def looks_black(self):
        """True when the camera is delivering pure black frames.

        The Macally cam (Z-Star chip) sometimes wedges and streams all-zero
        frames until it is physically unplugged and replugged; this lets the
        UI warn the operator instead of silently taking black photos.
        """
        with self._lock:
            return self._frame is not None and self._brightness < 0.5

    def get_frame(self):
        """Latest 4:3 BGR frame, or None if the camera hasn't warmed up yet."""
        with self._lock:
            return self._frame

    def save_photo(self, path):
        """Write the latest frame as a JPEG. Returns True on success."""
        frame = self.get_frame()
        if frame is None:
            return False
        return cv2.imwrite(path, frame, [cv2.IMWRITE_JPEG_QUALITY, 95])

    def close(self):
        self._running = False
        self._thread.join(timeout=2)
        self.cap.release()


def frame_to_surface(frame, size, mirror=False):
    """Convert a BGR frame to a pygame surface scaled to `size` (w, h)."""
    import pygame

    rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    if mirror:
        rgb = np.ascontiguousarray(rgb[:, ::-1])
    surf = pygame.image.frombuffer(rgb.tobytes(), (rgb.shape[1], rgb.shape[0]), "RGB")
    return pygame.transform.smoothscale(surf, size)


if __name__ == "__main__":
    # Quick standalone check: python3 camera.py [device]
    # Grabs a frame from the configured camera and reports on it.
    import sys
    import time
    import tomllib
    from pathlib import Path

    if len(sys.argv) > 1:
        device = sys.argv[1]
    else:
        with open(Path(__file__).parent / "config.toml", "rb") as f:
            device = tomllib.load(f)["camera"]["device"]

    print(f"Opening camera {device!r}...")
    c = Camera(device, 640, 480)
    time.sleep(2)
    frame = c.get_frame()
    if frame is None:
        print("PROBLEM: no frames arriving from the camera.")
    elif c.looks_black():
        print("PROBLEM: camera is streaming black frames. "
              "Unplug it, plug it back in, and try again.")
    else:
        h, w = frame.shape[:2]
        out = "/tmp/camera_test.jpg"
        c.save_photo(out)
        print(f"OK: {w}x{h} frames, test photo saved to {out}")
    c.close()