Open raw ↗"""Seed demo data for HAD Digital MVP.
Creates realistic demo data with French names, real treatment protocols,
3 patients with different cancer types and treatment plans.
"""
from datetime import datetime, timedelta, timezone
from database import get_db
from user_store import create_user
from ctcae_engine import get_ctcae_engine
from alert_engine import get_alert_engine
from audit_logger import log_action
def seed_demo_data():
"""Seed all demo data. Safe to call multiple times (idempotent)."""
db = get_db()
# Check if already seeded
count = db.fetchone("SELECT COUNT(*) as cnt FROM users")
if count and count["cnt"] > 0:
print("[seed] Demo data already exists, skipping.")
return
print("[seed] Seeding demo data...")
# Create patients FIRST (before users that reference them)
patients = _create_patients()
print(f"[seed] Created {len(patients)} patients")
# Create users (some reference patient IDs)
users = _create_users(patients)
print(f"[seed] Created {len(users)} users")
# Update patients with their primary oncologist
_update_patient_owners(users)
# Create episodes and treatment plans
episodes = _create_episodes(patients, users)
print(f"[seed] Created {len(episodes)} episodes")
# Create sample toxicity reports
reports = _create_reports(patients, users)
print(f"[seed] Created {len(reports)} toxicity reports")
# Create sample alerts
alerts = _create_alerts(patients, users)
print(f"[seed] Created {len(alerts)} alerts")
# Create sample messages
messages = _create_messages(patients, users)
print(f"[seed] Created {len(messages)} messages")
# Create timeline events
events = _create_timeline(patients, users)
print(f"[seed] Created {len(events)} timeline events")
log_action(
user_id=users["admin"]["id"],
action="seed_demo_data",
resource_type="system",
details="Demo data seeded successfully",
)
print("[seed] Demo data seeding complete!")
def _create_patients() -> list:
"""Create demo patients with realistic medical data (before users)."""
db = get_db()
patients = []
# Patient 1: Breast cancer - Jeanne Durand
db.execute(
"""INSERT INTO patients
(mrn, first_name, last_name, date_of_birth, gender, phone, address,
emergency_contact, cancer_type, cancer_stage, diagnosis_date, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
"MRN-2024-001", "Jeanne", "Durand", "1958-03-15", "F",
"+33 6 12 34 56 78", "15 Rue de la Paix, 75002 Paris",
"Jacques Durand (epoux): +33 6 98 76 54 32",
"Cancer du sein", "IIB", "2024-01-10", "active",
),
)
patients.append({"id": 1, "name": "Jeanne Durand", "mrn": "MRN-2024-001"})
# Patient 2: Lung cancer - Pierre Moreau
db.execute(
"""INSERT INTO patients
(mrn, first_name, last_name, date_of_birth, gender, phone, address,
emergency_contact, cancer_type, cancer_stage, diagnosis_date, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
"MRN-2024-002", "Pierre", "Moreau", "1965-07-22", "M",
"+33 6 23 45 67 89", "8 Avenue Victor Hugo, 69003 Lyon",
"Marie Moreau (epouse): +33 6 87 65 43 21",
"Cancer du poumon", "IIIA", "2024-02-05", "active",
),
)
patients.append({"id": 2, "name": "Pierre Moreau", "mrn": "MRN-2024-002"})
# Patient 3: Colorectal cancer - Marie-Claire Laurent
db.execute(
"""INSERT INTO patients
(mrn, first_name, last_name, date_of_birth, gender, phone, address,
emergency_contact, cancer_type, cancer_stage, diagnosis_date, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
"MRN-2024-003", "Marie-Claire", "Laurent", "1972-11-08", "F",
"+33 6 34 56 78 90", "22 Boulevard Gambetta, 13001 Marseille",
"Antoine Laurent (fils): +33 6 76 54 32 10",
"Cancer colorectal", "IIIB", "2024-03-01", "active",
),
)
patients.append({"id": 3, "name": "Marie-Claire Laurent", "mrn": "MRN-2024-003"})
db.commit()
return patients
def _create_users(patients: list) -> dict:
"""Create demo users with different roles."""
users = {}
# Admin
users["admin"] = {
"id": create_user(
username="admin", password="admin123", role="admin",
display_name="Marie Dupont", email="[email protected]",
), "role": "admin",
}
# Oncologists
users["oncologist1"] = {
"id": create_user(
username="dr.martin", password="demo123", role="oncologist",
display_name="Dr. Jean-Pierre Martin", email="[email protected]",
), "role": "oncologist",
}
users["oncologist2"] = {
"id": create_user(
username="dr.bernard", password="demo123", role="oncologist",
display_name="Dr. Sophie Bernard", email="[email protected]",
), "role": "oncologist",
}
# HAD Nurses
users["had_nurse1"] = {
"id": create_user(
username="inf.moret", password="demo123", role="had_nurse",
display_name="Infirmiere Claire Moret", email="[email protected]",
), "role": "had_nurse",
}
users["had_nurse2"] = {
"id": create_user(
username="inf.petit", password="demo123", role="had_nurse",
display_name="Infirmiere Lucie Petit", email="[email protected]",
), "role": "had_nurse",
}
# Community Nurse
users["community_nurse"] = {
"id": create_user(
username="inf.dubois", password="demo123", role="community_nurse",
display_name="Infirmiere Isabelle Dubois", email="[email protected]",
), "role": "community_nurse",
}
# GP
users["gp"] = {
"id": create_user(
username="dr.lefevre", password="demo123", role="gp",
display_name="Dr. Philippe Lefevre", email="[email protected]",
), "role": "gp",
}
# Pharmacist
users["pharmacist"] = {
"id": create_user(
username="ph.robert", password="demo123", role="pharmacist",
display_name="Pharmacien Nicolas Robert", email="[email protected]",
), "role": "pharmacist",
}
# Patients (reference patient IDs)
users["patient1"] = {
"id": create_user(
username="patient.durand", password="demo123", role="patient",
display_name="Jeanne Durand", email="[email protected]", patient_id=1,
), "role": "patient",
}
users["patient2"] = {
"id": create_user(
username="patient.moreau", password="demo123", role="patient",
display_name="Pierre Moreau", email="[email protected]", patient_id=2,
), "role": "patient",
}
users["patient3"] = {
"id": create_user(
username="patient.laurent", password="demo123", role="patient",
display_name="Marie-Claire Laurent", email="[email protected]", patient_id=3,
), "role": "patient",
}
# Caregiver
users["caregiver"] = {
"id": create_user(
username="aidant.durand", password="demo123", role="caregiver",
display_name="Jacques Durand (epoux)", email="[email protected]", patient_id=1,
), "role": "caregiver",
}
return users
def _update_patient_owners(users: dict):
"""Assign primary oncologists to patients."""
db = get_db()
db.execute("UPDATE patients SET primary_oncologist_id = ? WHERE id = ?", (users["oncologist1"]["id"], 1))
db.execute("UPDATE patients SET primary_oncologist_id = ? WHERE id = ?", (users["oncologist2"]["id"], 2))
db.execute("UPDATE patients SET primary_oncologist_id = ? WHERE id = ?", (users["oncologist1"]["id"], 3))
db.commit()
def _create_episodes(patients: list, users: dict) -> list:
"""Create treatment episodes and plans."""
db = get_db()
episodes = []
now = datetime.now(timezone.utc)
# Patient 1: Breast cancer - AC-T protocol
db.execute(
"INSERT INTO episodes (patient_id, episode_type, start_date, status, notes) VALUES (?, ?, ?, ?, ?)",
(1, "chemotherapy", (now - timedelta(days=60)).strftime("%Y-%m-%d"), "active",
"Protocole AC-T pour cancer du sein stade IIB"),
)
episodes.append({"id": 1, "patient_id": 1})
db.execute(
"INSERT INTO treatment_plans (patient_id, episode_id, protocol_name, regimen, cycle_count, current_cycle, cycle_length_days, drugs, start_date, status, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(1, 1, "AC-T", "Adriamycine-Cyclophosphamide suivie de Taxol", 8, 4, 21,
"Cycles 1-4: Doxorubicine 60mg/m2 + Cyclophosphamide 600mg/m2; Cycles 5-8: Paclitaxel 175mg/m2",
(now - timedelta(days=60)).strftime("%Y-%m-%d"), "active", "Cycles 1-4 AC termines. Paclitaxel en cours."),
)
# Patient 2: Lung cancer - Cisplatin-Pemetrexed
db.execute(
"INSERT INTO episodes (patient_id, episode_type, start_date, status, notes) VALUES (?, ?, ?, ?, ?)",
(2, "chemotherapy", (now - timedelta(days=30)).strftime("%Y-%m-%d"), "active",
"Protocole Cisplatine-Pemetrexed pour cancer du poumon non a petites cellules"),
)
episodes.append({"id": 2, "patient_id": 2})
db.execute(
"INSERT INTO treatment_plans (patient_id, episode_id, protocol_name, regimen, cycle_count, current_cycle, cycle_length_days, drugs, start_date, status, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(2, 2, "CisPem", "Cisplatine-Pemetrexed", 6, 2, 21,
"Cisplatine 75mg/m2 J1 + Pemetrexed 500mg/m2 J1",
(now - timedelta(days=30)).strftime("%Y-%m-%d"), "active", "2 cycles termines. Bonne tolerance initiale."),
)
# Patient 3: Colorectal cancer - FOLFOX
db.execute(
"INSERT INTO episodes (patient_id, episode_type, start_date, status, notes) VALUES (?, ?, ?, ?, ?)",
(3, "chemotherapy", (now - timedelta(days=45)).strftime("%Y-%m-%d"), "active",
"Protocole FOLFOX pour cancer colorectal stade IIIB"),
)
episodes.append({"id": 3, "patient_id": 3})
db.execute(
"INSERT INTO treatment_plans (patient_id, episode_id, protocol_name, regimen, cycle_count, current_cycle, cycle_length_days, drugs, start_date, status, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(3, 3, "FOLFOX6", "FOLFOX6 modifie", 12, 3, 14,
"Oxaliplatine 85mg/m2 J1 + Leucovorine 400mg/m2 J1 + 5-FU 400mg/m2 bolus J1 + 5-FU 2400mg/m2 46h",
(now - timedelta(days=45)).strftime("%Y-%m-%d"), "active", "3 cycles termines. Neuropathie peripherique G1 notee."),
)
db.commit()
return episodes
def _create_reports(patients: list, users: dict) -> list:
"""Create sample toxicity reports with grading."""
db = get_db()
engine = get_ctcae_engine()
reports = []
now = datetime.now(timezone.utc)
# Patient 1: Nausea G2
cur = db.execute(
"INSERT INTO toxicity_reports (patient_id, episode_id, treatment_plan_id, reporter_id, symptom_id, symptom_category, report_date, severity_score, notes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(1, 1, 1, users["had_nurse1"]["id"], "nausea", "gastrointestinal",
(now - timedelta(days=3)).strftime("%Y-%m-%d %H:%M:%S"), 5,
"Nausee moderee depuis 2 jours. Perte d appetit mais hydratation conservee.", "pending"))
rid = cur.lastrowid
reports.append({"id": rid, "patient_id": 1})
engine.grade_and_save(rid, 1, "nausea", {"severity_score": 5})
# Patient 1: Fatigue G3
cur = db.execute(
"INSERT INTO toxicity_reports (patient_id, episode_id, treatment_plan_id, reporter_id, symptom_id, symptom_category, report_date, severity_score, notes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(1, 1, 1, users["patient1"]["id"], "fatigue", "constitutional",
(now - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S"), 7,
"Fatigue severe empechant les activites quotidiennes.", "pending"))
rid = cur.lastrowid
reports.append({"id": rid, "patient_id": 1})
engine.grade_and_save(rid, 1, "fatigue", {"severity_score": 7})
# Patient 1: Alopecia G2
cur = db.execute(
"INSERT INTO toxicity_reports (patient_id, episode_id, treatment_plan_id, reporter_id, symptom_id, symptom_category, report_date, severity_score, notes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(1, 1, 1, users["had_nurse1"]["id"], "alopecia", "dermatologic",
(now - timedelta(days=5)).strftime("%Y-%m-%d %H:%M:%S"), None,
"Perte de cheveux significative depuis cycle 3.", "pending"))
rid = cur.lastrowid
reports.append({"id": rid, "patient_id": 1})
engine.grade_and_save(rid, 1, "alopecia", {"hair_loss_percent": 60})
# Patient 2: Neutropenia G3
cur = db.execute(
"INSERT INTO toxicity_reports (patient_id, episode_id, treatment_plan_id, reporter_id, symptom_id, symptom_category, report_date, lab_values, notes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(2, 2, 2, users["oncologist2"]["id"], "neutropenia", "hematologic",
(now - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"),
'{"anc_mm3": 450}', "NFS J14: ANC 450/mm3. Pas de fievre.", "pending"))
rid = cur.lastrowid
reports.append({"id": rid, "patient_id": 2})
engine.grade_and_save(rid, 2, "neutropenia", {"anc_mm3": 450})
# Patient 2: Anemia G2
cur = db.execute(
"INSERT INTO toxicity_reports (patient_id, episode_id, treatment_plan_id, reporter_id, symptom_id, symptom_category, report_date, lab_values, notes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(2, 2, 2, users["oncologist2"]["id"], "anemia", "hematologic",
(now - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"),
'{"hemoglobin_gdl": 9.2}', "Hb 9.2 g/dL. Fatigue moderee.", "pending"))
rid = cur.lastrowid
reports.append({"id": rid, "patient_id": 2})
engine.grade_and_save(rid, 2, "anemia", {"hemoglobin_gdl": 9.2})
# Patient 3: Peripheral neuropathy G2
cur = db.execute(
"INSERT INTO toxicity_reports (patient_id, episode_id, treatment_plan_id, reporter_id, symptom_id, symptom_category, report_date, severity_score, notes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(3, 3, 3, users["community_nurse"]["id"], "peripheral_neuropathy", "neurologic",
(now - timedelta(days=4)).strftime("%Y-%m-%d %H:%M:%S"), 5,
"Picotements dans les mains et les pieds. Difficulte a fermer les boutons.", "pending"))
rid = cur.lastrowid
reports.append({"id": rid, "patient_id": 3})
engine.grade_and_save(rid, 3, "peripheral_neuropathy", {"severity_score": 5})
# Patient 3: Diarrhea G1
cur = db.execute(
"INSERT INTO toxicity_reports (patient_id, episode_id, treatment_plan_id, reporter_id, symptom_id, symptom_category, report_date, severity_score, notes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(3, 3, 3, users["patient3"]["id"], "diarrhea", "gastrointestinal",
(now - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S"), None,
"3-4 selles liquides par jour depuis 3 jours.", "pending"))
rid = cur.lastrowid
reports.append({"id": rid, "patient_id": 3})
engine.grade_and_save(rid, 3, "diarrhea", {"stools_increase_per_day": 3})
db.commit()
return reports
def _create_alerts(patients: list, users: dict) -> list:
"""Create sample alerts from existing grades."""
db = get_db()
ae = get_alert_engine()
alerts = []
grades = db.fetchall("SELECT * FROM toxicity_grades ORDER BY id")
for g in grades:
if g["grade"] >= 2:
alert = ae.create_alert_for_grade(
grade_id=g["id"], patient_id=g["patient_id"], report_id=g["report_id"],
grade=g["grade"], symptom_id=g["symptom_id"],
display_name=g["symptom_id"].replace("_", " ").title(), reporter_id=users["admin"]["id"])
if alert:
alerts.append(alert)
return alerts
def _create_messages(patients: list, users: dict) -> list:
"""Create sample messages."""
db = get_db()
messages = []
msgs = [
(users["oncologist1"]["id"], users["had_nurse1"]["id"], 1, "Patient Jeanne Durand - Cycle 5",
"Bonjour Claire,\n\nJeanne Durand commence le Paclitaxel la semaine prochaine. Surveiller particulierement les neuropathies et les myalgies.\n\nLe bilan pre-therapeutique est OK.\n\nCordialement,\nDr. Martin", "care_team"),
(users["had_nurse1"]["id"], users["oncologist1"]["id"], 1, "Re: Patient Jeanne Durand - Cycle 5",
"Bonjour Docteur,\n\nBien note. J\'ai programme la visite pour lundi prochain.\nLa patiente signale une fatigue importante depuis le dernier cycle.\n\nCordialement,\nClaire Moret", "care_team"),
(users["gp"]["id"], users["oncologist2"]["id"], 2, "Pierre Moreau - Consultation de suivi",
"Cher collegue,\n\nJ\'ai revu Pierre Moreau aujourd\'hui. Il se plaint de fatigue et d\'une perte d\'appetit moderee.\nLe bilan sanguin montre une anemie a 9.2 g/dL.\n\nCordialement,\nDr. Lefevre", "care_team"),
(users["pharmacist"]["id"], users["had_nurse1"]["id"], 3, "Medicaments Marie-Claire Laurent",
"Bonjour,\n\nLes medicaments suivants sont prets pour la patiente:\n- Oxaliplatine 130mg\n- Leucovorine 400mg\n- 5-FU 2400mg\n\nLivraison prevue pour vendredi.\n\nCordialement,\nN. Robert", "care_team"),
(users["patient1"]["id"], users["had_nurse1"]["id"], 1, "Question sur les medicaments",
"Bonjour Claire,\n\nJe voulais vous demander si je peux prendre du paracetamol contre les maux de tete ? J\'ai aussi des nausees depuis hier.\n\nMerci,\nJeanne", "direct"),
]
for sender, recip, pid, subj, body, mtype in msgs:
cur = db.execute("INSERT INTO messages (sender_id, recipient_id, patient_id, subject, body, message_type) VALUES (?, ?, ?, ?, ?, ?)",
(sender, recip, pid, subj, body, mtype))
messages.append({"id": cur.lastrowid})
db.commit()
return messages
def _create_timeline(patients: list, users: dict) -> list:
"""Create timeline events."""
db = get_db()
events = []
now = datetime.now(timezone.utc)
data = [
(1, "treatment", 20, "Cycle 4 AC termine", "Doxorubicine 60mg/m2 + Cyclophosphamide 600mg/m2 administres sans incident.", users["oncologist1"]["id"]),
(1, "report", 3, "Rapport de toxicite soumis", "Nausee G2 signalee par l\'infirmiere.", users["had_nurse1"]["id"]),
(1, "grade", 3, "Grade CTCAE attribue", "Nausee: Grade 2 (moderee). Antiemetiques prescrits.", None),
(1, "alert", 2, "Alerte urgente generee", "Fatigue G3: notification envoyee a l\'equipe soignante.", None),
(1, "appointment", 1, "Consultation de suivi programmee", "Rendez-vous avec Dr. Martin pour evaluation pre-cycle 5.", users["oncologist1"]["id"]),
(2, "treatment", 10, "Cycle 2 CisPem termine", "Cisplatine 75mg/m2 + Pemetrexed 500mg/m2. Bonne tolerance.", users["oncologist2"]["id"]),
(2, "report", 1, "Bilan hematologique", "NFS J14: ANC 450/mm3 (G3), Hb 9.2 (G2).", users["oncologist2"]["id"]),
(2, "alert", 1, "Alerte emergency generee", "Neutropenie G3: notification immediate a l\'equipe.", None),
(3, "treatment", 7, "Cycle 3 FOLFOX termine", "Oxaliplatine 85mg/m2 + 5-FU. Neuropathie peripherique notee.", users["oncologist1"]["id"]),
(3, "report", 4, "Rapport neuropathie", "Neuropathie peripherique G2 signalee par la patiente.", users["community_nurse"]["id"]),
(3, "note", 3, "Note de suivi", "Patientese signale des difficultes a manipuler de petits objets. Dose d\'oxaliplatine a reevaluer.", users["oncologist1"]["id"]),
]
for pid, etype, days_ago, title, desc, created_by in data:
event_date = (now - timedelta(days=days_ago)).strftime("%Y-%m-%d %H:%M:%S")
cur = db.execute("INSERT INTO timeline_events (patient_id, event_type, event_date, title, description, created_by) VALUES (?, ?, ?, ?, ?, ?)",
(pid, etype, event_date, title, desc, created_by))
events.append({"id": cur.lastrowid})
db.commit()
return events
if __name__ == "__main__":
seed_demo_data()