diff --git a/.gitignore b/.gitignore index 70bd403..2db0abb 100644 --- a/.gitignore +++ b/.gitignore @@ -12,6 +12,8 @@ new_outputs/ reference_outputs/ ctt_log.txt *.png +# documentation screenshots are tracked +!docs/images/*.png .venv/ .env/ diff --git a/README.md b/README.md index c7dbb4b..bfec186 100644 --- a/README.md +++ b/README.md @@ -1,310 +1,21 @@ # Raspberry Pi Camera Tuning Tool (CTT) -Camera Tuning Tool for generating JSON tuning files for Raspberry Pi cameras. -Takes DNG calibration images and produces tuning parameters for the PiSP or VC4 -ISP platforms. - -## Installation - -Requires Python 3.11+. - -```bash -pip install rpi-ctt -``` - -For development, install in editable mode from the repository: - -```bash -pip install -e . -``` - -## Usage - -### Full calibration - -```bash -python3 -m ctt -i [-o ] [--name ] [-t pisp] [-t vc4] [-c config.json] -``` - -All outputs (JSON tuning file, log file, Macbeth chart PNGs) are written to the -output directory. If `-o` is not specified, the current directory is used. - -Output filenames are `_.json` and `_.log`. If -`--name` is not specified, the name is derived from the input directory. You -can pass `-t` multiple times or use a comma-separated list (e.g. `-t pisp,vc4`). -If no `-t` is given, both targets are run (e.g. `imx219_pisp.json` and -`imx219_vc4.json`). - -### ALSC-only calibration - -```bash -python3 -m ctt --alsc-only -i [-o ] [-t pisp] [-t vc4] -``` - -### Colour-only calibration (AWB + CCM) - -```bash -python3 -m ctt --colour-only -i [-o ] [-t pisp] [-t vc4] -``` - -### Update an existing tuning file - -Re-run calibrations using an existing file as a template. The target (pisp or vc4) is read from the tuning file; do not use `-t` with `--update`. The file is updated **in place**. - -```bash -python3 -m ctt -i --update - -python3 -m ctt -i --update -o --name imx219 -``` - -With `--alsc-only` or `--colour-only`, only that section is re-calibrated; all other algorithm blocks in the file are left unchanged. - -```bash -python3 -m ctt --alsc-only -i --update -``` - -### Use a custom template - -```bash -python3 -m ctt -i -o -t pisp --template my_base.json -``` - -### Convert between VC4 and PiSP - -Interpolates ALSC grids, swaps denoise/AGC/HDR blocks: - -```bash -python3 -m ctt --convert -t pisp input_vc4.json output_pisp.json -python3 -m ctt --convert -t vc4 input_pisp.json # in-place -``` - -### Prettify a tuning file - -```bash -python3 -m ctt --prettify [-t pisp|vc4] [output.json] -``` - -## Options - -| Flag | Description | -|------|-------------| -| `-i`, `--input` | Calibration image directory | -| `-o`, `--output` | Output directory (default: current directory) | -| `--name` | Base name for output files (default: derived from input directory) | -| `-t`, `--target` | Target platform(s): repeat for multiple (e.g. `-t pisp -t vc4`) or use `-t pisp,vc4`. Default: both. | -| `-c`, `--config` | Configuration file (see below) | -| `--template` | Custom template JSON file | -| `--update` | Existing tuning file to update in place (target taken from file; cannot use `-t`) | -| `--plot` | Show matplotlib debug plot for algorithm (e.g. `awb`, `alsc`, `geq`, `noise`). Can be repeated or comma-separated. | -| `--alsc-only` | Run only ALSC (lens shading) calibration | -| `--colour-only` | Run only AWB and CCM calibrations | -| `--convert` | Convert tuning file between VC4 and PiSP | -| `--prettify` | Prettify an existing tuning file | - -## Configuration file - -An optional JSON config file controls calibration behaviour. See -`apps/ctt/ctt/data/config_example.json` for the full set of options: - -```json -{ - "disable": [], - "plot": [], - "alsc": { - "do_alsc_colour": 1, - "luminance_strength": 0.8, - "max_gain": 8.0 - }, - "awb": { - "greyworld": 0 - }, - "blacklevel": -1, - "macbeth": { - "small": 0, - "show": 0 - } -} -``` - -- **disable** - List of algorithm keys to skip (e.g. `["rpi.noise"]`) -- **plot** - List of algorithm names for matplotlib debug plots. Use short names (`awb`, `alsc`, `geq`, `noise`) or full keys (`rpi.awb`, etc.). Supported: **rpi.awb** (colour curve hatspace + dehatspace), **rpi.alsc** (3D vignetting surfaces), **rpi.geq** (fit and scaled fit), **rpi.noise** (per-image noise fit). Can also be set via `--plot` on the command line. -- **alsc.do_alsc_colour** - Enable colour shading calibration (default: 1) -- **alsc.luminance_strength** - Luminance correction strength, 0.0-1.0 (default: 0.8) -- **alsc.max_gain** - Maximum lens shading gain (default: 8.0) -- **awb.greyworld** - Use grey world AWB instead of Macbeth-based (default: 0) -- **blacklevel** - Override black level; -1 to auto-detect (default: -1) -- **macbeth.small** - Use small Macbeth chart detection (default: 0) -- **macbeth.show** - Display detected Macbeth chart (default: 0) - -## Calibration images - -The tool looks only in the **root** of the input directory and uses **`.dng` files only** (no subdirectories, no JPEGs). - -- **ALSC images**: Filenames containing `alsc` and a colour temperature (e.g. `alsc_3000k_0.dng`). Uniform flat-field images at multiple colour temperatures. -- **Macbeth images**: Filenames must encode colour temperature and lux in the form `...k_l.dng` (e.g. `d65_5858k_1344l.dng`). Images of a Macbeth ColorChecker chart for AWB, CCM, noise, lux, and GEQ. - -If a file is skipped (e.g. missing colour temp/lux in the filename, or Macbeth chart not found in the image), the tool prints a short message (e.g. colour temp/lux not in filename, or Macbeth not found) with the filename. - -## Calibrations performed - -| Algorithm | Key | Description | -|-----------|-----|-------------| -| ALSC | `rpi.alsc` | Lens shading correction (colour and luminance) | -| AWB | `rpi.awb` | Auto white balance calibration | -| CCM | `rpi.ccm` | Colour correction matrices per illuminant | -| CAC | `rpi.cac` | Chromatic aberration correction (PiSP only) | -| Noise | `rpi.noise` | Noise profile characterisation | -| Lux | `rpi.lux` | Lux level calibration | -| GEQ | `rpi.geq` | Green equalisation threshold | - -## Web frontend (ctt-server) - -`ctt-server` is an optional web UI for capturing, tagging and tuning calibration -images, served from the Raspberry Pi. It runs as a single process on the Pi: the -server previews and captures DNGs in-process with Picamera2, files them with -CTT-correct filenames, runs the tuner in-process, and serves downloadable tuning -files with result visualisations. The client is just a web browser on any machine -on the network. - -### Install and run (on the Pi) - -```bash -pip install "rpi-ctt[server]" # from PyPI -# or, from a local checkout: -pip install -e ".[server]" -ctt-server # HTTPS on 0.0.0.0:5000 -``` - -`ctt-server` is **HTTPS-only**; on first run it generates a self-signed -certificate under `/.tls` (pass `--cert`/`--key` to use your own, or -`--port` to change the port). Browse to `https://:5000` and accept -the one-time self-signed warning. Picamera2 ships with Raspberry Pi OS and is -imported lazily; if you use a virtualenv, create it with `--system-site-packages` -(or `apt install python3-picamera2`) so picamera2 is visible. - -### Workflow - -1. **Project** — create one per sensor (e.g. `imx708_wide`); the name becomes the - output filename base (`imx708_wide_pisp.json`). -2. **Capture** — frame with the live preview and histogram, set exposure/gain (or - leave on auto), and capture. In Macbeth mode a live finder overlays the detected - chart and flags low confidence or a too-small chart. Each shot is filed with a - CTT-correct filename. -3. **Run** — pick targets (PiSP/VC4/both) and mode (full / ALSC-only / colour-only); - CTT progress streams live in the console. -4. **Results** — download the tuning `.json`/`.log` or the whole project as a zip, - and inspect the AWB curve, per-CT CCM matrices, ALSC shading heatmap and - lux/noise references parsed from the output JSON. - -When a supported lightbox is attached (see below), the capture page shows a -**Lightbox** control to pick the illuminant and set intensity; it is hidden when no -device is present. - -## Lightbox control - -`ctt.devices` provides a small, generic API for controlling an illumination lightbox -over USB from the Pi, so a calibration run can set repeatable illuminants -programmatically. Consumers use the abstract `Lightbox` interface and the -device-agnostic factory; concrete drivers (currently **Image Engineering -lightSTUDIO-S**) plug in behind it. A new lightbox is added as a driver package under -`apps/ctt/ctt/devices//` registered in `apps/ctt/ctt/devices/registry.py` — no -change to the generic API or its consumers. - -### Install (on the Pi) - -```bash -pip install "rpi-ctt[devices]" # from PyPI (or "rpi-ctt[server,devices]" with the web UI) -# or, from a local checkout: -pip install -e ".[devices]" # or ".[server,devices]" with the web UI -sudo apt install libusb-1.0-0 # pyusb's backend - -# allow non-root USB access (lightSTUDIO-S) -sudo cp apps/ctt/ctt/devices/lightstudio_s/contrib/99-lightstudio.rules /etc/udev/rules.d/ -sudo udevadm control --reload && sudo udevadm trigger -``` - -If the udev rule is missing but pyusb is installed and the device is plugged in, the -probe gets far enough to fail with `usb.core.USBError: [Errno 13] Access denied` -(rather than reporting absent) — install the rule above. If `udevadm trigger` doesn't -re-apply to the already-enumerated device, unplug and replug the lightbox so the rule -runs on re-enumeration. The user running the server must be in the `plugdev` group. - -### Use - -```python -from ctt.devices import get_lightbox - -with get_lightbox() as box: # first attached, supported lightbox - box.set_illuminant('D65') # name → channel, at its default intensity - box.set_intensity(4, 50) # channel 4 (D65) at 50 % - print(box.info()) - box.off() -``` - -```bash -ctt-lightbox probe # find the box, list illuminants -ctt-lightbox status -ctt-lightbox set 1 50 # channel 1 (F12) → 50 % -ctt-lightbox illuminant D65 # switch to D65 at its default -ctt-lightbox off -``` - -### lightSTUDIO-S channels - -| Ch | Illuminant | Default % | | Ch | Illuminant | Default % | -|----|------------|-----------|----|----|------------|-----------| -| 1 | F12 | 100 | | 5 | Halogen (10 lux) | 2 | -| 2 | F11 | 100 | | 6 | Halogen (100 lux) | 25 | -| 3 | D50 | 100 | | 7 | Halogen (400 lux) | 100 | -| 4 | D65 | 100 | | 8 | Halogen + blue filter (400 lux) | 100 | - -## Development - -### Linting and formatting - -The project uses [ruff](https://docs.astral.sh/ruff/) for linting and formatting: - -```bash -# Check for lint errors -python3 -m ruff check . - -# Auto-fix lint errors -python3 -m ruff check --fix . - -# Format code -python3 -m ruff format . - -# Check formatting without modifying files -python3 -m ruff format --check . -``` - -These same checks run in CI. To catch problems before committing, enable the -[pre-commit](https://pre-commit.com/) hooks (they run ruff lint + format on each -commit, mirroring CI): - -```bash -pip install -e ".[dev]" -pre-commit install # one-time, per clone -pre-commit run --all-files # optional: check the whole tree now -``` - -### Running tests - -Install with the test extra and run pytest: - -```bash -pip install -e ".[test]" -pytest -v -``` - -### Building a wheel package - -```bash -pip install build -python3 -m build -``` - -This produces a `.whl` file in the `dist/` directory. +Camera Tuning Tool for generating libcamera JSON tuning files for Raspberry Pi +cameras. Takes DNG calibration images and produces tuning parameters for the PiSP +or VC4 ISP platforms. + +## Documentation + +- [Installation](docs/installation.md) — installing from PyPI or a local + checkout, optional extras, Raspberry Pi notes +- [CTT server](docs/ctt-server.md) — web UI for capturing, tuning and + inspecting results, served from the Pi +- [CTT CLI](docs/ctt-cli.md) — command-line usage, options, configuration and + calibration image requirements +- [Device control](docs/device-control.md) — USB lightbox control (API and + `ctt-lightbox` CLI) +- [Developers](docs/developers.md) — repository layout, linting, tests and + packaging ## License diff --git a/apps/ctt-server/ctt_server/__main__.py b/apps/ctt-server/ctt_server/__main__.py index 28fbbf5..dd32397 100644 --- a/apps/ctt-server/ctt_server/__main__.py +++ b/apps/ctt-server/ctt_server/__main__.py @@ -52,11 +52,9 @@ def _ssl_context(args): if args.cert and args.key: return (args.cert, args.key) - from pathlib import Path + from .sessions import resolve_workspace - from .sessions import default_workspace - - ws = Path(args.workspace) if args.workspace else default_workspace() + ws = resolve_workspace(args.workspace) tls = ws / '.tls' tls.mkdir(parents=True, exist_ok=True) cert, key = tls / 'cert.pem', tls / 'key.pem' diff --git a/apps/ctt-server/ctt_server/app.py b/apps/ctt-server/ctt_server/app.py index c3aed5b..18b2913 100644 --- a/apps/ctt-server/ctt_server/app.py +++ b/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 difflib import json import logging from datetime import datetime @@ -36,7 +37,7 @@ platform_target, reload_shared_camera, ) -from .naming import NamingError, validate_filename +from .naming import NamingError, detect_type, validate_filename from .sessions import Project, Workspace logger = logging.getLogger(__name__) @@ -191,7 +192,7 @@ def lightbox_status() -> dict: # --- pages ------------------------------------------------------------- @app.route('/') def index(): - return redirect(url_for('projects')) + return render_template('index.html') @app.route('/projects') def projects(): @@ -289,24 +290,34 @@ def api_macbeth_deltae(): # --- live preview test (load a generated tuning into the camera) ------- @app.route('/projects//preview-test', methods=['POST']) def preview_test(name: str): - """Restart the camera with this project's generated tuning for live preview.""" + """Restart the camera with this project's generated (or custom) tuning for live preview.""" proj = get_project_or_404(name) + kind = (request.get_json(silent=True) or {}).get('kind', 'generated') target = platform_target() if target is None: return jsonify({'error': 'Could not determine the camera ISP platform.'}), 503 - json_path = proj.output_dir / f'{proj.name}_{target}.json' - if not json_path.exists(): - return jsonify( - { - 'error': f'No {target.upper()} tuning for this project — this Pi runs {target.upper()}. ' - f'Re-run CTT for {target}.' - } - ), 400 + if kind == 'custom': + json_path = proj.output_dir / f'{proj.name}_{target}_custom.json' + if not json_path.exists(): + return jsonify( + {'error': f'No custom tuning edits for {target.upper()} — save edits on the Tuning tab first.'} + ), 400 + else: + json_path = proj.output_dir / f'{proj.name}_{target}.json' + if not json_path.exists(): + return jsonify( + { + 'error': f'No {target.upper()} tuning for this project — this Pi runs {target.upper()}. ' + f'Re-run CTT for {target}.' + } + ), 400 try: - cam = reload_shared_camera(tuning_file=str(json_path), preview_max_width=1920) + # 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')) except CameraError as err: return jsonify({'error': str(err)}), 503 - return jsonify({'target': target, 'tuning': json_path.name, 'model': cam.model}) + return jsonify({'target': target, 'tuning': json_path.name, 'kind': kind, 'model': cam.model}) @app.route('/api/preview-default', methods=['POST']) def api_preview_default(): @@ -362,32 +373,49 @@ def api_set_lightbox(): @app.route('/projects//capture', methods=['POST']) def capture(name: str): + """Capture a still — or a burst of `frames` stills, saved as indexed + files (alsc_5000k_0/1/2..., d65_5000k_800l_0/1/2...) which CTT averages + internally during a run.""" proj = get_project_or_404(name) body = request.get_json(force=True) or {} image_type = body.get('image_type') try: - colour_temp = int(body.get('colour_temp')) + # Dark frames carry no tags; everything else needs a colour temp. + colour_temp = None if image_type == 'dark' else int(body.get('colour_temp')) lux = int(body['lux']) if body.get('lux') not in (None, '') else None label = body.get('label') or None - dng_bytes, jpg_bytes, meta = get_shared_camera().capture_still() - cap = proj.add_capture( - dng_bytes, image_type, colour_temp, lux=lux, label=label, controls=meta, jpeg_bytes=jpg_bytes - ) + frames = max(1, min(int(body.get('frames', 1) or 1), 16)) + shots = get_shared_camera().capture_burst(frames) + caps = [ + proj.add_capture( + dng_bytes, + image_type, + colour_temp, + lux=lux, + label=label, + controls=meta, + jpeg_bytes=jpg_bytes, + indexed=frames > 1, # burst frames get _ names; singles keep overwrite semantics + ) + for dng_bytes, jpg_bytes, meta in shots + ] except CameraError as err: return jsonify({'error': str(err)}), 503 except (TypeError, ValueError) as err: return jsonify({'error': str(err)}), 400 - return jsonify( + added = [ { 'filename': cap.filename, 'image_type': cap.image_type, 'colour_temp': cap.colour_temp, 'lux': cap.lux, 'label': cap.label, + 'valid': True, 'jpeg': 'saved', # a fresh capture always has a full-res JPEG - 'counts': proj.counts(), } - ) + for cap in caps + ] + return jsonify({'added': added, 'counts': proj.counts()}) @app.route('/projects//upload', methods=['POST']) def upload(name: str): @@ -434,6 +462,40 @@ def set_capture_excluded(name: str, filename: str): abort(404, f'No such capture: {filename}') return jsonify({'filename': cap.filename, 'excluded': cap.excluded}) + # Quick black level measurements, cached per (path, mtime) so repeated Run + # tab visits don't re-read multi-megabyte DNGs. + _blacklevel_cache: dict[str, dict] = {} + + @app.route('/projects//blacklevel') + def project_blacklevel(name: str): + """Black level measured from the project's dark frames (seeds the Run tab). + + Uses the same measurement helper as the calibration algorithm, so the + seeded value matches what a full run would compute. + """ + from ctt.algorithms.black_level import measure_dark_dng # noqa: PLC0415 - defer the ctt import + + proj = get_project_or_404(name) + excluded = {c.filename for c in proj.captures if c.excluded} + frames = [] + for p in sorted(proj.path.glob('*.dng')): + if detect_type(p.name) != 'dark' or p.name in excluded: + continue + mtime = p.stat().st_mtime + cached = _blacklevel_cache.get(str(p)) + if cached is None or cached['mtime'] != mtime: + try: + cached = {'mtime': mtime, 'frame': measure_dark_dng(str(p))} + except Exception as err: + logger.warning(f'Black level measurement failed for {p.name}: {err}') + continue + _blacklevel_cache[str(p)] = cached + frames.append(cached['frame']) + if not frames: + return jsonify({'black_level': None, 'frames': []}) + black_level = round(sum(f['black_level'] for f in frames) / len(frames)) + return jsonify({'black_level': black_level, 'frames': frames}) + def _develop_dng(dng: Path, thumb: bool = False) -> Response: """Develop a DNG into a preview JPEG with rawpy (won't match the ISP look). @@ -549,7 +611,10 @@ def preview_page(name: str): proj = get_project_or_404(name) files = ctt_runner.output_files(proj, ['pisp', 'vc4']) available = {t: f for t, f in files.items() if f['json']} - return render_template('preview.html', project=proj, files=available, runs=_run_info(available)) + customs = {t: _custom_tuning_path(proj, t).exists() for t in available} + return render_template( + 'preview.html', project=proj, files=available, runs=_run_info(available), customs=customs + ) # --- MTF (slanted-edge) -------------------------------------------------- # Chart captures live in /mtf/, which CTT runs never see (they only @@ -648,7 +713,7 @@ def download_archive(name: str): @app.route('/projects//download//') def download(name: str, kind: str, target: str): proj = get_project_or_404(name) - suffix = {'json': '.json', 'log': '.log'}.get(kind) + suffix = {'json': '.json', 'log': '.log', 'custom': '_custom.json'}.get(kind) if suffix is None: abort(404) path = proj.output_dir / f'{proj.name}_{target}{suffix}' @@ -656,4 +721,61 @@ def download(name: str, kind: str, target: str): abort(404) return send_file(path, as_attachment=True, download_name=path.name) + # --- custom (hand-edited) tuning files ---------------------------------- + def _custom_tuning_path(proj: Project, target: str) -> Path: + return proj.output_dir / f'{proj.name}_{target}_custom.json' + + def _tuning_state(proj: Project, target: str) -> dict: + """Original + custom tuning text for a target, with a unified diff.""" + if target not in ctt_runner.VALID_TARGETS: + abort(404) + json_path = proj.output_dir / f'{proj.name}_{target}.json' + if not json_path.exists(): + abort(404, 'No tuning file for that target') + original = json_path.read_text() + custom_path = _custom_tuning_path(proj, target) + custom = custom_path.read_text() if custom_path.exists() else None + diff = None + if custom is not None: + diff = '\n'.join( + difflib.unified_diff( + original.splitlines(), + custom.splitlines(), + fromfile=json_path.name, + tofile=custom_path.name, + lineterm='', + ) + ) + return {'original': original, 'custom': custom, 'diff': diff} + + @app.route('/projects//tuning-data/') + def tuning_data(name: str, target: str): + """The Tuning tab's contents: original + custom text and their diff.""" + proj = get_project_or_404(name) + return jsonify(_tuning_state(proj, target)) + + @app.route('/projects//tuning/custom/', methods=['POST']) + def save_custom_tuning(name: str, target: str): + """Save hand edits as the per-target custom tuning file (validated JSON).""" + proj = get_project_or_404(name) + if target not in ctt_runner.VALID_TARGETS: + abort(404) + text = (request.get_json(force=True) or {}).get('json', '') + try: + json.loads(text) + except json.JSONDecodeError as err: + return jsonify({'error': f'Invalid JSON: {err}'}), 400 + proj.output_dir.mkdir(parents=True, exist_ok=True) + _custom_tuning_path(proj, target).write_text(text) + return jsonify(_tuning_state(proj, target)) + + @app.route('/projects//tuning/custom//delete', methods=['POST']) + def delete_custom_tuning(name: str, target: str): + """Revert to the original tuning: remove the custom file.""" + proj = get_project_or_404(name) + if target not in ctt_runner.VALID_TARGETS: + abort(404) + _custom_tuning_path(proj, target).unlink(missing_ok=True) + return jsonify(_tuning_state(proj, target)) + return app diff --git a/apps/ctt-server/ctt_server/camera.py b/apps/ctt-server/ctt_server/camera.py index 1aa9442..5ec369d 100644 --- a/apps/ctt-server/ctt_server/camera.py +++ b/apps/ctt-server/ctt_server/camera.py @@ -9,8 +9,11 @@ # imported lazily so the package can still be imported (and the tuner run) on # machines without a camera, e.g. a desktop for development. +import contextlib import logging import os +import subprocess +import sys import tempfile import threading import time @@ -64,6 +67,7 @@ def __init__(self, preview_max_width: int = 1920, tuning_file: str | None = None self._lock = threading.Lock() self._ev = 0.0 # exposure compensation (EV); tracked here as metadata may omit it self._auto = True # AeEnable state; tracked so we report it reliably (AeLocked is ambiguous) + self._fps = 30.0 # framerate target; 0 = unconstrained (variable frame duration) # Optionally start the pipeline with a specific tuning file (e.g. a freshly # generated CTT tuning, for the Results-page live preview test). None = the # camera's built-in default tuning. @@ -73,30 +77,38 @@ def __init__(self, preview_max_width: int = 1920, tuning_file: str | None = None self._picam2 = Picamera2(tuning=tuning) else: self._picam2 = Picamera2() - self.model = self._picam2.camera_properties.get('Model', 'unknown') - # Full-resolution raw stream (full sensor field of view) → full-res DNGs. - # Pick the largest available raw sensor mode (the full readout), rather than - # sensor_resolution, since the pixel-array size isn't always a usable mode. - # Reading sensor_modes probes every mode, so cache the result per model and - # reuse it when the camera is reopened (e.g. a tuning reload). - raw_size = _RAW_SIZE_CACHE.get(self.model) - if raw_size is None: - largest = max(self._picam2.sensor_modes, key=lambda m: m['size'][0] * m['size'][1])['size'] - raw_size = (int(largest[0]), int(largest[1])) - _RAW_SIZE_CACHE[self.model] = raw_size - sensor_w, sensor_h = raw_size - self.resolution = (sensor_w, sensor_h) - self._raw_size = raw_size - # Preview derived from the full-res frame, scaled down preserving aspect. - prev_w = min(preview_max_width, sensor_w) & ~1 - prev_h = int(round(prev_w * sensor_h / sensor_w)) & ~1 - self._preview_size = (prev_w, prev_h) - # Sensor flip (applied at configure time via a libcamera Transform); affects - # both the live preview and the captured raw/DNG/PNG. - self._hflip = False - self._vflip = False - self._picam2.configure(self._video_config()) - self._picam2.start() + try: + self.model = self._picam2.camera_properties.get('Model', 'unknown') + # Full-resolution raw stream (full sensor field of view) → full-res DNGs. + # Pick the largest available raw sensor mode (the full readout), rather than + # sensor_resolution, since the pixel-array size isn't always a usable mode. + # Reading sensor_modes probes every mode, so cache the result per model and + # reuse it when the camera is reopened (e.g. a tuning reload). + raw_size = _RAW_SIZE_CACHE.get(self.model) + if raw_size is None: + largest = max(self._picam2.sensor_modes, key=lambda m: m['size'][0] * m['size'][1])['size'] + raw_size = (int(largest[0]), int(largest[1])) + _RAW_SIZE_CACHE[self.model] = raw_size + sensor_w, sensor_h = raw_size + self.resolution = (sensor_w, sensor_h) + self._raw_size = raw_size + # Preview derived from the full-res frame, scaled down preserving aspect. + prev_w = min(preview_max_width, sensor_w) & ~1 + prev_h = int(round(prev_w * sensor_h / sensor_w)) & ~1 + self._preview_size = (prev_w, prev_h) + # Sensor flip (applied at configure time via a libcamera Transform); affects + # both the live preview and the captured raw/DNG/PNG. + self._hflip = False + self._vflip = False + self._picam2.configure(self._video_config()) + self._picam2.start() + except Exception: + # A failed bring-up (e.g. a bad custom tuning rejected by libcamera) + # must release the device, or the process holds the camera acquired + # and every reopen fails until a restart. + with contextlib.suppress(Exception): + self._picam2.close() + raise # Allow auto-exposure to settle before the first frame. time.sleep(0.5) @@ -105,12 +117,28 @@ def _transform(self): return Transform(hflip=1 if self._hflip else 0, vflip=1 if self._vflip else 0) + def _frame_duration_limits(self) -> tuple[int, int]: + """FrameDurationLimits for the current fps target (0 = unconstrained). + + A fixed rate pins both limits (which also caps exposure at one frame, + and is clamped by libcamera to the control's advertised range); 0 spans + the camera's full FrameDurationLimits range so long exposures can + stretch the frame, instead of create_video_configuration's locked-30fps + default clamping them at ~33 ms. + """ + if self._fps: + frame_duration = int(round(1_000_000 / self._fps)) + return (frame_duration, frame_duration) + fd_min, fd_max, _ = self._picam2.camera_controls['FrameDurationLimits'] + return (int(fd_min), int(fd_max)) + def _video_config(self): return self._picam2.create_video_configuration( main={'size': self._preview_size, 'format': 'RGB888'}, raw={'size': self._raw_size}, transform=self._transform(), buffer_count=4, + controls={'FrameDurationLimits': self._frame_duration_limits()}, ) def set_transform(self, hflip: bool, vflip: bool) -> dict: @@ -126,6 +154,13 @@ def set_transform(self, hflip: bool, vflip: bool) -> dict: # --- controls ---------------------------------------------------------- def get_controls(self) -> dict: md = self._picam2.capture_metadata() + frame_duration = int(md.get('FrameDuration', 0)) + # The sensor's advertised control ranges (mode-dependent) are the + # absolute truth for the UI sliders. + ctrl = self._picam2.camera_controls + exp_min, exp_max, _ = ctrl['ExposureTime'] + gain_min, gain_max, _ = ctrl['AnalogueGain'] + fd_min, fd_max, _ = ctrl['FrameDurationLimits'] return { 'exposure': int(md.get('ExposureTime', 0)), 'gain': round(float(md.get('AnalogueGain', 0.0)), 3), @@ -134,6 +169,14 @@ def get_controls(self) -> dict: 'focus_fom': int(md.get('FocusFoM', 0)), # focus figure of merit (higher = sharper) 'ev': round(self._ev, 2), 'auto_exposure': self._auto, + 'fps': self._fps, # user target; 0 = unconstrained + 'measured_fps': round(1_000_000 / frame_duration, 1) if frame_duration else 0, + 'exposure_min': int(exp_min), + 'exposure_max': int(exp_max), + 'gain_min': round(float(gain_min), 3), + 'gain_max': round(float(gain_max), 3), + 'frame_duration_min': int(fd_min), + 'frame_duration_max': int(fd_max), } def set_controls(self, controls: dict) -> dict: @@ -152,6 +195,9 @@ def set_controls(self, controls: dict) -> dict: if 'ev' in controls and controls['ev'] is not None: self._ev = float(controls['ev']) new['ExposureValue'] = self._ev # AEC bias; only affects auto-exposure + if controls.get('fps') is not None: + self._fps = max(float(controls['fps']), 0.0) + new['FrameDurationLimits'] = self._frame_duration_limits() if 'awb' in controls: new['AwbEnable'] = bool(controls['awb']) if new: @@ -331,17 +377,24 @@ def capture_dng(self) -> tuple[bytes, dict]: return data, meta def capture_still(self, quality: int = 95) -> tuple[bytes, bytes, dict]: - """Capture one full-resolution frame as BOTH a DNG and a JPEG. + """Capture one full-resolution frame as BOTH a DNG and a JPEG.""" + return self.capture_burst(1, quality=quality)[0] + + def capture_burst(self, frames: int, quality: int = 95) -> list[tuple[bytes, bytes, dict]]: + """Capture a burst of full-resolution frames, each as a DNG + JPEG. The running video config has a full-resolution raw stream but only a - preview-size main stream, so a full-res JPEG needs a still mode. We switch - to a full-sensor still configuration (full main + full raw), grab a single - request, and produce the DNG from the raw stream and the JPEG from the main - stream of that same frame -- so the two match exactly -- then switch back to - the preview config. This costs a brief preview blip, like capture_png(). + preview-size main stream, so full-res JPEGs need a still mode. We switch + to a full-sensor still configuration (full main + full raw) once, grab + the requests back to back, and produce each frame's DNG from its raw + stream and JPEG from its main stream -- so the pairs match exactly -- + then switch back to the preview config. This costs a brief preview + blip, like capture_png(). """ import cv2 # noqa: PLC0415 + frames = max(1, min(int(frames), 16)) + shots = [] with self._lock: still = self._picam2.create_still_configuration( main={'size': self.resolution, 'format': 'RGB888'}, @@ -349,27 +402,34 @@ def capture_still(self, quality: int = 95) -> tuple[bytes, bytes, dict]: transform=self._transform(), ) self._picam2.switch_mode(still) - request = self._picam2.capture_request() try: - with tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / 'capture.dng' - request.save_dng(str(path)) - dng = path.read_bytes() - arr = request.make_array('main') # full-res RGB888 (BGR-ordered = cv2 native) - metadata = request.get_metadata() + for _ in range(frames): + request = self._picam2.capture_request() + try: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / 'capture.dng' + request.save_dng(str(path)) + dng = path.read_bytes() + arr = request.make_array('main') # full-res RGB888 (BGR-ordered = cv2 native) + metadata = request.get_metadata() + finally: + request.release() + shots.append((dng, arr, metadata)) finally: - request.release() self._picam2.switch_mode(self._video_config()) - ok, buf = cv2.imencode('.jpg', arr, [cv2.IMWRITE_JPEG_QUALITY, quality]) - if not ok: - raise CameraError('Failed to encode capture JPEG') - meta = { - 'exposure': int(metadata.get('ExposureTime', 0)), - 'gain': round(float(metadata.get('AnalogueGain', 0.0)), 3), - 'colour_temp': int(metadata.get('ColourTemperature', 0)), - 'lux': round(float(metadata.get('Lux', 0.0)), 1), - } - return dng, buf.tobytes(), meta + out = [] + for dng, arr, metadata in shots: + ok, buf = cv2.imencode('.jpg', arr, [cv2.IMWRITE_JPEG_QUALITY, quality]) + if not ok: + raise CameraError('Failed to encode capture JPEG') + meta = { + 'exposure': int(metadata.get('ExposureTime', 0)), + 'gain': round(float(metadata.get('AnalogueGain', 0.0)), 3), + 'colour_temp': int(metadata.get('ColourTemperature', 0)), + 'lux': round(float(metadata.get('Lux', 0.0)), 1), + } + out.append((dng, buf.tobytes(), meta)) + return out def health(self) -> dict: return { @@ -400,16 +460,65 @@ def get_shared_camera() -> Picamera2Camera: global _SHARED with _SHARED_LOCK: if _SHARED is None: - _SHARED = Picamera2Camera() + try: + _SHARED = Picamera2Camera() + except CameraError: + raise + except Exception as err: + raise CameraError(f'Failed to start the camera: {err}') from err return _SHARED -def reload_shared_camera(tuning_file: str | None = None, preview_max_width: int = 1920) -> Picamera2Camera: +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: """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. + 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. """ global _SHARED with _SHARED_LOCK: @@ -418,7 +527,20 @@ def reload_shared_camera(tuning_file: str | None = None, preview_max_width: int if _SHARED is not None: _SHARED.close() _SHARED = None - _SHARED = Picamera2Camera(preview_max_width=preview_max_width, tuning_file=tuning_file) + 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 + # tuning so the camera isn't left dead, then report. + if tuning_file is not None: + with contextlib.suppress(Exception): + _SHARED = Picamera2Camera(preview_max_width=preview_max_width) + if isinstance(err, CameraError): + raise + tuning_name = os.path.basename(tuning_file) if tuning_file else 'the default tuning' + raise CameraError(f'Failed to start the camera with {tuning_name}: {err}') from err return _SHARED diff --git a/apps/ctt-server/ctt_server/ctt_runner.py b/apps/ctt-server/ctt_server/ctt_runner.py index 635ac1c..46e27d3 100644 --- a/apps/ctt-server/ctt_server/ctt_runner.py +++ b/apps/ctt-server/ctt_server/ctt_runner.py @@ -21,7 +21,7 @@ from .sessions import Project VALID_TARGETS = ('pisp', 'vc4') -VALID_MODES = ('full', 'alsc-only', 'colour-only') +VALID_MODES = ('full', 'alsc-only', 'colour-only', 'blacklevel-only') # Attaching a handler to the process-global `ctt` logger is not safe to do # concurrently, and CTT is not designed for concurrent runs, so serialise. @@ -119,6 +119,7 @@ def run_ctt_stream( config_path = write_config(project, options or {}) alsc_only = mode == 'alsc-only' colour_only = mode == 'colour-only' + black_level_only = mode == 'blacklevel-only' # Honour per-capture exclusion: pass the include-list only when something # is excluded, so the default run still picks up stray on-disk DNGs that # aren't registered in project.json (exactly as before). @@ -150,6 +151,7 @@ def worker() -> None: targets, alsc_only=alsc_only, colour_only=colour_only, + black_level_only=black_level_only, images=images, ) except ArgError as err: @@ -177,6 +179,14 @@ def worker() -> None: break yield item thread.join() + # A successful run regenerates the originals, so hand edits no longer + # apply: discard each run target's custom tuning file. + if result.get('code', 1) == 0: + for t in targets: + custom = project.output_dir / f'{project.name}_{t}_custom.json' + if custom.exists(): + custom.unlink() + yield f'Discarded custom tuning edits for {t} (original regenerated)' yield f'CTT_EXIT {result.get("code", 1)}' finally: _run_lock.release() diff --git a/apps/ctt-server/ctt_server/naming.py b/apps/ctt-server/ctt_server/naming.py index 200e68f..58f6ed4 100644 --- a/apps/ctt-server/ctt_server/naming.py +++ b/apps/ctt-server/ctt_server/naming.py @@ -9,7 +9,9 @@ # # ALSC flat-field: alsc_k_.dng e.g. alsc_5000k_0.dng # Macbeth chart: