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: