fabaff/platypush
1
0
Fork 0
forked from platypush/platypush
Code Issues Pull requests Projects Releases Packages Wiki Activity

Refactored AssistantEvent.

Browse source 
`AssistantEvent.assistant` is now modelled as an opaque object that
behaves the following way:

- The underlying plugin name is saved under `event.args['_assistant']`.

- `event.assistant` is a property that returns the assistant instance
  via `get_plugin`.

- `event.assistant` is reported as a string (plugin qualified name) upon
  event dump.

This allows event hooks to easily use `event.assistant` to interact with
the underlying assistant and easily modify the conversation flow, while
event hook conditions can still be easily modelled as equality
operations between strings.
This commit is contained in:
Fabio Manganiello 2024-04-09 00:15:51 +02:00
parent a670f01647
commit 8378bee7c6
4 changed files with 108 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
platypush/message/event/__init__.py
View file

@ -257,26 +257,29 @@ class Event(Message):
return result
def as_dict(self):
"""
Converts the event into a dictionary
"""
args = copy.deepcopy(self.args)
flatten(args)
return {
'type': 'event',
'target': self.target,
'origin': self.origin if hasattr(self, 'origin') else None,
'id': self.id if hasattr(self, 'id') else None,
'_timestamp': self.timestamp,
'args': {'type': self.type, **args},
}
def __str__(self):
"""
Overrides the str() operator and converts
the message into a UTF-8 JSON string
"""
args = copy.deepcopy(self.args)
flatten(args)
return json.dumps(
{
'type': 'event',
'target': self.target,
'origin': self.origin if hasattr(self, 'origin') else None,
'id': self.id if hasattr(self, 'id') else None,
'_timestamp': self.timestamp,
'args': {'type': self.type, **args},
},
cls=self.Encoder,
)
return json.dumps(self.as_dict(), cls=self.Encoder)
@dataclass

48
platypush/message/event/assistant/__init__.py
View file

@ -1,27 +1,53 @@
import re
import sys
from typing import Optional
from typing import Optional, Union
from platypush.context import get_plugin
from platypush.message.event import Event
from platypush.plugins.assistant import AssistantPlugin
from platypush.utils import get_plugin_name_by_class
class AssistantEvent(Event):
"""Base class for assistant events"""
def __init__(self, *args, assistant: Optional[str] = None, **kwargs):
def __init__(
self, *args, assistant: Optional[Union[str, AssistantPlugin]] = None, **kwargs
):
"""
:param assistant: Name of the assistant plugin that triggered the event.
"""
super().__init__(*args, assistant=assistant, **kwargs)
assistant = assistant or kwargs.get('assistant')
if assistant:
assistant = (
assistant
if isinstance(assistant, str)
else get_plugin_name_by_class(assistant.__class__)
)
kwargs['_assistant'] = assistant
super().__init__(*args, **kwargs)
@property
def _assistant(self):
return (
get_plugin(self.args.get('assistant'))
if self.args.get('assistant')
else None
)
def assistant(self) -> Optional[AssistantPlugin]:
assistant = self.args.get('_assistant')
if not assistant:
return None
return get_plugin(assistant)
def as_dict(self):
evt_dict = super().as_dict()
evt_args = {**evt_dict['args']}
assistant = evt_args.pop('_assistant', None)
if assistant:
evt_args['assistant'] = assistant
return {
**evt_dict,
'args': evt_args,
}
class ConversationStartEvent(AssistantEvent):
@ -95,8 +121,8 @@ class SpeechRecognizedEvent(AssistantEvent):
"""
result = super().matches_condition(condition)
if result.is_match and self._assistant and 'phrase' in condition.args:
self._assistant.stop_conversation()
if result.is_match and self.assistant and 'phrase' in condition.args:
self.assistant.stop_conversation()
return result

6
platypush/plugins/__init__.py
View file

@ -122,6 +122,12 @@ class Plugin(EventGenerator, ExtensionWithManifest): # lgtm [py/missing-call-to
assert entities, 'entities plugin not initialized'
return entities
def __str__(self):
"""
:return: The qualified name of the plugin.
"""
return get_plugin_name_by_class(self.__class__)
def run(self, method, *args, **kwargs):
assert (
method in self.registered_actions

70
platypush/plugins/assistant/__init__.py
View file

@ -8,24 +8,7 @@ from typing import Any, Collection, Dict, Optional, Type
from platypush.context import get_bus, get_plugin
from platypush.entities.assistants import Assistant
from platypush.entities.managers.assistants import AssistantEntityManager
from platypush.message.event.assistant import (
AlarmEndEvent,
AlarmStartedEvent,
AlertEndEvent,
AlertStartedEvent,
AssistantEvent,
ConversationEndEvent,
ConversationStartEvent,
ConversationTimeoutEvent,
HotwordDetectedEvent,
MicMutedEvent,
MicUnmutedEvent,
NoResponseEvent,
ResponseEvent,
SpeechRecognizedEvent,
TimerEndEvent,
TimerStartedEvent,
)
from platypush.message.event import Event as AppEvent
from platypush.plugins import Plugin, action
from platypush.utils import get_plugin_name_by_class
@ -182,6 +165,17 @@ class AssistantPlugin(Plugin, AssistantEntityManager, ABC):
self.publish_entities([self])
return asdict(self._state)
@action
def render_response(self, text: str, *_, **__):
"""
Render a response text as audio over the configured TTS plugin.
:param text: Text to render.
"""
self._on_response_render_start(text)
self._render_response(text)
self._on_response_render_end()
def _get_tts_plugin(self):
if not self.tts_plugin:
return None
@ -201,11 +195,13 @@ class AssistantPlugin(Plugin, AssistantEntityManager, ABC):
audio.play(self._conversation_start_sound)
def _send_event(self, event_type: Type[AssistantEvent], **kwargs):
def _send_event(self, event_type: Type[AppEvent], **kwargs):
self.publish_entities([self])
get_bus().post(event_type(assistant=self._plugin_name, **kwargs))
def _on_conversation_start(self):
from platypush.message.event.assistant import ConversationStartEvent
self._last_response = None
self._last_query = None
self._conversation_running.set()
@ -213,66 +209,98 @@ class AssistantPlugin(Plugin, AssistantEntityManager, ABC):
self._play_conversation_start_sound()
def _on_conversation_end(self):
from platypush.message.event.assistant import ConversationEndEvent
self._conversation_running.clear()
self._send_event(ConversationEndEvent)
def _on_conversation_timeout(self):
from platypush.message.event.assistant import ConversationTimeoutEvent
self._last_response = None
self._last_query = None
self._conversation_running.clear()
self._send_event(ConversationTimeoutEvent)
def _on_no_response(self):
from platypush.message.event.assistant import NoResponseEvent
self._last_response = None
self._conversation_running.clear()
self._send_event(NoResponseEvent)
def _on_reponse_rendered(self, text: Optional[str]):
def _on_response_render_start(self, text: Optional[str]):
from platypush.message.event.assistant import ResponseEvent
self._last_response = text
self._send_event(ResponseEvent, response_text=text)
tts = self._get_tts_plugin()
def _render_response(self, text: Optional[str]):
tts = self._get_tts_plugin()
if tts and text:
self.stop_conversation()
tts.say(text=text, **self.tts_plugin_args)
def _on_response_render_end(self):
pass
def _on_hotword_detected(self, hotword: Optional[str]):
from platypush.message.event.assistant import HotwordDetectedEvent
self._send_event(HotwordDetectedEvent, hotword=hotword)
def _on_speech_recognized(self, phrase: Optional[str]):
from platypush.message.event.assistant import SpeechRecognizedEvent
phrase = (phrase or '').lower().strip()
self._last_query = phrase
self._send_event(SpeechRecognizedEvent, phrase=phrase)
def _on_alarm_start(self):
from platypush.message.event.assistant import AlarmStartedEvent
self._cur_alert_type = AlertType.ALARM
self._send_event(AlarmStartedEvent)
def _on_alarm_end(self):
from platypush.message.event.assistant import AlarmEndEvent
self._cur_alert_type = None
self._send_event(AlarmEndEvent)
def _on_timer_start(self):
from platypush.message.event.assistant import TimerStartedEvent
self._cur_alert_type = AlertType.TIMER
self._send_event(TimerStartedEvent)
def _on_timer_end(self):
from platypush.message.event.assistant import TimerEndEvent
self._cur_alert_type = None
self._send_event(TimerEndEvent)
def _on_alert_start(self):
from platypush.message.event.assistant import AlertStartedEvent
self._cur_alert_type = AlertType.ALERT
self._send_event(AlertStartedEvent)
def _on_alert_end(self):
from platypush.message.event.assistant import AlertEndEvent
self._cur_alert_type = None
self._send_event(AlertEndEvent)
def _on_mute(self):
from platypush.message.event.assistant import MicMutedEvent
self._is_muted = True
self._send_event(MicMutedEvent)
def _on_unmute(self):
from platypush.message.event.assistant import MicUnmutedEvent
self._is_muted = False
self._send_event(MicUnmutedEvent)