commit f582f317c0266cf4dc2683ff128a03a48a7103a0 Author: srtk Date: Fri Nov 21 21:34:32 2025 +0530 initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..350a679 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +app.png +logo.png diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..e69de29 diff --git a/server.py b/server.py new file mode 100644 index 0000000..e9d40df --- /dev/null +++ b/server.py @@ -0,0 +1,277 @@ +#!/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()