"""Generate 25 CSV test files covering all flag states.""" import csv import random import os from datetime import date, timedelta random.seed(42) DEVICE_TYPES = ["dexcom_g7", "dexcom_g6", "freestyle_libre_3", "omnipod_5"] PAYERS = ["Medicare Part B", "Medicaid - GA", "BCBS - FL", "Aetna", "UnitedHealth", "Cigna", "Humana"] COMPONENTS = {"dexcom_g7": "sensor", "dexcom_g6": "sensor", "freestyle_libre_3": "sensor", "omnipod_5": "pod"} # Shipment date ranges to trigger different flag states TODAY = date.today() DATE_BUCKETS = { "OK": (TODAY - timedelta(days=10), TODAY - timedelta(days=1)), "VISIT_DUE": (TODAY - timedelta(days=400), TODAY - timedelta(days=250)), # old visit, no recent qualifier "OUT_OF_COVERAGE": (TODAY - timedelta(days=600), TODAY - timedelta(days=500)), # way too old "REFILL_WINDOW": (TODAY - timedelta(days=30), TODAY - timedelta(days=25)), # inside resupply window } OUTPUT_DIR = os.path.dirname(os.path.abspath(__file__)) for i in range(1, 26): flag = random.choice(list(DATE_BUCKETS.keys())) bucket = DATE_BUCKETS[flag] delta = (bucket[1] - bucket[0]).days ship_date = bucket[0] + timedelta(days=random.randint(0, max(delta, 1))) device = random.choice(DEVICE_TYPES) component = COMPONENTS[device] payer = random.choice(PAYERS) quantity = random.choice([1, 2, 3, 6, 9, 14]) filename = f"sample-batch-{i:02d}-{flag.lower()}.csv" filepath = os.path.join(OUTPUT_DIR, filename) with open(filepath, "w", newline="") as f: writer = csv.writer(f) writer.writerow(["patient_id", "device_type", "shipment_date", "quantity", "payer", "component"]) # 3-8 rows per file num_rows = random.randint(3, 8) for j in range(num_rows): pid = f"PT-{1001 + (i-1) * 10 + j}" row_ship = ship_date + timedelta(days=random.randint(-5, 5)) writer.writerow([pid, device, row_ship.isoformat(), random.choice([1, 2, 3, 6, 9]), payer, component]) print(f"Wrote {filename} ({num_rows} rows, flag={flag})") print(f"\nDone — 25 files in {OUTPUT_DIR}")