Various

apps/ctt-server/ctt_server/app.py

@@ -7,6 +7,7 @@
 # Everything runs in one process (on the Pi). The remote client is just a
 # browser. Captures use Picamera2 directly; the tuner runs as a subprocess.
 
+import contextlib
 import difflib
 import json
 import logging
@@ -54,6 +55,17 @@
     logger.warning('rawpy not available: DNG-only captures will have no in-browser preview')
 
 
+# libcamera installs its built-in (system) tuning files here, one JSON per
+# sensor, split by ISP platform. A local build under /usr/local takes precedence
+# over the distro package, mirroring libcamera's own search order.
+_SYSTEM_TUNING_ROOTS = ('/usr/local/share', '/usr/share')
+
+
+def _system_tuning_dirs(target: str) -> list[Path]:
+    """Installed libcamera tuning directories for an ISP platform ('pisp'/'vc4')."""
+    return [Path(root) / 'libcamera' / 'ipa' / 'rpi' / target for root in _SYSTEM_TUNING_ROOTS]
+
+
 def _run_info(available: dict) -> dict:
     """Map target -> {label, epoch} from each tuning file's mtime, for the UI."""
     info = {}
@@ -325,9 +337,7 @@ def preview_test(name: str):
                     }
                 ), 400
         try:
-            # Hand-edited tunings are canary-tested in a subprocess first: a bad
-            # one would otherwise wedge this process's camera until a restart.
-            cam = reload_shared_camera(tuning_file=str(json_path), preview_max_width=1920, validate=(kind == 'custom'))
+            cam = reload_shared_camera(tuning_file=str(json_path), preview_max_width=1920)
         except CameraError as err:
             return jsonify({'error': str(err)}), 503
         return jsonify({'target': target, 'tuning': json_path.name, 'kind': kind, 'model': cam.model})
@@ -343,11 +353,15 @@ def api_preview_default():
 
     @app.route('/projects/<name>/preview-capture')
     def preview_capture(name: str):
-        """Download a full-resolution PNG still from the live preview camera."""
+        """Download a PNG snapshot of the live preview (zero shutter lag).
+
+        Grabs the frame currently on screen at the selected mode's preview
+        resolution, so manual exposure/gain is preserved (no mode-switch blip).
+        """
         proj = get_project_or_404(name)
         cam = camera_or_503()
         try:
-            png = cam.capture_png()
+            png = cam.capture_preview_png()
         except CameraError as err:
             abort(503, str(err))
         stamp = datetime.now().strftime('%Y%m%d_%H%M%S')
@@ -577,6 +591,7 @@ def run_stream(name: str):
         proj = get_project_or_404(name)
         targets = [t for t in request.args.get('targets', 'pisp,vc4').split(',') if t]
         mode = request.args.get('mode', 'full')
+        update = request.args.get('update') == '1'
         options = {
             'awb': {'greyworld': request.args.get('greyworld') == '1'},
             'alsc': {
@@ -597,7 +612,7 @@ def run_stream(name: str):
         }
 
         def generate():
-            for line in ctt_runner.run_ctt_stream(proj, targets, mode, options):
+            for line in ctt_runner.run_ctt_stream(proj, targets, mode, options, update=update):
                 yield f'data: {json.dumps(line)}\n\n'
 
         return Response(
@@ -606,6 +621,81 @@ def generate():
             headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
         )
 
+    @app.route('/projects/<name>/run/import-tuning', methods=['POST'])
+    def import_tuning(name: str):
+        """Import a tuning file as this project's tuning for its target.
+
+        The tuning comes from an uploaded file or, with ``system_path``, an
+        installed libcamera tuning. The target is read from the file itself; the
+        file then becomes {name}_{target}.json so a subsequent update run
+        refreshes it in place.
+        """
+        from ctt.core.runner import get_target_from_tuning_file
+        from ctt.utils.errors import ArgError
+
+        proj = get_project_or_404(name)
+        system_path = request.form.get('system_path')
+        if system_path:
+            # Restrict to the installed tuning directories so a request can't read
+            # arbitrary files off the Pi via this endpoint.
+            src = Path(system_path)
+            plat = platform_target()
+            allowed = plat is not None and src.resolve().parent in {d.resolve() for d in _system_tuning_dirs(plat)}
+            if not allowed or not src.is_file():
+                return jsonify({'error': 'Not an installed tuning file'}), 400
+            try:
+                text = src.read_text(encoding='utf-8')
+                json.loads(text)
+            except (OSError, UnicodeDecodeError, json.JSONDecodeError) as err:
+                return jsonify({'error': f'Could not read system tuning: {err}'}), 400
+        else:
+            upload = request.files.get('file')
+            if upload is None or not upload.filename:
+                return jsonify({'error': 'No file uploaded'}), 400
+            try:
+                text = upload.read().decode('utf-8')
+                json.loads(text)
+            except (UnicodeDecodeError, json.JSONDecodeError) as err:
+                return jsonify({'error': f'Invalid JSON: {err}'}), 400
+        # Stage to a temp file so the target can be read before we know the path.
+        proj.output_dir.mkdir(parents=True, exist_ok=True)
+        tmp = proj.output_dir / f'.import-{proj.name}.json'
+        tmp.write_text(text)
+        try:
+            target = get_target_from_tuning_file(str(tmp))
+        except ArgError as err:
+            tmp.unlink(missing_ok=True)
+            return jsonify({'error': str(err).strip()}), 400
+        tmp.replace(proj.output_dir / f'{proj.name}_{target}.json')
+        return jsonify({'target': target})
+
+    @app.route('/projects/<name>/run/system-tunings')
+    def system_tunings(name: str):
+        """List installed libcamera tunings for this Pi's ISP, for the update base.
+
+        Defaults the selection to the file matching the live sensor model (e.g.
+        imx708.json), so an update can be seeded from the camera's stock tuning.
+        """
+        get_project_or_404(name)
+        target = platform_target()
+        if target is None:
+            return jsonify({'error': 'Could not determine the camera ISP platform.'}), 503
+        model = None
+        with contextlib.suppress(CameraError):
+            model = get_shared_camera().model
+        files: list[dict] = []
+        seen: set[str] = set()
+        for d in _system_tuning_dirs(target):
+            if d.is_dir():
+                for f in sorted(d.glob('*.json')):
+                    if f.name not in seen:  # /usr/local entry wins over /usr/share
+                        seen.add(f.name)
+                        files.append({'name': f.name, 'path': str(f)})
+        default = next((f['path'] for f in files if model and f['name'] == f'{model}.json'), None)
+        if default is None and files:
+            default = files[0]['path']
+        return jsonify({'target': target, 'model': model, 'default': default, 'files': files})
+
     # --- results -----------------------------------------------------------
     @app.route('/projects/<name>/results')
     def results_page(name: str):

apps/ctt-server/ctt_server/camera.py

@@ -12,8 +12,6 @@
 import contextlib
 import logging
 import os
-import subprocess
-import sys
 import tempfile
 import threading
 import time
@@ -91,6 +89,7 @@ def __init__(self, preview_max_width: int = 1920, tuning_file: str | None = None
                         'size': size,
                         'bit_depth': int(m.get('bit_depth', 0)),
                         'fps': round(float(m.get('fps', 0.0)), 1),
+                        'unpacked': str(m['unpacked']),
                     }
                     if size not in by_size or entry['bit_depth'] > by_size[size]['bit_depth']:
                         by_size[size] = entry
@@ -128,6 +127,9 @@ def _apply_mode(self, mode: dict) -> None:
         w, h = (int(mode['size'][0]), int(mode['size'][1]))
         self._raw_size = (w, h)
         self._raw_bit_depth = int(mode.get('bit_depth', 0)) or None
+        # Unpacked Bayer format (e.g. 'SRGGB12') so the raw stream is delivered
+        # one sample per 16-bit word rather than MIPI-packed (the _CSI2P default).
+        self._raw_format = mode.get('unpacked')
         self.resolution = self._raw_size
         # Preview derived from the mode's frame, scaled down preserving aspect.
         prev_w = min(self._preview_max_width, w) & ~1
@@ -178,7 +180,7 @@ def _frame_duration_limits(self) -> tuple[int, int]:
     def _video_config(self):
         return self._picam2.create_video_configuration(
             main={'size': self._preview_size, 'format': 'RGB888'},
-            raw={'size': self._raw_size},
+            raw={'size': self._raw_size, 'format': self._raw_format},
             sensor=self._sensor_config(),
             transform=self._transform(),
             buffer_count=4,
@@ -282,6 +284,25 @@ def capture_png(self) -> bytes:
             raise CameraError('Failed to encode PNG')
         return buf.tobytes()
 
+    def capture_preview_png(self) -> bytes:
+        """Snapshot the current live preview frame as a PNG (zero shutter lag).
+
+        Grabs the running main stream straight from the pipeline — no mode switch —
+        so the snapshot is exactly the frame on screen, at the selected mode's preview
+        resolution, with the manual exposure/gain (and the ISP denoise state) left
+        untouched. Switching to a full-resolution still mode (capture_png) reverts
+        auto-exposure and steps the live preview's brightness, which is why the
+        on-screen snapshot avoids it.
+        """
+        import cv2  # noqa: PLC0415
+
+        with self._lock:
+            arr = self._picam2.capture_array('main')  # RGB888 == BGR-ordered == cv2 native
+        ok, buf = cv2.imencode('.png', arr)
+        if not ok:
+            raise CameraError('Failed to encode PNG')
+        return buf.tobytes()
+
     def mjpeg_frames(self, fps: float = 10.0):
         """Yield multipart MJPEG chunks for a streaming HTTP response."""
         period = 1.0 / max(fps, 1.0)
@@ -443,7 +464,7 @@ def capture_burst(self, frames: int, quality: int = 95) -> list[tuple[bytes, byt
         with self._lock:
             still = self._picam2.create_still_configuration(
                 main={'size': self.resolution, 'format': 'RGB888'},
-                raw={'size': self._raw_size},
+                raw={'size': self._raw_size, 'format': self._raw_format},
                 sensor=self._sensor_config(),
                 transform=self._transform(),
             )
@@ -518,56 +539,14 @@ def get_shared_camera() -> Picamera2Camera:
         return _SHARED
 
 
-def validate_tuning_file(path: str, timeout: float = 60.0) -> None:
-    """Test-load a tuning file by bringing the camera up in a throwaway subprocess.
-
-    A tuning the IPA rejects doesn't just fail the open: libcamera drops the
-    camera from the process-global camera manager, which cannot be recreated,
-    leaving this process camera-less until a restart. Hand-edited tunings are
-    therefore brought up in a sacrificial subprocess first, so the main process
-    only ever loads files known to work. The camera must be closed in this
-    process while the canary runs. Raises CameraError when the tuning is bad.
-    """
-    script = (
-        'import os, sys\n'
-        'from picamera2 import Picamera2\n'
-        'path = sys.argv[1]\n'
-        'tuning = Picamera2.load_tuning_file(os.path.basename(path), dir=os.path.dirname(path))\n'
-        'cam = Picamera2(tuning=tuning)\n'
-        'cam.configure(cam.create_video_configuration())\n'
-        'cam.start()\n'
-        'cam.stop()\n'
-        'cam.close()\n'
-    )
-    try:
-        proc = subprocess.run(
-            [sys.executable, '-c', script, str(path)],
-            capture_output=True,
-            text=True,
-            timeout=timeout,
-            check=False,
-        )
-    except subprocess.TimeoutExpired as err:
-        raise CameraError(f'Tuning validation timed out for {os.path.basename(path)}') from err
-    if proc.returncode != 0:
-        # Prefer libcamera's ERROR lines (the actual complaint) over the tail
-        # of the python traceback.
-        lines = [ln for ln in f'{proc.stderr}\n{proc.stdout}'.splitlines() if ln.strip()]
-        errors = [ln for ln in lines if 'ERROR' in ln]
-        detail = (errors or lines)[-1] if (errors or lines) else f'exit code {proc.returncode}'
-        raise CameraError(f'Tuning {os.path.basename(path)} failed to load: {detail}')
-
-
-def reload_shared_camera(
-    tuning_file: str | None = None, preview_max_width: int = 1920, validate: bool = False
-) -> Picamera2Camera:
+def reload_shared_camera(tuning_file: str | None = None, preview_max_width: int = 1920) -> Picamera2Camera:
     """Restart the shared camera with a given tuning file (None = default tuning).
 
     Only one Picamera2 can be open per process, so the current instance is closed
     first. A no-op when the camera already runs the requested tuning, which keeps the
-    capture page's "restore default" call cheap. validate=True canary-tests the
-    tuning in a subprocess first (see validate_tuning_file) — use it for any
-    tuning that isn't known-good, e.g. hand-edited custom files.
+    capture page's "restore default" call cheap. A tuning that fails to load (e.g. a
+    bad hand-edit) is reported as a CameraError and the camera is reopened with the
+    default tuning so it isn't left dead.
     """
     global _SHARED
     with _SHARED_LOCK:
@@ -577,8 +556,6 @@ def reload_shared_camera(
             _SHARED.close()
             _SHARED = None
         try:
-            if validate and tuning_file is not None:
-                validate_tuning_file(tuning_file)
             _SHARED = Picamera2Camera(preview_max_width=preview_max_width, tuning_file=tuning_file)
         except Exception as err:
             # The requested tuning failed to load: reopen with the default