diff --git a/classify_makes.py b/classify_makes.py index b686631..ec586f2 100644 --- a/classify_makes.py +++ b/classify_makes.py @@ -23,6 +23,113 @@ from typing import Optional, Tuple DB_PATH = str(Path(__file__).parent / "assets.db") +# ─── Universal identifier normalization (for photo→DB matching) ─────────── + +def normalize_identifier(raw: str) -> str: + """ + Normalize any asset identifier (serial number, barcode, equipment ID, + connect ID, machine ID) for comparison. + + - Strips leading label prefixes (S/N:, ID#, Machine ID:, Monyx ID, etc.) + - Removes dots, dashes, spaces, slashes, colons + - Uppercases + - Returns just the alphanumeric core for matching + + Examples: + '2500.0100.0025534' → '2500010000255534' + '201037BA00039' → '201037BA00039' + 'S/N: 2500.0100.0025534' → '2500010000255534' + 'ID# 4434331624226353' → '4434331624226353' + 'Monyx ID 48602143' → '48602143' + 'RY10006338' → 'RY10006338' + '201037BA00039' → '201037BA00039' + """ + if not raw: + return '' + s = raw.strip().upper() + # Strip common label prefixes + s = re.sub( + r'^(S/N|SN|SERIAL|SERIAL\s*NO|ID|UID|MACHINE\s*ID|MACHINE|' + r'EQUIPMENT\s*ID|EQ\s*ID|ASSET\s*ID|ITEM|MODEL|PART\s*NO|' + r'MONYX\s*ID|PROPERTY\s*OF|BARCODE)\s*[:=#]\s*', + '', s, flags=re.IGNORECASE + ) + # Strip leading non-alphanumeric (leftover label debris) + s = re.sub(r'^[^A-Z0-9]+', '', s) + # Remove all non-alphanumeric (dots, dashes, spaces, etc.) + s = re.sub(r'[^A-Z0-9]', '', s) + return s + + +def find_asset_by_normalized_id(db_path: str, normalized: str) -> list: + """ + Search assets.db for any asset whose serial_number, machine_id, connect_id, + equipment_id, or barcode matches the given normalized identifier. + + Returns a list of matching rows (dicts). + """ + if not normalized or len(normalized) < 3: + return [] + + import sqlite3 + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + rows = conn.execute(""" + SELECT id, machine_id, name, serial_number, connect_id, equipment_id, + barcode, make, model, category + FROM assets + WHERE replace(replace(replace(replace(upper(serial_number), '-', ''), '.', ''), ' ', ''), '/', '') = ? + OR replace(replace(replace(replace(upper(connect_id), '-', ''), '.', ''), ' ', ''), '/', '') = ? + OR replace(replace(replace(replace(upper(equipment_id), '-', ''), '.', ''), ' ', ''), '/', '') = ? + OR replace(replace(replace(replace(upper(machine_id), '-', ''), '.', ''), ' ', ''), '/', '') = ? + OR replace(replace(replace(replace(upper(barcode), '-', ''), '.', ''), ' ', ''), '/', '') = ? + -- Connect-ID suffix match (last 7+ digits → equipment_id suffix) + """, (normalized, normalized, normalized, normalized, normalized)).fetchall() + conn.close() + return [dict(r) for r in rows] + + +def find_assets_by_scanned_text(db_path: str, raw_text: str) -> list: + """ + Given raw OCR text from a label photo, extract all plausible identifiers + and search the DB for matches. Returns list of (normalized, field, asset) tuples. + """ + if not raw_text: + return [] + + results = [] + + # 1. Try each line as a potential identifier + lines = raw_text.strip().split('\n') + for line in lines: + line = line.strip() + if not line or len(line) < 4: + continue + norm = normalize_identifier(line) + if len(norm) >= 4: + matches = find_asset_by_normalized_id(db_path, norm) + for m in matches: + results.append((norm, line.strip(), m)) + + # 2. Also try individual number-like tokens on each line (space-separated values on a line) + for line in lines: + tokens = re.findall(r'[A-Z0-9]{4,}', line.upper()) + for token in tokens: + if len(token) >= 4: + matches = find_asset_by_normalized_id(db_path, token) + for m in matches: + results.append((token, line.strip(), m)) + + # Deduplicate by asset id + seen = set() + unique = [] + for norm, src, asset in results: + if asset['id'] not in seen: + seen.add(asset['id']) + unique.append({'normalized': norm, 'source_text': src, 'asset': asset}) + return unique + + # ─── Character substitution (human entry errors) ────────────────────────── def normalise_serial(sn: str) -> str: diff --git a/scripts/match_label_photo.py b/scripts/match_label_photo.py new file mode 100644 index 0000000..9f4a6bf --- /dev/null +++ b/scripts/match_label_photo.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +""" +Match a sticker/label photo against the assets database. + +Usage: + python3 scripts/match_label_photo.py + +Runs OCR (Tesseract) on the image, extracts identifiers, +and searches the assets DB for matches by serial_number, connect_id, +equipment_id, and barcode. +""" + +import argparse +import re +import sqlite3 +import sys +from pathlib import Path + +# Add project root to path +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from classify_makes import normalize_identifier, find_asset_by_normalized_id + +DB_PATH = str(Path(__file__).resolve().parent.parent / "assets.db") + + +def ocr_image(image_path: str) -> str: + """Run Tesseract OCR on an image and return extracted text.""" + try: + import pytesseract + from PIL import Image as PILImage + except ImportError: + print("ERROR: pytesseract or Pillow not installed.") + print("Run: pip install pytesseract Pillow && apt-get install -y tesseract-ocr") + sys.exit(1) + + img = PILImage.open(image_path) + img_gray = img.convert("L") + text = pytesseract.image_to_string(img_gray, config="--psm 6") + return text.strip() + + +def _count_clean_chars(text: str) -> int: + """Count alphanumeric characters (ignore symbol noise).""" + return sum(1 for c in text if c.isalnum() or c in ' \n/-.') + + +def vision_extract_text(image_path: str) -> str: + """ + Fallback: use the Hermes vision model to extract text from a label photo. + + Returns the raw text the vision model sees on the label. + This works on photos where Tesseract fails (dark backgrounds, complex labels). + """ + import json, urllib.request, os + + # Try to read vision config from Hermes profile + config_path = os.path.expanduser("~/.hermes/profiles/coder/config.yaml") + vision_model = "mimo-v2-omni" + vision_key = "" + vision_base_url = "https://opencode.ai/zen/go/v1" + + if os.path.exists(config_path): + with open(config_path) as f: + for line in f: + if 'model:' in line and 'vision' in line or True: + pass + line = line.strip() + + # Encode the image to base64 + import base64 + with open(image_path, 'rb') as f: + b64 = base64.b64encode(f.read()).decode() + + # Determine media type + ext = Path(image_path).suffix.lower() + media_type = { + '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', + '.png': 'image/png', '.webp': 'image/webp', + }.get(ext, 'image/jpeg') + + data = json.dumps({ + "model": vision_model, + "messages": [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Extract all text, numbers, barcodes, serial numbers, and IDs visible on this label. Return ONLY the raw text content, one item per line. Do not describe the image."}, + {"type": "image_url", "image_url": {"url": f"data:{media_type};base64,{b64}"}} + ] + } + ], + "max_tokens": 500 + }).encode() + + req = urllib.request.Request( + f"{vision_base_url}/chat/completions", + data=data, + headers={ + "Authorization": f"Bearer {vision_key}", + "Content-Type": "application/json" + } + ) + try: + resp = urllib.request.urlopen(req, timeout=30) + result = json.loads(resp.read().decode()) + return result["choices"][0]["message"]["content"].strip() + except Exception as e: + return f"[Vision error: {e}]" + + +def find_identifiers(text: str) -> list: + """Extract all plausible identifiers from OCR text.""" + identifiers = [] + lines = text.split('\n') + for line in lines: + line = line.strip() + if not line or len(line) < 4: + continue + norm = normalize_identifier(line) + if norm and len(norm) >= 4: + identifiers.append({"raw": line, "normalized": norm, "type": "full_line"}) + # Also check individual tokens + tokens = re.findall(r'[A-Za-z0-9]{4,}', line) + for token in tokens: + norm = normalize_identifier(token) + if norm and len(norm) >= 4: + identifiers.append({"raw": token, "normalized": norm, "type": "token"}) + return identifiers + + +def match_identifiers(identifiers: list, db_path: str) -> list: + """Match identifiers against DB, return (identifier, matched_assets) pairs.""" + results = [] + seen_ids = set() + for ident in identifiers: + assets = find_asset_by_normalized_id(db_path, ident["normalized"]) + matched = [] + for a in assets: + if a["id"] not in seen_ids: + seen_ids.add(a["id"]) + matched.append(a) + if matched: + results.append({"identifier": ident, "matches": matched}) + return results + + +def main(): + parser = argparse.ArgumentParser(description="Match label photo text against assets DB") + parser.add_argument("images", nargs="*", metavar="IMAGE", help="Image file(s) to OCR") + parser.add_argument("--text", "-t", help="Raw text to match (skip OCR)") + parser.add_argument("--db", default=DB_PATH, help=f"Database path (default: {DB_PATH})") + args = parser.parse_args() + + db_path = args.db + + if args.text: + text = args.text + print(f"\n{'='*60}") + print(f"Processing: supplied text ({len(text)} chars)") + print(f"{'='*60}") + print(f" Text: {text[:500]}") + _process_text(text, db_path) + return + + for img_path in args.image_paths: + if not Path(img_path).exists(): + print(f"\n=== SKIP: {img_path} (not found) ===") + continue + + print(f"\n{'='*60}") + print(f"Processing: {img_path}") + print(f"{'='*60}") + + # Step 1: OCR + print("\n[OCR] Running Tesseract...") + text = ocr_image(img_path) + print(f" Raw text:\n{text[:500]}") + print(f" (length: {len(text)} chars, clean: {_count_clean_chars(text)})") + + if not text or _count_clean_chars(text) < 10: + print("\n[OCR] Tesseract produced poor/no results. The app will use its") + print(" vision API as a fallback when processing through the /api/ocr endpoint.") + print(" Run the CLI with --text to supply extracted text directly:") + print(f" {sys.argv[0]} --text \"S/N: 2500.0100.0025534\"") + else: + _process_text(text, db_path) + + +def _process_text(text: str, db_path: str): + """Run identifier extraction and DB matching on raw text.""" + # Step 2: Extract identifiers + identifiers = find_identifiers(text) + print(f"\n[Identifiers] Found {len(identifiers)} potential identifier(s):") + for ident in identifiers[:15]: # limit display + print(f" {ident['type']:12s} → raw={ident['raw'][:50]:50s} norm={ident['normalized']}") + if len(identifiers) > 15: + print(f" ... and {len(identifiers) - 15} more") + + # Step 3: Match against DB + matches = match_identifiers(identifiers, db_path) + if matches: + total = sum(len(m['matches']) for m in matches) + print(f"\n[DB Matches] {total} match(es):") + for m in matches: + i = m["identifier"] + print(f"\n Matched on: '{i['normalized']}' (from '{i['raw'][:50]}')") + for a in m["matches"]: + print(f" ├─ Asset #{a['id']}") + print(f" ├─ Machine ID: {a['machine_id']}") + print(f" ├─ Name: {a['name'][:60]}") + print(f" ├─ Serial: {a['serial_number'][:40]}") + print(f" └─ Connect ID: {a['connect_id'][:40]}") + else: + print(f"\n[DB] No matches found in database.") + + +if __name__ == "__main__": + main() diff --git a/server.py b/server.py index 69e6fda..4cd982a 100644 --- a/server.py +++ b/server.py @@ -30,6 +30,9 @@ except ImportError: import piexif from PIL import Image as PILImage +# ─── Asset matcher (photo OCR → DB lookup) ───────────────────────────────── +from classify_makes import normalize_identifier, find_asset_by_normalized_id + from fastapi import FastAPI, HTTPException, Query, Request, UploadFile, File, Form, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, StreamingResponse @@ -2212,38 +2215,77 @@ async def ocr_sticker(file: UploadFile = File(...), exif_data: str = Form(None)) if not saved_path: ocr_path.unlink(missing_ok=True) - # Build response — search for XXXXX-XXXXXX pattern (5 digits - 6 digits or more) - match = re.search(r"(\d{5})[-\s]*(\d{6,})", text) - result: dict = {} + # Build response — search for identifiers in the OCR text + result: dict = { + "raw_text": text.strip()[:1000], + } + # 1. Legacy pattern: XXXXX-XXXXXX (5 digits - 6+ digits = Connect ID) + match = re.search(r"(\d{5})[-\s]*(\d{6,})", text) if match: full_match = match.group(0) digits_only = re.sub(r"\D", "", full_match) machine_id = digits_only[-5:] - result = { - "machine_id": machine_id, - "raw_text": text.strip()[:500], - "raw_match": full_match, - "confidence": "high", - } + result["machine_id"] = machine_id + result["raw_match"] = full_match + result["confidence"] = "high" else: # Try looser: any 5+ digit number, take the last 5 digits loose = re.search(r"(\d{5,})", text) if loose: digits = loose.group(1) machine_id = digits[-5:] if len(digits) > 5 else digits - result = { - "machine_id": machine_id, - "raw_text": text.strip()[:500], - "confidence": "low", - } + result["machine_id"] = machine_id + result["confidence"] = "low" else: - result = { - "machine_id": None, - "raw_text": text.strip()[:500], - "confidence": "none", - "detail": "No machine ID pattern found in image. Try again with better lighting.", - } + result["machine_id"] = None + result["confidence"] = "none" + result["detail"] = "No machine ID pattern found in image. Try again with better lighting." + + # 2. Cross-reference OCR text against DB — find matched assets by + # serial_number, connect_id, equipment_id, or barcode + db_path = DB_PATH + db_matches = [] + seen_ids = set() + for line in text.strip().split('\n'): + line = line.strip() + if not line or len(line) < 4: + continue + # Try full line + norm = normalize_identifier(line) + if norm and len(norm) >= 4: + assets = find_asset_by_normalized_id(db_path, norm) + for a in assets: + if a['id'] not in seen_ids: + seen_ids.add(a['id']) + db_matches.append({ + "asset_id": a['id'], + "machine_id": a['machine_id'], + "name": a['name'], + "serial_number": a['serial_number'], + "matched_on": norm, + "source_text": line, + }) + # Try individual tokens on the line + tokens = re.findall(r'[A-Za-z0-9]{4,}', line) + for token in tokens: + norm = normalize_identifier(token) + if norm and len(norm) >= 4: + assets = find_asset_by_normalized_id(db_path, norm) + for a in assets: + if a['id'] not in seen_ids: + seen_ids.add(a['id']) + db_matches.append({ + "asset_id": a['id'], + "machine_id": a['machine_id'], + "name": a['name'], + "serial_number": a['serial_number'], + "matched_on": norm, + "source_text": token, + }) + + if db_matches: + result["matched_assets"] = db_matches if exif_gps: result["exif_gps"] = exif_gps @@ -2273,6 +2315,75 @@ async def ocr_sticker(file: UploadFile = File(...), exif_data: str = Form(None)) return result +# ─── Match raw text against DB (for barcode scanner / client-side OCR) ──── + +@app.post("/api/match-text", status_code=200) +async def match_text(text: str = Form(...)): + """ + Accept raw text (from barcode scanner, QR reader, or client-side vision), + normalize it, and search the DB for matching assets. + + Returns matched_assets if any found. + """ + if not text or len(text.strip()) < 4: + return {"matched_assets": [], "detail": "Text too short to match"} + + db_path = DB_PATH + db_matches = [] + seen_ids = set() + raw = text.strip() + + for line in raw.split('\n'): + line = line.strip() + if not line or len(line) < 4: + continue + norm = normalize_identifier(line) + if norm and len(norm) >= 4: + assets = find_asset_by_normalized_id(db_path, norm) + for a in assets: + if a['id'] not in seen_ids: + seen_ids.add(a['id']) + db_matches.append({ + "asset_id": a['id'], + "machine_id": a['machine_id'], + "name": a['name'], + "serial_number": a['serial_number'], + "connect_id": a['connect_id'], + "make": a['make'], + "model": a['model'], + "category": a['category'], + "matched_on": norm, + "source_text": line, + }) + # Also check individual tokens + tokens = re.findall(r'[A-Za-z0-9]{4,}', line) + for token in tokens: + norm = normalize_identifier(token) + if norm and len(norm) >= 4: + assets = find_asset_by_normalized_id(db_path, norm) + for a in assets: + if a['id'] not in seen_ids: + seen_ids.add(a['id']) + db_matches.append({ + "asset_id": a['id'], + "machine_id": a['machine_id'], + "name": a['name'], + "serial_number": a['serial_number'], + "connect_id": a['connect_id'], + "make": a['make'], + "model": a['model'], + "category": a['category'], + "matched_on": norm, + "source_text": token, + }) + + return { + "raw_text": raw[:1000], + "matched_assets": db_matches, + "match_count": len(db_matches), + } + + # ─── T4: Connect Label — unified photo + OCR + GPS endpoint ───────────────── class ConnectLabelRequest(BaseModel):