#!/usr/bin/env python3 import socket import subprocess import sys import select import time import signal import threading import logging from zeroconf import ServiceInfo, Zeroconf # --------------------------- # Logging Configuration # --------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) logger = logging.getLogger("ProCamServer") # --------------------------- # Configuration # --------------------------- UDP_PORT = 5000 DEVICE = '/dev/video10' BUFFER_SIZE = 65535 HEARTBEAT_INTERVAL = 2 CONNECTION_TIMEOUT = 5 class StreamServer: def __init__(self, host_ip): self.host_ip = host_ip self.running = False self.ffmpeg_process = None self.socket = None self.zeroconf = None self.last_packet_time = 0 self.connected = False self.client_addr = None self.restart_attempts = 0 self.max_restart_attempts = 3 # ---------------------------------------------------------------------- def start_discovery(self): logger.info(f"Registering mDNS service on {self.host_ip}:{UDP_PORT}") desc = {'version': '2.0', 'device': DEVICE} info = ServiceInfo( "_procam._udp.local.", "ProCamLink._procam._udp.local.", addresses=[socket.inet_aton(self.host_ip)], port=UDP_PORT, properties=desc, server="desktop.local.", ) self.zeroconf = Zeroconf() self.zeroconf.register_service(info) return info # ---------------------------------------------------------------------- def start_ffmpeg(self): cmd = [ 'ffmpeg', '-hide_banner', '-loglevel', 'error', '-f', 'h264', '-i', 'pipe:0', '-vf', 'format=yuv420p', '-c:v', 'rawvideo', '-pix_fmt', 'yuyv422', '-f', 'v4l2', DEVICE ] try: self.ffmpeg_process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, bufsize=0 ) logger.info(f"FFmpeg started (PID {self.ffmpeg_process.pid})") self.restart_attempts = 0 return True except Exception as e: logger.error(f"Failed to start FFmpeg: {e}") return False # ---------------------------------------------------------------------- def restart_ffmpeg(self): if self.restart_attempts >= self.max_restart_attempts: logger.warning("Maximum FFmpeg restart attempts reached") return False self.restart_attempts += 1 logger.info(f"Restarting FFmpeg (attempt {self.restart_attempts})") if self.ffmpeg_process: try: self.ffmpeg_process.terminate() self.ffmpeg_process.wait(timeout=2) except: try: self.ffmpeg_process.kill() except: pass time.sleep(0.5) return self.start_ffmpeg() # ---------------------------------------------------------------------- def check_ffmpeg_health(self): if self.ffmpeg_process and self.ffmpeg_process.poll() is not None: return False return True # ---------------------------------------------------------------------- def send_heartbeat(self): while self.running: time.sleep(HEARTBEAT_INTERVAL) if self.connected and self.client_addr: try: self.socket.sendto(b'HEARTBEAT', self.client_addr) except: pass # ---------------------------------------------------------------------- def monitor_connection(self): while self.running: time.sleep(1) if self.last_packet_time > 0: elapsed = time.time() - self.last_packet_time if elapsed > CONNECTION_TIMEOUT and self.connected: logger.warning("Client connection lost") self.connected = False self.client_addr = None elif elapsed < CONNECTION_TIMEOUT and not self.connected and self.client_addr: logger.info(f"Client connected: {self.client_addr[0]}") self.connected = True # ---------------------------------------------------------------------- def run(self): self.running = True service_info = self.start_discovery() self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, BUFFER_SIZE * 8) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.socket.bind(("0.0.0.0", UDP_PORT)) self.socket.setblocking(0) logger.info("ProCam server initialized") logger.info(f"Listening on {self.host_ip}:{UDP_PORT}") logger.info(f"Output device: {DEVICE}") logger.info("Waiting for incoming stream...") if not self.start_ffmpeg(): return threading.Thread(target=self.send_heartbeat, daemon=True).start() threading.Thread(target=self.monitor_connection, daemon=True).start() stats_interval = time.time() stats_bytes = 0 total_bytes = 0 try: while self.running: if not self.check_ffmpeg_health(): logger.error("FFmpeg has stopped unexpectedly") if not self.restart_ffmpeg(): break ready = select.select([self.socket], [], [], 0.05) if ready[0]: try: data, addr = self.socket.recvfrom(BUFFER_SIZE) if data == b'HEARTBEAT': continue if not self.connected or self.client_addr != addr: self.client_addr = addr self.last_packet_time = time.time() try: if self.ffmpeg_process and self.ffmpeg_process.poll() is None: self.ffmpeg_process.stdin.write(data) self.ffmpeg_process.stdin.flush() stats_bytes += len(data) total_bytes += len(data) except (BrokenPipeError, IOError, OSError): logger.error("FFmpeg pipe error encountered") if not self.restart_ffmpeg(): break except Exception: pass if time.time() - stats_interval > 5: if stats_bytes > 0: mbps = (stats_bytes * 8) / (5 * 1_000_000) logger.info(f"Throughput: {mbps:.2f} Mbps | Total received: {total_bytes / (1024*1024):.1f} MB") stats_interval = time.time() stats_bytes = 0 except KeyboardInterrupt: logger.info("Shutdown signal received, stopping server...") finally: self.cleanup(service_info) # ---------------------------------------------------------------------- def cleanup(self, service_info): self.running = False if self.ffmpeg_process: try: self.ffmpeg_process.terminate() self.ffmpeg_process.wait(timeout=2) except: try: self.ffmpeg_process.kill() except: pass if self.socket: self.socket.close() if self.zeroconf: try: self.zeroconf.unregister_service(service_info) self.zeroconf.close() except: pass logger.info("Cleanup complete. Server stopped.") # ---------------------------------------------------------------------- if __name__ == "__main__": host_ip = '192.168.1.8' server = StreamServer(host_ip) def signal_handler(sig, frame): server.running = False signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) server.run()