import math def rule_missing_status_after_boot(frames, ctx): if len(frames) < 3: return None has_boot = any(f['action'] == "BootNotification" for f in frames) has_status = any(f['action'] == "StatusNotification" for f in frames) if has_boot and not has_status: return { "id": "NO_STATUS_AFTER_BOOT", "severity": "warning", "confidence": "medium", "title": "Séquence de démarrage non conforme", "explication": "Aucun StatusNotification n'a été reçu après le BootNotification.", "action": "Envoyez un TriggerMessage(StatusNotification)." } return None def rule_bad_websocket_ping(frames, ctx): for f in reversed(frames): if f['action'] == "GetConfiguration" and f['direction'] == "OUT": config = f['payload'].get('configurationKey', []) for item in config: if item.get('key') == 'WebSocketPingInterval' and item.get('value') == '0': return { "id": "BAD_PING_INTERVAL", "severity": "critical", "confidence": "high", "title": "WebSocketPingInterval désactivé (0)", "explication": "La borne n'envoie pas de ping. Derrière une box FAI (NAT), la connexion sera coupée.", "action": "Corrigez cette valeur à 60 depuis l'onglet Audit Config." } return None def check_electrical(borne_config_dict, installer_config): findings = [] max_c_str = borne_config_dict.get("MaxCurrent", {}).get("value") disj_str = str(installer_config.get("disj_calibre", "0")) p_decl_str = str(installer_config.get("borne_puissance", "0")) raccordement = installer_config.get("raccordement", "") if not (max_c_str and disj_str and p_decl_str): return findings try: max_c = int(max_c_str) disj = int(disj_str) p_decl = int(p_decl_str) # a) Sécurité disjoncteur if disj < max_c: findings.append({ "id": "ELEC_UNDER_PROTECTED", "severity": "critical", "confidence": "high", "title": "Protection sous-dimensionnée", "explication": f"Disjoncteur {disj}A inférieur au courant max borne {max_c}A.", "action": "Risque de déclenchement en charge. Vérifier le calibre et la section de câble." }) # b) Puissance réelle vs déclarée if "mono" in raccordement: p_reelle = 230 * max_c else: p_reelle = 400 * max_c * math.sqrt(3) p_reelle_kw = round(p_reelle / 1000, 1) if p_reelle_kw < p_decl * 0.85: findings.append({ "id": "ELEC_POWER_BRIDLED", "severity": "critical", "confidence": "high", "title": "Borne bridée logiquement", "explication": f"Borne bridée à {p_reelle_kw} kW ({max_c}A x {400 if 'tri' in raccordement else 230}V) alors que {p_decl} kW sont déclarés.", "action": "Le firmware limite le courant. Vérifier le paramétrage usine ou une limitation logicielle." }) # c) 22kW sur Mono if p_decl > 7.4 and "mono" in raccordement: findings.append({ "id": "ELEC_MONO_OVERPOWER", "severity": "critical", "confidence": "high", "title": "Raccordement incohérent", "explication": f"Puissance {p_decl}kW déclarée sur du monophasé.", "action": "Au-delà de 7.4kW, le triphasé est obligatoire." }) except ValueError: pass return findings def analyze_frames(frames, cp_data, installer_config=None): findings = [] # Rècles comportementales rules = [rule_missing_status_after_boot, rule_bad_websocket_ping] for rule in rules: res = rule(frames, cp_data) if res: findings.append(res) # Règles électriques (nécessitent la config installateur) if installer_config: borne_config_dict = {} for f in reversed(frames): if f['action'] == "GetConfiguration" and f['direction'] == "OUT": borne_config_dict = {k["key"]: k for k in f['payload'].get('configurationKey', [])} break findings.extend(check_electrical(borne_config_dict, installer_config)) return findings