#!/usr/bin/env python3 """ 3DS Webcam Bridge - PC Server Receives camera streams from 3DS and exposes them as virtual webcams. Supports mono and stereo modes with multiple output configurations. """ import argparse import io import socket import struct import sys import time import platform from typing import Optional, Tuple import numpy as np from PIL import Image # Backend selection BACKEND_PYVIRTUALCAM = "pyvirtualcam" BACKEND_PYFAKEWEBCAM = "pyfakewebcam" BACKEND_FFMPEG = "ffmpeg" # Protocol constants PROTO_VERSION = 0x01 PROTO_FLAG_STEREO = 0x01 # Global state g_verbose = False def log(msg): """Print log message.""" print(f"[INFO] {msg}") def log_verbose(msg): """Print verbose log message.""" if g_verbose: print(f"[VERBOSE] {msg}") def log_error(msg): """Print error message.""" print(f"[ERROR] {msg}", file=sys.stderr) def recv_all(conn: socket.socket, n: int) -> bytes: """Receive exactly n bytes from socket.""" data = b'' while len(data) < n: chunk = conn.recv(n - len(data)) if not chunk: raise ConnectionError("Connection closed by client") data += chunk return data def read_frame(conn: socket.socket) -> Tuple[Optional[Image.Image], Optional[Image.Image]]: """ Read a frame from the 3DS. Returns (left_image, right_image) where right_image is None for mono mode. """ # Read header header = recv_all(conn, 2) flags = header[0] version = header[1] if version != PROTO_VERSION: log_error(f"Unsupported protocol version: {version}") return None, None stereo = (flags & PROTO_FLAG_STEREO) != 0 if not stereo: # Mono frame: [4-byte length][JPEG] length_bytes = recv_all(conn, 4) length = struct.unpack('>I', length_bytes)[0] log_verbose(f"Receiving mono frame: {length} bytes") jpeg_data = recv_all(conn, length) left_img = Image.open(io.BytesIO(jpeg_data)).convert('RGB') return left_img, None else: # Stereo frame: [2B wL][2B hL][2B wR][2B hR][4B left_len][4B right_len][left_jpeg][right_jpeg] dims = recv_all(conn, 8) wL, hL, wR, hR = struct.unpack('>HHHH', dims) lengths = recv_all(conn, 8) left_len, right_len = struct.unpack('>II', lengths) log_verbose(f"Receiving stereo frame: left={wL}x{hL} ({left_len} bytes), right={wR}x{hR} ({right_len} bytes)") left_data = recv_all(conn, left_len) right_data = recv_all(conn, right_len) left_img = Image.open(io.BytesIO(left_data)).convert('RGB') right_img = Image.open(io.BytesIO(right_data)).convert('RGB') return left_img, right_img def compose_sbs(left: Image.Image, right: Image.Image, width: int, height: int) -> np.ndarray: """Compose side-by-side stereo image.""" left_resized = left.resize((width, height), Image.BILINEAR) right_resized = right.resize((width, height), Image.BILINEAR) left_arr = np.array(left_resized) right_arr = np.array(right_resized) # Concatenate horizontally sbs = np.concatenate([left_arr, right_arr], axis=1) return sbs def compose_anaglyph(left: Image.Image, right: Image.Image, width: int, height: int) -> np.ndarray: """Compose red-cyan anaglyph image.""" left_resized = left.resize((width, height), Image.BILINEAR) right_resized = right.resize((width, height), Image.BILINEAR) left_arr = np.array(left_resized) right_arr = np.array(right_resized) # Red-cyan anaglyph: red from left, green+blue from right anaglyph = np.zeros_like(left_arr) anaglyph[:, :, 0] = left_arr[:, :, 0] # Red from left anaglyph[:, :, 1] = right_arr[:, :, 1] # Green from right anaglyph[:, :, 2] = right_arr[:, :, 2] # Blue from right return anaglyph class VirtualCameraBackend: """Base class for virtual camera backends.""" def __init__(self, args): self.args = args def send_frame(self, frame: np.ndarray): """Send a frame to the virtual camera.""" raise NotImplementedError def send_dual_frames(self, left: np.ndarray, right: np.ndarray): """Send dual frames (for dual device mode).""" raise NotImplementedError def close(self): """Close the backend.""" pass class PyVirtualCamBackend(VirtualCameraBackend): """Backend using pyvirtualcam.""" def __init__(self, args): super().__init__(args) try: import pyvirtualcam self.pyvirtualcam = pyvirtualcam except ImportError: raise RuntimeError("pyvirtualcam is not installed. Install with: pip install pyvirtualcam") self.camera = None self.camera_left = None self.camera_right = None def _open_single_camera(self, width, height, fps, device=None): """Open a single pyvirtualcam device.""" kwargs = { 'width': width, 'height': height, 'fps': fps, 'print_fps': g_verbose } if device is not None: # Device can be a name (string) or index (int) try: device_int = int(device) kwargs['device'] = device_int except ValueError: kwargs['device'] = device return self.pyvirtualcam.Camera(**kwargs) def setup_single(self): """Setup single camera output.""" width = self.args.width height = self.args.height # For SBS mode, double the width if self.args.stereo_mode == 'sbs': width = width * 2 self.camera = self._open_single_camera( width=width, height=height, fps=self.args.fps, device=self.args.device ) log(f"Opened virtual camera: {width}x{height} @ {self.args.fps} FPS") def setup_dual(self): """Setup dual camera output.""" devices = self.args.dual_devices.split(',') if len(devices) != 2: raise ValueError("dual_devices must specify exactly 2 devices separated by comma") left_device = devices[0].strip() right_device = devices[1].strip() self.camera_left = self._open_single_camera( width=self.args.width, height=self.args.height, fps=self.args.fps, device=left_device if left_device else None ) self.camera_right = self._open_single_camera( width=self.args.width, height=self.args.height, fps=self.args.fps, device=right_device if right_device else None ) log(f"Opened dual virtual cameras: {self.args.width}x{self.args.height} @ {self.args.fps} FPS") def send_frame(self, frame: np.ndarray): """Send frame to single camera.""" if self.camera is None: self.setup_single() self.camera.send(frame) def send_dual_frames(self, left: np.ndarray, right: np.ndarray): """Send frames to dual cameras.""" if self.camera_left is None or self.camera_right is None: self.setup_dual() self.camera_left.send(left) self.camera_right.send(right) def close(self): """Close cameras.""" if self.camera: self.camera.close() if self.camera_left: self.camera_left.close() if self.camera_right: self.camera_right.close() class PyFakeWebcamBackend(VirtualCameraBackend): """Backend using pyfakewebcam (Linux only).""" def __init__(self, args): super().__init__(args) if platform.system() != 'Linux': raise RuntimeError("pyfakewebcam is only available on Linux") try: import pyfakewebcam self.pyfakewebcam = pyfakewebcam except ImportError: raise RuntimeError("pyfakewebcam is not installed. Install with: pip install pyfakewebcam") self.camera = None self.camera_left = None self.camera_right = None def setup_single(self): """Setup single camera output.""" width = self.args.width height = self.args.height if self.args.stereo_mode == 'sbs': width = width * 2 device = self.args.device if self.args.device else '/dev/video0' self.camera = self.pyfakewebcam.FakeWebcam(device, width, height) log(f"Opened pyfakewebcam device: {device} ({width}x{height})") def setup_dual(self): """Setup dual camera output.""" devices = self.args.dual_devices.split(',') if len(devices) != 2: raise ValueError("dual_devices must specify exactly 2 devices separated by comma") self.camera_left = self.pyfakewebcam.FakeWebcam( devices[0].strip(), self.args.width, self.args.height) self.camera_right = self.pyfakewebcam.FakeWebcam( devices[1].strip(), self.args.width, self.args.height) log(f"Opened dual pyfakewebcam devices: {devices[0]}, {devices[1]}") def send_frame(self, frame: np.ndarray): """Send frame to single camera.""" if self.camera is None: self.setup_single() self.camera.schedule_frame(frame) def send_dual_frames(self, left: np.ndarray, right: np.ndarray): """Send frames to dual cameras.""" if self.camera_left is None or self.camera_right is None: self.setup_dual() self.camera_left.schedule_frame(left) self.camera_right.schedule_frame(right) class FFmpegBackend(VirtualCameraBackend): """Backend using ffmpeg pipe to v4l2 (Linux only).""" def __init__(self, args): super().__init__(args) if platform.system() != 'Linux': raise RuntimeError("ffmpeg backend is only available on Linux") import subprocess self.subprocess = subprocess self.process = None self.process_left = None self.process_right = None def setup_single(self): """Setup single ffmpeg output.""" width = self.args.width height = self.args.height if self.args.stereo_mode == 'sbs': width = width * 2 device = self.args.device if self.args.device else '/dev/video0' cmd = [ 'ffmpeg', '-f', 'rawvideo', '-pixel_format', 'rgb24', '-video_size', f'{width}x{height}', '-framerate', str(self.args.fps), '-i', '-', '-f', 'v4l2', device ] self.process = self.subprocess.Popen(cmd, stdin=self.subprocess.PIPE) log(f"Opened ffmpeg pipe to {device} ({width}x{height})") def setup_dual(self): """Setup dual ffmpeg output.""" devices = self.args.dual_devices.split(',') if len(devices) != 2: raise ValueError("dual_devices must specify exactly 2 devices") cmd_template = [ 'ffmpeg', '-f', 'rawvideo', '-pixel_format', 'rgb24', '-video_size', f'{self.args.width}x{self.args.height}', '-framerate', str(self.args.fps), '-i', '-', '-f', 'v4l2', None # Device placeholder ] cmd_left = cmd_template[:] cmd_left[-1] = devices[0].strip() self.process_left = self.subprocess.Popen(cmd_left, stdin=self.subprocess.PIPE) cmd_right = cmd_template[:] cmd_right[-1] = devices[1].strip() self.process_right = self.subprocess.Popen(cmd_right, stdin=self.subprocess.PIPE) log(f"Opened dual ffmpeg pipes to {devices[0]}, {devices[1]}") def send_frame(self, frame: np.ndarray): """Send frame via ffmpeg.""" if self.process is None: self.setup_single() self.process.stdin.write(frame.tobytes()) self.process.stdin.flush() def send_dual_frames(self, left: np.ndarray, right: np.ndarray): """Send frames via dual ffmpeg pipes.""" if self.process_left is None or self.process_right is None: self.setup_dual() self.process_left.stdin.write(left.tobytes()) self.process_left.stdin.flush() self.process_right.stdin.write(right.tobytes()) self.process_right.stdin.flush() def close(self): """Close ffmpeg processes.""" if self.process: self.process.stdin.close() self.process.wait() if self.process_left: self.process_left.stdin.close() self.process_left.wait() if self.process_right: self.process_right.stdin.close() self.process_right.wait() def create_backend(args) -> VirtualCameraBackend: """Create appropriate backend based on preference.""" backend_name = args.prefer_backend if backend_name == BACKEND_PYVIRTUALCAM: try: return PyVirtualCamBackend(args) except RuntimeError as e: log_error(f"Failed to create pyvirtualcam backend: {e}") if platform.system() == 'Linux': log("Falling back to pyfakewebcam...") backend_name = BACKEND_PYFAKEWEBCAM else: raise if backend_name == BACKEND_PYFAKEWEBCAM: try: return PyFakeWebcamBackend(args) except RuntimeError as e: log_error(f"Failed to create pyfakewebcam backend: {e}") log("Falling back to ffmpeg...") backend_name = BACKEND_FFMPEG if backend_name == BACKEND_FFMPEG: return FFmpegBackend(args) raise RuntimeError(f"No suitable backend found") def process_stream(conn: socket.socket, addr, args): """Process incoming stream from 3DS.""" log(f"Client connected from {addr}") backend = create_backend(args) frame_count = 0 start_time = time.time() try: while True: # Read frame from 3DS left_img, right_img = read_frame(conn) if left_img is None: break # Process based on mode if right_img is None: # Mono mode log_verbose("Processing mono frame") left_resized = left_img.resize((args.width, args.height), Image.BILINEAR) frame = np.array(left_resized) backend.send_frame(frame) else: # Stereo mode log_verbose("Processing stereo frame") if args.stereo_mode == 'off': # Send left eye only left_resized = left_img.resize((args.width, args.height), Image.BILINEAR) frame = np.array(left_resized) backend.send_frame(frame) elif args.stereo_mode == 'sbs': # Side-by-side frame = compose_sbs(left_img, right_img, args.width, args.height) backend.send_frame(frame) elif args.stereo_mode == 'anaglyph': # Red-cyan anaglyph frame = compose_anaglyph(left_img, right_img, args.width, args.height) backend.send_frame(frame) elif args.stereo_mode == 'dual': # Dual devices left_resized = left_img.resize((args.width, args.height), Image.BILINEAR) right_resized = right_img.resize((args.width, args.height), Image.BILINEAR) left_arr = np.array(left_resized) right_arr = np.array(right_resized) backend.send_dual_frames(left_arr, right_arr) frame_count += 1 if frame_count % 30 == 0: elapsed = time.time() - start_time fps = frame_count / elapsed log(f"Processed {frame_count} frames ({fps:.1f} FPS)") except ConnectionError as e: log_error(f"Connection error: {e}") except KeyboardInterrupt: log("Interrupted by user") except Exception as e: log_error(f"Error processing stream: {e}") import traceback traceback.print_exc() finally: backend.close() log(f"Client disconnected: {addr}") def main(): global g_verbose parser = argparse.ArgumentParser( description='3DS Webcam Bridge - PC Server', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Linux - mono mode with default settings %(prog)s --device /dev/video0 # Linux - side-by-side stereo %(prog)s --stereo-mode sbs --width 400 --height 240 --device /dev/video0 # Linux - dual devices for left/right eyes %(prog)s --stereo-mode dual --dual-devices /dev/video0,/dev/video1 # Windows - side-by-side with OBS Virtual Camera %(prog)s --stereo-mode sbs --device "OBS Virtual Camera" # Windows - dual OBS virtual cameras %(prog)s --stereo-mode dual --dual-devices "OBS Virtual Camera,OBS Virtual Camera 2" """ ) parser.add_argument('--listen', default='0.0.0.0', help='Listen address (default: 0.0.0.0)') parser.add_argument('--port', type=int, default=9000, help='Listen port (default: 9000)') parser.add_argument('--mode', choices=['mono', 'auto'], default='auto', help='Force mono or auto-detect (default: auto)') parser.add_argument('--stereo-mode', choices=['off', 'sbs', 'anaglyph', 'dual'], default='off', help='Stereo output mode (default: off)') parser.add_argument('--device', default=None, help='Virtual camera device (Linux: /dev/videoX, Windows: device name or index)') parser.add_argument('--dual-devices', default=None, help='Dual devices for stereo (comma-separated)') parser.add_argument('--width', type=int, default=640, help='Output width per eye (default: 640)') parser.add_argument('--height', type=int, default=480, help='Output height (default: 480)') parser.add_argument('--fps', type=int, default=15, help='Output FPS (default: 15)') parser.add_argument('--prefer-backend', choices=[BACKEND_PYVIRTUALCAM, BACKEND_PYFAKEWEBCAM, BACKEND_FFMPEG], default=BACKEND_PYVIRTUALCAM, help='Preferred backend (default: pyvirtualcam)') parser.add_argument('--verbose', action='store_true', help='Verbose output') args = parser.parse_args() g_verbose = args.verbose # Validate arguments if args.stereo_mode == 'dual' and not args.dual_devices: parser.error("--dual-devices is required when using --stereo-mode dual") # Create listening socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: sock.bind((args.listen, args.port)) sock.listen(1) log(f"Listening on {args.listen}:{args.port}") log(f"Stereo mode: {args.stereo_mode}") log(f"Backend: {args.prefer_backend}") while True: log("Waiting for connection...") conn, addr = sock.accept() try: process_stream(conn, addr, args) finally: conn.close() except KeyboardInterrupt: log("\nShutting down...") except Exception as e: log_error(f"Server error: {e}") import traceback traceback.print_exc() finally: sock.close() if __name__ == '__main__': main()