Spaces:
Runtime error
Runtime error
File size: 6,306 Bytes
c1dcc23 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """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()
|