from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse from app.ocpp.router import CustomChargePoint from app.core.session_manager import session_mgr from app.core.diagnostics import analyze_frames, check_electrical from ocpp.v16 import call as v16_call import logging import json import time logging.basicConfig(level=logging.INFO) app = FastAPI(title="Check-My-IRVE API", version="3.5.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class SocketAdapter: def __init__(self, websocket: WebSocket, cp_id: str): self._ws = websocket self.cp_id = cp_id async def recv(self): msg = await self._ws.receive_text() try: parsed = json.loads(msg) msg_type = parsed[0] unique_id = parsed[1] if msg_type == 2: action, payload = parsed[2], parsed[3] if len(parsed) > 3 else {} elif msg_type == 3: action, payload = None, parsed[2] if len(parsed) > 2 else {} elif msg_type == 4: action, payload = None, {"errorCode": parsed[2]} else: return msg await session_mgr.log_frame(self.cp_id, "IN", msg_type, action, unique_id, payload) except Exception as e: logging.error(f"Erreur parsing IN: {e}") return msg async def send(self, message): try: parsed = json.loads(message) msg_type = parsed[0] unique_id = parsed[1] if msg_type == 2: action, payload = parsed[2], parsed[3] if len(parsed) > 3 else {} elif msg_type == 3: action, payload = None, parsed[2] if len(parsed) > 2 else {} elif msg_type == 4: action, payload = None, {"errorCode": parsed[2]} else: return await self._ws.send_text(message) await session_mgr.log_frame(self.cp_id, "OUT", msg_type, action, unique_id, payload) except Exception as e: logging.error(f"Erreur parsing OUT: {e}") await self._ws.send_text(message) active_charge_points = {} @app.websocket("/ocpp/{client_id}") async def websocket_endpoint(websocket: WebSocket, client_id: str): subprotocol = websocket.headers.get("sec-websocket-protocol", "Unknown") await websocket.accept(subprotocol="ocpp1.6") adapter = SocketAdapter(websocket, client_id) cp = CustomChargePoint(client_id, adapter) cp.subprotocol = subprotocol active_charge_points[client_id] = cp if session_mgr.redis: data = await session_mgr.redis.hget("chargepoints", client_id) if data: cp_data = json.loads(data) cp_data["reconnections"] = cp_data.get("reconnections", 0) + 1 await session_mgr.redis.hset("chargepoints", client_id, json.dumps(cp_data)) try: await cp.start() except WebSocketDisconnect as e: logging.info(f"Borne {client_id} déconnectée (Code: {e.code}).") await session_mgr.update_disconnect(client_id, e.code) await session_mgr.update_cp_status(client_id, "Disconnected") if client_id in active_charge_points: del active_charge_points[client_id] except Exception as e: logging.error(f"Erreur OCPP sur {client_id}: {e}") @app.get("/api/chargepoints") async def get_cps(): return await session_mgr.get_cps() @app.get("/api/chargepoints/{cp_id}/frames") async def get_frames(cp_id: str): return await session_mgr.get_frames(cp_id) @app.delete("/api/chargepoints/{cp_id}/frames") async def purge_frames(cp_id: str): await session_mgr.purge_frames(cp_id) return {"status": "success"} @app.post("/api/chargepoints/{cp_id}/diagnostics") async def get_diagnostics(cp_id: str, payload: dict): frames = await session_mgr.get_frames(cp_id) cps = await session_mgr.get_cps() cp_data = next((cp for cp in cps if cp["id"] == cp_id), {}) installer_config = payload.get("installer_config", {}) return analyze_frames(frames, cp_data, installer_config) @app.get("/api/chargepoints/{cp_id}/report", response_class=HTMLResponse) async def get_report(cp_id: str): frames = await session_mgr.get_frames(cp_id) cps = await session_mgr.get_cps() cp_data = next((cp for cp in cps if cp["id"] == cp_id), {}) findings = analyze_frames(frames, cp_data, {}) html = f"""
Borne: {cp_id}
{json.dumps(frames[:20], indent=2)}
"""
return HTMLResponse(content=html)
@app.post("/api/chargepoints/{cp_id}/audit")
async def audit_config(payload: dict):
borne_config_dict = {k["key"]: k for k in payload.get("borne_config", {}).get("configurationKey", [])}
installer_config = payload.get("installer_config", {})
audit_result = []
REFERENCE_CONFIG = {
"HeartbeatInterval": {"expected": "300", "severity": "warning", "readonly": False},
"WebSocketPingInterval": {"expected": "60", "severity": "critical", "readonly": False},
"MeterValueSampleInterval": {"expected": "10", "severity": "info", "readonly": False},
"ConnectionTimeOut": {"expected": "60", "severity": "warning", "readonly": False}
}
for key, ref in REFERENCE_CONFIG.items():
borne_val = borne_config_dict.get(key, {}).get("value", "Manquant")
audit_result.append({
"key": key, "borne_val": borne_val, "expected_val": ref["expected"],
"match": borne_val == ref["expected"], "severity": ref["severity"], "readonly": ref["readonly"]
})
elec_findings = check_electrical(borne_config_dict, installer_config)
for f in elec_findings:
audit_result.append({
"key": f['title'], "borne_val": f['explication'], "expected_val": f['action'],
"match": False, "severity": f['severity'], "readonly": True
})
return audit_result
@app.post("/api/chargepoints/{cp_id}/command")
async def send_command(cp_id: str, command: dict):
if cp_id not in active_charge_points: return {"error": "CP not connected"}
cp = active_charge_points[cp_id]
action = command.get("action")
payload = command.get("payload", {})
try:
if action == "TriggerMessage": req = v16_call.TriggerMessagePayload(requested_message=payload.get("requested_message", "StatusNotification"), connector_id=payload.get("connector_id", 0))
elif action == "GetConfiguration": req = v16_call.GetConfigurationPayload(key=payload.get("key", []))
elif action == "RemoteStartTransaction": req = v16_call.RemoteStartTransactionPayload(id_tag="CHARGEMAP_PASS", connector_id=1)
elif action == "Reset": req = v16_call.ResetPayload(type="Soft")
else: return {"error": "Unsupported"}
resp = await cp.call(req)
return {"status": "success", "response": resp}
except Exception as e: return {"error": str(e)}
@app.websocket("/api/live")
async def websocket_live(websocket: WebSocket):
await websocket.accept()
if not session_mgr.redis:
await websocket.send_text(json.dumps({"error": "Redis indisponible"}))
await websocket.close()
return
pubsub = session_mgr.redis.pubsub()
await pubsub.subscribe("live_frames")
try:
async for message in pubsub.listen():
if message['type'] == 'message': await websocket.send_text(message['data'].decode('utf-8'))
except: pass
finally: await pubsub.unsubscribe("live_frames")