248 lines
8.3 KiB
Python
248 lines
8.3 KiB
Python
"""Sentence platform for nodered."""
|
|
|
|
import asyncio
|
|
from enum import Enum
|
|
import logging
|
|
from typing import Any
|
|
|
|
from hassil.expression import Sentence
|
|
from hassil.recognize import RecognizeResult
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.websocket_api.connection import ActiveConnection
|
|
from homeassistant.components.websocket_api.decorators import (
|
|
async_response,
|
|
require_admin,
|
|
websocket_command,
|
|
)
|
|
from homeassistant.components.websocket_api.messages import (
|
|
error_message,
|
|
event_message,
|
|
result_message,
|
|
)
|
|
from homeassistant.const import CONF_ID, CONF_TYPE
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers import config_validation as cv
|
|
|
|
from .const import CONF_SERVER_ID
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
response_futures: dict[str, asyncio.Future] = {}
|
|
response_lock = asyncio.Lock()
|
|
|
|
|
|
class ResponseType(Enum):
|
|
"""Enumeration for sentence response types used in Node-RED triggers."""
|
|
|
|
FIXED = "fixed"
|
|
DYNAMIC = "dynamic"
|
|
|
|
|
|
@require_admin
|
|
@websocket_command(
|
|
{
|
|
vol.Required(CONF_TYPE): "nodered/sentence",
|
|
vol.Required(CONF_SERVER_ID): cv.string,
|
|
vol.Required("sentences", default=[]): [cv.string],
|
|
vol.Optional("response", default="Done"): cv.string,
|
|
vol.Optional("response_type", default=ResponseType.FIXED): vol.Coerce(
|
|
ResponseType
|
|
),
|
|
vol.Optional("response_timeout", default=1): cv.positive_float,
|
|
}
|
|
)
|
|
@async_response
|
|
async def websocket_sentence(
|
|
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
|
|
) -> None:
|
|
"""Create sentence trigger."""
|
|
message_id = msg[CONF_ID]
|
|
sentences = msg["sentences"]
|
|
response = msg["response"]
|
|
response_timeout = msg["response_timeout"]
|
|
response_type = msg["response_type"]
|
|
|
|
@callback
|
|
async def handle_trigger(
|
|
sentence: str,
|
|
result: RecognizeResult | None = None,
|
|
device_id: str | None = None,
|
|
) -> Any:
|
|
"""Handle Sentence trigger.
|
|
|
|
RecognizeResult was added in 2023.8.0
|
|
device_id was added in 2024.4.0.
|
|
"""
|
|
# RecognizeResult in 2024.12 is not serializable,
|
|
# so we need to convert it to a serializable format
|
|
serialized = convert_recognize_result_to_dict(result)
|
|
|
|
_LOGGER.debug("Sentence trigger: %s", sentence)
|
|
connection.send_message(
|
|
event_message(
|
|
message_id,
|
|
{
|
|
"data": {
|
|
"sentence": sentence,
|
|
"result": serialized,
|
|
"deviceId": device_id,
|
|
"responseId": message_id,
|
|
}
|
|
},
|
|
)
|
|
)
|
|
|
|
if response_type == ResponseType.DYNAMIC:
|
|
async with response_lock:
|
|
if message_id not in response_futures:
|
|
response_futures[message_id] = hass.loop.create_future()
|
|
|
|
try:
|
|
result = await asyncio.wait_for(
|
|
response_futures[message_id], response_timeout
|
|
)
|
|
_LOGGER.debug(
|
|
"Sentence response %s received with response: %s",
|
|
message_id,
|
|
result,
|
|
)
|
|
except TimeoutError:
|
|
_LOGGER.debug(
|
|
"Timeout reached for sentence response %s. Continuing...",
|
|
message_id,
|
|
)
|
|
# Remove the event from the dictionary after timeout
|
|
del response_futures[message_id]
|
|
else:
|
|
return result
|
|
|
|
return response
|
|
|
|
def remove_trigger() -> None:
|
|
"""Remove sentence trigger."""
|
|
|
|
async def _remove_future() -> None:
|
|
async with response_lock:
|
|
response_futures.pop(message_id, None)
|
|
|
|
hass.async_create_task(_remove_future())
|
|
_remove_trigger()
|
|
_LOGGER.info("Sentence trigger removed: %s", sentences)
|
|
|
|
try:
|
|
from homeassistant.components.conversation.agent_manager import ( # noqa: PLC0415
|
|
get_agent_manager,
|
|
)
|
|
from homeassistant.components.conversation.trigger import ( # noqa: PLC0415
|
|
TriggerDetails,
|
|
)
|
|
|
|
manager = get_agent_manager(hass)
|
|
if manager is None:
|
|
raise ValueError("Conversation integration not loaded") # noqa: TRY301
|
|
|
|
_remove_trigger = manager.register_trigger(
|
|
TriggerDetails(sentences, handle_trigger) # type: ignore[arg-type]
|
|
)
|
|
except ImportError:
|
|
# Fallback for Home Assistant versions before 2025.10.0
|
|
from homeassistant.components.conversation.default_agent import ( # noqa: PLC0415
|
|
DATA_DEFAULT_ENTITY, # type: ignore[import]
|
|
DefaultAgent,
|
|
)
|
|
|
|
default_agent = hass.data[DATA_DEFAULT_ENTITY]
|
|
if not isinstance(default_agent, DefaultAgent):
|
|
raise TypeError("default_agent is not an instance of DefaultAgent") # noqa: B904
|
|
|
|
_remove_trigger = default_agent.register_trigger(sentences, handle_trigger) # type: ignore[arg-type]
|
|
except ValueError as err:
|
|
connection.send_message(error_message(message_id, "value_error", str(err)))
|
|
return
|
|
except TypeError as err:
|
|
connection.send_message(error_message(message_id, "type_error", str(err)))
|
|
return
|
|
except RuntimeError as err:
|
|
connection.send_message(error_message(message_id, "runtime_error", str(err)))
|
|
return
|
|
|
|
if response_type == ResponseType.FIXED:
|
|
_LOGGER.info("Sentence trigger created: %s", sentences)
|
|
else:
|
|
_LOGGER.info(
|
|
"Sentence trigger created: %s with dynamic response #%s and "
|
|
"timeout of %s seconds",
|
|
sentences,
|
|
message_id,
|
|
response_timeout,
|
|
)
|
|
connection.subscriptions[msg[CONF_ID]] = remove_trigger
|
|
connection.send_message(result_message(message_id))
|
|
|
|
|
|
@require_admin
|
|
@websocket_command(
|
|
{
|
|
vol.Required(CONF_TYPE): "nodered/sentence_response",
|
|
vol.Required("response_id"): cv.positive_int,
|
|
vol.Required("response"): cv.string,
|
|
}
|
|
)
|
|
@async_response
|
|
async def websocket_sentence_response(
|
|
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
|
|
) -> None:
|
|
"""Send response to sentence trigger."""
|
|
message_id = msg[CONF_ID]
|
|
response_id = msg["response_id"]
|
|
response = msg["response"]
|
|
|
|
async with response_lock:
|
|
# Print the current response_futures keys (ids)
|
|
if response_id in response_futures:
|
|
# Set the message as the result
|
|
response_futures[response_id].set_result(response)
|
|
# Remove the event from the dictionary after dispatching
|
|
del response_futures[response_id]
|
|
_LOGGER.info("Sentence response received: %s", response)
|
|
connection.send_message(result_message(message_id))
|
|
else:
|
|
message = f"Sentence response not found for id: {response_id}"
|
|
_LOGGER.warning(message)
|
|
connection.send_message(
|
|
error_message(message_id, "sentence_response_not_found", message),
|
|
)
|
|
|
|
|
|
def convert_recognize_result_to_dict(result: Any) -> dict:
|
|
"""Serialize a RecognizeResult object into a JSON-serializable dictionary."""
|
|
|
|
def serialize(obj: Any) -> Any:
|
|
if isinstance(obj, Sentence):
|
|
# Custom serialization for Sentence
|
|
return {
|
|
"text": obj.text,
|
|
"pattern": (
|
|
obj.pattern.pattern
|
|
if hasattr(obj, "pattern") and obj.pattern is not None
|
|
else None
|
|
),
|
|
}
|
|
if hasattr(obj, "__dict__"):
|
|
# For objects with attributes, serialize attributes
|
|
return {key: serialize(value) for key, value in vars(obj).items()}
|
|
if isinstance(obj, list):
|
|
# Recursively handle lists
|
|
return [serialize(item) for item in obj]
|
|
if isinstance(obj, dict):
|
|
# Recursively handle dictionaries
|
|
return {key: serialize(value) for key, value in obj.items()}
|
|
if isinstance(obj, (int, float, str, type(None))):
|
|
# Primitive types are already serializable
|
|
return obj
|
|
# Fallback for non-serializable types
|
|
return str(obj)
|
|
|
|
return serialize(result)
|