nicoboy/wireclaw-pizero
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 1080p@10fps + sideband UDP 5602 + textoverlay

Browse Source 
- Pipeline 1920x1080@10fps avec level=(string)4 (auto via _h264_level)
- Sideband thread UDP port 5602 : JSON {ts,width,height,fps,encoder,temp_c} @ 1 Hz
- textoverlay "temp=XX.XC  10fps  1920x1080" haut-gauche avant encodeur
- .env.dev : WIDTH=1920 HEIGHT=1080 FPS=10 BITRATE_KBPS=2500
- _load_env() : .env base + .env.dev override
- Fallback hw→sw conservé, level auto 3.2/4 selon résolution

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
nicoboy
2026-06-09 11:45:35 +02:00
parent e2ceeded8a
commit ca4aa13d1d
3 changed files with 109 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
camera_server.py
View File

@ -3,13 +3,17 @@
WireClaw — Pi Zero 2W camera server
Flask API port 5000 : /health /settings
RTP H264 → Jetson via GStreamer subprocess
Sideband JSON UDP port 5602 → Jetson (1 Hz)
Encoder strategy: v4l2h264enc hardware (default) → x264enc software (auto-fallback)
Config: .env (base) overridden by .env.dev if present
"""
import json
import os
import shlex
import signal
import socket
import subprocess
import threading
import time
@ -17,20 +21,38 @@ import logging
from flask import Flask, jsonify, request
# --- .env loading -------------------------------------------------------------
def _load_env(path: str) -> None:
if not os.path.exists(path):
return
with open(path) as f:
for raw in f:
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, val = line.partition("=")
os.environ[key.strip()] = val.strip()
_load_env(".env")
_load_env(".env.dev") # overrides .env when present
# --- Configuration -----------------------------------------------------------
DEFAULTS: dict = {
"jetson_host": "192.168.1.84",
"jetson_port": 5600,
"width": 640,
"height": 480,
"fps": 15,
"bitrate_kbps": 1000,
"jetson_host": os.environ.get("JETSON_HOST", "192.168.1.84"),
"jetson_port": int(os.environ.get("JETSON_PORT", "5600")),
"width": int(os.environ.get("WIDTH", "640")),
"height": int(os.environ.get("HEIGHT", "480")),
"fps": int(os.environ.get("FPS", "15")),
"bitrate_kbps": int(os.environ.get("BITRATE_KBPS", "1000")),
# "auto" tries hw then sw; "hw" forces v4l2h264enc; "sw" forces x264enc
"encoder": "auto",
"encoder": os.environ.get("ENCODER", "auto"),
}
FLASK_PORT = 5000
FLASK_PORT = 5000
SIDEBAND_PORT = 5602
# --- Logging -----------------------------------------------------------------
@ -40,37 +62,92 @@ logging.basicConfig(
)
log = logging.getLogger(__name__)
# --- Helpers -----------------------------------------------------------------
def _get_temp_celsius() -> float | None:
"""Read SoC temperature via vcgencmd. Returns None if unavailable."""
try:
out = subprocess.check_output(["vcgencmd", "measure_temp"], text=True, timeout=1)
# output: "temp=45.2'C"
return float(out.strip().split("=")[1].replace("'C", ""))
except Exception:
return None
def _h264_level(w: int, h: int) -> str:
"""Minimum H.264 level for the given resolution."""
# level 4 required for anything above 720p
return "4" if w * h > 1280 * 720 else "3.2"
# --- Pipeline builder --------------------------------------------------------
def _build_pipeline(cfg: dict, encoder: str) -> str:
def _build_pipeline(cfg: dict, encoder: str, overlay_text: str) -> str:
w, h, fps = cfg["width"], cfg["height"], cfg["fps"]
host, port = cfg["jetson_host"], cfg["jetson_port"]
level = _h264_level(w, h)
sink = f"rtph264pay config-interval=1 pt=96 ! udpsink host={host} port={port}"
# shlex-safe: repr() wraps in single quotes, preserving spaces inside text value
overlay = (
f"textoverlay text={overlay_text!r} "
f"valignment=top halignment=left "
f'font-desc="Sans Bold 24" draw-shadow=true'
)
if encoder == "hw":
bitrate_bps = cfg["bitrate_kbps"] * 1000
# NV12 format avoids a CPU color-conversion step before v4l2convert
# level=(string)3.2 is required — default level 1.0 caps at 176x144
# Explicit NV12 cap after v4l2convert so textoverlay and v4l2h264enc agree on format
return (
f"libcamerasrc ! "
f"video/x-raw,width={w},height={h},format=NV12,framerate={fps}/1 ! "
f"v4l2convert ! "
f"v4l2convert ! video/x-raw,format=NV12 ! "
f"{overlay} ! "
f"v4l2h264enc video-bitrate={bitrate_bps} "
f'extra-controls="controls,repeat_sequence_header=1" ! '
f"video/x-h264,level=(string)3.2 ! "
f"video/x-h264,level=(string){level} ! "
f"h264parse ! "
f"{sink}"
)
# sw
# sw — videoconvert handles any format before and after textoverlay
return (
f"libcamerasrc ! "
f"video/x-raw,width={w},height={h},framerate={fps}/1 ! "
f"videoconvert ! "
f"{overlay} ! "
f"videoconvert ! "
f"x264enc tune=zerolatency bitrate={cfg['bitrate_kbps']} ! "
f"{sink}"
)
# --- Sideband UDP thread ------------------------------------------------------
def _sideband_worker(gst: "GstManager") -> None:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
log.info("Sideband UDP ready (→ port %d)", SIDEBAND_PORT)
try:
while not gst._shutdown.is_set():
cfg = gst.get_cfg()
temp = _get_temp_celsius()
payload = json.dumps({
"ts": time.time(),
"width": cfg["width"],
"height": cfg["height"],
"fps": cfg["fps"],
"encoder": gst.encoder_active or "unknown",
"temp_c": temp,
}).encode()
try:
sock.sendto(payload, (cfg["jetson_host"], SIDEBAND_PORT))
except OSError as exc:
log.debug("sideband send: %s", exc)
time.sleep(1)
finally:
sock.close()
# --- GStreamer process manager -----------------------------------------------
class GstManager:
@ -78,7 +155,6 @@ class GstManager:
self._cfg: dict = dict(DEFAULTS)
self._lock = threading.Lock()
self._proc: subprocess.Popen | None = None
self._restart_event = threading.Event()
self._shutdown = threading.Event()
self._thread: threading.Thread | None = None
self.encoder_active: str | None = None
@ -88,6 +164,7 @@ class GstManager:
def start(self) -> None:
self._thread = threading.Thread(target=self._worker, name="gst-rtp", daemon=True)
self._thread.start()
threading.Thread(target=_sideband_worker, args=(self,), name="sideband", daemon=True).start()
def stop(self) -> None:
self._shutdown.set()
@ -121,7 +198,12 @@ class GstManager:
def _launch(self, encoder: str) -> "subprocess.Popen | None":
with self._lock:
cfg = dict(self._cfg)
pipeline = _build_pipeline(cfg, encoder)
temp = _get_temp_celsius()
temp_str = f"{temp:.1f}C" if temp is not None else "?C"
overlay_text = f"temp={temp_str} {cfg['fps']}fps {cfg['width']}x{cfg['height']}"
pipeline = _build_pipeline(cfg, encoder, overlay_text)
cmd = ["gst-launch-1.0", "-e"] + shlex.split(pipeline)
log.info("GStreamer [%s] pipeline: %s", encoder, pipeline)
try:
@ -159,8 +241,7 @@ class GstManager:
self.encoder_active = enc
log.info("RTP stream running (encoder=%s)", enc)
stderr_t = threading.Thread(target=self._drain_stderr, args=(proc,), daemon=True)
stderr_t.start()
threading.Thread(target=self._drain_stderr, args=(proc,), daemon=True).start()
proc.wait()
rc = proc.returncode
@ -192,8 +273,8 @@ app = Flask(__name__)
def health():
alive = _gst.is_alive()
return jsonify({
"status": "ok" if alive else "degraded",
"rtp": alive,
"status": "ok" if alive else "degraded",
"rtp": alive,
"encoder": _gst.encoder_active,
})