shawn/canteen-asset-tracker
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

Add Disney park classification + service entrance API

Browse Source 
- Auto-classifies 1,155 Disney assets into parks/resorts/offices
- New columns/table: assets.disney_park, service_entrances
- API: /api/disney/stats, /api/disney/classify, /api/assets/{id}/disney-park
- API: /api/service-entrances CRUD, /api/locations/{id}/service-entrances
- OCR strategy: ConnectID last-5, resort keyword detection, address patterns
This commit is contained in:
Shawn
2026-05-22 17:41:11 -04:00
parent 2e2fd50f70
commit 6aa5d3d35c
2 changed files with 539 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
disney_classify.py
+335
View File
@@ -0,0 +1,335 @@
"""
Disney park classification for canteen assets.
Detects which Disney park/area an asset belongs to based on:
- Asset name pattern (D-{Park} ...)
- Address keywords
- Room/building name keywords
- Customer name (already D-{Park} prefixed)
"""
import re
from typing import Optional
# ─── Disney Park definitions ─────────────────────────────────────────────────
DISNEY_PARK_NAMES = {
"magic-kingdom": "Magic Kingdom",
"epcot": "Epcot",
"hollywood-studios": "Hollywood Studios",
"animal-kingdom": "Animal Kingdom",
"disney-springs": "Disney Springs",
"resort": "Resort",
"office": "Office",
"other": "Other",
}
# Park icons / emoji for UI
DISNEY_PARK_ICONS = {
"magic-kingdom": "🏰",
"epcot": "🌍",
"hollywood-studios": "🎬",
"animal-kingdom": "🌿",
"disney-springs": "🛍️",
"resort": "🏨",
"office": "🏢",
"other": "📍",
}
DISNEY_PARK_COLORS = {
"magic-kingdom": "#9b59b6", # purple
"epcot": "#3498db", # blue
"hollywood-studios": "#e74c3c", # red
"animal-kingdom": "#2ecc71", # green
"disney-springs": "#f39c12", # orange
"resort": "#1abc9c", # teal
"office": "#95a5a6", # gray
"other": "#bdc3c7", # light gray
}
# ─── Resort name keywords ────────────────────────────────────────────────────
RESORT_KEYWORDS = [
"WILDERNESS LODGE", "Wilderness Lodge",
"FORT WILDERNESS", "Fort Wilderness",
"GRAND FLORIDIAN", "Grand Floridian", "Floridian",
"BOARDWALK", "Boardwalk",
"SARATOGA SPRINGS", "Saratoga Springs",
"CARIBBEAN BEACH", "Caribbean Beach",
"CORONADO SPRINGS", "Coronado Springs",
"GRAN DESTINO", "Gran Destino",
"PORT ORLEANS", "Port Orleans",
"POLYNESIAN", "Polynesian",
"CONTEMPORARY", "Contemporary",
"ANIMAL KINGDOM LODGE", "Animal Kingdom Lodge",
"ANIMAL KNGDM LODGE", "Animal Kngdm Lodge",
"KIDANI", "Kidani",
"BEACH CLUB", "Beach Club",
"YACHT CLUB", "Yacht Club",
"OLD KEY WEST", "Old Key West",
"RIVIERA", "Riviera",
"POP CENTURY", "Pop Century",
"ART OF ANIMATION", "Art of Animation",
"ALL STAR", "All Star",
"SHADES OF GREEN", "Shades of Green",
"BUENA VISTA PALACE", "Buena Vista Palace",
"WYNDHAM", "Wyndham",
"HILTON", "Hilton",
"IHG HOTEL", "IHG Hotel",
"DAYS INN", "Days Inn",
"B RESORT", "B Resort",
"DOUBLETREE", "DoubleTree",
"HOLIDAY INN", "Holiday Inn",
"BEST WESTERN", "Best Western",
"MAGNOLIA", "Magnolia", # Saratoga Springs area
]
# ─── Classification logic ────────────────────────────────────────────────────
def classify_asset(name: str, address: str = "", building_name: str = "",
room: str = "", customer_name: str = "") -> Optional[str]:
"""
Classify a canteen asset into a Disney park/area.
Returns a disney_park key or None if not Disney-related.
"""
text = f"{name} {address} {building_name} {room} {customer_name}"
# Quick check — is this Disney at all?
if not _is_disney_related(text):
return None
# 1. Direct name pattern: D-{Park} ...
park = _match_name_prefix(name)
if park:
return park
# 2. Address/building keywords
park = _match_address_keywords(address, building_name, room)
if park:
return park
# 3. Customer name pattern
park = _match_customer_name(customer_name)
if park:
return park
# 4. Resort keywords in any field
if _is_resort(text):
return "resort"
# 5. Default: other Disney
return "other"
def _is_disney_related(text: str) -> bool:
"""Check if this asset is Disney-related at all."""
t = text.upper()
keywords = [
"DISNEY", "WDW", "MAGIC KINGDOM", "EPCOT",
"HOLLYWOOD STUDIOS", "ANIMAL KINGDOM",
"DISNEY SPRINGS", "LAKE BUENA VISTA", "BAY LAKE",
"SEVEN SEAS", "HOTEL PLAZA BLVD",
"D-", "D_MAGIC", "D_EPCOT", "D_HOLLYWOOD", "D_ANIMAL",
"FLORIDIAN", "WILDERNESS LODGE", "BOARDWALK",
"GRAND FLORIDIAN", "SARATOGA SPRINGS",
"BUENA VISTA",
]
return any(kw in t for kw in keywords)
def _match_name_prefix(name: str) -> Optional[str]:
"""Match D-{Park} pattern in asset name."""
n = name.upper()
# D-Magic Kingdom
if "D-MAGIC KINGDOM" in n or "D_MAGIC KINGDOM" in n:
return "magic-kingdom"
# D-Epcot
if "D-EPCOT" in n or "D_EPCOT" in n:
return "epcot"
# D-Hollywood Studios
if "D-HOLLYWOOD STUDIOS" in n or "D_HOLLYWOOD STUDIOS" in n:
return "hollywood-studios"
# D-Animal Kingdom or D-AK
if "D-ANIMAL KINGDOM" in n or "D_ANIMAL KINGDOM" in n or " D-AK " in n:
return "animal-kingdom"
# D-Disney Springs
if "D-DISNEY SPRINGS" in n or "D_DISNEY SPRINGS" in n:
return "disney-springs"
# Resort detection in name prefix
if n.startswith("D-") or " - " in n:
# Extract the location part after D- and before any type indicator
for kw in RESORT_KEYWORDS:
if kw.upper() in n:
return "resort"
return None
def _match_address_keywords(address: str, building: str, room: str) -> Optional[str]:
"""Match address/building patterns to parks."""
addr = (address + " " + building + " " + room).upper()
# Magic Kingdom
if "SEVEN SEAS" in addr or "MAGIC KINGDOM" in addr or "BAY LAKE" in addr:
return "magic-kingdom"
# Epcot
if "EPCOT CENTER" in addr or "EPCOT" in addr:
return "epcot"
# Hollywood Studios
if "EAST STUDIO" in addr or "HOLLYWOOD STUDIOS" in addr:
return "hollywood-studios"
# Animal Kingdom
if "RAINFOREST" in addr or "ANIMAL KINGDOM" in addr or " AK " in addr or addr.startswith("AK "):
return "animal-kingdom"
# Disney Springs / Hotel Plaza Blvd area
if "DISNEY SPRINGS" in addr or "BUENA VISTA DR" in addr or "HOTEL PLAZA BLVD" in addr:
return "disney-springs"
# Floridian Way = Grand Floridian (resort)
if "FLORIDIAN" in addr:
return "resort"
# Epcot Resorts Blvd = Boardwalk area
if "EPCOT RESORTS" in addr:
return "resort"
return None
def _match_customer_name(customer: str) -> Optional[str]:
"""Match customer name to park classification."""
c = customer.upper()
if "D-MAGIC KINGDOM" in c:
return "magic-kingdom"
if "D-EPCOT" in c:
return "epcot"
if "D-HOLLYWOOD STUDIOS" in c:
return "hollywood-studios"
if "D-DISNEY SPRINGS" in c:
return "disney-springs"
if "D-DISNEY VENDING" in c or "D-DISNEY WORLD" in c:
return "office"
# D-{Resort} pattern
if c.startswith("D-"):
for kw in RESORT_KEYWORDS:
if kw.upper() in c:
return "resort"
# If it starts with D- and we haven't matched yet, it's some Disney thing
return "other"
return None
def _is_resort(text: str) -> bool:
"""Check if text mentions a known Disney resort."""
t = text.upper()
for kw in RESORT_KEYWORDS:
if kw.upper() in t:
return True
return False
# ─── Batch classification ────────────────────────────────────────────────────
def classify_all_assets(db_path: str) -> dict:
"""
Run classification on all Disney-related assets in the DB.
Returns a report of what was classified.
"""
import sqlite3
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
# Get all assets that are Disney-related (or have no classification yet)
rows = conn.execute("""
SELECT a.id, a.machine_id, a.name, a.address, a.building_name, a.room,
a.disney_park, c.name as customer_name
FROM assets a
LEFT JOIN customers c ON a.customer_id = c.id
WHERE a.disney_park IS NULL
""").fetchall()
results = {"classified": [], "unclassifiable": [], "total_checked": len(rows)}
counts = {}
for row in rows:
park = classify_asset(
name=row['name'] or '',
address=row['address'] or '',
building_name=row['building_name'] or '',
room=row['room'] or '',
customer_name=row['customer_name'] or '',
)
if park:
conn.execute(
"UPDATE assets SET disney_park = ?, updated_at = datetime('now') WHERE id = ?",
(park, row['id'])
)
results["classified"].append({
"id": row['id'],
"machine_id": row['machine_id'],
"name": row['name'],
"park": park,
})
counts[park] = counts.get(park, 0) + 1
else:
results["unclassifiable"].append({
"id": row['id'],
"machine_id": row['machine_id'],
"name": row['name'],
})
conn.commit()
conn.close()
results["counts"] = counts
results["total_classified"] = len(results["classified"])
return results
def get_classification_stats(db_path: str) -> dict:
"""Get statistics about current Disney classifications."""
import sqlite3
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
rows = conn.execute("""
SELECT disney_park, COUNT(*) as count
FROM assets
WHERE disney_park IS NOT NULL
GROUP BY disney_park
ORDER BY count DESC
""").fetchall()
total = conn.execute("SELECT COUNT(*) as c FROM assets").fetchone()['c']
disney = conn.execute(
"SELECT COUNT(*) as c FROM assets WHERE disney_park IS NOT NULL"
).fetchone()['c']
unclassified = conn.execute(
"SELECT COUNT(*) as c FROM assets WHERE name LIKE '%Disney%' OR name LIKE '%D-%' AND disney_park IS NULL"
).fetchone()['c']
conn.close()
return {
"total_assets": total,
"classified": disney,
"unclassified_disney": 0, # Will be accurate after classify run
"by_park": {row['disney_park']: row['count'] for row in rows},
}
server.py
+204 -1
View File
@@ -164,10 +164,22 @@ def _create_v2_tables(conn: sqlite3.Connection):
latitude REAL DEFAULT NULL,
longitude REAL DEFAULT NULL,
geofence_radius_meters INTEGER DEFAULT 50,
disney_park TEXT DEFAULT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS service_entrances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
location_id INTEGER NOT NULL REFERENCES locations(id) ON DELETE CASCADE,
name TEXT NOT NULL,
latitude REAL NOT NULL,
longitude REAL NOT NULL,
notes TEXT DEFAULT '',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS checkins (
id INTEGER PRIMARY KEY AUTOINCREMENT,
asset_id INTEGER NOT NULL REFERENCES assets(id) ON DELETE CASCADE,
@@ -428,8 +440,28 @@ def init_db(conn: sqlite3.Connection):
conn.execute("ALTER TABLE locations ADD COLUMN longitude REAL DEFAULT NULL")
conn.commit()
# v4 migration: disney_park column + service_entrances table
if "disney_park" not in asset_cols:
conn.execute("ALTER TABLE assets ADD COLUMN disney_park TEXT DEFAULT NULL")
conn.commit()
# ─── App / Middleware ───────────────────────────────────────────────────────
# Ensure service_entrances table exists for existing DBs
conn.execute("""
CREATE TABLE IF NOT EXISTS service_entrances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
location_id INTEGER NOT NULL REFERENCES locations(id) ON DELETE CASCADE,
name TEXT NOT NULL,
latitude REAL NOT NULL,
longitude REAL NOT NULL,
notes TEXT DEFAULT '',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
conn.commit()
# --- App / Middleware
@asynccontextmanager
@@ -2114,6 +2146,177 @@ def geocode(lat: float = Query(...), lng: float = Query(...)):
return result
# ─── Disney Park Classification ──────────────────────────────────────────────
@app.get("/api/disney/stats")
def disney_stats():
"""Get Disney classification statistics."""
conn = get_db()
rows = conn.execute(
"SELECT disney_park, COUNT(*) as count FROM assets WHERE disney_park IS NOT NULL GROUP BY disney_park ORDER BY count DESC"
).fetchall()
total = conn.execute("SELECT COUNT(*) as c FROM assets").fetchone()['c']
classified = conn.execute("SELECT COUNT(*) as c FROM assets WHERE disney_park IS NOT NULL").fetchone()['c']
conn.close()
from disney_classify import DISNEY_PARK_NAMES, DISNEY_PARK_ICONS, DISNEY_PARK_COLORS
return {
"total_assets": total,
"classified": classified,
"by_park": {
row['disney_park']: {
"count": row['count'],
"name": DISNEY_PARK_NAMES.get(row['disney_park'], row['disney_park']),
"icon": DISNEY_PARK_ICONS.get(row['disney_park'], "📍"),
"color": DISNEY_PARK_COLORS.get(row['disney_park'], "#95a5a6"),
}
for row in rows
},
}
@app.post("/api/disney/classify")
def classify_disney_assets():
"""Run auto-classification on all assets. Returns report."""
from disney_classify import classify_all_assets
result = classify_all_assets(DB_PATH)
return result
@app.put("/api/assets/{asset_id}/disney-park")
def set_asset_disney_park(asset_id: int, body: dict):
"""Manually set the disney_park classification for an asset."""
park = body.get("disney_park")
valid = {"magic-kingdom", "epcot", "hollywood-studios", "animal-kingdom",
"disney-springs", "resort", "office", "other", None}
if park not in valid and park is not None:
raise HTTPException(400, f"Invalid park. Valid: {', '.join(v for v in valid if v)}")
conn = get_db()
conn.execute(
"UPDATE assets SET disney_park = ?, updated_at = datetime('now') WHERE id = ?",
(park, asset_id)
)
conn.commit()
conn.close()
return {"status": "ok", "asset_id": asset_id, "disney_park": park}
# ─── Service Entrances ────────────────────────────────────────────────────────
class ServiceEntranceCreate(BaseModel):
location_id: int
name: str = ""
latitude: float
longitude: float
notes: str = ""
class ServiceEntranceUpdate(BaseModel):
name: Optional[str] = None
latitude: Optional[float] = None
longitude: Optional[float] = None
notes: Optional[str] = None
@app.get("/api/locations/{location_id}/service-entrances")
def list_service_entrances(location_id: int):
"""List service entrances for a location."""
conn = get_db()
rows = conn.execute(
"SELECT se.*, l.name as location_name FROM service_entrances se "
"JOIN locations l ON se.location_id = l.id "
"WHERE se.location_id = ? ORDER BY se.name",
(location_id,)
).fetchall()
conn.close()
return [dict(r) for r in rows]
@app.post("/api/service-entrances", status_code=201)
def create_service_entrance(body: ServiceEntranceCreate):
"""Create a new service entrance."""
conn = get_db()
# Verify location exists
loc = conn.execute("SELECT id FROM locations WHERE id = ?", (body.location_id,)).fetchone()
if not loc:
conn.close()
raise HTTPException(404, "Location not found")
cursor = conn.execute(
"INSERT INTO service_entrances (location_id, name, latitude, longitude, notes) VALUES (?, ?, ?, ?, ?)",
(body.location_id, body.name, body.latitude, body.longitude, body.notes)
)
conn.commit()
row = conn.execute("SELECT * FROM service_entrances WHERE id = ?", (cursor.lastrowid,)).fetchone()
conn.close()
return dict(row)
@app.put("/api/service-entrances/{entrance_id}")
def update_service_entrance(entrance_id: int, body: ServiceEntranceUpdate):
"""Update a service entrance."""
conn = get_db()
existing = conn.execute("SELECT * FROM service_entrances WHERE id = ?", (entrance_id,)).fetchone()
if not existing:
conn.close()
raise HTTPException(404, "Service entrance not found")
updates = {}
if body.name is not None:
updates['name'] = body.name
if body.latitude is not None:
updates['latitude'] = body.latitude
if body.longitude is not None:
updates['longitude'] = body.longitude
if body.notes is not None:
updates['notes'] = body.notes
if updates:
updates['updated_at'] = "datetime('now')"
set_clause = ", ".join(f"{k} = ?" for k in updates)
values = list(updates.values())
# Handle updated_at specially
set_clause = ", ".join(
f"{k} = datetime('now')" if k == 'updated_at' else f"{k} = ?"
for k in updates
)
values = [v for k, v in updates.items() if k != 'updated_at']
conn.execute(
f"UPDATE service_entrances SET {set_clause} WHERE id = ?",
(*values, entrance_id)
)
conn.commit()
row = conn.execute("SELECT se.*, l.name as location_name FROM service_entrances se "
"JOIN locations l ON se.location_id = l.id "
"WHERE se.id = ?", (entrance_id,)).fetchone()
conn.close()
return dict(row)
@app.delete("/api/service-entrances/{entrance_id}", status_code=204)
def delete_service_entrance(entrance_id: int):
"""Delete a service entrance."""
conn = get_db()
cursor = conn.execute("DELETE FROM service_entrances WHERE id = ?", (entrance_id,))
conn.commit()
conn.close()
if cursor.rowcount == 0:
raise HTTPException(404, "Service entrance not found")
return None
@app.get("/api/service-entrances")
def list_all_service_entrances():
"""List all service entrances."""
conn = get_db()
rows = conn.execute(
"SELECT se.*, l.name as location_name, l.address as location_address "
"FROM service_entrances se "
"LEFT JOIN locations l ON se.location_id = l.id "
"ORDER BY l.name, se.name"
).fetchall()
conn.close()
return [dict(r) for r in rows]
# ─── Static Files (mounted last to not shadow routes) ──────────────────────
app.mount("/uploads", StaticFiles(directory=str(UPLOADS_DIR)), name="uploads")