sttil/Signal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Signal/docs/superpowers/plans/2026-06-07-signal-csv-generator.md
Kisa bcb1a10fe6 feat: demo MVP — 90/60/45 priority tiers, doc state machine, confirm visit workflow 
- New CoverageFlag enum: SUPPLY_LAPSED, VISIT_REQUIRED, TRANSFER_PENDING,
  RENEWAL_CRITICAL/ELEVATED/SOON, RESUPPLY_READY, ACTIVE
- Doc state machine: 5-item payer-dependent status per patient (SWO, Visit,
  PECOS, PA, Diagnosis) with cascade chain
- Confirm Visit endpoint: staff enters prescriber-confirmed date, persisted in
  Supabase confirmed_visits table, survives all future CSV imports
- Supabase migration: 001_add_confirmed_visits.sql (run manually in SQL editor)
- Frontend: Badge rebuilt for 8 flags, DocStatusBar 5-dot display,
  ConfirmVisitModal, expandable WorklistTable rows
- Legal: LOI, NDA, BAA drafts at pitch/legal/ for Nixon Law Group review
- Compliance docs: privacy policy, incident response, data handling
- CSV generator: market_data.json + PA/NJ generator scripts
- 15/15 tests passing

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-07 11:29:33 -04:00

872 lines
33 KiB
Markdown
Raw Blame History

# Signal Demo MVP — CSV Generator Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Build a parameterized CSV generator system that produces PA-market (25 files) and NJ-market (25 files) test sets with state-change patients, transferred patients, and all normalizer challenge types. Also deliver the Bug Hunter report.
**Architecture:** `generate_supplier_set.py` is the core reusable generator. It reads `market_data.json` for state-specific payer names and geography. State-specific wrappers (`generate_pa_set.py`, `generate_nj_set.py`) call it with preset args. The Bug Hunter is a script that runs all 50 CSVs through the Signal normalizer and returns a structured failure report.
**Tech Stack:** Python 3.11 (stdlib only — no external dependencies). Signal normalizer lives at `python-backend/api/normalizer.py`.
**Key paths:**
- Reference generator: `test-data/generate_samples.py` — read this before writing anything
- Output: `test-data/pa-set/` (25 files) and `test-data/nj-set/` (25 files)
- Normalizer import path (from test-data dir): `sys.path.insert(0, '../python-backend')`
**Before starting:** Read `test-data/generate_samples.py` in full. The new generators follow the same HEADER_VARIANTS pattern and DATE_BUCKETS approach, extended for PA/NJ payers, doc fields, and state-change/transfer scenarios.
---
## File Map
**Create:**
- `test-data/market_data.json` — state-to-payer-names config
- `test-data/generate_supplier_set.py` — parameterized generator (accepts --state, --supplier, --patients, --challenge-level)
- `test-data/generate_pa_set.py` — PA market wrapper
- `test-data/generate_nj_set.py` — NJ market wrapper
- `test-data/bug_hunter.py` — runs all 50 CSVs through normalizer, outputs failure report
- `test-data/pa-set/` — 25 generated CSV files
- `test-data/nj-set/` — 25 generated CSV files
---
## Task 1: market_data.json
- [ ] **Step 1: Create market_data.json**
Create `test-data/market_data.json`:
```json
{
"PA": {
"supplier_name": "Gaboro DME",
"branch_prefix": "PA",
"branch_id": "PA-001",
"state": "PA",
"payers": {
"medicare_ffs": ["Medicare Part B", "Medicare", "CMS"],
"medicare_advantage": ["UPMC Health Plan", "Highmark BlueCross BlueShield", "Aetna PA Medicare"],
"medicaid": ["PA Medicaid", "Keystone First", "AmeriHealth Caritas PA", "UPMC for You"],
"commercial": ["Independence Blue Cross", "Highmark BCBS PA", "Aetna PA", "Cigna PA", "Geisinger"]
},
"geographies": ["Philadelphia", "Bucks County", "Montgomery County", "Chester County"],
"prescribers": ["Penn Medicine", "Jefferson Health", "Temple Health", "Drexel Medicine"]
},
"NJ": {
"supplier_name": "Gaboro DME",
"branch_prefix": "NJ",
"branch_id": "NJ-002",
"state": "NJ",
"payers": {
"medicare_ffs": ["Medicare Part B", "Medicare", "Medicare FFS NJ"],
"medicare_advantage": ["Aetna NJ Medicare", "UnitedHealthcare NJ Medicare", "Horizon BCBS NJ Medicare"],
"medicaid": ["NJ FamilyCare", "WellCare NJ", "AmeriHealth NJ"],
"commercial": ["Horizon BCBS NJ", "Aetna NJ", "Cigna NJ", "Oxford Health", "UnitedHealthcare NJ"]
},
"geographies": ["Bergen County", "Essex County", "Hudson County", "Passaic County"],
"prescribers": ["Hackensack Meridian", "RWJBarnabas", "Atlantic Health", "Valley Health"]
}
}
```
- [ ] **Step 2: Commit**
```bash
cd /Users/sttil-solutions/projects/signal
git add test-data/market_data.json
git commit -m "feat: market_data.json — PA/NJ payer and geography config"
```
---
## Task 2: generate_supplier_set.py (parameterized core generator)
- [ ] **Step 1: Read the reference generator**
Read `test-data/generate_samples.py` in full before writing. Key patterns to follow:
- HEADER_VARIANTS list structure (tuple of column names + date format + extras dict)
- DATE_BUCKETS for status targeting
- Random seed for reproducibility
- Output naming pattern
- [ ] **Step 2: Create generate_supplier_set.py**
Create `test-data/generate_supplier_set.py`:
```python
"""
generate_supplier_set.py — Parameterized Signal CSV generator.
Produces N files for a given state/supplier/challenge-level combination.
Usage:
python generate_supplier_set.py --state PA --supplier "Gaboro DME" --patients 30 --challenge-level high --output pa-set
python generate_supplier_set.py --state NJ --supplier "Gaboro DME" --patients 25 --challenge-level high --output nj-set
Challenge levels:
low: canonical headers, clean dates, valid payers
medium: header variants, mixed date formats, some nulls
high: all variants + nulls + malformed dates + transfer patients + state-change patients
"""
import argparse
import csv
import json
import os
import random
from datetime import date, timedelta
random.seed(42)
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
MARKET_DATA_PATH = os.path.join(SCRIPT_DIR, "market_data.json")
DEVICE_TYPES = ["dexcom_g7", "dexcom_g6", "freestyle_libre_3", "freestyle_libre_2"]
COMPONENTS = {
"dexcom_g7": "sensor", "dexcom_g6": "sensor",
"freestyle_libre_3": "sensor", "freestyle_libre_2": "sensor",
}
TODAY = date.today()
DATE_BUCKETS = {
"active": (TODAY - timedelta(days=10), TODAY - timedelta(days=1)),
"renewal_soon": (TODAY - timedelta(days=270), TODAY - timedelta(days=260)),
"renewal_elevated": (TODAY - timedelta(days=300), TODAY - timedelta(days=290)),
"renewal_critical": (TODAY - timedelta(days=315), TODAY - timedelta(days=310)),
"visit_required": (TODAY - timedelta(days=400), TODAY - timedelta(days=380)),
"supply_lapsed": (TODAY - timedelta(days=600), TODAY - timedelta(days=500)),
}
# Header variants — (patient_id_col, device_col, date_col, qty_col, payer_col, component_col, date_fmt, extras_factory)
# extras_factory(state_data) → dict of extra columns to include
BASE_HEADER_VARIANTS = [
# 1 — canonical
("patient_id", "device_type", "shipment_date", "quantity", "payer", "component",
"%Y-%m-%d", lambda sd: {}),
# 2 — Brightree-style
("Patient ID", "Item Description", "Service Date", "Qty", "Insurance Name", "Item Type",
"%m/%d/%Y", lambda sd: {"Prescriber NPI": "1234567890", "Branch": sd["branch_id"]}),
# 3 — all caps
("PT_ID", "DEVICE", "SHIP DATE", "UNITS", "CARRIER", "TYPE",
"%Y-%m-%d", lambda sd: {}),
# 4 — MRN + text date
("MRN", "Product Name", "Dispense Date", "Qty Dispensed", "Plan Name", "Supply Type",
"%d-%b-%Y", lambda sd: {"Supplier": sd["supplier_name"], "State": sd["state"]}),
# 5 — account_number
("Account Number", "Product", "Fill Date", "Count", "Primary Payer", "component",
"%m/%d/%y", lambda sd: {}),
# 6 — external ref + ISO datetime
("External Patient Ref", "Item", "Date of Service", "Qty Shipped", "Insurance", "item_type",
"%Y-%m-%dT%H:%M:%S", lambda sd: {"Notes": "batch export", "Region": sd["geographies"][0]}),
# 7 — Acct # + YYYYMMDD
("Acct #", "DME", "Order Date", "quantity", "plan", "component_type",
"%Y%m%d", lambda sd: {}),
# 8 — patientid no space + dispense date
("patientid", "devicetype", "dispensedate", "qty", "payername", "supplytype",
"%m/%d/%Y", lambda sd: {}),
# 9 — pt_id + dos
("pt_id", "product_type", "dos", "qty", "ins_name", "component",
"%Y-%m-%d", lambda sd: {}),
# 10 — account_no + hcpcs
("account_no", "hcpcs_description", "service_date", "units", "primary_payer", "supply_type",
"%m-%d-%Y", lambda sd: {"HCPCS Code": "A9277"}),
# 11 — patient_account + commercial
("patient_account", "description", "ship_date", "quantity_dispensed", "carrier", "component",
"%Y-%m-%d", lambda sd: {"Account Manager": "Staff"}),
# 12 — id + product
("id", "product", "fill_date", "qty_shipped", "payer", "item_type",
"%m/%d/%Y", lambda sd: {}),
# 13 — PT ID spaces
("PT ID", "Device Type", "Shipment Date", "Quantity", "Insurance", "Component",
"%Y-%m-%d", lambda sd: {"Region": sd["geographies"][1] if len(sd["geographies"]) > 1 else ""}),
# 14 — lowercase no space
("patientid", "devicetype", "dispensedate", "qty", "payername", "supplytype",
"%m/%d/%Y", lambda sd: {}),
# 15 — account_number + d/m/y
("account_number", "item_description", "order_date", "units_dispensed", "plan_name", "component",
"%d/%m/%Y", lambda sd: {"Facility": f"{sd['supplier_name']} Main"}),
# 16 — member_id
("member_id", "Product Type", "Service Date", "Qty", "Insurance Name", "Component Type",
"%Y-%m-%d", lambda sd: {}),
# 17 — acct no
("acct no", "dme description", "last ship date", "qty dispensed", "payer name", "type",
"%m/%d/%Y", lambda sd: {"Billing Cycle": "Monthly"}),
# 18 — external_patient_ref
("external_patient_ref", "equipment description", "service_date", "count", "primary_payer", "supply_type",
"%Y-%m-%dT%H:%M:%S", lambda sd: {}),
# 19 — with doc columns: swo_status, pa_status
("patient_id", "device_type", "shipment_date", "quantity", "payer", "component",
"%Y-%m-%d", lambda sd: {"swo_status": "On File", "pa_status": "Not Required", "diagnosis_on_file": "Yes"}),
# 20 — with visit_date column
("patient_id", "device_type", "shipment_date", "quantity", "payer", "component",
"%Y-%m-%d", lambda sd: {"visit_date": "", "pecos_verified": "Yes", "swo_status": "On File"}),
# 21 — PA status column variants
("Patient ID", "Item Description", "Service Date", "Qty", "Insurance Name", "Item Type",
"%m/%d/%Y", lambda sd: {"auth_status": "Approved", "swo_status": "On File"}),
# 22 — Prior Auth column
("account_no", "device_type", "shipment_date", "qty", "insurance", "component",
"%Y-%m-%d", lambda sd: {"Prior Auth": "Pending", "diagnosis_on_file": "Yes"}),
# 23 — with notes + auth numbers (noise columns)
("patient_id", "device_type", "shipment_date", "quantity", "payer", "component",
"%Y-%m-%d", lambda sd: {"Notes": "see chart", "Auth Number": "A12345", "Staff": "K.F."}),
# 24 — missing optional columns (no component, no quantity)
("patient_id", "device_type", "shipment_date", None, "payer", None,
"%Y-%m-%d", lambda sd: {}),
# 25 — transfer patient column
("patient_id", "device_type", "shipment_date", "quantity", "payer", "component",
"%Y-%m-%d", lambda sd: {"Transfer_From": "Prior Supplier LLC", "swo_status": "Pending"}),
]
def _pick_payer(state_data: dict, payer_type: str) -> str:
payers = state_data["payers"].get(payer_type, [])
return random.choice(payers) if payers else "Medicare Part B"
def _format_date(d: date, fmt: str) -> str:
return d.strftime(fmt)
def _random_date_in_bucket(bucket: tuple[date, date]) -> date:
start, end = bucket
days = (end - start).days
return start + timedelta(days=random.randint(0, days))
def _device_display(device: str, header_style: str) -> str:
"""Map device type to display string matching the header style."""
display_map = {
"dexcom_g7": {
"canonical": "dexcom_g7",
"display": "Dexcom G7",
"short": "G7",
},
"dexcom_g6": {
"canonical": "dexcom_g6",
"display": "Dexcom G6",
"short": "G6",
},
"freestyle_libre_3": {
"canonical": "freestyle_libre_3",
"display": "FreeStyle Libre 3",
"short": "FSL3",
},
"freestyle_libre_2": {
"canonical": "freestyle_libre_2",
"display": "FreeStyle Libre 2",
"short": "FSL2",
},
}
dm = display_map.get(device, {"canonical": device, "display": device, "short": device})
if header_style in ("Item Description", "Product Name", "product", "description", "dme description", "equipment description"):
return dm["display"]
if header_style in ("DEVICE", "DME"):
return dm["short"]
return dm["canonical"]
def _payer_display(payer_str: str, header_style: str) -> str:
return payer_str # payer strings are already in display format from market_data.json
def generate_state_change_patients(state_data: dict, file_index: int, num_files: int) -> list[dict]:
"""
Generate 6 state-change patients whose status evolves across the file sequence.
Returns patient records that change based on file_index.
"""
patients = []
progress = file_index / max(num_files - 1, 1) # 0.0 to 1.0
# Patient A: starts RENEWAL_SOON, confirmed in file 10, ACTIVE by file 20
if file_index < 10:
shipment_date = TODAY - timedelta(days=270)
visit_date = "" # No confirmed visit yet
swo_status = "On File"
elif file_index < 20:
shipment_date = TODAY - timedelta(days=10)
visit_date = (TODAY - timedelta(days=5)).strftime("%Y-%m-%d")
swo_status = "On File"
else:
shipment_date = TODAY - timedelta(days=5)
visit_date = (TODAY - timedelta(days=5)).strftime("%Y-%m-%d")
swo_status = "On File"
patients.append({
"patient_id": "SC-A-001",
"device_type": "dexcom_g7",
"shipment_date": shipment_date,
"quantity": 1,
"payer": _pick_payer(state_data, "medicare_ffs"),
"component": "sensor",
"visit_date": visit_date,
"swo_status": swo_status,
"pecos_verified": "Yes",
"pa_status": "Not Required",
"diagnosis_on_file": "Yes",
})
# Patient B: starts ACTIVE, slides to RENEWAL_SOON by file 15
base_visit_offset = 170 - int(progress * 90) # 170 → 80 days ago
patients.append({
"patient_id": "SC-B-002",
"device_type": "freestyle_libre_3",
"shipment_date": TODAY - timedelta(days=5),
"quantity": 1,
"payer": _pick_payer(state_data, "medicaid"),
"component": "sensor",
"visit_date": (TODAY - timedelta(days=base_visit_offset)).strftime("%Y-%m-%d"),
"swo_status": "On File",
"pecos_verified": "Yes",
"pa_status": "Approved",
"diagnosis_on_file": "Yes",
})
# Patient C: transfer patient appearing mid-sequence, clears by file 18
if file_index < 12:
transfer_from = "Prior Supplier LLC"
swo_status_c = "Pending"
else:
transfer_from = ""
swo_status_c = "On File"
patients.append({
"patient_id": "SC-C-003",
"device_type": "dexcom_g6",
"shipment_date": TODAY - timedelta(days=30),
"quantity": 1,
"payer": _pick_payer(state_data, "commercial"),
"component": "sensor",
"Transfer_From": transfer_from,
"swo_status": swo_status_c,
"pa_status": "Pending" if file_index < 12 else "Approved",
"diagnosis_on_file": "Yes" if file_index >= 12 else "No",
})
# Patient D: SUPPLY_LAPSED → new shipment after file 8 → ACTIVE
if file_index < 8:
shipment_d = TODAY - timedelta(days=600)
else:
shipment_d = TODAY - timedelta(days=5)
patients.append({
"patient_id": "SC-D-004",
"device_type": "dexcom_g7",
"shipment_date": shipment_d,
"quantity": 1,
"payer": _pick_payer(state_data, "medicare_advantage"),
"component": "sensor",
"swo_status": "On File",
"pa_status": "Approved" if file_index >= 8 else "Pending",
"diagnosis_on_file": "Yes",
})
# Patients E and F: simple active patients that stay stable (control group)
patients.append({
"patient_id": "SC-E-005",
"device_type": "freestyle_libre_2",
"shipment_date": TODAY - timedelta(days=3),
"quantity": 1,
"payer": _pick_payer(state_data, "medicare_ffs"),
"component": "sensor",
"visit_date": (TODAY - timedelta(days=30)).strftime("%Y-%m-%d"),
"swo_status": "On File",
"pecos_verified": "Yes",
"pa_status": "Not Required",
"diagnosis_on_file": "Yes",
})
patients.append({
"patient_id": "SC-F-006",
"device_type": "dexcom_g7",
"shipment_date": TODAY - timedelta(days=7),
"quantity": 1,
"payer": _pick_payer(state_data, "commercial"),
"component": "sensor",
"swo_status": "On File",
"pa_status": "Approved",
"diagnosis_on_file": "Yes",
})
return patients
def generate_challenge_rows(state_data: dict, challenge_level: str, file_index: int) -> list[dict]:
"""Generate challenge-type rows based on challenge_level."""
if challenge_level == "low":
return []
challenges = []
if challenge_level in ("medium", "high"):
# Null patient_id (should be skipped)
challenges.append({
"patient_id": "",
"device_type": "dexcom_g7",
"shipment_date": TODAY - timedelta(days=10),
"quantity": 1,
"payer": _pick_payer(state_data, "medicare_ffs"),
"component": "sensor",
})
if challenge_level == "high":
# Future shipment date (edge case)
challenges.append({
"patient_id": f"FUTURE-{file_index:03d}",
"device_type": "dexcom_g7",
"shipment_date": TODAY + timedelta(days=30),
"quantity": 1,
"payer": _pick_payer(state_data, "medicare_ffs"),
"component": "sensor",
})
# Zero quantity
challenges.append({
"patient_id": f"ZERQTY-{file_index:03d}",
"device_type": "freestyle_libre_3",
"shipment_date": TODAY - timedelta(days=5),
"quantity": 0,
"payer": _pick_payer(state_data, "commercial"),
"component": "sensor",
})
# Duplicate patient_id (second one should be caught or processed)
challenges.append({
"patient_id": f"DUP-PT-{file_index:03d}",
"device_type": "dexcom_g7",
"shipment_date": TODAY - timedelta(days=5),
"quantity": 1,
"payer": _pick_payer(state_data, "medicare_ffs"),
"component": "sensor",
})
challenges.append({
"patient_id": f"DUP-PT-{file_index:03d}", # duplicate
"device_type": "freestyle_libre_3",
"shipment_date": TODAY - timedelta(days=3),
"quantity": 1,
"payer": _pick_payer(state_data, "commercial"),
"component": "sensor",
})
return challenges
def generate_patient_batch(
state_data: dict,
patient_count: int,
file_index: int,
num_files: int,
challenge_level: str,
) -> list[dict]:
"""Generate the main patient batch for one file."""
patients = []
payer_types = ["medicare_ffs", "medicare_ffs", "medicare_advantage", "medicaid", "commercial"]
status_buckets = [
"active", "active", "active", "active",
"renewal_soon", "renewal_elevated", "renewal_critical",
"visit_required", "supply_lapsed",
]
for i in range(patient_count):
device = random.choice(DEVICE_TYPES)
payer_type = random.choice(payer_types)
payer_str = _pick_payer(state_data, payer_type)
bucket_name = status_buckets[i % len(status_buckets)]
shipment_date = _random_date_in_bucket(DATE_BUCKETS[bucket_name])
qty = random.choice([1, 1, 1, 2, 3])
geo = random.choice(state_data["geographies"])
prescriber = random.choice(state_data["prescribers"])
patient = {
"patient_id": f"{state_data['branch_prefix']}-{file_index:02d}-{i+1:04d}",
"device_type": device,
"shipment_date": shipment_date,
"quantity": qty,
"payer": payer_str,
"component": COMPONENTS.get(device, "sensor"),
}
# Add doc fields for PA/NJ scenarios
if payer_type == "medicare_ffs":
patient.update({
"swo_status": random.choice(["On File", "On File", "Pending"]),
"pecos_verified": random.choice(["Yes", "Yes", "No"]),
"pa_status": "Not Required",
"diagnosis_on_file": random.choice(["Yes", "Yes", "No"]),
})
elif payer_type == "medicare_advantage":
patient.update({
"swo_status": random.choice(["On File", "Pending"]),
"pecos_verified": random.choice(["Yes", "No"]),
"pa_status": random.choice(["Approved", "Pending", "Denied"]),
"diagnosis_on_file": random.choice(["Yes", "No"]),
})
else:
patient.update({
"swo_status": random.choice(["On File", "Pending", "Expired"]),
"pa_status": random.choice(["Approved", "Pending", "Not Required"]),
"diagnosis_on_file": random.choice(["Yes", "No"]),
})
# Add visit date for some patients
if random.random() > 0.4:
visit_offset = random.randint(30, 200)
patient["visit_date"] = (TODAY - timedelta(days=visit_offset)).strftime("%Y-%m-%d")
else:
patient["visit_date"] = ""
patients.append(patient)
return patients
def write_csv_file(
output_path: str,
patients: list[dict],
variant: tuple,
state_data: dict,
challenge_level: str,
file_index: int,
) -> None:
"""Write one CSV file using the given header variant."""
pid_col, dev_col, date_col, qty_col, payer_col, comp_col, date_fmt, extras_factory = variant
extras = extras_factory(state_data)
headers = [h for h in [pid_col, dev_col, date_col, qty_col, payer_col, comp_col] if h is not None]
# Add doc columns if they appear in patients
doc_cols = ["visit_date", "swo_status", "pecos_verified", "pa_status", "diagnosis_on_file", "Transfer_From"]
doc_headers_in_extras = {k for k in extras if k in doc_cols}
extra_headers = [k for k in extras if k not in doc_cols]
# Malformed date files (2 per set at indices 7 and 14)
if challenge_level == "high" and file_index in (7, 14):
malformed_patients = []
for p in patients[:3]:
mp = dict(p)
mp["_malformed_date"] = True
malformed_patients.append(mp)
patients = malformed_patients + patients[3:]
with open(output_path, "w", newline="") as f:
all_headers = headers + list(doc_headers_in_extras) + extra_headers
# Some files: include doc columns explicitly
if any(col in extras for col in doc_cols):
for col in doc_cols:
if col in extras and col not in all_headers:
all_headers.append(col)
writer = csv.DictWriter(f, fieldnames=all_headers, extrasaction="ignore")
writer.writeheader()
for p in patients:
row = {}
# Map patient dict to column headers
if pid_col:
row[pid_col] = p.get("patient_id", "")
if dev_col:
row[dev_col] = _device_display(p.get("device_type", "dexcom_g7"), dev_col)
if date_col:
sd = p.get("shipment_date", TODAY)
if isinstance(sd, date):
if hasattr(p, "_malformed_date") or p.get("_malformed_date"):
row[date_col] = random.choice(["not on file", "pending", "TBD"])
else:
row[date_col] = _format_date(sd, date_fmt)
else:
row[date_col] = str(sd)
if qty_col:
row[qty_col] = p.get("quantity", 1)
if payer_col:
row[payer_col] = p.get("payer", "Medicare Part B")
if comp_col:
row[comp_col] = p.get("component", "sensor")
# Doc fields
for col in doc_cols:
if col in all_headers:
row[col] = p.get(col, "")
# Extras
for k, v in extras.items():
if k in all_headers:
row[k] = v
writer.writerow(row)
def generate_set(
state: str,
supplier: str,
patients: int,
challenge_level: str,
output_dir: str,
num_files: int = 25,
) -> None:
"""Generate num_files CSV files for the given state."""
with open(MARKET_DATA_PATH) as f:
market = json.load(f)
state_data = market.get(state.upper())
if not state_data:
raise ValueError(f"State '{state}' not found in market_data.json. Available: {list(market.keys())}")
state_data["supplier_name"] = supplier # allow override
os.makedirs(output_dir, exist_ok=True)
for i in range(num_files):
# Rotate through header variants
variant = BASE_HEADER_VARIANTS[i % len(BASE_HEADER_VARIANTS)]
# Core patient batch
patient_rows = generate_patient_batch(state_data, patients, i, num_files, challenge_level)
# State-change patients (appear in every file)
sc_patients = generate_state_change_patients(state_data, i, num_files)
all_patients = sc_patients + patient_rows
# Challenge rows
if challenge_level == "high":
challenge_rows = generate_challenge_rows(state_data, challenge_level, i)
all_patients = all_patients + challenge_rows
filename = f"batch-{i+1:02d}-{state.lower()}-{challenge_level}.csv"
output_path = os.path.join(output_dir, filename)
write_csv_file(output_path, all_patients, variant, state_data, challenge_level, i)
print(f" Written: {filename} ({len(all_patients)} rows)")
print(f"\nGenerated {num_files} files in {output_dir}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Generate Signal CSV test files")
parser.add_argument("--state", default="PA")
parser.add_argument("--supplier", default="Gaboro DME")
parser.add_argument("--patients", type=int, default=30)
parser.add_argument("--challenge-level", choices=["low", "medium", "high"], default="high")
parser.add_argument("--output", default="pa-set")
parser.add_argument("--files", type=int, default=25)
args = parser.parse_args()
output_path = os.path.join(SCRIPT_DIR, args.output)
generate_set(
state=args.state,
supplier=args.supplier,
patients=args.patients,
challenge_level=args.challenge_level,
output_dir=output_path,
num_files=args.files,
)
```
- [ ] **Step 3: Commit**
```bash
cd /Users/sttil-solutions/projects/signal
git add test-data/generate_supplier_set.py
git commit -m "feat: parameterized CSV generator with state-change patients and challenge types"
```
---
## Task 3: State-specific wrappers + generate PA and NJ sets
- [ ] **Step 1: Create generate_pa_set.py**
Create `test-data/generate_pa_set.py`:
```python
"""Generate 25 PA-market CSV files (Philadelphia/Southeastern PA)."""
import os, sys
sys.path.insert(0, os.path.dirname(__file__))
from generate_supplier_set import generate_set
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if __name__ == "__main__":
generate_set(
state="PA",
supplier="Gaboro DME",
patients=30,
challenge_level="high",
output_dir=os.path.join(SCRIPT_DIR, "pa-set"),
num_files=25,
)
print("PA set complete.")
```
- [ ] **Step 2: Create generate_nj_set.py**
Create `test-data/generate_nj_set.py`:
```python
"""Generate 25 NJ-market CSV files (Northern NJ / Metro NY)."""
import os, sys
sys.path.insert(0, os.path.dirname(__file__))
from generate_supplier_set import generate_set
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if __name__ == "__main__":
generate_set(
state="NJ",
supplier="Gaboro DME",
patients=25,
challenge_level="high",
output_dir=os.path.join(SCRIPT_DIR, "nj-set"),
num_files=25,
)
print("NJ set complete.")
```
- [ ] **Step 3: Run both generators**
```bash
cd /Users/sttil-solutions/projects/signal/test-data
python generate_pa_set.py
python generate_nj_set.py
```
Expected: 25 files in `test-data/pa-set/` and 25 files in `test-data/nj-set/`.
Verify file count:
```bash
ls test-data/pa-set/ | wc -l && ls test-data/nj-set/ | wc -l
```
Expected: `25` on each line.
- [ ] **Step 4: Commit**
```bash
cd /Users/sttil-solutions/projects/signal
git add test-data/generate_pa_set.py test-data/generate_nj_set.py test-data/pa-set/ test-data/nj-set/
git commit -m "feat: generate PA and NJ CSV test sets (50 files total)"
```
---
## Task 4: Bug Hunter — run all 50 CSVs through normalizer
- [ ] **Step 1: Create bug_hunter.py**
Create `test-data/bug_hunter.py`:
```python
"""
bug_hunter.py — Run all pa-set and nj-set CSVs through the Signal normalizer.
Reports: filename, error type, line, recommendation.
Usage:
python test-data/bug_hunter.py > test-data/bug-report.txt
"""
import os
import sys
import json
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(SCRIPT_DIR, "..", "python-backend"))
from api.normalizer import normalize_csv
SETS = [
os.path.join(SCRIPT_DIR, "pa-set"),
os.path.join(SCRIPT_DIR, "nj-set"),
]
RESULTS = {
"total_files": 0,
"total_records": 0,
"total_skipped": 0,
"files_with_errors": [],
"all_skip_reasons": [],
}
def hunt(directory: str) -> None:
if not os.path.exists(directory):
print(f"SKIP: {directory} does not exist")
return
files = sorted(f for f in os.listdir(directory) if f.endswith(".csv"))
for filename in files:
path = os.path.join(directory, filename)
RESULTS["total_files"] += 1
try:
text = open(path, encoding="utf-8", errors="replace").read()
records, skipped, mapping_summary = normalize_csv(text)
RESULTS["total_records"] += len(records)
RESULTS["total_skipped"] += len(skipped)
required_missing = mapping_summary.get("required_missing", [])
file_issues = []
if required_missing:
file_issues.append({
"type": "REQUIRED_COLUMNS_MISSING",
"detail": f"Missing: {required_missing}",
"recommendation": "Update HEADER_MAP in normalizer.py to cover these column names",
})
for reason in skipped:
file_issues.append({
"type": "ROW_SKIPPED",
"detail": reason,
"recommendation": "Check date format, device name, or patient_id for this row",
})
RESULTS["all_skip_reasons"].append(f"{filename}: {reason}")
if file_issues:
RESULTS["files_with_errors"].append({
"file": filename,
"records_processed": len(records),
"issues": file_issues,
})
except Exception as e:
RESULTS["files_with_errors"].append({
"file": filename,
"records_processed": 0,
"issues": [{"type": "EXCEPTION", "detail": str(e), "recommendation": "Fix the exception before pilot"}],
})
print(f"Processed directory: {directory} ({len(files)} files)")
if __name__ == "__main__":
for d in SETS:
hunt(d)
print("\n=== BUG HUNTER REPORT ===")
print(f"Total files: {RESULTS['total_files']}")
print(f"Total records processed: {RESULTS['total_records']}")
print(f"Total rows skipped: {RESULTS['total_skipped']}")
print(f"Files with issues: {len(RESULTS['files_with_errors'])}")
if RESULTS["files_with_errors"]:
print("\n--- FILES WITH ISSUES ---")
for entry in RESULTS["files_with_errors"]:
print(f"\nFile: {entry['file']} ({entry['records_processed']} records OK)")
for issue in entry["issues"]:
print(f" [{issue['type']}] {issue['detail']}")
print(f" Fix: {issue['recommendation']}")
else:
print("\nAll files processed successfully. Zero issues found.")
print("\n=== END REPORT ===")
```
- [ ] **Step 2: Run Bug Hunter**
```bash
cd /Users/sttil-solutions/projects/signal && python test-data/bug_hunter.py 2>&1 | tee test-data/bug-report.txt
```
- [ ] **Step 3: Fix issues found by Bug Hunter**
For each issue in the bug report:
- `REQUIRED_COLUMNS_MISSING`: add the missing column name variants to `HEADER_MAP` in `python-backend/api/normalizer.py`
- `ROW_SKIPPED` due to date format: add the date format to `DATE_FORMATS` list in normalizer.py
- `ROW_SKIPPED` due to device name: add the device alias to `DEVICE_MAP` in normalizer.py
- `EXCEPTION`: fix the root cause in the generator or normalizer
After fixes, re-run Bug Hunter until output shows "Zero issues found" or only expected skips (intentionally malformed rows).
- [ ] **Step 4: Commit**
```bash
cd /Users/sttil-solutions/projects/signal
git add test-data/bug_hunter.py test-data/bug-report.txt
git commit -m "feat: bug_hunter.py — 50-CSV normalizer stress test; report saved"
```
Reference in a new issue View git blame Copy permalink