Nouvelle page
"""Fallback Conversation Agent"""
from __future__ import annotations
import logging
import requests
from homeassistant.components import conversation
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.util import ulid
from home_assistant_intents import get_languages
from homeassistant.helpers import (
config_validation as cv,
intent,
)
from .const import (
CONF_DEBUG_LEVEL,
CONF_PRIMARY_AGENT,
CONF_FALLBACK_AGENT,
DEBUG_LEVEL_NO_DEBUG,
DEBUG_LEVEL_LOW_DEBUG,
DEBUG_LEVEL_VERBOSE_DEBUG,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
# hass.data key for agent.
DATA_AGENT = "agent"
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Fallback Conversation Agent from a config entry."""
agent = FallbackConversationAgent(hass, entry)
conversation.async_set_agent(hass, entry, agent)
return True
def post_to_api(url, json):
return requests.post(url, json=json)
class FallbackConversationAgent(conversation.AbstractConversationAgent):
"""Fallback Conversation Agent."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize the agent."""
self.hass = hass
self.entry = entry
@property
def supported_languages(self) -> list[str]:
"""Return a list of supported languages."""
return get_languages()
async def async_process(
self, user_input: conversation.ConversationInput
) -> conversation.ConversationResult:
url = 'http://192.168.122.1:10500'
data = {'text': user_input.text}
response_from_api = await self.hass.async_add_executor_job(post_to_api, url, data) #task.executor(requests.post, )
response = intent.IntentResponse(language=user_input.language)
response.async_set_speech(response_from_api.text)
return conversation.ConversationResult(
conversation_id=None,
response=response
)
"""Config flow for Fallback Conversation integration."""
from __future__ import annotations
import logging
# from types import MappingProxyType
from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, async_get_hass, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.selector import (
ConversationAgentSelector,
ConversationAgentSelectorConfig,
SelectSelector,
SelectSelectorConfig,
SelectOptionDict,
SelectSelectorMode,
)
from .const import (
CONF_DEBUG_LEVEL,
CONF_PRIMARY_AGENT,
CONF_FALLBACK_AGENT,
DEBUG_LEVEL_NO_DEBUG,
DEBUG_LEVEL_LOW_DEBUG,
DEBUG_LEVEL_VERBOSE_DEBUG,
DOMAIN,
DEFAULT_NAME,
DEFAULT_DEBUG_LEVEL,
)
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): str,
vol.Optional(CONF_DEBUG_LEVEL, default=DEFAULT_DEBUG_LEVEL): SelectSelector(
SelectSelectorConfig(
options=[
SelectOptionDict(value=DEBUG_LEVEL_NO_DEBUG, label="No Debug"),
SelectOptionDict(value=DEBUG_LEVEL_LOW_DEBUG, label="Some Debug"),
SelectOptionDict(value=DEBUG_LEVEL_VERBOSE_DEBUG, label="Verbose Debug"),
],
mode=SelectSelectorMode.DROPDOWN
),
),
}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Fallback Agent config flow."""
VERSION = 1
async def async_step_user(self, user_input: dict[str, Any] | None = None) -> FlowResult:
"""Handle the initial step."""
_LOGGER.debug("ConfigFlow::user_input %s", user_input)
if user_input is None:
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
)
return self.async_create_entry(
title=user_input.get(CONF_NAME, DEFAULT_NAME),
data=user_input,
)
@staticmethod
@callback
def async_get_options_flow(config_entry: config_entries.ConfigEntry) -> config_entries.OptionsFlow:
"""Create the options flow."""
return OptionsFlow(config_entry)
class OptionsFlow(config_entries.OptionsFlow):
"""Fallback config flow options handler."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
self._options = dict(config_entry.data)
self._options.update(dict(config_entry.options))
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options."""
if user_input is not None:
self._options.update(user_input)
return self.async_create_entry(
title=user_input.get(CONF_NAME, DEFAULT_NAME),
data=self._options,
)
schema = await self.fallback_config_option_schema(self._options)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(schema),
)
async def fallback_config_option_schema(self, options: dict) -> dict:
"""Return a schema for Fallback options."""
return {
vol.Required(
CONF_DEBUG_LEVEL,
description={"suggested_value": options.get(CONF_DEBUG_LEVEL, DEFAULT_DEBUG_LEVEL)},
default=DEFAULT_DEBUG_LEVEL,
): SelectSelector(
SelectSelectorConfig(
options=[
SelectOptionDict(value=DEBUG_LEVEL_NO_DEBUG, label="No Debug"),
SelectOptionDict(value=DEBUG_LEVEL_LOW_DEBUG, label="Some Debug"),
SelectOptionDict(value=DEBUG_LEVEL_VERBOSE_DEBUG, label="Verbose Debug"),
],
mode=SelectSelectorMode.DROPDOWN
),
),
};
No Comments