feat: add tap sensitivity slider entity for Home Assistant

PROJECT_PLAN.md

@@ -70,6 +70,7 @@
 - [x] ESPHome 协议服务器实现
 - [x] mDNS 服务发现（自动被 Home Assistant 发现）
 - [x] 本地唤醒词检测（microWakeWord）
+- [x] 拍一拍唤醒（IMU 加速度检测，仅无线版本）
 - [x] 音频流传输到 Home Assistant
 - [x] TTS 音频播放
 - [x] 停止词检测
@@ -584,7 +585,7 @@ VAD_DB_OFF = -45  # 停止检测阈值
 - ❌ 用户说 "记住这个" → 保存动作
 - ❌ 用户说 "做刚才的动作" → 播放录制的动作
 
-### Phase 20 - 环境感知响应 (未实现) ❌
+### Phase 20 - 环境感知响应 (部分实现) 🟡
 
 **目标**: 利用 IMU 传感器感知环境变化并做出响应。
 
@@ -592,11 +593,49 @@ VAD_DB_OFF = -45  # 停止检测阈值
 - ✅ `mini.imu["accelerometer"]` - 加速度计 (Phase 7 已实现为实体)
 - ✅ `mini.imu["gyroscope"]` - 陀螺仪 (Phase 7 已实现为实体)
 
+**已实现功能**:
+
+| 检测事件 | 响应动作 | 实现状态 |
+|---------|---------|---------|
+| 拍一拍唤醒 | 进入持续对话模式 | ✅ 已实现 |
+| 再次拍一拍 | 退出持续对话模式 | ✅ 已实现 |
+
+**拍一拍唤醒 vs 语音唤醒**:
+
+| 唤醒方式 | 对话模式 | 说明 |
+|---------|---------|------|
+| 语音唤醒 (Okay Nabu) | 单次对话 | 每次对话需要重新说唤醒词 |
+| 拍一拍唤醒 | 持续对话 | TTS 结束后自动继续监听，再拍一次退出 |
+
+**技术实现**:
+- `tap_detector.py` - IMU 加速度突变检测
+- `satellite.py:_tap_conversation_mode` - 持续对话模式标志
+- 阈值: 2.0g (可配置)
+- 冷却时间: 1.0s (防止重复触发)
+- 仅限无线版本 (Wireless) 可用
+
+```python
+# satellite.py - 持续对话模式
+def wakeup_from_tap(self):
+    if self._tap_conversation_mode:
+        # 第二次拍 - 退出持续对话
+        self._tap_conversation_mode = False
+        self._reachy_on_idle()
+    else:
+        # 第一次拍 - 进入持续对话
+        self._tap_conversation_mode = True
+        self.send_messages([VoiceAssistantRequest(start=True)])
+
+def _tts_finished(self):
+    if self._tap_conversation_mode:
+        # 持续对话模式：自动继续监听
+        self.send_messages([VoiceAssistantRequest(start=True)])
+```
+
 **未实现功能**:
 
 | 检测事件 | 响应动作 | 实现状态 |
 |---------|---------|---------|
-| 被拍打/敲击 | 播放惊讶动作 + 语音 "哎呀!" | ❌ 未实现 |
 | 被摇晃 | 播放晕眩动作 + 语音 "别晃我~" | ❌ 未实现 |
 | 倾斜/倒下 | 播放求助动作 + 语音 "我倒了，帮帮我" | ❌ 未实现 |
 | 长时间静止 | 进入休眠动画 | ❌ 未实现 |

reachy_mini_ha_voice/entity_registry.py

@@ -13,6 +13,7 @@ from .entity_extensions import SensorEntity, SwitchEntity, SelectEntity, ButtonE
 if TYPE_CHECKING:
     from .reachy_controller import ReachyController
     from .camera_server import MJPEGCameraServer
+    from .tap_detector import TapDetector
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -79,6 +80,8 @@ ENTITY_KEYS: Dict[str, int] = {
     "agc_max_gain": 1201,
     "noise_suppression": 1202,
     "echo_cancellation_converged": 1203,
+    # Phase 20: Tap detection
+    "tap_sensitivity": 1300,
 }
 
 
@@ -100,6 +103,7 @@ class EntityRegistry:
         reachy_controller: "ReachyController",
         camera_server: Optional["MJPEGCameraServer"] = None,
         play_emotion_callback: Optional[Callable[[str], None]] = None,
+        tap_detector: Optional["TapDetector"] = None,
     ):
         """Initialize the entity registry.
 
@@ -108,11 +112,13 @@ class EntityRegistry:
             reachy_controller: The ReachyController instance
             camera_server: Optional camera server for camera entity
             play_emotion_callback: Optional callback for playing emotions
+            tap_detector: Optional tap detector for sensitivity control
         """
         self.server = server
         self.reachy_controller = reachy_controller
         self.camera_server = camera_server
         self._play_emotion_callback = play_emotion_callback
+        self.tap_detector = tap_detector
 
         # Emotion state
         self._current_emotion = "None"
@@ -145,6 +151,7 @@ class EntityRegistry:
         # Phase 11 (LED control) disabled - LEDs are inside the robot and not visible
         self._setup_phase12_entities(entities)
         # Phase 13-14 (head_joints, passive_joints) removed - not needed
+        self._setup_phase20_entities(entities)
 
         _LOGGER.info("All entities registered: %d total", len(entities))
 
@@ -726,6 +733,40 @@ class EntityRegistry:
 
         _LOGGER.debug("Phase 12 entities registered: agc_enabled, agc_max_gain, noise_suppression, echo_cancellation_converged")
 
+    def _setup_phase20_entities(self, entities: List) -> None:
+        """Setup Phase 20 entities: Tap detection settings (Wireless only)."""
+        from .tap_detector import TAP_THRESHOLD_G_MIN, TAP_THRESHOLD_G_MAX, TAP_THRESHOLD_G_DEFAULT
+
+        def get_tap_sensitivity() -> float:
+            """Get current tap sensitivity threshold in g."""
+            if self.tap_detector:
+                return self.tap_detector.threshold_g
+            return TAP_THRESHOLD_G_DEFAULT
+
+        def set_tap_sensitivity(value: float) -> None:
+            """Set tap sensitivity threshold in g."""
+            if self.tap_detector:
+                self.tap_detector.threshold_g = value
+                _LOGGER.info("Tap sensitivity set to %.2fg", value)
+
+        entities.append(NumberEntity(
+            server=self.server,
+            key=get_entity_key("tap_sensitivity"),
+            name="Tap Sensitivity",
+            object_id="tap_sensitivity",
+            min_value=TAP_THRESHOLD_G_MIN,
+            max_value=TAP_THRESHOLD_G_MAX,
+            step=0.1,
+            icon="mdi:gesture-tap",
+            unit_of_measurement="g",
+            mode=2,  # Slider mode
+            entity_category=1,  # config
+            value_getter=get_tap_sensitivity,
+            value_setter=set_tap_sensitivity,
+        ))
+
+        _LOGGER.debug("Phase 20 entities registered: tap_sensitivity")
+
     def find_entity_references(self, entities: List) -> None:
         """Find and store references to special entities from existing list.
 

reachy_mini_ha_voice/models.py

@@ -73,6 +73,7 @@ class ServerState:
     reachy_mini: Optional[object] = None
     motion_enabled: bool = True
     motion: Optional[object] = None  # ReachyMiniMotion instance
+    tap_detector: Optional[object] = None  # TapDetector instance (Wireless only)
 
     media_player_entity: "Optional[MediaPlayerEntity]" = None
     satellite: "Optional[VoiceSatelliteProtocol]" = None

reachy_mini_ha_voice/motion.py

@@ -112,6 +112,26 @@ class ReachyMiniMotion:
         self._movement_manager.set_state(RobotState.LISTENING)
         _LOGGER.debug("Reachy Mini: Listening pose")
 
+    def on_continue_listening(self):
+        """Called when continuing to listen in tap conversation mode.
+        
+        Performs a small nod to indicate ready for next input.
+        Non-blocking: command sent to MovementManager.
+        """
+        if self._movement_manager is None:
+            return
+
+        self._movement_manager.set_state(RobotState.LISTENING)
+        
+        # Small nod to indicate ready
+        action = PendingAction(
+            name="continue_nod",
+            target_pitch=math.radians(8),  # Small nod down
+            duration=0.25,
+        )
+        self._movement_manager.queue_action(action)
+        _LOGGER.debug("Reachy Mini: Continue listening (nod)")
+
     def on_thinking(self):
         """Called when processing speech - thinking pose.
 

reachy_mini_ha_voice/satellite.py

@@ -75,16 +75,21 @@ class VoiceSatelliteProtocol(APIServer):
         self._continue_conversation = False
         self._timer_finished = False
         self._external_wake_words: Dict[str, VoiceAssistantExternalWakeWord] = {}
+        
+        # Tap-to-talk continuous conversation mode
+        self._tap_conversation_mode = False  # When True, auto-continue after TTS
 
         # Initialize Reachy controller
         self.reachy_controller = ReachyController(state.reachy_mini)
 
-        # Initialize entity registry
+        # Initialize entity registry (tap_detector from state if available)
+        tap_detector = getattr(state, 'tap_detector', None)
         self._entity_registry = EntityRegistry(
             server=self,
             reachy_controller=self.reachy_controller,
             camera_server=camera_server,
             play_emotion_callback=self._play_emotion,
+            tap_detector=tap_detector,
         )
 
         # Only setup entities once (check if already initialized)
@@ -330,6 +335,44 @@ class VoiceSatelliteProtocol(APIServer):
         self._is_streaming_audio = True
         self.state.tts_player.play(self.state.wakeup_sound)
 
+    def wakeup_from_tap(self) -> None:
+        """Trigger wake-up from tap detection (no wake word).
+        
+        First tap: Enter continuous conversation mode
+        Second tap (while in conversation): Exit continuous conversation mode
+        """
+        # If already in tap conversation mode, exit it
+        if self._tap_conversation_mode:
+            _LOGGER.info("Tap detected - exiting continuous conversation mode")
+            self._tap_conversation_mode = False
+            self._is_streaming_audio = False
+            # Stop any ongoing TTS
+            self.state.tts_player.stop()
+            self.unduck()
+            self._reachy_on_idle()
+            return
+        
+        if self._timer_finished:
+            # Stop timer instead
+            self._timer_finished = False
+            self.state.tts_player.stop()
+            _LOGGER.debug("Stopping timer finished sound")
+            return
+
+        _LOGGER.info("Tap detected - entering continuous conversation mode")
+        self._tap_conversation_mode = True
+
+        self.send_messages(
+            [VoiceAssistantRequest(start=True, wake_word_phrase="tap")]
+        )
+        self.duck()
+        self._is_streaming_audio = True
+        self.state.tts_player.play(self.state.wakeup_sound)
+
+    def is_tap_conversation_active(self) -> bool:
+        """Check if tap-triggered continuous conversation is active."""
+        return self._tap_conversation_mode
+
     def stop(self) -> None:
         self.state.active_wake_words.discard(self.state.stop_word.id)
         self.state.tts_player.stop()
@@ -364,10 +407,21 @@ class VoiceSatelliteProtocol(APIServer):
         self.state.active_wake_words.discard(self.state.stop_word.id)
         self.send_messages([VoiceAssistantAnnounceFinished()])
 
-        if self._continue_conversation:
+        # Check if should continue conversation
+        # 1. HA requested continue (intent not matched)
+        # 2. Tap conversation mode is active
+        should_continue = self._continue_conversation or self._tap_conversation_mode
+
+        if should_continue:
             self.send_messages([VoiceAssistantRequest(start=True)])
             self._is_streaming_audio = True
-            _LOGGER.debug("Continuing conversation")
+            
+            if self._tap_conversation_mode:
+                _LOGGER.debug("Continuing tap conversation mode")
+                # Provide feedback for continuous conversation mode
+                self._tap_continue_feedback()
+            else:
+                _LOGGER.debug("Continuing conversation (HA requested)")
         else:
             self.unduck()
             _LOGGER.debug("TTS response finished")
@@ -511,6 +565,23 @@ class VoiceSatelliteProtocol(APIServer):
         except Exception as e:
             _LOGGER.error("Reachy Mini motion error: %s", e)
 
+    def _tap_continue_feedback(self) -> None:
+        """Provide feedback when continuing conversation in tap mode.
+        
+        Plays a short sound and triggers a nod to indicate ready for next input.
+        """
+        try:
+            # Play the wakeup sound (short beep) to indicate listening
+            self.state.tts_player.play(self.state.wakeup_sound)
+            _LOGGER.debug("Tap continue feedback: sound played")
+            
+            # Trigger a small nod to indicate ready for input
+            if self.state.motion_enabled and self.state.motion:
+                self.state.motion.on_continue_listening()
+                _LOGGER.debug("Tap continue feedback: continue listening pose")
+        except Exception as e:
+            _LOGGER.error("Tap continue feedback error: %s", e)
+
     def _reachy_on_timer_finished(self) -> None:
         """Called when a timer finishes."""
         if not self.state.motion_enabled or not self.state.reachy_mini:

reachy_mini_ha_voice/tap_detector.py

@@ -0,0 +1,191 @@
+"""Tap detection using IMU accelerometer data.
+
+This module provides tap/knock detection for Reachy Mini (Wireless version only).
+When a tap is detected, it can trigger the voice assistant wake-up.
+"""
+
+import logging
+import math
+import threading
+import time
+from typing import Callable, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from reachy_mini import ReachyMini
+
+_LOGGER = logging.getLogger(__name__)
+
+# Tap detection parameters
+TAP_THRESHOLD_G_DEFAULT = 2.0  # Default acceleration threshold in g
+TAP_THRESHOLD_G_MIN = 0.5  # Minimum threshold (very sensitive)
+TAP_THRESHOLD_G_MAX = 5.0  # Maximum threshold (less sensitive)
+TAP_COOLDOWN_SECONDS = 1.0  # Minimum time between tap detections
+TAP_DETECTION_RATE_HZ = 50  # IMU polling rate
+
+
+class TapDetector:
+    """Detects taps/knocks on Reachy Mini using IMU accelerometer."""
+
+    def __init__(
+        self,
+        reachy_mini: Optional["ReachyMini"] = None,
+        on_tap_callback: Optional[Callable[[], None]] = None,
+        threshold_g: float = TAP_THRESHOLD_G_DEFAULT,
+        cooldown_seconds: float = TAP_COOLDOWN_SECONDS,
+    ):
+        """Initialize tap detector.
+        
+        Args:
+            reachy_mini: Reachy Mini robot instance
+            on_tap_callback: Callback function when tap is detected
+            threshold_g: Acceleration threshold in g units (0.5-5.0)
+            cooldown_seconds: Minimum time between tap detections
+        """
+        self.reachy_mini = reachy_mini
+        self._on_tap_callback = on_tap_callback
+        self._threshold_g = threshold_g
+        self._threshold_ms2 = threshold_g * 9.81
+        self._cooldown_seconds = cooldown_seconds
+        
+        self._running = False
+        self._thread: Optional[threading.Thread] = None
+        self._last_tap_time: float = 0.0
+        self._enabled = True
+        self._baseline_magnitude: float = 9.81
+
+    @property
+    def threshold_g(self) -> float:
+        """Get current threshold in g units."""
+        return self._threshold_g
+
+    @threshold_g.setter
+    def threshold_g(self, value: float) -> None:
+        """Set threshold in g units (clamped to valid range)."""
+        value = max(TAP_THRESHOLD_G_MIN, min(TAP_THRESHOLD_G_MAX, value))
+        self._threshold_g = value
+        self._threshold_ms2 = value * 9.81
+        _LOGGER.info("Tap threshold set to %.2fg", value)
+
+    def set_reachy_mini(self, reachy_mini: "ReachyMini") -> None:
+        """Set the Reachy Mini instance."""
+        self.reachy_mini = reachy_mini
+
+    def set_callback(self, callback: Callable[[], None]) -> None:
+        """Set the tap detection callback."""
+        self._on_tap_callback = callback
+
+    def set_enabled(self, enabled: bool) -> None:
+        """Enable or disable tap detection."""
+        self._enabled = enabled
+        _LOGGER.info("Tap detection %s", "enabled" if enabled else "disabled")
+
+    @property
+    def is_running(self) -> bool:
+        """Check if tap detector is running."""
+        return self._running
+
+    def start(self) -> None:
+        """Start tap detection thread."""
+        if self._running:
+            _LOGGER.warning("TapDetector already running")
+            return
+        
+        if self.reachy_mini is None:
+            _LOGGER.warning("Cannot start TapDetector: no Reachy Mini instance")
+            return
+        
+        # Check if IMU is available (Wireless version only)
+        try:
+            imu_data = self.reachy_mini.imu
+            if imu_data is None:
+                _LOGGER.warning(
+                    "IMU not available - tap detection disabled "
+                    "(only available on Wireless version)"
+                )
+                return
+        except Exception as e:
+            _LOGGER.warning("Failed to check IMU availability: %s", e)
+            return
+        
+        self._running = True
+        self._thread = threading.Thread(
+            target=self._detection_loop,
+            daemon=True,
+            name="tap-detector"
+        )
+        self._thread.start()
+        _LOGGER.info("TapDetector started (threshold=%.1fg)", self._threshold_ms2 / 9.81)
+
+    def stop(self) -> None:
+        """Stop tap detection thread."""
+        self._running = False
+        if self._thread is not None:
+            self._thread.join(timeout=2.0)
+            self._thread = None
+        _LOGGER.info("TapDetector stopped")
+
+    def _calibrate_baseline(self) -> None:
+        """Calibrate baseline acceleration (gravity)."""
+        samples = []
+        for _ in range(20):
+            if not self._running or self.reachy_mini is None:
+                break
+            try:
+                imu = self.reachy_mini.imu
+                if imu and "accelerometer" in imu:
+                    ax, ay, az = imu["accelerometer"]
+                    samples.append(math.sqrt(ax*ax + ay*ay + az*az))
+            except Exception:
+                pass
+            time.sleep(0.02)
+        
+        if samples:
+            self._baseline_magnitude = sum(samples) / len(samples)
+            _LOGGER.info("IMU baseline calibrated: %.2f m/s²", self._baseline_magnitude)
+
+    def _detection_loop(self) -> None:
+        """Main detection loop running in background thread."""
+        _LOGGER.debug("Tap detection loop started")
+        
+        # Calibration phase
+        self._calibrate_baseline()
+        
+        interval = 1.0 / TAP_DETECTION_RATE_HZ
+        
+        while self._running:
+            try:
+                if not self._enabled or self.reachy_mini is None:
+                    time.sleep(interval)
+                    continue
+                
+                imu = self.reachy_mini.imu
+                if not imu or "accelerometer" not in imu:
+                    time.sleep(interval)
+                    continue
+                
+                # Get accelerometer data
+                ax, ay, az = imu["accelerometer"]
+                magnitude = math.sqrt(ax*ax + ay*ay + az*az)
+                
+                # Detect tap: sudden spike above baseline + threshold
+                delta = abs(magnitude - self._baseline_magnitude)
+                
+                if delta > self._threshold_ms2:
+                    now = time.time()
+                    if now - self._last_tap_time > self._cooldown_seconds:
+                        self._last_tap_time = now
+                        _LOGGER.info("Tap detected! (delta=%.2f m/s²)", delta)
+                        
+                        if self._on_tap_callback:
+                            try:
+                                self._on_tap_callback()
+                            except Exception as e:
+                                _LOGGER.error("Tap callback error: %s", e)
+                
+                time.sleep(interval)
+                
+            except Exception as e:
+                _LOGGER.debug("Tap detection error: %s", e)
+                time.sleep(0.1)
+        
+        _LOGGER.debug("Tap detection loop stopped")

reachy_mini_ha_voice/voice_assistant.py

@@ -26,6 +26,7 @@ from .util import get_mac
 from .zeroconf import HomeAssistantZeroconf
 from .motion import ReachyMiniMotion
 from .camera_server import MJPEGCameraServer
+from .tap_detector import TapDetector
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -75,6 +76,8 @@ class VoiceAssistantService:
         self._state: Optional[ServerState] = None
         self._motion = ReachyMiniMotion(reachy_mini)
         self._camera_server: Optional[MJPEGCameraServer] = None
+        self._tap_detector: Optional[TapDetector] = None
+        self._last_tap_wakeup: float = 0.0  # For refractory period
 
     async def start(self) -> None:
         """Start the voice assistant service."""
@@ -165,6 +168,18 @@ class VoiceAssistantService:
         if self._motion is not None:
             self._motion.start()
 
+        # Start tap detector for "tap to wake" (Wireless version only)
+        if self.reachy_mini is not None:
+            self._tap_detector = TapDetector(
+                reachy_mini=self.reachy_mini,
+                on_tap_callback=self._on_tap_detected,
+                threshold_g=2.0,
+                cooldown_seconds=1.0,
+            )
+            self._tap_detector.start()
+            # Store tap_detector in state for entity registry access
+            self._state.tap_detector = self._tap_detector
+
         # Start audio processing thread (non-daemon for proper cleanup)
         self._running = True
         self._audio_thread = threading.Thread(
@@ -247,7 +262,12 @@ class VoiceAssistantService:
             await self._camera_server.stop()
             self._camera_server = None
 
-        # 8. Shutdown motion executor
+        # 8. Stop tap detector
+        if self._tap_detector:
+            self._tap_detector.stop()
+            self._tap_detector = None
+
+        # 9. Shutdown motion executor
         if self._motion:
             self._motion.shutdown()
 
@@ -619,3 +639,35 @@ class VoiceAssistantService:
         if stopped and (self._state.stop_word.id in self._state.active_wake_words):
             _LOGGER.info("Stop word detected")
             self._state.satellite.stop()
+
+    def _on_tap_detected(self) -> None:
+        """Callback when tap is detected on the robot.
+        
+        First tap: Enter continuous conversation mode
+        Second tap: Exit continuous conversation mode
+        """
+        if self._state is None or self._state.satellite is None:
+            return
+        
+        # Check refractory period
+        now = time.monotonic()
+        if now - self._last_tap_wakeup < self._state.refractory_seconds:
+            _LOGGER.debug("Tap ignored (refractory period)")
+            return
+        
+        self._last_tap_wakeup = now
+        
+        # Check if we're exiting conversation mode
+        is_exiting = self._state.satellite.is_tap_conversation_active()
+        
+        # Trigger tap handling in satellite (handles mode toggle)
+        self._state.satellite.wakeup_from_tap()
+        
+        # Trigger motion feedback
+        if self._motion is not None:
+            if is_exiting:
+                # Exiting conversation - return to idle
+                self._motion.on_idle()
+            else:
+                # Starting conversation
+                self._motion.on_wakeup()