#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ pi_gateway.py ============= Gateway de la Raspberry Pi 5 para FAB ENCLOSER (version con deteccion IA). Unifica en UN SOLO proceso: - Puente BLE con la XIAO ESP32-S3 (Bleak) - Deteccion de fallos de impresion (YOLO + Picamera2) [spaghetti] - WebSocket hacia la app PyQt6 (estado XIAO + ALERTAS de IA + comandos) - Stream MJPEG del frame ANOTADO por YOLO (/video) - Notificaciones MQTT + ntfy (igual que tu detectar_web_alerta.py) - /health para el monitor de la app Por que un solo proceso: La camara CSI solo puede abrirla un proceso a la vez. La deteccion YOLO y el stream hacia la app comparten la MISMA instancia de Picamera2. El frame que se transmite es el ya anotado por YOLO, asi la app ve exactamente lo que la IA detecta. ---------------------------------------------------------------------- Endpoints: GET /health -> estado de la Pi, BLE e IA GET /video -> MJPEG del frame anotado por YOLO WS /ws -> estado XIAO + alertas IA (push) y comandos (recibe) Mensajes que la app recibe por WebSocket: Estado del XIAO (igual que antes): {"t":..,"h":..,"heater":..,...,"ts":..} Alerta de IA (nuevo, se distingue por el campo "type"): {"type":"ai_alert","status":"SPAGHETTI DETECTADO","msg":"...","ts":..} Estado de deteccion en vivo (nuevo): {"type":"ai_status","status":"Impresion normal","ts":..} Comandos que la app envia (igual que antes): {"cmd":"H:1"} {"cmd":"SP:45"} {"cmd":"EMERGENCY"} ... ---------------------------------------------------------------------- Dependencias (Raspberry Pi OS Bookworm): sudo apt install -y python3-picamera2 pip install bleak fastapi "uvicorn[standard]" ultralytics \ opencv-python paho-mqtt requests --break-system-packages ---------------------------------------------------------------------- """ import asyncio import json import time import threading import logging from contextlib import asynccontextmanager from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import StreamingResponse, JSONResponse import uvicorn from bleak import BleakClient, BleakScanner # --- Vision / IA ---------------------------------------------------- try: from ultralytics import YOLO from picamera2 import Picamera2 import cv2 import requests import paho.mqtt.client as mqtt VISION_AVAILABLE = True except ImportError as e: VISION_AVAILABLE = False _vision_err = str(e) # ==================================================================== # CONFIGURACION # ==================================================================== # --- BLE de la XIAO (mismos UUIDs del .ino) --- DEVICE_NAME = "FAB_ENCLOSER" SERVICE_UUID = "6e400001-b5a3-f393-e0a9-e50e24dcca9e" RX_CHAR_UUID = "6e400002-b5a3-f393-e0a9-e50e24dcca9e" # Pi -> XIAO TX_CHAR_UUID = "6e400003-b5a3-f393-e0a9-e50e24dcca9e" # XIAO -> Pi # --- Servidor web --- HTTP_HOST = "0.0.0.0" HTTP_PORT = 8000 # --- Deteccion IA (de tu detectar_web_alerta.py) --- MODEL_PATH = "best.pt" # o "best_ncnn_model" MQTT_TOPIC = "impresora/errores/spaghetti" MQTT_BROKER = "localhost" NTFY_URL = "https://ntfy.sh/monitoreo_impresion_chuy" CONF_THRESHOLD = 0.25 IMG_SIZE = 320 ALERT_COOLDOWN = 60 # s entre notificaciones externas DETECTION_TIME_REQUIRED = 3 # s continuos para confirmar spaghetti CAM_SIZE = (640, 480) logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") log = logging.getLogger("pi_gateway") # ==================================================================== # ESTADO COMPARTIDO # ==================================================================== class Hub: def __init__(self): self.last_state = {} # ultimo estado del XIAO self.ws_clients = set() self.ble_client = None self.ble_connected = False self.loop = None # IA self.ai_status = "Iniciando..." self.ai_last_alert = 0.0 async def broadcast(self, message: dict): if not self.ws_clients: return data = json.dumps(message) dead = [] for ws in self.ws_clients: try: await ws.send_text(data) except Exception: dead.append(ws) for ws in dead: self.ws_clients.discard(ws) def broadcast_threadsafe(self, message: dict): """Para llamar desde el hilo de deteccion (no-async).""" if self.loop: asyncio.run_coroutine_threadsafe(self.broadcast(message), self.loop) async def send_to_xiao(self, command: str): if not (self.ble_client and self.ble_connected): log.warning("BLE no conectado; comando descartado: %s", command) return payload = (command + "\n").encode() try: await self.ble_client.write_gatt_char(RX_CHAR_UUID, payload, response=False) except Exception as e: log.error("Error enviando a XIAO '%s': %s", command, e) hub = Hub() # ==================================================================== # CLIENTE BLE HACIA LA XIAO # ==================================================================== class BleBridge: def __init__(self, hub: Hub): self.hub = hub self._buffer = "" self._stop = False def _on_notify(self, _char, data: bytearray): self._buffer += data.decode(errors="ignore") while "\n" in self._buffer: line, self._buffer = self._buffer.split("\n", 1) line = line.strip() if not line: continue try: state = json.loads(line) except json.JSONDecodeError: continue state["ts"] = time.time() self.hub.last_state = state if self.hub.loop: asyncio.run_coroutine_threadsafe( self.hub.broadcast(state), self.hub.loop) async def run(self): while not self._stop: log.info("Buscando XIAO (%s)...", DEVICE_NAME) device = await BleakScanner.find_device_by_name( DEVICE_NAME, timeout=10.0) if device is None: log.warning("XIAO no encontrada; reintento en 5 s.") await asyncio.sleep(5) continue try: async with BleakClient(device) as client: self.hub.ble_client = client self.hub.ble_connected = True await client.start_notify(TX_CHAR_UUID, self._on_notify) log.info("XIAO conectada por BLE.") while client.is_connected and not self._stop: await asyncio.sleep(1) except Exception as e: log.error("Conexion BLE caida: %s", e) finally: self.hub.ble_connected = False self.hub.ble_client = None await asyncio.sleep(3) def stop(self): self._stop = True ble_bridge = BleBridge(hub) # ==================================================================== # DETECCION IA (YOLO + Picamera2) en hilo propio # ==================================================================== class Detector: """ Corre YOLO sobre la camara CSI. Mantiene el ultimo frame ANOTADO (en JPEG) para el stream MJPEG, actualiza el estado y dispara alertas. Fusion de tu detectar_web_alerta.py con el gateway. """ def __init__(self, hub: Hub): self.hub = hub self._stop = False self._latest_jpeg = None self._lock = threading.Condition() self._detection_start = None self.mqtt_client = None # ---- notificaciones (igual que tu codigo) ---- def _notify_user(self, message): try: r = requests.post( NTFY_URL, data=message.encode("utf-8"), headers={"Title": "ALERTA IMPRESION 3D", "Priority": "urgent", "Tags": "warning"}, timeout=10) log.info("ntfy status: %s", r.status_code) except Exception as e: log.error("Error ntfy: %s", e) def _setup_mqtt(self): try: c = mqtt.Client() c.connect(MQTT_BROKER, 1883, 60) log.info("MQTT conectado") return c except Exception as e: log.warning("No se pudo conectar MQTT: %s", e) return None def _send_alert(self, message): now = time.time() if now - self.hub.ai_last_alert < ALERT_COOLDOWN: return self.hub.ai_last_alert = now log.warning("ALERTA: %s", message) # MQTT if self.mqtt_client is not None: try: self.mqtt_client.publish(MQTT_TOPIC, message) except Exception as e: log.error("Error MQTT: %s", e) # ntfy self._notify_user(message) # WebSocket -> app PyQt6 (panel de Alertas) self.hub.broadcast_threadsafe({ "type": "ai_alert", "status": "SPAGHETTI DETECTADO", "msg": message, "ts": now, }) def get_jpeg(self): with self._lock: self._lock.wait(timeout=5) return self._latest_jpeg def run(self): if not VISION_AVAILABLE: log.error("Vision no disponible: %s", _vision_err) return model = YOLO(MODEL_PATH) self.mqtt_client = self._setup_mqtt() picam2 = Picamera2() config = picam2.create_preview_configuration( main={"size": CAM_SIZE, "format": "RGB888"}) picam2.configure(config) picam2.start() log.info("Camara + YOLO iniciados.") last_pushed_status = None while not self._stop: try: frame = picam2.capture_array() if frame is None: continue # Escala de grises -> 3 canales (igual que tu pipeline) gray = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY) gray_3ch = cv2.cvtColor(gray, cv2.COLOR_GRAY2RGB) results = model.predict(gray_3ch, imgsz=IMG_SIZE, conf=CONF_THRESHOLD, verbose=False) annotated = results[0].plot() detected = False if results[0].boxes is not None: for box in results[0].boxes: if float(box.conf[0]) >= CONF_THRESHOLD: detected = True break # Anti falsos positivos (tu logica) now = time.time() if detected: if self._detection_start is None: self._detection_start = now elapsed = now - self._detection_start self.hub.ai_status = ( f"Detectando posible spaghetti ({elapsed:.1f}s)") if elapsed >= DETECTION_TIME_REQUIRED: self.hub.ai_status = "SPAGHETTI DETECTADO" self._send_alert("Spaghetti detectado en impresion 3D") else: self._detection_start = None self.hub.ai_status = "Impresion normal" # Empuja cambios de estado de IA a la app (sin spam) if self.hub.ai_status != last_pushed_status: last_pushed_status = self.hub.ai_status self.hub.broadcast_threadsafe({ "type": "ai_status", "status": self.hub.ai_status, "ts": now, }) # Texto sobre el frame color = (0, 255, 0) if "SPAGHETTI" in self.hub.ai_status: color = (0, 0, 255) elif "Detectando" in self.hub.ai_status: color = (0, 255, 255) cv2.putText(annotated, self.hub.ai_status, (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 0.8, color, 2) # Codifica a JPEG para el stream ok, buf = cv2.imencode(".jpg", annotated) if ok: with self._lock: self._latest_jpeg = buf.tobytes() self._lock.notify_all() except Exception as e: log.error("Error en loop de deteccion: %s", e) time.sleep(1) picam2.stop() def stop(self): self._stop = True detector = Detector(hub) # ==================================================================== # FASTAPI # ==================================================================== @asynccontextmanager async def lifespan(app: FastAPI): hub.loop = asyncio.get_event_loop() # Hilo de deteccion (camara + YOLO) det_thread = threading.Thread(target=detector.run, daemon=True) det_thread.start() # Puente BLE ble_task = asyncio.create_task(ble_bridge.run()) log.info("Gateway iniciado en http://%s:%d", HTTP_HOST, HTTP_PORT) yield ble_bridge.stop() detector.stop() ble_task.cancel() app = FastAPI(lifespan=lifespan) @app.get("/health") async def health(): return JSONResponse({ "status": "ok", "ble_connected": hub.ble_connected, "ws_clients": len(hub.ws_clients), "vision": VISION_AVAILABLE, "ai_status": hub.ai_status, "last_state_age": (time.time() - hub.last_state["ts"]) if hub.last_state.get("ts") else None, }) @app.get("/video") async def video(): def generate(): boundary = b"--frame\r\n" while True: jpeg = detector.get_jpeg() if jpeg is None: continue yield (boundary + b"Content-Type: image/jpeg\r\n\r\n" + jpeg + b"\r\n") return StreamingResponse( generate(), media_type="multipart/x-mixed-replace; boundary=frame") @app.websocket("/ws") async def websocket_endpoint(ws: WebSocket): await ws.accept() hub.ws_clients.add(ws) log.info("App conectada por WS (%d activas).", len(hub.ws_clients)) # Estado inicial inmediato if hub.last_state: await ws.send_text(json.dumps(hub.last_state)) await ws.send_text(json.dumps( {"type": "ai_status", "status": hub.ai_status, "ts": time.time()})) try: while True: raw = await ws.receive_text() try: msg = json.loads(raw) except json.JSONDecodeError: continue cmd = msg.get("cmd") if cmd: await hub.send_to_xiao(cmd) except WebSocketDisconnect: pass finally: hub.ws_clients.discard(ws) log.info("App desconectada (%d activas).", len(hub.ws_clients)) if __name__ == "__main__": uvicorn.run(app, host=HTTP_HOST, port=HTTP_PORT, log_level="info")