From 68753b71ec6cc5d0a1158f08fa040c7f539668e3 Mon Sep 17 00:00:00 2001 From: Naushir Patuck Date: Thu, 1 Oct 2026 14:06:41 +0100 Subject: [PATCH 1/3] ctt: break the Macbeth-search reference cycle that pinned image memory A failed chart search returned its exception, whose traceback reached every calling frame and was kept alive in a cycle through the caller's message local. Large image arrays and the camera itself then survived until the cyclic collector ran, which rarely happens, so returning the message text instead cuts a single target's peak from 12.7 GB to 6.7 GB for both targets. --- ctt/detection/macbeth.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/ctt/detection/macbeth.py b/ctt/detection/macbeth.py index 5f76e2c..6f4f676 100644 --- a/ctt/detection/macbeth.py +++ b/ctt/detection/macbeth.py @@ -413,4 +413,8 @@ def get_macbeth_chart(img: np.ndarray, ref_data: tuple) -> tuple: return (max_cor, best_map_col_norm, fit_coords, success_msg) except MacbethError as error: - return (0, None, None, error) + # Return the text, not the exception: its traceback references this frame and, + # through it, every caller up to the run itself. The caller keeps the message in + # a local, which would close that into a reference cycle and pin the (large) + # image arrays in those frames until the cyclic garbage collector happens to run. + return (0, None, None, str(error)) From 0476c776c9b1d25d5b9ff7278b9e39920a0116b3 Mon Sep 17 00:00:00 2001 From: Naushir Patuck Date: Thu, 1 Oct 2026 14:06:41 +0100 Subject: [PATCH 2/3] ctt: reduce calibration frames at load to the statistics they need ALSC frames now keep only their grid cell means, dark frames their channel statistics, and Macbeth bursts their exact uint32 sums, from which lux rebuilds the identical float64 average. Results are bit-identical, and a 50 MP run with both targets now peaks at 2.8 GB rather than 6.7 GB. --- ctt/algorithms/alsc.py | 50 ++++++++++++++++++++++++++--------- ctt/algorithms/black_level.py | 18 +++++++++++-- ctt/algorithms/lux.py | 2 +- ctt/core/camera.py | 21 ++++++++++++++- ctt/core/image.py | 12 +++++++++ ctt/core/image_loader.py | 17 ++++++++---- ctt/core/runner.py | 2 +- docs/ctt-cli.md | 7 +++++ tests/test_alsc.py | 41 ++++++++++++++++++++++++++-- tests/test_black_level.py | 9 ++++++- tests/test_burst.py | 9 ++++--- 11 files changed, 160 insertions(+), 28 deletions(-) diff --git a/ctt/algorithms/alsc.py b/ctt/algorithms/alsc.py index ebe576d..1c1c095 100644 --- a/ctt/algorithms/alsc.py +++ b/ctt/algorithms/alsc.py @@ -259,21 +259,22 @@ def alsc( """Calculate g/r and g/b for grid points for a single image.""" cam.log += f'\nProcessing image: {img.name}' grid_w, grid_h = grid_size - # Get channels in correct order. - channels = [img.channels[i] for i in img.order] - # Calculate size of single rectangle; divisions ensure final row/column of cells has non-zero pixels. - w, h = img.w / 2, img.h / 2 - dx, dy = int((w - 1) // (grid_w - 1)), int((h - 1) // (grid_h - 1)) + w, h, dx, dy = alsc_cell_size(img, grid_size) + if img.alsc_grids is not None: + try: + g_grid, r_grid, b_grid = img.alsc_grids[tuple(grid_size)] + except KeyError: + raise ValueError(f'{img.name}: ALSC cell means were not computed for a {grid_size} grid') from None + else: + g_grid, r_grid, b_grid = alsc_cell_means(img, grid_size) - # Average the green channels into one. - av_ch_g = np.mean((channels[1:3]), axis=0) if do_alsc_colour: - # Obtain grid_w x grid_h grid of intensities for each channel and subtract black level. + # Subtract black level from the grid_w x grid_h cell intensities of each channel. # Floor at 1 so a dark/vignetted cell at or below black level can't make a ratio # divide by zero (inf/nan) or go negative (which would flip the min-normalisation). - g = np.maximum(get_grid(av_ch_g, dx, dy, grid_size) - img.blacklevel_16, 1) - r = np.maximum(get_grid(channels[0], dx, dy, grid_size) - img.blacklevel_16, 1) - b = np.maximum(get_grid(channels[3], dx, dy, grid_size) - img.blacklevel_16, 1) + g = np.maximum(g_grid - img.blacklevel_16, 1) + r = np.maximum(r_grid - img.blacklevel_16, 1) + b = np.maximum(b_grid - img.blacklevel_16, 1) # Calculate ratios as 32-bit for medianBlur; then median blur to remove peaks. cr = np.reshape(g / r, (grid_h, grid_w)).astype('float32') cb = np.reshape(g / b, (grid_h, grid_w)).astype('float32') @@ -290,7 +291,7 @@ def alsc( else: # Only perform calculations for luminance shading. Floor at 1 (see colour branch). - g = np.maximum(get_grid(av_ch_g, dx, dy, grid_size) - img.blacklevel_16, 1) + g = np.maximum(g_grid - img.blacklevel_16, 1) cg = np.reshape(1 / g, (grid_h, grid_w)).astype('float32') cg = cv2.medianBlur(cg, 3).astype('float64') cg = cg / np.min(cg) @@ -299,6 +300,31 @@ def alsc( return img.col, None, None, cg_clamp, (w, h, dx, dy) +def alsc_cell_size(img: Image, grid_size: tuple[int, int]) -> tuple: + """Channel size and grid cell size; divisions ensure the final row/column of cells has pixels.""" + grid_w, grid_h = grid_size + w, h = img.w / 2, img.h / 2 + return w, h, int((w - 1) // (grid_w - 1)), int((h - 1) // (grid_h - 1)) + + +def alsc_cell_means(img: Image, grid_size: tuple[int, int]) -> tuple: + """Green (Gr/Gb averaged), red and blue cell means of an ALSC image, before black level. + + These are all that the ALSC calibration reads from an image's channels, so the + loader can keep them in place of the full-resolution channels (Image.alsc_grids). + """ + _, _, dx, dy = alsc_cell_size(img, grid_size) + # Get channels in correct order. + channels = [img.channels[i] for i in img.order] + # Average the green channels into one. + av_ch_g = np.mean((channels[1:3]), axis=0) + return ( + get_grid(av_ch_g, dx, dy, grid_size), + get_grid(channels[0], dx, dy, grid_size), + get_grid(channels[3], dx, dy, grid_size), + ) + + def get_grid(chan: np.ndarray, dx: int, dy: int, grid_size: tuple[int, int]) -> np.ndarray: """Compress channel down to a grid of the requested size.""" grid_w, grid_h = grid_size diff --git a/ctt/algorithms/black_level.py b/ctt/algorithms/black_level.py index 8a0361c..ca90209 100644 --- a/ctt/algorithms/black_level.py +++ b/ctt/algorithms/black_level.py @@ -35,6 +35,18 @@ _METADATA_DELTA_LIMIT = 0.01 * (2**16) +def reduce_dark_image(img: Image, drop_channels: bool = True) -> None: + """Keep the per-channel mean and std of a dark frame, all that is measured from it. + + With drop_channels the full-resolution channels are then released, so a run can + hold its dark frames without holding their pixels. + """ + img.channel_means = [float(np.mean(ch)) for ch in img.channels] + img.channel_stds = [float(np.std(ch)) for ch in img.channels] + if drop_channels: + img.channels = [] + + def measure_dark_image(img: Image) -> dict: """Per-channel black level means of a loaded dark frame, 16-bit scaled. @@ -42,8 +54,10 @@ def measure_dark_image(img: Image) -> dict: mapping established in image_loader.dng_load_image. Mono sensors (pattern 128) report a single 'y' value instead of 'r'/'g'/'b'. """ - means = [float(np.mean(img.channels[i])) for i in img.order] - stds = [float(np.std(img.channels[i])) for i in img.order] + if img.channel_means is None: + reduce_dark_image(img, drop_channels=False) + means = [img.channel_means[i] for i in img.order] + stds = [img.channel_stds[i] for i in img.order] out = { 'name': img.name, 'black_level': float(np.mean(means)), diff --git a/ctt/algorithms/lux.py b/ctt/algorithms/lux.py index 47631f4..623c0ce 100644 --- a/ctt/algorithms/lux.py +++ b/ctt/algorithms/lux.py @@ -51,7 +51,7 @@ def run(self) -> dict | None: # in the metrics so the Results page can plot it. samples = [] for img in cam.imgs: - y = lux_calc(cam, img, [img.patches[i] for i in img.order], [img.channels[i] for i in img.order]) + y = lux_calc(cam, img, [img.patches[i] for i in img.order], [img.channel_values(i) for i in img.order]) slope = y / (img.lux * img.exposure * img.againQ8_norm) samples.append({'name': img.name, 'ct': int(img.col), 'lux': int(img.lux), 'y': y, 'slope': slope}) diff --git a/ctt/core/camera.py b/ctt/core/camera.py index 3c9632e..8765f77 100644 --- a/ctt/core/camera.py +++ b/ctt/core/camera.py @@ -9,6 +9,8 @@ import time from pathlib import Path +from ..algorithms.alsc import alsc_cell_means +from ..algorithms.black_level import reduce_dark_image from ..output.json_formatter import pretty_print from ..utils.tools import get_photos from .image_loader import load_image, load_image_group @@ -124,8 +126,20 @@ def write_log(self, filename: str | None) -> None: logfile.write(str(self.log)) def add_imgs( - self, directory: str, mac_config: tuple, blacklevel: int = -1, images: list[str] | None = None + self, + directory: str, + mac_config: tuple, + blacklevel: int = -1, + images: list[str] | None = None, + alsc_grid_size: tuple[int, int] | None = None, ) -> None: + """Load and classify the calibration images in directory. + + Frames are reduced at load to what their calibrations read, as a run cannot + hold every full-resolution frame of a high-resolution sensor: dark frames keep + their channel statistics, Macbeth bursts their exact sums (load_image_group), + and, when alsc_grid_size is given, ALSC frames their cell means on that grid. + """ self.log_new_sec('Image Loading', cal=False) logger.info(f'\nLoading images from {directory}') self.log += f'\nDirectory: {directory}' @@ -164,6 +178,10 @@ def add_imgs( if col is not None: img.col = col img.name = filename + if alsc_grid_size is not None: + grid = tuple(alsc_grid_size) + img.alsc_grids = {grid: alsc_cell_means(img, grid)} + img.channels = [] self.log += f'\nColour temperature: {col} K' self.imgs_alsc.append(img) if blacklevel != -1: @@ -189,6 +207,7 @@ def add_imgs( # Dark frames need no chart detection or demosaic; only the raw # channel statistics are consumed (black level measurement). img = load_image(self, address, mac=False, demosaic=False) + reduce_dark_image(img) self.log += '\nIdentified as a dark frame' img.name = filename self.imgs_dark.append(img) diff --git a/ctt/core/image.py b/ctt/core/image.py index 3df775f..75cfdbc 100644 --- a/ctt/core/image.py +++ b/ctt/core/image.py @@ -36,8 +36,20 @@ class Image: patch_size: int | None = None frames_averaged: int = 1 # burst frames averaged into this image (in-CTT, by filename group) patches_single: list | None = None # patches from one burst frame (true noise statistics) + # Reductions held in place of the full-resolution channels once those are dropped + # at load (a frame's channels are ~100 MB at 50 MP, so a run cannot keep them all). + channel_sums: list | None = None # exact integer burst sums; see channel_values() + channel_means: list | None = None # per-channel mean, storage order (dark frames) + channel_stds: list | None = None # per-channel std, storage order (dark frames) + alsc_grids: dict | None = None # grid_size -> (g, r, b) cell means, before black level ver: int = 0 + def channel_values(self, i: int) -> np.ndarray: + """Channel i in storage order, rebuilding a burst average from its exact sums.""" + if self.channel_sums is not None: + return self.channel_sums[i] / self.frames_averaged + return self.channels[i] + def get_patches(self, cen_coords: list, size: int | None = None) -> int: cen_coords = list(np.array(cen_coords[0]).astype(np.int32)) self.cen_coords = cen_coords diff --git a/ctt/core/image_loader.py b/ctt/core/image_loader.py index 28eef88..02b10f6 100644 --- a/ctt/core/image_loader.py +++ b/ctt/core/image_loader.py @@ -251,20 +251,27 @@ def load_image_group( return load_image(cam, im_strs[0], mac_config, demosaic=demosaic) # Average the frames one at a time: holding a whole burst in memory at once - # is a sizeable chunk of a Pi's RAM. float64 sums of uint16 data are exact, - # so this matches np.mean over the stacked frames bit for bit. + # is a sizeable chunk of a Pi's RAM. The uint16 frames are summed exactly in + # uint32 (enough for 65537 frames), and float64 division of the exact sums + # matches np.mean over the stacked frames bit for bit. base = dng_load_image(cam, im_strs[0], demosaic=demosaic) single_channels = base.channels - sums = [ch.astype(np.float64) for ch in single_channels] + sums = [ch.astype(np.uint32) for ch in single_channels] for im_str in im_strs[1:]: img = dng_load_image(cam, im_str, demosaic=False) for i, ch in enumerate(img.channels): sums[i] += ch - base.channels = [s / len(im_strs) for s in sums] + del img + base.channel_sums = sums base.frames_averaged = len(im_strs) + base.channels = [base.channel_values(i) for i in range(len(sums))] cam.log += f'\nAveraged {len(im_strs)} burst frames' - if not _detect_macbeth(cam, base, mac_config, base.name): + detected = _detect_macbeth(cam, base, mac_config, base.name) + # Chart detection and patch sampling were the last full-resolution uses of the + # float64 average; later consumers rebuild it from the sums via channel_values(). + base.channels = [] + if not detected: return None # Patches of one un-averaged frame, sampled at the same chart coordinates diff --git a/ctt/core/runner.py b/ctt/core/runner.py index 0218ad8..56d21db 100644 --- a/ctt/core/runner.py +++ b/ctt/core/runner.py @@ -212,7 +212,7 @@ def _mode_label() -> str: cam = Camera(json_output, json=json_template) cam.output_dir = output_dir cam.log_user_input(json_output, directory, config, log_output) - cam.add_imgs(directory, mac_config, blacklevel, images=images) + cam.add_imgs(directory, mac_config, blacklevel, images=images, alsc_grid_size=grid_size) # Infer ALSC-only when only ALSC images present (e.g. mono LSC-only from DNGs), # and black-level-only when the directory holds nothing but dark frames. if len(cam.imgs) == 0 and len(cam.imgs_cac) == 0 and len(cam.imgs_alsc) > 0: diff --git a/docs/ctt-cli.md b/docs/ctt-cli.md index 6d1f328..b62ebd6 100644 --- a/docs/ctt-cli.md +++ b/docs/ctt-cli.md @@ -160,6 +160,13 @@ 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. +Images are reduced at load to the statistics their calibrations use, so memory +use scales mainly with the number of Macbeth images (or burst groups) rather +than the total file count. ALSC frames keep only their grid cell means and dark +frames their channel statistics. A Macbeth image holds roughly 2 bytes per +pixel, or 4 per pixel for a burst group. As a guide, a 50 MP sensor with 8 +Macbeth burst groups, 20 ALSC frames and 5 dark frames peaks at about 3 GB. + ## Calibrations performed | Algorithm | Key | Description | diff --git a/tests/test_alsc.py b/tests/test_alsc.py index e44321c..6d9713a 100644 --- a/tests/test_alsc.py +++ b/tests/test_alsc.py @@ -2,11 +2,14 @@ # # Copyright (C) 2026, Raspberry Pi # -# Tests for ALSC post-correction residual prediction. +# Tests for ALSC table calculation and post-correction residual prediction. import numpy as np +import pytest -from ctt.algorithms.alsc import alsc_residuals, get_grid +from ctt.algorithms.alsc import alsc, alsc_cell_means, alsc_residuals, get_grid +from ctt.core.camera import Camera +from ctt.core.image import Image GRID = (16, 12) @@ -23,6 +26,40 @@ def test_get_grid_uint16_matches_float_reference(): np.testing.assert_allclose(out, ref) +def _alsc_image(reduced: bool) -> Image: + """A synthetic uint16 flat-field, optionally reduced to its cell means as at load.""" + rng = np.random.default_rng(3) + img = Image() + img.name = 'alsc_5000k_0.dng' + img.col = 5000 + img.w, img.h = 2 * 330, 2 * 250 + img.order = (2, 0, 3, 1) + img.blacklevel_16 = 4096 + img.channels = [rng.integers(8000, 60000, (250, 330), dtype=np.uint16) for _ in range(4)] + if reduced: + img.alsc_grids = {GRID: alsc_cell_means(img, GRID)} + img.channels = [] + return img + + +@pytest.mark.parametrize('colour', [True, False]) +def test_reduced_image_gives_identical_tables(colour): + # The loader keeps only the cell means of an ALSC frame; the tables calculated + # from them must match the full-channel calculation bit for bit. + full = alsc(Camera('out.json', json={}), _alsc_image(reduced=False), colour, GRID) + reduced = alsc(Camera('out.json', json={}), _alsc_image(reduced=True), colour, GRID) + for a, b in zip(full, reduced, strict=True): + if a is None: + assert b is None + else: + np.testing.assert_array_equal(np.asarray(a), np.asarray(b)) + + +def test_reduced_image_rejects_other_grid(): + with pytest.raises(ValueError, match='not computed'): + alsc(Camera('out.json', json={}), _alsc_image(reduced=True), True, (32, 32)) + + def _vignetted(corner_level: float) -> np.ndarray: """A radial flat-field: 1.0 at the exact grid centre, corner_level at the corners.""" grid_w, grid_h = GRID diff --git a/tests/test_black_level.py b/tests/test_black_level.py index fdb4fc1..17349ab 100644 --- a/tests/test_black_level.py +++ b/tests/test_black_level.py @@ -115,10 +115,17 @@ def test_dark_file_lands_in_imgs_dark(self, tmp_path, monkeypatch): monkeypatch.setattr( camera_mod, 'load_image', - lambda cam, address, mac_config=None, mac=True, demosaic=True: types.SimpleNamespace(), + lambda cam, address, mac_config=None, mac=True, demosaic=True: types.SimpleNamespace( + channels=[np.full((4, 4), v, dtype=np.uint16) for v in (10, 20, 30, 40)] + ), ) cam = camera_mod.Camera('out.json', json={}) cam.add_imgs(str(tmp_path) + '/', (0, 0)) assert [i.name for i in cam.imgs_dark] == ['dark_0.dng'] + # Dark frames are reduced to their channel statistics at load. + dark = cam.imgs_dark[0] + assert dark.channel_means == [10.0, 20.0, 30.0, 40.0] + assert dark.channel_stds == [0.0] * 4 + assert dark.channels == [] assert cam.imgs == [] # the dark frame must not be treated as Macbeth assert len(cam.imgs_alsc) == 1 diff --git a/tests/test_burst.py b/tests/test_burst.py index d4fb6c5..6eedf13 100644 --- a/tests/test_burst.py +++ b/tests/test_burst.py @@ -88,7 +88,7 @@ def _stub_loader(self, monkeypatch, values): def fake_load(cam, im_str, demosaic=True): img = Image() value = next(calls) - img.channels = [np.full((64, 64), value, dtype=np.float64) for _ in range(4)] + img.channels = [np.full((64, 64), value, dtype=np.uint16) for _ in range(4)] img.sigbits = 12 img.blacklevel_16 = 0 img.name = im_str.split('/')[-1] @@ -102,7 +102,7 @@ def fake_load(cam, im_str, demosaic=True): return loader_mod def test_group_averages_channels_and_keeps_single_patches(self, tmp_path, monkeypatch): - loader_mod = self._stub_loader(monkeypatch, [1000.0, 3000.0]) + loader_mod = self._stub_loader(monkeypatch, [1000, 3000]) cam = Camera('out.json', json={}) img = loader_mod.load_image_group(cam, ['/x/a_5000k_800l_0.dng', '/x/a_5000k_800l_1.dng'], (0, 0)) assert img is not None @@ -136,9 +136,12 @@ def fake_load(cam, im_str, demosaic=True): cam = Camera('out.json', json={}) img = loader_mod.load_image_group(cam, [f'/x/a_5000k_800l_{i}.dng' for i in range(8)], (0, 0)) + # Only the exact sums are kept after detection; the average rebuilt from them + # is what later consumers (lux) see. + assert img.channels == [] for i in range(4): expected = np.mean([f[i] for f in frames], axis=0) - np.testing.assert_array_equal(img.channels[i], expected) + np.testing.assert_array_equal(img.channel_values(i), expected) def test_single_member_group_is_plain_load(self, tmp_path, monkeypatch): loader_mod = self._stub_loader(monkeypatch, [3000.0]) From a7e55e3332966c587c82d7e59b9672813bd44d15 Mon Sep 17 00:00:00 2001 From: Naushir Patuck Date: Thu, 1 Oct 2026 14:13:37 +0100 Subject: [PATCH 3/3] ctt-server: capture the Preview-tab PNG at the selected sensor mode's resolution The "Capture PNG (full res)" button had saved the downscaled preview frame since the zero-shutter-lag change, so it now takes a still at the selected mode's resolution. That still is held to the preview frame's exposure, gain, colour gains and ISP controls, and the user's exposure and white balance settings are re-applied afterwards, so the live preview no longer steps after a snapshot. --- apps/ctt_server/app.py | 8 +- apps/ctt_server/camera.py | 64 ++++++++++------ apps/ctt_server/templates/preview.html | 2 +- docs/ctt-server.md | 6 +- tests/test_ctt_server_capture.py | 101 ++++++++++++++++++++----- 5 files changed, 130 insertions(+), 51 deletions(-) diff --git a/apps/ctt_server/app.py b/apps/ctt_server/app.py index 4183616..a924d86 100644 --- a/apps/ctt_server/app.py +++ b/apps/ctt_server/app.py @@ -476,15 +476,15 @@ def api_preview_default(): @app.route('/projects//preview-capture') def preview_capture(name: str): - """Download a PNG snapshot of the live preview (zero shutter lag). + """Download a PNG snapshot of the live preview at the selected mode's resolution. - Grabs the frame currently on screen at the selected mode's preview - resolution, so manual exposure/gain is preserved (no mode-switch blip). + Captured at the preview's exposure, gain and white balance, which are left + unchanged on the live preview afterwards. """ proj = get_project_or_404(name) cam = camera_or_503() try: - png = cam.capture_preview_png() + png = cam.capture_png() except CameraError as err: abort(503, str(err)) stamp = datetime.now().strftime('%Y%m%d_%H%M%S') diff --git a/apps/ctt_server/camera.py b/apps/ctt_server/camera.py index b85b84a..a883647 100644 --- a/apps/ctt_server/camera.py +++ b/apps/ctt_server/camera.py @@ -66,6 +66,10 @@ def __init__(self, preview_max_width: int = 1920, tuning_file: str | None = None 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) + # Manual exposure/gain and white balance as last set, re-applied after a still + # capture: reconfiguring the pipeline resets controls to the config's defaults. + self._manual_exposure: dict = {} + self._awb_controls: dict = {} # 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. @@ -232,6 +236,7 @@ def set_controls(self, controls: dict) -> dict: self._auto = bool(controls['auto_exposure']) if self._auto: new['AeEnable'] = True + self._manual_exposure = {} else: # Manual: disable AEC and apply the requested exposure/gain. new['AeEnable'] = False @@ -239,6 +244,7 @@ def set_controls(self, controls: dict) -> dict: new['ExposureTime'] = int(controls['exposure']) if controls.get('gain') is not None: new['AnalogueGain'] = float(controls['gain']) + self._manual_exposure.update({k: new[k] for k in ('ExposureTime', 'AnalogueGain') if k in new}) 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 @@ -247,11 +253,13 @@ def set_controls(self, controls: dict) -> dict: new['FrameDurationLimits'] = self._frame_duration_limits() if 'awb' in controls: new['AwbEnable'] = bool(controls['awb']) + self._awb_controls = {} if new['AwbEnable'] else {'AwbEnable': False} if controls.get('colour_gains') is not None: # Explicit gains imply manual white balance. r_gain, b_gain = controls['colour_gains'] new['AwbEnable'] = False new['ColourGains'] = (float(r_gain), float(b_gain)) + self._awb_controls = {'AwbEnable': False, 'ColourGains': new['ColourGains']} if new: self._picam2.set_controls(new) time.sleep(0.3) # let the pipeline apply the new controls @@ -269,46 +277,52 @@ def capture_jpeg(self, quality: int = 95) -> bytes: return buf.tobytes() def capture_png(self) -> bytes: - """Capture a full-resolution processed still (current tuning applied) as PNG. - - Briefly switches the pipeline to a full-sensor-resolution still mode, then - back to the preview/video config — so the result is the full field of view - at native resolution, not the downscaled preview stream. + """Snapshot the live preview as a PNG at the selected sensor mode's resolution. + + The live main stream is only preview-sized, so the frame comes from a brief + switch to a still configuration at the mode's resolution. That still is held + to the preview frame's exposure, gain and colour gains, with the preview's ISP + controls (e.g. denoise), so it matches what is on screen. Switching back + resets the controls to the config's defaults, so the user's exposure and white + balance settings are re-applied and the live preview carries on unchanged. """ import cv2 # noqa: PLC0415 with self._lock: + md = self._picam2.capture_metadata() + video = self._video_config() + controls = dict(video.get('controls', {})) + controls.update( + AeEnable=False, + ExposureTime=int(md['ExposureTime']), + AnalogueGain=float(md['AnalogueGain']), + ) + if md.get('ColourGains'): + controls.update(AwbEnable=False, ColourGains=tuple(float(g) for g in md['ColourGains'])) still = self._picam2.create_still_configuration( main={'size': self.resolution, 'format': 'RGB888'}, sensor=self._sensor_config(), transform=self._transform(), + controls=controls, ) - arr = self._picam2.switch_mode_and_capture_array(still, 'main') + try: + arr = self._picam2.switch_mode_and_capture_array(still, 'main') + finally: + self._picam2.set_controls( + { + 'AeEnable': self._auto, + 'AwbEnable': True, + 'ExposureValue': self._ev, + **self._manual_exposure, + **self._awb_controls, + } + ) # Picamera2 'RGB888' arrays are BGR-ordered, which is exactly what cv2 wants. ok, buf = cv2.imencode('.png', arr) if not ok: 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 capture_still_frames(self, frames: int) -> list: """Capture a burst of full-resolution processed frames as arrays. diff --git a/apps/ctt_server/templates/preview.html b/apps/ctt_server/templates/preview.html index 734a55c..ab2ceef 100644 --- a/apps/ctt_server/templates/preview.html +++ b/apps/ctt_server/templates/preview.html @@ -37,7 +37,7 @@

Live preview test

diff --git a/docs/ctt-server.md b/docs/ctt-server.md index 3b51bf3..ccb1eb4 100644 --- a/docs/ctt-server.md +++ b/docs/ctt-server.md @@ -182,8 +182,8 @@ restores the default tuning. While previewing you can: -- inspect detail with a click-and-hold loupe magnifier, and capture a - full-resolution PNG; +- inspect detail with a click-and-hold loupe magnifier, and capture a PNG + snapshot of the preview at the selected sensor mode's resolution; - control exposure — auto with EV compensation, or manual exposure time and analogue gain, plus an FPS limit (0 = unconstrained, allowing long exposures) and H/V flips; @@ -424,7 +424,7 @@ tagging or sweeping against a bad reading. The web UI shows out-of-range values | POST | `/projects//tuning/custom//delete` | Revert: remove the custom file | | POST | `/projects//preview-test` | Restart the camera with this project's tuning; `{"kind": "generated"\|"custom"}` (custom files are canary-tested in a subprocess first) | | POST | `/api/preview-default` | Restore the built-in default tuning | -| GET | `/projects//preview-capture` | Full-resolution PNG still from the live preview | +| GET | `/projects//preview-capture` | PNG snapshot of the live preview at the selected sensor mode's resolution | ### MTF diff --git a/tests/test_ctt_server_capture.py b/tests/test_ctt_server_capture.py index 771ad2b..c10c076 100644 --- a/tests/test_ctt_server_capture.py +++ b/tests/test_ctt_server_capture.py @@ -2,46 +2,111 @@ # # Copyright (C) 2026, Raspberry Pi # -# The Preview-tab snapshot is a zero-shutter-lag grab of the live preview frame: -# it reads the running main stream with NO mode switch, so the manual exposure/gain -# and the ISP denoise state survive (capture_png's still-mode switch reverts auto- -# exposure and steps the brightness). A fake Picamera2 lets this run without hardware. +# The Preview-tab PNG is a snapshot of the live preview at the selected sensor mode's +# resolution. The live main stream is only preview-sized, so it comes from a brief +# still-mode switch; the still must be held to the preview frame's exposure, gain and +# colour gains, and the user's controls re-applied afterwards (reconfiguring resets +# them). A fake Picamera2 lets this run without hardware. import threading import numpy as np +import pytest from ctt_server.camera import Picamera2Camera +PREVIEW_MD = {'ExposureTime': 12345, 'AnalogueGain': 2.5, 'ColourGains': (1.8, 1.6)} + class FakePicam2: def __init__(self): + self.still_configs = [] self.switch_calls = [] - self.array_calls = [] + self.set_calls = [] + + def capture_metadata(self): + return dict(PREVIEW_MD) - def capture_array(self, stream): - self.array_calls.append(stream) - return np.zeros((4, 4, 3), dtype=np.uint8) + def create_video_configuration(self, **kwargs): + return {'controls': {'NoiseReductionMode': 'Fast', **kwargs.get('controls', {})}} - def switch_mode(self, cfg): - self.switch_calls.append(cfg) + def create_still_configuration(self, **kwargs): + self.still_configs.append(kwargs) + return kwargs def switch_mode_and_capture_array(self, cfg, stream): - self.switch_calls.append(cfg) - return np.zeros((4, 4, 3), dtype=np.uint8) + self.switch_calls.append(stream) + w, h = cfg['main']['size'] + return np.zeros((h, w, 3), dtype=np.uint8) + def set_controls(self, controls): + self.set_calls.append(dict(controls)) -def _camera(fake): + +def _camera(fake, monkeypatch): """A Picamera2Camera bound to a fake picam2, bypassing the hardware __init__.""" cam = object.__new__(Picamera2Camera) cam._picam2 = fake cam._lock = threading.Lock() + cam._raw_size = cam.resolution = (64, 48) + cam._raw_format = None + cam._raw_bit_depth = 10 + cam._preview_size = (32, 24) + cam._fps = 30.0 + cam._ev = 0.5 + cam._auto = True + cam._manual_exposure = {} + cam._awb_controls = {} + monkeypatch.setattr(Picamera2Camera, '_transform', lambda self: None) + monkeypatch.setattr(Picamera2Camera, '_frame_duration_limits', lambda self: (33333, 33333)) return cam -def test_preview_snapshot_grabs_running_stream_without_switching(): - fake = FakePicam2() - png = _camera(fake).capture_preview_png() +def _png_size(png: bytes) -> tuple[int, int]: assert png[:8] == b'\x89PNG\r\n\x1a\n' # a real PNG - assert fake.array_calls == ['main'] # grabbed the live preview frame - assert fake.switch_calls == [] # zero shutter lag: no mode switch + return int.from_bytes(png[16:20], 'big'), int.from_bytes(png[20:24], 'big') + + +def test_snapshot_is_at_the_selected_mode_resolution(monkeypatch): + fake = FakePicam2() + assert _png_size(_camera(fake, monkeypatch).capture_png()) == (64, 48) + assert fake.still_configs[0]['main']['size'] == (64, 48) + + +def test_snapshot_still_matches_the_preview_frame(monkeypatch): + fake = FakePicam2() + _camera(fake, monkeypatch).capture_png() + controls = fake.still_configs[0]['controls'] + assert controls['AeEnable'] is False + assert controls['ExposureTime'] == 12345 + assert controls['AnalogueGain'] == 2.5 + assert controls['AwbEnable'] is False + assert controls['ColourGains'] == (1.8, 1.6) + assert controls['NoiseReductionMode'] == 'Fast' # the preview's ISP controls, not the still defaults + + +@pytest.mark.parametrize( + 'auto, manual, awb, expected', + [ + (True, {}, {}, {'AeEnable': True, 'AwbEnable': True, 'ExposureValue': 0.5}), + ( + False, + {'ExposureTime': 20000, 'AnalogueGain': 4.0}, + {'AwbEnable': False, 'ColourGains': (2.0, 1.5)}, + { + 'AeEnable': False, + 'AwbEnable': False, + 'ExposureValue': 0.5, + 'ExposureTime': 20000, + 'AnalogueGain': 4.0, + 'ColourGains': (2.0, 1.5), + }, + ), + ], +) +def test_user_controls_restored_after_the_snapshot(monkeypatch, auto, manual, awb, expected): + fake = FakePicam2() + cam = _camera(fake, monkeypatch) + cam._auto, cam._manual_exposure, cam._awb_controls = auto, manual, awb + cam.capture_png() + assert fake.set_calls == [expected]