chenyuxuan/arcs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: migrate project into clean repository

Browse Source
This commit is contained in:
yuuux
2026-08-13 16:50:52 +08:00
commit d1d25a09e7
27405 changed files with 9422808 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
tools/assets_cpy.py Executable file
View File

@@ -0,0 +1,53 @@
#!/usr/bin/env python3
import os
import json
import shutil
import glob
import sys
from pathlib import Path
def copy_assets(config_path, output_dir):
"""Copy assets based on the configuration"""
# Create output directory if it doesn't exist
os.makedirs(output_dir, exist_ok=True)
# Load the configuration
with open(config_path, 'r') as f:
config = json.load(f)
# Process each copy operation
for op in config.get('copy_operations', []):
source = op['source']
destination = os.path.join(output_dir, op['destination'])
# Create destination directory if it doesn't exist
dest_dir = os.path.dirname(destination)
if dest_dir:
os.makedirs(dest_dir, exist_ok=True)
print(f"Copying {source} to {destination}")
# Handle wildcard in source
if '*' in source:
# For glob patterns, copy each matching file
for src_file in glob.glob(source):
if os.path.isfile(src_file):
dest = os.path.join(dest_dir, os.path.basename(src_file)) if os.path.isdir(destination) else destination
shutil.copy2(src_file, dest)
else:
# For single files
if os.path.isfile(source):
shutil.copy2(source, destination)
else:
print(f"Warning: Source file not found: {source}", file=sys.stderr)
if __name__ == "__main__":
if len(sys.argv) != 3:
print(f"Usage: {sys.argv[0]} <config_path> <output_dir>")
sys.exit(1)
config_path = sys.argv[1]
output_dir = sys.argv[2]
copy_assets(config_path, output_dir)

224
tools/audio/README_GUI.md Normal file
View File

@@ -0,0 +1,224 @@
# USB CDC 录音工具 GUI 版本使用说明
## 功能特性
### 🎯 核心功能
- **实时波形显示**：以图形方式显示录音波形
- **文件保存可选**：可选择是否保存音频文件
- **MD5 校验**：自动验证数据完整性
- **状态监控**：实时显示录音状态、字节数、序列号
### ✨ 新增功能：文件保存选项
现在你可以选择是否保存音频文件到磁盘：
- **✅ 保存音频文件**（默认）：录音数据会保存到指定的文件
- **❌ 不保存文件**：仅显示实时波形，不占用磁盘空间
## 使用方法
### 启动工具
```bash
cd /media/oem/Data1/project/arcs/voiceassistant/tools/audio
python3 serial_capture_gui.py
```
### 配置参数
1. **串口设置**
- 串口路径：默认 `/dev/ttyACM0`
- 波特率：默认 `921600`（可选择其他波特率）
2. **文件保存选项**
- ✅ **勾选**"保存音频文件"：录音会保存到文件
- ❌ **取消勾选**：仅显示波形，不保存文件
3. **输出文件**（仅当启用保存时）
- 输入文件路径或点击"浏览..."选择
- 支持格式：`.pcm`、`.wav` 等
### 操作流程
#### 场景1：录音并保存文件
1. ✅ 勾选"保存音频文件"
2. 设置输出文件路径
3. 点击"开始录音"
4. 观察实时波形
5. 点击"停止录音"
6. 查看MD5校验结果
7. 音频文件已保存到指定位置
#### 场景2：仅查看波形（不保存）
1. ❌ 取消勾选"保存音频文件"
2. 输出文件路径会自动禁用（灰色）
3. 点击"开始录音"
4. 观察实时波形
5. 点击"停止录音"
6. 查看MD5校验结果
7. **无文件保存，不占用磁盘空间**
## 界面说明
```
┌─────────────────────────────────────────────┐
│ 配置区域 │
│ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ │
│ 串口: /dev/ttyACM0 波特率: 921600 │
│ ☑ 保存音频文件 │
│ 输出文件: output.pcm [浏览...] │
│ │
│ 控制按钮 │
│ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ │
│ [开始录音] [停止录音] │
│ │
│ 实时波形显示 │
│ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ │
│ ┌───────────────────────────────────────┐ │
│ │ 📊 音频波形（实时更新） │ │
│ │ ╱╲ ╱╲ ╱╲ │ │
│ │ ╱ ╲ ╱ ╲ ╱ ╲ │ │
│ │ ╱ ╲ ╲ │ │
│ └───────────────────────────────────────┘ │
│ │
│ 状态信息 │
│ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ │
│ 状态: 正在录音并保存到: output.pcm │
│ 已接收: 102400 字节 序列号: 10 │
│ PC MD5: abc123def456... │
│ 设备 MD5: abc123def456... │
└─────────────────────────────────────────────┘
```
## 状态信息说明
### 录音时的状态
| 状态文本 | 含义 |
|---------|------|
| `正在录音并保存到: xxx.pcm` | 正在录音并保存文件 |
| `正在录音（仅显示波形，不保存文件）` | 仅查看波形，不保存 |
### 完成时的状态
| 状态文本 | 含义 | 颜色 |
|---------|------|------|
| `录音完成，MD5校验通过 ✓` | 文件已保存，校验成功 | 🟢 绿色 |
| `录音完成（未保存文件），MD5校验通过 ✓` | 未保存文件，但数据完整 | 🟢 绿色 |
| `录音完成，MD5校验失败 ✗` | 文件已保存，但数据损坏 | 🔴 红色 |
| `录音完成（未保存文件），MD5校验失败 ✗` | 未保存文件，数据损坏 | 🔴 红色 |
## 使用场景
### 💾 需要保存文件的场景
- 永久保存录音数据
- 后续音频分析
- 存档备份
### 👁️ 仅查看波形的场景
- 快速测试设备连接
- 检查音频信号质量
- 调试麦克风或音频输入
- 临时监控，不需要保存
## 技术细节
### 文件保存逻辑
```python
# 启用保存文件
☑ 保存音频文件
↓
打开文件 → 写入PCM数据 → 关闭文件
# 不保存文件
☐ 保存音频文件
↓
不打开文件 → 数据只用于显示波形和MD5计算
```
### 波形显示
- **采样率假设**：16 kHz
- **显示范围**：最近 1 秒的音频数据
- **幅度范围**：-32768 到 32767（16位PCM）
- **更新频率**：每 100ms 刷新一次
### MD5 校验
无论是否保存文件，**MD5 校验始终会进行**：
- PC端：对接收到的所有PCM数据计算MD5
- 设备端：对发送的所有PCM数据计算MD5
- 两者对比，验证传输完整性
## 常见问题
### Q1: 不保存文件时，MD5 还有意义吗？
**A**: 有！MD5校验用于验证**数据传输的完整性**，即使不保存文件，也能确认数据在传输过程中没有损坏或丢失。
### Q2: 波形显示会影响性能吗？
**A**: 波形显示使用队列缓冲和限流机制，对性能影响极小。如果队列满，旧数据会被自动丢弃。
### Q3: 可以中途切换是否保存文件吗？
**A**: 不可以。必须在开始录音前设置。如需更改，请停止录音后重新配置。
### Q4: 不保存文件能节省多少空间？
**A**:
- **录音 1 分钟**（16kHz，16bit，单声道）≈ 1.92 MB
- **录音 10 分钟** ≈ 19.2 MB
- **录音 1 小时** ≈ 115 MB
选择"不保存文件"可以完全避免磁盘占用。
## 依赖库
```bash
# 安装依赖（如果缺少）
pip3 install pyserial numpy matplotlib
```
已包含在标准安装中：
- `tkinter` - GUI 框架（Python 标准库）
- `serial` - 串口通信
- `numpy` - 数据处理
- `matplotlib` - 波形绘制
## 与命令行版本对比
| 特性 | GUI 版本 | 命令行版本 |
|------|---------|-----------|
| 波形显示 | ✅ 实时图形 | ❌ 无 |
| 文件保存选项 | ✅ 可选 | ❌ 总是保存 |
| 使用难度 | 🟢 简单（点击） | 🟡 需要参数 |
| 状态监控 | ✅ 图形化 | 📝 文本输出 |
| 适用场景 | 交互式使用 | 自动化脚本 |
## 快捷操作
### 快速测试（不保存文件）
1. 取消勾选"保存音频文件"
2. 点击"开始录音"
3. 观察波形是否正常
4. 点击"停止录音"
### 正式录音（保存文件）
1. 勾选"保存音频文件"
2. 点击"浏览..."选择保存位置
3. 点击"开始录音"
4. 等待录音完成
5. 点击"停止录音"
6. 确认 MD5 校验通过 ✓
---
**提示**：首次使用建议先取消勾选"保存音频文件"进行测试，确认连接正常后再保存文件。

558
tools/audio/serial_capture.py Executable file
View File

@@ -0,0 +1,558 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
USB CDC串口录音脚本 - 使用自定义协议
详细协议规范请参考：docs/usb_cdc_protocol.md
"""
import serial
import argparse
import sys
import os
import time
import hashlib
import struct
import threading
from datetime import datetime
from enum import IntEnum
class FrameType(IntEnum):
"""帧类型定义"""
CMD_REQUEST = 0x01
CMD_RESPONSE = 0x02
AUDIO_DATA = 0x03
MD5_DATA = 0x04
class CommandID(IntEnum):
"""命令码定义"""
START_RECORD = 0x01
STOP_RECORD = 0x02
QUERY_STATUS = 0x03
class StatusCode(IntEnum):
"""状态码定义"""
OK = 0x00
ERROR = 0x01
BUSY = 0x02
UNSUPPORTED = 0x03
class ParseState(IntEnum):
"""解析状态机"""
MAGIC1 = 0
MAGIC2 = 1
TYPE = 2
LENGTH = 3
DATA = 4
CHECKSUM = 5
class CDCProtocol:
"""CDC协议处理类"""
MAGIC_MSB = 0xAA
MAGIC_LSB = 0x55
HEADER_SIZE = 7
# 数据长度限制 - 区分命令帧和音频帧
MAX_CMD_DATA_SIZE = 64 # 命令/响应/MD5最大数据长度
def __init__(self):
self.reset()
def reset(self):
"""重置解析器状态"""
self.state = ParseState.MAGIC1
self.frame_type = 0
self.data_length = 0
self.data = bytearray()
self.length_buf = bytearray()
self.checksum = 0
@staticmethod
def calc_checksum(data):
"""计算XOR校验和"""
checksum = 0
for byte in data:
checksum ^= byte
return checksum
def build_frame(self, frame_type, data=b''):
"""构建协议帧"""
# 构建帧头
frame = bytearray()
frame.append(self.MAGIC_MSB)
frame.append(self.MAGIC_LSB)
frame.append(frame_type)
# 添加长度（小端序）
frame.extend(struct.pack('<I', len(data)))
# 添加数据
frame.extend(data)
# 计算并添加校验和
checksum = self.calc_checksum(frame)
frame.append(checksum)
return bytes(frame)
def build_cmd_request(self, cmd_id, params=b''):
"""构建命令请求帧"""
data = bytearray()
data.append(cmd_id)
data.extend(params)
return self.build_frame(FrameType.CMD_REQUEST, bytes(data))
def parse_byte(self, byte):
"""
解析单个字节
返回：None=需要更多数据，dict=帧解析完成
"""
if self.state == ParseState.MAGIC1:
if byte == self.MAGIC_MSB:
self.state = ParseState.MAGIC2
# 否则继续等待
elif self.state == ParseState.MAGIC2:
if byte == self.MAGIC_LSB:
self.state = ParseState.TYPE
else:
# 魔数错误，重新开始
self.reset()
if byte == self.MAGIC_MSB:
self.state = ParseState.MAGIC2
elif self.state == ParseState.TYPE:
self.frame_type = byte
self.state = ParseState.LENGTH
self.length_buf = bytearray()
elif self.state == ParseState.LENGTH:
self.length_buf.append(byte)
if len(self.length_buf) >= 4:
# 小端序解析长度
self.data_length = struct.unpack('<I', self.length_buf)[0]
self.data = bytearray()
if self.data_length > 0:
self.state = ParseState.DATA
else:
self.state = ParseState.CHECKSUM
elif self.state == ParseState.DATA:
self.data.append(byte)
if len(self.data) >= self.data_length:
self.state = ParseState.CHECKSUM
elif self.state == ParseState.CHECKSUM:
self.checksum = byte
# 音频帧（Type=0x03）的校验位固定为0x00，不进行校验
if self.frame_type == FrameType.AUDIO_DATA:
# 音频帧不验证校验和，固定填充0x00
pass
else:
# 其他帧验证校验和
frame_for_check = bytearray()
frame_for_check.append(self.MAGIC_MSB)
frame_for_check.append(self.MAGIC_LSB)
frame_for_check.append(self.frame_type)
frame_for_check.extend(struct.pack('<I', self.data_length))
frame_for_check.extend(self.data)
calc_checksum = self.calc_checksum(frame_for_check)
if calc_checksum != self.checksum:
# 校验失败
print(f"[警告] 校验和错误: 期望0x{calc_checksum:02X}, 收到0x{self.checksum:02X}")
self.reset()
return None
# 帧解析完成
frame = {
'type': self.frame_type,
'data': bytes(self.data)
}
# 重置状态以准备下一帧
self.reset()
return frame
return None
class AudioRecorder:
"""音频录音器"""
def __init__(self, port, baudrate, output_file, timeout=1):
self.port = port
self.baudrate = baudrate
self.output_file = output_file
self.timeout = timeout
self.serial = None
self.protocol = CDCProtocol()
self.audio_file = None
self.md5_ctx = hashlib.md5()
self.byte_count = 0
self.seq_num = 0
self.device_md5 = None
self.running = False
self.recording = False
self.rx_thread = None
self.lock = threading.Lock()
def connect(self):
"""连接串口"""
try:
self.serial = serial.Serial(
port=self.port,
baudrate=self.baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=self.timeout
)
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 串口已打开: {self.port}")
print(f"波特率: {self.baudrate}")
print(f"输出文件: {self.output_file}")
print("-" * 50)
# 清空串口缓存
self.serial.reset_input_buffer()
self.serial.reset_output_buffer()
print("串口缓存已清空")
return True
except serial.SerialException as e:
print(f"串口错误: {e}", file=sys.stderr)
return False
def disconnect(self):
"""断开串口"""
if self.serial and self.serial.is_open:
self.serial.close()
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 串口已关闭")
def send_command(self, cmd_id, params=b''):
"""发送命令"""
if not self.serial or not self.serial.is_open:
return False
try:
frame = self.protocol.build_cmd_request(cmd_id, params)
self.serial.write(frame)
self.serial.flush()
cmd_name = {
CommandID.START_RECORD: "开始录音",
CommandID.STOP_RECORD: "停止录音",
CommandID.QUERY_STATUS: "查询状态"
}.get(cmd_id, f"未知命令(0x{cmd_id:02X})")
print(f"[命令] 发送: {cmd_name}")
return True
except Exception as e:
print(f"[错误] 发送命令失败: {e}")
return False
def handle_cmd_response(self, data):
"""处理命令响应"""
if len(data) < 2:
print("[警告] 命令响应数据过短")
return
cmd_id = data[0]
status = data[1]
response_data = data[2:] if len(data) > 2 else b''
cmd_name = {
CommandID.START_RECORD: "开始录音",
CommandID.STOP_RECORD: "停止录音",
CommandID.QUERY_STATUS: "查询状态"
}.get(cmd_id, f"0x{cmd_id:02X}")
status_name = {
StatusCode.OK: "成功",
StatusCode.ERROR: "失败",
StatusCode.BUSY: "忙碌",
StatusCode.UNSUPPORTED: "不支持"
}.get(status, f"0x{status:02X}")
print(f"[响应] {cmd_name}: {status_name}")
# 如果是状态查询响应，解析详细信息
if cmd_id == CommandID.QUERY_STATUS and status == StatusCode.OK and len(response_data) >= 9:
state = response_data[0]
total_bytes = struct.unpack('<I', response_data[1:5])[0]
seq_num = struct.unpack('<I', response_data[5:9])[0]
state_name = "正在录音" if state == 0x01 else "空闲"
print(f" 状态: {state_name}")
print(f" 已传输: {total_bytes} 字节")
print(f" 序列号: {seq_num}")
def handle_audio_data(self, data):
"""处理音频数据"""
if len(data) < 4:
print("[警告] 音频数据过短")
return
# 解析序列号
seq_num = struct.unpack('<I', data[0:4])[0]
pcm_data = data[4:]
# 检查序列号
if seq_num != self.seq_num:
print(f"\n[警告] 序列号跳变: 期望{self.seq_num}, 收到{seq_num}")
self.seq_num = seq_num
# 写入文件
if self.audio_file:
self.audio_file.write(pcm_data)
self.audio_file.flush()
# 更新MD5
self.md5_ctx.update(pcm_data)
# 更新统计
self.byte_count += len(pcm_data)
self.seq_num += 1
# 打印进度
print(f"\r已接收: {self.byte_count} 字节 (序列号: {seq_num})", end='', flush=True)
def handle_md5_data(self, data):
"""处理MD5数据"""
if len(data) != 16:
print(f"\n[警告] MD5数据长度错误: {len(data)}")
return
self.device_md5 = data.hex()
print(f"\n[MD5] 设备端: {self.device_md5}")
def handle_frame(self, frame):
"""处理接收到的帧"""
frame_type = frame['type']
data = frame['data']
if frame_type == FrameType.CMD_RESPONSE:
self.handle_cmd_response(data)
elif frame_type == FrameType.AUDIO_DATA:
self.handle_audio_data(data)
elif frame_type == FrameType.MD5_DATA:
self.handle_md5_data(data)
else:
print(f"[警告] 未知帧类型: 0x{frame_type:02X}")
def receive_thread_func(self):
"""接收线程函数"""
parser = CDCProtocol()
while self.running:
try:
if self.serial and self.serial.is_open and self.serial.in_waiting > 0:
data = self.serial.read(self.serial.in_waiting)
# 逐字节解析
for byte in data:
frame = parser.parse_byte(byte)
if frame:
with self.lock:
self.handle_frame(frame)
else:
time.sleep(0.01)
except Exception as e:
print(f"\n[错误] 接收线程异常: {e}")
break
def start_recording(self):
"""开始录音"""
# 打开输出文件
try:
self.audio_file = open(self.output_file, 'wb')
except IOError as e:
print(f"文件错误: {e}", file=sys.stderr)
return False
# 重置统计
self.byte_count = 0
self.seq_num = 0
self.md5_ctx = hashlib.md5()
self.device_md5 = None
# 启动接收线程
self.running = True
self.rx_thread = threading.Thread(target=self.receive_thread_func, daemon=True)
self.rx_thread.start()
# 发送开始录音命令
print("-" * 50)
if not self.send_command(CommandID.START_RECORD):
self.running = False
if self.audio_file:
self.audio_file.close()
return False
# 等待响应
time.sleep(0.5)
self.recording = True
print("开始接收音频数据... (按 Ctrl+C 停止)")
print("-" * 50)
return True
def stop_recording(self):
"""停止录音"""
if not self.recording:
return
print("\n\n停止录音...")
print("-" * 50)
# 发送停止录音命令
self.send_command(CommandID.STOP_RECORD)
# 等待剩余数据和MD5
print("等待设备发送MD5...")
time.sleep(2)
# 停止接收线程
self.recording = False
self.running = False
if self.rx_thread:
self.rx_thread.join(timeout=3)
# 关闭文件
if self.audio_file:
self.audio_file.close()
self.audio_file = None
# 计算本地MD5
local_md5 = self.md5_ctx.hexdigest()
print("-" * 50)
print(f"录音完成!")
print(f"总字节数: {self.byte_count}")
print(f"总帧数: {self.seq_num}")
print(f"PC端MD5: {local_md5}")
if self.device_md5:
print(f"设备端MD5: {self.device_md5}")
if local_md5 == self.device_md5:
print("MD5校验: 通过 ✓")
else:
print("MD5校验: 失败 ✗")
print("[警告] 数据可能在传输过程中损坏!")
else:
print("设备端MD5: 未收到")
print("-" * 50)
def run(self):
"""运行录音"""
if not self.connect():
return False
try:
if not self.start_recording():
return False
# 等待用户中断
while self.recording:
time.sleep(0.1)
except KeyboardInterrupt:
print("\n\n用户中断")
finally:
self.stop_recording()
self.disconnect()
return True
def main():
parser = argparse.ArgumentParser(
description='USB CDC串口录音工具 - 使用自定义协议',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''
使用示例:
# Linux
%(prog)s -p /dev/ttyACM0 -b 115200 -o output.pcm
# Windows
%(prog)s -p COM3 -b 115200 -o output.pcm
# 指定超时时间
%(prog)s -p /dev/ttyACM0 -b 921600 -o data.pcm -t 2
协议说明:
本工具使用自定义二进制协议进行通信，详见 docs/usb_cdc_protocol.md
命令流程:
1. PC发送"开始录音"命令
2. 设备响应确认
3. 设备持续发送音频数据帧（带序列号）
4. PC发送"停止录音"命令
5. 设备发送MD5校验值
6. PC对比MD5验证数据完整性
'''
)
parser.add_argument(
'-p', '--port',
required=True,
help='串口号 (Linux: /dev/ttyACM0, Windows: COM3)'
)
parser.add_argument(
'-b', '--baudrate',
type=int,
required=True,
help='波特率 (如: 9600, 115200, 921600)'
)
parser.add_argument(
'-o', '--output',
required=True,
help='输出文件路径 (.pcm)'
)
parser.add_argument(
'-t', '--timeout',
type=float,
default=1,
help='串口读取超时时间(秒), 默认: 1'
)
args = parser.parse_args()
# 检查输出目录是否存在
output_dir = os.path.dirname(args.output)
if output_dir and not os.path.exists(output_dir):
print(f"创建输出目录: {output_dir}")
os.makedirs(output_dir, exist_ok=True)
# 创建录音器并运行
recorder = AudioRecorder(args.port, args.baudrate, args.output, args.timeout)
success = recorder.run()
sys.exit(0 if success else 1)
if __name__ == '__main__':
main()

902
tools/audio/serial_capture_gui.py Executable file
View File

@@ -0,0 +1,902 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
USB CDC串口录音工具 - GUI版本
支持实时波形显示和录音控制
"""
import sys
import os
import serial
import time
import hashlib
import struct
import threading
import queue
from datetime import datetime
from enum import IntEnum
import numpy as np
# GUI相关
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from matplotlib.figure import Figure
import matplotlib.animation as animation
class FrameType(IntEnum):
"""帧类型定义"""
CMD_REQUEST = 0x01
CMD_RESPONSE = 0x02
AUDIO_DATA = 0x03
MD5_DATA = 0x04
class CommandID(IntEnum):
"""命令码定义"""
START_RECORD = 0x01
STOP_RECORD = 0x02
QUERY_STATUS = 0x03
class StatusCode(IntEnum):
"""状态码定义"""
OK = 0x00
ERROR = 0x01
BUSY = 0x02
UNSUPPORTED = 0x03
class ParseState(IntEnum):
"""解析状态机"""
MAGIC1 = 0
MAGIC2 = 1
TYPE = 2
LENGTH = 3
DATA = 4
CHECKSUM = 5
class CDCProtocol:
"""CDC协议处理类"""
MAGIC_MSB = 0xAA
MAGIC_LSB = 0x55
HEADER_SIZE = 7
MAX_CMD_DATA_SIZE = 64
MAX_AUDIO_DATA_SIZE = 65536
MAX_DATA_SIZE = MAX_AUDIO_DATA_SIZE
def __init__(self):
self.reset()
def reset(self):
"""重置解析器状态"""
self.state = ParseState.MAGIC1
self.frame_type = 0
self.data_length = 0
self.data = bytearray()
self.length_buf = bytearray()
self.checksum = 0
@staticmethod
def calc_checksum(data):
"""计算XOR校验和"""
checksum = 0
for byte in data:
checksum ^= byte
return checksum
def build_frame(self, frame_type, data=b''):
"""构建协议帧"""
frame = bytearray()
frame.append(self.MAGIC_MSB)
frame.append(self.MAGIC_LSB)
frame.append(frame_type)
frame.extend(struct.pack('<I', len(data)))
frame.extend(data)
checksum = self.calc_checksum(frame)
frame.append(checksum)
return bytes(frame)
def build_cmd_request(self, cmd_id, params=b''):
"""构建命令请求帧"""
data = bytearray()
data.append(cmd_id)
data.extend(params)
return self.build_frame(FrameType.CMD_REQUEST, bytes(data))
def parse_byte(self, byte):
"""解析单个字节"""
if self.state == ParseState.MAGIC1:
if byte == self.MAGIC_MSB:
self.state = ParseState.MAGIC2
elif self.state == ParseState.MAGIC2:
if byte == self.MAGIC_LSB:
self.state = ParseState.TYPE
else:
self.reset()
if byte == self.MAGIC_MSB:
self.state = ParseState.MAGIC2
elif self.state == ParseState.TYPE:
self.frame_type = byte
self.state = ParseState.LENGTH
self.length_buf = bytearray()
elif self.state == ParseState.LENGTH:
self.length_buf.append(byte)
if len(self.length_buf) >= 4:
self.data_length = struct.unpack('<I', self.length_buf)[0]
if self.data_length > self.MAX_DATA_SIZE:
self.reset()
return None
self.data = bytearray()
if self.data_length > 0:
self.state = ParseState.DATA
else:
self.state = ParseState.CHECKSUM
elif self.state == ParseState.DATA:
self.data.append(byte)
if len(self.data) >= self.data_length:
self.state = ParseState.CHECKSUM
elif self.state == ParseState.CHECKSUM:
self.checksum = byte
# 音频帧不验证校验和
if self.frame_type == FrameType.AUDIO_DATA:
pass
else:
# 其他帧验证校验和
frame_for_check = bytearray()
frame_for_check.append(self.MAGIC_MSB)
frame_for_check.append(self.MAGIC_LSB)
frame_for_check.append(self.frame_type)
frame_for_check.extend(struct.pack('<I', self.data_length))
frame_for_check.extend(self.data)
calc_checksum = self.calc_checksum(frame_for_check)
if calc_checksum != self.checksum:
self.reset()
return None
# 帧解析完成
frame = {
'type': self.frame_type,
'data': bytes(self.data)
}
self.reset()
return frame
return None
class AudioRecorderGUI:
"""带GUI的音频录音器"""
def __init__(self, root):
self.root = root
self.root.title("USB CDC 录音工具")
self.root.geometry("900x700")
# 录音状态
self.recording = False
self.serial = None
self.rx_thread = None
self.running = False
self.audio_file = None
self.byte_count = 0
self.seq_num = 0
self.md5_ctx = None
self.device_md5 = None
self.is_saving_file = False # 记录当前是否在保存文件
# 音频数据队列（用于波形显示）
self.audio_queue = queue.Queue(maxsize=100)
self.waveform_data = {} # 字典，key为通道号，value为numpy数组
self.max_waveform_samples = 16000 # 显示1秒的数据（假设16kHz采样率）
self.x_axis_max = 16000 # X轴显示的最大样本数（可通过滚轮调节）
self.x_axis_auto_fit = False # 是否自动适应显示所有数据
self.x_view_position = 0 # X轴视图起始位置（用于滚动条）
self.auto_scroll = True # 是否自动滚动到最新数据
self.updating_scrollbar = False # 标志：正在程序更新滚动条（防止触发回调）
# 波形线条（多通道）
self.waveform_lines = []
self.channel_colors = ['b', 'r', 'g', 'orange', 'purple', 'cyan', 'magenta', 'brown']
# Y轴范围控制（每个通道独立）
self.y_limits = {} # {channel: (ymin, ymax)}
# 协议解析器
self.protocol = CDCProtocol()
self.lock = threading.Lock()
# 创建GUI
self.create_widgets()
def create_widgets(self):
"""创建GUI组件"""
# === 顶部：配置区域 ===
config_frame = ttk.LabelFrame(self.root, text="配置", padding=10)
config_frame.pack(fill=tk.X, padx=10, pady=5)
# 串口选择
ttk.Label(config_frame, text="串口:").grid(row=0, column=0, sticky=tk.W, padx=5)
self.port_var = tk.StringVar(value="/dev/ttyACM0")
port_entry = ttk.Entry(config_frame, textvariable=self.port_var, width=20)
port_entry.grid(row=0, column=1, padx=5)
# 波特率
ttk.Label(config_frame, text="波特率:").grid(row=0, column=2, sticky=tk.W, padx=5)
self.baud_var = tk.StringVar(value="921600")
baud_combo = ttk.Combobox(config_frame, textvariable=self.baud_var, width=15,
values=["115200", "460800", "921600", "1500000"])
baud_combo.grid(row=0, column=3, padx=5)
# 采样率
ttk.Label(config_frame, text="采样率:").grid(row=0, column=4, sticky=tk.W, padx=5)
self.sample_rate_var = tk.StringVar(value="16000")
sample_rate_combo = ttk.Combobox(config_frame, textvariable=self.sample_rate_var, width=10,
values=["8000", "16000", "32000", "48000"],
state="readonly")
sample_rate_combo.grid(row=0, column=5, padx=5)
# 位深
ttk.Label(config_frame, text="位深:").grid(row=0, column=6, sticky=tk.W, padx=5)
self.bit_depth_var = tk.StringVar(value="16")
bit_depth_combo = ttk.Combobox(config_frame, textvariable=self.bit_depth_var, width=8,
values=["16"],
state="readonly")
bit_depth_combo.grid(row=0, column=7, padx=5)
# 通道数
ttk.Label(config_frame, text="通道数:").grid(row=0, column=8, sticky=tk.W, padx=5)
self.channels_var = tk.StringVar(value="1")
channels_combo = ttk.Combobox(config_frame, textvariable=self.channels_var, width=8,
values=["1", "2", "3", "4", "5", "6", "8"],
state="readonly")
channels_combo.grid(row=0, column=9, padx=5)
channels_combo.bind("<<ComboboxSelected>>", self.on_channels_changed)
# 是否保存文件选项和输出文件（同一行）
self.save_file_var = tk.BooleanVar(value=True)
save_check = ttk.Checkbutton(config_frame, text="保存文件:",
variable=self.save_file_var,
command=self.toggle_file_controls)
save_check.grid(row=1, column=0, sticky=tk.W, padx=5, pady=5)
self.output_var = tk.StringVar(value="output.pcm")
self.output_entry = ttk.Entry(config_frame, textvariable=self.output_var, width=35)
self.output_entry.grid(row=1, column=1, columnspan=2, padx=5, pady=5, sticky=tk.W)
self.browse_btn = ttk.Button(config_frame, text="浏览...", command=self.browse_output_file)
self.browse_btn.grid(row=1, column=3, padx=5, pady=5)
# === 中间：控制按钮区域 ===
control_frame = ttk.Frame(self.root, padding=10)
control_frame.pack(fill=tk.X, padx=10, pady=5)
self.start_btn = ttk.Button(control_frame, text="开始录音", command=self.start_recording,
state=tk.NORMAL, width=15)
self.start_btn.pack(side=tk.LEFT, padx=5)
self.stop_btn = ttk.Button(control_frame, text="停止录音", command=self.stop_recording,
state=tk.DISABLED, width=15)
self.stop_btn.pack(side=tk.LEFT, padx=5)
# === 波形显示区域 ===
waveform_frame = ttk.LabelFrame(self.root, text="实时波形", padding=10)
waveform_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
# 创建matplotlib图形（初始为空，稍后根据通道数创建子图）
self.fig = Figure(figsize=(10, 6), dpi=100)
self.axes = [] # 多个子图的列表
# 将matplotlib嵌入到tkinter
self.canvas = FigureCanvasTkAgg(self.fig, master=waveform_frame)
self.canvas.draw()
self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)
# 添加横向滚动条
scrollbar_frame = ttk.Frame(waveform_frame)
scrollbar_frame.pack(fill=tk.X, pady=(5, 0))
ttk.Label(scrollbar_frame, text="时间轴:").pack(side=tk.LEFT, padx=5)
self.x_scrollbar = ttk.Scale(scrollbar_frame, from_=0, to=100, orient=tk.HORIZONTAL,
command=self.on_x_scroll)
self.x_scrollbar.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)
self.auto_scroll_var = tk.BooleanVar(value=True)
self.auto_scroll_check = ttk.Checkbutton(scrollbar_frame, text="自动滚动",
variable=self.auto_scroll_var,
command=self.toggle_auto_scroll)
self.auto_scroll_check.pack(side=tk.LEFT, padx=5)
# 初始化波形线（多通道）- 必须在canvas创建后
self.init_waveform_lines()
# 绑定鼠标滚轮事件用于Y轴范围调节
self.canvas.mpl_connect('scroll_event', self.on_scroll)
# === 底部：状态信息区域 ===
status_frame = ttk.LabelFrame(self.root, text="状态信息", padding=10)
status_frame.pack(fill=tk.X, padx=10, pady=5)
# 状态标签
self.status_label = ttk.Label(status_frame, text="就绪", foreground="green", font=("Arial", 10, "bold"))
self.status_label.grid(row=0, column=0, columnspan=4, sticky=tk.W, pady=5)
# 统计信息
ttk.Label(status_frame, text="已接收:").grid(row=1, column=0, sticky=tk.W)
self.bytes_label = ttk.Label(status_frame, text="0 字节", width=20)
self.bytes_label.grid(row=1, column=1, sticky=tk.W)
ttk.Label(status_frame, text="序列号:").grid(row=1, column=2, sticky=tk.W, padx=20)
self.seq_label = ttk.Label(status_frame, text="0", width=15)
self.seq_label.grid(row=1, column=3, sticky=tk.W)
ttk.Label(status_frame, text="PC MD5:").grid(row=2, column=0, sticky=tk.W)
self.pc_md5_label = ttk.Label(status_frame, text="-", width=40)
self.pc_md5_label.grid(row=2, column=1, columnspan=3, sticky=tk.W)
ttk.Label(status_frame, text="设备 MD5:").grid(row=3, column=0, sticky=tk.W)
self.device_md5_label = ttk.Label(status_frame, text="-", width=40)
self.device_md5_label.grid(row=3, column=1, columnspan=3, sticky=tk.W)
# 启动波形更新动画
self.ani = animation.FuncAnimation(self.fig, self.update_waveform,
interval=100, blit=False)
def init_waveform_lines(self):
"""初始化波形线（每个通道独立子图）"""
# 清除所有旧的子图
self.fig.clear()
self.axes = []
self.waveform_lines = []
# 获取通道数
num_channels = int(self.channels_var.get())
# 为每个通道创建独立的子图
for ch in range(num_channels):
# 创建子图 (行数, 列数, 索引)
ax = self.fig.add_subplot(num_channels, 1, ch + 1)
# 设置子图属性 - 将通道标签放在左边作为Y轴标签
ax.set_ylabel(f'CH{ch}', fontsize=10, fontweight='bold', rotation=0,
labelpad=20, va='center')
ax.grid(True, alpha=0.3)
ax.set_ylim(-32768, 32767)
# 设置Y轴刻度标签字体大小
ax.tick_params(axis='y', labelsize=7)
ax.tick_params(axis='x', labelsize=7)
# 只在最后一个子图显示X轴标签
if ch == num_channels - 1:
ax.set_xlabel('样本', fontsize=8)
else:
ax.set_xticklabels([])
# 为该子图创建波形线
color = self.channel_colors[ch % len(self.channel_colors)]
line, = ax.plot([], [], color=color, linewidth=0.8)
self.axes.append(ax)
self.waveform_lines.append(line)
# 初始化该通道的Y轴范围
self.y_limits[ch] = (-32768, 32767)
# 调整子图间距 - 更紧凑的布局
self.fig.subplots_adjust(left=0.08, right=0.98, top=0.98, bottom=0.05, hspace=0.15)
# 初始化波形数据字典
self.waveform_data = {ch: np.array([]) for ch in range(num_channels)}
self.canvas.draw()
def on_channels_changed(self, event=None):
"""通道数改变时的回调"""
if not self.recording:
self.init_waveform_lines()
# 更新最大样本数（根据采样率）
sample_rate = int(self.sample_rate_var.get())
self.max_waveform_samples = sample_rate # 显示1秒数据
self.x_axis_max = sample_rate # 同步更新X轴最大值
def on_x_scroll(self, value):
"""X轴滚动条回调"""
# 当用户拖动滚动条时，禁用自动滚动
if self.auto_scroll_var.get():
self.auto_scroll_var.set(False)
self.auto_scroll = False
# 更新视图位置（value 是0-100的百分比）
self.x_view_position = float(value)
def toggle_auto_scroll(self):
"""切换自动滚动模式"""
self.auto_scroll = self.auto_scroll_var.get()
if self.auto_scroll:
# 启用自动滚动时，将视图移到最右侧
self.x_scrollbar.set(100)
self.x_view_position = 100
def on_scroll(self, event):
"""鼠标滚轮事件 - 调整Y轴或X轴范围"""
if event.y is None:
return
target_channel = None
target_ax = None
# 先尝试检测鼠标是否在某个子图内
if event.inaxes is not None:
for ch, ax in enumerate(self.axes):
if event.inaxes == ax:
target_channel = ch
target_ax = ax
break
# 如果不在子图内，检测鼠标Y坐标在哪个子图的垂直范围内（包括Y轴区域和间隙）
if target_channel is None:
for ch, ax in enumerate(self.axes):
bbox = ax.get_window_extent()
# 扩大检测范围：上下各扩展10像素
if bbox.y0 - 10 <= event.y <= bbox.y1 + 10:
target_channel = ch
target_ax = ax
break
# 如果找到了目标通道
if target_channel is not None and target_ax is not None:
bbox = target_ax.get_window_extent()
# 判断鼠标在Y轴左侧还是右侧（波形区域）
# Y轴区域大约在子图左边界后60像素内
y_axis_right_edge = bbox.x0 + 60
if event.x is not None and event.x > y_axis_right_edge:
# 鼠标在波形区域（Y轴右侧）- 调节X轴范围（所有通道同步）
zoom_factor = 0.9 if event.button == 'up' else 1.1
if self.x_axis_auto_fit and event.button == 'up':
# 当前是自动适应模式，向上滚动开始缩放
sample_rate = int(self.sample_rate_var.get())
self.x_axis_max = int(sample_rate * zoom_factor)
self.x_axis_auto_fit = False
else:
# 计算新的X轴最大值
new_x_max = self.x_axis_max * zoom_factor
# 限制范围：最小800个样本，最大100000个样本
new_x_max = max(800, min(100000, int(new_x_max)))
# 如果缩小到最小值，切换到自动适应模式
if new_x_max <= 800 and event.button == 'down':
self.x_axis_auto_fit = True
else:
self.x_axis_auto_fit = False
self.x_axis_max = new_x_max
# 不在这里直接设置xlim，让update_waveform处理
self.canvas.draw_idle()
else:
# 鼠标在Y轴区域 - 调节该通道的Y轴范围
# 获取当前Y轴范围
ymin, ymax = self.y_limits.get(target_channel, (-32768, 32767))
y_range = ymax - ymin
y_center = (ymin + ymax) / 2
# 滚轮向上：缩小范围（放大显示）
# 滚轮向下：扩大范围（缩小显示）
zoom_factor = 0.9 if event.button == 'up' else 1.1
# 计算新的Y轴范围
new_range = y_range * zoom_factor
# 限制最小和最大范围
new_range = max(100, min(65536, new_range))
# 计算新的Y轴上下限（保持中心不变）
new_ymin = y_center - new_range / 2
new_ymax = y_center + new_range / 2
# 更新Y轴范围
self.y_limits[target_channel] = (new_ymin, new_ymax)
target_ax.set_ylim(new_ymin, new_ymax)
self.canvas.draw_idle()
def parse_interleaved_pcm(self, pcm_data):
"""
解析交叉排列的PCM数据
输入格式: ch0 ch1 ch2 ... ch0 ch1 ch2 ...
返回: {0: [ch0_samples], 1: [ch1_samples], ...}
"""
num_channels = int(self.channels_var.get())
bit_depth = int(self.bit_depth_var.get())
# 将bytes转换为int16数组
if bit_depth == 16:
samples = np.frombuffer(pcm_data, dtype=np.int16)
else:
# 未来可以扩展支持其他位深
samples = np.frombuffer(pcm_data, dtype=np.int16)
# 分离各通道数据
channel_data = {}
for ch in range(num_channels):
channel_data[ch] = samples[ch::num_channels]
return channel_data
def browse_output_file(self):
"""浏览输出文件"""
filename = filedialog.asksaveasfilename(
defaultextension=".pcm",
filetypes=[("PCM files", "*.pcm"), ("WAV files", "*.wav"), ("All files", "*.*")]
)
if filename:
self.output_var.set(filename)
def toggle_file_controls(self):
"""切换文件控件的启用状态"""
if self.save_file_var.get():
# 启用文件控件
self.output_entry.config(state=tk.NORMAL)
self.browse_btn.config(state=tk.NORMAL)
else:
# 禁用文件控件
self.output_entry.config(state=tk.DISABLED)
self.browse_btn.config(state=tk.DISABLED)
def update_status(self, message, color="black"):
"""更新状态信息"""
self.status_label.config(text=message, foreground=color)
def update_statistics(self):
"""更新统计信息"""
self.bytes_label.config(text=f"{self.byte_count} 字节")
self.seq_label.config(text=str(self.seq_num))
def update_waveform(self, frame):
"""更新波形显示（多通道支持）"""
try:
# 从队列获取新的音频数据
while not self.audio_queue.empty():
try:
new_data = self.audio_queue.get_nowait()
# 解析交叉排列的多通道数据
channel_data = self.parse_interleaved_pcm(new_data)
# 更新每个通道的数据
for ch, samples in channel_data.items():
if ch in self.waveform_data:
# 拼接新数据
self.waveform_data[ch] = np.concatenate([self.waveform_data[ch], samples])
# 限制最大缓冲区为1000000个样本
max_buffer_size = 1000000
if len(self.waveform_data[ch]) > max_buffer_size:
self.waveform_data[ch] = self.waveform_data[ch][-max_buffer_size:]
except queue.Empty:
break
# 更新每条波形线和对应子图的X轴
num_channels = int(self.channels_var.get())
for ch in range(num_channels):
if ch < len(self.waveform_lines) and ch < len(self.axes) and ch in self.waveform_data:
data = self.waveform_data[ch]
if len(data) > 0:
# 更新波形线数据
self.waveform_lines[ch].set_data(range(len(data)), data)
# 为每个子图设置X轴范围
if self.x_axis_auto_fit:
# 自动适应模式：显示所有数据
self.axes[ch].set_xlim(0, max(len(data), 1000))
else:
# 固定窗口模式 - 根据滚动条位置显示
if len(data) <= self.x_axis_max:
# 数据不足，从0开始显示
self.axes[ch].set_xlim(0, max(self.x_axis_max, 1000))
else:
# 根据滚动条位置计算显示范围
if self.auto_scroll:
# 自动滚动：显示最新数据
x_end = len(data)
x_start = max(0, x_end - self.x_axis_max)
# 同步更新滚动条位置（不触发回调）
self.x_scrollbar.set(100)
else:
# 手动滚动：根据滚动条位置显示
total_range = len(data) - self.x_axis_max
x_start = int((self.x_view_position / 100.0) * total_range)
x_start = max(0, min(x_start, total_range))
x_end = x_start + self.x_axis_max
self.axes[ch].set_xlim(x_start, x_end)
# 应用该通道的Y轴范围
if ch in self.y_limits:
self.axes[ch].set_ylim(self.y_limits[ch])
except Exception as e:
print(f"波形更新错误: {e}")
return tuple(self.waveform_lines) if self.waveform_lines else ()
def send_command(self, cmd_id, params=b''):
"""发送命令"""
if not self.serial or not self.serial.is_open:
return False
frame = self.protocol.build_cmd_request(cmd_id, params)
try:
self.serial.write(frame)
return True
except Exception as e:
self.update_status(f"发送命令失败: {e}", "red")
return False
def handle_cmd_response(self, data):
"""处理命令响应"""
if len(data) < 2:
return
cmd_id = data[0]
status = data[1]
status_str = {0: "成功", 1: "失败", 2: "忙碌", 3: "不支持"}.get(status, "未知")
cmd_str = {1: "开始录音", 2: "停止录音", 3: "查询状态"}.get(cmd_id, f"命令{cmd_id}")
message = f"{cmd_str}: {status_str}"
color = "green" if status == 0 else "red"
self.update_status(message, color)
def handle_audio_data(self, data):
"""处理音频数据"""
if len(data) < 4:
return
# 解析序列号
seq_num = struct.unpack('<I', data[0:4])[0]
pcm_data = data[4:]
# 检查序列号
if seq_num != self.seq_num:
self.seq_num = seq_num
# 写入文件（仅当启用保存时）
if self.audio_file:
self.audio_file.write(pcm_data)
self.audio_file.flush()
# 更新MD5（无论是否保存文件都需要计算）
self.md5_ctx.update(pcm_data)
# 更新统计
self.byte_count += len(pcm_data)
self.seq_num += 1
# 更新GUI
self.root.after(0, self.update_statistics)
# 添加到波形队列（无论是否保存文件都显示波形）
try:
self.audio_queue.put_nowait(pcm_data)
except queue.Full:
# 队列满了，丢弃旧数据
try:
self.audio_queue.get_nowait()
self.audio_queue.put_nowait(pcm_data)
except:
pass
def handle_md5_data(self, data):
"""处理MD5数据"""
if len(data) != 16:
return
self.device_md5 = data.hex()
self.device_md5_label.config(text=self.device_md5)
# 比较MD5
pc_md5 = self.md5_ctx.hexdigest()
self.pc_md5_label.config(text=pc_md5)
# 显示结果
if pc_md5 == self.device_md5:
if self.is_saving_file:
self.update_status("录音完成，MD5校验通过 ✓", "green")
else:
self.update_status("录音完成（未保存文件），MD5校验通过 ✓", "green")
else:
if self.is_saving_file:
self.update_status("录音完成，MD5校验失败 ✗", "red")
else:
self.update_status("录音完成（未保存文件），MD5校验失败 ✗", "red")
def handle_frame(self, frame):
"""处理接收到的帧"""
frame_type = frame['type']
data = frame['data']
if frame_type == FrameType.CMD_RESPONSE:
self.handle_cmd_response(data)
elif frame_type == FrameType.AUDIO_DATA:
self.handle_audio_data(data)
elif frame_type == FrameType.MD5_DATA:
self.handle_md5_data(data)
def receive_thread_func(self):
"""接收线程函数"""
parser = CDCProtocol()
while self.running:
try:
if self.serial and self.serial.is_open and self.serial.in_waiting > 0:
data = self.serial.read(self.serial.in_waiting)
# 逐字节解析
for byte in data:
frame = parser.parse_byte(byte)
if frame:
with self.lock:
self.handle_frame(frame)
else:
time.sleep(0.01)
except Exception as e:
print(f"接收线程异常: {e}")
break
def start_recording(self):
"""开始录音"""
port = self.port_var.get()
baudrate = int(self.baud_var.get())
save_file = self.save_file_var.get()
output_file = self.output_var.get()
# 打开串口
try:
self.serial = serial.Serial(port, baudrate, timeout=1)
self.update_status(f"串口已打开: {port}", "green")
except Exception as e:
messagebox.showerror("错误", f"无法打开串口: {e}")
return
# 打开输出文件（仅当启用保存时）
if save_file:
try:
self.audio_file = open(output_file, 'wb')
except IOError as e:
messagebox.showerror("错误", f"无法创建文件: {e}")
self.serial.close()
return
else:
self.audio_file = None # 不保存文件
# 重置统计
self.byte_count = 0
self.seq_num = 0
self.md5_ctx = hashlib.md5()
self.device_md5 = None
# 重置波形数据为字典格式（多通道）
num_channels = int(self.channels_var.get())
self.waveform_data = {ch: np.array([]) for ch in range(num_channels)}
# 重置X轴最大值和自动适应模式
sample_rate = int(self.sample_rate_var.get())
self.x_axis_max = sample_rate
self.x_axis_auto_fit = False # 开始录音时使用固定窗口模式
# 重置滚动条状态
self.x_view_position = 0
self.auto_scroll = True
self.auto_scroll_var.set(True)
self.x_scrollbar.set(100)
self.pc_md5_label.config(text="-")
self.device_md5_label.config(text="-")
self.update_statistics()
# 启动接收线程
self.running = True
self.rx_thread = threading.Thread(target=self.receive_thread_func, daemon=True)
self.rx_thread.start()
# 发送开始录音命令
time.sleep(0.1)
if not self.send_command(CommandID.START_RECORD):
self.update_status("发送开始录音命令失败", "red")
self.running = False
if self.audio_file:
self.audio_file.close()
if self.serial:
self.serial.close()
return
self.recording = True
self.is_saving_file = save_file # 记录是否保存文件
self.start_btn.config(state=tk.DISABLED)
self.stop_btn.config(state=tk.NORMAL)
# 更新状态信息
if save_file:
self.update_status(f"正在录音并保存到: {output_file}", "blue")
else:
self.update_status("正在录音（仅显示波形，不保存文件）", "blue")
def stop_recording(self):
"""停止录音"""
if not self.recording:
return
self.recording = False
self.update_status("正在停止录音...", "orange")
# 发送停止录音命令
if self.send_command(CommandID.STOP_RECORD):
# 等待MD5数据
time.sleep(1)
# 停止接收线程
self.running = False
if self.rx_thread:
self.rx_thread.join(timeout=2)
# 关闭文件和串口
if self.audio_file:
self.audio_file.close()
self.audio_file = None
if self.serial and self.serial.is_open:
self.serial.close()
self.serial = None
self.start_btn.config(state=tk.NORMAL)
self.stop_btn.config(state=tk.DISABLED)
# 如果没有收到MD5数据，显示停止信息
if not self.device_md5:
if self.is_saving_file:
self.update_status("录音已停止", "orange")
else:
self.update_status("录音已停止（未保存文件）", "orange")
def on_closing(self):
"""窗口关闭事件"""
if self.recording:
if messagebox.askokcancel("退出", "录音正在进行，确定要退出吗？"):
self.stop_recording()
time.sleep(0.5)
self.root.destroy()
else:
self.root.destroy()
def main():
"""主函数"""
root = tk.Tk()
app = AudioRecorderGUI(root)
root.protocol("WM_DELETE_WINDOW", app.on_closing)
root.mainloop()
if __name__ == "__main__":
main()

313
tools/audio/usb_cdc_protocol.md Normal file
View File

@@ -0,0 +1,313 @@
# USB CDC 串口通信协议规范
## 概述
本协议定义了上位机（PC）与设备之间通过USB CDC进行音频数据传输和命令控制的通信格式。协议采用二进制帧格式，支持命令、音频数据和MD5校验等功能。
## 协议版本
- 版本号：1.0
- 更新日期：2026-01-07
## 帧格式
### 基本帧结构
```
+--------+--------+--------+--------+--------+--------+
| Magic | Type | Length | Data | Check |
| 2 Bytes| 1 Byte | 4 Bytes | N Bytes| 1 Byte |
+--------+--------+--------+--------+--------+--------+
```
**注意：** 音频数据帧（Type=0x03）的Check字段固定填充为0x00，不进行实际校验计算，保持协议格式一致性。
### 字段说明
| 字段 | 长度 | 说明 |
|------|------|------|
| Magic | 2 字节 | 魔数，固定为 `0xAA55`（大端序），用于帧同步 |
| Type | 1 字节 | 数据类型，见下表 |
| Length | 4 字节 | 数据长度（小端序），仅指Data字段的长度<br>**命令/响应/MD5帧**: ≤64字节<br>|
| Data | N 字节 | 实际数据内容，长度由Length字段指定 |
| Check | 1 字节 | 校验和，对整个帧（除Check字段外）进行XOR校验<br>**音频帧固定填充0x00** |
**数据长度限制说明：**
- 命令请求/响应/MD5帧的Data字段最大64字节
- 这种区分设计可节省非音频帧的内存占用
### 数据类型（Type）
| 类型值 | 名称 | 方向 | 说明 |
|--------|------|------|------|
| 0x01 | CMD_REQUEST | PC → 设备 | 命令请求 |
| 0x02 | CMD_RESPONSE | 设备 → PC | 命令响应 |
| 0x03 | AUDIO_DATA | 设备 → PC | 音频数据 |
| 0x04 | MD5_DATA | 设备 → PC | MD5校验值 |
## 命令定义
### 命令码
| 命令码 | 名称 | 说明 |
|--------|------|------|
| 0x01 | CMD_START_RECORD | 开始录音 |
| 0x02 | CMD_STOP_RECORD | 停止录音 |
| 0x03 | CMD_QUERY_STATUS | 查询状态 |
### 命令请求格式（CMD_REQUEST）
**Data字段格式：**
```
+--------+--------+
| CmdID | Params |
| 1 Byte | N Bytes|
+--------+--------+
```
- **CmdID**: 命令码
- **Params**: 命令参数（可选，长度根据具体命令而定）
### 命令响应格式（CMD_RESPONSE）
**Data字段格式：**
```
+--------+--------+--------+
| CmdID | Status | Data |
| 1 Byte | 1 Byte | N Bytes|
+--------+--------+--------+
```
- **CmdID**: 对应的命令码
- **Status**: 状态码（0x00=成功，0x01=失败，0x02=忙碌，0x03=不支持）
- **Data**: 响应数据（可选）
### 状态码定义
| 状态码 | 名称 | 说明 |
|--------|------|------|
| 0x00 | STATUS_OK | 命令执行成功 |
| 0x01 | STATUS_ERROR | 命令执行失败 |
| 0x02 | STATUS_BUSY | 设备忙碌 |
| 0x03 | STATUS_UNSUPPORTED | 不支持的命令 |
## 数据格式
### 音频数据（AUDIO_DATA）
**Data字段格式：**
```
+--------+--------+
| SeqNum | PCM |
| 4 Bytes| N Bytes|
+--------+--------+
```
- **SeqNum**: 序列号（小端序），用于检测丢包，从0开始递增
- **PCM**: 原始PCM音频数据
**注意：** 音频数据帧保留Check校验字节，但固定填充为0x00，不进行实际校验计算。这样保持协议格式一致性，数据完整性由最终的MD5校验保证。
### MD5数据（MD5_DATA）
**Data字段格式：**
```
+--------+
| MD5 |
|16 Bytes|
+--------+
```
- **MD5**: 音频数据的MD5哈希值（128位）
## 通信流程
### 开始录音流程
```
PC 设备
| |
|---- CMD_START_RECORD ---->|
| | 开始录音
|<--- CMD_RESPONSE(OK) -----|
| |
|<--- AUDIO_DATA(SeqNum=0)--|
|<--- AUDIO_DATA(SeqNum=1)--|
|<--- AUDIO_DATA(SeqNum=2)--|
| ... |
```
### 停止录音流程
```
PC 设备
| |
|---- CMD_STOP_RECORD ----->|
| | 停止录音
|<--- AUDIO_DATA(last) -----|
|<--- MD5_DATA -------------|
|<--- CMD_RESPONSE(OK) -----|
| |
```
### 查询状态流程
```
PC 设备
| |
|---- CMD_QUERY_STATUS ---->|
| |
|<--- CMD_RESPONSE ---------|
| (Status + Data) |
| |
```
**查询状态响应Data格式：**
```
+--------+--------+--------+--------+
| State | TotalBytes | SeqNum |
| 1 Byte | 4 Bytes | 4 Bytes|
+--------+--------+--------+--------+
```
- **State**: 当前状态（0x00=空闲，0x01=正在录音）
- **TotalBytes**: 已传输的总字节数（小端序）
- **SeqNum**: 当前序列号（小端序）
## 校验算法
### XOR校验
对除Check字段外的所有字节进行XOR运算：
```c
uint8_t calculate_checksum(uint8_t *data, uint32_t len) {
uint8_t checksum = 0;
for (uint32_t i = 0; i < len; i++) {
checksum ^= data[i];
}
return checksum;
}
```
### MD5校验
使用标准MD5算法对所有音频数据进行校验：
1. 设备端：对发送的所有PCM数据进行MD5计算
2. 停止录音后，设备发送MD5_DATA帧
3. PC端：对接收的所有PCM数据进行MD5计算并与设备端发送的MD5对比
## 示例
### 示例1：开始录音命令
**PC发送：**
```
AA 55 01 01 00 00 00 01 FA
```
解析：
- Magic: `AA 55`
- Type: `01` (CMD_REQUEST)
- Length: `01 00 00 00` (1字节，小端序)
- Data: `01` (CMD_START_RECORD)
- Check: `FA` (XOR校验)
**设备响应：**
```
AA 55 02 02 00 00 00 01 00 F8
```
解析：
- Magic: `AA 55`
- Type: `02` (CMD_RESPONSE)
- Length: `02 00 00 00` (2字节)
- Data: `01 00` (CmdID=0x01, Status=0x00成功)
- Check: `F8`
### 示例2：音频数据
**设备发送：**
```
AA 55 03 08 00 00 00 00 00 00 00 XX XX XX XX
```
解析：
- Magic: `AA 55`
- Type: `03` (AUDIO_DATA)
- Length: `08 00 00 00` (8字节)
- Data: `00 00 00 00 XX XX XX XX` (SeqNum=0, PCM数据4字节)
- Check: **无** (音频帧不包含校验字节)
### 示例3：MD5数据
**设备发送：**
```
AA 55 04 10 00 00 00 [16字节MD5] XX
```
解析：
- Magic: `AA 55`
- Type: `04` (MD5_DATA)
- Length: `10 00 00 00` (16字节)
- Data: 16字节MD5值
- Check: `XX`
## 错误处理
### 帧错误
1. **魔数错误**：丢弃当前字节，继续搜索魔数
2. **校验错误**：丢弃整个帧，请求重传（可选）
3. **长度异常**：丢弃整个帧
### 超时处理
- **命令超时**：PC端发送命令后，2秒内未收到响应视为超时
- **数据超时**：录音过程中，5秒内未收到数据视为断连
### 序列号检查
- PC端应检查音频数据的SeqNum是否连续
- 如果发现跳号，说明有数据丢失
## 实现注意事项
### 设备端（C语言）
1. **内存管理**：使用环形缓冲区管理数据帧
2. **多线程**：命令处理和数据发送应分开不同任务
3. **优先级**：命令处理优先级应高于音频数据发送
4. **MD5计算**：使用增量方式计算MD5，避免一次性读取所有数据
### PC端（Python）
1. **异步接收**：使用独立线程接收数据
2. **帧解析**：实现状态机进行帧解析，处理粘包/拆包
3. **超时机制**：实现命令超时和数据超时检测
4. **MD5对比**：实时计算接收数据的MD5，录音结束后对比
## 扩展性
协议预留了扩展空间：
1. **命令扩展**：可以添加新的命令码（0x04-0xFF）
2. **数据类型扩展**：可以添加新的数据类型（0x05-0xFF）
3. **参数扩展**：每个命令的Params字段可以自定义格式
## 参考实现
- 设备端实现：[src/middleware/usb/app_usb_cdc.c](../src/middleware/usb/app_usb_cdc.c)
- PC端实现：[tools/audio/serial_capture.py](../tools/audio/serial_capture.py)
- 协议库：[src/middleware/usb/cdc_protocol.c](../src/middleware/usb/cdc_protocol.c)
## 版本历史
| 版本 | 日期 | 说明 |
|------|------|------|
| 1.0 | 2026-01-07 | 初始版本，定义基本协议格式和命令 |
## 联系方式
如有问题或建议，请联系开发团队。

450
tools/audio/usb_cdc_quick_start.md Normal file
View File

@@ -0,0 +1,450 @@
# USB CDC 录音系统快速开始指南
## 概述
本文档介绍如何使用新的USB CDC串口录音系统，该系统使用自定义二进制协议进行通信，支持：
- 命令控制（开始/停止录音、查询状态）
- 音频数据传输（带序列号防丢包）
- MD5校验（确保数据完整性）
- 无需依赖adb shell命令
## 系统架构
```
┌──────────────┐ ┌──────────────┐
│ PC端 │ │ 设备端 │
│ │ │ │
│ Python脚本 │ ◄──USB CDC协议──► │ C代码 │
│ │ │ │
│ - 发送命令 │ │ - 接收命令 │
│ - 接收数据 │ │ - 发送音频 │
│ - 计算MD5 │ │ - 计算MD5 │
└──────────────┘ └──────────────┘
```
## 文件说明
### 协议相关文件
| 文件 | 说明 |
|------|------|
| [docs/usb_cdc_protocol.md](usb_cdc_protocol.md) | 完整的协议规范文档 |
| [src/middleware/usb/cdc_protocol.h](../src/middleware/usb/cdc_protocol.h) | 协议头文件（C） |
| [src/middleware/usb/cdc_protocol.c](../src/middleware/usb/cdc_protocol.c) | 协议实现（C） |
| [src/middleware/usb/app_usb_cdc.h](../src/middleware/usb/app_usb_cdc.h) | USB CDC接口头文件 |
| [src/middleware/usb/app_usb_cdc.c](../src/middleware/usb/app_usb_cdc.c) | USB CDC实现文件 |
| [tools/audio/serial_capture.py](../tools/audio/serial_capture.py) | PC端录音脚本（Python） |
## PC端使用说明
### 1. 环境准备
确保已安装Python 3和pyserial库：
```bash
# 安装依赖
pip3 install pyserial
# 或者使用系统包管理器
# Ubuntu/Debian
sudo apt-get install python3-serial
# macOS
brew install python3
pip3 install pyserial
```
### 2. 查找串口设备
**Linux:**
```bash
# 查看所有串口设备
ls /dev/ttyACM* /dev/ttyUSB*
# 通常USB CDC设备显示为 /dev/ttyACM0
```
**Windows:**
```powershell
# 在设备管理器中查看 "端口(COM和LPT)"
# 通常显示为 COM3, COM4 等
```
**macOS:**
```bash
# 查看所有串口设备
ls /dev/tty.*
# USB CDC设备通常显示为 /dev/tty.usbmodemXXXX
```
### 3. 录音操作
**基本用法：**
```bash
# Linux
python3 tools/audio/serial_capture.py -p /dev/ttyACM0 -b 115200 -o output.pcm
# Windows
python tools/audio/serial_capture.py -p COM3 -b 115200 -o output.pcm
# macOS
python3 tools/audio/serial_capture.py -p /dev/tty.usbmodem14201 -b 115200 -o output.pcm
```
**完整示例：**
```bash
# 指定所有参数
python3 tools/audio/serial_capture.py \
--port /dev/ttyACM0 \
--baudrate 115200 \
--output recordings/audio_$(date +%Y%m%d_%H%M%S).pcm \
--timeout 2
```
### 4. 录音流程
脚本运行后会自动执行以下步骤：
1. **连接串口**
```
[2026-01-07 10:30:15] 串口已打开: /dev/ttyACM0
波特率: 115200
输出文件: output.pcm
--------------------------------------------------
串口缓存已清空
```
2. **发送开始命令**
```
--------------------------------------------------
[命令] 发送: 开始录音
[响应] 开始录音: 成功
开始接收音频数据... (按 Ctrl+C 停止)
--------------------------------------------------
```
3. **接收音频数据**
```
已接收: 102400 字节 (序列号: 12)
```
4. **停止录音（按Ctrl+C）**
```
用户中断
停止录音...
--------------------------------------------------
[命令] 发送: 停止录音
等待设备发送MD5...
[MD5] 设备端: a1b2c3d4e5f6789012345678abcdef01
[响应] 停止录音: 成功
```
5. **验证MD5**
```
--------------------------------------------------
录音完成!
总字节数: 102400
总帧数: 13
PC端MD5: a1b2c3d4e5f6789012345678abcdef01
设备端MD5: a1b2c3d4e5f6789012345678abcdef01
MD5校验: 通过 ✓
--------------------------------------------------
```
### 5. 播放录音
录音文件为原始PCM格式，可以使用以下工具播放：
**使用ffplay（推荐）：**
```bash
# 16kHz, 16位, 单声道
ffplay -f s16le -ar 16000 -ac 1 output.pcm
# 48kHz, 16位, 双声道
ffplay -f s16le -ar 48000 -ac 2 output.pcm
```
**转换为WAV格式：**
```bash
# 使用ffmpeg转换
ffmpeg -f s16le -ar 16000 -ac 1 -i output.pcm output.wav
```
**使用sox播放：**
```bash
play -t raw -r 16000 -e signed -b 16 -c 1 output.pcm
```
## 设备端使用说明
### 1. 添加源文件到编译系统
确保以下文件被包含在编译中：
```cmake
# CMakeLists.txt
target_sources(your_target PRIVATE
src/middleware/usb/cdc_protocol.c
src/middleware/usb/app_usb_cdc.c
)
target_include_directories(your_target PRIVATE
src/middleware/usb
)
```
### 2. 初始化USB CDC
在应用初始化代码中调用：
```c
#include "app_usb_cdc.h"
// 在main或init函数中
int ret = app_usb_cdc_init();
if (ret != 0) {
LISA_LOGE("APP", "Failed to initialize USB CDC");
return ret;
}
```
### 3. 发送音频数据
在音频采集回调中发送数据：
```c
#include "app_usb_cdc.h"
void audio_callback(uint8_t *pcm_data, uint32_t len)
{
// 只有在录音状态才发送
if (app_usb_cdc_is_recording()) {
// 数据会被自动封装成协议帧并发送
app_usb_cdc_audio_write(pcm_data, len);
}
}
```
### 4. 配置MD5计算（可选）
如果系统中有mbedtls，可以在编译配置中启用：
```kconfig
# Kconfig或.config
CONFIG_MBEDTLS=y
```
如果没有mbedtls，代码会使用占位函数（返回全0），建议集成一个MD5实现。
### 5. 查看状态
```c
// 检查是否正在录音
bool recording = app_usb_cdc_is_recording();
// 获取已发送的字节数
uint32_t bytes = app_usb_cdc_get_total_bytes();
// 获取当前序列号
uint32_t seq = app_usb_cdc_get_seq_num();
LISA_LOGI("APP", "Recording=%d, Bytes=%lu, Seq=%lu",
recording, bytes, seq);
```
## 协议详解
### 基本帧格式
```
+--------+--------+--------+--------+--------+--------+
| Magic | Type | Length | Data | Check |
| 2 Bytes| 1 Byte | 4 Bytes | N Bytes| 1 Byte |
+--------+--------+--------+--------+--------+--------+
0xAA55 0xXX Little Endian ... XOR
```
### 数据类型
- `0x01` - 命令请求（PC → 设备）
- `0x02` - 命令响应（设备 → PC）
- `0x03` - 音频数据（设备 → PC）
- `0x04` - MD5数据（设备 → PC）
### 命令码
- `0x01` - 开始录音
- `0x02` - 停止录音
- `0x03` - 查询状态
详细协议说明请参考：[usb_cdc_protocol.md](usb_cdc_protocol.md)
## 常见问题
### Q1: 串口打开失败
**错误：** `串口错误: [Errno 13] Permission denied: '/dev/ttyACM0'`
**解决方法：**
```bash
# 临时解决：添加当前用户到dialout组
sudo usermod -a -G dialout $USER
# 注销并重新登录
# 或者临时使用sudo（不推荐）
sudo python3 tools/audio/serial_capture.py -p /dev/ttyACM0 -b 115200 -o output.pcm
```
### Q2: MD5校验失败
**可能原因：**
1. 设备端没有正确实现MD5计算（使用的是占位函数）
2. 传输过程中有数据丢失
3. 序列号跳变
**解决方法：**
1. 在设备端集成完整的MD5库（mbedtls或其他）
2. 检查是否有序列号跳变的警告
3. 尝试降低波特率或使用更短的传输距离
### Q3: 数据接收不完整
**症状：** 序列号跳变警告
**解决方法：**
1. 检查USB连接是否稳定
2. 降低音频数据发送速率
3. 增大设备端的队列长度（`CDC_AUDIO_QUEUE_LENGTH`）
4. 检查PC端CPU占用率
### Q4: 设备无响应
**检查步骤：**
```bash
# 1. 确认设备已连接
ls -l /dev/ttyACM*
# 2. 检查设备是否可读写
cat /dev/ttyACM0
# 应该能看到一些数据（如果设备在发送）
# 3. 尝试手动发送命令（用于调试）
echo -ne '\xAA\x55\x01\x01\x00\x00\x00\x01\xFA' > /dev/ttyACM0
```
### Q5: 如何调试协议
**PC端调试：**
在脚本中添加调试输出：
```python
# 在parse_byte函数中添加
print(f"State: {self.state}, Byte: 0x{byte:02X}")
```
**设备端调试：**
```c
// 启用协议日志
#define CDC_PROTO_DEBUG 1
// 在关键位置添加日志
LISA_LOGI(TAG, "Frame type: 0x%02X, length: %lu",
frame->header.type, frame->header.length);
```
## 性能参数
### 典型配置
| 参数 | 值 | 说明 |
|------|-----|------|
| 波特率 | 115200 | 标准USB CDC速度 |
| 音频帧大小 | 1024-4096字节 | 推荐2048字节 |
| 队列长度 | 50 | 可根据内存调整 |
| 最大数据长度 | 4096字节 | 单帧最大负载 |
### 理论吞吐量
```
波特率: 115200 bps
有效数据率: ~11.5 KB/s (考虑协议开销)
单帧开销: 8字节（帧头+校验）+ 4字节（序列号）= 12字节
数据负载: 2048字节
帧效率: 2048 / (2048 + 12) ≈ 99.4%
支持音频格式:
- 16kHz, 16bit, Mono: 32 KB/s (需要更高波特率)
- 8kHz, 16bit, Mono: 16 KB/s (可支持)
```
**建议：** 对于高采样率音频（如16kHz以上），考虑使用更高的波特率或USB bulk传输模式。
## 扩展开发
### 添加新命令
1. **在协议头文件中定义命令码：**
```c
// cdc_protocol.h
typedef enum {
CDC_CMD_START_RECORD = 0x01,
CDC_CMD_STOP_RECORD = 0x02,
CDC_CMD_QUERY_STATUS = 0x03,
CDC_CMD_SET_GAIN = 0x04, // 新命令：设置增益
} cdc_cmd_id_t;
```
2. **在设备端实现命令处理：**
```c
// app_usb_cdc.c
case CDC_CMD_SET_GAIN:
if (frame->header.length >= 2) {
uint8_t gain = cmd->params[0];
// 处理增益设置
set_audio_gain(gain);
response_len = cdc_proto_build_cmd_response(...);
}
break;
```
3. **在PC端脚本中添加命令：**
```python
# serial_capture.py
class CommandID(IntEnum):
START_RECORD = 0x01
STOP_RECORD = 0x02
QUERY_STATUS = 0x03
SET_GAIN = 0x04 # 新命令
# 添加发送方法
def set_gain(self, gain_value):
params = bytes([gain_value])
return self.send_command(CommandID.SET_GAIN, params)
```
## 参考资料
- [USB CDC协议规范](usb_cdc_protocol.md) - 完整的协议文档
- [TinyUSB文档](https://docs.tinyusb.org/) - USB协议栈参考
- [Python pyserial文档](https://pyserial.readthedocs.io/) - 串口通信库
## 技术支持
如有问题或建议，请：
1. 查阅协议文档：`docs/usb_cdc_protocol.md`
2. 查看代码注释：头文件中有详细的API说明
3. 联系开发团队
---
**版本：** 1.0
**更新日期：** 2026-01-07
**作者：** LISTENAI开发团队

BIN
tools/cskburn/cskburn Normal file
View File

Binary file not shown.

BIN
tools/cskburn/cskburn.exe Normal file
View File

Binary file not shown.

175
tools/make_lpk.py Executable file
View File

@@ -0,0 +1,175 @@
#!/usr/bin/env python3
"""Build manifest.json and package all referenced images into an .lpk archive."""
import argparse
import hashlib
import json
from pathlib import Path
import re
import zipfile
from typing import Dict, List
MANIFEST_VERSION = 2
CHIP = "arcs"
def compute_md5(path: Path) -> str:
digest = hashlib.md5()
with path.open("rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
digest.update(chunk)
return digest.hexdigest()
def resolve_vars(file_spec: str, vars_map: Dict[str, Path]) -> str:
expanded = file_spec
for key, value in vars_map.items():
expanded = expanded.replace(f"${{{key}}}", str(value))
unresolved = re.findall(r"\$\{([^}]+)\}", expanded)
if unresolved:
missing = ", ".join(sorted(set(unresolved)))
raise ValueError(f"missing --var for: {missing}")
return expanded
def resolve_image_path(
table_path: Path, file_spec: str, vars_map: Dict[str, Path]
) -> Path:
expanded = resolve_vars(file_spec, vars_map)
path = Path(expanded)
if not path.is_absolute():
path = (table_path.parent / path).resolve()
return path
def load_partition_table(table_path: Path) -> List[Dict]:
with table_path.open("r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, list):
raise ValueError("partition table must be a JSON array")
return data
def collect_images(
table_path: Path, vars_map: Dict[str, Path], no_boot: bool = False
) -> List[Dict]:
images = load_partition_table(table_path)
collected = []
for image in images:
if no_boot and image.get("name") == "boot":
continue
file_spec = image.get("file")
if not file_spec:
raise ValueError("partition entry missing 'file' field")
source_path = resolve_image_path(table_path, file_spec, vars_map)
if not source_path.is_file():
raise FileNotFoundError(f"missing image file: {source_path}")
collected.append(
{
"name": image.get("name"),
"addr": image.get("addr"),
"file": source_path.name,
"source_path": source_path,
"md5": compute_md5(source_path),
}
)
return collected
def build_manifest(images: List[Dict]) -> Dict:
manifest = {"manifest": MANIFEST_VERSION, "chip": CHIP, "images": []}
for image in images:
manifest["images"].append(
{
"name": image.get("name"),
"addr": image.get("addr"),
"file": f"./{image.get('file')}",
"md5": image.get("md5"),
}
)
return manifest
def package_lpk(manifest: Dict, images: List[Dict], output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
manifest_text = json.dumps(manifest, indent=4) + "\n"
with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("manifest.json", manifest_text)
for image in images:
source_path = image["source_path"]
print(
f"packing image: {image['name']} ({image['addr']}) md5={image['md5']} -> {image['file']}"
)
zf.write(source_path, arcname=image["file"])
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Generate manifest.json and package an .lpk archive."
)
parser.add_argument(
"output",
type=Path,
help="Output .lpk path",
)
parser.add_argument(
"--partition-table",
type=Path,
required=True,
help="Partition table JSON file; file paths are relative to the JSON by default.",
)
parser.add_argument(
"--no-boot",
action="store_true",
help="Exclude the boot image from the package.",
)
parser.add_argument(
"--var",
action="append",
default=[],
metavar="KEY=VALUE",
help="Variable mapping for ${KEY}; VALUE is relative to cwd by default.",
)
return parser.parse_args()
def parse_vars(var_args: List[str]) -> Dict[str, Path]:
vars_map: Dict[str, Path] = {}
for item in var_args:
if "=" not in item:
raise ValueError(f"invalid --var '{item}', expected KEY=VALUE")
key, value = item.split("=", 1)
key = key.strip()
value = value.strip()
if not key:
raise ValueError(f"invalid --var '{item}', empty KEY")
value_path = Path(value)
if not value_path.is_absolute():
value_path = (Path.cwd() / value_path).resolve()
vars_map[key] = value_path
return vars_map
def main() -> None:
args = parse_args()
table_path = args.partition_table
vars_map = parse_vars(args.var)
images = collect_images(table_path, vars_map, no_boot=args.no_boot)
manifest = build_manifest(images)
package_lpk(manifest, images, args.output)
print(f"wrote lpk to {args.output}")
if __name__ == "__main__":
main()

478
tools/romfs.py Normal file
View File

@@ -0,0 +1,478 @@
#!/usr/bin/env python3
import argparse
import os
import stat
import struct
import sys
MAGIC = b"-rom1fs-"
SUPERBLOCK_SIZE = 16
ALIGN = 16
FS_PAD = 1024
TYPE_HARDLINK = 0
TYPE_DIR = 1
TYPE_FILE = 2
TYPE_SYMLINK = 3
TYPE_BLOCK = 4
TYPE_CHAR = 5
TYPE_SOCKET = 6
TYPE_FIFO = 7
def align_up(value, align):
return (value + (align - 1)) & ~(align - 1)
def be32(value):
return struct.pack(">I", value & 0xFFFFFFFF)
def sum_be32_words(data):
if len(data) % 4 != 0:
raise ValueError("data length must be multiple of 4 bytes")
total = 0
for i in range(0, len(data), 4):
total = (total + struct.unpack(">I", data[i:i + 4])[0]) & 0xFFFFFFFF
return total
def checksum_block(data):
total = sum_be32_words(data)
return (-total) & 0xFFFFFFFF
class Node:
def __init__(self, name, node_type, st=None):
self.name = name
self.node_type = node_type
self.st = st
self.exec_bit = False
self.data = b""
self.children = []
self.link_target = None
self.spec_info = 0
self.size = 0
self.offset = None
def name_bytes(self):
return self.name.encode("utf-8")
def header_meta_len(self):
name_len = len(self.name_bytes()) + 1
return SUPERBLOCK_SIZE + align_up(name_len, ALIGN)
def data_len(self):
if self.node_type in (TYPE_FILE, TYPE_SYMLINK):
return self.size
return 0
def total_len(self):
return self.header_meta_len() + align_up(self.data_len(), ALIGN)
def build_tree(root_dir):
inode_map = {}
def build_node(path, name, parent):
st = os.lstat(path)
mode = st.st_mode
exec_bit = bool(mode & 0o111)
if stat.S_ISDIR(mode):
node = Node(name, TYPE_DIR, st=st)
node.exec_bit = exec_bit
entries = []
with os.scandir(path) as it:
for entry in it:
entries.append(entry.name)
entries.sort()
# "." and ".." are hardlinks to current and parent directories.
dot = Node(".", TYPE_HARDLINK)
dot.link_target = node
dotdot = Node("..", TYPE_HARDLINK)
dotdot.link_target = parent if parent is not None else node
node.children.extend([dot, dotdot])
for entry_name in entries:
child_path = os.path.join(path, entry_name)
node.children.append(build_node(child_path, entry_name, node))
return node
if stat.S_ISLNK(mode):
node = Node(name, TYPE_SYMLINK, st=st)
node.exec_bit = exec_bit
target = os.readlink(path)
node.data = target.encode("utf-8")
node.size = len(node.data)
return node
if stat.S_ISREG(mode):
inode_key = (st.st_dev, st.st_ino)
if inode_key in inode_map:
link = Node(name, TYPE_HARDLINK, st=st)
link.link_target = inode_map[inode_key]
return link
node = Node(name, TYPE_FILE, st=st)
node.exec_bit = exec_bit
with open(path, "rb") as f:
node.data = f.read()
node.size = len(node.data)
inode_map[inode_key] = node
return node
if stat.S_ISBLK(mode):
node = Node(name, TYPE_BLOCK, st=st)
node.exec_bit = exec_bit
node.spec_info = (os.major(st.st_rdev) << 16) | os.minor(st.st_rdev)
return node
if stat.S_ISCHR(mode):
node = Node(name, TYPE_CHAR, st=st)
node.exec_bit = exec_bit
node.spec_info = (os.major(st.st_rdev) << 16) | os.minor(st.st_rdev)
return node
if stat.S_ISSOCK(mode):
node = Node(name, TYPE_SOCKET, st=st)
node.exec_bit = exec_bit
return node
if stat.S_ISFIFO(mode):
node = Node(name, TYPE_FIFO, st=st)
node.exec_bit = exec_bit
return node
raise ValueError(f"unsupported file type: {path}")
root = Node("", TYPE_DIR)
root.exec_bit = True
entries = []
with os.scandir(root_dir) as it:
for entry in it:
entries.append(entry.name)
entries.sort()
dot = Node(".", TYPE_HARDLINK)
dot.link_target = root
dotdot = Node("..", TYPE_HARDLINK)
dotdot.link_target = root
root.children.extend([dot, dotdot])
for entry_name in entries:
child_path = os.path.join(root_dir, entry_name)
root.children.append(build_node(child_path, entry_name, root))
return root
def assign_offsets(node, start_offset):
node.offset = start_offset
next_offset = start_offset + node.total_len()
if node.node_type == TYPE_DIR:
for child in node.children:
next_offset = assign_offsets(child, next_offset)
return next_offset
def write_node(node, image, next_offset):
if node.link_target is not None:
node.spec_info = node.link_target.offset
elif node.node_type == TYPE_DIR:
node.spec_info = node.children[0].offset if node.children else 0
type_bits = node.node_type & 0x7
mode_bits = type_bits | (0x8 if node.exec_bit else 0x0)
next_field = (next_offset or 0) | mode_bits
header = [
be32(next_field),
be32(node.spec_info),
be32(node.size),
be32(0),
]
name_bytes = node.name_bytes() + b"\x00"
name_padded = name_bytes + b"\x00" * (align_up(len(name_bytes), ALIGN) - len(name_bytes))
meta = b"".join(header) + name_padded
chksum = checksum_block(meta)
meta = meta[:12] + be32(chksum) + meta[16:]
image.extend(meta)
if node.data_len():
data = node.data
image.extend(data)
image.extend(b"\x00" * (align_up(len(data), ALIGN) - len(data)))
def serialize(root, volume_name):
volume_bytes = volume_name.encode("utf-8") + b"\x00"
volume_padded = volume_bytes + b"\x00" * (align_up(len(volume_bytes), ALIGN) - len(volume_bytes))
start_offset = SUPERBLOCK_SIZE + len(volume_padded)
end_offset = assign_offsets(root, start_offset)
full_size = align_up(end_offset, FS_PAD)
image = bytearray()
image.extend(MAGIC)
image.extend(be32(full_size))
image.extend(be32(0))
image.extend(volume_padded)
def emit_dir(node):
if node.node_type != TYPE_DIR:
return
for i, child in enumerate(node.children):
next_child = node.children[i + 1] if i + 1 < len(node.children) else None
next_offset = next_child.offset if next_child else 0
write_node(child, image, next_offset)
if child.node_type == TYPE_DIR:
emit_dir(child)
write_node(root, image, 0)
emit_dir(root)
if len(image) < full_size:
image.extend(b"\x00" * (full_size - len(image)))
checksum_len = min(512, full_size)
checksum_data = bytes(image[:checksum_len])
chksum = checksum_block(checksum_data)
image[12:16] = be32(chksum)
return image
def read_u32(data, offset):
if offset + 4 > len(data):
raise ValueError("unexpected end of image")
return struct.unpack(">I", data[offset:offset + 4])[0]
def parse_superblock(data):
if len(data) < SUPERBLOCK_SIZE:
raise ValueError("image too small")
if data[:8] != MAGIC:
raise ValueError("invalid romfs magic")
full_size = read_u32(data, 8)
if full_size == 0 or full_size > len(data):
raise ValueError("invalid romfs size")
if full_size % 4 != 0:
raise ValueError("invalid romfs size alignment")
checksum_len = min(512, full_size)
if checksum_len % 4 != 0:
raise ValueError("invalid checksum region size")
if sum_be32_words(data[:checksum_len]) != 0:
raise ValueError("romfs superblock checksum mismatch")
name_start = SUPERBLOCK_SIZE
name_end = data.find(b"\x00", name_start, full_size)
if name_end == -1:
raise ValueError("volume name not terminated")
volume_name = data[name_start:name_end].decode("utf-8", errors="replace")
name_padded_len = align_up((name_end - name_start) + 1, ALIGN)
root_offset = SUPERBLOCK_SIZE + name_padded_len
if root_offset % ALIGN != 0:
raise ValueError("root offset not aligned")
return full_size, volume_name, root_offset
def parse_header(data, offset, full_size):
if offset % ALIGN != 0:
raise ValueError(f"unaligned header at {offset}")
if offset + SUPERBLOCK_SIZE > full_size:
raise ValueError("header exceeds image size")
next_field = read_u32(data, offset)
spec_info = read_u32(data, offset + 4)
size = read_u32(data, offset + 8)
checksum = read_u32(data, offset + 12)
name_start = offset + SUPERBLOCK_SIZE
name_end = data.find(b"\x00", name_start, full_size)
if name_end == -1:
raise ValueError("unterminated name")
name = data[name_start:name_end].decode("utf-8", errors="replace")
name_padded_len = align_up((name_end - name_start) + 1, ALIGN)
meta_len = SUPERBLOCK_SIZE + name_padded_len
if offset + meta_len > full_size:
raise ValueError("header metadata exceeds image size")
meta = data[offset:offset + meta_len]
if sum_be32_words(meta) != 0:
raise ValueError(f"header checksum mismatch at {offset}")
mode_bits = next_field & 0xF
next_offset = next_field & ~0xF
node_type = mode_bits & 0x7
exec_bit = bool(mode_bits & 0x8)
data_offset = offset + meta_len
data_offset = offset + meta_len
if data_offset + size > full_size:
raise ValueError("file data exceeds image size")
return {
"offset": offset,
"next_offset": next_offset,
"spec_info": spec_info,
"size": size,
"checksum": checksum,
"name": name,
"node_type": node_type,
"exec_bit": exec_bit,
"meta_len": meta_len,
"data_offset": data_offset,
}
def traverse_directory(data, start_offset, full_size, base_path, entries, visited):
offset = start_offset
while offset:
if offset in visited:
raise ValueError(f"loop detected at {offset}")
visited.add(offset)
if offset >= full_size:
raise ValueError(f"header offset out of bounds: {offset}")
info = parse_header(data, offset, full_size)
name = info["name"]
node_type = info["node_type"]
entry_path = os.path.join(base_path, name) if name else base_path
entries.append((entry_path, info))
if node_type == TYPE_DIR:
child_offset = info["spec_info"]
if child_offset and (child_offset % ALIGN != 0 or child_offset >= full_size):
raise ValueError(f"invalid directory entry offset: {child_offset}")
if name not in (".", ".."):
traverse_directory(data, child_offset, full_size, entry_path, entries, visited)
next_offset = info["next_offset"]
if next_offset and (next_offset % ALIGN != 0 or next_offset >= full_size):
raise ValueError(f"invalid next offset: {next_offset}")
offset = next_offset
def list_image(image_path):
with open(image_path, "rb") as f:
data = f.read()
full_size, volume_name, root_offset = parse_superblock(data)
entries = []
visited = set()
root_info = parse_header(data, root_offset, full_size)
if root_info["node_type"] != TYPE_DIR:
raise ValueError("root node is not a directory")
entries.append(("/", root_info))
traverse_directory(data, root_info["spec_info"], full_size, "/", entries, visited)
print(f"Volume: {volume_name}")
for path, info in entries:
if os.path.basename(path) in (".", ".."):
continue
node_type = info["node_type"]
size = info["size"]
if node_type == TYPE_DIR:
kind = "dir"
elif node_type == TYPE_FILE:
kind = "file"
elif node_type == TYPE_SYMLINK:
kind = "symlink"
elif node_type == TYPE_HARDLINK:
kind = "hardlink"
elif node_type == TYPE_BLOCK:
kind = "block"
elif node_type == TYPE_CHAR:
kind = "char"
elif node_type == TYPE_SOCKET:
kind = "socket"
else:
kind = "fifo"
line = f"0x{info['offset']:08x} {kind:8} {size:10} {path}"
if node_type == TYPE_SYMLINK:
target = data[info["data_offset"]:info["data_offset"] + size].decode("utf-8", errors="replace")
line += f" -> {target}"
print(line)
def extract_image(image_path, output_dir):
with open(image_path, "rb") as f:
data = f.read()
full_size, volume_name, root_offset = parse_superblock(data)
os.makedirs(output_dir, exist_ok=True)
entries = []
visited = set()
root_info = parse_header(data, root_offset, full_size)
if root_info["node_type"] != TYPE_DIR:
raise ValueError("root node is not a directory")
traverse_directory(data, root_info["spec_info"], full_size, output_dir, entries, visited)
hardlinks = []
offset_to_path = {}
for path, info in entries:
node_type = info["node_type"]
name = os.path.basename(path)
if name in (".", ".."):
continue
if name == "" or os.path.sep in name or (os.path.altsep and os.path.altsep in name):
raise ValueError(f"invalid entry name: {name!r}")
offset_to_path[info["offset"]] = path
if node_type == TYPE_DIR:
os.makedirs(path, exist_ok=True)
elif node_type == TYPE_FILE:
os.makedirs(os.path.dirname(path), exist_ok=True)
start = info["data_offset"]
end = start + info["size"]
with open(path, "wb") as f:
f.write(data[start:end])
elif node_type == TYPE_SYMLINK:
os.makedirs(os.path.dirname(path), exist_ok=True)
start = info["data_offset"]
end = start + info["size"]
target = data[start:end].decode("utf-8", errors="replace")
os.symlink(target, path)
elif node_type == TYPE_HARDLINK:
hardlinks.append((path, info["spec_info"]))
elif node_type == TYPE_BLOCK:
os.makedirs(os.path.dirname(path), exist_ok=True)
os.mknod(path, stat.S_IFBLK | 0o600, info["spec_info"])
elif node_type == TYPE_CHAR:
os.makedirs(os.path.dirname(path), exist_ok=True)
os.mknod(path, stat.S_IFCHR | 0o600, info["spec_info"])
elif node_type == TYPE_FIFO:
os.makedirs(os.path.dirname(path), exist_ok=True)
os.mkfifo(path)
elif node_type == TYPE_SOCKET:
continue
else:
raise ValueError(f"unsupported node type {node_type}")
for path, target_offset in hardlinks:
name = os.path.basename(path)
if name in (".", ".."):
continue
target_path = offset_to_path.get(target_offset)
if target_path is None:
raise ValueError(f"hardlink target not found at {target_offset}")
os.makedirs(os.path.dirname(path), exist_ok=True)
os.link(target_path, path)
def pack_image(input_dir, image_path, volume_name):
root = build_tree(input_dir)
image = serialize(root, volume_name)
with open(image_path, "wb") as f:
f.write(image)
def main(argv=None):
parser = argparse.ArgumentParser(description="ROMFS pack/list/extract tool")
subparsers = parser.add_subparsers(dest="command", required=True)
pack_parser = subparsers.add_parser("pack", help="pack a directory into ROMFS image")
pack_parser.add_argument("-i", "--input", required=True, help="input directory")
pack_parser.add_argument("-o", "--output", required=True, help="output image path")
pack_parser.add_argument("-n", "--name", default="romfs", help="volume name")
list_parser = subparsers.add_parser("list", help="list contents of ROMFS image")
list_parser.add_argument("-i", "--input", required=True, help="input image path")
extract_parser = subparsers.add_parser("extract", help="extract ROMFS image")
extract_parser.add_argument("-i", "--input", required=True, help="input image path")
extract_parser.add_argument("-o", "--output", required=True, help="output directory")
args = parser.parse_args(argv)
if args.command == "pack":
pack_image(args.input, args.output, args.name)
elif args.command == "list":
list_image(args.input)
elif args.command == "extract":
extract_image(args.input, args.output)
else:
parser.error("unknown command")
if __name__ == "__main__":
main()