Skip to content

Commit

Permalink
improve_typing
Browse files Browse the repository at this point in the history
  • Loading branch information
@JarbasAl
JarbasAl committed Jul 22, 2024
1 parent b603751 commit bb456c0
Show file tree
Hide file tree
Showing 10 changed files with 108 additions and 114 deletions.
14 changes: 2 additions & 12 deletions ovos_core/intent_services/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
from collections import namedtuple
from typing import Tuple, Callable
from ovos_bus_client.message import Message
from ovos_bus_client.session import SessionManager
Expand All @@ -31,20 +30,11 @@
from ovos_utils.log import LOG, deprecated, log_deprecation
from ovos_utils.metrics import Stopwatch
from ovos_workshop.intents import open_intent_envelope

# Intent match response tuple containing
# intent_service: Name of the service that matched the intent
# intent_type: intent name (used to call intent handler over the message bus)
# intent_data: data provided by the intent match
# skill_id: the skill this handler belongs to
IntentMatch = namedtuple('IntentMatch',
['intent_service', 'intent_type',
'intent_data', 'skill_id', 'utterance']
)
from ovos_plugin_manager.templates.pipeline import IntentMatch


class IntentService:
"""Mycroft intent service. parses utterances using a variety of systems.
"""OVOS intent service. parses utterances using a variety of systems.
The intent service also provides the internal API for registering and
querying the intent service.
Expand Down
5 changes: 3 additions & 2 deletions ovos_core/intent_services/adapt_service.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,8 @@
from adapt.engine import IntentDeterminationEngine
from ovos_config.config import Configuration
from ovos_bus_client.message import Message
import ovos_core.intent_services

from ovos_plugin_manager.templates.pipeline import IntentMatch
from ovos_bus_client.session import IntentContextManager as ContextManager, \
SessionManager
from ovos_utils import flatten_list
Expand Down Expand Up @@ -248,7 +249,7 @@ def take_best(intent, utt):
sess.context.update_context(ents)

skill_id = best_intent['intent_type'].split(":")[0]
ret = ovos_core.intent_services.IntentMatch(
ret = IntentMatch(
'Adapt', best_intent['intent_type'], best_intent, skill_id,
best_intent['utterance']
)
Expand Down
16 changes: 8 additions & 8 deletions ovos_core/intent_services/commonqa_service.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,17 +2,17 @@
from dataclasses import dataclass
from os.path import dirname
from threading import Event
from typing import Dict
from typing import Dict, Optional

from ovos_classifiers.opm.heuristics import BM25MultipleChoiceSolver

import ovos_core.intent_services
from ovos_bus_client.message import Message
from ovos_bus_client.session import SessionManager
from ovos_config.config import Configuration
from ovos_utils import flatten_list
from ovos_utils.log import LOG
from ovos_workshop.app import OVOSAbstractApplication
from ovos_plugin_manager.templates.pipeline import IntentMatch


@dataclass
Expand Down Expand Up @@ -76,7 +76,7 @@ def is_question_like(self, utterance: str, lang: str):
# require a "question word"
return self.voc_match(utterance, "QuestionWord", lang)

def match(self, utterances: str, lang: str, message: Message):
def match(self, utterances: str, lang: str, message: Message) -> Optional[IntentMatch]:
"""
Send common query request and select best response
Expand Down Expand Up @@ -114,11 +114,11 @@ def match(self, utterances: str, lang: str, message: Message):
message.data["utterance"] = utterance
answered, skill_id = self.handle_question(message)
if answered:
match = ovos_core.intent_services.IntentMatch(intent_service='CommonQuery',
intent_type=True,
intent_data={},
skill_id=skill_id,
utterance=utterance)
match = IntentMatch(intent_service='CommonQuery',
intent_type=True,
intent_data={},
skill_id=skill_id,
utterance=utterance)
break
return match

Expand Down
16 changes: 8 additions & 8 deletions ovos_core/intent_services/converse_service.py
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import time
from threading import Event

import ovos_core.intent_services
from ovos_plugin_manager.templates.pipeline import IntentMatch
from ovos_bus_client.message import Message
from ovos_bus_client.session import SessionManager, UtteranceState
from ovos_bus_client.util import get_message_lang
Expand Down Expand Up @@ -311,7 +311,7 @@ def converse(self, utterances, skill_id, lang, message):
f'increasing "max_skill_runtime" in mycroft.conf might help alleviate this issue')
return False

def converse_with_skills(self, utterances, lang, message):
def converse_with_skills(self, utterances, lang, message) -> Optional[IntentMatch]:
"""Give active skills a chance at the utterance
Args:
Expand All @@ -335,12 +335,12 @@ def converse_with_skills(self, utterances, lang, message):
continue
if self.converse(utterances, skill_id, lang, message):
state = session.utterance_states.get(skill_id, UtteranceState.INTENT)
return ovos_core.intent_services.IntentMatch(intent_service='Converse',
intent_type=state != UtteranceState.RESPONSE,
# intent_type == True -> emit "ovos.utterance.handled"
intent_data={},
skill_id=skill_id,
utterance=utterances[0])
return IntentMatch(intent_service='Converse',
intent_type=state != UtteranceState.RESPONSE,
# intent_type == True -> emit "ovos.utterance.handled"
intent_data={},
skill_id=skill_id,
utterance=utterances[0])
return None

def handle_get_response_enable(self, message):
Expand Down
22 changes: 12 additions & 10 deletions ovos_core/intent_services/fallback_service.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,13 @@
"""Intent service for Mycroft's fallback system."""
import operator
from collections import namedtuple
from typing import Optional

import time
from ovos_config import Configuration

import ovos_core.intent_services

from ovos_plugin_manager.templates.pipeline import IntentMatch
from ovos_utils import flatten_list
from ovos_utils.log import LOG
from ovos_bus_client.session import SessionManager
Expand Down Expand Up @@ -157,7 +159,7 @@ def attempt_fallback(self, utterances, skill_id, lang, message):
f'increasing "max_skill_runtime" in mycroft.conf might help alleviate this issue')
return False

def _fallback_range(self, utterances, lang, message, fb_range):
def _fallback_range(self, utterances, lang, message, fb_range) -> Optional[IntentMatch]:
"""Send fallback request for a specified priority range.
Args:
Expand Down Expand Up @@ -187,24 +189,24 @@ def _fallback_range(self, utterances, lang, message, fb_range):
continue
result = self.attempt_fallback(utterances, skill_id, lang, message)
if result:
return ovos_core.intent_services.IntentMatch(intent_service='Fallback',
intent_type=None,
intent_data={},
skill_id=skill_id,
utterance=utterances[0])
return IntentMatch(intent_service='Fallback',
intent_type=None,
intent_data={},
skill_id=skill_id,
utterance=utterances[0])
return None

def high_prio(self, utterances, lang, message):
def high_prio(self, utterances, lang, message) -> Optional[IntentMatch]:
"""Pre-padatious fallbacks."""
return self._fallback_range(utterances, lang, message,
FallbackRange(0, 5))

def medium_prio(self, utterances, lang, message):
def medium_prio(self, utterances, lang, message) -> Optional[IntentMatch]:
"""General fallbacks."""
return self._fallback_range(utterances, lang, message,
FallbackRange(5, 90))

def low_prio(self, utterances, lang, message):
def low_prio(self, utterances, lang, message) -> Optional[IntentMatch]:
"""Low prio fallbacks with general matching such as chat-bot."""
return self._fallback_range(utterances, lang, message,
FallbackRange(90, 101))
Expand Down
114 changes: 57 additions & 57 deletions ovos_core/intent_services/ocp_service.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
from threading import Lock, RLock
from typing import List, Tuple, Optional, Union

import ovos_core.intent_services
from ovos_plugin_manager.templates.pipeline import IntentMatch
from ovos_workshop.app import OVOSAbstractApplication

try:
Expand Down Expand Up @@ -350,7 +350,7 @@ def handle_player_state_update(self, message: Message):
self.update_player_proxy(player)

# pipeline
def match_high(self, utterances: List[str], lang: str, message: Message = None):
def match_high(self, utterances: List[str], lang: str, message: Message = None) -> Optional[IntentMatch]:
""" exact matches only, handles playback control
recommended after high confidence intents pipeline stage """
if lang not in self.intent_matchers:
Expand Down Expand Up @@ -385,13 +385,13 @@ def match_high(self, utterances: List[str], lang: str, message: Message = None):
else:
return None

return ovos_core.intent_services.IntentMatch(intent_service="OCP_intents",
intent_type=f'ocp:{match["name"]}',
intent_data=match,
skill_id=OCP_ID,
utterance=utterance)
return IntentMatch(intent_service="OCP_intents",
intent_type=f'ocp:{match["name"]}',
intent_data=match,
skill_id=OCP_ID,
utterance=utterance)

def match_medium(self, utterances: List[str], lang: str, message: Message = None):
def match_medium(self, utterances: List[str], lang: str, message: Message = None) -> Optional[IntentMatch]:
""" match a utterance via classifiers,
recommended before common_qa pipeline stage"""
utterance = utterances[0].lower()
Expand All @@ -410,17 +410,17 @@ def match_medium(self, utterances: List[str], lang: str, message: Message = None
# extract the query string
query = self.remove_voc(utterance, "Play", lang).strip()

return ovos_core.intent_services.IntentMatch(intent_service="OCP_media",
intent_type=f"ocp:play",
intent_data={"media_type": media_type,
"entities": ents,
"query": query,
"is_ocp_conf": bconf,
"conf": confidence},
skill_id=OCP_ID,
utterance=utterance)

def match_fallback(self, utterances: List[str], lang: str, message: Message = None):
return IntentMatch(intent_service="OCP_media",
intent_type=f"ocp:play",
intent_data={"media_type": media_type,
"entities": ents,
"query": query,
"is_ocp_conf": bconf,
"conf": confidence},
skill_id=OCP_ID,
utterance=utterance)

def match_fallback(self, utterances: List[str], lang: str, message: Message = None) -> Optional[IntentMatch]:
""" match an utterance via presence of known OCP keywords,
recommended before fallback_low pipeline stage"""
utterance = utterances[0].lower()
Expand All @@ -437,39 +437,39 @@ def match_fallback(self, utterances: List[str], lang: str, message: Message = No
# extract the query string
query = self.remove_voc(utterance, "Play", lang).strip()

return ovos_core.intent_services.IntentMatch(intent_service="OCP_fallback",
intent_type=f"ocp:play",
intent_data={"media_type": media_type,
"entities": ents,
"query": query,
"conf": float(confidence)},
skill_id=OCP_ID,
utterance=utterance)
return IntentMatch(intent_service="OCP_fallback",
intent_type=f"ocp:play",
intent_data={"media_type": media_type,
"entities": ents,
"query": query,
"conf": float(confidence)},
skill_id=OCP_ID,
utterance=utterance)

def _process_play_query(self, utterance: str, lang: str, match: dict = None,
message: Optional[Message] = None):
message: Optional[Message] = None) -> Optional[IntentMatch]:

match = match or {}
player = self.get_player(message)
# if media is currently paused, empty string means "resume playback"
if player.player_state == PlayerState.PAUSED and \
self._should_resume(utterance, lang, message=message):
return ovos_core.intent_services.IntentMatch(intent_service="OCP_intents",
intent_type=f"ocp:resume",
intent_data=match,
skill_id=OCP_ID,
utterance=utterance)
return IntentMatch(intent_service="OCP_intents",
intent_type=f"ocp:resume",
intent_data=match,
skill_id=OCP_ID,
utterance=utterance)

if not utterance:
# user just said "play", we are missing the search query
phrase = self.get_response("play.what", num_retries=2)
if not phrase:
# let the error intent handler take action
return ovos_core.intent_services.IntentMatch(intent_service="OCP_intents",
intent_type=f"ocp:search_error",
intent_data=match,
skill_id=OCP_ID,
utterance=utterance)
return IntentMatch(intent_service="OCP_intents",
intent_type=f"ocp:search_error",
intent_data=match,
skill_id=OCP_ID,
utterance=utterance)

sess = SessionManager.get(message)
# if a skill was explicitly requested, search it first
Expand All @@ -489,18 +489,18 @@ def _process_play_query(self, utterance: str, lang: str, match: dict = None,

ents = OCPFeaturizer.extract_entities(utterance)

return ovos_core.intent_services.IntentMatch(intent_service="OCP_intents",
intent_type=f"ocp:play",
intent_data={"media_type": media_type,
"query": query,
"entities": ents,
"skills": valid_skills,
"conf": match["conf"],
"media_conf": float(conf),
# "results": results,
"lang": lang},
skill_id=OCP_ID,
utterance=utterance)
return IntentMatch(intent_service="OCP_intents",
intent_type=f"ocp:play",
intent_data={"media_type": media_type,
"query": query,
"entities": ents,
"skills": valid_skills,
"conf": match["conf"],
"media_conf": float(conf),
# "results": results,
"lang": lang},
skill_id=OCP_ID,
utterance=utterance)

# bus api
def handle_search_query(self, message: Message):
Expand Down Expand Up @@ -1070,7 +1070,7 @@ def _handle_legacy_audio_end(self, message: Message):
############
# Legacy Mycroft CommonPlay skills

def match_legacy(self, utterances: List[str], lang: str, message: Message = None):
def match_legacy(self, utterances: List[str], lang: str, message: Message = None) -> Optional[IntentMatch]:
""" match legacy mycroft common play skills (must import from deprecated mycroft module)
not recommended, legacy support only
Expand All @@ -1089,12 +1089,12 @@ def match_legacy(self, utterances: List[str], lang: str, message: Message = None
if match["name"] == "play":
LOG.info(f"Legacy Mycroft CommonPlay match: {match}")
utterance = match["entities"].pop("query")
return ovos_core.intent_services.IntentMatch(intent_service="OCP_media",
intent_type=f"ocp:legacy_cps",
intent_data={"query": utterance,
"conf": 0.7},
skill_id=OCP_ID,
utterance=utterance)
return IntentMatch(intent_service="OCP_media",
intent_type=f"ocp:legacy_cps",
intent_data={"query": utterance,
"conf": 0.7},
skill_id=OCP_ID,
utterance=utterance)

def handle_legacy_cps(self, message: Message):
"""intent handler for legacy CPS matches"""
Expand Down
Loading

0 comments on commit bb456c0

Please sign in to comment.