Hama myScan Linux driver tools: scan mode, frame decoding, stitching
UVC extension-unit control (mouse/scan mode, LEDs, exposure), JPEG header fix and embedded laser record decoding, session recorder, and a stitcher that combines laser dead reckoning with phase correlation and ECC refinement. FINDINGS.md holds the reverse-engineered protocol and laser calibration. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
erik
Claude Opus 5.5
@@ -0,0 +1,80 @@
|
||||
"""Decode Hama myScan / Dacuda scanner-mouse MJPEG frames.
|
||||
|
||||
The firmware hides a per-frame sensor record in the JPEG headers (reverse
|
||||
engineered from OVJpegSrcPreProc / DaReadMouseData in Omnivision's
|
||||
UVCControl.framework shipped with the Dacuda Scanner Mouse Mac app):
|
||||
|
||||
* SOF0 height/width bytes carry 4 data bytes; the true size is 640x296.
|
||||
* Entries 0x36..0x44 (relative to the FF marker) of the second DQT carry
|
||||
15 more data bytes; the true table value there is 0x19.
|
||||
"""
|
||||
import struct
|
||||
from dataclasses import dataclass
|
||||
|
||||
WIDTH, HEIGHT = 640, 296
|
||||
|
||||
|
||||
def _spcal(v, mode, bit):
|
||||
# OVSPCal: undo the encoding that keeps quant-table bytes legal.
|
||||
if mode == 2:
|
||||
r = max(v - 1, 0) * 4 + 2
|
||||
elif mode == 3:
|
||||
r = max(v - 1, 0) * 4 + 3
|
||||
elif mode == 1:
|
||||
r = v * 4 + 1
|
||||
else:
|
||||
r = v * 4
|
||||
return ((r & 0xFB) | ((bit & 1) << 2)) & 0xFF
|
||||
|
||||
|
||||
@dataclass
|
||||
class SensorRecord:
|
||||
counter: int
|
||||
v: tuple # six big-endian 16-bit values; [2] and [5] are not transmitted
|
||||
b14: int
|
||||
b15: int
|
||||
b16: int
|
||||
flags: int # five flag bits (byte 17)
|
||||
raw: bytes
|
||||
|
||||
|
||||
def fix_and_extract(jpeg: bytes):
|
||||
"""Return (clean_jpeg, SensorRecord | None)."""
|
||||
buf = bytearray(jpeg)
|
||||
s = bytearray(32)
|
||||
have = 0
|
||||
dqt = 0
|
||||
i = 0
|
||||
n = min(len(buf), 40000)
|
||||
while i < n - 1:
|
||||
if buf[i] == 0xFF:
|
||||
m = buf[i + 1]
|
||||
if m == 0xD8:
|
||||
have |= 1
|
||||
elif m == 0xC0:
|
||||
have |= 2
|
||||
s[0], s[0x14], s[0x15], s[0x16] = buf[i + 5:i + 9]
|
||||
buf[i + 5:i + 9] = struct.pack('>HH', HEIGHT, WIDTH)
|
||||
elif m == 0xDB:
|
||||
dqt += 1
|
||||
if dqt == 2:
|
||||
have |= 4
|
||||
d = buf[i + 0x36:i + 0x45]
|
||||
for dst, src in zip((0x18, 0x19, 0x1A, 2, 3, 4, 5, 8, 9, 10, 11, 14, 15, 16, 17), d):
|
||||
s[dst] = src
|
||||
buf[i + 0x36:i + 0x45] = b'\x19' * 15
|
||||
if have == 7:
|
||||
break
|
||||
i += 1
|
||||
if have != 7:
|
||||
return bytes(buf), None
|
||||
|
||||
# Each data byte gets a 2-bit mode from bytes 0x14..0x16 and one bit from 0x18..0x1a.
|
||||
plan = [(2, 0x14, 6, 0x18, 5), (3, 0x14, 4, 0x18, 4), (4, 0x14, 2, 0x18, 3), (5, 0x14, 0, 0x18, 2),
|
||||
(8, 0x15, 6, 0x18, 1), (9, 0x15, 4, 0x19, 5), (10, 0x15, 2, 0x19, 4), (11, 0x15, 0, 0x19, 3),
|
||||
(14, 0x16, 6, 0x19, 2), (15, 0x16, 4, 0x19, 1), (16, 0x16, 2, 0x1A, 5), (17, 0x16, 0, 0x1A, 4)]
|
||||
for dst, mreg, msh, breg, bsh in plan:
|
||||
s[dst] = _spcal(s[dst], (s[mreg] >> msh) & 3, s[breg] >> bsh)
|
||||
vals = tuple((s[2 * k + 2] << 8) | s[2 * k + 3] for k in range(6))
|
||||
rec = SensorRecord(s[0], vals, s[14], s[15], s[16], s[17] & 0x1F, bytes(s[:20]))
|
||||
return bytes(buf), rec
|
||||
@@ -0,0 +1,145 @@
|
||||
"""Assemble scan frames into a page image.
|
||||
|
||||
Poses come from dead reckoning on the two laser sensors embedded in each
|
||||
frame's record (see frame.py), then each frame is refined against the canvas
|
||||
built so far with ECC (Euclidean model) before it is blended in.
|
||||
|
||||
Laser calibration (from scan_text, 2026-10-02; fit against image registration):
|
||||
v0/v1 = laser 1 cumulative (a, b) counts, v3/v4 = laser 2; 16-bit wrapping.
|
||||
content shift between consecutive frames, in frame pixels:
|
||||
dx = -0.022*a - 0.355*b, dy = 0.324*a + 0.002*b (a, b = mean of both lasers)
|
||||
image rotation (rad) = ROT_GAIN * (db1 - db2)
|
||||
"""
|
||||
import numpy as np
|
||||
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
|
||||
|
||||
|
||||
def _s16(v):
|
||||
v = np.asarray(v, np.int64) & 0xFFFF
|
||||
return np.where(v >= 0x8000, v - 0x10000, v)
|
||||
|
||||
|
||||
def laser_deltas(v):
|
||||
"""v: (N, 6) record values -> (N, 4) per-frame deltas (a1, b1, a2, b2); first row zero."""
|
||||
p = _s16(np.asarray(v)[:, [0, 1, 3, 4]])
|
||||
d = np.diff(p, axis=0, prepend=p[:1])
|
||||
return (d + 0x8000) % 0x10000 - 0x8000
|
||||
|
||||
|
||||
def _aff(theta, tx, ty):
|
||||
c, s = np.cos(theta), np.sin(theta)
|
||||
return np.array([[c, -s, tx], [s, c, ty], [0, 0, 1.0]])
|
||||
|
||||
|
||||
def laser_shift(d):
|
||||
"""Content shift (dx, dy) in frame pixels predicted from one frame's laser deltas."""
|
||||
return SHIFT @ ((d[0] + d[2]) / 2, (d[1] + d[3]) / 2)
|
||||
|
||||
|
||||
def relative_motion(d, center, rot_gain=ROT_GAIN, shift=None):
|
||||
"""3x3 transform mapping current-frame pixels to previous-frame pixels.
|
||||
shift overrides the laser-predicted content shift."""
|
||||
dx, dy = laser_shift(d) if shift is None else shift
|
||||
th = rot_gain * (d[1] - d[3])
|
||||
cx, cy = center
|
||||
# content moved by (dx, dy) and rotated by th about the center: p_prev = R(-th)(p - c - d) + c
|
||||
return _aff(0, cx, cy) @ _aff(-th, 0, 0) @ _aff(0, -cx - dx, -cy - dy)
|
||||
|
||||
|
||||
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
|
||||
return np.minimum.outer(y, x).astype(np.float32).clip(max=40) / 40
|
||||
|
||||
|
||||
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
|
||||
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.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.n = 0
|
||||
self.stats = {"refined": 0, "rejected": 0, "skipped": 0, "image_shift": 0}
|
||||
|
||||
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]
|
||||
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
|
||||
right = int(max(0, q[0].max() + pad - self.size[0])) if q[0].max() > self.size[0] - 50 else 0
|
||||
bottom = int(max(0, q[1].max() + pad - self.size[1])) if q[1].max() > self.size[1] - 50 else 0
|
||||
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]
|
||||
shift = _aff(0, left, top)
|
||||
self.pose = shift @ self.pose
|
||||
return shift @ M
|
||||
return M
|
||||
|
||||
def _refine(self, frame, 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)
|
||||
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)
|
||||
W = np.eye(2, 3, dtype=np.float32)
|
||||
try:
|
||||
_, W = cv2.findTransformECC(patch, frame.astype(np.float32), W, cv2.MOTION_EUCLIDEAN,
|
||||
(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 60, 1e-4),
|
||||
mask.astype(np.uint8), 5)
|
||||
except cv2.error:
|
||||
self.stats["rejected"] += 1
|
||||
return M
|
||||
# patch(x) ~ frame(W x), patch pixel x sits at canvas M x, so frame pixel y sits at M W^-1 y.
|
||||
W3 = np.vstack([W, [0, 0, 1]]).astype(np.float64)
|
||||
if abs(np.arctan2(W3[1, 0], W3[0, 0])) > np.radians(3) or np.hypot(W3[0, 2], W3[1, 2]) > 25:
|
||||
self.stats["rejected"] += 1
|
||||
return M
|
||||
self.stats["refined"] += 1
|
||||
return M @ np.linalg.inv(W3)
|
||||
|
||||
def add(self, frame, deltas):
|
||||
frame = frame.astype(np.float32)
|
||||
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.
|
||||
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:
|
||||
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)
|
||||
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.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)
|
||||
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)
|
||||
+104
@@ -0,0 +1,104 @@
|
||||
"""Register access for the Hama myScan / Dacuda scanner mouse via its UVC extension unit.
|
||||
|
||||
Reverse engineered from Omnivision's UVCControl.framework (Dacuda Scanner Mouse
|
||||
for Mac 2.4). Extension unit 6 (GUID dd880f8a-1cba-4954-8a25-f7875967f0f7):
|
||||
|
||||
selector 1 20 bytes, GET only device info block
|
||||
selector 2 5 bytes, GET/SET register access: [cmd, addr_lo, addr_hi, value, 0]
|
||||
cmd 0x00/0x80 = bridge ("USB") register write/read
|
||||
cmd 0x01/0x81 = SCCB register write/read
|
||||
reads: SET_CUR the request, then GET_CUR; value in byte 3
|
||||
selector 3 33 bytes, GET/SET buffer access (write first, then read)
|
||||
|
||||
The SCCB slave used by cmd 0x01/0x81 is selected by bridge register 0xF1
|
||||
(0x42 = the OV image sensor). The app's SCCB "device 0x88" is a pseudo-ID that
|
||||
maps to the bridge registers (cmd 0x00/0x80); that is where the mouse MCU's
|
||||
command/status registers (0xD8, 0xBC, ...) live. The app reaches the image
|
||||
sensor through the MCU's proxy registers (0xBD/0xBE/0xB8).
|
||||
"""
|
||||
import ctypes
|
||||
import fcntl
|
||||
import os
|
||||
import time
|
||||
|
||||
UNIT = 6
|
||||
SEL_INFO, SEL_REG, SEL_BUF = 1, 2, 3
|
||||
_SET_CUR, _GET_CUR = 0x01, 0x81
|
||||
_UVCIOC_CTRL_QUERY = 0xC0107521 # _IOWR('u', 0x21, struct uvc_xu_control_query)
|
||||
|
||||
MCU = 0x88
|
||||
SENSOR = 0x42
|
||||
|
||||
|
||||
class _Query(ctypes.Structure):
|
||||
_fields_ = [("unit", ctypes.c_uint8), ("selector", ctypes.c_uint8),
|
||||
("query", ctypes.c_uint8), ("size", ctypes.c_uint16),
|
||||
("data", ctypes.c_void_p)]
|
||||
|
||||
|
||||
class XU:
|
||||
def __init__(self, dev="/dev/video4"):
|
||||
self.fd = os.open(dev, os.O_RDWR)
|
||||
|
||||
def close(self):
|
||||
os.close(self.fd)
|
||||
|
||||
def _query(self, sel, query, data):
|
||||
buf = (ctypes.c_uint8 * len(data))(*data)
|
||||
q = _Query(UNIT, sel, query, len(data), ctypes.cast(buf, ctypes.c_void_p))
|
||||
fcntl.ioctl(self.fd, _UVCIOC_CTRL_QUERY, q)
|
||||
return bytes(buf)
|
||||
|
||||
def info(self):
|
||||
return self._query(SEL_INFO, _GET_CUR, bytes(20))
|
||||
|
||||
def _reg(self, cmd, addr, value=0, read=False):
|
||||
req = bytes([cmd, addr & 0xFF, (addr >> 8) & 0xFF, value & 0xFF, 0])
|
||||
self._query(SEL_REG, _SET_CUR, req)
|
||||
if read:
|
||||
return self._query(SEL_REG, _GET_CUR, bytes(5))[3]
|
||||
|
||||
def bridge_read(self, addr):
|
||||
return self._reg(0x80, addr, read=True)
|
||||
|
||||
def bridge_write(self, addr, value):
|
||||
self._reg(0x00, addr, value)
|
||||
|
||||
def sccb_read(self, addr):
|
||||
return self._reg(0x81, addr, read=True)
|
||||
|
||||
def sccb_write(self, addr, value):
|
||||
self._reg(0x01, addr, value)
|
||||
|
||||
# --- mouse MCU (app's pseudo SCCB id 0x88 = bridge registers), as used by Hal::OVMouseManager ---
|
||||
|
||||
def mcu_write_verified(self, reg, value, tries=10):
|
||||
"""Write + read back with retries, like the app's helper at 0x138b020."""
|
||||
for _ in range(tries):
|
||||
self.bridge_write(reg, value)
|
||||
if self.bridge_read(reg) == value:
|
||||
return True
|
||||
time.sleep(0.01)
|
||||
return False
|
||||
|
||||
def mcu_command(self, value, ack, tries=100):
|
||||
"""Write the MCU command register 0xD8 and poll until it reads `ack`."""
|
||||
self.bridge_write(0xD8, value)
|
||||
got = None
|
||||
for _ in range(tries):
|
||||
got = self.bridge_read(0xD8)
|
||||
if got == ack:
|
||||
return True, got
|
||||
return False, got
|
||||
|
||||
def disable_mouse(self): # enter standby: needed before scanning
|
||||
return self.mcu_command(0x00, 0xFF)
|
||||
|
||||
def start_scan(self): # LEDs on + sensor data embedded in frames
|
||||
return self.mcu_command(0xE1, 0x99)
|
||||
|
||||
def enable_mouse(self): # back to normal mouse mode
|
||||
return self.mcu_command(0xE0, 0x11)
|
||||
|
||||
def scan_button(self):
|
||||
return self.bridge_read(0xBC) & 1
|
||||
Reference in New Issue
Block a user