86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
"""Logique métier GSPARC — kilométrage, consommation, anomalies."""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
PRODUIT_ATTENDU = "Gasoual Normal"
|
||
|
||
|
||
def calculer_kilometrage(compteur_avant: float, compteur_apres: float) -> float:
|
||
"""Calcule la distance parcourue entre deux approvisionnements."""
|
||
if compteur_avant is None or compteur_apres is None:
|
||
return 0.0
|
||
distance = compteur_apres - compteur_avant
|
||
return max(0, distance)
|
||
|
||
|
||
def calculer_consommation(quantite_litres: float, km_parcourus: float) -> float | None:
|
||
"""Calcule la consommation en L/100km."""
|
||
if km_parcourus <= 0 or quantite_litres <= 0:
|
||
return None
|
||
return round((quantite_litres / km_parcourus) * 100, 2)
|
||
|
||
|
||
def detecter_anomalies(approvisionnement: dict, dernier_appro: dict | None = None) -> list[str]:
|
||
"""Détecte les anomalies sur un approvisionnement.
|
||
|
||
Retourne une liste de codes d'anomalie :
|
||
- KM_NEGATIF : kilométrage nul ou négatif
|
||
- PRODUIT_INVALIDE : produit différent de Gasoual Normal
|
||
- DOUBLON : même numéro de reçu
|
||
- KM_INCOHERENT : kilométrage en baisse ou bond suspect
|
||
- VEHICULE_INCONNU : matricule non reconnu (à vérifier en amont)
|
||
- CARTE_INCONNUE : carte non reconnue (à vérifier en amont)
|
||
"""
|
||
anomalies = []
|
||
|
||
km_parcouru = approvisionnement.get("الكلم_المقطوع", 0) or 0
|
||
produit = approvisionnement.get("المنتج", "")
|
||
|
||
# Anomalie 1 : kilométrage négatif
|
||
if km_parcouru <= 0:
|
||
anomalies.append("KM_NEGATIF")
|
||
|
||
# Anomalie 2 : produit non conforme
|
||
if produit and produit.strip() != PRODUIT_ATTENDU:
|
||
anomalies.append("PRODUIT_INVALIDE")
|
||
|
||
# Anomalie 3 : incohérence de kilométrage avec approvisionnement précédent
|
||
if dernier_appro:
|
||
km_precedent = dernier_appro.get("الكلم_المقطوع", 0) or 0
|
||
compteur_avant = approvisionnement.get("العداد_قبل", 0) or 0
|
||
compteur_precedent = dernier_appro.get("العداد_بعد", 0) or 0
|
||
|
||
if compteur_avant < compteur_precedent:
|
||
anomalies.append("KM_INCOHERENT")
|
||
|
||
return anomalies
|
||
|
||
|
||
def analyser_tendance(valeurs: list[float]) -> dict:
|
||
"""Analyse la tendance de consommation."""
|
||
if not valeurs or len(valeurs) < 2:
|
||
return {"tendance": "stable", "variation_pct": 0}
|
||
|
||
moyenne = sum(valeurs) / len(valeurs)
|
||
dernieres = valeurs[-3:] if len(valeurs) >= 3 else valeurs
|
||
moyenne_recente = sum(dernieres) / len(dernieres)
|
||
|
||
if moyenne > 0:
|
||
variation = ((moyenne_recente - moyenne) / moyenne) * 100
|
||
else:
|
||
variation = 0
|
||
|
||
if variation > 10:
|
||
tendance = "hausse"
|
||
elif variation < -10:
|
||
tendance = "baisse"
|
||
else:
|
||
tendance = "stable"
|
||
|
||
return {
|
||
"tendance": tendance,
|
||
"variation_pct": round(variation, 1),
|
||
"moyenne_generale": round(moyenne, 2),
|
||
"moyenne_recente": round(moyenne_recente, 2),
|
||
} |