Spaces:
Sleeping
Sleeping
File size: 5,743 Bytes
1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 1f8d682 55da826 | 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 143 144 145 146 | """
MODULE: DATA INJECTOR (FLUX CONTINU)
====================================
Responsabilité : Injection contrôlée de nouveaux lots de données.
Mode : Append (Ajout à la suite sans écraser).
Style : Professionnel (Sans Emoji).
"""
import pandas as pd
import numpy as np
import datetime as dt
from faker import Faker
import streamlit as st
import random
class DataInjector:
def __init__(self, client, sheet_name):
self.client = client
self.sheet_name = sheet_name
self.fake = Faker("fr_FR")
# Services de l'hôpital
self.svc_codes = ["URG", "MED", "PED", "REA", "INF", "ISO", "CAR", "NEU", "TRA"]
def _append_to_sheet(self, df, worksheet_name):
"""Ajoute les données à la suite des lignes existantes dans Google Sheets"""
try:
sh = self.client.open(self.sheet_name)
worksheet = sh.worksheet(worksheet_name)
# Conversion en string pour l'API Google Sheets
df_export = df.copy()
for col in df_export.select_dtypes(include=['datetime64', 'object', 'datetimetz']).columns:
df_export[col] = df_export[col].astype(str)
# Extraction des valeurs brutes
values = df_export.values.tolist()
# Appel API
worksheet.append_rows(values)
return True
except Exception as e:
st.error(f"Erreur lors de l'ajout dans '{worksheet_name}': {e}")
return False
def generate_batch(self, days=7):
"""
Génère un lot de données pour X jours.
Volume estimé pour 7 jours : ~63 lignes Ressources + ~400 lignes Admissions + Stocks.
"""
# On simule à partir d'aujourd'hui pour avoir des données "fraîches"
start_date = dt.date.today()
dates = pd.date_range(start_date, periods=days, freq="D")
# 1. SCÉNARIOS (Météo/Épidémie)
scen_rows = []
for d in dates:
# Petit hasard pour rendre ça vivant
r = np.random.random()
if r < 0.1: sc = "Grippe"
elif r < 0.05: sc = "Canicule"
else: sc = "Normal"
scen_rows.append({"date": d, "scenario": sc, "intensity": 1 if sc != "Normal" else 0})
df_scen = pd.DataFrame(scen_rows)
# 2. ADMISSIONS (Le flux patients)
adm_rows = []
for d in dates:
for svc in self.svc_codes:
# Variation aléatoire pour tester la sensibilité de l'IA
n_patients = np.random.poisson(lam=random.randint(3, 12))
adm_rows.append({
"date": d,
"service_code": svc,
"nb_admissions": n_patients
})
df_adm = pd.DataFrame(adm_rows)
# 3. RESSOURCES (L'état des lits)
res_rows = []
for d in dates:
for svc in self.svc_codes:
# Simulation d'une occupation variable
lits_tot = 150
occup = np.random.randint(80, 155) # Parfois ça déborde (>150)
# Injection d'anomalies pour tester le module DQ
is_anomaly = 1 if occup > lits_tot else 0
anomaly_type = "suroccupation" if is_anomaly else None
res_rows.append({
"date": d,
"service_code": svc,
"lits_total": lits_tot,
"lits_occupes": occup,
"personnel_disponible": np.random.randint(20, 35),
"personnel_requis": 30,
"taux_occupation": round(occup/lits_tot, 2),
"dq_is_anomaly": is_anomaly,
"dq_anomaly_type": anomaly_type
})
df_res = pd.DataFrame(res_rows)
return df_scen, df_adm, df_res
def run(self, days_to_add):
scen, adm, res = self.generate_batch(days=days_to_add)
total_rows = len(scen) + len(adm) + len(res)
with st.status(f"Injection de {total_rows} nouvelles lignes...", expanded=True) as status:
st.write("Ajout au calendrier...")
self._append_to_sheet(scen, "scenarios_calendrier")
st.write(f"Ajout : {len(adm)} admissions...")
self._append_to_sheet(adm, "admissions_jour")
st.write(f"Ajout : {len(res)} relevés de ressources...")
self._append_to_sheet(res, "ressources_jour")
status.update(label="Injection terminée !", state="complete")
return total_rows
# --- INTERFACE ---
def show_data_injector(client, sheet_name):
st.markdown("## INJECTEUR DE FLUX (LIVE TEST)")
st.caption("Ajoute des données récentes pour tester la réaction de l'IA et du Dashboard.")
col1, col2 = st.columns([2, 1])
with col1:
days = st.slider("Jours à simuler", min_value=1, max_value=14, value=7,
help="7 jours génèrent environ 500 nouvelles lignes.")
with col2:
# Estimation du volume
st.metric("Volume estimé", f"~{days * 9 * 6} lignes", delta="Nouveaux cas")
# MODIFICATION ICI : Suppression de type='primary' pour éviter la couleur rouge
if st.button("Injecter les données maintenant"):
injector = DataInjector(client, sheet_name)
count = injector.run(days)
st.success(f"Opération réussie. {count} nouvelles lignes ont été ajoutées à la base.")
st.warning("Note : Relancez l'ETL (Feature Engineering) pour que l'IA prenne en compte ces nouvelles données.") |