#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Weishi No. 1000 -> Logitech webcam -> OCR -> CSV (English, adjustable logging). Python 3.10+ / Windows 11. See README.md for installation and controls. """ from __future__ import annotations import argparse import csv import json import re import shutil import sys import time from collections import Counter, defaultdict, deque from concurrent.futures import ThreadPoolExecutor from datetime import datetime from pathlib import Path try: import cv2 import numpy as np import pytesseract except ImportError as exc: print(f"Missing Python package: {exc}") print("Install with: python -m pip install -r requirements.txt") raise SystemExit(1) ROOT = Path(__file__).resolve().parent CONFIG_FILE = ROOT / "weishi_installningar.json" DATA_DIR = ROOT / "matningar" DEBUG_DIR = ROOT / "felsokning" FIELDS = ("rate", "amplitude", "beat_error") NAMES = {"rate": "Rate (s/d)", "amplitude": "Amplitude (deg)", "beat_error": "Beat error (ms)"} COLORS = {"rate": (85, 220, 90), "amplitude": (255, 180, 40), "beat_error": (220, 90, 255)} VIEW_MAX_WIDTH = 1250 VIEW_MAX_HEIGHT = 760 OCR_PERIOD_SECONDS = 2.5 DEFAULT_LOG_INTERVAL = 10 MIN_LOG_INTERVAL = 10 MAX_LOG_INTERVAL = 3600 LATEST_RATE_DEBUG = DEBUG_DIR / "senaste_rate_debug.png" LATEST_AMPLITUDE_DEBUG = DEBUG_DIR / "senaste_amplitude_debug.png" LATEST_BEAT_ERROR_DEBUG = DEBUG_DIR / "senaste_beat_error_debug.png" # Keep the most recent COMPLETED OCR pass, even if another pass is running. # Missing or rejected OCR values from that pass are represented by 99. # Do not discard an accepted reading just because OCR took over 10 seconds. def find_tesseract() -> str | None: candidates = [ __import__("os").environ.get("TESSERACT_CMD"), shutil.which("tesseract"), r"C:\Program Files\Tesseract-OCR\tesseract.exe", r"C:\Program Files (x86)\Tesseract-OCR\tesseract.exe", ] for item in candidates: if item and (Path(item).is_file() or shutil.which(str(item))): return str(item) return None def parse_number(field: str, raw: str) -> tuple[int | float | None, bool]: """Parse strict numeric ROI. Returns value, whether BE decimal was inferred.""" s = raw.strip().replace("−", "-").replace("–", "-") s = s.replace(" ", "").replace("\n", "").replace(",", ".") # Optional units if user did not crop them out perfectly. s = re.sub(r"(?i)(s/d|ms|°|º)", "", s) if field in ("rate", "amplitude"): # OCR may confuse e.g. + and whitespace; do NOT guess missing negatives. pattern = r"[+-]?\d{1,3}" if field == "rate" else r"\d{2,3}" if not re.fullmatch(pattern, s): return None, False n = int(s) if field == "rate" and -999 <= n <= 999: return n, False if field == "amplitude" and 100 <= n <= 360: return n, False elif field == "beat_error": if re.fullmatch(r"\d[.]\d", s): n = float(s) return (n, False) if 0.0 <= n <= 9.9 else (None, False) # On pixelated LCDs OCR sometimes drops the decimal point, e.g. 04 -> 0.4. # Only a two-digit result can be reconstructed, and it receives less weight. if re.fullmatch(r"\d{2}", s): return int(s) / 10.0, True return None, False def make_variants(crop: np.ndarray, field: str | None = None) -> dict[str, np.ndarray]: if crop.size == 0: return {} # The fine LCD grid can make OCR misread "-16" as "-7". # For Rate, smooth the grid at ORIGINAL resolution before upscaling. # These variants have been checked against a real camera screenshot. if field == "rate": gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY) denoised = { "rate_gauss7": cv2.GaussianBlur(gray, (7, 7), 0), "rate_median5": cv2.medianBlur(gray, 5), "rate_gauss9": cv2.GaussianBlur(gray, (9, 9), 0), } result = {} for name, img in denoised.items(): _, bw = cv2.threshold(img, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) result[name] = cv2.copyMakeBorder( bw, 15, 15, 15, 15, cv2.BORDER_CONSTANT, value=255) # Extra upscaling can help when digits are smaller in the camera image. result["rate_gauss7_2x"] = cv2.resize( result["rate_gauss7"], None, fx=2, fy=2, interpolation=cv2.INTER_NEAREST) return result if field == "amplitude": # Smooth the LCD grid BEFORE upscaling. Earlier versions # often found the right digits but rejected them due to too few votes. gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY) prefiltered = { "amp_gauss3": cv2.GaussianBlur(gray, (3, 3), 0), "amp_gauss5": cv2.GaussianBlur(gray, (5, 5), 0), "amp_median3": cv2.medianBlur(gray, 3), "amp_gauss7": cv2.GaussianBlur(gray, (7, 7), 0), } result = {} for name, img in prefiltered.items(): _, bw = cv2.threshold(img, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) # First reduce grid noise, then upscale 2x using nearest-neighbor. bw = cv2.resize(bw, None, fx=2, fy=2, interpolation=cv2.INTER_NEAREST) result[name] = cv2.copyMakeBorder( bw, 20, 20, 20, 20, cv2.BORDER_CONSTANT, value=255) return result if field == "beat_error": # The decimal point in 0.1 can disappear with 4x upscaling. # Keep Beat Error OCR at original resolution and smooth the grid lightly. gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY) imgs = { "be_gray": gray, "be_median3": cv2.medianBlur(gray, 3), "be_gauss3": cv2.GaussianBlur(gray, (3, 3), 0), "be_gauss5": cv2.GaussianBlur(gray, (5, 5), 0), } result = {} for name, img in imgs.items(): _, bw = cv2.threshold(img, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) result[name] = cv2.copyMakeBorder( bw, 15, 15, 15, 15, cv2.BORDER_CONSTANT, value=255) return result b = crop[:, :, 0] # Weishi usually has a blue backlit LCD. g = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY) # Upscale BEFORE thresholding to make dot-matrix digits easier to recognize. scale = 4 if max(crop.shape[:2]) < 600 else 3 b = cv2.resize(b, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC) g = cv2.resize(g, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC) b = cv2.GaussianBlur(b, (3, 3), 0) g = cv2.GaussianBlur(g, (3, 3), 0) cl = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)).apply(b) _, blue_bw = cv2.threshold(b, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) _, blue_clahe_bw = cv2.threshold(cl, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) _, gray_bw = cv2.threshold(g, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) adaptive = cv2.adaptiveThreshold(g, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 31, 8) imgs = { "bla_otsu": blue_bw, "bla_kontrast": blue_clahe_bw, "gra_otsu": gray_bw, "adaptiv": adaptive, "bla_inverterad": 255 - blue_bw, } # White margin keeps digits at the edge from getting clipped by Tesseract. return {k: cv2.copyMakeBorder(v, 24, 24, 24, 24, cv2.BORDER_CONSTANT, value=255) for k, v in imgs.items()} def detect_rate_sign(crop: np.ndarray) -> dict: """Find + or - as a separate LCD glyph before OCR reads the digits. A Rate with a missing sign is intentionally rejected: on the Weishi LCD, OCR can miss '-' and turn a negative measurement into a positive one. The Rate selection must contain the FULL sign and all digits, but no 's/d'. """ empty = {"sign": None, "digit_x": None, "reason": "Sign cannot be identified"} if crop.size == 0 or crop.shape[0] < 25 or crop.shape[1] < 45: return empty gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY) # Blurring in ORIGINAL pixel size reduces the blue LCD's fine square grid. blur = cv2.GaussianBlur(gray, (7, 7), 0) _, mask = cv2.threshold(blur, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU) num, _, stats, _ = cv2.connectedComponentsWithStats(mask) height, width = mask.shape min_area = max(18, int(height * width * 0.0012)) components = [tuple(map(int, r)) for r in stats[1:] if int(r[4]) >= min_area and int(r[2]) >= 5 and int(r[3]) >= 3] if not components: return empty max_height = max(c[3] for c in components) # Digits are normally the tallest connected shapes inside the selection. digits = sorted((c for c in components if c[3] >= 0.78 * max_height and c[3] >= 0.32 * height), key=lambda c: c[0]) if not digits: return empty first_digit = digits[0] digit_x = first_digit[0] digit_height = float(first_digit[3]) digit_center_y = first_digit[1] + first_digit[3] / 2 candidates = [] for x, y, w, h, area in components: if x + w > digit_x - 3: continue if w < max(9, int(digit_height * 0.20)): continue # Isolated plus/minus signs are vertically centered against the digits. center_error = abs(y + h / 2 - digit_center_y) / digit_height if center_error > 0.24: continue ratio = h / digit_height aspect = w / max(1, h) if ratio <= 0.30 and aspect >= 1.8: candidates.append((x, "-", center_error, ratio, aspect)) elif 0.33 <= ratio <= 0.85 and 0.60 <= aspect <= 1.65: # A plus has a tall vertical stroke AND a horizontal crossbar. part = mask[y:y+h, x:x+w] mid_y = part[int(h * .33):int(h * .67), :] mid_x = part[:, int(w * .33):int(w * .67)] horizontal = (mid_y.sum(axis=1) / 255).max(initial=0) vertical = (mid_x.sum(axis=0) / 255).max(initial=0) if horizontal >= 0.60 * w and vertical >= 0.60 * h: candidates.append((x, "+", center_error, ratio, aspect)) if not candidates: return {"sign": None, "digit_x": digit_x, "reason": "No clear + or - sign before the digits"} candidates.sort(key=lambda a: a[0]) sign = candidates[0][1] return {"sign": sign, "digit_x": digit_x, "reason": "Sign detected separately from OCR"} def recognize(field: str, crop: np.ndarray) -> dict: sign_check = detect_rate_sign(crop) if field == "rate" else None variants = make_variants(crop, field) if field == "rate": # Critical fix: OCR reads DIGITS ONLY, never the + or - sign. # Otherwise the plus sign can look like a four (e.g., +8 -> +48). # The sign is already detected separately by detect_rate_sign(). variants = {} digit_x = sign_check["digit_x"] if digit_x is not None: x = max(0, int(digit_x)) if x < crop.shape[1] - 10: digit_variants = make_variants(crop[:, x:], "rate") variants = {"siffra_" + name: img for name, img in digit_variants.items()} # Doubling the image size with light blur can clarify a zero digit. variants["siffra_gauss9_2x"] = cv2.resize( digit_variants["rate_gauss9"], None, fx=2, fy=2, interpolation=cv2.INTER_NEAREST) votes: Counter = Counter() supporting_variants: dict[int | float, set[str]] = defaultdict(set) variants_count: Counter = Counter() traces: list[dict] = [] samples: dict[str, list[str]] = defaultdict(list) # Rate signs/BE decimal are essential. Each box should include numbers only, # not the neighboring label, graph or unit. whitelist = "0123456789" if field in ("rate", "amplitude") else "0123456789.," for name, img in variants.items(): # PSM 7: single line; PSM 13: raw line. Better than page OCR. for psm in ((7, 8, 13) if field in ("rate", "beat_error") else (7, 13)): config = f"--oem 3 --psm {psm} -c tessedit_char_whitelist={whitelist}" try: data = pytesseract.image_to_data( img, lang="eng", config=config, output_type=pytesseract.Output.DICT, timeout=3) tokens = [] confidences = [] for token, conf in zip(data["text"], data["conf"]): token = token.strip() if token: tokens.append(token) try: confidences.append(max(0.0, float(conf))) except (ValueError, TypeError): pass raw = "".join(tokens) confidence = (sum(confidences) / len(confidences)) if confidences else 0 except (pytesseract.TesseractError, RuntimeError) as exc: raw, confidence = "", 0 traces.append({"variant": name, "psm": psm, "error": str(exc)}) continue value, inferred = parse_number(field, raw) if field == "rate": if value is not None and sign_check["sign"] is not None: # Plus/minus comes from separate shape analysis; OCR reads only digits. value = abs(int(value)) * (1 if sign_check["sign"] == "+" else -1) elif value == 0 and sign_check["sign"] is None: # The Weishi usually displays zero without a sign. value = 0 else: # Reject other unsigned values rather than guessing the sign. value = None traces.append({"variant": name, "psm": psm, "ocr": raw, "confidence": round(confidence, 1), "parsed": value, "decimal_inferred": inferred}) if value is None: continue # A missed decimal is accepted only if several variants agree. score = max(0.2, confidence / 100.0) * (0.58 if inferred else 1.0) votes[value] += score variants_count[value] += 1 supporting_variants[value].add(name) samples[value].append(raw) if votes: best, score = votes.most_common(1)[0] count = variants_count[best] # Rate requires at least TWO preprocessing variants (not just different PSMs). # Other fields use their own acceptance rules. if field == "rate": accepted = ((sign_check["sign"] is not None or best == 0) and len(supporting_variants[best]) >= 2 and score >= 0.8) elif field == "amplitude": # Amplitude is always unsigned (100..360). # Tesseract may report confidence=0 even for clearly readable LCD digits. # One preprocessing variant is not enough: a 3 can otherwise # be misread as 9. Require TWO independent preprocessing variants. # Accept low-confidence votes when the variants agree. accepted = len(supporting_variants[best]) >= 2 else: accepted = count >= 2 or score >= 0.90 return {"value": best if accepted else None, "proposed": best, "votes": count, "variants": len(supporting_variants[best]), "score": round(score, 2), "raw": samples[best][0], "trace": traces, "sign_check": sign_check} return {"value": None, "proposed": None, "votes": 0, "variants": 0, "score": 0, "raw": "", "trace": traces, "sign_check": sign_check} def crop_area(frame: np.ndarray, box: list[float]) -> np.ndarray: h, w = frame.shape[:2] x1 = max(0, min(w - 1, int(round(box[0] * w)))) y1 = max(0, min(h - 1, int(round(box[1] * h)))) x2 = max(x1 + 1, min(w, int(round((box[0] + box[2]) * w)))) y2 = max(y1 + 1, min(h, int(round((box[1] + box[3]) * h)))) return frame[y1:y2, x1:x2].copy() def empty_field(reason: str = "No current OCR reading") -> dict: """Return a missing field in the same format as recognize().""" return {"value": None, "raw": "", "proposed": None, "votes": 0, "variants": 0, "score": 0.0, "trace": [], "error": reason} def analyze(frame: np.ndarray, regions: dict) -> dict: """Read each field independently so one OCR failure does not discard the others.""" started = time.monotonic() values = {} debug_rate_crop = None debug_amplitude_crop = None debug_beat_error_crop = None for field in FIELDS: if field not in regions: values[field] = empty_field("ROI not selected") continue try: crop = crop_area(frame, regions[field]) if field == 'rate': debug_rate_crop = crop.copy() elif field == 'amplitude': debug_amplitude_crop = crop.copy() elif field == 'beat_error': debug_beat_error_crop = crop.copy() values[field] = recognize(field, crop) except Exception as exc: values[field] = empty_field(str(exc)) return {"time": started, "wall_time": datetime.now().astimezone().isoformat( timespec="seconds"), "fields": values, "regions": regions, "debug_rate_crop": debug_rate_crop, "debug_amplitude_crop": debug_amplitude_crop, "debug_beat_error_crop": debug_beat_error_crop} def get_view_size(frame: np.ndarray) -> tuple[int, int]: h, w = frame.shape[:2] scale = min(VIEW_MAX_WIDTH / w, VIEW_MAX_HEIGHT / h, 1.0) return max(1, int(round(w * scale))), max(1, int(round(h * scale))) def select_region(frame: np.ndarray, field: str) -> list[float] | None: w, h = get_view_size(frame) display = cv2.resize(frame, (w, h), interpolation=cv2.INTER_AREA) title = f"SELECT DIGITS ONLY: {NAMES[field]} - ENTER=confirm, ESC=cancel" cv2.namedWindow(title, cv2.WINDOW_NORMAL) cv2.resizeWindow(title, w, h) try: x, y, bw, bh = cv2.selectROI(title, display, showCrosshair=True, fromCenter=False) finally: cv2.destroyWindow(title) if bw < 5 or bh < 5: return None # Normalized crop coordinates remain valid across webcam resolutions. return [x / w, y / h, bw / w, bh / h] def format_preview_value(field: str, value) -> str: if value is None: return "---" if field == "rate": if isinstance(value, int) and value > 0: return f"+{value}" return str(value) if field == "beat_error" and isinstance(value, float): return f"{value:.1f}".replace(".", ",") return str(value) def build_rate_debug_panel(rate_crop: np.ndarray | None, rate_result: dict | None) -> np.ndarray: panel_w, panel_h = 330, 170 panel = np.full((panel_h, panel_w, 3), 22, dtype=np.uint8) cv2.rectangle(panel, (0, 0), (panel_w - 1, panel_h - 1), (80, 80, 80), 1) cv2.putText(panel, "RATE debug", (12, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.70, (255, 255, 255), 2, cv2.LINE_AA) crop_area_img = panel[34:126, 10:200] crop_area_img[:] = (35, 35, 35) if rate_crop is not None and getattr(rate_crop, 'size', 0) > 0: ch, cw = rate_crop.shape[:2] scale = min(crop_area_img.shape[1] / max(1, cw), crop_area_img.shape[0] / max(1, ch)) nw = max(1, int(round(cw * scale))) nh = max(1, int(round(ch * scale))) resized = cv2.resize(rate_crop, (nw, nh), interpolation=cv2.INTER_AREA) x = (crop_area_img.shape[1] - nw) // 2 y = (crop_area_img.shape[0] - nh) // 2 crop_area_img[y:y + nh, x:x + nw] = resized else: cv2.putText(panel, "No Rate image", (18, 82), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (180, 180, 180), 1, cv2.LINE_AA) value = None if not rate_result else rate_result.get('value') proposed = None if not rate_result else rate_result.get('proposed') raw = "" if not rate_result else str(rate_result.get('raw') or "") score = 0.0 if not rate_result else float(rate_result.get('score') or 0.0) votes = 0 if not rate_result else int(rate_result.get('votes') or 0) variants = 0 if not rate_result else int(rate_result.get('variants') or 0) sign = None if rate_result and isinstance(rate_result.get('sign_check'), dict): sign = rate_result['sign_check'].get('sign') display = format_preview_value('rate', value) if value is not None else '99' color = (110, 240, 120) if value is not None else (0, 200, 255) cv2.putText(panel, f"OCR: {display}", (214, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.80, color, 2, cv2.LINE_AA) cv2.putText(panel, f"Raw: {raw or '-'}", (214, 88), cv2.FONT_HERSHEY_SIMPLEX, 0.53, (235, 235, 235), 1, cv2.LINE_AA) cv2.putText(panel, f"Candidate: {proposed if proposed is not None else '-'}", (214, 110), cv2.FONT_HERSHEY_SIMPLEX, 0.50, (220, 220, 220), 1, cv2.LINE_AA) sign_text = sign if sign is not None else ("unsigned 0" if value == 0 else "?") cv2.putText(panel, f"Sign: {sign_text}", (214, 132), cv2.FONT_HERSHEY_SIMPLEX, 0.50, (220, 220, 220), 1, cv2.LINE_AA) cv2.putText(panel, f"Score:{score:.2f} V:{votes} Var:{variants}", (12, 150), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (180, 220, 255), 1, cv2.LINE_AA) return panel def save_latest_rate_debug(panel: np.ndarray | None): if panel is None: return DEBUG_DIR.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(LATEST_RATE_DEBUG), panel) def build_amplitude_debug_panel(amplitude_crop: np.ndarray | None, amplitude_result: dict | None) -> np.ndarray: """Show the exact Amplitude crop used by the last completed OCR pass.""" panel_w, panel_h = 330, 170 panel = np.full((panel_h, panel_w, 3), 22, dtype=np.uint8) cv2.rectangle(panel, (0, 0), (panel_w - 1, panel_h - 1), (80, 80, 80), 1) cv2.putText(panel, "AMPLITUDE debug", (12, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.70, (255, 255, 255), 2, cv2.LINE_AA) crop_display = panel[34:126, 10:200] crop_display[:] = (35, 35, 35) if amplitude_crop is not None and getattr(amplitude_crop, 'size', 0) > 0: ch, cw = amplitude_crop.shape[:2] scale = min(crop_display.shape[1] / max(1, cw), crop_display.shape[0] / max(1, ch)) nw = max(1, int(round(cw * scale))) nh = max(1, int(round(ch * scale))) resized = cv2.resize(amplitude_crop, (nw, nh), interpolation=cv2.INTER_AREA) x = (crop_display.shape[1] - nw) // 2 y = (crop_display.shape[0] - nh) // 2 crop_display[y:y + nh, x:x + nw] = resized else: cv2.putText(panel, "No Amplitude image", (17, 82), cv2.FONT_HERSHEY_SIMPLEX, 0.46, (180, 180, 180), 1, cv2.LINE_AA) value = None if not amplitude_result else amplitude_result.get('value') proposed = None if not amplitude_result else amplitude_result.get('proposed') raw = "" if not amplitude_result else str(amplitude_result.get('raw') or "") score = 0.0 if not amplitude_result else float(amplitude_result.get('score') or 0.0) votes = 0 if not amplitude_result else int(amplitude_result.get('votes') or 0) variants = 0 if not amplitude_result else int(amplitude_result.get('variants') or 0) display = str(value) if value is not None else "99" color = (110, 240, 120) if value is not None else (0, 200, 255) cv2.putText(panel, f"OCR: {display}", (214, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.72, color, 2, cv2.LINE_AA) cv2.putText(panel, f"Raw: {raw or '-'}"[:18], (214, 88), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (235, 235, 235), 1, cv2.LINE_AA) cv2.putText(panel, f"Candidate: {proposed if proposed is not None else '-'}", (214, 110), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (220, 220, 220), 1, cv2.LINE_AA) cv2.putText(panel, "Accepted" if value is not None else "Rejected", (214, 132), cv2.FONT_HERSHEY_SIMPLEX, 0.48, color, 1, cv2.LINE_AA) cv2.putText(panel, f"Score:{score:.2f} V:{votes} Var:{variants}", (12, 150), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (180, 220, 255), 1, cv2.LINE_AA) return panel def save_latest_amplitude_debug(panel: np.ndarray | None): if panel is None: return DEBUG_DIR.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(LATEST_AMPLITUDE_DEBUG), panel) def build_beat_error_debug_panel(beat_crop: np.ndarray | None, beat_result: dict | None) -> np.ndarray: panel_w, panel_h = 330, 170 panel = np.full((panel_h, panel_w, 3), 22, dtype=np.uint8) cv2.rectangle(panel, (0, 0), (panel_w - 1, panel_h - 1), (80, 80, 80), 1) cv2.putText(panel, "BEAT ERROR debug", (12, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.62, (255, 255, 255), 2, cv2.LINE_AA) crop_display = panel[34:126, 10:200] crop_display[:] = (35, 35, 35) if beat_crop is not None and getattr(beat_crop, 'size', 0) > 0: ch, cw = beat_crop.shape[:2] scale = min(crop_display.shape[1] / max(1, cw), crop_display.shape[0] / max(1, ch)) nw = max(1, int(round(cw * scale))) nh = max(1, int(round(ch * scale))) resized = cv2.resize(beat_crop, (nw, nh), interpolation=cv2.INTER_AREA) x = (crop_display.shape[1] - nw) // 2 y = (crop_display.shape[0] - nh) // 2 crop_display[y:y + nh, x:x + nw] = resized else: cv2.putText(panel, "No Beat image", (24, 82), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (180, 180, 180), 1, cv2.LINE_AA) value = None if not beat_result else beat_result.get('value') proposed = None if not beat_result else beat_result.get('proposed') raw = "" if not beat_result else str(beat_result.get('raw') or "") score = 0.0 if not beat_result else float(beat_result.get('score') or 0.0) votes = 0 if not beat_result else int(beat_result.get('votes') or 0) variants = 0 if not beat_result else int(beat_result.get('variants') or 0) display = format_preview_value('beat_error', value) if value is not None else '99' color = (110, 240, 120) if value is not None else (0, 200, 255) cv2.putText(panel, f"OCR: {display}", (214, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.72, color, 2, cv2.LINE_AA) cv2.putText(panel, f"Raw: {raw or '-'}"[:18], (214, 88), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (235, 235, 235), 1, cv2.LINE_AA) cv2.putText(panel, f"Candidate: {proposed if proposed is not None else '-'}", (214, 110), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (220, 220, 220), 1, cv2.LINE_AA) cv2.putText(panel, "Accepted" if value is not None else "Rejected", (214, 132), cv2.FONT_HERSHEY_SIMPLEX, 0.48, color, 1, cv2.LINE_AA) cv2.putText(panel, f"Score:{score:.2f} V:{votes} Var:{variants}", (12, 150), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (180, 220, 255), 1, cv2.LINE_AA) return panel def save_latest_beat_error_debug(panel: np.ndarray | None): if panel is None: return DEBUG_DIR.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(LATEST_BEAT_ERROR_DEBUG), panel) def draw_preview(frame: np.ndarray, regions: dict, result: dict | None, auto_save: bool, save_period: float, focus: str, info: str, rate_panel: np.ndarray | None = None, amplitude_panel: np.ndarray | None = None, beat_error_panel: np.ndarray | None = None, interval_entry: str | None = None, interval_error: str = "") -> np.ndarray: image = frame.copy() h, w = image.shape[:2] for field in FIELDS: box = regions.get(field) if not box: continue x = int(box[0] * w) y = int(box[1] * h) x2 = int((box[0] + box[2]) * w) y2 = int((box[1] + box[3]) * h) color = COLORS[field] cv2.rectangle(image, (x, y), (x2, y2), color, 3) # Keep the field label away from the OCR crop; only drawn on preview. val = None if not result else result["fields"][field]["value"] display = format_preview_value(field, val) short = field.upper() + ": " + (display if val is not None else "---") ty = y - 9 if y >= 22 else y2 + 21 cv2.putText(image, short, (x, ty), cv2.FONT_HERSHEY_SIMPLEX, 0.70, color, 2, cv2.LINE_AA) view_w, view_h = get_view_size(image) image = cv2.resize(image, (view_w, view_h), interpolation=cv2.INTER_AREA) # HUD occupies the bottom of the preview, without changing captured pixels. overlay = image.copy() strip_h = min(94, view_h) cv2.rectangle(overlay, (0, view_h - strip_h), (view_w, view_h), (18, 18, 18), -1) cv2.addWeighted(overlay, 0.78, image, 0.22, 0, image) base = view_h - strip_h + 23 cv2.putText(image, "1/2/3 = select ROI | R = all ROIs | D = save debug images", (12, base), cv2.FONT_HERSHEY_SIMPLEX, 0.54, (255, 255, 255), 1, cv2.LINE_AA) cv2.putText(image, "S = save now | A = auto save | I = interval | F = autofocus | +/- = focus | Q = quit", (12, base + 23), cv2.FONT_HERSHEY_SIMPLEX, 0.49, (230, 230, 230), 1, cv2.LINE_AA) state = f"Weishi Logger | Auto save: {'ON' if auto_save else 'OFF'} / {save_period:g}s Focus: {focus} {info}" cv2.putText(image, state[:130], (12, base + 46), cv2.FONT_HERSHEY_SIMPLEX, 0.50, (130, 235, 255), 1, cv2.LINE_AA) # Three debug panels along the bottom: Rate | Amplitude | Beat Error. panels = [panel for panel in (rate_panel, amplitude_panel, beat_error_panel) if panel is not None and getattr(panel, 'size', 0) > 0] if panels: gap = 8 margin = 10 base_pw = max(p.shape[1] for p in panels) base_ph = max(p.shape[0] for p in panels) total_w = sum(p.shape[1] for p in panels) + gap * (len(panels) - 1) available_w = max(1, view_w - 2 * margin) available_h = max(1, strip_h - 8) common_scale = min(1.0, available_w / max(1, total_w), available_h / max(1, base_ph)) scaled_panels = [] for panel in panels: if common_scale < 1.0: panel = cv2.resize(panel, (max(1, int(round(panel.shape[1] * common_scale))), max(1, int(round(panel.shape[0] * common_scale)))), interpolation=cv2.INTER_AREA) scaled_panels.append(panel) total_scaled_w = sum(p.shape[1] for p in scaled_panels) + gap * (len(scaled_panels) - 1) start_x = max(margin, (view_w - total_scaled_w) // 2) for idx, panel in enumerate(scaled_panels): ph, pw = panel.shape[:2] x = start_x + sum(p.shape[1] for p in scaled_panels[:idx]) + gap * idx y = view_h - ph - 4 image[y:y + ph, x:x + pw] = panel # Native OpenCV numeric input: no extra GUI toolkit is required. # Leave the camera and all three debug panels running while editing. if interval_entry is not None: x0, y0 = 12, 12 box_w = min(610, max(0, view_w - 24)) box_h = min(130, max(0, view_h - 24)) if box_w > 140 and box_h > 90: cv2.rectangle(image, (x0, y0), (x0 + box_w, y0 + box_h), (22, 22, 22), -1) cv2.rectangle(image, (x0, y0), (x0 + box_w, y0 + box_h), (100, 240, 240), 2) cv2.putText(image, "LOGGING INTERVAL (10 - 3600 seconds)", (x0 + 12, y0 + 28), cv2.FONT_HERSHEY_SIMPLEX, 0.62, (255, 255, 255), 2, cv2.LINE_AA) cv2.putText(image, f"Seconds: {interval_entry or '_'}", (x0 + 12, y0 + 66), cv2.FONT_HERSHEY_SIMPLEX, 0.82, (120, 240, 255), 2, cv2.LINE_AA) cv2.putText(image, interval_error or "Type number, ENTER = apply, ESC = cancel, BACKSPACE = erase", (x0 + 12, y0 + 105), cv2.FONT_HERSHEY_SIMPLEX, 0.43, (80, 170, 255) if interval_error else (210, 210, 210), 1, cv2.LINE_AA) return image def read_config() -> dict: """Load existing ROI settings and optional interval (old files still work).""" try: data = json.loads(CONFIG_FILE.read_text(encoding="utf-8")) return data if isinstance(data, dict) else {} except (FileNotFoundError, ValueError, OSError, TypeError): return {} def write_config(data: dict) -> None: CONFIG_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8") def load_regions() -> dict: config = read_config() regions = config.get("regions") if not isinstance(regions, dict): return {} return {k: v for k, v in regions.items() if k in FIELDS and isinstance(v, list) and len(v) == 4 and all(isinstance(n, (float, int)) and not isinstance(n, bool) for n in v) and all(0 <= n <= 1 for n in v)} def save_regions(regions: dict): data = read_config() data["regions"] = regions write_config(data) def is_valid_log_interval(value) -> bool: return isinstance(value, int) and not isinstance(value, bool) and ( MIN_LOG_INTERVAL <= value <= MAX_LOG_INTERVAL) def load_log_interval() -> int: """Default to 10 seconds for previous installs and corrupt settings.""" interval = read_config().get("log_interval_seconds", DEFAULT_LOG_INTERVAL) return interval if is_valid_log_interval(interval) else DEFAULT_LOG_INTERVAL def save_log_interval(seconds: int) -> None: if not is_valid_log_interval(seconds): raise ValueError("Logging interval must be between 10 and 3600 seconds") data = read_config() data["log_interval_seconds"] = seconds write_config(data) def snapshot_for_csv(result: dict | None, regions: dict, now: float | None = None) -> dict: """Return the three values from the latest completed OCR pass. Rate OCR already uses several independent image variants and a separate sign check. No extra stability gate is needed before saving an accepted Rate reading. If no new OCR pass has finished when a row is due, repeat the latest completed reading. Only fields that failed in that pass become 99. """ values = {field: 99 for field in FIELDS} if not result or result.get("regions") != regions: return values fields = result.get("fields", {}) for field in FIELDS: value = (fields.get(field) or {}).get("value") if value is None or isinstance(value, bool) or not isinstance(value, (int, float)): continue # Keep only values that recognize() has actually accepted. # Do not add a separate rate_stable gate, which previously produced R=99 # even when the green Rate label displayed a valid OCR reading. values[field] = value return values def store_csv(values: dict, timestamp: str | None = None) -> Path: """Always save a CSV row, using 99 for missing or rejected readings. timestamp is the CSV logging time, not the latest OCR capture time. """ DATA_DIR.mkdir(parents=True, exist_ok=True) now = datetime.now().astimezone() path = DATA_DIR / f"weishi_{now:%Y-%m-%d}.csv" is_new = not path.exists() or path.stat().st_size == 0 def get(field: str): number = values.get(field) return 99 if number is None else number beat_error = get("beat_error") row = [timestamp or now.isoformat(timespec="seconds"), get("rate"), get("amplitude"), "99" if beat_error == 99 else f"{beat_error:.1f}".replace(".", ",")] with path.open("a", newline="", encoding="utf-8-sig" if is_new else "utf-8") as fp: writer = csv.writer(fp, delimiter=";") if is_new: writer.writerow(["timestamp", "rate_s_per_day", "amplitude_deg", "beat_error_ms"]) writer.writerow(row) return path def save_debug(frame: np.ndarray, regions: dict, result: dict | None) -> Path: directory = DEBUG_DIR / datetime.now().strftime("%Y%m%d_%H%M%S_%f") directory.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(directory / "kamera_180grader.jpg"), frame) for field in FIELDS: if field in regions: crop = crop_area(frame, regions[field]) cv2.imwrite(str(directory / f"{field}_original.png"), crop) for variant, image in make_variants(crop, field).items(): cv2.imwrite(str(directory / f"{field}_{variant}.png"), image) json_result = result if isinstance(result, dict): json_result = dict(result) rate_crop = json_result.pop('debug_rate_crop', None) if isinstance(rate_crop, np.ndarray) and rate_crop.size > 0: cv2.imwrite(str(directory / 'rate_senaste_ocr_crop.png'), rate_crop) amplitude_crop = json_result.pop('debug_amplitude_crop', None) if isinstance(amplitude_crop, np.ndarray) and amplitude_crop.size > 0: cv2.imwrite(str(directory / 'amplitude_senaste_ocr_crop.png'), amplitude_crop) beat_crop = json_result.pop('debug_beat_error_crop', None) if isinstance(beat_crop, np.ndarray) and beat_crop.size > 0: cv2.imwrite(str(directory / 'beat_error_senaste_ocr_crop.png'), beat_crop) (directory / "ocr_resultat.json").write_text( json.dumps(json_result, ensure_ascii=False, indent=2, default=str), encoding="utf-8") return directory def open_camera(index: int): if sys.platform == "win32": cap = cv2.VideoCapture(index, cv2.CAP_DSHOW) if not cap.isOpened(): cap.release() cap = cv2.VideoCapture(index, cv2.CAP_MSMF) else: cap = cv2.VideoCapture(index) if not cap.isOpened(): raise RuntimeError(f"Cannot open camera {index}. Try --camera 1 or 2.") cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1920) cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080) cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) return cap def main(): p = argparse.ArgumentParser(description="Read a Weishi No. 1000 using a webcam and log data to CSV") p.add_argument("--camera", type=int, default=0, help="Camera index (0, 1, 2 ...)") p.add_argument("--image", type=Path, help="Use a still image instead of a webcam") p.add_argument("--no-rotate", action="store_true", help="Do NOT rotate the image by 180 degrees") p.add_argument("--interval", type=int, default=None, help="Override saved log interval (10-3600 seconds)") p.add_argument("--focus", type=int, default=216, help="Manual focus 0..255 (default: 216); -1 = leave unchanged") args = p.parse_args() if args.interval is not None and not is_valid_log_interval(args.interval): p.error("--interval must be an integer between 10 and 3600 seconds") if args.focus != -1 and not 0 <= args.focus <= 255: p.error("--focus must be -1 or between 0 and 255") tesseract = find_tesseract() if not tesseract: print("\nTesseract OCR not found. Install Tesseract for Windows and restart.") print("https://github.com/UB-Mannheim/tesseract/wiki") print("Alternatively, set TESSERACT_CMD to the full path of tesseract.exe") return 2 pytesseract.pytesseract.tesseract_cmd = tesseract print(f"OCR: {tesseract}") regions = load_regions() log_interval = args.interval if args.interval is not None else load_log_interval() interval_entry = None interval_error = "" print(f"Logging interval: {log_interval} seconds (press I in the camera window to change)") # Image testing must not silently add repeated rows to the real measurement log. auto_save = args.image is None # Independent CSV timer: logging does NOT depend on OCR completion speed. # Log a row even when OCR fails or the readings have not changed. next_save_at = time.monotonic() + log_interval last_started = 0.0 last_result = None last_rate_panel = None last_amplitude_panel = None last_beat_error_panel = None info = "Press R to select all three ROIs" if len(regions) < 3 else "Ready to read" focus_value = 40.0 auto_focus = True cap = None static = None if args.image: static = cv2.imread(str(args.image)) if static is None: print(f"Could not read image: {args.image}") return 2 else: try: cap = open_camera(args.camera) except RuntimeError as exc: print(exc) return 2 print("Camera started. If the image is blurry, press F for autofocus or +/- to adjust manual focus.") time.sleep(0.7) focus_value = cap.get(cv2.CAP_PROP_FOCUS) if focus_value < 0 or focus_value > 255: focus_value = 40.0 if args.focus >= 0: cap.set(cv2.CAP_PROP_AUTOFOCUS, 0) if cap.set(cv2.CAP_PROP_FOCUS, args.focus): focus_value = float(args.focus) auto_focus = False print(f"Manual focus set to {args.focus}") else: print(f"Camera could not set focus to {args.focus}; adjust with +/-") print("R = select all ROIs, 1/2/3 = one ROI, I = logging interval, D = debug, S = save, Q = quit") print(f"CSV files: {DATA_DIR}") cv2.namedWindow("Weishi Logger", cv2.WINDOW_NORMAL) executor = ThreadPoolExecutor(max_workers=1) future = None # Cache the latest *rotated* frame only. Crops are taken from the original pixels. try: while True: if static is not None: frame = static.copy() else: ok, frame = cap.read() if not ok: print("Camera returned no image. Exiting.") break if not args.no_rotate: frame = cv2.rotate(frame, cv2.ROTATE_180) if future is not None and future.done(): try: candidate = future.result() # Do not reuse values from obsolete ROI selections. last_result = candidate if candidate.get("regions") == regions else None if last_result: last_rate_panel = build_rate_debug_panel( last_result.get('debug_rate_crop'), (last_result.get('fields') or {}).get('rate')) save_latest_rate_debug(last_rate_panel) last_amplitude_panel = build_amplitude_debug_panel( last_result.get('debug_amplitude_crop'), (last_result.get('fields') or {}).get('amplitude')) save_latest_amplitude_debug(last_amplitude_panel) last_beat_error_panel = build_beat_error_debug_panel( last_result.get('debug_beat_error_crop'), (last_result.get('fields') or {}).get('beat_error')) save_latest_beat_error_debug(last_beat_error_panel) else: last_rate_panel = None last_amplitude_panel = None last_beat_error_panel = None except Exception as exc: print(f"OCR error: {exc}") last_result = None last_rate_panel = None last_amplitude_panel = None last_beat_error_panel = None info = "OCR error (CSV will still log 99)" future = None if last_result: parts = last_result["fields"] text_values = [] saved_values = snapshot_for_csv(last_result, regions) for field, title in (("rate", "R"), ("amplitude", "A"), ("beat_error", "BE")): text_values.append(f"{title}={saved_values[field]}") info = " ".join(text_values) + " (99 = missing/uncertain)" else: info = "No valid OCR image (CSV will still be saved)" now = time.monotonic() # IMPORTANT: save CSV here, NOT inside the OCR completion block. # This logs on the chosen schedule, even without OCR results. if auto_save and now >= next_save_at: fresh = snapshot_for_csv(last_result, regions, now) stamp = datetime.now().astimezone().isoformat(timespec="seconds") path = store_csv(fresh, stamp) print(f"{stamp} | Saved {path.name}: " f"Rate={fresh['rate']}, Amplitude={fresh['amplitude']}, " f"Beat error={fresh['beat_error']}") # Keep the selected schedule despite variable OCR processing time. # A blocked GUI can delay a row, but missing OCR results must not # prevent it from being written. Skip deadlines already passed. while next_save_at <= now: next_save_at += log_interval if (len(regions) > 0 and future is None and now - last_started >= OCR_PERIOD_SECONDS): future = executor.submit(analyze, frame.copy(), dict(regions)) last_started = now focus_text = "auto" if auto_focus else f"manual {focus_value:.0f}" shown = draw_preview(frame, regions, last_result, auto_save, log_interval, focus_text, info, last_rate_panel, last_amplitude_panel, last_beat_error_panel, interval_entry, interval_error) cv2.imshow("Weishi Logger", shown) key = cv2.waitKey(30) & 0xFF if key == 255: continue if interval_entry is not None: if key == 27: # ESC: cancel without changing anything. interval_entry = None interval_error = "" info = f"Logging interval unchanged: {log_interval}s" elif key in (13, 10): # ENTER: apply if valid. seconds = int(interval_entry) if interval_entry else 0 if not is_valid_log_interval(seconds): interval_error = "Enter a whole number between 10 and 3600" else: log_interval = seconds next_save_at = time.monotonic() + log_interval interval_entry = None interval_error = "" try: save_log_interval(log_interval) info = f"Logging interval saved: {log_interval}s" except OSError as exc: info = f"Interval set to {log_interval}s (settings save failed)" print(f"Could not save settings: {exc}") print(f"Logging interval: {log_interval} seconds") elif key in (8, 127): # BACKSPACE. interval_entry = interval_entry[:-1] interval_error = "" elif ord("0") <= key <= ord("9") and len(interval_entry) < 4: interval_entry += chr(key) interval_error = "" continue # Never treat number keys as ROI shortcuts while editing. if key in (ord("q"), ord("Q"), 27): break if key in (ord("i"), ord("I")): interval_entry = "" interval_error = "" print(f"Enter logging interval (10-3600 seconds); current {log_interval}s") continue if key in (ord("r"), ord("R")): # Freeze one frame, mark three independent numeric-only boxes. for f in FIELDS: box = select_region(frame, f) if box is not None: regions[f] = box save_regions(regions) print(f"Selected: {NAMES[f]}") last_result = None last_rate_panel = None last_amplitude_panel = None last_beat_error_panel = None last_started = 0 info = "ROIs saved" elif key in (ord("1"), ord("2"), ord("3")): f = FIELDS[int(chr(key)) - 1] box = select_region(frame, f) if box is not None: regions[f] = box save_regions(regions) last_result = None last_rate_panel = None last_amplitude_panel = None last_started = 0 info = f"ROI updated: {f}" elif key in (ord("d"), ord("D")): folder = save_debug(frame, regions, last_result) info = f"Debug images saved: {folder.name}" print(f"Debug images: {folder}") elif key in (ord("s"), ord("S")): values = snapshot_for_csv(last_result, regions) path = store_csv(values) info = "Saved manually (99 = missing): " + path.name print(f"Saved manually: {path}, {values}") elif key in (ord("a"), ord("A")): auto_save = not auto_save if auto_save: next_save_at = time.monotonic() + log_interval info = "Auto save ON" if auto_save else "Auto save OFF" elif key in (ord("f"), ord("F")): if cap is not None: auto_focus = not auto_focus success = cap.set(cv2.CAP_PROP_AUTOFOCUS, 1 if auto_focus else 0) info = ("Autofocus ON" if auto_focus else "Autofocus OFF") + ( "" if success else " (camera may not support this)") print(info) elif key in (ord("+"), ord("="), ord("-"), ord("_")): if cap is not None: if auto_focus: cap.set(cv2.CAP_PROP_AUTOFOCUS, 0) auto_focus = False delta = 5 if key in (ord("+"), ord("=")) else -5 focus_value = float(max(0, min(255, focus_value + delta))) success = cap.set(cv2.CAP_PROP_FOCUS, focus_value) info = f"Focus {focus_value:.0f}" + ( "" if success else " (camera may not support this)") print(info) finally: # Do not wait for full OCR chain on window closing. executor.shutdown(wait=False, cancel_futures=True) if cap is not None: cap.release() cv2.destroyAllWindows() return 0 if __name__ == "__main__": raise SystemExit(main())