Spaces:
Sleeping
Sleeping
| """ | |
| 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.") |