From 4f33afb445371517b7311e365b7dc9607e1a237e Mon Sep 17 00:00:00 2001 From: Shawn Date: Fri, 22 May 2026 19:02:04 -0400 Subject: [PATCH] =?UTF-8?q?classify=5Fmakes.py:=20serial=20number=20?= =?UTF-8?q?=E2=86=92=20make/model/class=20classification?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Classifies Unknown-make assets by serial number prefix patterns, cross-referenced against Cantaloupe Excel export data. Handles: - Character substitution errors (S→5, I→1, O→0) - Double-strike/extra character cleanup (I5SS → 155) - 8 rule sets: Vendo, Crane (167/168/186/187/222/221/47x), DN/Dixie Narco, Royal (20xx/BA/CA/PA), USI, AMS, VE - Confidence levels: high/medium/low - Dry-run + --apply modes - --stats for post-classification summary Result: 121/130 unknown machines classified (93%) 76 Vendo, 14 DN, 12 Crane, 8 Royal, 6 USI, 1 AMS 9 remaining: garbage serials (a, 00, 520) + one-offs --- classify_makes.py | 495 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 495 insertions(+) create mode 100644 classify_makes.py diff --git a/classify_makes.py b/classify_makes.py new file mode 100644 index 0000000..b72a0a8 --- /dev/null +++ b/classify_makes.py @@ -0,0 +1,495 @@ +""" +Serial number → Make, Model, and Class classification for canteen assets. + +Uses serial number prefix patterns cross-referenced against Cantaloupe export data +to identify Unknown machines and classify GF Bev vs GF Food. + +GF = Glass Front (vending industry term). + - GF Bev = glass-front beverage machines (mostly DN/Dixie Narco) + - GF Food = glass-front food/snack machines (ALL Crane) + +Usage: + python3 classify_makes.py # dry-run (report only) + python3 classify_makes.py --apply # update DB + python3 classify_makes.py --stats # show classification stats +""" + +import sqlite3 +import re +import sys +from pathlib import Path +from typing import Optional, Tuple + +DB_PATH = str(Path(__file__).parent / "assets.db") + + +# ─── Character substitution (human entry errors) ────────────────────────── + +def normalise_serial(sn: str) -> str: + """ + Normalise a serial number, applying common human-entry substitutions. + S→5, I→1, O→0 are the most common errors when hand-entering serial plates. + + Also handles double-strike errors (I5SS → I5S, i.e. extra S typed twice). + Returns the cleaned serial. + """ + if not sn: + return '' + s = sn.strip().upper() + # Common substitutions (order matters — do before other processing) + s = s.replace('I', '1') # I→1 + s = s.replace('O', '0') # O→0 + s = s.replace('S', '5') # S→5 + + # Handle double-strike: I5SS → 1555 → collapse extra 5 + # Pattern: 1555 → 155 (extra digit from double-strike or repeat key) + s = re.sub(r'1555+', '155', s) + # Remove dashes for pattern matching + return s + + +def clean_for_pattern(sn: str) -> str: + """Remove dashes and whitespace for pattern matching.""" + return normalise_serial(sn).replace('-', '').strip() + + +# ─── Rule definitions ───────────────────────────────────────────────────── + +# Each rule is: (pattern_fn, make, model, gf_class, confidence) +# pattern_fn takes cleaned serial and returns True/False +# gf_class is the inferred GF classification (GF Bev, GF Food, Bev, Snack, etc.) + +def _make_rule(prefixes, make, model, gf_class, confidence='high', + min_len=None, max_len=None): + """Factory for prefix-based rules.""" + def match(sn_clean): + if min_len and len(sn_clean) < min_len: + return False + if max_len and len(sn_clean) > max_len: + return False + for p in prefixes: + if sn_clean.startswith(p): + return True + return False + return (match, make, model, gf_class, confidence) + + +# ─── VENDO ───────────────────────────────────────────────────────────────── +# Serial format: 000XXXXXX (9-digit, starts with 000) +VENDO_RULE = _make_rule(['000'], 'Vendo', '621/721/821', 'Bev', + min_len=8, max_len=10) + +# ─── DN / DIXIE NARCO ───────────────────────────────────────────────────── +# Serial format: 11XXXXXXXXXX (12-13 digit, starts with 11, various sub-prefixes) +DN_PREFIXES = [ + '112301', '112304', '112402', '112404', '112408', '112503', + '112601', '112602', '112603', '112006', '112010', '112011', + '111808', '111812', '111904', '111905', '111906', '112111', + '112206', '112510', + # Other DN 11-prefixed: 11 + 4-digit year/week code +] +DN_RULE = _make_rule(DN_PREFIXES, 'DN', 'BevMax/5800/3800/200E', 'GF Bev', + min_len=10, max_len=14) + +# Broader DN catch: any 12-13 digit serial starting with '11' +def _dn_broad(sn_clean): + return len(sn_clean) >= 11 and sn_clean.startswith('11') and sn_clean.isdigit() +DN_BROAD = (_dn_broad, 'DN', 'Unknown', 'GF Bev', 'medium') + +# ─── CRANE ───────────────────────────────────────────────────────────────── +# Multiple serial formats for Crane/National machines + +# 9-digit: 167XXXXXX, 168XXXXXX (National 167/168 series) +CRANE_167_168 = _make_rule(['167', '168'], 'Crane', '15x/16x', 'Snack', + min_len=9, max_len=10) + +# 9-digit: 186XXXXXX, 187XXXXXX (Crane Merchant Media) +CRANE_186_187 = _make_rule(['186', '187'], 'Crane', 'Merchant Media', 'Snack', + min_len=9, max_len=10) + +# 9-digit: 180XXXXXX, 181XXXXXX +CRANE_180_181 = _make_rule(['180', '181'], 'Crane', 'Merchant Media', 'Snack', + min_len=9, max_len=10) + +# 12-digit: 222XXXXXXXXX (Crane Merchant Media, 186, 187) +def _crane_222(sn_clean): + return len(sn_clean) >= 11 and sn_clean.startswith('222') and sn_clean.isdigit() +CRANE_222 = (_crane_222, 'Crane', 'Merchant Media', 'Snack', 'high') + +# 12-digit: 221XXXXXXXXX (Crane Merchant Media, 472) +def _crane_221(sn_clean): + return len(sn_clean) >= 11 and sn_clean.startswith('221') and sn_clean.isdigit() +CRANE_221 = (_crane_221, 'Crane', 'Merchant Media', 'Snack/GF Food', 'high') + +# 471/472: dash or 9-digit +def _crane_47(sn_clean): + return (sn_clean.startswith('471') or sn_clean.startswith('472')) and len(sn_clean) >= 8 +CRANE_47 = (_crane_47, 'Crane', '471/472', 'GF Food', 'high') + +# Dash format: 168-XXXXXX, 167-XXXXXX +def _crane_dash(orig_sn): + """Check original serial (with dashes) for Crane dash patterns.""" + if not orig_sn: + return False + s = orig_sn.strip() + for prefix in ['168-', '167-', '472-', '471-', '449-', '186-', '187-']: + if s.startswith(prefix): + return True + return False +CRANE_DASH = (_crane_dash, 'Crane', '15x/16x', 'Snack', 'high') + +# ─── ROYAL ───────────────────────────────────────────────────────────────── +# Format 1: 20YYMMCAXXXXX or 20YYWWBAXXXXX (year+week+code+sequence) +def _royal_20xx(sn_clean): + return (sn_clean.startswith('200') or sn_clean.startswith('201')) and len(sn_clean) >= 10 +ROYAL_20XX = (_royal_20xx, 'Royal', 'GIII', 'Bev', 'high') + +# Format 2: 1[5-9]WW [AL/BL/etc] XXXXX (old Royal format) +def _royal_old(sn_clean): + """Match Royal old format: 15WW AL XXXXX etc.""" + return bool(re.match(r'^1[5-9]\d{2}[A-Z]{2}\d{5}$', sn_clean)) +ROYAL_OLD = (_royal_old, 'Royal', 'GIII', 'Bev', 'medium') + +# 20xx with BA/CA/PA codes in original format +def _royal_code(orig_sn): + if not orig_sn: + return False + s = orig_sn.strip().upper() + return bool(re.search(r'(BA|CA|PA)\d{5}', s)) +ROYAL_CODE = (_royal_code, 'Royal', 'GIII', 'Bev', 'high') + +# ─── USI ─────────────────────────────────────────────────────────────────── +# 12-digit serials, often starting with 12, 14, 15 +def _usi_12digit(sn_clean): + return len(sn_clean) == 12 and sn_clean.isdigit() and sn_clean[:2] in ('12', '14', '15') +USI_12 = (_usi_12digit, 'USI', 'Mercato/Evoke/30xx', 'Snack', 'medium') + +# 7-digit serials (older USI) +def _usi_7digit(sn_clean): + return len(sn_clean) == 7 and sn_clean.isdigit() and sn_clean.startswith('13') +USI_7 = (_usi_7digit, 'USI', '30xx', 'Snack', 'medium') + +# ─── AMS ─────────────────────────────────────────────────────────────────── +# Dash format: 1-XXXXXXXX or 1-XXXX-XXXX +def _ams_dash(orig_sn): + if not orig_sn: + return False + s = orig_sn.strip() + return bool(re.match(r'^1-\d{4,8}', s)) or bool(re.match(r'^1-\d{4}-\d{4}', s)) +AMS_DASH = (_ams_dash, 'AMS', '3561/Sensit 3', 'Snack', 'high') + +# AMS 11-digit: 1118XXXXXXXX, 1121XXXXXXXX +AMS_LONG = _make_rule(['111809', '111811', '112111', '112034'], 'AMS', '3561/Sensit 3', 'Snack', + min_len=10, max_len=14) + +# ─── VE ──────────────────────────────────────────────────────────────────── +# VE serials: often short, with revision patterns +def _ve_pattern(sn_clean): + return bool(re.match(r'^[A-Z]\d{7}', sn_clean)) +VE_PATTERN = (_ve_pattern, 'VE', 'Revision Door', 'Snack', 'low') + +# ─── Edge cases / near-misses ────────────────────────────────────────────── + +# 8-digit 00XXXXXX → likely Vendo missing one leading zero (Vendo is 000XXXXXX) +def _vendo_8digit(sn_clean): + return len(sn_clean) == 8 and sn_clean.startswith('00') and sn_clean.isdigit() +VENDO_8 = (_vendo_8digit, 'Vendo', '621/721/821', 'Bev', 'medium') + +# BA/PA suffix without year prefix → Royal +def _royal_suffix(orig_sn): + if not orig_sn: + return False + s = orig_sn.strip().upper() + return bool(re.search(r'\d{6,8}(BA|PA|CA)$', s)) +ROYAL_SUFFIX = (_royal_suffix, 'Royal', 'GIII', 'Bev', 'medium') + +# RY prefix → Royal abbreviation +def _royal_ry(orig_sn): + if not orig_sn: + return False + s = orig_sn.strip().upper() + return s.startswith('RY') and len(s) >= 6 +ROYAL_RY = (_royal_ry, 'Royal', 'GIII', 'Bev', 'low') + +# ─── RULE COLLECTION (ordered: first match wins) ─────────────────────────── + +RULES = [ + # (description, rule_tuple) + ('Vendo 000-', VENDO_RULE), + ('Crane 167/168', CRANE_167_168), + ('Crane 186/187', CRANE_186_187), + ('Crane 180/181', CRANE_180_181), + ('Crane 222-', CRANE_222), + ('Crane 221-', CRANE_221), + ('Crane 47x', CRANE_47), + ('Crane dash', CRANE_DASH), + ('Royal 20xx', ROYAL_20XX), + ('Royal old format', ROYAL_OLD), + ('Royal BA/CA code', ROYAL_CODE), + ('USI 12-digit', USI_12), + ('USI 7-digit', USI_7), + ('AMS dash', AMS_DASH), + ('AMS long', AMS_LONG), + ('DN specific prefixes', DN_RULE), + ('DN broad 11x', DN_BROAD), + ('Vendo 8-digit 00', VENDO_8), + ('Royal BA/PA suffix', ROYAL_SUFFIX), + ('Royal RY prefix', ROYAL_RY), + ('VE pattern', VE_PATTERN), +] + + +# ─── Classification logic ────────────────────────────────────────────────── + +def classify_by_serial(serial_number: str) -> Optional[dict]: + """ + Attempt to classify an asset by its serial number. + Returns a dict with make, model, gf_class, confidence, rule_name + or None if no rule matches. + """ + if not serial_number or not serial_number.strip(): + return None + + orig = serial_number.strip() + clean = clean_for_pattern(orig) + + for rule_name, (pattern_fn, make, model, gf_class, confidence) in RULES: + try: + if pattern_fn(clean if 'dash' not in rule_name.lower() and + 'code' not in rule_name.lower() + else orig): + return { + 'make': make, + 'model': model, + 'gf_class': gf_class, + 'confidence': confidence, + 'rule': rule_name, + } + except Exception: + continue + + return None + + +def classify_unknown_assets(db_path: str, apply: bool = False) -> dict: + """ + Find all assets with Unknown/empty make and attempt to classify by serial. + + Args: + db_path: Path to assets.db + apply: If True, actually UPDATE the DB. If False, dry-run report. + + Returns a report dict. + """ + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + # Find Unknown-make assets with non-empty serials + rows = conn.execute(""" + SELECT id, machine_id, name, serial_number, make, model, category + FROM assets + WHERE (make = 'Unknown' OR make IS NULL OR make = '') + AND serial_number IS NOT NULL + AND serial_number != '' + ORDER BY serial_number + """).fetchall() + + results = { + 'total_unknown': len(rows), + 'classified': [], + 'unmatched': [], + 'by_make': {}, + 'by_rule': {}, + 'by_confidence': {'high': 0, 'medium': 0, 'low': 0}, + } + + for row in rows: + classification = classify_by_serial(row['serial_number']) + + if classification and classification['confidence'] != 'low': + entry = { + 'id': row['id'], + 'machine_id': row['machine_id'], + 'name': row['name'], + 'serial': row['serial_number'], + 'current_make': row['make'], + 'current_model': row['model'], + 'current_category': row['category'], + **classification, + } + + # Infer the best category/class if current is generic + if row['category'] in ('Other', 'Unknown', '', None): + entry['suggested_category'] = classification['gf_class'] + else: + entry['suggested_category'] = row['category'] + + results['classified'].append(entry) + results['by_make'][classification['make']] = \ + results['by_make'].get(classification['make'], 0) + 1 + results['by_rule'][classification['rule']] = \ + results['by_rule'].get(classification['rule'], 0) + 1 + results['by_confidence'][classification['confidence']] += 1 + else: + results['unmatched'].append({ + 'id': row['id'], + 'machine_id': row['machine_id'], + 'name': row['name'], + 'serial': row['serial_number'], + 'reason': classification['rule'] if classification else 'no rule matched', + }) + + # Apply updates if requested + if apply and results['classified']: + updated = 0 + for entry in results['classified']: + if entry['confidence'] == 'low': + continue # Skip low-confidence matches + conn.execute(""" + UPDATE assets + SET make = ?, + model = CASE WHEN model = 'Unknown' OR model IS NULL OR model = '' + THEN ? ELSE model END, + category = CASE WHEN category = 'Other' OR category IS NULL OR category = '' + THEN ? ELSE category END, + updated_at = datetime('now') + WHERE id = ? + """, ( + entry['make'], + entry['model'], + entry['suggested_category'], + entry['id'], + )) + updated += 1 + conn.commit() + results['applied'] = updated + + conn.close() + return results + + +def get_stats(db_path: str) -> dict: + """Get current classification statistics.""" + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + total = conn.execute("SELECT COUNT(*) as c FROM assets").fetchone()['c'] + unknown = conn.execute( + "SELECT COUNT(*) as c FROM assets WHERE make = 'Unknown' OR make IS NULL OR make = ''" + ).fetchone()['c'] + + by_make = {} + for r in conn.execute( + "SELECT make, COUNT(*) as cnt FROM assets GROUP BY make ORDER BY cnt DESC" + ): + by_make[r['make']] = r['cnt'] + + by_category = {} + for r in conn.execute( + "SELECT category, COUNT(*) as cnt FROM assets GROUP BY category ORDER BY cnt DESC" + ): + by_category[r['category']] = r['cnt'] + + conn.close() + return { + 'total': total, + 'unknown_make': unknown, + 'by_make': by_make, + 'by_category': by_category, + } + + +# ─── CLI ──────────────────────────────────────────────────────────────────── + +def main(): + import argparse + parser = argparse.ArgumentParser( + description='Classify Unknown machines by serial number pattern' + ) + parser.add_argument('--apply', action='store_true', + help='Actually update the database (default: dry-run)') + parser.add_argument('--stats', action='store_true', + help='Show classification statistics and exit') + parser.add_argument('--db', default=DB_PATH, + help=f'Database path (default: {DB_PATH})') + args = parser.parse_args() + + if args.stats: + stats = get_stats(args.db) + print("=== Classification Statistics ===") + print(f"Total assets: {stats['total']}") + print(f"Unknown make: {stats['unknown_make']} " + f"({stats['unknown_make']/stats['total']*100:.1f}%)") + print() + print("By Make:") + for make, cnt in sorted(stats['by_make'].items(), + key=lambda x: x[1], reverse=True): + bar = '█' * (cnt // 20) + print(f" {make:<15} {cnt:>5} {bar}") + print() + print("By Category:") + for cat, cnt in sorted(stats['by_category'].items(), + key=lambda x: x[1], reverse=True): + print(f" {cat:<15} {cnt:>5}") + return + + mode = "DRY-RUN" if not args.apply else "APPLY" + print(f"=== Serial Number Classification ({mode}) ===\n") + + result = classify_unknown_assets(args.db, apply=args.apply) + + print(f"Unknown-make assets checked: {result['total_unknown']}") + print(f"Classified: {len(result['classified'])}") + print(f"Unmatched: {len(result['unmatched'])}") + print() + + if result['classified']: + print("=== By Make ===") + for make, cnt in sorted(result['by_make'].items(), + key=lambda x: x[1], reverse=True): + print(f" → {make}: {cnt}") + + print() + print("=== By Rule ===") + for rule, cnt in sorted(result['by_rule'].items(), + key=lambda x: x[1], reverse=True): + print(f" {rule}: {cnt}") + + print() + print("=== By Confidence ===") + for level in ['high', 'medium', 'low']: + cnt = result['by_confidence'].get(level, 0) + if cnt: + print(f" {level}: {cnt}") + + print() + print("=== Classified Assets ===") + for e in result['classified']: + flag = '' + if e['confidence'] == 'medium': + flag = ' ⚠️' + print(f" MID={e['machine_id']:>8} SN={e['serial']:>16} " + f"→ {e['make']:<8} {e['model']:<25} " + f"({e['rule']}) [{e['confidence']}]{flag}") + + if result['unmatched']: + print() + print("=== Unmatched (needs manual review) ===") + for e in result['unmatched']: + print(f" MID={e['machine_id']:>8} SN={e['serial']:>16} " + f"Name={e['name'][:50]}") + + if args.apply: + print() + print(f"✅ Applied {result.get('applied', 0)} updates to database.") + + print() + + # Show post-classification stats + stats = get_stats(args.db) + print(f"After classification: {stats['unknown_make']} Unknown make remaining " + f"(of {stats['total']} total)") + + +if __name__ == '__main__': + main()