566 lines
23 KiB
Python
566 lines
23 KiB
Python
#!/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())
|