a9e4f913ec
L'ancienne sonde abandonnait définitivement après 15s et restait bloquée en mode synthétique même quand le flux réel arrivait juste après (latence Mycelium variable, attente du premier keyframe H.264). Remplacé par un thread de fond (gstreamer_watcher) qui retente en continu et bascule la source active dès qu'une frame valide est lue, y compris après une perte de flux en cours de route. Démarrage MJPEG immédiat, plus d'attente bloquante.
303 lines
10 KiB
Python
303 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Réception flux H.264 UDP depuis Pi Zero (port 5600) via GStreamer.
|
|
Inférence YOLOv8n sur chaque frame, exposition MJPEG annoté sur port 5601.
|
|
Thread sideband UDP 5602 : reçoit JSON Pi Zero (résolution, fps, temp, CPU...)
|
|
et affiche un overlay semi-transparent sur chaque frame.
|
|
|
|
Fallback automatique vers webcam locale (cv2.VideoCapture(0)) si le flux
|
|
UDP n'est pas disponible, pour tester sans le Pi Zero.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import cv2
|
|
from ultralytics import YOLO
|
|
|
|
MJPEG_PORT = 5601
|
|
SIDEBAND_PORT = 5602
|
|
MJPEG_QUALITY = 70
|
|
MODEL_PATH = Path(__file__).parent / "yolov8n.pt"
|
|
|
|
# Pipeline GStreamer : RTP H.264 UDP → BGR frames
|
|
# Deux udpsrc (IPv4 0.0.0.0 + IPv6 ::) fusionnés par funnel : un socket IPv6 "::"
|
|
# seul ne reçoit pas les paquets IPv4 sur ce Jetson (IPV6_V6ONLY actif, pas de
|
|
# dual-stack), donc on écoute les deux familles en parallèle sur le même port.
|
|
# IPv4 : réseau local Freebox (dev) — IPv6 : Mycelium (prod, 4G).
|
|
GSTREAMER_PIPELINE = (
|
|
"funnel name=f "
|
|
"! application/x-rtp,payload=96 "
|
|
"! rtph264depay "
|
|
"! avdec_h264 "
|
|
"! videoconvert "
|
|
"! video/x-raw,format=BGR "
|
|
"! appsink drop=true max-buffers=1 "
|
|
"udpsrc address=0.0.0.0 port=5600 ! f. "
|
|
"udpsrc address=:: port=5600 ! f."
|
|
)
|
|
|
|
# Frame annotée partagée entre le thread de capture et le serveur MJPEG
|
|
_frame_lock = threading.Lock()
|
|
_current_frame = None
|
|
_running = True
|
|
|
|
# Source active partagée entre le watcher GStreamer et la boucle de capture :
|
|
# "synthetic" par défaut, bascule vers "gstreamer" dès qu'une frame réelle arrive.
|
|
_source_lock = threading.Lock()
|
|
_active_cap: cv2.VideoCapture | None = None
|
|
_active_source = "synthetic"
|
|
|
|
# Dernières stats Pi Zero reçues via UDP 5602 (thread-safe)
|
|
_stats_lock = threading.Lock()
|
|
_pi_stats: dict | None = None
|
|
|
|
|
|
def sideband_loop() -> None:
|
|
"""Écoute UDP 5602, stocke le dernier JSON Pi Zero dans _pi_stats."""
|
|
global _pi_stats
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
sock.settimeout(1.0)
|
|
sock.bind(("0.0.0.0", SIDEBAND_PORT))
|
|
print(f"[SIDEBAND] Écoute UDP sur port {SIDEBAND_PORT}")
|
|
while _running:
|
|
try:
|
|
data, _ = sock.recvfrom(4096)
|
|
payload = json.loads(data.decode("utf-8"))
|
|
with _stats_lock:
|
|
_pi_stats = payload
|
|
except socket.timeout:
|
|
continue
|
|
except (json.JSONDecodeError, UnicodeDecodeError) as e:
|
|
print(f"[SIDEBAND] JSON invalide : {e}")
|
|
sock.close()
|
|
print("[SIDEBAND] Arrêt.")
|
|
|
|
|
|
def draw_overlay(frame: np.ndarray) -> np.ndarray:
|
|
"""Dessine l'overlay stats Pi Zero (fond noir semi-transparent) sur le frame."""
|
|
with _stats_lock:
|
|
stats = _pi_stats
|
|
if stats is None:
|
|
return frame
|
|
|
|
w_res = stats.get("width", "?")
|
|
h_res = stats.get("height", "?")
|
|
fps = stats.get("fps", "?")
|
|
enc = stats.get("encoder", "?")
|
|
temp = stats.get("temp_c", "?")
|
|
cpu = stats.get("cpu_pct", "?")
|
|
throttled = stats.get("throttled", False)
|
|
throttle_str = "THROTTLED" if throttled else "ok"
|
|
|
|
line1 = f"{w_res}x{h_res} | {fps}fps | {enc} | {temp}C"
|
|
line2 = f"CPU {cpu}% | {throttle_str}"
|
|
|
|
font = cv2.FONT_HERSHEY_SIMPLEX
|
|
scale = 0.55
|
|
thickness = 1
|
|
margin = 6
|
|
|
|
(tw1, th1), _ = cv2.getTextSize(line1, font, scale, thickness)
|
|
(tw2, th2), _ = cv2.getTextSize(line2, font, scale, thickness)
|
|
box_w = max(tw1, tw2) + margin * 2
|
|
box_h = th1 + th2 + margin * 3
|
|
|
|
# Copie pour ne pas altérer le frame original (réutilisé en mode synthétique)
|
|
out = frame.copy()
|
|
bg = out.copy()
|
|
cv2.rectangle(bg, (0, 0), (box_w, box_h), (0, 0, 0), -1)
|
|
cv2.addWeighted(bg, 0.55, out, 0.45, 0, out)
|
|
|
|
cv2.putText(out, line1, (margin, margin + th1),
|
|
font, scale, (255, 255, 255), thickness, cv2.LINE_AA)
|
|
cv2.putText(out, line2, (margin, margin * 2 + th1 + th2),
|
|
font, scale, (255, 255, 255), thickness, cv2.LINE_AA)
|
|
|
|
return out
|
|
|
|
|
|
def gstreamer_watcher() -> None:
|
|
"""
|
|
Thread de fond permanent : tant que la source active n'est pas "gstreamer",
|
|
tente d'ouvrir le pipeline et de lire une première frame (le read() bloque
|
|
naturellement jusqu'à ce que des paquets valides arrivent, ce qui remplace
|
|
l'ancien timeout fixe). Bascule la source active dès qu'une frame arrive,
|
|
y compris après un démarrage en mode synthétique ou une perte de flux.
|
|
"""
|
|
global _active_cap, _active_source
|
|
|
|
while _running:
|
|
with _source_lock:
|
|
already_connected = _active_source == "gstreamer"
|
|
if already_connected:
|
|
time.sleep(1)
|
|
continue
|
|
|
|
cap = cv2.VideoCapture(GSTREAMER_PIPELINE, cv2.CAP_GSTREAMER)
|
|
if not cap.isOpened():
|
|
cap.release()
|
|
time.sleep(2)
|
|
continue
|
|
|
|
ret, _ = cap.read() # bloque jusqu'à la première frame valide
|
|
if not ret:
|
|
cap.release()
|
|
time.sleep(2)
|
|
continue
|
|
|
|
print("[CAM] Flux GStreamer UDP reçu.")
|
|
with _source_lock:
|
|
_active_cap = cap
|
|
_active_source = "gstreamer"
|
|
|
|
|
|
def _make_waiting_frame() -> np.ndarray:
|
|
"""Génère un frame noir 640x480 avec message d'attente."""
|
|
frame = np.zeros((480, 640, 3), dtype=np.uint8)
|
|
cv2.putText(frame, "En attente flux Pi Zero...", (50, 220),
|
|
cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 255, 255), 2)
|
|
cv2.putText(frame, "UDP 5600 non disponible", (50, 260),
|
|
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (100, 100, 255), 2)
|
|
return frame
|
|
|
|
|
|
def capture_loop(model: YOLO) -> None:
|
|
"""
|
|
Lit les frames de la source active (gstreamer ou synthétique), passe dans
|
|
YOLOv8, stocke le résultat annoté. La source est mise à jour en tâche de
|
|
fond par gstreamer_watcher() : ce thread n'a pas besoin d'attendre le flux
|
|
réel au démarrage, il bascule dessus dès qu'il devient disponible.
|
|
"""
|
|
global _current_frame, _running, _active_cap, _active_source
|
|
|
|
print("[CAM] Démarrage en mode synthétique — connexion GStreamer en tâche de fond.")
|
|
synthetic_frame = _make_waiting_frame()
|
|
|
|
while _running:
|
|
with _source_lock:
|
|
cap, source = _active_cap, _active_source
|
|
|
|
if source == "synthetic":
|
|
frame = synthetic_frame
|
|
time.sleep(1 / 10) # 10 fps synthétiques
|
|
else:
|
|
ret, frame = cap.read()
|
|
if not ret:
|
|
print("[CAM] Flux GStreamer perdu, retour en mode synthétique.")
|
|
cap.release()
|
|
with _source_lock:
|
|
_active_cap = None
|
|
_active_source = "synthetic"
|
|
continue
|
|
|
|
# Inférence YOLOv8 — skip en mode synthétique (frame vide sans objet)
|
|
if model is not None and source != "synthetic":
|
|
results = model(frame, verbose=False)
|
|
annotated = results[0].plot()
|
|
else:
|
|
annotated = frame
|
|
|
|
# Overlay stats Pi Zero (ne modifie pas annotated en place)
|
|
annotated = draw_overlay(annotated)
|
|
|
|
with _frame_lock:
|
|
_current_frame = annotated
|
|
|
|
with _source_lock:
|
|
if _active_cap is not None:
|
|
_active_cap.release()
|
|
print("[CAM] Capture terminée.")
|
|
|
|
|
|
class MJPEGHandler(BaseHTTPRequestHandler):
|
|
def log_message(self, format, *args):
|
|
pass # Silencieux pour ne pas polluer les logs WireClaw
|
|
|
|
def do_GET(self):
|
|
if self.path != "/":
|
|
self.send_error(404)
|
|
return
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=frame")
|
|
self.end_headers()
|
|
|
|
while _running:
|
|
with _frame_lock:
|
|
frame = _current_frame
|
|
|
|
if frame is None:
|
|
time.sleep(0.05)
|
|
continue
|
|
|
|
ok, jpeg = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, MJPEG_QUALITY])
|
|
if not ok:
|
|
continue
|
|
data = jpeg.tobytes()
|
|
|
|
try:
|
|
self.wfile.write(
|
|
f"--frame\r\nContent-Type: image/jpeg\r\nContent-Length: {len(data)}\r\n\r\n".encode()
|
|
+ data
|
|
+ b"\r\n"
|
|
)
|
|
self.wfile.flush()
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
break
|
|
|
|
time.sleep(1 / 30)
|
|
|
|
|
|
def main():
|
|
global _running
|
|
|
|
parser = argparse.ArgumentParser(description="WireClaw — réception caméra + YOLOv8")
|
|
parser.add_argument("--no-yolo", action="store_true", help="Désactiver l'inférence YOLOv8 (flux brut)")
|
|
args = parser.parse_args()
|
|
|
|
print(f"[CAM] Chargement modèle YOLOv8 : {MODEL_PATH}")
|
|
model = YOLO(str(MODEL_PATH))
|
|
|
|
if args.no_yolo:
|
|
# Mode flux brut : remplace l'inférence par une copie directe
|
|
model = None
|
|
print("[CAM] Mode flux brut (YOLOv8 désactivé)")
|
|
|
|
# Thread sideband : stats Pi Zero via UDP 5602
|
|
ts = threading.Thread(target=sideband_loop, daemon=True)
|
|
ts.start()
|
|
|
|
# Thread de fond : connexion/reconnexion continue au flux GStreamer réel
|
|
tg = threading.Thread(target=gstreamer_watcher, daemon=True)
|
|
tg.start()
|
|
|
|
# Thread de capture + inférence (démarre immédiatement en mode synthétique)
|
|
t = threading.Thread(target=capture_loop, args=(model,), daemon=True)
|
|
t.start()
|
|
|
|
server = HTTPServer(("0.0.0.0", MJPEG_PORT), MJPEGHandler)
|
|
print(f"[CAM] MJPEG stream sur http://0.0.0.0:{MJPEG_PORT}/")
|
|
print(f"[CAM] Ouvrir dans un navigateur ou VLC : http://192.168.1.84:{MJPEG_PORT}/")
|
|
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
_running = False
|
|
server.shutdown()
|
|
print("[CAM] Arrêt.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|