Live button-driven scanning with EEPROM calibration

- ScanSession: scan-mode context manager with bright-pixel auto-exposure and
  sensor colour gains; tools/scan.py scans on the scan button with live stitching
- Read-only EEPROM access and calib.py parser (laser geometry, intrinsics,
  colour gains, black/white flat-field samples); flat-field correction in Stitcher
- OV7740 sensor access via the MCU proxy; colour gains from EEPROM fix green cast
- Stitcher: colour support, high-passed registration (robust on blank paper)
- RE notes on the Mac app's colour, exposure, stitching and button handling

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
erik
2026-10-02 11:13:59 +00:00
co-authored by Claude Opus 5.5
parent e1c1012c1c
commit e70d56d957
11 changed files with 919 additions and 82 deletions
+132
View File
@@ -0,0 +1,132 @@
"""Per-device calibration stored in the mouse's EEPROM (written at the factory).
Layout (from the Mac app, see RE_colour_stitch.md 1.4): 0x3000 u16 LE blob length N,
0x3002 4-byte checksum, 0x3006 N-byte boost::archive binary blob (i386 sizes) holding
Roi, Sensor, Undistortion, Color and Info params. Fields are located by structure, not
fully deserialised; offsets below were checked against one unit only.
Colour calibration is two sets of point samples, black and white, each
`u16 x, u16 y, u32 3, u8 b, g, r` in raw-frame coordinates, which the app fits with
NURBS surfaces and uses for a per-pixel linear correction (`slope*in + intercept`,
white -> colorDestBrightImage = 250).
"""
import os
import struct
import numpy as np
EEPROM_BASE = 0x3000
DEST_WHITE = 250.0
CACHE = os.path.join(os.path.dirname(__file__), "..", "calib", "eeprom_3000.bin")
def read_eeprom(x, cache=CACHE):
"""Header + blob, cached on disk after the first (~15 s) read."""
if cache and os.path.exists(cache):
return open(cache, "rb").read()
head = x.eeprom_read(EEPROM_BASE, 6)
n = struct.unpack("<H", head[:2])[0]
if not 0 < n <= 20000:
raise ValueError(f"no calibration blob (length {n})")
data = head + x.eeprom_read(EEPROM_BASE + 6, n)
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
open(cache, "wb").write(data)
return data
def _sample_runs(blob, min_len=50):
"""Find runs of 11-byte colour samples (the u32 3 marks each one)."""
runs, i = [], 4
while i < len(blob) - 11:
j = i
while blob[j:j + 4] == b"\x03\x00\x00\x00":
j += 11
if (j - i) // 11 >= min_len:
runs.append((i - 4, (j - i) // 11))
i = j
else:
i += 1
return runs
def parse(data):
blob = data[6:]
if blob[4:26] != b"serialization::archive":
raise ValueError("not a boost archive")
d = lambda o: struct.unpack_from("<d", blob, o)[0] # noqa: E731
runs = _sample_runs(blob)
if len(runs) != 2:
raise ValueError(f"expected black+white sample runs, found {len(runs)}")
samples = [np.array([struct.unpack_from("<2HI3B", blob, s + 11 * k) for k in range(n)],
np.float32)[:, [0, 1, 3, 4, 5]] for s, n in runs]
end = runs[1][0] + 11 * runs[1][1]
return {
"roi": struct.unpack_from("<4i", blob, 0x29), # x0, y0, x1, y1
"meters_per_unit": (d(0x3B), d(0x43)), # left, right laser
"laser_left_m": (d(0x5B), d(0x63)),
"laser_right_m": (d(0x73), d(0x7B)),
"focal_px": d(0x85), "principal_px": (d(0x8D), d(0x95)),
"pixel_size_m": d(0x111),
"black": samples[0], "white": samples[1], # columns x, y, b, g, r
"gains_rgb": struct.unpack_from("<3i", blob, end + 4), # sensor regs 1, 2, 3
}
def _bspline_basis(t, n_ctrl, length):
"""Uniform cubic B-spline basis (len(t) x n_ctrl) on [0, length)."""
seg = n_ctrl - 3
u = np.clip(np.asarray(t, np.float64) / length * seg, 0, seg - 1e-9)
k = u.astype(int)
f = u - k
w = np.stack([(1 - f) ** 3, 3 * f ** 3 - 6 * f ** 2 + 4, -3 * f ** 3 + 3 * f ** 2 + 3 * f + 1, f ** 3], 1) / 6
B = np.zeros((len(u), n_ctrl))
for j in range(4):
B[np.arange(len(u)), k + j] = w[:, j]
return B
def fit_surface(s, shape, ctrl=(6, 10), ridge=1.0):
"""Smooth surface (H x W x 3) through scattered samples (x, y, b, g, r)."""
h, w = shape
ny, nx = ctrl
By, Bx = _bspline_basis(s[:, 1], ny, h), _bspline_basis(s[:, 0], nx, w)
A = (By[:, :, None] * Bx[:, None, :]).reshape(len(s), -1)
# Ridge towards the mean keeps sparsely sampled corners from running off.
mean = s[:, 2:].mean(0)
C = np.linalg.solve(A.T @ A + ridge * np.eye(A.shape[1]), A.T @ (s[:, 2:] - mean)) + mean
Gy, Gx = _bspline_basis(np.arange(h), ny, h), _bspline_basis(np.arange(w), nx, w)
return np.einsum("yi,ijc,xj->yxc", Gy, C.reshape(ny, nx, 3), Gx).astype(np.float32)
class FlatField:
"""Per-pixel colour correction from the EEPROM black/white surfaces.
Calling the object divides a frame by the white surface normalised to mean 1 per
channel: that removes the LED/lens falloff whatever the current exposure and sensor
gains, leaving overall colour balance to the stitcher's white patch. `apply` is the
app's full linear correction, valid only at the factory exposure and gains."""
def __init__(self, cal, shape=(296, 640)):
self.black = fit_surface(cal["black"], shape)
self.white = fit_surface(cal["white"], shape)
self.shape = self.white / self.white.reshape(-1, 3).mean(0)
self.inv = 1 / self.shape
self.inv_gray = 1 / self.shape.mean(2)
def __call__(self, frame):
return frame.astype(np.float32) * (self.inv if frame.ndim == 3 else self.inv_gray)
def apply(self, frame, level=1.0):
f = frame.astype(np.float32)
b, w = self.black * level, self.white * level
if f.ndim == 2:
b, w = b.mean(2), w.mean(2)
return (f - b) * (DEST_WHITE / np.maximum(w - b, 1))
def load(x=None, cache=CACHE):
if x is None and not (cache and os.path.exists(cache)):
from .xu import XU
x = XU()
return parse(read_eeprom(x, cache))
+79
View File
@@ -0,0 +1,79 @@
"""Scan-mode session: enter scan mode, stream decoded frames with auto-exposure,
and always restore mouse mode on exit."""
import os
import time
import cv2
import numpy as np
from .frame import fix_and_extract
from .xu import XU
def quiet_libjpeg():
"""libjpeg warns on every frame (the firmware's headers); send fd 2 to /dev/null
and return a Python stream for the real stderr."""
import sys
sys.stderr = os.fdopen(os.dup(2), "w")
os.dup2(os.open(os.devnull, os.O_WRONLY), 2)
def bright_level(im, pct=95):
"""Brightness of the paper: the `pct` percentile of the brightest channel."""
px = im[::4, ::4]
px = px.reshape(-1, 3).max(1) if px.ndim == 3 else px
return float(np.percentile(px, pct))
class ScanSession:
def __init__(self, video="/dev/video4", target=200, exposure=None, xu=None, gains=None):
"""gains: sensor colour gains for regs 1/2/3 (the EEPROM's gainR/G/B), or None."""
self.video, self.target, self.fixed, self.gains = video, target, exposure, gains
self.x = xu or XU(video)
self.e6 = exposure or 40
self.cap = None
def __enter__(self):
x = self.x
if not (x.disable_mouse()[0] and x.mcu_write_verified(0x2D, 0x60)
and x.mcu_write_verified(0xE7, 0x28) and x.mcu_write_verified(0xE6, self.e6)
and x.start_scan()[0]):
x.enable_mouse()
raise RuntimeError("failed to enter scan mode")
if self.gains and not x.set_color_gains(*self.gains):
print("warning: could not set sensor colour gains", flush=True)
self.cap = cv2.VideoCapture(self.video, cv2.CAP_V4L2)
self.cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*"MJPG"))
self.cap.set(cv2.CAP_PROP_CONVERT_RGB, 0)
return self
def __exit__(self, *exc):
if self.cap is not None:
self.cap.release()
self.cap = None
self.restored = self.x.enable_mouse()[0]
return False
def frames(self, color=True):
"""Yield (t, n, image, record); image is BGR (or grayscale) 640x296."""
t0, n = time.time(), 0
flag = cv2.IMREAD_COLOR if color else cv2.IMREAD_GRAYSCALE
while True:
ok, f = self.cap.read()
if not ok:
continue
clean, rec = fix_and_extract(f.tobytes())
im = cv2.imdecode(np.frombuffer(clean, np.uint8), flag)
if im is None or rec is None:
continue
yield time.time() - t0, n, im, rec
# Proportional auto-exposure every 3 frames (brightness is ~linear in E6). It
# regulates the bright end (paper) of the strongest channel, so no channel clips;
# clipping would break the linear flat-field correction.
if self.fixed is None and n % 3 == 0:
level = bright_level(im)
if level > 1:
new = int(min(255, max(2, round(self.e6 * (self.target / level) ** 0.7))))
if new != self.e6 and self.x.mcu_write_verified(0xE6, new):
self.e6 = new
n += 1
+84 -30
View File
@@ -16,6 +16,7 @@ import cv2
SHIFT = np.array([[-0.0222, -0.3551], # dx from (a, b)
[0.3239, 0.0020]]) # dy from (a, b)
ROT_GAIN = -4.4e-4
MIN_PHASE_RESP = 0.2 # phase-correlation peak needed to override the lasers
def _s16(v):
@@ -50,6 +51,16 @@ def relative_motion(d, center, rot_gain=ROT_GAIN, shift=None):
return _aff(0, cx, cy) @ _aff(-th, 0, 0) @ _aff(0, -cx - dx, -cy - dy)
def _highpass(img, mask=None, sigma=12):
"""Remove illumination (vignetting, LED falloff) so registration sees only page texture.
With a mask, the local mean is taken over masked pixels only."""
if mask is None:
return img - cv2.GaussianBlur(img, (0, 0), sigma)
m = mask.astype(np.float32)
mean = cv2.GaussianBlur(img * m, (0, 0), sigma) / np.maximum(cv2.GaussianBlur(m, (0, 0), sigma), 1e-3)
return (img - mean) * m
def _feather(h, w):
y = np.minimum(np.arange(h), np.arange(h)[::-1]) + 1
x = np.minimum(np.arange(w), np.arange(w)[::-1]) + 1
@@ -57,22 +68,31 @@ def _feather(h, w):
class Stitcher:
def __init__(self, frame_shape, margin=600, refine=True, rot_gain=ROT_GAIN):
self.h, self.w = frame_shape
self.refine, self.rot_gain = refine, rot_gain
"""Feed frames (grayscale or BGR) with their raw record values; registration runs on
grayscale, blending keeps all channels. Dropped frames are fine: the laser
positions are cumulative, so deltas are taken against the last frame added."""
def __init__(self, frame_shape, margin=600, refine=True, rot_gain=ROT_GAIN, flat=None):
self.h, self.w = frame_shape[:2]
self.ch = frame_shape[2] if len(frame_shape) > 2 else 1
self.refine, self.rot_gain, self.flat = refine, rot_gain, flat
self.size = (self.w + 2 * margin, self.h + 2 * margin) # grows as needed
self.acc = np.zeros(self.size[::-1], np.float32)
self.wsum = np.zeros_like(self.acc)
self.acc = np.zeros(self.size[::-1] + ((self.ch,) if self.ch > 1 else ()), np.float32)
self.wsum = np.zeros(self.size[::-1], np.float32)
self.pose = _aff(0, margin, margin) # frame -> canvas
self.feather = _feather(self.h, self.w)
self.window = cv2.createHanningWindow((self.w, self.h), cv2.CV_32F)
self.prev = None
self.prev = None # previous frame, high-passed
self.prev_v = None # previous raw record values
self.n = 0
self.stats = {"refined": 0, "rejected": 0, "skipped": 0, "image_shift": 0}
def _corners(self, M):
c = np.array([[0, 0, 1], [self.w, 0, 1], [0, self.h, 1], [self.w, self.h, 1.0]]).T
return (M @ c)[:2]
def _grow(self, M):
corners = np.array([[0, 0, 1], [self.w, 0, 1], [0, self.h, 1], [self.w, self.h, 1.0]]).T
q = (M @ corners)[:2]
q = self._corners(M)
pad = 400
left = int(max(0, pad - q[0].min())) if q[0].min() < 50 else 0
top = int(max(0, pad - q[1].min())) if q[1].min() < 50 else 0
@@ -81,25 +101,37 @@ class Stitcher:
if left or top or right or bottom:
self.acc = cv2.copyMakeBorder(self.acc, top, bottom, left, right, cv2.BORDER_CONSTANT, 0)
self.wsum = cv2.copyMakeBorder(self.wsum, top, bottom, left, right, cv2.BORDER_CONSTANT, 0)
self.size = self.acc.shape[::-1]
self.size = self.wsum.shape[::-1]
shift = _aff(0, left, top)
self.pose = shift @ self.pose
return shift @ M
return M
def _refine(self, frame, M):
def _roi(self, M):
"""Canvas rectangle (x0, y0, x1, y1) covered by a frame at pose M."""
q = self._corners(M)
x0, y0 = np.floor(q.min(1)).astype(int) - 1
x1, y1 = np.ceil(q.max(1)).astype(int) + 2
return max(x0, 0), max(y0, 0), min(x1, self.size[0]), min(y1, self.size[1])
def _gray_canvas(self, x0, y0, x1, y1):
acc = self.acc[y0:y1, x0:x1]
return acc.mean(axis=2) if self.ch > 1 else acc
def _refine(self, gray, M):
# Canvas content seen through the predicted pose, in frame coordinates.
Mi = np.linalg.inv(M)[:2]
acc = cv2.warpAffine(self.acc, Mi, (self.w, self.h), flags=cv2.INTER_LINEAR)
ws = cv2.warpAffine(self.wsum, Mi, (self.w, self.h), flags=cv2.INTER_LINEAR)
x0, y0, x1, y1 = self._roi(M)
Mi = np.linalg.inv(_aff(0, -x0, -y0) @ M)[:2]
acc = cv2.warpAffine(self._gray_canvas(x0, y0, x1, y1), Mi, (self.w, self.h), flags=cv2.INTER_LINEAR)
ws = cv2.warpAffine(self.wsum[y0:y1, x0:x1], Mi, (self.w, self.h), flags=cv2.INTER_LINEAR)
mask = ws > 0.3
if mask.mean() < 0.3:
self.stats["skipped"] += 1
return M
patch = np.where(mask, acc / np.maximum(ws, 1e-6), 0).astype(np.float32)
patch = _highpass(np.where(mask, acc / np.maximum(ws, 1e-6), 0).astype(np.float32), mask)
W = np.eye(2, 3, dtype=np.float32)
try:
_, W = cv2.findTransformECC(patch, frame.astype(np.float32), W, cv2.MOTION_EUCLIDEAN,
_, W = cv2.findTransformECC(patch, gray, W, cv2.MOTION_EUCLIDEAN,
(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 60, 1e-4),
mask.astype(np.uint8), 5)
except cv2.error:
@@ -113,33 +145,55 @@ class Stitcher:
self.stats["refined"] += 1
return M @ np.linalg.inv(W3)
def add(self, frame, deltas):
frame = frame.astype(np.float32)
def add(self, frame, v):
"""frame: HxW or HxWx3 uint8; v: the record's six raw values (cumulative laser counts)."""
frame = frame.astype(np.float32) if self.flat is None else self.flat(frame)
gray = frame.mean(axis=2) if frame.ndim == 3 else frame
gray = np.ascontiguousarray(gray, np.float32)
hp = _highpass(gray)
v = np.asarray(v)
if self.n:
# At speed the lasers can disagree with each other (one over-counts), so check
# the translation against frame-to-frame phase correlation.
deltas = laser_deltas(np.stack([self.prev_v, v]))[1]
# The lasers sometimes miscount at speed (one or both), so check their translation
# against frame-to-frame phase correlation. High-passed frames keep the fixed
# illumination pattern from reading as "no motion" on blank paper.
shift = None
(dx, dy), resp = cv2.phaseCorrelate(self.prev, frame, self.window)
if resp > 0.2 and np.hypot(*(laser_shift(deltas) - (dx, dy))) > 2:
(dx, dy), resp = cv2.phaseCorrelate(self.prev, hp, self.window)
if resp > MIN_PHASE_RESP and np.hypot(*(laser_shift(deltas) - (dx, dy))) > 2:
shift = (dx, dy)
self.stats["image_shift"] += 1
M = self.pose @ relative_motion(deltas, (self.w / 2, self.h / 2), self.rot_gain, shift)
M = self._grow(M)
if self.refine:
M = self._refine(frame, M)
M = self._refine(hp, M)
else:
M = self.pose
self.pose = M
self.prev = frame
self.acc += cv2.warpAffine(frame * self.feather, M[:2], self.size, flags=cv2.INTER_LINEAR)
self.wsum += cv2.warpAffine(self.feather, M[:2], self.size, flags=cv2.INTER_LINEAR)
self.pose, self.prev, self.prev_v = M, hp, v
x0, y0, x1, y1 = self._roi(M)
Ml = (_aff(0, -x0, -y0) @ M)[:2]
wf = self.feather[..., None] if frame.ndim == 3 else self.feather
self.acc[y0:y1, x0:x1] += cv2.warpAffine(frame * wf, Ml, (x1 - x0, y1 - y0), flags=cv2.INTER_LINEAR)
self.wsum[y0:y1, x0:x1] += cv2.warpAffine(self.feather, Ml, (x1 - x0, y1 - y0), flags=cv2.INTER_LINEAR)
self.n += 1
def image(self, background=255):
out = np.where(self.wsum > 0.05, self.acc / np.maximum(self.wsum, 1e-6), background)
out = out.clip(0, 255).astype(np.uint8)
ys, xs = np.nonzero(self.wsum > 0.05)
def image(self, white_balance=True, background=255):
covered = self.wsum > 0.05
ws = np.maximum(self.wsum, 1e-6)
out = self.acc / (ws[..., None] if self.ch > 1 else ws)
if white_balance:
out = white_patch(out, covered)
m = covered[..., None] if self.ch > 1 else covered
out = np.where(m, out, background).clip(0, 255).astype(np.uint8)
ys, xs = np.nonzero(covered)
if len(ys):
out = out[ys.min():ys.max() + 1, xs.min():xs.max() + 1]
# Frames arrive transposed and mirrored relative to the page.
return cv2.flip(cv2.transpose(out), -1)
def white_patch(img, mask, pct=97, white=245):
"""Scale each channel so its bright percentile (the paper) maps to `white`."""
img = img.astype(np.float32)
px = img[mask]
ref = np.percentile(px, pct, axis=0)
return img * (white / np.maximum(ref, 1))
+59
View File
@@ -102,3 +102,62 @@ class XU:
def scan_button(self):
return self.bridge_read(0xBC) & 1
# --- EEPROM (per-device calibration), read-only; protocol from UVCControl EEPROMRead (uvc: 0x1bb9).
# Selector-3 commands: 01 = set addr+len, 02 = start read, 03 = next 32 bytes, 05 = end.
# (The write path uses 06/07/09 and is deliberately not implemented.)
def _buf(self, *cmd, read=False):
req = bytes(cmd) + bytes(33 - len(cmd))
self._query(SEL_BUF, _SET_CUR, req)
if read:
return self._query(SEL_BUF, _GET_CUR, bytes(33))
def eeprom_read(self, addr, length):
self._buf(0x01, addr & 0xFF, addr >> 8, length & 0xFF, length >> 8, read=True)
self._buf(0x02)
out = bytearray()
try:
while len(out) < length:
out += self._buf(0x03, read=True)[1:33]
finally:
self._buf(0x05)
return bytes(out[:length])
# --- image sensor (OV7740 at SCCB 0x42) through the MCU proxy BD/BE/B8, as the app's
# OVMouseManager does (0x1390a30 write, 0x1390860 read). Only answers in standby/scan mode.
def _proxy_idle(self):
for _ in range(100):
if self.bridge_read(0xB8) == 0:
return True
time.sleep(0.01)
return False
def sensor_read(self, reg):
if not self._proxy_idle():
return None
self.bridge_write(0xBD, reg)
self.bridge_write(0xB8, 2)
for _ in range(100):
if self.bridge_read(0xB8) == 0:
break
return self.bridge_read(0xBE)
def sensor_write(self, reg, value, tries=10):
"""Write and read back with retries, like sccbWrite(0x42, ...)."""
for _ in range(tries):
if not self._proxy_idle():
return False
self.bridge_write(0xBD, reg)
self.bridge_write(0xBE, value)
self.bridge_write(0xB8, 1)
time.sleep(0.01)
if self.sensor_read(reg) == value:
return True
return False
def set_color_gains(self, reg1, reg2, reg3):
"""OV7740 colour gains: reg 0x01 = blue, 0x02 = red, 0x03 = green (0x40 = 1x).
The app writes the EEPROM's gainR/G/B to regs 1/2/3 in that order."""
return all(self.sensor_write(r, v) for r, v in ((1, reg1), (2, reg2), (3, reg3)))