# Python实现多摄像头麦克风独立控制的完整方案 ## 基于多线程的实现（推荐用于Python） ```python import threading import time import subprocess import cv2 import pyaudio import os from datetime import datetime class CameraController: def __init__(self, camera_id, device_path=None): self.camera_id = camera_id self.device_path = device_path or f"/dev/video{camera_id}" self.is_running = False self.cap = None def start_capture(self): """启动摄像头捕获""" try: self.cap = cv2.VideoCapture(self.camera_id) if not self.cap.isOpened(): print(f"摄像头 {self.camera_id} 打开失败") return False self.is_running = True print(f"摄像头 {self.camera_id} 开始捕获") # 设置摄像头参数 self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640) self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) self.cap.set(cv2.CAP_PROP_FPS, 30) return True except Exception as e: print(f"摄像头 {self.camera_id} 启动错误: {e}") return False def capture_loop(self): """摄像头捕获循环""" frame_count = 0 while self.is_running: ret, frame = self.cap.read() if ret: frame_count += 1 # 保存图片示例（实际应用中可能不需要每帧保存） if frame_count % 100 == 0: filename = f"camera_{self.camera_id}_frame_{frame_count}.jpg" cv2.imwrite(filename, frame) print(f"摄像头 {self.camera_id} 保存帧: {filename}") # 这里可以添加图像处理代码 # process_frame(frame) time.sleep(0.033) # 约30fps def stop_capture(self): """停止摄像头捕获""" self.is_running = False if self.cap: self.cap.release() print(f"摄像头 {self.camera_id} 停止捕获") class MicrophoneController: def __init__(self, mic_id, device_index=None): self.mic_id = mic_id self.device_index = device_index or mic_id self.is_recording = False self.audio = pyaudio.PyAudio() self.stream = None def start_recording(self, output_file=None): """开始录音""" if not output_file: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_file = f"mic_{self.mic_id}_{timestamp}.wav" try: self.stream = self.audio.open( format=pyaudio.paInt16, channels=2, rate=44100, input=True, input_device_index=self.device_index, frames_per_buffer=1024 ) self.is_recording = True self.output_file = output_file print(f"麦克风 {self.mic_id} 开始录音: {output_file}") return True except Exception as e: print(f"麦克风 {self.mic_id} 启动错误: {e}") return False def record_loop(self, duration=10): """录音循环""" frames = [] start_time = time.time() while self.is_recording and (time.time() - start_time) < duration: try: data = self.stream.read(1024) frames.append(data) except Exception as e: print(f"麦克风 {self.mic_id} 录音错误: {e}") break # 保存录音文件 if frames: self._save_audio(frames) def _save_audio(self, frames): """保存音频数据到文件""" import wave wf = wave.open(self.output_file, 'wb') wf.setnchannels(2) wf.setsampwidth(self.audio.get_sample_size(pyaudio.paInt16)) wf.setframerate(44100) wf.writeframes(b''.join(frames)) wf.close() print(f"麦克风 {self.mic_id} 录音保存: {self.output_file}") def stop_recording(self): """停止录音""" self.is_recording = False if self.stream: self.stream.stop_stream() self.stream.close() self.audio.terminate() print(f"麦克风 {self.mic_id} 停止录音") def camera_thread_function(camera_id): """摄像头线程函数""" camera = CameraController(camera_id) if camera.start_capture(): camera.capture_loop() camera.stop_capture() def microphone_thread_function(mic_id, duration=30): """麦克风线程函数""" mic = MicrophoneController(mic_id) if mic.start_recording(): mic.record_loop(duration) mic.stop_recording() def discover_devices(): """发现系统中的音视频设备""" print("=== 系统设备发现 ===") # 发现摄像头 print("\n摄像头设备:") for i in range(5): cap = cv2.VideoCapture(i) if cap.isOpened(): print(f" 摄像头 {i}: 可用") cap.release() else: print(f" 摄像头 {i}: 不可用") # 发现麦克风 print("\n音频设备:") audio = pyaudio.PyAudio() for i in range(audio.get_device_count()): device_info = audio.get_device_info_by_index(i) if device_info['maxInputChannels'] > 0: print(f" 输入设备 {i}: {device_info['name']}") audio.terminate() def main_multithreading(): """多线程主函数""" print("启动多线程音视频控制系统...") # 发现设备 discover_devices() # 创建线程列表 threads = [] # 启动摄像头线程 camera_threads = [] for cam_id in [0, 1]: # 假设有两个摄像头 thread = threading.Thread(target=camera_thread_function, args=(cam_id,)) thread.daemon = True threads.append(thread) camera_threads.append(thread) # 启动麦克风线程 mic_threads = [] for mic_id in [1, 2]: # 假设有两个麦克风 thread = threading.Thread(target=microphone_thread_function, args=(mic_id, 30)) thread.daemon = True threads.append(thread) mic_threads.append(thread) # 启动所有线程 for thread in threads: thread.start() print("所有设备线程已启动，运行30秒...") # 主线程等待 try: time.sleep(30) print("程序运行完成") except KeyboardInterrupt: print("用户中断程序") # 基于多进程的实现（更稳定） import multiprocessing def camera_process_function(camera_id): """摄像头进程函数""" camera = CameraController(camera_id) if camera.start_capture(): camera.capture_loop() camera.stop_capture() def microphone_process_function(mic_id): """麦克风进程函数""" mic = MicrophoneController(mic_id) if mic.start_recording(): mic.record_loop(30) mic.stop_recording() def main_multiprocessing(): """多进程主函数""" print("启动多进程音视频控制系统...") processes = [] # 创建摄像头进程 for cam_id in [0, 1]: process = multiprocessing.Process(target=camera_process_function, args=(cam_id,)) processes.append(process) # 创建麦克风进程 for mic_id in [1, 2]: process = multiprocessing.Process(target=microphone_process_function, args=(mic_id,)) processes.append(process) # 启动所有进程 for process in processes: process.start() print("所有设备进程已启动") # 等待所有进程完成 for process in processes: process.join() print("所有进程执行完成") # 高级功能：使用队列进行进程间通信 def camera_process_with_queue(camera_id, output_queue): """带队列通信的摄像头进程""" camera = CameraController(camera_id) if camera.start_capture(): frame_count = 0 while camera.is_running and frame_count < 100: # 限制帧数用于演示 ret, frame = camera.cap.read() if ret: frame_count += 1 # 发送帧信息到队列 if frame_count % 10 == 0: output_queue.put({ 'camera_id': camera_id, 'frame_count': frame_count, 'timestamp': time.time() }) time.sleep(0.1) camera.stop_capture() output_queue.put({'camera_id': camera_id, 'status': 'finished'}) if __name__ == "__main__": # 安装所需包的命令： # pip install opencv-python pyaudio print("选择运行模式:") print("1. 多线程模式") print("2. 多进程模式") choice = input("请输入选择 (1 或 2): ") if choice == "1": main_multithreading() elif choice == "2": main_multiprocessing() else: print("无效选择，使用多线程模式") main_multithreading() ``` ## 设备配置和故障处理 ```python # 设备配置工具 class DeviceManager: @staticmethod def list_video_devices(): """列出所有视频设备""" import glob video_devices = glob.glob('/dev/video*') print("视频设备列表:") for device in sorted(video_devices): print(f" {device}") return video_devices @staticmethod def list_audio_devices(): """列出所有音频设备""" audio = pyaudio.PyAudio() print("音频输入设备列表:") for i in range(audio.get_device_count()): info = audio.get_device_info_by_index(i) if info['maxInputChannels'] > 0: print(f" 设备 {i}: {info['name']} (输入通道: {info['maxInputChannels']})") audio.terminate() @staticmethod def test_camera(camera_id): """测试摄像头""" try: cap = cv2.VideoCapture(camera_id) if cap.isOpened(): ret, frame = cap.read() if ret: print(f"摄像头 {camera_id} 测试成功") cv2.imwrite(f'test_camera_{camera_id}.jpg', frame) else: print(f"摄像头 {camera_id} 无法读取帧") else: print(f"摄像头 {camera_id} 无法打开") cap.release() except Exception as e: print(f"摄像头 {camera_id} 测试失败: {e}") @staticmethod def test_microphone(mic_id): """测试麦克风""" try: audio = pyaudio.PyAudio() stream = audio.open( format=pyaudio.paInt16, channels=1, rate=44100, input=True, input_device_index=mic_id, frames_per_buffer=1024 ) print(f"麦克风 {mic_id} 测试成功") stream.stop_stream() stream.close() audio.terminate() except Exception as e: print(f"麦克风 {mic_id} 测试失败: {e}") # 使用示例 if __name__ == "__main__": # 设备检测 DeviceManager.list_video_devices() DeviceManager.list_audio_devices() # 设备测试 DeviceManager.test_camera(0) DeviceManager.test_camera(1) DeviceManager.test_microphone(1) DeviceManager.test_microphone(2) ``` ## 关键特性说明 ### 1. **多线程优势** - 资源共享方便 - 创建和销毁开销小 - Python GIL限制但I/O操作影响不大 ### 2. **多进程优势** - 真正的并行执行 - 进程崩溃互不影响 - 更好的CPU利用率 ### 3. **设备独立性** - 每个摄像头/麦克风独立控制 - 独立的错误处理 - 独立的参数配置 ### 4. **扩展建议** - 可根据需要添加Web界面 - 可集成RTMP推流 - 可添加AI分析功能 这个Python实现提供了完整的设备控制框架，您可以根据具体硬件配置调整设备ID和参数。