Spaces:
Sleeping
Sleeping
File size: 2,681 Bytes
7ddb64a | 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 | import random
from api.models.schemas import DiagnosisResultPerDisease, DiagnosisResponse
# Interfacage du modèle de ML (Spark) avec l'API FastAPI.
# Pour l'instant, on simule les prédictions avec des valeurs aléatoires.
MOCK_CONDITIONS = [
"Malaria", "Typhoid Fever", "Influenza", "Dengue Fever",
"Pneumonia", "Gastroenteritis", "Meningitis", "Cholera"
]
def get_mock_diagnosis(symptoms: list[str], age: int, sex: str, medical_history: str, geolocation: str) -> DiagnosisResponse:
"""
Simule la prediction du modèle de ML en retournant 3 conditions médicales aléatoires avec des probabilités et une recommandation.
Fonction reelle: model.predict_proba(encode_symptoms(symptoms))
"""
# Choisir 3 conditions au hasard parmi la liste, avec des probabilités aléatoires
selected = random.sample(MOCK_CONDITIONS, 3)
# Genere les probabilités aléatoires et les normalise pour que la somme soit 1
raw = sorted([random.random() for _ in selected], reverse=True)
total = sum(raw)
probabilities = [round(p / total, 2) for p in raw]
statistics = [
DiagnosisResultPerDisease(
name=disease,
probability=prob,
level="level_high" if prob > 0.7 else "level_medium" if prob < 0.4 else "level_low",
symptoms=symptoms
)
for disease, prob in zip(selected, probabilities)
]
return DiagnosisResponse(
statistics=statistics,
recommendation=_get_recommendation(statistics[0].level)
)
def _get_recommendation(severity: str) -> str:
return {
"level_high": "Veuillez consulter un médecin immédiatement ou rendez-vous aux urgences.",
"level_moderate": "Veuillez surveiller vos symptômes et consulter un médecin si ceux-ci s'aggravent.",
"level_low": "Restez au repos et hydratez-vous. Consultez un médecin si les symptômes persistent."
}.get(severity, "Consultez un professionnel de santé.")
# === Chargeur de modèle réel (a implémenter plus tard) ===
# import pickle
#
# _model = None
#
# def load_model():
# global _model
# with open("saved_models/spark_model.pkl", "rb") as f:
# _model = pickle.load(f)
#
# def encode_symptoms(symptoms: list[str]) -> list[int]:
# # Convertit la liste de symptômes en un vecteur binaire pour le modèle ML.
# ALL_SYMPTOMS = [...] # full list from your dataset
# return [1 if s in symptoms else 0 for s in ALL_SYMPTOMS]
#
# def get_real_diagnosis(symptoms: list[str]) -> DiagnosisResponse:
# vector = encode_symptoms(symptoms)
# probas = _model.predict_proba([vector])[0]
# classes = _model.classes_
# ...
|