From 6256b4ecd6ffc3fd73d8abd0adaa6457ecb705e7 Mon Sep 17 00:00:00 2001 From: srtk Date: Sun, 30 Nov 2025 14:42:09 +0530 Subject: [PATCH] add working mic functionality --- server.py | 371 ++++++++++++++++++------------------------------------ 1 file changed, 121 insertions(+), 250 deletions(-) diff --git a/server.py b/server.py index e9d40df..6f125bd 100644 --- a/server.py +++ b/server.py @@ -1,277 +1,148 @@ #!/usr/bin/env python3 -import socket -import subprocess -import sys -import select -import time -import signal -import threading -import logging +import socket, subprocess, sys, time, signal, threading, select, math 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 +CONF = { + 'video_port': 5000, 'audio_port': 5001, + 'video_dev': '/dev/video20', 'audio_sink': 'AirLink_Sink', 'audio_src': 'AirLink_Mic' +} - -class StreamServer: - def __init__(self, host_ip): - self.host_ip = host_ip +class AirLinkServer: + def __init__(self): + self.ip = self.get_lan_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 - ] + self.procs = {'video': None, 'audio': None, 'ffmpeg_audio': None} + self.sockets = {'video': None, 'audio': None} + self.stats = {'v_bytes': 0, 'a_bytes': 0, 'v_pkts': 0, 'a_pkts': 0, 'last_time': time.time()} + self.zc = None + self.lock = threading.Lock() + def get_lan_ip(self): + """Finds LAN IP by prioritizing physical interfaces over VPN/Tun interfaces.""" 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 + # List global IPv4 addresses, prioritize eth/wlan + cmd = "ip -4 -o addr show scope global | awk '{print $2,$4}'" + out = subprocess.check_output(cmd, shell=True).decode().strip().split('\n') + candidates = [] + for line in out: + if not line: continue + iface, ip = line.split()[0], line.split()[1].split('/')[0] + if iface.startswith(('e', 'w')): return ip # Found physical interface + candidates.append(ip) + return candidates[0] if candidates else '127.0.0.1' + except: + # Fallback to standard socket connect + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: s.connect(("8.8.8.8", 80)); return s.getsockname()[0] + except: return '127.0.0.1' + finally: s.close() + + def setup_system(self): + print(f"[INIT] Configuring system on {self.ip}...") + try: + subprocess.run("sudo modprobe -r v4l2loopback; sudo modprobe v4l2loopback video_nr=20 card_label=AirLink_Cam exclusive_caps=1", shell=True, check=True, stderr=subprocess.DEVNULL) + print(f"[INFO] Video Device: {CONF['video_dev']}") + + subprocess.run(f"pactl load-module module-null-sink sink_name={CONF['audio_sink']} sink_properties=device.description='AirLink_Audio' rate=48000 channels=1", shell=True, stderr=subprocess.DEVNULL) + subprocess.run(f"pactl load-module module-remap-source source_name={CONF['audio_src']} master={CONF['audio_sink']}.monitor source_properties=device.description='AirLink_Mic' channels=1", shell=True, stderr=subprocess.DEVNULL) + print(f"[INFO] Audio Source: {CONF['audio_src']}") return True - except Exception as e: - logger.error(f"Failed to start FFmpeg: {e}") + print(f"[ERR] Setup failed: {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: + def paint_black(self): + with self.lock: + if self.procs['video']: + self.procs['video'].terminate() + self.procs['video'] = None try: - self.ffmpeg_process.terminate() - self.ffmpeg_process.wait(timeout=2) - except: - try: - self.ffmpeg_process.kill() - except: - pass + subprocess.run(['ffmpeg', '-hide_banner', '-loglevel', 'error', '-f', 'lavfi', '-i', 'color=black:s=1280x720', '-pix_fmt', 'yuyv422', '-frames:v', '5', '-f', 'v4l2', CONF['video_dev']], check=False) + print("[INFO] Video stream stopped (Black Screen)") + except: pass - time.sleep(0.5) - return self.start_ffmpeg() + def start_pipeline(self, p_type): + with self.lock: + if p_type == 'video' and (not self.procs['video'] or self.procs['video'].poll() is not None): + self.procs['video'] = subprocess.Popen( + ['ffmpeg', '-fflags', 'nobuffer', '-flags', 'low_delay', '-analyzeduration', '0', '-probesize', '32', + '-hide_banner', '-loglevel', 'fatal', '-f', 'h264', '-i', 'pipe:0', + '-vf', 'scale=1280:720:force_original_aspect_ratio=decrease,pad=1280:720:(ow-iw)/2:(oh-ih)/2', + '-vsync', '0', '-copyts', '-pix_fmt', 'yuyv422', '-f', 'v4l2', CONF['video_dev']], + stdin=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0) + + elif p_type == 'audio' and (not self.procs['audio'] or self.procs['audio'].poll() is not None): + self.procs['ffmpeg_audio'] = subprocess.Popen( + ['ffmpeg', '-hide_banner', '-loglevel', 'fatal', '-fflags', 'nobuffer', '-f', 's16le', '-ar', '48000', '-ac', '1', '-i', 'pipe:0', '-f', 's16le', '-ar', '48000', '-ac', '1', 'pipe:1'], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0) + self.procs['audio'] = subprocess.Popen( + ['pacat', '--playback', f"--device={CONF['audio_sink']}", '--rate=48000', '--channels=1', '--format=s16le', '--latency-msec=20'], + stdin=self.procs['ffmpeg_audio'].stdout, stderr=subprocess.DEVNULL, bufsize=0) - # ---------------------------------------------------------------------- - - 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 log_stats(self): + now = time.time() + delta = now - self.stats['last_time'] + if delta >= 1.0: + v_mbps = (self.stats['v_bytes'] * 8) / (1000 * 1000) / delta + a_kbps = (self.stats['a_bytes'] * 8) / 1000 / delta + print(f"[STATS] Video: {v_mbps:6.2f} Mbps ({self.stats['v_pkts']} pps) | Audio: {a_kbps:6.2f} kbps ({self.stats['a_pkts']} pps)") + self.stats = {'v_bytes': 0, 'a_bytes': 0, 'v_pkts': 0, 'a_pkts': 0, 'last_time': now} def run(self): + if not self.setup_system(): return self.running = True - service_info = self.start_discovery() + + # Discovery + info = ServiceInfo("_procam._udp.local.", "AirLink._procam._udp.local.", addresses=[socket.inet_aton(self.ip)], port=CONF['video_port'], properties={'version': '2.0'}) + self.zc = Zeroconf(); self.zc.register_service(info) - 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) + # Sockets + for k, port, buf in [('video', CONF['video_port'], 4*1024*1024), ('audio', CONF['audio_port'], 512*1024)]: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, buf) + s.bind(("0.0.0.0", port)) + s.setblocking(0) + self.sockets[k] = s - 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...") + self.start_pipeline('video'); self.start_pipeline('audio') + print(f"\n[READY] Server listening on {self.ip}:{CONF['video_port']}\n") - if not self.start_ffmpeg(): - return + while self.running: + self.log_stats() + # Ensure audio is alive + if self.procs['audio'] and self.procs['audio'].poll() is not None: self.start_pipeline('audio') - 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: + ready, _, _ = select.select(list(self.sockets.values()), [], [], 0.1) + for sock in ready: try: - self.ffmpeg_process.kill() - except: - pass + data, addr = sock.recvfrom(65535) + + # Control Signals + if len(data) < 16: + if data == b'PING': sock.sendto(b'PONG', addr); continue + if data == b'HEARTBEAT': sock.sendto(b'ACK', addr); continue + if data in [b'PAUSE', b'BYE']: self.paint_black(); continue - if self.socket: - self.socket.close() + # Data Routing + if sock == self.sockets['video']: + self.stats['v_bytes'] += len(data); self.stats['v_pkts'] += 1 + if not self.procs['video'] or self.procs['video'].poll() is not None: self.start_pipeline('video') + if self.procs['video']: self.procs['video'].stdin.write(data); self.procs['video'].stdin.flush() + + elif sock == self.sockets['audio']: + self.stats['a_bytes'] += len(data); self.stats['a_pkts'] += 1 + if self.procs['ffmpeg_audio']: self.procs['ffmpeg_audio'].stdin.write(data); self.procs['ffmpeg_audio'].stdin.flush() + except Exception: pass - if self.zeroconf: - try: - self.zeroconf.unregister_service(service_info) - self.zeroconf.close() - except: - pass - - logger.info("Cleanup complete. Server stopped.") - - -# ---------------------------------------------------------------------- + def cleanup(self): + self.running = False + self.paint_black() + if self.zc: self.zc.close() + for mod in [CONF['audio_sink'], CONF['audio_src']]: subprocess.run(f"pactl unload-module {mod}", shell=True, stderr=subprocess.DEVNULL) + print("\n[INFO] Cleanup complete. Exiting.") 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() + server = AirLinkServer() + signal.signal(signal.SIGINT, lambda s,f: server.cleanup() or sys.exit(0)) + server.run() \ No newline at end of file