✨ feature: send vad and asr in opus.

justfile

@@ -1,3 +1,6 @@
+light:
+  sudo /home/pi/.local/share/../bin/uv run src/chatbot/tests/test_light.py
+
 setting:
   uv run streamlit run src/chatbot/webui.py
 

pyproject.toml

@@ -14,7 +14,6 @@ dependencies = [
     "openai>=1.82.0",
     "httpx[http2,socks]>=0.28.1",
     "dotenv>=0.9.9",
-    "rpi-gpio>=0.7.1",
     "aiohttp>=3.12.7",
     "aiofiles>=24.1.0",
     "websockets>=15.0.1",
@@ -25,6 +24,8 @@ dependencies = [
     "fastapi>=0.115.12",
     "uvicorn>=0.34.3",
     "soundfile>=0.13.1",
+    "rpi-ws281x>=5.0.0; platform_machine=='armv7l' or platform_machine=='aarch64'",
+    "rpi-gpio>=0.7.1; platform_machine=='armv7l' or platform_machine=='aarch64'",
 ]
 authors = [{ name = "MrXnneHang", email = "XnneHang@gmail.com" }]
 

src/chatbot/api/async_api.py

@@ -15,7 +15,7 @@
 from chatbot._dictionary import session_keys
 from chatbot.config_manager import ServiceSettings, load_settings_file
 from chatbot.console.logger import Logger
-from chatbot.tools.audio import file_to_wav, play_opus_file
+from chatbot.tools.audio import file_to_opus, file_to_wav, play_opus_file
 
 if TYPE_CHECKING:
     from chatbot._typing import VadResponse
@@ -204,3 +204,8 @@ async def async_get_vad_response(audio_path: Path) -> VadResponse:
 async def async_file_to_wav(input_path: Path, output_path: Path):
     loop = asyncio.get_event_loop()
     await loop.run_in_executor(None, file_to_wav, input_path, output_path)
+
+
+async def async_file_to_opus(input_path: Path, output_path: Path):
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, file_to_opus, input_path, output_path)

src/chatbot/chatter/voice_recorder.py

@@ -8,7 +8,7 @@
 import sounddevice as sd
 import soundfile as sf
 
-from chatbot.api.async_api import async_get_asr_response, async_get_vad_response
+from chatbot.api.async_api import async_file_to_opus, async_get_asr_response, async_get_vad_response
 from chatbot.chatter import CHANNELS, INITIAL_WAIT_TIME, MAX_RECORDING_TIME, RATE, SEGMENT_DURATION, SILENCE_THRESHOLD
 from chatbot.console.logger import Logger
 from chatbot.tools.timed_helper import get_time_tag_with_millis
@@ -41,11 +41,29 @@ def audio_callback(
             # 将音频帧添加到临时缓冲区
             self.audio_frames.append(indata.copy())
 
-    async def save_wav(self, audio_data: NDArray[np.float32], file_path: Path) -> Path:
-        """将音频数据保存为WAV文件"""
-        sf.write(file_path, audio_data, RATE, format="WAV", subtype="PCM_16")  # type: ignore[arg-type]
-        Logger.info(f"已保存WAV文件: {file_path}")
-        return file_path
+    async def save_audio(self, audio_data: NDArray[np.float32], file_path: Path) -> Path:
+        """将音频数据保存为Opus文件"""
+        # 先保存为临时WAV文件
+        temp_wav_path = file_path.with_suffix(".wav")
+        sf.write(temp_wav_path, audio_data, RATE, format="WAV", subtype="PCM_16")  # type: ignore
+        Logger.info(f"已保存临时WAV文件: {temp_wav_path}")
+
+        # 转换为Opus
+        opus_path = file_path.with_suffix(".opus")
+        await async_file_to_opus(temp_wav_path, opus_path)
+
+        # 确保文件写入完成
+        await asyncio.sleep(0.1)
+
+        # 验证文件是否存在且有内容
+        if not opus_path.exists() or opus_path.stat().st_size == 0:
+            raise RuntimeError(f"Opus文件创建失败或为空: {opus_path}")
+
+        # 删除临时WAV文件
+        temp_wav_path.unlink()
+
+        Logger.info(f"已保存Opus文件: {opus_path} (大小: {opus_path.stat().st_size} 字节)")
+        return opus_path
 
     async def process_segments(self) -> None:
         """按顺序处理音频片段进行VAD检测"""
@@ -59,9 +77,9 @@ async def process_segments(self) -> None:
                 continue
             # 获取最早的片段进行处理
             segment_data, segment_index = self.segments_to_process.pop(0)
-            segment_path = self.cache_dir / f"segment_{segment_index}_{get_time_tag_with_millis()}.wav"
+            segment_path = self.cache_dir / f"segment_{segment_index}_{get_time_tag_with_millis()}.opus"
             # 保存片段为WAV文件
-            await self.save_wav(segment_data, segment_path)
+            await self.save_audio(segment_data, segment_path)
             # 发送到VAD服务
             vad_result = await async_get_vad_response(segment_path)
             # 清理临时文件
@@ -75,6 +93,8 @@ async def process_segments(self) -> None:
             audio_length_ms = vad_result["audio_length"]
             # 获取时间戳数组
             timestamps = vad_result.get("timestamp", [])
+            if len(timestamps) == 0:
+                timestamps = [[0, 10]]  # 设置一个较小的默认值，避免后续计算出错
             if timestamps and len(timestamps) > 0:
                 # 找到最后一个语音段的结束时间
                 last_voice_activity_ms = timestamps[-1][-1] if timestamps[-1] else 0
@@ -187,14 +207,14 @@ async def save_and_process_full_audio(self, full_audio: NDArray[np.float32]) ->
             return None
         # 保存完整的WAV文件
         time_tag = get_time_tag_with_millis()
-        full_wav_path = self.cache_dir / f"full_audio_{time_tag}.wav"
-        await self.save_wav(full_audio, full_wav_path)
+        full_audio_path = self.cache_dir / f"full_audio_{time_tag}.opus"
+        await self.save_audio(full_audio, full_audio_path)
         # 发送到ASR服务
-        Logger.info(f"发送完整音频文件到ASR服务: {full_wav_path}")
-        response = await async_get_asr_response(full_wav_path)
+        Logger.info(f"发送完整音频文件到ASR服务: {full_audio_path}")
+        response = await async_get_asr_response(full_audio_path)
         # 清理临时文件
-        if full_wav_path.exists():
-            full_wav_path.unlink()
+        if full_audio_path.exists():
+            full_audio_path.unlink()
         # 处理ASR结果
         if response:
             Logger.info(f"ASR识别结果: {response.strip()}")

src/chatbot/tests/test_light.py

@@ -0,0 +1,143 @@
+# type: ignore
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+
+import RPi.GPIO as GPIO
+from rpi_ws281x import Color, PixelStrip
+
+# Reduce logging
+logging.basicConfig(level=logging.ERROR)
+
+# LED灯带参数
+LED_COUNT = 255  # LED灯带上的LED数量
+LED_PIN = 18  # GPIO引脚
+pin_to_monitor = 23  # 监控的引脚
+LED_FREQ_HZ = 800000  # LED信号频率
+LED_DMA = 10  # DMA通道
+LED_BRIGHTNESS = 10  # LED亮度（0-255）
+LED_INVERT = False  # 是否反转信号
+LED_CHANNEL = 0  # 通道
+
+# 设置 GPIO 引脚
+GPIO.setmode(GPIO.BCM)
+GPIO.setup(pin_to_monitor, GPIO.IN, pull_up_down=GPIO.PUD_UP)
+
+# 初始化LED灯带
+strip = PixelStrip(LED_COUNT, LED_PIN, LED_FREQ_HZ, LED_DMA, LED_INVERT, LED_BRIGHTNESS, LED_CHANNEL)
+strip.begin()
+current_brightness = 10
+
+# 全局状态变量（替代 streamlit session_state）
+app_state = {
+    "starttime": None,
+    "PIN": False,
+    "light_state": False,  # False表示降低亮度，True表示提高亮度
+    "led_on": False,
+}
+
+
+async def set_all_pixels(color):
+    """设置所有LED为指定颜色"""
+    for i in range(strip.numPixels()):
+        strip.setPixelColor(i, color)
+    strip.show()
+    await asyncio.sleep(0.1)
+
+
+async def set_brightness(value):
+    """设置灯带亮度并应用"""
+    global current_brightness
+    current_brightness = max(10, min(255, value))  # 确保亮度在10-255之间
+    strip.setBrightness(current_brightness)
+    strip.show()
+    print(f"Brightness set to: {current_brightness}/255")
+    await asyncio.sleep(0.1)
+
+
+async def monitor_io():
+    while True:
+        try:
+            # 当引脚为高电平时时间计数器触发
+            if GPIO.input(pin_to_monitor) == GPIO.HIGH:
+                if not app_state["PIN"]:
+                    app_state["PIN"] = True
+                    app_state["starttime"] = time.time()  # 计时器开始计时
+                    app_state["light_state"] = not app_state["light_state"]  # 转换长按模式
+                    print(f"Button pressed, light_state: {app_state['light_state']}")
+            else:
+                if app_state["PIN"]:
+                    app_state["PIN"] = False
+                    print("Button released")
+
+            # Add a small sleep to prevent CPU hogging
+            await asyncio.sleep(0.1)
+        except Exception as e:
+            print(f"Error in monitor_io: {e}")
+            await asyncio.sleep(0.1)
+
+
+async def light_set():
+    while True:
+        try:
+            # 当检测到持续的高电平时
+            if app_state["PIN"] and app_state["starttime"] is not None:
+                elapsed_time = time.time() - app_state["starttime"]
+                if elapsed_time > 1:
+                    if app_state["light_state"]:
+                        print("light up")
+                        await set_brightness(current_brightness + 10)
+                    else:
+                        print("light down")
+                        await set_brightness(current_brightness - 10)
+
+            else:
+                if app_state["starttime"] is not None:
+                    elapsed_time = time.time() - app_state["starttime"]
+                    if elapsed_time < 1:
+                        print("Trigger - Short press detected")
+                        app_state["led_on"] = not app_state["led_on"]
+                        if app_state["led_on"]:
+                            print("light on")
+                            await set_all_pixels(Color(255, 255, 255))
+                        else:
+                            print("light off")
+                            await set_all_pixels(Color(0, 0, 0))
+                    app_state["starttime"] = None
+
+            await asyncio.sleep(0.1)
+        except Exception as e:
+            print(f"Error in light_set: {e}")
+            await asyncio.sleep(0.1)
+
+
+async def main():
+    print("Starting GPIO monitoring...")
+    print(f"Monitoring pin {pin_to_monitor}")
+    print("Press Ctrl+C to stop")
+
+    try:
+        # 创建任务
+        task1 = asyncio.create_task(monitor_io())
+        task2 = asyncio.create_task(light_set())
+
+        # 等待任务完成
+        await asyncio.gather(task1, task2)
+    except asyncio.CancelledError:
+        print("Tasks cancelled")
+    except Exception as e:
+        print(f"Error in main: {e}")
+
+
+try:
+    asyncio.run(main())
+except KeyboardInterrupt:
+    print("\nMonitoring stopped by user")
+except Exception as e:
+    print(f"Unexpected error: {e}")
+finally:
+    print("Cleaning up GPIO...")
+    GPIO.cleanup()  # 清理所有使用过的 GPIO
+    print("GPIO cleanup completed")

src/chatbot/tools/audio.py

@@ -3,6 +3,8 @@
 import subprocess
 from typing import TYPE_CHECKING
 
+from chatbot.console.logger import Logger
+
 if TYPE_CHECKING:
     from pathlib import Path
 
@@ -52,3 +54,39 @@ def file_to_wav(input_path: Path, output_path: Path) -> None:
         print("Error: ffmpeg not found. Please ensure ffmpeg is installed and added to PATH.")
     except Exception as e:
         print(f"Error converting {input_path} to WAV: {e}")
+
+
+def file_to_opus(input_path: Path, output_path: Path):
+    """
+    使用 ffmpeg 将 * 文件转换为 Opus 文件，并应用指定的参数，使用 run_shell_command 执行命令。
+    可以处理 MP4, MP3, AAC, FLAC, etc. 等 FFmpeg 支持的格式。
+    """
+    # TODO 这一步可能比较久,但是只在结束时输出, 可以考虑用 wepxct 和 pexpect
+    if not input_path.exists():
+        raise FileNotFoundError(f"输入文件不存在: {input_path}")
+
+    if input_path.suffix.lower() == ".opus":
+        Logger.info("文件已经是 Opus 格式，无需转换。")
+        return input_path
+
+    command = [
+        "ffmpeg",
+        "-y",  # 强制覆盖输出文件
+        "-i",
+        str(input_path.absolute()),
+        "-vn",  # 禁用视频流
+        "-c:a",
+        "libopus",  # 音频编码器，Opus
+        str(output_path.absolute()),
+    ]
+    try:
+        subprocess.run(
+            command,
+            check=True,
+            stdout=subprocess.DEVNULL,  # 屏蔽子进程标准输出
+            stderr=subprocess.DEVNULL,  # 屏蔽子进程错误输出
+        )
+    except FileNotFoundError:
+        print("Error: ffmpeg not found. Please ensure ffmpeg is installed and added to PATH.")
+    except subprocess.CalledProcessError as e:
+        print(f"Error converting {input_path} to Opus: {e}")