classify_makes.py: serial number → make/model/class classification
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
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user