【CI130x 离在线】Linux平台从命名管道接收PCM格式数据并转为WAV文件——Python脚本实现

我来为您提供一个完整的解决方案，使用Python实现从命名管道读取PCM数据并转换为WAV文件。

方案概述

我将使用Python实现，因为它有良好的音频处理库支持。程序会监听命名管道，当检测到数据流时开始录制，在数据流中断时自动结束并保存为WAV文件。

#!/usr/bin/env python3
"""
PCM数据采集和WAV转换程序
从命名管道 /tmp/VCM_Send 读取PCM数据并转换为WAV文件
PCM格式: 16kHz, 单声道, 16bit
"""import os
import sys
import time
import signal
import wave
import argparse
from pathlib import Pathclass PCMtoWAVConverter:def __init__(self, pipe_path, output_dir, sample_rate=16000, channels=1, sample_width=2):self.pipe_path = pipe_pathself.output_dir = Path(output_dir)self.sample_rate = sample_rateself.channels = channelsself.sample_width = sample_width# 创建输出目录self.output_dir.mkdir(exist_ok=True)# 确保命名管道存在self._ensure_pipe_exists()self.is_recording = Falseself.current_data = bytearray()def _ensure_pipe_exists(self):"""确保命名管道存在，如果不存在则创建"""if not os.path.exists(self.pipe_path):print(f"创建命名管道: {self.pipe_path}")os.mkfifo(self.pipe_path)elif not os.path.ispipe(self.pipe_path):print(f"错误: {self.pipe_path} 存在但不是命名管道")sys.exit(1)def generate_filename(self):"""生成基于时间戳的输出文件名"""timestamp = time.strftime("%Y%m%d_%H%M%S")return self.output_dir / f"recording_{timestamp}.wav"def save_as_wav(self, data, filename):"""将PCM数据保存为WAV文件"""try:with wave.open(str(filename), 'wb') as wav_file:wav_file.setnchannels(self.channels)wav_file.setsampwidth(self.sample_width)wav_file.setframerate(self.sample_rate)wav_file.writeframes(data)file_size = len(data)duration = file_size / (self.sample_rate * self.channels * self.sample_width)print(f"已保存: {filename} (时长: {duration:.2f}秒, 大小: {file_size}字节)")except Exception as e:print(f"保存WAV文件时出错: {e}")def monitor_pipe(self, silence_timeout=3.0, chunk_size=1024):"""监控命名管道并录制数据参数:silence_timeout: 静默超时时间(秒)，超过此时间无数据则认为录制结束chunk_size: 每次读取的数据块大小"""print(f"开始监控命名管道: {self.pipe_path}")print(f"参数: 采样率={self.sample_rate}Hz, 声道数={self.channels}, 位深度={self.sample_width*8}bit")print(f"静默超时: {silence_timeout}秒")print("按 Ctrl+C 停止程序")last_data_time = Nonetry:while True:# 以二进制模式打开命名管道with open(self.pipe_path, 'rb') as pipe:print("检测到数据流，开始录制...")self.is_recording = Trueself.current_data = bytearray()last_data_time = time.time()try:while True:data = pipe.read(chunk_size)current_time = time.time()if data:self.current_data.extend(data)last_data_time = current_timeelse:# 检查是否超时if current_time - last_data_time > silence_timeout:print("检测到数据流结束")break# 短暂休眠避免CPU占用过高time.sleep(0.1)except BrokenPipeError:print("管道连接中断")# 保存录制的数据if len(self.current_data) > 0:filename = self.generate_filename()self.save_as_wav(bytes(self.current_data), filename)self.is_recording = Falseexcept KeyboardInterrupt:print("\n程序被用户中断")except Exception as e:print(f"监控过程中出错: {e}")def continuous_monitor(self, silence_timeout=3.0, chunk_size=1024):"""持续监控模式，每次数据流都会创建新的WAV文件"""print("进入持续监控模式...")self.monitor_pipe(silence_timeout, chunk_size)def signal_handler(sig, frame):"""处理Ctrl+C信号"""print('\n程序退出')sys.exit(0)def main():parser = argparse.ArgumentParser(description='从命名管道读取PCM数据并转换为WAV文件')parser.add_argument('--pipe', default='/tmp/VCM_Send', help='命名管道路径 (默认: /tmp/VCM_Send)')parser.add_argument('--output', default='./recordings', help='输出目录 (默认: ./recordings)')parser.add_argument('--timeout', type=float, default=3.0, help='静默超时时间(秒) (默认: 3.0)')parser.add_argument('--chunk-size', type=int, default=1024, help='读取块大小 (默认: 1024)')args = parser.parse_args()# 注册信号处理signal.signal(signal.SIGINT, signal_handler)# 创建转换器实例converter = PCMtoWAVConverter(pipe_path=args.pipe,output_dir=args.output,sample_rate=16000,channels=1,sample_width=2)# 开始监控converter.continuous_monitor(silence_timeout=args.timeout,chunk_size=args.chunk_size)if __name__ == "__main__":main()

使用方法

1. 保存脚本：

sudo chmod +x pcm_to_wav.py

2. 运行程序：

# 基本用法
python3 pcm_to_wav.py# 自定义参数
python3 pcm_to_wav.py --output /path/to/output --timeout 5.0 --chunk-size 2048

3. 向管道发送数据测试：

# 在一个终端运行采集程序
python3 pcm_to_wav.py# 在另一个终端发送测试数据
cat test.pcm > /tmp/VCM_Send

自动启动脚本（systemd服务）

创建systemd服务文件 /etc/systemd/system/pcm-recorder.service：

[Unit]
Description=PCM to WAV Recorder
After=network.target[Service]
Type=simple
User=root
ExecStart=/usr/bin/python3 /path/to/pcm_to_wav.py --output /var/recordings --timeout 5.0
Restart=always
RestartSec=5[Install]
WantedBy=multi-user.target

启用服务：

sudo systemctl daemon-reload
sudo systemctl enable pcm-recorder.service
sudo systemctl start pcm-recorder.service

技术特点

1. 自动结束检测：使用超时机制检测数据流结束

2. 持续监控：程序会持续运行，每次数据流都会创建新的WAV文件

3. 错误处理：完善的异常处理机制

4. 灵活配置：支持命令行参数配置

5. 文件命名：基于时间戳的自动文件命名