reachy_mini_ha_voice/satellite.py

"""Voice satellite protocol for Reachy Mini."""

import hashlib
import logging
import math
import posixpath
import shutil
import time
from collections.abc import Iterable
from typing import Dict, Optional, Set, Union, TYPE_CHECKING
from urllib.parse import urlparse, urlunparse
from urllib.request import urlopen

if TYPE_CHECKING:
    from .camera_server import MJPEGCameraServer

# pylint: disable=no-name-in-module
from aioesphomeapi.api_pb2 import (  # type: ignore[attr-defined]
    ButtonCommandRequest,
    CameraImageRequest,
    DeviceInfoRequest,
    DeviceInfoResponse,
    ListEntitiesDoneResponse,
    ListEntitiesRequest,
    MediaPlayerCommandRequest,
    NumberCommandRequest,
    SelectCommandRequest,
    SubscribeHomeAssistantStatesRequest,
    SubscribeStatesRequest,
    SwitchCommandRequest,
    VoiceAssistantAnnounceFinished,
    VoiceAssistantAnnounceRequest,
    VoiceAssistantAudio,
    VoiceAssistantConfigurationRequest,
    VoiceAssistantConfigurationResponse,
    VoiceAssistantEventResponse,
    VoiceAssistantExternalWakeWord,
    VoiceAssistantRequest,
    VoiceAssistantSetConfiguration,
    VoiceAssistantTimerEventResponse,
    VoiceAssistantWakeWord,
)
from aioesphomeapi.model import (
    VoiceAssistantEventType,
    VoiceAssistantFeature,
    VoiceAssistantTimerEventType,
)
from google.protobuf import message
from pymicro_wakeword import MicroWakeWord
from pyopen_wakeword import OpenWakeWord

from .api_server import APIServer
from .entity import MediaPlayerEntity
from .entity_registry import EntityRegistry, get_entity_key
from .models import AvailableWakeWord, ServerState, WakeWordType
from .util import call_all
from .reachy_controller import ReachyController

_LOGGER = logging.getLogger(__name__)


class VoiceSatelliteProtocol(APIServer):
    """Voice satellite protocol handler for ESPHome."""

    def __init__(self, state: ServerState, camera_server: Optional["MJPEGCameraServer"] = None) -> None:
        super().__init__(state.name)
        self.state = state
        self.state.satellite = self
        self.camera_server = camera_server

        # Initialize streaming state early (before entity setup)
        # This is needed because audio processing thread checks this attribute
        self._is_streaming_audio = False
        self._in_pipeline = False  # True when voice pipeline is active (listening/processing/speaking)
        self._tts_url: Optional[str] = None
        self._tts_played = False
        self._continue_conversation = False
        self._timer_finished = False
        self._external_wake_words: Dict[str, VoiceAssistantExternalWakeWord] = {}

        # Tap-to-talk continuous conversation mode (REMOVED - too many false triggers)
        # Continuous conversation is now controlled via Home Assistant switch
        # self._tap_conversation_mode = False

        # Conversation tracking for continuous conversation
        self._conversation_id: Optional[str] = None
        self._conversation_timeout = 300.0  # 5 minutes, same as ESPHome default
        self._last_conversation_time = 0.0

        # Initialize Reachy controller
        self.reachy_controller = ReachyController(state.reachy_mini)

        # Connect MovementManager to ReachyController for pose control from HA
        if state.motion is not None and state.motion.movement_manager is not None:
            self.reachy_controller.set_movement_manager(state.motion.movement_manager)

            # Setup speech sway callback for audio-driven head motion
            def sway_callback(sway: dict) -> None:
                mm = state.motion.movement_manager
                if mm is not None:
                    mm.set_speech_sway(
                        sway.get("x_m", 0.0),
                        sway.get("y_m", 0.0),
                        sway.get("z_m", 0.0),
                        sway.get("roll_rad", 0.0),
                        sway.get("pitch_rad", 0.0),
                        sway.get("yaw_rad", 0.0),
                    )

            state.tts_player.set_sway_callback(sway_callback)
            _LOGGER.info("Speech sway callback configured for TTS player")

        # Initialize entity registry
        self._entity_registry = EntityRegistry(
            server=self,
            reachy_controller=self.reachy_controller,
            camera_server=camera_server,
            play_emotion_callback=self._play_emotion,
        )

        # Connect gesture state callback
        if camera_server:
            camera_server.set_gesture_state_callback(self._entity_registry.update_gesture_state)

        # Only setup entities once (check if already initialized)
        # This prevents duplicate entity registration on reconnection
        if not getattr(self.state, '_entities_initialized', False):
            if self.state.media_player_entity is None:
                self.state.media_player_entity = MediaPlayerEntity(
                    server=self,
                    key=get_entity_key("reachy_mini_media_player"),
                    name="Media Player",
                    object_id="reachy_mini_media_player",
                    music_player=state.music_player,
                    announce_player=state.tts_player,
                )
                self.state.entities.append(self.state.media_player_entity)

            # Setup all entities using the registry
            self._entity_registry.setup_all_entities(self.state.entities)

            # Mark entities as initialized
            self.state._entities_initialized = True
            _LOGGER.info("Entities initialized: %d total", len(self.state.entities))
        else:
            _LOGGER.debug("Entities already initialized, skipping setup")
            # Update server reference in existing entities
            for entity in self.state.entities:
                entity.server = self

    def handle_voice_event(
        self, event_type: VoiceAssistantEventType, data: Dict[str, str]
    ) -> None:
        _LOGGER.debug("Voice event: type=%s, data=%s", event_type.name, data)

        if event_type == VoiceAssistantEventType.VOICE_ASSISTANT_RUN_START:
            self._tts_url = data.get("url")
            self._tts_played = False
            self._continue_conversation = False
            # Reachy Mini: Start listening animation
            self._reachy_on_listening()

            # Note: TTS URL requires HA authentication, cannot pre-download
            # Speaking animation uses JSON-defined multi-frequency sway instead

        elif event_type in (
            VoiceAssistantEventType.VOICE_ASSISTANT_STT_VAD_END,
            VoiceAssistantEventType.VOICE_ASSISTANT_STT_END,
        ):
            self._is_streaming_audio = False
            # Reachy Mini: Stop listening, start thinking
            self._reachy_on_thinking()

        elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_PROGRESS:
            if data.get("tts_start_streaming") == "1":
                # Start streaming early
                self.play_tts()

        elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_END:
            if data.get("continue_conversation") == "1":
                self._continue_conversation = True

        elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_TTS_START:
            # Reachy Mini: Start speaking animation (JSON-defined multi-frequency sway)
            _LOGGER.debug("TTS_START event received, triggering speaking animation")
            self._reachy_on_speaking()

        elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_TTS_END:
            self._tts_url = data.get("url")
            self.play_tts()

        elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_RUN_END:
            # Pipeline run ended
            self._tts_played = False
            self._is_streaming_audio = False

            # Check if should continue conversation
            self._handle_run_end()

    def handle_timer_event(
        self,
        event_type: VoiceAssistantTimerEventType,
        msg: VoiceAssistantTimerEventResponse,
    ) -> None:
        _LOGGER.debug("Timer event: type=%s", event_type.name)

        if event_type == VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_FINISHED:
            if not self._timer_finished:
                self.state.active_wake_words.add(self.state.stop_word.id)
                self._timer_finished = True
                self.duck()
                self._play_timer_finished()
                # Reachy Mini: Timer finished animation
                self._reachy_on_timer_finished()

    def handle_message(self, msg: message.Message) -> Iterable[message.Message]:
        if isinstance(msg, VoiceAssistantEventResponse):
            # Pipeline event
            data: Dict[str, str] = {}
            for arg in msg.data:
                data[arg.name] = arg.value
            self.handle_voice_event(VoiceAssistantEventType(msg.event_type), data)

        elif isinstance(msg, VoiceAssistantAnnounceRequest):
            _LOGGER.debug("Announcing: %s", msg.text)
            assert self.state.media_player_entity is not None

            urls = []
            if msg.preannounce_media_id:
                urls.append(msg.preannounce_media_id)
            urls.append(msg.media_id)

            self.state.active_wake_words.add(self.state.stop_word.id)
            self._continue_conversation = msg.start_conversation
            self.duck()

            yield from self.state.media_player_entity.play(
                urls, announcement=True, done_callback=self._tts_finished
            )

        elif isinstance(msg, VoiceAssistantTimerEventResponse):
            self.handle_timer_event(VoiceAssistantTimerEventType(msg.event_type), msg)

        elif isinstance(msg, DeviceInfoRequest):
            yield DeviceInfoResponse(
                uses_password=False,
                name=self.state.name,
                mac_address=self.state.mac_address,
                voice_assistant_feature_flags=(
                    VoiceAssistantFeature.VOICE_ASSISTANT
                    | VoiceAssistantFeature.API_AUDIO
                    | VoiceAssistantFeature.ANNOUNCE
                    | VoiceAssistantFeature.START_CONVERSATION
                    | VoiceAssistantFeature.TIMERS
                ),
            )

        elif isinstance(
            msg,
            (
                ListEntitiesRequest,
                SubscribeHomeAssistantStatesRequest,
                SubscribeStatesRequest,
                MediaPlayerCommandRequest,
                NumberCommandRequest,
                SwitchCommandRequest,
                SelectCommandRequest,
                ButtonCommandRequest,
                CameraImageRequest,
            ),
        ):
            for entity in self.state.entities:
                yield from entity.handle_message(msg)

            if isinstance(msg, ListEntitiesRequest):
                yield ListEntitiesDoneResponse()

        elif isinstance(msg, VoiceAssistantConfigurationRequest):
            available_wake_words = [
                VoiceAssistantWakeWord(
                    id=ww.id,
                    wake_word=ww.wake_word,
                    trained_languages=ww.trained_languages,
                )
                for ww in self.state.available_wake_words.values()
            ]

            for eww in msg.external_wake_words:
                if eww.model_type != "micro":
                    continue

                available_wake_words.append(
                    VoiceAssistantWakeWord(
                        id=eww.id,
                        wake_word=eww.wake_word,
                        trained_languages=eww.trained_languages,
                    )
                )
                self._external_wake_words[eww.id] = eww

            yield VoiceAssistantConfigurationResponse(
                available_wake_words=available_wake_words,
                active_wake_words=[
                    ww.id
                    for ww in self.state.wake_words.values()
                    if ww.id in self.state.active_wake_words
                ],
                max_active_wake_words=2,
            )

            _LOGGER.info("Connected to Home Assistant")

        elif isinstance(msg, VoiceAssistantSetConfiguration):
            # Change active wake words
            active_wake_words: Set[str] = set()

            for wake_word_id in msg.active_wake_words:
                if wake_word_id in self.state.wake_words:
                    # Already loaded, just add to active set
                    active_wake_words.add(wake_word_id)
                    continue

                model_info = self.state.available_wake_words.get(wake_word_id)
                if not model_info:
                    # Check external wake words (may require download)
                    external_wake_word = self._external_wake_words.get(wake_word_id)
                    if not external_wake_word:
                        _LOGGER.warning("Wake word not found: %s", wake_word_id)
                        continue

                    model_info = self._download_external_wake_word(external_wake_word)
                    if not model_info:
                        continue

                    self.state.available_wake_words[wake_word_id] = model_info

                _LOGGER.debug("Loading wake word: %s", model_info.wake_word_path)
                loaded_model = model_info.load()
                # Set id attribute on the model for later identification
                setattr(loaded_model, 'id', wake_word_id)
                self.state.wake_words[wake_word_id] = loaded_model
                _LOGGER.info("Wake word loaded: %s", wake_word_id)
                active_wake_words.add(wake_word_id)
                # Don't break - load ALL requested wake words, not just the first one

            self.state.active_wake_words = active_wake_words
            _LOGGER.debug("Active wake words: %s", active_wake_words)

            self.state.preferences.active_wake_words = list(active_wake_words)
            self.state.save_preferences()
            self.state.wake_words_changed = True

    def handle_audio(self, audio_chunk: bytes) -> None:
        if not self._is_streaming_audio:
            return
        self.send_messages([VoiceAssistantAudio(data=audio_chunk)])

    def _get_or_create_conversation_id(self) -> str:
        """Get existing conversation_id or create a new one.

        Reuses conversation_id if within timeout period, otherwise creates new one.
        """
        now = time.time()
        if (self._conversation_id is None or
                now - self._last_conversation_time > self._conversation_timeout):
            # Create new conversation_id
            import uuid
            self._conversation_id = str(uuid.uuid4())
            _LOGGER.debug("Created new conversation_id: %s", self._conversation_id)

        self._last_conversation_time = now
        return self._conversation_id

    def _clear_conversation(self) -> None:
        """Clear conversation state when exiting conversation mode."""
        self._conversation_id = None
        self._continue_conversation = False

    def wakeup(self, wake_word: Union[MicroWakeWord, OpenWakeWord]) -> None:
        """Handle wake word detection - start voice pipeline.

        Only called when in idle state (checked by voice_assistant.py).
        """
        if self._timer_finished:
            # Stop timer instead
            self._timer_finished = False
            self.state.tts_player.stop()
            _LOGGER.debug("Stopping timer finished sound")
            return

        # Mark pipeline as active
        self._in_pipeline = True
        
        wake_word_phrase = wake_word.wake_word
        _LOGGER.debug("Detected wake word: %s", wake_word_phrase)

        # Turn toward sound source using DOA (Direction of Arrival)
        self._turn_to_sound_source()

        # Get or create conversation_id for context tracking
        conv_id = self._get_or_create_conversation_id()

        self.send_messages(
            [VoiceAssistantRequest(
                start=True,
                wake_word_phrase=wake_word_phrase,
                conversation_id=conv_id,
            )]
        )
        self.duck()
        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.

        NOTE: This method is DISABLED. Tap-to-wake caused too many false triggers.
        Continuous conversation is now controlled via Home Assistant switch.
        """
        _LOGGER.warning("wakeup_from_tap() called but tap wake is disabled")
        return

    def is_tap_conversation_active(self) -> bool:
        """Check if tap-triggered continuous conversation is active.

        NOTE: Tap wake is DISABLED. This always returns False.
        """
        return False

    def stop(self) -> None:
        """Stop current TTS playback (e.g., user said stop word)."""
        self.state.active_wake_words.discard(self.state.stop_word.id)
        self.state.tts_player.stop()

        if self._timer_finished:
            self._timer_finished = False
            _LOGGER.debug("Stopping timer finished sound")
        else:
            _LOGGER.debug("TTS response stopped manually")

        # Send announce finished to HA
        self.send_messages([VoiceAssistantAnnounceFinished()])
        # Note: RUN_END event will handle the rest

    def play_tts(self) -> None:
        if (not self._tts_url) or self._tts_played:
            return

        self._tts_played = True
        _LOGGER.debug("Playing TTS response: %s", self._tts_url)

        self.state.active_wake_words.add(self.state.stop_word.id)
        self.state.tts_player.play(self._tts_url, done_callback=self._tts_finished)

    def duck(self) -> None:
        _LOGGER.debug("Ducking music")
        self.state.music_player.duck()
        # Pause Sendspin to prevent audio conflicts during voice interaction
        self.state.music_player.pause_sendspin()

    def unduck(self) -> None:
        _LOGGER.debug("Unducking music")
        self.state.music_player.unduck()
        # Resume Sendspin audio
        self.state.music_player.resume_sendspin()

    def _tts_finished(self) -> None:
        """Called when TTS audio playback finishes.

        Following reference project pattern: handle continue conversation here.
        """
        self.state.active_wake_words.discard(self.state.stop_word.id)
        self.send_messages([VoiceAssistantAnnounceFinished()])

        # Check if should continue conversation
        # 1. Our switch is ON: Always continue (unconditional)
        # 2. Our switch is OFF: Follow HA's continue_conversation request
        continuous_mode = self.state.preferences.continuous_conversation
        should_continue = continuous_mode or self._continue_conversation

        if should_continue:
            _LOGGER.debug("Continuing conversation (our_switch=%s, ha_request=%s)",
                         continuous_mode, self._continue_conversation)

            # Play prompt sound to indicate ready for next input
            self.state.tts_player.play(self.state.wakeup_sound)

            # Use same conversation_id for context continuity
            conv_id = self._get_or_create_conversation_id()
            self.send_messages([VoiceAssistantRequest(
                start=True,
                conversation_id=conv_id,
            )])
            self._is_streaming_audio = True

            # Stay in listening mode
            self._reachy_on_listening()
        else:
            self._clear_conversation()
            self.unduck()
            _LOGGER.debug("Conversation finished")
            
            # Mark pipeline as inactive - ready for new wake word
            self._in_pipeline = False

            # Reachy Mini: Return to idle
            self._reachy_on_idle()

    def _handle_run_end(self) -> None:
        """Handle pipeline RUN_END event.

        Following reference project pattern: call _tts_finished if TTS wasn't played.
        """
        if not self._tts_played:
            self._tts_finished()

        self._tts_played = False

    def _play_timer_finished(self) -> None:
        if not self._timer_finished:
            self.unduck()
            return

        self.state.tts_player.play(
            self.state.timer_finished_sound,
            done_callback=lambda: call_all(
                lambda: time.sleep(1.0), self._play_timer_finished
            ),
        )

    def connection_lost(self, exc):
        super().connection_lost(exc)
        _LOGGER.info("Disconnected from Home Assistant")
        # Clear streaming state on disconnect
        self._is_streaming_audio = False
        self._in_pipeline = False
        self._tts_url = None
        self._tts_played = False
        self._continue_conversation = False

    def _download_external_wake_word(
        self, external_wake_word: VoiceAssistantExternalWakeWord
    ) -> Optional[AvailableWakeWord]:
        eww_dir = self.state.download_dir / "external_wake_words"
        eww_dir.mkdir(parents=True, exist_ok=True)

        config_path = eww_dir / f"{external_wake_word.id}.json"
        should_download_config = not config_path.exists()

        # Check if we need to download the model file
        model_path = eww_dir / f"{external_wake_word.id}.tflite"
        should_download_model = True

        if model_path.exists():
            model_size = model_path.stat().st_size
            if model_size == external_wake_word.model_size:
                with open(model_path, "rb") as model_file:
                    model_hash = hashlib.sha256(model_file.read()).hexdigest()

                if model_hash == external_wake_word.model_hash:
                    should_download_model = False
                    _LOGGER.debug(
                        "Model size and hash match for %s. Skipping download.",
                        external_wake_word.id,
                    )

        if should_download_config or should_download_model:
            # Download config
            _LOGGER.debug("Downloading %s to %s", external_wake_word.url, config_path)
            with urlopen(external_wake_word.url) as request:
                if request.status != 200:
                    _LOGGER.warning(
                        "Failed to download: %s, status=%s",
                        external_wake_word.url,
                        request.status,
                    )
                    return None

                with open(config_path, "wb") as model_file:
                    shutil.copyfileobj(request, model_file)

        if should_download_model:
            # Download model file
            parsed_url = urlparse(external_wake_word.url)
            parsed_url = parsed_url._replace(
                path=posixpath.join(posixpath.dirname(parsed_url.path), model_path.name)
            )
            model_url = urlunparse(parsed_url)

            _LOGGER.debug("Downloading %s to %s", model_url, model_path)
            with urlopen(model_url) as request:
                if request.status != 200:
                    _LOGGER.warning(
                        "Failed to download: %s, status=%s", model_url, request.status
                    )
                    return None

                with open(model_path, "wb") as model_file:
                    shutil.copyfileobj(request, model_file)

        return AvailableWakeWord(
            id=external_wake_word.id,
            type=WakeWordType.MICRO_WAKE_WORD,
            wake_word=external_wake_word.wake_word,
            trained_languages=external_wake_word.trained_languages,
            wake_word_path=config_path,
        )

    # -------------------------------------------------------------------------
    # Reachy Mini Motion Control
    # -------------------------------------------------------------------------

    def _turn_to_sound_source(self) -> None:
        """Turn robot head toward sound source using DOA at wakeup.

        This is called once at wakeup to orient the robot toward the speaker.
        Face tracking will take over after the initial turn.

        DOA angle convention (from SDK):
        - 0 radians = left (Y+ direction in head frame)
        - π/2 radians = front (X+ direction in head frame)
        - π radians = right (Y- direction in head frame)

        The SDK uses: p_head = [sin(doa), cos(doa), 0]
        So we need to convert this to yaw angle.

        Note: We don't check speech_detected because by the time wake word
        detection completes, the user may have stopped speaking.
        """
        if not self.state.motion_enabled or not self.state.reachy_mini:
            _LOGGER.info("DOA turn-to-sound: motion disabled or no robot")
            return

        try:
            # Get DOA from reachy_controller (only read once)
            doa = self.reachy_controller.get_doa_angle()
            if doa is None:
                _LOGGER.info("DOA not available, skipping turn-to-sound")
                return

            angle_rad, speech_detected = doa
            _LOGGER.debug("DOA raw: angle=%.3f rad (%.1f°), speech=%s",
                         angle_rad, math.degrees(angle_rad), speech_detected)

            # Convert DOA to direction vector in head frame
            # SDK convention: p_head = [sin(doa), cos(doa), 0]
            # where X+ is front, Y+ is left
            dir_x = math.sin(angle_rad)  # Front component
            dir_y = math.cos(angle_rad)  # Left component

            # Calculate yaw angle from direction vector
            # DOA convention: 0 = left, π/2 = front, π = right
            # Robot yaw: positive = turn left, negative = turn right
            # yaw = doa - π/2 maps: left(0) → -90°, front(π/2) → 0°, right(π) → +90°
            yaw_rad = angle_rad - math.pi / 2
            yaw_deg = math.degrees(yaw_rad)

            _LOGGER.debug("DOA direction: x=%.2f, y=%.2f, yaw=%.1f°",
                         dir_x, dir_y, yaw_deg)

            # Only turn if angle is significant (> 10°) to avoid noise
            DOA_THRESHOLD_DEG = 10.0
            if abs(yaw_deg) < DOA_THRESHOLD_DEG:
                _LOGGER.debug("DOA angle %.1f° below threshold (%.1f°), skipping turn",
                             yaw_deg, DOA_THRESHOLD_DEG)
                return

            # Apply 80% of DOA angle as conservative strategy
            # This accounts for potential DOA inaccuracy
            DOA_SCALE = 0.8
            target_yaw_deg = yaw_deg * DOA_SCALE

            _LOGGER.info("Turning toward sound source: DOA=%.1f°, target=%.1f°",
                         yaw_deg, target_yaw_deg)

            # Use MovementManager to turn (non-blocking)
            if self.state.motion and self.state.motion.movement_manager:
                self.state.motion.movement_manager.turn_to_angle(
                    target_yaw_deg,
                    duration=0.5  # Quick turn
                )
        except Exception as e:
            _LOGGER.error("Error in turn-to-sound: %s", e)

    def _reachy_on_listening(self) -> None:
        """Called when listening for speech (HA state: Listening)."""
        # Enable high-frequency face tracking during listening
        self._set_conversation_mode(True)

        # Resume face tracking (may have been paused during speaking)
        if self.camera_server is not None:
            try:
                self.camera_server.set_face_tracking_enabled(True)
            except Exception as e:
                _LOGGER.debug("Failed to resume face tracking: %s", e)

        if not self.state.motion_enabled or not self.state.reachy_mini:
            return
        try:
            _LOGGER.debug("Reachy Mini: Listening animation")
            if self.state.motion:
                self.state.motion.on_listening()
        except Exception as e:
            _LOGGER.error("Reachy Mini motion error: %s", e)

    def _reachy_on_thinking(self) -> None:
        """Called when processing speech (HA state: Processing)."""
        # Resume face tracking (may have been paused during speaking)
        if self.camera_server is not None:
            try:
                self.camera_server.set_face_tracking_enabled(True)
            except Exception as e:
                _LOGGER.debug("Failed to resume face tracking: %s", e)

        if not self.state.motion_enabled or not self.state.reachy_mini:
            return
        try:
            _LOGGER.debug("Reachy Mini: Thinking animation")
            if self.state.motion:
                self.state.motion.on_thinking()
        except Exception as e:
            _LOGGER.error("Reachy Mini motion error: %s", e)

    def _reachy_on_speaking(self) -> None:
        """Called when TTS is playing (HA state: Responding)."""
        # Pause face tracking during speaking - robot will use speaking animation instead
        if self.camera_server is not None:
            try:
                self.camera_server.set_face_tracking_enabled(False)
                _LOGGER.debug("Face tracking paused during speaking")
            except Exception as e:
                _LOGGER.debug("Failed to pause face tracking: %s", e)

        if not self.state.motion_enabled:
            _LOGGER.warning("Motion disabled, skipping speaking animation")
            return
        if not self.state.reachy_mini:
            _LOGGER.warning("No reachy_mini instance, skipping speaking animation")
            return
        if not self.state.motion:
            _LOGGER.warning("No motion controller, skipping speaking animation")
            return

        try:
            _LOGGER.debug("Reachy Mini: Starting speaking animation")
            self.state.motion.on_speaking_start()
        except Exception as e:
            _LOGGER.error("Reachy Mini motion error: %s", e)

    def _reachy_on_idle(self) -> None:
        """Called when returning to idle state (HA state: Idle)."""
        # Disable high-frequency face tracking, switch to adaptive mode
        self._set_conversation_mode(False)

        # Resume face tracking (may have been paused during speaking)
        if self.camera_server is not None:
            try:
                self.camera_server.set_face_tracking_enabled(True)
            except Exception as e:
                _LOGGER.debug("Failed to resume face tracking: %s", e)

        if not self.state.motion_enabled or not self.state.reachy_mini:
            return
        try:
            _LOGGER.debug("Reachy Mini: Idle animation")
            if self.state.motion:
                self.state.motion.on_idle()
        except Exception as e:
            _LOGGER.error("Reachy Mini motion error: %s", e)

    def _set_conversation_mode(self, in_conversation: bool) -> None:
        """Set conversation mode for adaptive face tracking.

        When in conversation, face tracking runs at high frequency.
        When idle, face tracking uses adaptive rate to save CPU.
        """
        if self.camera_server is not None:
            try:
                self.camera_server.set_conversation_mode(in_conversation)
            except Exception as e:
                _LOGGER.debug("Failed to set conversation mode: %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:
            return
        try:
            _LOGGER.debug("Reachy Mini: Timer finished animation")
            if self.state.motion:
                self.state.motion.on_timer_finished()
        except Exception as e:
            _LOGGER.error("Reachy Mini motion error: %s", e)

    def _play_emotion(self, emotion_name: str) -> None:
        """Play an emotion/expression from the emotions library.

        Args:
            emotion_name: Name of the emotion (e.g., "happy1", "sad1", etc.)
        """
        try:
            import requests

            # Get WLAN IP from daemon status
            wlan_ip = "localhost"
            if self.state.reachy_mini is not None:
                try:
                    status = self.state.reachy_mini.client.get_status(wait=False)
                    wlan_ip = status.get('wlan_ip', 'localhost')
                except Exception:
                    wlan_ip = "localhost"

            # Call the emotion playback API
            # Dataset: pollen-robotics/reachy-mini-emotions-library
            base_url = f"http://{wlan_ip}:8000/api/move/play/recorded-move-dataset"
            dataset = "pollen-robotics/reachy-mini-emotions-library"
            url = f"{base_url}/{dataset}/{emotion_name}"

            response = requests.post(url, timeout=5)
            if response.status_code == 200:
                result = response.json()
                move_uuid = result.get('uuid')
                _LOGGER.info(f"Playing emotion: {emotion_name} (uuid={move_uuid})")
            else:
                _LOGGER.warning(f"Failed to play emotion {emotion_name}: HTTP {response.status_code}")

        except Exception as e:
            _LOGGER.error(f"Error playing emotion {emotion_name}: {e}")