#!/usr/bin/env python3 """camera_bench.py - measure the tower camera for the capture redesign. Subcommands (one JSON object per line on stdout): info key features, ranges, versions maxfps FMT LIMIT full-res FMT at link LIMIT (bytes/s): payload, max fps stream FMT LIMIT FPS SECS continuous stream: delivered, incomplete, id gaps, stalls ae FMT LIMIT FPS DIR RATE auto-exposure step response (DIR dark|bright, RATE = IntensityControllerRate) awb FMT LIMIT FPS FACTOR RATE auto white balance step response: force both ratios to FACTOR x the auto result, then release (RATE = BalanceWhiteAutoRate) wb FMT LIMIT does BalanceRatio change the delivered pixels? save FMT LIMIT N DIR save N settled full-res frames as raw .bin + .json restore XML load the profile XML (as the program does) and check frames flow Never saves a user set or writes an XML. Every run leaves the camera in whatever state the subcommand set; `restore` puts it back on the profile. """ import json import os import statistics import sys import threading import time import inspect # vmbpy looks for the transport layers when it is imported, so the path must be # set before the import (tower 121 has it system-wide; tower 111 does not). os.environ.setdefault("GENICAM_GENTL64_PATH", "/opt/VimbaX/cti") import vmbpy import vmbpy.util.runtime_type_check as _rtc # vmbpy 1.2.0's runtime type check resolves its own forward references # ("Camera") with typing.get_type_hints, which fails under Python 3.14's # deferred annotations - start_streaming() raises NameError before a frame # arrives. Skip the check, in this process only. def _no_hints(self, func, *args, **kwargs): b = inspect.signature(func).bind(*args, **kwargs) b.apply_defaults() return b.arguments, {} _rtc.RuntimeTypeCheckEnable._RuntimeTypeCheckEnable__dismantle_sig = _no_hints # Default: the first real camera. The SDK's simulator transport layer offers # DEV_Cam1..3, which are not cameras (as in scripts/export_camera_profile.py). CAM_ID = os.environ.get("CAM_ID") EXPOSURE_CAP_US = float(os.environ.get("EXPOSURE_CAP_US", "20000")) def out(**kw): print(json.dumps(kw), flush=True) def f(cam, n): return cam.get_feature_by_name(n) def get(cam, n): try: v = f(cam, n).get() return v if isinstance(v, (int, float, bool, str)) else str(v) except Exception: return None def rng(cam, n): try: return list(f(cam, n).get_range()) except Exception: return None def setf(cam, n, v): f(cam, n).set(v) return get(cam, n) def bytes_per_px(fmt): return {"RGB8": 3, "BGR8": 3, "BayerRG8": 1, "Mono8": 1, "BayerRG12p": 1.5, "BayerRG10p": 1.25, "BayerRG12": 2, "BayerRG10": 2}[fmt] def full_res(cam, fmt, limit): """Binning 1, full sensor, FMT, link LIMIT; frame rate at its maximum.""" setf(cam, "BinningSelector", "Digital") # Each axis's minimum can depend on the other: try both orders. for _ in range(2): for n in ("BinningVertical", "BinningHorizontal"): try: setf(cam, n, 1) except Exception: pass if get(cam, "BinningHorizontal") != 1 or get(cam, "BinningVertical") != 1: raise RuntimeError(f"binning refused: {get(cam, 'BinningHorizontal')}x{get(cam, 'BinningVertical')}") setf(cam, "OffsetX", 0) setf(cam, "OffsetY", 0) setf(cam, "Width", get(cam, "WidthMax")) setf(cam, "Height", get(cam, "HeightMax")) setf(cam, "PixelFormat", fmt) setf(cam, "DeviceLinkThroughputLimitMode", "On") setf(cam, "DeviceLinkThroughputLimit", int(limit)) # Cap auto-exposure at 20 ms (tower 121's profile value), so the frame rate # measures the link, not a long exposure: tower 111's profile allows 9 s. if get(cam, "ExposureAutoMax") > EXPOSURE_CAP_US: setf(cam, "ExposureAutoMax", EXPOSURE_CAP_US) if get(cam, "ExposureTime") > EXPOSURE_CAP_US: setf(cam, "ExposureTime", EXPOSURE_CAP_US) setf(cam, "AcquisitionFrameRateEnable", True) hi = rng(cam, "AcquisitionFrameRate")[1] return hi def set_fps(cam, fps): lo, hi = rng(cam, "AcquisitionFrameRate") return setf(cam, "AcquisitionFrameRate", min(max(fps, lo), hi)) def chunks_on(cam): # ChunkEnable is writable only while ChunkModeActive is off (measured, fw 14.3). setf(cam, "ChunkModeActive", False) for sel in ("ExposureTime", "Gain", "Timestamp", "BalanceRatioRed", "BalanceRatioBlue"): setf(cam, "ChunkSelector", sel) setf(cam, "ChunkEnable", True) setf(cam, "ChunkModeActive", True) def auto_on(cam): setf(cam, "ExposureAuto", "Continuous") setf(cam, "GainAuto", "Continuous") # Brightness from a sparse sample of the frame buffer, taken in C (bytes # slicing and translate/count), so the handler stays well under a millisecond. # Per format: which bytes to sample, and how they map to 0..255. # RGB8 / BayerRG8: every byte is a value. # BayerRG12p: the third byte of each 3-byte pair is the high 8 bits of the # second pixel (odd columns: G on even rows, B on odd rows - no R sites). # BayerRG12: little-endian 16-bit; the high byte holds bits 8..11 (x16). _HI = bytes(1 if v >= 250 else 0 for v in range(256)) _HI12 = bytes(1 if v >= 15 else 0 for v in range(256)) _SAMPLE_STRIDE = 997 def sample_brightness(mv, fmt, nbytes): """(mean on a 0..255 scale, share of samples at or near white) or (None, None).""" if fmt in ("RGB8", "BGR8", "BayerRG8", "Mono8"): s, scale, tab = bytes(mv[0:nbytes:_SAMPLE_STRIDE]), 1, _HI elif fmt == "BayerRG12p": s, scale, tab = bytes(mv[2:nbytes:3 * 331]), 1, _HI elif fmt == "BayerRG12": s, scale, tab = bytes(mv[1:nbytes:2 * 499]), 16, _HI12 else: return None, None if not s: return None, None return sum(s) * scale / len(s), s.translate(tab).count(1) / len(s) class Recorder: """Frame handler: metadata only, buffer straight back - so the measurement cannot cause the drops it is counting. Saving copies only when asked.""" def __init__(self, fmt=None, w=0, h=0): # [host_t, frame_id, complete, exposure_us, gain_db, dev_ts_ns, # brightness_mean, clipped_share, wb_red, wb_blue] self.rows = [] self.lock = threading.Lock() self.want_pixels = 0 self.pixels = [] self.fmt = fmt self.nbytes = int(w * h * bytes_per_px(fmt)) if fmt else 0 def __call__(self, cam, stream, frame): t = time.monotonic() ok = frame.get_status() == vmbpy.FrameStatus.Complete row = [t, frame.get_id(), int(ok), None, None, frame.get_timestamp(), None, None, None, None] if ok: def cb(feats): try: row[3] = feats.get_feature_by_name("ChunkExposureTime").get() row[4] = feats.get_feature_by_name("ChunkGain").get() row[8] = feats.get_feature_by_name("ChunkBalanceRatioRed").get() row[9] = feats.get_feature_by_name("ChunkBalanceRatioBlue").get() except Exception: pass try: frame.access_chunk_data(cb) except Exception: pass if self.fmt and self.nbytes: try: row[6], row[7] = sample_brightness(memoryview(frame.get_buffer()), self.fmt, self.nbytes) except Exception: pass if self.want_pixels > 0: self.pixels.append((bytes(memoryview(frame.get_buffer())[: self.nbytes]), row[3], row[4])) self.want_pixels -= 1 with self.lock: self.rows.append(row) cam.queue_frame(frame) def snapshot(self): with self.lock: return list(self.rows) def summarize(rows, secs, payload): comp = [r for r in rows if r[2]] ids = [r[1] for r in rows if r[1] is not None] missing = sum(max(0, b - a - 1) for a, b in zip(ids, ids[1:])) ts = [r[0] for r in rows] gaps = [b - a for a, b in zip(ts, ts[1:])] return dict(frames=len(rows), complete=len(comp), incomplete=len(rows) - len(comp), missing_ids=missing, fps=round(len(comp) / secs, 3), mb_s=round(len(comp) * payload / secs / 1e6, 1), max_gap_s=round(max(gaps), 3) if gaps else None, gaps_over_1s=sum(1 for g in gaps if g > 1.0), median_gap_s=round(statistics.median(gaps), 4) if gaps else None) def run_stream(cam, rec, secs, stall_s=5.0): """Stream for SECS; stop early if nothing arrives for STALL_S.""" t0 = time.monotonic() cam.start_streaming(rec, buffer_count=5) stalled = False try: while time.monotonic() - t0 < secs: time.sleep(0.2) rows = rec.snapshot() last = rows[-1][0] if rows else t0 if time.monotonic() - last > stall_s and time.monotonic() - t0 > stall_s: stalled = True break finally: cam.stop_streaming() return time.monotonic() - t0, stalled def eff(e, g): """Exposure x linear gain: the brightness the auto-exposure is steering.""" return e * 10 ** (g / 20.0) def cmd_info(cam): names = ["DeviceModelName", "DeviceSerialNumber", "DeviceFirmwareVersion", "DeviceTemperature", "PixelFormat", "SensorBitDepth", "BinningHorizontal", "BinningVertical", "BinningHorizontalMode", "Width", "Height", "WidthMax", "HeightMax", "PayloadSize", "DeviceLinkThroughputLimit", "AcquisitionFrameRate", "ExposureTime", "ExposureAuto", "ExposureAutoMin", "ExposureAutoMax", "Gain", "GainAuto", "GainAutoMax", "IntensityAutoPrecedence", "IntensityControllerRate", "IntensityControllerTarget", "IntensityControllerTolerance", "IntensityControllerAlgorithm", "IntensityControllerRegion", "BalanceWhiteAuto", "BalanceRatio", "ChunkModeActive"] out(cmd="info", values={n: get(cam, n) for n in names}, ranges={n: rng(cam, n) for n in ["AcquisitionFrameRate", "DeviceLinkThroughputLimit", "IntensityControllerRate", "IntensityControllerTarget", "IntensityControllerTolerance", "BalanceRatio", "ExposureTime", "Gain"]}, vmbpy=vmbpy.__version__) def cmd_maxfps(cam, fmt, limit): auto_on(cam) hi = full_res(cam, fmt, float(limit)) out(cmd="maxfps", fmt=fmt, limit=int(float(limit)), width=get(cam, "Width"), height=get(cam, "Height"), payload=get(cam, "PayloadSize"), max_fps=round(hi, 3), exposure_us=get(cam, "ExposureTime")) def cmd_stream(cam, fmt, limit, fps, secs): auto_on(cam) full_res(cam, fmt, float(limit)) actual = set_fps(cam, float(fps)) chunks_on(cam) payload = get(cam, "PayloadSize") rec = Recorder(fmt, get(cam, "Width"), get(cam, "Height")) dur, stalled = run_stream(cam, rec, float(secs)) rows = rec.snapshot() s = summarize(rows, dur, payload) exps = [r[3] for r in rows if r[3] is not None] out(cmd="stream", fmt=fmt, limit=int(float(limit)), fps_set=round(actual, 3), secs=round(dur, 1), payload=payload, stalled_early=stalled, chunk_frames=len(exps), exposure_range=[min(exps), max(exps)] if exps else None, **s) def cmd_ae(cam, fmt, limit, fps, direction, rate): full_res(cam, fmt, float(limit)) actual = set_fps(cam, float(fps)) chunks_on(cam) setf(cam, "IntensityControllerRate", float(rate)) auto_on(cam) emin, emax = get(cam, "ExposureAutoMin"), get(cam, "ExposureAutoMax") gmax = get(cam, "GainAutoMax") rec = Recorder(fmt, get(cam, "Width"), get(cam, "Height")) cam.start_streaming(rec, buffer_count=5) try: time.sleep(8.0) # let it settle on the scene pre = [r for r in rec.snapshot() if r[3] is not None][-5:] base_e = statistics.median(r[3] for r in pre) base_g = statistics.median(r[4] for r in pre) base_mean = statistics.median(r[6] for r in pre) if pre and pre[0][6] is not None else None base_clip = statistics.median(r[7] for r in pre) if pre and pre[0][7] is not None else None base = eff(base_e, base_g) # Force a x16 step in brightness, using gain when exposure hits its limit. setf(cam, "ExposureAuto", "Off") setf(cam, "GainAuto", "Off") if direction == "dark": e = max(emin, base / 16) g = 0.0 else: e = min(emax, base * 16) g = min(gmax, max(0.0, 20 * __import__("math").log10(base * 16 / e))) setf(cam, "ExposureTime", e) setf(cam, "Gain", g) forced = eff(e, g) time.sleep(2.0) t0 = time.monotonic() auto_on(cam) time.sleep(15.0) finally: cam.stop_streaming() rows = [r for r in rec.snapshot() if r[0] >= t0 and r[3] is not None] # [s since release, exposure_us, gain_db, brightness 0..255, clipped share] series = [[round(r[0] - t0, 3), round(r[3], 1), round(r[4], 2), None if r[6] is None else round(r[6], 2), None if r[7] is None else round(r[7], 4)] for r in rows] vals = [(r[3], r[4]) for r in rows] def same(a, b): return abs(a[0] - b[0]) <= 0.005 * max(a[0], 1) and abs(a[1] - b[1]) <= 0.05 # Rule under test: first frame after which exposure+gain stay put for 3 frames. first3 = None for i in range(len(vals) - 2): if same(vals[i], vals[i + 1]) and same(vals[i + 1], vals[i + 2]): first3 = i + 2 break # Truth: the frame after which nothing changes to the end of the run. last_change = 0 for i in range(1, len(vals)): if not same(vals[i], vals[i - 1]): last_change = i first_ok = next((i for i, v in enumerate(vals) if abs(eff(*v) / base - 1) <= 0.10), None) fin = eff(*vals[-1]) if vals else None out(cmd="ae", fmt=fmt, fps_set=round(actual, 3), direction=direction, rate=rate, base_exposure_us=round(base_e, 1), base_gain_db=round(base_g, 2), base_brightness=None if base_mean is None else round(base_mean, 2), base_clipped=None if base_clip is None else round(base_clip, 4), target=get(cam, "IntensityControllerTarget"), tolerance=get(cam, "IntensityControllerTolerance"), forced_factor=round(forced / base, 3), final_vs_base=round(fin / base, 3) if fin else None, frames=len(vals), stable3_frame=first3, stable3_s=series[first3][0] if first3 is not None else None, settled_frame=last_change, settled_s=series[last_change][0] if series else None, within10pct_frame=first_ok, within10pct_s=series[first_ok][0] if first_ok is not None else None, series=series) def cmd_awb(cam, fmt, limit, fps, factor, rate): """Auto white balance step response. Exposure and gain are locked after auto-exposure settles, so only white balance moves. The auto result on the unchanged scene is the reference; both ratios are then forced to FACTOR x that and released.""" full_res(cam, fmt, float(limit)) actual = set_fps(cam, float(fps)) chunks_on(cam) auto_on(cam) setf(cam, "BalanceWhiteAutoRate", float(rate)) rec = Recorder(fmt, get(cam, "Width"), get(cam, "Height")) cam.start_streaming(rec, buffer_count=5) try: time.sleep(6.0) pre = [r for r in rec.snapshot() if r[3] is not None][-5:] setf(cam, "ExposureAuto", "Off") setf(cam, "GainAuto", "Off") setf(cam, "ExposureTime", statistics.median(r[3] for r in pre)) setf(cam, "Gain", statistics.median(r[4] for r in pre)) setf(cam, "BalanceWhiteAuto", "Continuous") t_ref = time.monotonic() time.sleep(8.0) # auto white balance finds its answer on the still scene ref_rows = [r for r in rec.snapshot() if r[0] >= t_ref + 5.0 and r[8] is not None] ref_r = statistics.median(r[8] for r in ref_rows) ref_b = statistics.median(r[9] for r in ref_rows) spread_r = (max(r[8] for r in ref_rows) - min(r[8] for r in ref_rows)) / ref_r spread_b = (max(r[9] for r in ref_rows) - min(r[9] for r in ref_rows)) / ref_b setf(cam, "BalanceWhiteAuto", "Off") lo, hi = rng(cam, "BalanceRatio") for sel, ref in (("Red", ref_r), ("Blue", ref_b)): setf(cam, "BalanceRatioSelector", sel) setf(cam, "BalanceRatio", min(hi, max(lo, ref * float(factor)))) time.sleep(2.0) t0 = time.monotonic() setf(cam, "BalanceWhiteAuto", "Continuous") time.sleep(15.0) finally: cam.stop_streaming() rows = [r for r in rec.snapshot() if r[0] >= t0 and r[8] is not None] series = [[round(r[0] - t0, 3), round(r[8], 4), round(r[9], 4)] for r in rows] def within(r, tol): return abs(r[8] / ref_r - 1) <= tol and abs(r[9] / ref_b - 1) <= tol first5 = next((i for i, r in enumerate(rows) if within(r, 0.05)), None) stay5 = next((i for i in range(len(rows)) if all(within(x, 0.05) for x in rows[i:])), None) last_change = 0 for i in range(1, len(rows)): if rows[i][8] != rows[i - 1][8] or rows[i][9] != rows[i - 1][9]: last_change = i out(cmd="awb", fmt=fmt, fps_set=round(actual, 3), factor=float(factor), rate=float(rate), ref_red=round(ref_r, 4), ref_blue=round(ref_b, 4), ref_spread_red=round(spread_r, 4), ref_spread_blue=round(spread_b, 4), frames=len(rows), within5pct_frame=first5, within5pct_s=series[first5][0] if first5 is not None else None, stays5pct_frame=stay5, stays5pct_s=series[stay5][0] if stay5 is not None else None, settled_frame=last_change, settled_s=series[last_change][0] if series else None, series=series) def px12p(buf, i): """Pixel i of a BayerRG12p buffer (two pixels in three bytes).""" j = (i >> 1) * 3 if i & 1 == 0: return buf[j] | ((buf[j + 1] & 0x0F) << 8) return (buf[j + 1] >> 4) | (buf[j + 2] << 4) def mean_sites(buf, w, h, fmt, step=16): """Mean of R, G, B sites over a sparse grid (no numpy on the tower venv).""" sums = {"R": [0, 0], "G": [0, 0], "B": [0, 0]} if fmt == "BayerRG12p": # 12-bit values, reported on the 0..255 scale (/16) for y in range(0, h - 1, step): for x in range(0, w - 1, step): i = y * w + x sums["R"][0] += px12p(buf, i) / 16; sums["R"][1] += 1 sums["G"][0] += px12p(buf, i + 1) / 16; sums["G"][1] += 1 sums["B"][0] += px12p(buf, i + w + 1) / 16; sums["B"][1] += 1 elif fmt == "BayerRG8": for y in range(0, h - 1, step): for x in range(0, w - 1, step): i = y * w + x sums["R"][0] += buf[i]; sums["R"][1] += 1 sums["G"][0] += buf[i + 1]; sums["G"][1] += 1 sums["B"][0] += buf[i + w + 1]; sums["B"][1] += 1 else: # RGB8 for y in range(0, h, step): for x in range(0, w, step): i = (y * w + x) * 3 for k, c in enumerate("RGB"): sums[c][0] += buf[i + k]; sums[c][1] += 1 return {c: round(s / n, 2) for c, (s, n) in sums.items()} def grab(cam, fmt, n=3): w, h = get(cam, "Width"), get(cam, "Height") rec = Recorder(fmt, w, h) rec.want_pixels = n cam.start_streaming(rec, buffer_count=5) try: time.sleep(1.0) # skip frames exposed before the change rec.pixels.clear() rec.want_pixels = n t0 = time.monotonic() while rec.want_pixels > 0 and time.monotonic() - t0 < 15: time.sleep(0.05) finally: cam.stop_streaming() return rec.pixels, w, h def cmd_wb(cam, fmt, limit): full_res(cam, fmt, float(limit)) chunks_on(cam) auto_on(cam) rec = Recorder() cam.start_streaming(rec, buffer_count=5) time.sleep(6.0) cam.stop_streaming() pre = [r for r in rec.snapshot() if r[3] is not None][-3:] setf(cam, "ExposureAuto", "Off") setf(cam, "GainAuto", "Off") setf(cam, "ExposureTime", statistics.median(r[3] for r in pre)) setf(cam, "Gain", statistics.median(r[4] for r in pre)) setf(cam, "BalanceWhiteAuto", "Off") res = {} for sel in ("Red", "Blue"): setf(cam, "BalanceRatioSelector", sel) orig = get(cam, "BalanceRatio") for ratio in (1.0, 2.0): setf(cam, "BalanceRatio", ratio) px, w, h = grab(cam, fmt, 2) res[f"{sel}={ratio}"] = [mean_sites(p[0], w, h, fmt) for p in px] setf(cam, "BalanceRatio", orig) out(cmd="wb", fmt=fmt, exposure_us=get(cam, "ExposureTime"), gain_db=get(cam, "Gain"), means=res) def cmd_save(cam, fmt, limit, n, d): full_res(cam, fmt, float(limit)) chunks_on(cam) auto_on(cam) rec = Recorder() cam.start_streaming(rec, buffer_count=5) time.sleep(6.0) cam.stop_streaming() px, w, h = grab(cam, fmt, int(n)) os.makedirs(d, exist_ok=True) for i, (buf, e, g) in enumerate(px): base = os.path.join(d, f"{fmt}_{i}") with open(base + ".bin", "wb") as fh: fh.write(buf) with open(base + ".json", "w") as fh: json.dump(dict(fmt=fmt, width=w, height=h, exposure_us=e, gain_db=g), fh) out(cmd="save", fmt=fmt, saved=len(px), width=w, height=h, bytes=len(px[0][0]) if px else 0, means=[mean_sites(p[0], w, h, fmt) for p in px if fmt in ("BayerRG8", "BayerRG12p", "RGB8")]) def cmd_restore(cam, xml): cam.load_settings(xml, vmbpy.PersistType.All) rec = Recorder() dur, stalled = run_stream(cam, rec, 10.0) s = summarize(rec.snapshot(), dur, get(cam, "PayloadSize")) out(cmd="restore", xml=xml, stalled_early=stalled, values={n: get(cam, n) for n in ["PixelFormat", "BinningHorizontal", "BinningVertical", "Width", "Height", "AcquisitionFrameRate", "DeviceLinkThroughputLimit", "ExposureAuto", "GainAuto", "IntensityControllerRate", "ChunkModeActive", "BalanceWhiteAuto"]}, **s) def main(): cmd, args = sys.argv[1], sys.argv[2:] with vmbpy.VmbSystem.get_instance() as vmb: if CAM_ID: cam = vmb.get_camera_by_id(CAM_ID) else: real = [c for c in vmb.get_all_cameras() if not c.get_id().startswith("DEV_Cam")] if not real: out(cmd=cmd, error="no real camera found") return 1 cam = real[0] with cam: t0 = time.monotonic() try: {"info": cmd_info, "maxfps": cmd_maxfps, "stream": cmd_stream, "ae": cmd_ae, "awb": cmd_awb, "wb": cmd_wb, "save": cmd_save, "restore": cmd_restore}[cmd](cam, *args) except Exception as e: out(cmd=cmd, args=args, error=f"{type(e).__name__}: {e}") return 1 finally: print(json.dumps(dict(cmd=cmd, elapsed_s=round(time.monotonic() - t0, 1))), file=sys.stderr) return 0 if __name__ == "__main__": sys.exit(main())