AirLink/server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add working mic functionality

Browse source
This commit is contained in:
srtk 2025-11-30 14:42:09 +05:30
parent f582f317c0
commit 6256b4ecd6
1 changed files with 121 additions and 250 deletions
Download patch file Download diff file Expand all files Collapse all files

357
server.py
View file

@ -1,277 +1,148 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import socket import socket, subprocess, sys, time, signal, threading, select, math
import subprocess
import sys
import select
import time
import signal
import threading
import logging
from zeroconf import ServiceInfo, Zeroconf 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 # Configuration
# --------------------------- CONF = {
UDP_PORT = 5000 'video_port': 5000, 'audio_port': 5001,
DEVICE = '/dev/video10' 'video_dev': '/dev/video20', 'audio_sink': 'AirLink_Sink', 'audio_src': 'AirLink_Mic'
BUFFER_SIZE = 65535 }
HEARTBEAT_INTERVAL = 2
CONNECTION_TIMEOUT = 5
class AirLinkServer:
class StreamServer: def __init__(self):
def __init__(self, host_ip): self.ip = self.get_lan_ip()
self.host_ip = host_ip
self.running = False self.running = False
self.ffmpeg_process = None self.procs = {'video': None, 'audio': None, 'ffmpeg_audio': None}
self.socket = None self.sockets = {'video': None, 'audio': None}
self.zeroconf = None self.stats = {'v_bytes': 0, 'a_bytes': 0, 'v_pkts': 0, 'a_pkts': 0, 'last_time': time.time()}
self.zc = None
self.last_packet_time = 0 self.lock = threading.Lock()
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
]
def get_lan_ip(self):
"""Finds LAN IP by prioritizing physical interfaces over VPN/Tun interfaces."""
try: try:
self.ffmpeg_process = subprocess.Popen( # List global IPv4 addresses, prioritize eth/wlan
cmd, cmd = "ip -4 -o addr show scope global | awk '{print $2,$4}'"
stdin=subprocess.PIPE, out = subprocess.check_output(cmd, shell=True).decode().strip().split('\n')
stderr=subprocess.PIPE, candidates = []
stdout=subprocess.DEVNULL, for line in out:
bufsize=0 if not line: continue
) iface, ip = line.split()[0], line.split()[1].split('/')[0]
logger.info(f"FFmpeg started (PID {self.ffmpeg_process.pid})") if iface.startswith(('e', 'w')): return ip # Found physical interface
self.restart_attempts = 0 candidates.append(ip)
return True 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: except Exception as e:
logger.error(f"Failed to start FFmpeg: {e}") print(f"[ERR] Setup failed: {e}")
return False return False
# ---------------------------------------------------------------------- def paint_black(self):
with self.lock:
def restart_ffmpeg(self): if self.procs['video']:
if self.restart_attempts >= self.max_restart_attempts: self.procs['video'].terminate()
logger.warning("Maximum FFmpeg restart attempts reached") self.procs['video'] = None
return False
self.restart_attempts += 1
logger.info(f"Restarting FFmpeg (attempt {self.restart_attempts})")
if self.ffmpeg_process:
try: try:
self.ffmpeg_process.terminate() 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)
self.ffmpeg_process.wait(timeout=2) print("[INFO] Video stream stopped (Black Screen)")
except: except: pass
try:
self.ffmpeg_process.kill()
except:
pass
time.sleep(0.5) def start_pipeline(self, p_type):
return self.start_ffmpeg() 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): def log_stats(self):
if self.ffmpeg_process and self.ffmpeg_process.poll() is not None: now = time.time()
return False delta = now - self.stats['last_time']
return True 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)")
def send_heartbeat(self): self.stats = {'v_bytes': 0, 'a_bytes': 0, 'v_pkts': 0, 'a_pkts': 0, 'last_time': now}
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): def run(self):
if not self.setup_system(): return
self.running = True self.running = True
service_info = self.start_discovery()
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # Discovery
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, BUFFER_SIZE * 8) info = ServiceInfo("_procam._udp.local.", "AirLink._procam._udp.local.", addresses=[socket.inet_aton(self.ip)], port=CONF['video_port'], properties={'version': '2.0'})
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.zc = Zeroconf(); self.zc.register_service(info)
self.socket.bind(("0.0.0.0", UDP_PORT))
self.socket.setblocking(0)
logger.info("ProCam server initialized") # Sockets
logger.info(f"Listening on {self.host_ip}:{UDP_PORT}") for k, port, buf in [('video', CONF['video_port'], 4*1024*1024), ('audio', CONF['audio_port'], 512*1024)]:
logger.info(f"Output device: {DEVICE}") s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
logger.info("Waiting for incoming stream...") s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, buf)
s.bind(("0.0.0.0", port))
s.setblocking(0)
self.sockets[k] = s
if not self.start_ffmpeg(): self.start_pipeline('video'); self.start_pipeline('audio')
return print(f"\n[READY] Server listening on {self.ip}:{CONF['video_port']}\n")
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: while self.running:
if not self.check_ffmpeg_health(): self.log_stats()
logger.error("FFmpeg has stopped unexpectedly") # Ensure audio is alive
if not self.restart_ffmpeg(): if self.procs['audio'] and self.procs['audio'].poll() is not None: self.start_pipeline('audio')
break
ready = select.select([self.socket], [], [], 0.05) ready, _, _ = select.select(list(self.sockets.values()), [], [], 0.1)
if ready[0]: for sock in ready:
try: try:
data, addr = self.socket.recvfrom(BUFFER_SIZE) data, addr = sock.recvfrom(65535)
if data == b'HEARTBEAT': # Control Signals
continue 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 not self.connected or self.client_addr != addr: # Data Routing
self.client_addr = addr 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()
self.last_packet_time = time.time() 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
try: def cleanup(self):
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 self.running = False
self.paint_black()
if self.ffmpeg_process: if self.zc: self.zc.close()
try: for mod in [CONF['audio_sink'], CONF['audio_src']]: subprocess.run(f"pactl unload-module {mod}", shell=True, stderr=subprocess.DEVNULL)
self.ffmpeg_process.terminate() print("\n[INFO] Cleanup complete. Exiting.")
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__": if __name__ == "__main__":
host_ip = '192.168.1.8' server = AirLinkServer()
server = StreamServer(host_ip) signal.signal(signal.SIGINT, lambda s,f: server.cleanup() or sys.exit(0))
def signal_handler(sig, frame):
server.running = False
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
server.run() server.run()