From 909baec1e764ef844337e0cc6be85964768c83ab Mon Sep 17 00:00:00 2001 From: erik Date: Fri, 9 Oct 2026 12:05:22 +0000 Subject: [PATCH] Desktop app with live scanning and OCR tools/app.py (myscan/app.py, Qt/PySide6) shows the camera and the page as it is stitched, Start/Stop or the scan button (F5), then reads the finished page with RapidOCR (myscan/ocr.py; bundled multilingual models, offline). Lines are outlined on the page by confidence and linked to an editable text pane that can be copied or saved. Existing images can be opened or dropped, so the app is useful without the mouse. The scan loop moves from tools/scan.py to myscan/scanner.py, shared by both. Co-Authored-By: Claude Opus 5.5 --- README.md | 7 + myscan/app.py | 562 ++++++++++++++++++++++++++++++++++++++++++++++ myscan/ocr.py | 96 ++++++++ myscan/scanner.py | 191 ++++++++++++++++ requirements.txt | 4 + tools/app.py | 13 ++ tools/scan.py | 145 +----------- 7 files changed, 880 insertions(+), 138 deletions(-) create mode 100644 myscan/app.py create mode 100644 myscan/ocr.py create mode 100644 myscan/scanner.py create mode 100644 tools/app.py diff --git a/README.md b/README.md index b2448a8..7932464 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,13 @@ Linux driver tools for the Hama myScan (00052343) scanner mouse, a Dacuda design `--view [PORT]` serves a live view at `http://:8000/`: the camera, the page as it is stitched (re-solved every 1.5 s, current position outlined), Start/Stop buttons, PNG/JPEG download of the finished page and a Screenshot button (camera + page + status as one PNG) +- `tools/app.py` — desktop app (Qt/PySide6; Linux, Windows, macOS): live camera and page view while scanning + (Start/Stop or the scan button, F5), then OCR of the finished page with RapidOCR (`myscan/ocr.py`; + multilingual PP-OCRv6 models bundled in the wheel, works offline). Recognised lines are outlined on the page by + confidence; click one to select its text, and the cursor in the text highlights its line. The text is editable + and can be copied or saved as .txt. `python tools/app.py [image]` also opens and reads existing images (Open, or + drag and drop), which is all it does on machines without the mouse +- `myscan/scanner.py` — the scan loop shared by `scan.py` and the app (finds the mouse's `/dev/videoN` itself) - `myscan/live.py`, `myscan/webview.py` — the live preview and its browser viewer (standard library only) - `tools/record.py` — record a scan session: `python tools/record.py -d 25 -o scan_text` - `tools/stitch.py` — stitch a recording: `python tools/stitch.py scan_text` -> `scan_text/page.png` diff --git a/myscan/app.py b/myscan/app.py new file mode 100644 index 0000000..d77bb41 --- /dev/null +++ b/myscan/app.py @@ -0,0 +1,562 @@ +"""Desktop app (Qt, via PySide6): live view while scanning, and OCR of the page. + +Left, the camera and the page as it is being stitched (the same preview as the browser +viewer); after a scan the finished page is loaded and read with RapidOCR (myscan/ocr.py). +Recognised lines are outlined on the page, coloured by confidence; clicking one selects +its text on the right, and moving the cursor in the text highlights its line on the +page. The text can be edited, copied and saved. Images can also be opened or dropped +onto the window, so OCR works on any platform; scanning needs Linux and the mouse. + +Run: python tools/app.py [image] +""" +import os +import sys +import threading +import time + +import cv2 +import numpy as np +from PySide6.QtCore import QObject, QPointF, QRectF, Qt, QThread, Signal +from PySide6.QtGui import (QAction, QBrush, QColor, QGuiApplication, QImage, QKeySequence, QPainter, QPen, + QPixmap, QPolygonF, QTextCursor) +from PySide6.QtWidgets import (QApplication, QFileDialog, QGraphicsPixmapItem, QGraphicsPolygonItem, + QGraphicsScene, QGraphicsView, QLabel, QMainWindow, QMessageBox, QPlainTextEdit, + QSizePolicy, QSplitter, QToolBar, QVBoxLayout, QWidget) + +from .ocr import Ocr, layout + +IMAGE_FILTER = "Images (*.png *.jpg *.jpeg *.tif *.tiff *.bmp *.webp)" + + +def to_qimage(img): + """BGR or grayscale uint8 array -> QImage (a copy).""" + img = np.ascontiguousarray(img) + h, w = img.shape[:2] + fmt = QImage.Format_Grayscale8 if img.ndim == 2 else QImage.Format_BGR888 + return QImage(img.data, w, h, img.strides[0], fmt).copy() + + +def imread(path): + """cv2.imread that also takes non-ASCII paths on Windows.""" + return cv2.imdecode(np.fromfile(path, np.uint8), cv2.IMREAD_COLOR) + + +def imwrite(path, img): + ok, buf = cv2.imencode(os.path.splitext(path)[1] or ".png", img) + if not ok: + raise OSError(f"cannot encode {path}") + buf.tofile(path) + + +def scanner_device(): + """The myScan's capture device, or None (always None off Linux).""" + if not sys.platform.startswith("linux"): + return None + from .scanner import find_video + return find_video() + + +class Bridge(QObject): + """The viewer interface the scan loop talks to (see myscan/scanner.py), turned into + Qt signals; it is called from the scan threads, the slots run on the GUI thread.""" + frame = Signal(QImage) + page = Signal(QImage) + status = Signal(dict) + result = Signal(str) + + def __init__(self): + super().__init__() + self.state = {"state": "idle"} + self.start_event, self.stop_event = threading.Event(), threading.Event() + + def publish(self, name, bgr): + (self.frame if name == "frame" else self.page).emit(to_qimage(bgr)) + + def set_status(self, **kw): + self.state = {**self.state, **kw} + self.status.emit(dict(self.state)) + + def set_result(self, path): + self.result.emit(path) + + def take_start(self): + if self.start_event.is_set(): + self.start_event.clear() + return True + return False + + def take_stop(self): + if self.stop_event.is_set(): + self.stop_event.clear() + return True + return False + + +class ScanThread(QThread): + """Opens the mouse, then scans a page per Start (or scan-button press) until quit.""" + failed = Signal(str) + + def __init__(self, bridge, opts): + super().__init__() + self.bridge, self.opts, self.quit = bridge, opts, threading.Event() + + def run(self): + from .session import quiet_libjpeg + quiet_libjpeg() + from .scanner import Scanner + try: + self.bridge.set_status(state="starting", message="reading calibration") + scanner = Scanner(self.opts, self.bridge) + except Exception as e: + self.failed.emit(f"cannot open the scanner: {e}") + return + try: + while scanner.wait_press(self.quit): + scanner.scan() + except Exception as e: + self.failed.emit(f"scan failed: {e}") + finally: + scanner.close() + + def stop(self): + self.quit.set() + self.bridge.stop_event.set() + self.wait() + + +class OcrThread(QThread): + done = Signal(object, float) # lines, seconds + failed = Signal(str) + + def __init__(self, ocr, img): + super().__init__() + self.ocr, self.img = ocr, img + + def run(self): + t = time.time() + try: + self.done.emit(self.ocr(self.img), time.time() - t) + except Exception as e: + self.failed.emit(str(e)) + + +class LineItem(QGraphicsPolygonItem): + def __init__(self, index, box, score, on_click): + super().__init__(QPolygonF([QPointF(*p) for p in box])) + self.index, self.on_click = index, on_click + c = QColor(46, 160, 67) if score >= 0.9 else QColor(219, 140, 0) if score >= 0.7 else QColor(214, 40, 40) + self.color = c + self.setAcceptHoverEvents(True) + self.setCursor(Qt.PointingHandCursor) + self.setToolTip(f"{score:.0%}") + self.set_active(False) + + def set_active(self, on): + pen = QPen(QColor(30, 110, 230) if on else self.color, 3 if on else 1.5) + pen.setCosmetic(True) + self.setPen(pen) + fill = QColor(30, 110, 230, 70) if on else QColor(self.color.red(), self.color.green(), self.color.blue(), 25) + self.setBrush(QBrush(fill)) + + def mousePressEvent(self, ev): + self.on_click(self.index) + + +class PageView(QGraphicsView): + """Zoomable, pannable page with the OCR line boxes on top. Keeps the page fitted to + the window until the user zooms.""" + lineClicked = Signal(int) + + def __init__(self): + super().__init__() + self.setScene(QGraphicsScene(self)) + self.setRenderHints(QPainter.Antialiasing | QPainter.SmoothPixmapTransform) + self.setDragMode(QGraphicsView.ScrollHandDrag) + self.setTransformationAnchor(QGraphicsView.AnchorUnderMouse) + self.setBackgroundBrush(QColor(30, 32, 36)) + self.pixmap = QGraphicsPixmapItem() + self.pixmap.setTransformationMode(Qt.SmoothTransformation) + self.scene().addItem(self.pixmap) + self.items_, self.active, self.auto_fit = [], None, True + self.boxes_visible = True + + def set_image(self, qimg, keep_view=False): + self.pixmap.setPixmap(QPixmap.fromImage(qimg)) + self.scene().setSceneRect(QRectF(self.pixmap.pixmap().rect())) + if not keep_view: + self.auto_fit = True + if self.auto_fit: + self.fit() + + def set_lines(self, lines): + for it in self.items_: + self.scene().removeItem(it) + self.items_ = [LineItem(i, ln.box, ln.score, self.lineClicked.emit) for i, ln in enumerate(lines)] + for it in self.items_: + it.setVisible(self.boxes_visible) + self.scene().addItem(it) + self.active = None + + def show_boxes(self, on): + self.boxes_visible = on + for it in self.items_: + it.setVisible(on) + + def highlight(self, i): + if self.active is not None and self.active < len(self.items_): + self.items_[self.active].set_active(False) + self.active = i + if i is not None and i < len(self.items_): + self.items_[i].set_active(True) + if not self.auto_fit: + self.ensureVisible(self.items_[i], 40, 40) + + def fit(self): + self.auto_fit = True + if not self.pixmap.pixmap().isNull(): + self.fitInView(self.pixmap, Qt.KeepAspectRatio) + + def zoom(self, f): + self.auto_fit = False + self.scale(f, f) + + def wheelEvent(self, ev): + self.zoom(1.25 ** (ev.angleDelta().y() / 120)) + + def resizeEvent(self, ev): + super().resizeEvent(ev) + if self.auto_fit: + self.fit() + + +class ImageLabel(QLabel): + """A QLabel showing an image scaled to fit, keeping its aspect ratio.""" + + def __init__(self, placeholder): + super().__init__(placeholder) + self.setAlignment(Qt.AlignCenter) + self.setMinimumSize(80, 60) + self.setSizePolicy(QSizePolicy.Ignored, QSizePolicy.Ignored) + self.setStyleSheet("background: #1e2024; color: #8b919a;") + self.img = None + + def set_image(self, qimg): + self.img = qimg + self._update() + + def resizeEvent(self, ev): + super().resizeEvent(ev) + self._update() + + def _update(self): + if self.img is not None: + self.setPixmap(QPixmap.fromImage(self.img).scaled(self.size(), Qt.KeepAspectRatio, + Qt.SmoothTransformation)) + + +def pane(title, widget): + w = QWidget() + lay = QVBoxLayout(w) + lay.setContentsMargins(4, 4, 4, 4) + lay.setSpacing(3) + head = QLabel(title) + head.setStyleSheet("color: #8b919a; font-size: 11px; font-weight: 600; text-transform: uppercase;") + lay.addWidget(head) + lay.addWidget(widget, 1) + return w, head + + +class MainWindow(QMainWindow): + def __init__(self, opts=None): + super().__init__() + self.setWindowTitle("myScan") + self.resize(1400, 850) + self.setAcceptDrops(True) + self.opts = opts + self.ocr = Ocr() + self.image, self.path, self.lines, self.spans = None, None, [], [] + self.ocr_thread, self.scan_thread, self.scanning = None, None, False + self.bridge = Bridge() + self.bridge.frame.connect(self.on_frame) + self.bridge.page.connect(self.on_preview) + self.bridge.status.connect(self.on_status) + self.bridge.result.connect(self.on_result) + self._build() + self.device = scanner_device() + if self.device is None: + self.act_scan.setEnabled(False) + self.act_scan.setToolTip("No myScan found (scanning needs Linux and the mouse plugged in)") + self.camera_pane.hide() + self.set_message("No scanner found: open or drop an image to read it") + else: + self.start_scanner() + + # ---- layout ---- + + def _build(self): + tb = QToolBar("Main") + tb.setMovable(False) + tb.setToolButtonStyle(Qt.ToolButtonTextOnly) + self.addToolBar(tb) + + def act(text, slot, key=None, tip=None, checkable=False): + a = QAction(text, self) + a.triggered.connect(slot) + if key: + a.setShortcut(QKeySequence(key)) + if tip: + a.setToolTip(f"{tip} ({QKeySequence(key).toString(QKeySequence.NativeText)})" if key else tip) + a.setCheckable(checkable) + tb.addAction(a) + return a + + self.act_scan = act("Start scan", self.toggle_scan, "F5", "Start or stop a scan (or press the mouse's scan button)") + tb.addSeparator() + act("Open…", self.open_image, QKeySequence.Open, "Open an image to read") + self.act_save = act("Save image…", self.save_image, QKeySequence.Save, "Save the page image") + tb.addSeparator() + self.act_ocr = act("Read text", self.run_ocr, "Ctrl+R", "Run OCR on the page") + self.act_auto = act("Auto", lambda: None, tip="Read the text after every scan", checkable=True) + self.act_auto.setChecked(True) + self.act_copy = act("Copy text", self.copy_text, "Ctrl+Shift+C", "Copy all text to the clipboard") + self.act_savetxt = act("Save text…", self.save_text, "Ctrl+Shift+S", "Save the text as a .txt file") + tb.addSeparator() + self.act_boxes = act("Boxes", lambda on: self.view.show_boxes(on), "Ctrl+B", "Show recognised lines on the page", True) + self.act_boxes.setChecked(True) + act("Fit", lambda: self.view.fit(), "Ctrl+0", "Fit the page to the window") + act("Zoom +", lambda: self.view.zoom(1.25), QKeySequence.ZoomIn) + act("Zoom −", lambda: self.view.zoom(0.8), QKeySequence.ZoomOut) + for a in (self.act_save, self.act_ocr, self.act_copy, self.act_savetxt): + a.setEnabled(False) + + self.camera = ImageLabel("camera") + self.view = PageView() + self.view.lineClicked.connect(self.select_line) + self.text = QPlainTextEdit() + self.text.setPlaceholderText("Recognised text appears here; scan a page, or open or drop an image.") + self.text.setStyleSheet("font-size: 14px;") + self.text.setLineWrapMode(QPlainTextEdit.NoWrap) # one OCR line per text line + self.text.cursorPositionChanged.connect(self.on_cursor) + self.text.textChanged.connect(self.on_text_edited) + self._setting_text = False + + self.camera_pane, _ = pane("Camera", self.camera) + page_pane, self.page_head = pane("Page", self.view) + text_pane, self.text_head = pane("Text", self.text) + main = QSplitter(Qt.Horizontal) + for w in (self.camera_pane, page_pane, text_pane): + main.addWidget(w) + main.setSizes([170, 750, 480]) + self.setCentralWidget(main) + self.status_label = QLabel() + self.statusBar().addWidget(self.status_label, 1) + + def set_message(self, msg): + self.status_label.setText(msg) + + # ---- scanning ---- + + def start_scanner(self): + self.scan_thread = ScanThread(self.bridge, self.opts) + self.scan_thread.failed.connect(self.on_scan_failed) + self.scan_thread.start() + + def toggle_scan(self): + if not self.act_scan.isEnabled(): + return + (self.bridge.stop_event if self.scanning else self.bridge.start_event).set() + + def on_scan_failed(self, msg): + self.act_scan.setEnabled(False) + self.scanning = False + self.set_message(msg) + QMessageBox.warning(self, "myScan", msg) + + def on_frame(self, qimg): + self.camera.set_image(qimg) + + def on_preview(self, qimg): + if self.scanning: + self.view.set_image(qimg, keep_view=True) + + def on_status(self, s): + state = s.get("state", "idle") + if state == "scanning" and not self.scanning: # a new scan: clear the old page + self.scanning = True + self.view.set_lines([]) + self.view.set_image(QImage()) + self.set_text("", []) + self.lines, self.image, self.path = [], None, None + for a in (self.act_save, self.act_ocr, self.act_copy, self.act_savetxt): + a.setEnabled(False) + self.page_head.setText("Page (scanning)") + elif state in ("idle", "done"): + self.scanning = False + self.act_scan.setText("Stop scan" if state == "scanning" else "Start scan") + self.act_scan.setEnabled(state in ("idle", "scanning", "done")) + parts = [state] + if state in ("scanning", "assembling"): + if "t" in s: + parts.append(f"{s['t']:.1f} s") + if "frames" in s: + parts.append(f"{s['frames']} frames") + if s.get("backlog"): + parts.append(f"backlog {s['backlog']}") + if "loops" in s: + parts.append(f"{s['loops']} loop matches") + if s.get("message"): + parts.append(s["message"]) + self.set_message(" · ".join(parts)) + + def on_result(self, path): + self.scanning = False + self.load(path, auto_ocr=self.act_auto.isChecked()) + + # ---- images ---- + + def load(self, path, auto_ocr=True): + img = imread(path) + if img is None: + QMessageBox.warning(self, "myScan", f"cannot read {path}") + return + self.set_page(img, path) + if auto_ocr: + self.run_ocr() + + def set_page(self, img, path=None): + self.image, self.path, self.lines = img, path, [] + self.view.set_lines([]) + self.view.set_image(to_qimage(img)) + self.set_text("", []) + name = os.path.basename(path) if path else "page" + self.page_head.setText(f"Page — {name} ({img.shape[1]}×{img.shape[0]})") + self.setWindowTitle(f"myScan — {name}") + self.act_save.setEnabled(True) + self.act_ocr.setEnabled(True) + + def open_image(self): + path, _ = QFileDialog.getOpenFileName(self, "Open image", self.default_dir(), IMAGE_FILTER) + if path: + self.load(path) + + def save_image(self): + if self.image is None: + return + base = os.path.splitext(self.path)[0] if self.path else os.path.join(self.default_dir(), "page") + path, _ = QFileDialog.getSaveFileName(self, "Save image", base + ".png", IMAGE_FILTER) + if path: + imwrite(path, self.image) + self.set_message(f"saved {path}") + + def default_dir(self): + if self.path: + return os.path.dirname(self.path) + out = getattr(self.opts, "outdir", None) or "scans" + return out if os.path.isdir(out) else os.getcwd() + + def dragEnterEvent(self, ev): + if ev.mimeData().hasUrls(): + ev.acceptProposedAction() + + def dropEvent(self, ev): + urls = [u.toLocalFile() for u in ev.mimeData().urls() if u.isLocalFile()] + if urls: + self.load(urls[0]) + + # ---- OCR ---- + + def run_ocr(self): + if self.image is None or (self.ocr_thread and self.ocr_thread.isRunning()): + return + self.act_ocr.setEnabled(False) + self.text_head.setText("Text — reading…") + self.set_message("reading text…") + self.ocr_thread = OcrThread(self.ocr, self.image) + self.ocr_thread.done.connect(self.on_ocr) + self.ocr_thread.failed.connect(self.on_ocr_failed) + self.ocr_thread.start() + + def on_ocr(self, lines, secs): + self.act_ocr.setEnabled(self.image is not None) + self.lines = lines + text, spans = layout(lines) + self.view.set_lines(lines) + self.set_text(text, spans) + conf = f", mean confidence {np.mean([ln.score for ln in lines]):.0%}" if lines else "" + self.text_head.setText(f"Text — {len(lines)} lines{conf}") + self.set_message(f"read {len(lines)} lines in {secs:.1f} s") + self.act_copy.setEnabled(bool(text)) + self.act_savetxt.setEnabled(bool(text)) + + def on_ocr_failed(self, msg): + self.act_ocr.setEnabled(self.image is not None) + self.text_head.setText("Text") + self.set_message(f"OCR failed: {msg}") + + def set_text(self, text, spans): + self._setting_text = True + self.text.setPlainText(text) + self._setting_text = False + self.spans = spans + + def on_text_edited(self): + if not self._setting_text and self.spans: # positions no longer match the boxes + self.spans = [] + self.view.highlight(None) + self.act_copy.setEnabled(bool(self.text.toPlainText())) + self.act_savetxt.setEnabled(bool(self.text.toPlainText())) + + def on_cursor(self): + pos = self.text.textCursor().position() + hit = next((i for i, (a, b) in enumerate(self.spans) if a <= pos <= b), None) + if hit is not None or not self.spans: + self.view.highlight(hit) + + def select_line(self, i): + if i >= len(self.spans): + return + a, b = self.spans[i] + c = self.text.textCursor() + c.setPosition(a) + c.setPosition(b, QTextCursor.KeepAnchor) + self.text.setTextCursor(c) + self.text.setFocus() + self.view.highlight(i) + + def copy_text(self): + QGuiApplication.clipboard().setText(self.text.toPlainText()) + self.set_message("text copied to the clipboard") + + def save_text(self): + base = os.path.splitext(self.path)[0] if self.path else os.path.join(self.default_dir(), "page") + path, _ = QFileDialog.getSaveFileName(self, "Save text", base + ".txt", "Text (*.txt)") + if path: + with open(path, "w", encoding="utf-8") as f: + f.write(self.text.toPlainText() + "\n") + self.set_message(f"saved {path}") + + def closeEvent(self, ev): + if self.scan_thread and self.scan_thread.isRunning(): + self.set_message("finishing the scan and restoring mouse mode…") + QApplication.processEvents() + self.scan_thread.stop() # always leave the mouse in mouse mode + if self.ocr_thread: + self.ocr_thread.wait() + super().closeEvent(ev) + + +def main(argv=None, opts=None): + import argparse + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("image", nargs="?", help="image to open and read") + ap.add_argument("-o", "--outdir", default="scans", help="where scans are saved") + ap.add_argument("--video", help="capture device (default: find the myScan)") + ap.add_argument("--gray", action="store_true", help="scan in grayscale") + a = ap.parse_args(argv) + app = QApplication(sys.argv[:1]) + app.setApplicationName("myScan") + w = MainWindow(a) + w.show() + if a.image: + w.load(a.image) + return app.exec() diff --git a/myscan/ocr.py b/myscan/ocr.py new file mode 100644 index 0000000..1ac1311 --- /dev/null +++ b/myscan/ocr.py @@ -0,0 +1,96 @@ +"""OCR of scanned pages with RapidOCR (PaddleOCR's PP-OCR models on ONNX Runtime). + +pip-installable on Linux, Windows and macOS with no system packages; the default +multilingual PP-OCRv6 models ship inside the wheel, so it works offline. They read +Latin script (English, Swedish, German, French, ...) as well as Chinese and Japanese. +Independent of the scanner hardware, like myscan.stitching. + + lines = Ocr()(bgr) # [Line(box, text, score)], box = 4x2 corner points + text, spans = layout(lines) # plain text; spans[i] = (start, end) of lines[i] in it +""" +import threading +from dataclasses import dataclass + +import numpy as np + + +@dataclass +class Line: + box: np.ndarray # 4x2 float corners (clockwise from top left), image pixels + text: str + score: float + + @property + def cy(self): + return float(self.box[:, 1].mean()) + + @property + def left(self): + return float(self.box[:, 0].min()) + + @property + def height(self): + """Text height along the box's left edge (not the axis-aligned extent, which + grows with the slant of a skewed page).""" + return float(np.hypot(*(self.box[3] - self.box[0]))) + + +class Ocr: + """Lazily loads the engine (about 0.3 s) on first use; calls are serialised.""" + + def __init__(self, **params): + self.params = {"Global.log_level": "warning", **params} + self._engine, self._lock = None, threading.Lock() + + def engine(self): + if self._engine is None: + from rapidocr import RapidOCR + self._engine = RapidOCR(params=self.params) + return self._engine + + def __call__(self, bgr): + with self._lock: + r = self.engine()(bgr) + if r.boxes is None: + return [] + return [Line(np.asarray(b, float), t, float(s)) for b, t, s in zip(r.boxes, r.txts, r.scores)] + + +def rows(lines): + """Group lines into visual rows (top to bottom), each sorted left to right. A line + joins a row when its centre is within half a line height of the row's.""" + out = [] + for ln in sorted(lines, key=lambda ln: ln.cy): + for row in out[-3:]: + if abs(ln.cy - np.mean([r.cy for r in row])) < 0.5 * min(ln.height, row[0].height): + row.append(ln) + break + else: + out.append([ln]) + return [sorted(row, key=lambda ln: ln.left) for row in out] + + +def layout(lines): + """Reading-order text: rows joined by newlines, with a blank line where the step to + the next row is over 1.5 times the typical line pitch (a new paragraph); items in a + row are separated by a tab. Returns (text, spans), spans[i] = (start, end) of + lines[i] in text.""" + index = {id(ln): i for i, ln in enumerate(lines)} + rs = rows(lines) + cys = [float(np.mean([ln.cy for ln in row])) for row in rs] + steps = np.diff(cys) + pitch = float(np.median(steps)) if len(steps) else 0.0 + spans, parts, pos = [None] * len(lines), [], 0 + for k, row in enumerate(rs): + if k: + sep = "\n\n" if steps[k - 1] > 1.5 * max(pitch, 0.6 * row[0].height) else "\n" + parts.append(sep) + pos += len(sep) + for j, ln in enumerate(row): + if j: + parts.append("\t") + pos += 1 + spans[index[id(ln)]] = (pos, pos + len(ln.text)) + parts.append(ln.text) + pos += len(ln.text) + return "".join(parts), spans diff --git a/myscan/scanner.py b/myscan/scanner.py new file mode 100644 index 0000000..dfa6de1 --- /dev/null +++ b/myscan/scanner.py @@ -0,0 +1,191 @@ +"""The scan loop shared by tools/scan.py and the desktop app (tools/app.py). + +Scanner opens the mouse and its calibration once; Scanner.scan() captures one page +until the scan button (or the viewer's Stop) is pressed, assembles it and saves it as +/scan-YYYYmmdd-HHMMSS.png. A viewer (webview.Viewer, or the app's Qt bridge) +gets the live frames, the page preview and the status through publish(), set_status(), +set_result(), take_start() and take_stop(). +""" +import csv +import glob +import os +import queue +import threading +import time +from types import SimpleNamespace + +import cv2 + +from . import calib +from .live import LiveSession +from .session import ScanSession +from .stitching import Corrector, PageGraph +from .xu import XU + +OPTIONS = dict(outdir="scans", video=None, target=200, exposure=None, gray=False, no_wb=False, + no_flat=False, no_undistort=False, save_frames=False, max_time=180) + + +def find_video(vid="274f", pid="c001"): + """The capture node (/dev/videoN) of the first attached myScan, or None.""" + for node in sorted(glob.glob("/sys/class/video4linux/video*"), + key=lambda p: int(p.rsplit("video", 1)[1])): + try: + usb = os.path.realpath(os.path.join(node, "device", "..")) + ids = [open(os.path.join(usb, f)).read().strip() for f in ("idVendor", "idProduct")] + index = open(os.path.join(node, "index")).read().strip() + except OSError: + continue + if ids == [vid, pid] and index == "0": + return "/dev/" + os.path.basename(node) + return None + + +def wait_press(x, viewer=None, quit=None, poll=0.05): + """Block until the scan button goes down then up again, or Start is clicked. + Returns False instead if the `quit` event gets set.""" + prev = 1 + while not (quit and quit.is_set()): + if viewer and viewer.take_start(): + return True + b = x.scan_button() + if b and not prev: + break + prev = b + time.sleep(poll) + else: + return False + while x.scan_button(): + time.sleep(poll) + return True + + +def scan_page(x, a, frames_dir=None, corrector=None, gains=None, viewer=None): + """Capture until the button is pressed again. Frames are registered (and saved) on a + worker thread so capture never waits; returns (PageGraph or None, LiveSession or None).""" + g, live, rows, work, saves = None, None, [], queue.Queue(), queue.Queue() + + def worker(): + while (item := work.get()) is not None: + n, im, v = item + f = g.add(im, v) + if live: + live.on_frame(f) + + def saver(): + while (item := saves.get()) is not None: + cv2.imwrite(os.path.join(frames_dir, f"f{item[0]:05d}.png"), item[1]) + th = None + sv = threading.Thread(target=saver, daemon=True) if frames_dir else None + if sv: + sv.start() + if viewer: + viewer.take_stop() # ignore a Stop clicked before the scan started + with ScanSession(a.video, a.target, a.exposure, xu=x, gains=gains) as s: + down, last = 0, 0.0 + for t, n, im, rec in s.frames(color=not a.gray): + if g is None: + g = PageGraph(im.shape, corrector) + if viewer: + live = LiveSession(g, viewer) + th = threading.Thread(target=worker, daemon=True) + th.start() + work.put((n, im, rec.v)) + if sv: + saves.put((n, im)) + if live: + live.raw = im + if frames_dir: + rows.append([round(t, 4), n, rec.counter, *rec.v, rec.b14, rec.b15, rec.b16, rec.flags, s.e6, + round(float(im.mean()), 1), rec.raw.hex()]) + # Stop on a press: two "down" reads in a row (debounce), ignoring the first + # 0.7 s so a bouncing start press can't end the scan. + if n % 2 == 0: + down = down + 1 if x.scan_button() else 0 + if down >= 2 and t > 0.7: + break + if viewer and viewer.take_stop(): + break + if t >= a.max_time: + print("max scan time reached", flush=True) + break + if viewer: + viewer.set_status(t=t, frames=n + 1, backlog=work.qsize()) + if t - last >= 1: + print(f"\r {t:5.1f}s {n + 1} frames ({g.n} registered) E6={s.e6}", end="", flush=True) + last = t + print() + if viewer: + viewer.set_status(state="assembling", message="finishing registration") + if th: + work.put(None) + th.join() + if sv: + saves.put(None) + sv.join() + if frames_dir: # same layout as tools/record.py, so tools/stitch.py can replay it + with open(os.path.join(frames_dir, "records.csv"), "w", newline="") as fh: + w = csv.writer(fh) + w.writerow(["t", "frame", "counter", "v0", "v1", "v2", "v3", "v4", "v5", + "b14", "b15", "b16", "flags", "e6", "brightness", "raw"]) + w.writerows(rows) + return g, live + + +class Scanner: + """The mouse, its calibration and the scan options (OPTIONS; any attribute object + such as an argparse namespace, missing ones take the defaults).""" + + def __init__(self, opts=None, viewer=None): + a = SimpleNamespace(**{**OPTIONS, **(vars(opts) if opts is not None else {})}) + a.video = a.video or find_video() or "/dev/video4" + self.a, self.viewer = a, viewer + os.makedirs(a.outdir, exist_ok=True) + self.x = XU(a.video) + print("reading calibration from the mouse's EEPROM (cached after the first time)", flush=True) + cal = calib.load(self.x) + self.corrector = Corrector(None if a.no_flat else calib.FlatField(cal), + None if a.no_undistort else calib.Undistort(cal)) + self.gains = cal["gains_rgb"] + + def wait_press(self, quit=None): + if self.viewer: + self.viewer.set_status(state="idle", message="press the scan button or Start") + return wait_press(self.x, self.viewer, quit) + + def scan(self): + """Scan one page; returns the saved PNG's path, or None if nothing was captured.""" + a, viewer = self.a, self.viewer + print("scanning - press the scan button again to stop", flush=True) + if viewer: + viewer.set_status(state="scanning", message="press the scan button or Stop to finish") + name = time.strftime("scan-%Y%m%d-%H%M%S") + frames_dir = os.path.join(a.outdir, name) if a.save_frames else None + if frames_dir: + os.makedirs(frames_dir) + g, live = scan_page(self.x, a, frames_dir, self.corrector, self.gains, viewer) + if live: + live.stop_solving() + live.close() + if g is None or g.n < 2: + print("no frames captured") + if viewer: + viewer.set_status(state="idle", message="no frames captured") + return None + print("assembling the page...", flush=True) + t0 = time.time() + if viewer: + viewer.set_status(message="solving") + page = g.image(white_balance=not a.no_wb) + path = os.path.join(a.outdir, name + ".png") + cv2.imwrite(path, page) + print(f"saved {path} ({time.time() - t0:.1f} s; {g.stats})", flush=True) + if viewer: + s = min(1.0, 1600 / max(page.shape[:2])) + viewer.publish("page", cv2.resize(page, None, fx=s, fy=s, interpolation=cv2.INTER_AREA)) + viewer.set_result(path) + viewer.set_status(state="done", message=f"saved {path}", backlog=0) + return path + + def close(self): + self.x.close() diff --git a/requirements.txt b/requirements.txt index 1fb92b7..54cec66 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,7 @@ numpy opencv-python-headless scipy +# desktop app (tools/app.py) and OCR +PySide6-Essentials +rapidocr +onnxruntime diff --git a/tools/app.py b/tools/app.py new file mode 100644 index 0000000..96d06b0 --- /dev/null +++ b/tools/app.py @@ -0,0 +1,13 @@ +#!/usr/bin/env python3 +"""Desktop app: live view while scanning, and OCR of the scanned page (myscan/app.py). + + python tools/app.py [image]""" +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from myscan.app import main # noqa: E402 + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/scan.py b/tools/scan.py index 9930ff2..bbea895 100644 --- a/tools/scan.py +++ b/tools/scan.py @@ -3,116 +3,21 @@ mouse over the page, press it again to stop. Frames are registered while you scan; after the stop press the page is assembled from all of them at once (see myscan/stitching/pagegraph.py) and saved as /scan-YYYYmmdd-HHMMSS.png. The mouse goes -back to mouse mode between scans (and on Ctrl-C). +back to mouse mode between scans (and on Ctrl-C). The loop itself is myscan/scanner.py. With --view, a browser viewer (myscan/webview.py) shows the camera and the page as it is being stitched, and has Start/Stop buttons that work like the scan button.""" import argparse -import csv import os -import queue import sys -import threading -import time sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from myscan.session import ScanSession, quiet_libjpeg # noqa: E402 +from myscan.session import quiet_libjpeg # noqa: E402 quiet_libjpeg() -import cv2 # noqa: E402 -from myscan import calib # noqa: E402 -from myscan.live import LiveSession # noqa: E402 -from myscan.stitching import Corrector, PageGraph # noqa: E402 -from myscan.xu import XU # noqa: E402 - - -def wait_press(x, viewer=None, poll=0.05): - """Block until the scan button goes down then up again, or Start is clicked.""" - prev = 1 - while True: - if viewer and viewer.take_start(): - return - b = x.scan_button() - if b and not prev: - break - prev = b - time.sleep(poll) - while x.scan_button(): - time.sleep(poll) - - -def scan_page(x, a, frames_dir=None, corrector=None, gains=None, viewer=None): - """Capture until the button is pressed again. Frames are registered (and saved) on a - worker thread so capture never waits; returns (PageGraph or None, LiveSession or None).""" - g, live, rows, work, saves = None, None, [], queue.Queue(), queue.Queue() - - def worker(): - while (item := work.get()) is not None: - n, im, v = item - f = g.add(im, v) - if live: - live.on_frame(f) - - def saver(): - while (item := saves.get()) is not None: - cv2.imwrite(os.path.join(frames_dir, f"f{item[0]:05d}.png"), item[1]) - th = None - sv = threading.Thread(target=saver, daemon=True) if frames_dir else None - if sv: - sv.start() - if viewer: - viewer.take_stop() # ignore a Stop clicked before the scan started - with ScanSession(a.video, a.target, a.exposure, xu=x, gains=gains) as s: - down, last = 0, 0.0 - for t, n, im, rec in s.frames(color=not a.gray): - if g is None: - g = PageGraph(im.shape, corrector) - if viewer: - live = LiveSession(g, viewer) - th = threading.Thread(target=worker, daemon=True) - th.start() - work.put((n, im, rec.v)) - if sv: - saves.put((n, im)) - if live: - live.raw = im - if frames_dir: - rows.append([round(t, 4), n, rec.counter, *rec.v, rec.b14, rec.b15, rec.b16, rec.flags, s.e6, - round(float(im.mean()), 1), rec.raw.hex()]) - # Stop on a press: two "down" reads in a row (debounce), ignoring the first - # 0.7 s so a bouncing start press can't end the scan. - if n % 2 == 0: - down = down + 1 if x.scan_button() else 0 - if down >= 2 and t > 0.7: - break - if viewer and viewer.take_stop(): - break - if t >= a.max_time: - print("max scan time reached", flush=True) - break - if viewer: - viewer.set_status(t=t, frames=n + 1, backlog=work.qsize()) - if t - last >= 1: - print(f"\r {t:5.1f}s {n + 1} frames ({g.n} registered) E6={s.e6}", end="", flush=True) - last = t - print() - if viewer: - viewer.set_status(state="assembling", message="finishing registration") - if th: - work.put(None) - th.join() - if sv: - saves.put(None) - sv.join() - if frames_dir: # same layout as tools/record.py, so tools/stitch.py can replay it - with open(os.path.join(frames_dir, "records.csv"), "w", newline="") as fh: - w = csv.writer(fh) - w.writerow(["t", "frame", "counter", "v0", "v1", "v2", "v3", "v4", "v5", - "b14", "b15", "b16", "flags", "e6", "brightness", "raw"]) - w.writerows(rows) - return g, live +from myscan.scanner import Scanner # noqa: E402 def main(): @@ -120,7 +25,7 @@ def main(): ap.add_argument("-o", "--outdir", default="scans") ap.add_argument("--once", action="store_true", help="exit after one page") ap.add_argument("--now", action="store_true", help="start the first scan immediately") - ap.add_argument("--video", default="/dev/video4") + ap.add_argument("--video", help="capture device (default: find the myScan, else /dev/video4)") ap.add_argument("--target", type=float, default=200, help="auto-exposure target for paper (95th pct of brightest channel)") ap.add_argument("--exposure", type=int, help="fixed E6 value instead of auto-exposure") ap.add_argument("--gray", action="store_true") @@ -132,56 +37,20 @@ def main(): ap.add_argument("--view", type=int, nargs="?", const=8000, metavar="PORT", help="serve a live browser view on PORT (default 8000)") a = ap.parse_args() - os.makedirs(a.outdir, exist_ok=True) viewer = None if a.view is not None: from myscan.webview import Viewer viewer = Viewer(a.view) print("live view: " + " ".join(viewer.urls()), flush=True) - x = XU(a.video) - print("reading calibration from the mouse's EEPROM (cached after the first time)", flush=True) - cal = calib.load(x) - corrector = Corrector(None if a.no_flat else calib.FlatField(cal), - None if a.no_undistort else calib.Undistort(cal)) - gains = cal["gains_rgb"] + scanner = Scanner(a, viewer) first = True try: while True: if not (first and a.now): print("press the scan button to start (Ctrl-C to quit)", flush=True) - if viewer: - viewer.set_status(state="idle", message="press the scan button or Start") - wait_press(x, viewer) + scanner.wait_press() first = False - print("scanning - press the scan button again to stop", flush=True) - if viewer: - viewer.set_status(state="scanning", message="press the scan button or Stop to finish") - name = time.strftime("scan-%Y%m%d-%H%M%S") - frames_dir = os.path.join(a.outdir, name) if a.save_frames else None - if frames_dir: - os.makedirs(frames_dir) - g, live = scan_page(x, a, frames_dir, corrector, gains, viewer) - if live: - live.stop_solving() - live.close() - if g is None or g.n < 2: - print("no frames captured") - if viewer: - viewer.set_status(state="idle", message="no frames captured") - else: - print("assembling the page...", flush=True) - t0 = time.time() - if viewer: - viewer.set_status(message="solving") - page = g.image(white_balance=not a.no_wb) - path = os.path.join(a.outdir, name + ".png") - cv2.imwrite(path, page) - print(f"saved {path} ({time.time() - t0:.1f} s; {g.stats})", flush=True) - if viewer: - s = min(1.0, 1600 / max(page.shape[:2])) - viewer.publish("page", cv2.resize(page, None, fx=s, fy=s, interpolation=cv2.INTER_AREA)) - viewer.set_result(path) - viewer.set_status(state="done", message=f"saved {path}", backlog=0) + scanner.scan() if a.once: break except KeyboardInterrupt: