Python 控制摄像头和麦克风

一、控制摄像头（OpenCV）

1. 安装依赖

pip install opencv-python
pip install numpy

2. 基础摄像头控制

import cv2
import numpy as np

class CameraController:
    def __init__(self, camera_index=0):
        """初始化摄像头"""
        self.camera = cv2.VideoCapture(camera_index)
        
        if not self.camera.isOpened():
            raise Exception("无法打开摄像头")
        
        # 设置摄像头参数
        self.camera.set(cv2.CAP_PROP_FRAME_WIDTH, 1920)  # 设置宽度
        self.camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080)  # 设置高度
        self.camera.set(cv2.CAP_PROP_FPS, 30)  # 设置帧率
        
    def get_frame(self):
        """读取一帧"""
        ret, frame = self.camera.read()
        if not ret:
            return None
        return frame
    
    def show_live_video(self):
        """显示实时视频"""
        print("按 'q' 退出, 's' 截图")
        
        while True:
            frame = self.get_frame()
            if frame is None:
                break
            
            # 显示帧
            cv2.imshow('Camera', frame)
            
            key = cv2.waitKey(1) & 0xFF
            if key == ord('q'):  # 按q退出
                break
            elif key == ord('s'):  # 按s截图
                cv2.imwrite('screenshot.jpg', frame)
                print("截图已保存")
        
        self.release()
    
    def save_video(self, filename='output.avi', duration=10):
        """录制视频"""
        # 定义编码器
        fourcc = cv2.VideoWriter_fourcc(*'XVID')
        fps = 30
        width = int(self.camera.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(self.camera.get(cv2.CAP_PROP_FRAME_HEIGHT))
        
        out = cv2.VideoWriter(filename, fourcc, fps, (width, height))
        
        print(f"开始录制 {duration} 秒...")
        frame_count = 0
        max_frames = duration * fps
        
        while frame_count < max_frames:
            frame = self.get_frame()
            if frame is None:
                break
            
            out.write(frame)
            cv2.imshow('Recording', frame)
            
            frame_count += 1
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
        
        out.release()
        cv2.destroyAllWindows()
        print(f"视频已保存: {filename}")
    
    def get_camera_info(self):
        """获取摄像头信息"""
        width = self.camera.get(cv2.CAP_PROP_FRAME_WIDTH)
        height = self.camera.get(cv2.CAP_PROP_FRAME_HEIGHT)
        fps = self.camera.get(cv2.CAP_PROP_FPS)
        
        print(f"摄像头分辨率: {int(width)}x{int(height)}")
        print(f"帧率: {int(fps)} FPS")
    
    def release(self):
        """释放摄像头"""
        self.camera.release()
        cv2.destroyAllWindows()

# 使用示例
if __name__ == "__main__":
    cam = CameraController(0)  # 0是默认摄像头
    cam.get_camera_info()
    cam.show_live_video()

3. 列出所有摄像头

def list_cameras():
    """列出所有可用摄像头"""
    index = 0
    cameras = []
    
    while True:
        cap = cv2.VideoCapture(index)
        if not cap.isOpened():
            break
        cameras.append(index)
        print(f"摄像头 {index}: 可用")
        cap.release()
        index += 1
    
    return cameras

# 使用
available_cameras = list_cameras()

二、控制麦克风（PyAudio）

1. 安装依赖

pip install pyaudio
pip install numpy
pip install wave

2. 基础麦克风控制

import pyaudio
import wave
import numpy as np

class MicrophoneController:
    def __init__(self, device_index=None):
        """初始化麦克风"""
        self.audio = pyaudio.PyAudio()
        self.device_index = device_index
        
        # 音频参数
        self.format = pyaudio.paInt16  # 16位深度
        self.channels = 1  # 单声道
        self.rate = 44100  # 采样率 44.1kHz
        self.chunk = 1024  # 缓冲区大小
        
    def list_devices(self):
        """列出所有音频设备"""
        print("\n可用音频输入设备：")
        for i in range(self.audio.get_device_count()):
            info = self.audio.get_device_info_by_index(i)
            if info['maxInputChannels'] > 0:  # 输入设备
                print(f"设备 {i}: {info['name']}")
                print(f"  输入声道: {info['maxInputChannels']}")
                print(f"  默认采样率: {int(info['defaultSampleRate'])} Hz\n")
    
    def record_audio(self, filename='output.wav', duration=5):
        """录制音频"""
        stream = self.audio.open(
            format=self.format,
            channels=self.channels,
            rate=self.rate,
            input=True,
            input_device_index=self.device_index,
            frames_per_buffer=self.chunk
        )
        
        print(f"开始录音 {duration} 秒...")
        frames = []
        
        for i in range(0, int(self.rate / self.chunk * duration)):
            data = stream.read(self.chunk)
            frames.append(data)
        
        print("录音完成")
        stream.stop_stream()
        stream.close()
        
        # 保存为WAV文件
        wf = wave.open(filename, 'wb')
        wf.setnchannels(self.channels)
        wf.setsampwidth(self.audio.get_sample_size(self.format))
        wf.setframerate(self.rate)
        wf.writeframes(b''.join(frames))
        wf.close()
        
        print(f"音频已保存: {filename}")
    
    def record_with_visualization(self, duration=10):
        """录音并实时显示音量"""
        stream = self.audio.open(
            format=self.format,
            channels=self.channels,
            rate=self.rate,
            input=True,
            input_device_index=self.device_index,
            frames_per_buffer=self.chunk
        )
        
        print(f"开始录音 {duration} 秒 (实时音量显示)...")
        frames = []
        
        for i in range(0, int(self.rate / self.chunk * duration)):
            data = stream.read(self.chunk)
            frames.append(data)
            
            # 计算音量
            audio_data = np.frombuffer(data, dtype=np.int16)
            volume = np.abs(audio_data).mean()
            
            # 可视化音量（0-50的条形图）
            bar_length = int(volume / 100)
            bar = '█' * min(bar_length, 50)
            print(f"\r音量: {bar:<50} {int(volume)}", end='', flush=True)
        
        print("\n录音完成")
        stream.stop_stream()
        stream.close()
        
        return frames
    
    def real_time_monitor(self):
        """实时监听麦克风音量"""
        stream = self.audio.open(
            format=self.format,
            channels=self.channels,
            rate=self.rate,
            input=True,
            input_device_index=self.device_index,
            frames_per_buffer=self.chunk
        )
        
        print("实时监听麦克风（按 Ctrl+C 停止）...")
        
        try:
            while True:
                data = stream.read(self.chunk)
                audio_data = np.frombuffer(data, dtype=np.int16)
                volume = np.abs(audio_data).mean()
                
                # 可视化
                bar_length = int(volume / 100)
                bar = '█' * min(bar_length, 50)
                print(f"\r音量: {bar:<50} {int(volume)}", end='', flush=True)
                
        except KeyboardInterrupt:
            print("\n停止监听")
        
        stream.stop_stream()
        stream.close()
    
    def close(self):
        """关闭PyAudio"""
        self.audio.terminate()

# 使用示例
if __name__ == "__main__":
    mic = MicrophoneController()
    
    # 列出所有设备
    mic.list_devices()
    
    # 录音5秒
    mic.record_audio('test.wav', duration=5)
    
    # 实时监听
    # mic.real_time_monitor()
    
    mic.close()

三、同时录制视频和音频

import cv2
import pyaudio
import wave
import threading
import numpy as np

class VideoAudioRecorder:
    def __init__(self, camera_index=0, audio_device=None):
        """初始化视频和音频录制器"""
        # 摄像头
        self.camera = cv2.VideoCapture(camera_index)
        self.camera.set(cv2.CAP_PROP_FRAME_WIDTH, 1920)
        self.camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080)
        
        # 麦克风
        self.audio = pyaudio.PyAudio()
        self.audio_device = audio_device
        self.audio_format = pyaudio.paInt16
        self.channels = 1
        self.rate = 44100
        self.chunk = 1024
        
        self.is_recording = False
        self.audio_frames = []
        
    def record_audio_thread(self):
        """音频录制线程"""
        stream = self.audio.open(
            format=self.audio_format,
            channels=self.channels,
            rate=self.rate,
            input=True,
            input_device_index=self.audio_device,
            frames_per_buffer=self.chunk
        )
        
        while self.is_recording:
            data = stream.read(self.chunk)
            self.audio_frames.append(data)
        
        stream.stop_stream()
        stream.close()
    
    def record(self, video_filename='output.avi', audio_filename='output.wav', duration=10):
        """同时录制视频和音频"""
        # 视频编码器
        fourcc = cv2.VideoWriter_fourcc(*'XVID')
        fps = 30
        width = int(self.camera.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(self.camera.get(cv2.CAP_PROP_FRAME_HEIGHT))
        video_writer = cv2.VideoWriter(video_filename, fourcc, fps, (width, height))
        
        # 开始音频录制线程
        self.is_recording = True
        self.audio_frames = []
        audio_thread = threading.Thread(target=self.record_audio_thread)
        audio_thread.start()
        
        print(f"开始录制 {duration} 秒...")
        frame_count = 0
        max_frames = duration * fps
        
        while frame_count < max_frames:
            ret, frame = self.camera.read()
            if not ret:
                break
            
            video_writer.write(frame)
            cv2.imshow('Recording', frame)
            
            frame_count += 1
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
        
        # 停止录制
        self.is_recording = False
        audio_thread.join()
        
        video_writer.release()
        cv2.destroyAllWindows()
        
        # 保存音频
        wf = wave.open(audio_filename, 'wb')
        wf.setnchannels(self.channels)
        wf.setsampwidth(self.audio.get_sample_size(self.audio_format))
        wf.setframerate(self.rate)
        wf.writeframes(b''.join(self.audio_frames))
        wf.close()
        
        print(f"视频已保存: {video_filename}")
        print(f"音频已保存: {audio_filename}")
    
    def release(self):
        """释放资源"""
        self.camera.release()
        self.audio.terminate()

# 使用示例
if __name__ == "__main__":
    recorder = VideoAudioRecorder(camera_index=0)
    recorder.record(duration=10)
    recorder.release()

四、使用 sounddevice（更简单的音频库）

pip install sounddevice
pip install scipy

import sounddevice as sd
import scipy.io.wavfile as wav
import numpy as np

class SimpleMicController:
    def __init__(self, sample_rate=44100):
        self.sample_rate = sample_rate
    
    def list_devices(self):
        """列出所有设备"""
        print(sd.query_devices())
    
    def record(self, duration=5, filename='output.wav'):
        """录音"""
        print(f"录音 {duration} 秒...")
        recording = sd.rec(
            int(duration * self.sample_rate),
            samplerate=self.sample_rate,
            channels=1,
            dtype='int16'
        )
        sd.wait()  # 等待录音完成
        
        # 保存
        wav.write(filename, self.sample_rate, recording)
        print(f"已保存: {filename}")
    
    def play_audio(self, filename):
        """播放音频"""
        sample_rate, data = wav.read(filename)
        sd.play(data, sample_rate)
        sd.wait()

# 使用
mic = SimpleMicController()
mic.list_devices()
mic.record(duration=5)

五、完整示例：带GUI的录制器

import cv2
import pyaudio
import wave
import threading
from datetime import datetime

class MediaRecorder:
    def __init__(self):
        self.camera = cv2.VideoCapture(0)
        self.audio = pyaudio.PyAudio()
        self.is_recording = False
        
    def start_recording(self):
        """开始录制"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        video_file = f"video_{timestamp}.avi"
        audio_file = f"audio_{timestamp}.wav"
        
        print("按 's' 开始/停止录制，按 'q' 退出")
        
        while True:
            ret, frame = self.camera.read()
            if not ret:
                break
            
            # 显示状态
            status = "录制中" if self.is_recording else "待机"
            cv2.putText(frame, status, (10, 30), 
                       cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
            
            cv2.imshow('Media Recorder', frame)
            
            key = cv2.waitKey(1) & 0xFF
            if key == ord('q'):
                break
            elif key == ord('s'):
                if not self.is_recording:
                    print("开始录制...")
                    # 实现录制逻辑
                else:
                    print("停止录制")
        
        self.camera.release()
        cv2.destroyAllWindows()

# 运行
recorder = MediaRecorder()
recorder.start_recording()

选择建议