Spaces:
Runtime error
Runtime error
| """Generate a SMALL SYNTHETIC dataset so the system is demonstrable on a clean | |
| clone WITHOUT committing any real footage-derived data (challenge rule). | |
| Outputs: | |
| data/sample_events.jsonl ~200 events, valid against the schema (deliverable) | |
| data/demo_pos.csv a handful of POS rows aligned so some visits convert | |
| These are HAND-SYNTHESISED (clearly labelled), not produced from the CCTV clips. | |
| The real pipeline writes the real data/events.jsonl + data/pos_transactions.csv | |
| (both git-ignored). The API computes identically over either. | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import json | |
| import random | |
| import uuid | |
| from datetime import datetime, timedelta, timezone | |
| from pathlib import Path | |
| random.seed(2026) | |
| STORE = "STORE_BLR_002" | |
| BASE = datetime(2026, 4, 10, 14, 50, 0, tzinfo=timezone.utc) # 20:20 IST | |
| FLOOR_ZONES = [("SKINCARE", "skin", "CAM_FLOOR_01"), | |
| ("MAKEUP", "makeup", "CAM_FLOOR_02"), | |
| ("NAIL_FRAGRANCE", "personal-care", "CAM_FLOOR_02"), | |
| ("MAKEUP_STUDIO", "makeup", "CAM_FLOOR_01")] | |
| DATA = Path(__file__).resolve().parent.parent / "data" | |
| def iso(dt: datetime) -> str: | |
| return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") | |
| def ev(events, vid, etype, t, *, cam, zone=None, dwell=0, staff=False, conf=0.9, qd=None): | |
| seq = sum(1 for e in events if e["visitor_id"] == vid) + 1 | |
| events.append({ | |
| "event_id": str(uuid.uuid4()), "store_id": STORE, "camera_id": cam, | |
| "visitor_id": vid, "event_type": etype, "timestamp": iso(t), "zone_id": zone, | |
| "dwell_ms": dwell, "is_staff": staff, "confidence": round(conf, 3), | |
| "metadata": {"queue_depth": qd, "sku_zone": zone, "session_seq": seq}, | |
| }) | |
| def build(): | |
| events: list[dict] = [] | |
| pos: list[dict] = [] | |
| converters_billing: list[datetime] = [] | |
| t = BASE | |
| # 2 staff members moving through zones (must be EXCLUDED from metrics) | |
| for i, (cam, zone) in enumerate([("CAM_BACK_01", "BACKROOM"), ("CAM_BILLING_01", "BILLING")]): | |
| vid = f"VIS_staff{i}" | |
| st = BASE + timedelta(seconds=5 + i) | |
| ev(events, vid, "ZONE_ENTER", st, cam=cam, zone=zone, staff=True, conf=0.8) | |
| for k in range(1, 4): # long presence -> dwell pings | |
| ev(events, vid, "ZONE_DWELL", st + timedelta(seconds=30 * k), cam=cam, | |
| zone=zone, dwell=30000 * k, staff=True, conf=0.8) | |
| ev(events, vid, "ZONE_EXIT", st + timedelta(seconds=120), cam=cam, zone=zone, | |
| dwell=120000, staff=True, conf=0.8) | |
| n_customers = 46 | |
| for i in range(n_customers): | |
| vid = f"VIS_{uuid.uuid4().hex[:8]}" | |
| t = t + timedelta(seconds=random.randint(20, 90)) # spread over ~40 min | |
| conf = round(random.uniform(0.35, 0.95), 3) | |
| ev(events, vid, "ENTRY", t, cam="CAM_ENTRY_01", conf=conf) | |
| # ~65% browse a named zone | |
| browsed = random.random() < 0.65 | |
| last = t | |
| if browsed: | |
| zone, dept, cam = random.choice(FLOOR_ZONES) | |
| zt = t + timedelta(seconds=random.randint(8, 25)) | |
| dwell_s = random.randint(15, 95) | |
| ev(events, vid, "ZONE_ENTER", zt, cam=cam, zone=zone, conf=conf) | |
| for k in range(1, dwell_s // 30 + 1): # ZONE_DWELL every 30s | |
| ev(events, vid, "ZONE_DWELL", zt + timedelta(seconds=30 * k), cam=cam, | |
| zone=zone, dwell=30000 * k, conf=conf) | |
| ev(events, vid, "ZONE_EXIT", zt + timedelta(seconds=dwell_s), cam=cam, | |
| zone=zone, dwell=dwell_s * 1000, conf=conf) | |
| last = zt + timedelta(seconds=dwell_s) | |
| # of browsers, ~55% reach billing | |
| if browsed and random.random() < 0.55: | |
| qd = random.randint(0, 5) | |
| bt = last + timedelta(seconds=random.randint(5, 20)) | |
| etype = "BILLING_QUEUE_JOIN" if qd > 0 else "ZONE_ENTER" | |
| ev(events, vid, etype, bt, cam="CAM_BILLING_01", zone="BILLING", qd=qd, conf=conf) | |
| purchase = random.random() < 0.6 | |
| if purchase: | |
| converters_billing.append(bt) | |
| ev(events, vid, "ZONE_EXIT", bt + timedelta(seconds=random.randint(30, 80)), | |
| cam="CAM_BILLING_01", zone="BILLING", dwell=random.randint(30, 80) * 1000, conf=conf) | |
| else: | |
| ev(events, vid, "BILLING_QUEUE_ABANDON", | |
| bt + timedelta(seconds=random.randint(40, 120)), | |
| cam="CAM_BILLING_01", zone="BILLING", qd=qd, conf=conf) | |
| # one explicit re-entry: exit then REENTRY with the same visitor_id | |
| if i == 3: | |
| xt = last + timedelta(seconds=30) | |
| ev(events, vid, "EXIT", xt, cam="CAM_ENTRY_01", conf=conf) | |
| ev(events, vid, "REENTRY", xt + timedelta(seconds=90), cam="CAM_ENTRY_01", conf=conf) | |
| # POS: align a txn shortly AFTER each converter's billing time (-> conversion) | |
| n = 0 | |
| for bt in converters_billing: | |
| n += 1 | |
| pos.append({"store_id": STORE, "transaction_id": f"TXN_demo_{n:03d}", | |
| "timestamp": iso(bt + timedelta(seconds=random.randint(30, 180))), | |
| "basket_value_inr": round(random.uniform(199, 3200), 2)}) | |
| # a couple of extra (non-correlated) txns earlier in the day | |
| for j in range(2): | |
| n += 1 | |
| pos.append({"store_id": STORE, "transaction_id": f"TXN_demo_{n:03d}", | |
| "timestamp": iso(BASE - timedelta(hours=2, minutes=17 * j)), | |
| "basket_value_inr": round(random.uniform(149, 1800), 2)}) | |
| events.sort(key=lambda e: e["timestamp"]) | |
| pos.sort(key=lambda r: r["timestamp"]) | |
| return events, pos | |
| def main(): | |
| DATA.mkdir(parents=True, exist_ok=True) | |
| events, pos = build() | |
| with (DATA / "sample_events.jsonl").open("w", encoding="utf-8") as fh: | |
| for e in events: | |
| fh.write(json.dumps(e, separators=(",", ":")) + "\n") | |
| with (DATA / "demo_pos.csv").open("w", newline="", encoding="utf-8") as fh: | |
| w = csv.DictWriter(fh, fieldnames=["store_id", "transaction_id", "timestamp", "basket_value_inr"]) | |
| w.writeheader() | |
| w.writerows(pos) | |
| print(f"wrote {len(events)} synthetic events -> data/sample_events.jsonl") | |
| print(f"wrote {len(pos)} synthetic POS rows -> data/demo_pos.csv") | |
| if __name__ == "__main__": | |
| main() | |