conversation.py

Go to the documentation of this file.

"""Support for Wyoming intent recognition services."""
 
import logging
 
from wyoming.asr import Transcript
from wyoming.client import AsyncTcpClient
from wyoming.handle import Handled, NotHandled
from wyoming.info import HandleProgram, IntentProgram
from wyoming.intent import Intent, NotRecognized
 
from homeassistant.components import conversation
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import intent
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import ulid
 
from .const import DOMAIN
from .data import WyomingService
from .error import WyomingError
from .models import DomainDataItem
 
_LOGGER = logging.getLogger(__name__)
 
 
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up Wyoming conversation."""
    item: DomainDataItem = hass.data[DOMAIN][config_entry.entry_id]
    async_add_entities(
        [
            WyomingConversationEntity(config_entry, item.service),
        ]
    )
 
 
class WyomingConversationEntity(
    conversation.ConversationEntity, conversation.AbstractConversationAgent
):
    """Wyoming conversation agent."""
 
    _attr_has_entity_name = True
 
    def __init__(
        self,
        config_entry: ConfigEntry,
        service: WyomingService,
    ) -> None:
        """Set up provider."""
        super().__init__()
 
        self.serviceservice = service
 
        self._intent_service_intent_service: IntentProgram | None = None
        self._handle_service_handle_service: HandleProgram | None = None
 
        for maybe_intent in self.serviceservice.info.intent:
            if maybe_intent.installed:
                self._intent_service_intent_service = maybe_intent
                break
 
        for maybe_handle in self.serviceservice.info.handle:
            if maybe_handle.installed:
                self._handle_service_handle_service = maybe_handle
                break
 
        model_languages: set[str] = set()
 
        if self._intent_service_intent_service is not None:
            for intent_model in self._intent_service_intent_service.models:
                if intent_model.installed:
                    model_languages.update(intent_model.languages)
 
            self._attr_name_attr_name = self._intent_service_intent_service.name
            self._attr_supported_features_attr_supported_features_attr_supported_features = (
                conversation.ConversationEntityFeature.CONTROL
            )
        elif self._handle_service_handle_service is not None:
            for handle_model in self._handle_service_handle_service.models:
                if handle_model.installed:
                    model_languages.update(handle_model.languages)
 
            self._attr_name_attr_name = self._handle_service_handle_service.name
 
        self._supported_languages_supported_languages = list(model_languages)
        self._attr_unique_id_attr_unique_id = f"{config_entry.entry_id}-conversation"
 
    @property
    def supported_languages(self) -> list[str]:
        """Return a list of supported languages."""
        return self._supported_languages_supported_languages
 
    async def async_process(
        self, user_input: conversation.ConversationInput
    ) -> conversation.ConversationResult:
        """Process a sentence."""
        conversation_id = user_input.conversation_id or ulid.ulid_now()
        intent_response = intent.IntentResponse(language=user_input.language)
 
        try:
            async with AsyncTcpClient(self.serviceservice.host, self.serviceservice.port) as client:
                await client.write_event(
                    Transcript(
                        user_input.text, context={"conversation_id": conversation_id}
                    ).event()
                )
 
                while True:
                    event = await client.read_event()
                    if event is None:
                        _LOGGER.debug("Connection lost")
                        intent_response.async_set_error(
                            intent.IntentResponseErrorCode.UNKNOWN,
                            "Connection to service was lost",
                        )
                        return conversation.ConversationResult(
                            response=intent_response,
                            conversation_id=user_input.conversation_id,
                        )
 
                    if Intent.is_type(event.type):
                        # Success
                        recognized_intent = Intent.from_event(event)
                        _LOGGER.debug("Recognized intent: %s", recognized_intent)
 
                        intent_type = recognized_intent.name
                        intent_slots = {
                            e.name: {"value": e.value}
                            for e in recognized_intent.entities
                        }
                        intent_response = await intent.async_handle(
                            self.hasshass,
                            DOMAIN,
                            intent_type,
                            intent_slots,
                            text_input=user_input.text,
                            language=user_input.language,
                        )
 
                        if (not intent_response.speech) and recognized_intent.text:
                            intent_response.async_set_speech(recognized_intent.text)
 
                        break
 
                    if NotRecognized.is_type(event.type):
                        not_recognized = NotRecognized.from_event(event)
                        intent_response.async_set_error(
                            intent.IntentResponseErrorCode.NO_INTENT_MATCH,
                            not_recognized.text,
                        )
                        break
 
                    if Handled.is_type(event.type):
                        # Success
                        handled = Handled.from_event(event)
                        intent_response.async_set_speech(handled.text)
                        break
 
                    if NotHandled.is_type(event.type):
                        not_handled = NotHandled.from_event(event)
                        intent_response.async_set_error(
                            intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
                            not_handled.text,
                        )
                        break
 
        except (OSError, WyomingError) as err:
            _LOGGER.exception("Unexpected error while communicating with service")
            intent_response.async_set_error(
                intent.IntentResponseErrorCode.UNKNOWN,
                f"Error communicating with service: {err}",
            )
            return conversation.ConversationResult(
                response=intent_response,
                conversation_id=user_input.conversation_id,
            )
        except intent.IntentError as err:
            _LOGGER.exception("Unexpected error while handling intent")
            intent_response.async_set_error(
                intent.IntentResponseErrorCode.FAILED_TO_HANDLE,
                f"Error handling intent: {err}",
            )
            return conversation.ConversationResult(
                response=intent_response,
                conversation_id=user_input.conversation_id,
            )
 
        # Success
        return conversation.ConversationResult(
            response=intent_response, conversation_id=conversation_id
        )