| """Read one picture the way QR Scan Difficulty read every code, with the five of its eight scanners that run on Linux. |
| |
| As in the measurement: |
| * the picture is read in gray and shrunk, never enlarged, so its long side is at most each frame size (1920 down to |
| 224 pixels), with OpenCV's INTER_AREA; sizes that come out the same are read once; |
| * each frame is staged as a gray PNG; zxing-cpp, WeChat's scanner and OpenCV's two detectors read it in a worker |
| process (one OpenCV thread, library defaults, QR codes only), and ZXing reads it in a Java process the way |
| Android's Settings app calls it; |
| * a decode that runs past TIMEOUT_S counts as a miss: the worker is replaced and the rest is read without that call. |
| No measured picture had transparency; here a transparent part counts as white, as on a printed page. |
| |
| A scanner reads a frame when one of the contents it returns is the picture's content. The measurement knew each code's |
| content; here it is the content the most scanners return (ties: the most frames, then text over raw bytes, then the |
| shortest). Raw bytes and the text a scanner makes of them (ISO-8859-1, QR's default, or Shift JIS) are one content. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import base64 |
| import hashlib |
| import itertools |
| import json |
| import os |
| import queue |
| import subprocess |
| import tempfile |
| import threading |
| import time |
| from multiprocessing import get_context |
| from pathlib import Path |
|
|
| os.environ.setdefault("OPENCV_IO_MAX_IMAGE_PIXELS", "50000000") |
| VENDOR = Path(os.environ.get("QSD_VENDOR", Path(__file__).resolve().parent.parent / "vendor")) |
| SELF_TEST = Path(__file__).with_name("selftest.png") |
| FRAMES = [1920, 1280, 896, 640, 448, 320, 224] |
| POOL = ["zxing-cpp", "wechat", "opencv", "opencv-aruco"] |
| SCANNERS = [*POOL, "zxing-java"] |
| TIMEOUT_S = 20 |
| STAGE_TIMEOUT_S = 30 |
| DEADLINE_S = 40 |
| |
|
|
|
|
| class PictureError(Exception): |
| """A picture this server cannot open.""" |
|
|
|
|
| class _TooBig(Exception): |
| """More pixels than OPENCV_IO_MAX_IMAGE_PIXELS.""" |
|
|
|
|
| def frame_sizes(h: int, w: int) -> dict[tuple[int, int], list[int]]: |
| """{(w, h): [frame sizes]} for the frames, never enlarging (the measurement's rule, rounding included).""" |
| sizes: dict[tuple[int, int], list[int]] = {} |
| for f in FRAMES: |
| k = min(1.0, f / max(h, w)) |
| sizes.setdefault((max(1, round(w * k)), max(1, round(h * k))), []).append(f) |
| return sizes |
|
|
|
|
| |
|
|
| def _decoders(wechat_dir: str): |
| import cv2 |
| import zxingcpp |
|
|
| cv2.setNumThreads(1) |
| d = Path(wechat_dir) |
| wechat = cv2.wechat_qrcode.WeChatQRCode(str(d / "detect.prototxt"), str(d / "detect.caffemodel"), |
| str(d / "sr.prototxt"), str(d / "sr.caffemodel")) |
| qr_format = zxingcpp.BarcodeFormat.QRCode |
| dropped = (zxingcpp.BarcodeFormat.MicroQRCode, zxingcpp.BarcodeFormat.RMQRCode) |
|
|
| def zxing_cpp(bgr): |
| out = [] |
| for r in zxingcpp.read_barcodes(cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY), formats=qr_format): |
| if r.format in dropped or not getattr(r, "valid", True): |
| continue |
| out.append(bytes(r.bytes) if r.bytes is not None else (r.text or "").encode("utf-8")) |
| return out |
|
|
| def wechat_qr(bgr): |
| try: |
| texts, _ = wechat.detectAndDecode(bgr) |
| except UnicodeDecodeError as e: |
| return [bytes(e.object)] |
| return [t.encode("utf-8") for t in texts if t] |
|
|
| def single(detector): |
| def read(bgr): |
| text, _, _ = detector.detectAndDecode(cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY)) |
| return [text.encode("utf-8")] if text else [] |
| return read |
|
|
| return cv2, {"zxing-cpp": zxing_cpp, "wechat": wechat_qr, |
| "opencv": single(cv2.QRCodeDetector()), "opencv-aruco": single(cv2.QRCodeDetectorAruco())} |
|
|
|
|
| def _open(cv2, path: str): |
| """The picture in gray as the measurement opened it, any transparent part on white; None if it can't be opened.""" |
| import numpy as np |
|
|
| try: |
| gray = cv2.imread(path, cv2.IMREAD_GRAYSCALE) |
| if gray is None or not path.endswith((".png", ".webp")): |
| return gray |
| full = cv2.imread(path, cv2.IMREAD_UNCHANGED) |
| except cv2.error as e: |
| if "MAX_IMAGE_PIXELS" in str(e): |
| raise _TooBig from e |
| return None |
| if full is None or full.ndim != 3 or full.shape[2] != 4: |
| return gray |
| top = np.iinfo(full.dtype).max |
| if full[..., 3].min() == top: |
| return gray |
| a = full[..., 3].astype(np.float32) / top |
| g = cv2.cvtColor(full[..., :3], cv2.COLOR_BGR2GRAY).astype(np.float32) / top |
| return np.rint(255 * (g * a + (1 - a))).astype(np.uint8) |
|
|
|
|
| def _child(conn, wechat_dir: str) -> None: |
| try: |
| cv2, decoders = _decoders(wechat_dir) |
| except Exception as e: |
| conn.send(("broken", repr(e))) |
| return |
| conn.send(("ready",)) |
| while True: |
| try: |
| msg = conn.recv() |
| except EOFError: |
| return |
| if msg[0] == "stage": |
| _, path, outdir = msg |
| try: |
| gray = _open(cv2, path) |
| if gray is None: |
| conn.send(("unreadable",)) |
| continue |
| frames = [] |
| for k, ((w, h), labels) in enumerate(frame_sizes(*gray.shape).items()): |
| g = gray if (h, w) == gray.shape else cv2.resize(gray, (w, h), interpolation=cv2.INTER_AREA) |
| p = os.path.join(outdir, f"{k}.png") |
| cv2.imwrite(p, g, [cv2.IMWRITE_PNG_COMPRESSION, 1]) |
| frames.append((w, h, labels, p)) |
| conn.send(("staged", gray.shape[1], gray.shape[0], frames)) |
| except _TooBig: |
| conn.send(("too big",)) |
| except Exception: |
| conn.send(("unreadable",)) |
| elif msg[0] == "decode": |
| _, frames, skip = msg |
| for k, (_, _, _, p) in enumerate(frames): |
| todo = [n for n in POOL if (k, n) not in skip] |
| if not todo: |
| continue |
| bgr = cv2.cvtColor(cv2.imread(p, cv2.IMREAD_GRAYSCALE), cv2.COLOR_GRAY2BGR) |
| for n in todo: |
| conn.send(("start", k, n)) |
| try: |
| payloads = decoders[n](bgr) |
| except Exception: |
| payloads = [] |
| conn.send(("done", k, n, payloads)) |
| conn.send(("end",)) |
|
|
|
|
| class _Worker: |
| """One decoder worker process. A stuck one is killed; a new one starts when the next message is sent.""" |
|
|
| def __init__(self, ctx): |
| self._ctx = ctx |
| self._spawn() |
|
|
| def _spawn(self) -> None: |
| self.conn, child = self._ctx.Pipe() |
| self.proc = self._ctx.Process(target=_child, args=(child, str(VENDOR / "wechat")), daemon=True) |
| self.proc.start() |
| child.close() |
| msg = self._recv(180) |
| if msg is None or msg[0] != "ready": |
| raise RuntimeError(f"decoder worker did not start: {msg}") |
|
|
| def close(self) -> None: |
| self.proc.kill() |
| self.proc.join(5) |
| self.conn.close() |
|
|
| def _send(self, msg) -> None: |
| """Send to a live worker: one that was killed, or died while idle (killed for memory, say), is replaced.""" |
| if self.conn.closed or not self.proc.is_alive(): |
| self.close() |
| self._spawn() |
| try: |
| self.conn.send(msg) |
| except OSError: |
| self.close() |
| self._spawn() |
| self.conn.send(msg) |
|
|
| def _recv(self, timeout: float): |
| try: |
| if not self.conn.poll(timeout): |
| return None |
| return self.conn.recv() |
| except (EOFError, OSError): |
| return None |
|
|
| def stage(self, path: str, outdir: str): |
| t0 = time.monotonic() |
| self._send(("stage", path, outdir)) |
| msg = self._recv(STAGE_TIMEOUT_S) |
| if msg is None: |
| slow = time.monotonic() - t0 >= STAGE_TIMEOUT_S |
| self.close() |
| raise PictureError("This picture took too long to open." if slow else "This picture could not be opened.") |
| if msg[0] == "too big": |
| raise PictureError("This picture is over 50 megapixels.") |
| if msg[0] == "unreadable": |
| raise PictureError("This picture could not be opened.") |
| return msg[1], msg[2], msg[3] |
|
|
| def decode(self, frames, deadline: float): |
| """{(frame index, scanner): payloads}, and whether the deadline (or crashes) cut the reading short.""" |
| got: dict[tuple[int, str], list[bytes]] = {} |
| missed: set[tuple[int, str]] = set() |
| crashes = 0 |
| while True: |
| self._send(("decode", frames, missed | set(got))) |
| current = None |
| while True: |
| msg = self._recv(max(0.0, min(TIMEOUT_S, deadline - time.monotonic()))) |
| if msg is None: |
| break |
| if msg[0] == "start": |
| current = (msg[1], msg[2]) |
| elif msg[0] == "done": |
| got[(msg[1], msg[2])] = msg[3] |
| current = None |
| elif msg[0] == "end": |
| return got, False |
| |
| self.close() |
| if time.monotonic() >= deadline: |
| return got, True |
| if current is not None: |
| missed.add(current) |
| else: |
| crashes += 1 |
| if crashes > 1: |
| return got, True |
|
|
|
|
| |
|
|
| class _Java: |
| def __init__(self): |
| cp = os.pathsep.join([str(VENDOR / "zxing" / "core-3.5.2.jar"), str(VENDOR / "zxing" / "classes")]) |
| self.cmd = ["java", "-Djava.awt.headless=true", "-Xmx512m", "-cp", cp, "ZxingServer"] |
| self.lock = threading.Lock() |
| self.ids = itertools.count() |
| self._spawn() |
|
|
| def _spawn(self) -> None: |
| self.proc = subprocess.Popen(self.cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, |
| stderr=subprocess.DEVNULL, text=True, encoding="utf-8", bufsize=1) |
| self.lines: queue.Queue = queue.Queue() |
| threading.Thread(target=self._pump, args=(self.proc.stdout, self.lines), daemon=True).start() |
|
|
| @staticmethod |
| def _pump(stream, lines: queue.Queue) -> None: |
| for line in stream: |
| lines.put(line) |
| lines.put(None) |
|
|
| def close(self) -> None: |
| self.proc.kill() |
| self.proc.wait(5) |
| try: |
| self.proc.stdin.close() |
| except OSError: |
| pass |
|
|
| def read(self, paths: list[str], deadline: float) -> list[list[bytes] | None]: |
| """Per frame, ZXing's contents (empty: no read, a timeout included); None where the deadline left it unread.""" |
| out: list[list[bytes] | None] = [None] * len(paths) |
| deaths = 0 |
| for k, p in enumerate(paths): |
| |
| if not self.lock.acquire(timeout=max(0.0, deadline - time.monotonic())): |
| break |
| try: |
| left = deadline - time.monotonic() |
| if left <= 0: |
| break |
| if self.proc.poll() is not None: |
| self.close() |
| self._spawn() |
| rid = next(self.ids) |
| obj, died = None, False |
| try: |
| self.proc.stdin.write(f"{rid}\t{p}\n") |
| self.proc.stdin.flush() |
| line = self.lines.get(timeout=min(TIMEOUT_S, left)) |
| died = line is None |
| obj = json.loads(line) if line else None |
| except OSError: |
| died = True |
| except (ValueError, queue.Empty): |
| pass |
| if not isinstance(obj, dict) or obj.get("id") != rid: |
| self.close() |
| self._spawn() |
| deaths += died |
| if deaths > 1: |
| break |
| out[k] = [] if time.monotonic() < deadline else None |
| continue |
| out[k] = [base64.b64decode(c) for c in obj.get("contents", [])] |
| finally: |
| self.lock.release() |
| return out |
|
|
|
|
| |
|
|
| def _texts(p: bytes) -> set[bytes]: |
| """In UTF-8, the texts a scanner may make of raw bytes that aren't UTF-8: ISO-8859-1 (QR byte mode's default) |
| and, where they decode, Shift JIS (QR kanji mode).""" |
| try: |
| p.decode("utf-8") |
| return set() |
| except UnicodeDecodeError: |
| pass |
| out = {p.decode("latin-1").encode("utf-8")} |
| try: |
| out.add(p.decode("shift_jis").encode("utf-8")) |
| except UnicodeDecodeError: |
| pass |
| return out |
|
|
|
|
| def _same(p: bytes, q: bytes) -> bool: |
| """One content: the same bytes, or raw bytes and a text made of them.""" |
| return p == q or p in _texts(q) or q in _texts(p) |
|
|
|
|
| def _is_text(p: bytes) -> bool: |
| try: |
| p.decode("utf-8") |
| return True |
| except UnicodeDecodeError: |
| return False |
|
|
|
|
| def _consensus(contents: dict[tuple[str, int], list[bytes]]) -> bytes | None: |
| forms = {p for ps in contents.values() for p in ps} |
| if not forms: |
| return None |
|
|
| def votes(c: bytes) -> tuple[int, int]: |
| hits = [(s, k) for (s, k), ps in contents.items() if any(_same(p, c) for p in ps)] |
| return len({s for s, _ in hits}), len(hits) |
|
|
| return max(forms, key=lambda c: (*votes(c), _is_text(c), -len(c), c)) |
|
|
|
|
| def describe(content: bytes | None) -> dict | None: |
| """The content as the page may show it: printable text (UTF-8, or else ISO-8859-1), otherwise only its length.""" |
| if content is None: |
| return None |
| try: |
| text = content.decode("utf-8") |
| except UnicodeDecodeError: |
| text = content.decode("latin-1") |
| if all(c.isprintable() or c in "\r\n\t" for c in text): |
| return {"text": text[:300], "cut": len(text) > 300, "bytes": len(content)} |
| return {"text": None, "cut": False, "bytes": len(content)} |
|
|
|
|
| class Reader: |
| """Reads pictures, `slots` at a time (one decoder worker each); the ZXing process is shared.""" |
|
|
| def __init__(self, slots: int = 2): |
| ctx = get_context("spawn") |
| self.java = _Java() |
| self.workers = [_Worker(ctx) for _ in range(slots)] |
| self.free: queue.Queue = queue.Queue() |
| for w in self.workers: |
| self.free.put(w) |
| self._self_test() |
|
|
| def _self_test(self) -> None: |
| """Refuse to start, rather than place every picture wrongly, unless all five read a clean code at every size.""" |
| r = self.read(str(SELF_TEST)) |
| missed = [n for n in SCANNERS if not all(r["reads"][n])] |
| if missed or r["partial"]: |
| self.close() |
| raise RuntimeError(f"self-test failed: {', '.join(missed) or 'the reading ran out of time'}") |
|
|
| def close(self) -> None: |
| for w in self.workers: |
| w.close() |
| self.java.close() |
|
|
| def read(self, path: str, truth: set[str] | None = None) -> dict: |
| """Read one picture file. `truth` (sha256 hex digests of the content) replaces the consensus, as the |
| measurement scored known codes; the server never passes it.""" |
| t0 = time.monotonic() |
| deadline = t0 + DEADLINE_S |
| worker = self.free.get() |
| try: |
| with tempfile.TemporaryDirectory(prefix="qsd-") as tmp: |
| w, h, frames = worker.stage(path, tmp) |
| java: dict = {"r": [None] * len(frames)} |
|
|
| def zxing_java(): |
| java["r"] = self.java.read([f[3] for f in frames], deadline) |
|
|
| t = threading.Thread(target=zxing_java, daemon=True) |
| t.start() |
| try: |
| pool, partial = worker.decode(frames, deadline) |
| finally: |
| t.join() |
| finally: |
| self.free.put(worker) |
| contents = {(n, k): p for (k, n), p in pool.items()} |
| for k, p in enumerate(java["r"]): |
| contents[("zxing-java", k)] = p or [] |
| partial = partial or p is None |
| if truth is None: |
| content = _consensus(contents) |
| def hit(ps): return content is not None and any(_same(p, content) for p in ps) |
| else: |
| content = None |
| def hit(ps): return any(hashlib.sha256(p).hexdigest() in truth for p in ps) |
| reads = {n: [0] * len(FRAMES) for n in SCANNERS} |
| sizes = [None] * len(FRAMES) |
| for k, (fw, fh, labels, _) in enumerate(frames): |
| for f in labels: |
| fi = FRAMES.index(f) |
| sizes[fi] = [fw, fh] |
| for n in SCANNERS: |
| reads[n][fi] = int(hit(contents.get((n, k), []))) |
| return {"size": [w, h], "frames": sizes, "reads": reads, "content": describe(content), "partial": partial, |
| "ms": round(1000 * (time.monotonic() - t0))} |
|
|
