515 lines
20 KiB
Python
515 lines
20 KiB
Python
"""
|
|
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', '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', '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/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', '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')
|
|
|
|
# ─── Short DN serals (8-digit formats seen in DB) ──────────────────────────
|
|
# DN has 8-digit serials: 11440039, 11495117, 24110016, 24220045, etc.
|
|
# Narrow prefix rules to avoid false positives with Royal (1100xxxx) or Crane
|
|
|
|
def _dn_8digit_short(sn_clean):
|
|
"""8-digit DN serials with specific known prefixes."""
|
|
return (len(sn_clean) == 8 and sn_clean.isdigit() and
|
|
(sn_clean.startswith('114') or sn_clean.startswith('241') or
|
|
sn_clean.startswith('1149') or sn_clean.startswith('2411')))
|
|
DN_SHORT_8 = (_dn_8digit_short, 'DN', 'BevMax/5800/3800/200E', 'Bev', 'medium')
|
|
|
|
# 10-digit 76/77 prefix → DN (e.g., 76820565BD, 76920191BD, 77090262AE)
|
|
def _dn_76_77(sn_clean):
|
|
"""10-digit serials starting with 76 or 77 → DN."""
|
|
return len(sn_clean) == 10 and sn_clean.isdigit() and sn_clean[:2] in ('76', '77')
|
|
DN_76_77 = (_dn_76_77, 'DN', 'BevMax/5800/3800/200E', 'Bev', 'medium')
|
|
|
|
# ─── 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 short 8-digit', DN_SHORT_8),
|
|
('DN 76/77 10-digit', DN_76_77),
|
|
('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()
|