grammar: fuzzy wake gate and command matching
word-boundary wake stripping that's lenient on the coined word 'claudedo' (despaced-prefix match) without swallowing the command's spaces. data-driven phrase->action map; number words normalized to digits; 'target' aliases 'switch'. Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
parent
da7c39c4f2
commit
947b30c22e
159
src/claudedo/grammar.py
Normal file
159
src/claudedo/grammar.py
Normal file
@ -0,0 +1,159 @@
|
||||
"""wake-phrase gate + command grammar matching (fuzzy, data-driven).
|
||||
|
||||
the matcher is lenient by design: whisper renders the coined word "claudedo"
|
||||
inconsistently, so wake-phrase detection normalizes case, strips spaces/punctuation,
|
||||
and accepts close variants. number words are normalized to digits before matching.
|
||||
|
||||
flow: transcript -> strip_wake() returns the command remainder (or None if no wake
|
||||
phrase in listen mode) -> match_command() maps the remainder to an Action.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from difflib import SequenceMatcher
|
||||
|
||||
_PUNCT = re.compile(r"[^a-z0-9 ]+")
|
||||
_WS = re.compile(r"\s+")
|
||||
|
||||
_NUMBER_WORDS = {
|
||||
"zero": "0", "oh": "0",
|
||||
"one": "1", "won": "1",
|
||||
"two": "2", "to": "2", "too": "2",
|
||||
"three": "3", "tree": "3",
|
||||
"four": "4", "for": "4", "fore": "4",
|
||||
}
|
||||
|
||||
_INDEX_WORDS = {"1": 1, "2": 2, "3": 3, "4": 4}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Action:
|
||||
"""a matched command: a name plus an optional argument.
|
||||
|
||||
names: yes, no, select, approve, deny, submit, type, mode, switch, cancel.
|
||||
arg carries the select index (int), the literal text for ``type``, the mode for
|
||||
``mode``, or the session short-name for ``switch``.
|
||||
"""
|
||||
|
||||
name: str
|
||||
arg: object = None
|
||||
|
||||
|
||||
def normalize(text: str) -> str:
|
||||
"""lowercase, strip punctuation, collapse whitespace, map number words to digits."""
|
||||
text = text.lower().strip()
|
||||
text = _PUNCT.sub(" ", text)
|
||||
text = _WS.sub(" ", text).strip()
|
||||
if not text:
|
||||
return ""
|
||||
tokens = [_NUMBER_WORDS.get(tok, tok) for tok in text.split(" ")]
|
||||
return " ".join(tokens)
|
||||
|
||||
|
||||
def _ratio(a: str, b: str) -> float:
|
||||
return SequenceMatcher(None, a, b).ratio()
|
||||
|
||||
|
||||
def _wake_variants(phrase: str) -> set[str]:
|
||||
"""spaced and despaced forms of a wake phrase for lenient matching."""
|
||||
norm = normalize(phrase)
|
||||
return {norm, norm.replace(" ", "")}
|
||||
|
||||
|
||||
def strip_wake(transcript: str, wake_phrases: list[str], threshold: float,
|
||||
require_wake: bool) -> str | None:
|
||||
"""return the command remainder after the wake phrase.
|
||||
|
||||
if ``require_wake`` (listen mode) and no wake phrase is found at the start,
|
||||
return None so the daemon discards the utterance. if not required (ptt mode),
|
||||
a leading wake phrase is stripped when present but its absence is fine.
|
||||
|
||||
matches leniently on a despaced prefix (whisper splits/joins the coined word
|
||||
inconsistently) but always slices the remainder on a WORD boundary of the
|
||||
spaced, normalized transcript — so the command portion keeps its spaces.
|
||||
"""
|
||||
norm = normalize(transcript)
|
||||
if not norm:
|
||||
return None if require_wake else ""
|
||||
words = norm.split(" ")
|
||||
|
||||
best_remainder: str | None = None
|
||||
best_score = 0.0
|
||||
for phrase in wake_phrases:
|
||||
variants = _wake_variants(phrase)
|
||||
max_words = phrase.count(" ") + 2
|
||||
for take in range(1, min(max_words, len(words)) + 1):
|
||||
head_despaced = "".join(words[:take])
|
||||
for variant in variants:
|
||||
if not variant:
|
||||
continue
|
||||
score = _ratio(head_despaced, variant)
|
||||
if score >= threshold and score > best_score:
|
||||
best_score = score
|
||||
best_remainder = " ".join(words[take:]).strip()
|
||||
|
||||
if best_remainder is not None:
|
||||
return best_remainder
|
||||
return None if require_wake else norm
|
||||
|
||||
|
||||
def _fuzzy_in(token: str, options: tuple[str, ...], threshold: float) -> bool:
|
||||
return any(_ratio(token, opt) >= threshold for opt in options)
|
||||
|
||||
|
||||
def match_command(remainder: str, threshold: float) -> Action | None:
|
||||
"""map a normalized command remainder to an Action, or None if unrecognized."""
|
||||
remainder = remainder.strip()
|
||||
if not remainder:
|
||||
return None
|
||||
tokens = remainder.split(" ")
|
||||
head = tokens[0]
|
||||
rest = tokens[1:]
|
||||
|
||||
if head in _INDEX_WORDS:
|
||||
return Action("select", _INDEX_WORDS[head])
|
||||
|
||||
if _fuzzy_in(head, ("yes", "yeah", "yep", "yup"), threshold):
|
||||
return Action("yes")
|
||||
if _fuzzy_in(head, ("no", "nope", "nah"), threshold):
|
||||
return Action("no")
|
||||
if _fuzzy_in(head, ("approve", "allow"), threshold):
|
||||
return Action("approve")
|
||||
if _fuzzy_in(head, ("deny", "reject"), threshold):
|
||||
return Action("deny")
|
||||
if _fuzzy_in(head, ("send", "enter", "submit"), threshold):
|
||||
return Action("submit")
|
||||
if _fuzzy_in(head, ("cancel", "escape", "stop"), threshold):
|
||||
return Action("cancel")
|
||||
|
||||
if _fuzzy_in(head, ("select", "option", "choose", "number"), threshold) and rest:
|
||||
if rest[0] in _INDEX_WORDS:
|
||||
return Action("select", _INDEX_WORDS[rest[0]])
|
||||
|
||||
if _fuzzy_in(head, ("type", "dictate", "write"), threshold):
|
||||
text = " ".join(rest).strip()
|
||||
return Action("type", text) if text else None
|
||||
|
||||
if _fuzzy_in(head, ("mode",), threshold) and rest:
|
||||
if _fuzzy_in(rest[0], ("ptt",), threshold) or "push" in rest[0]:
|
||||
return Action("mode", "ptt")
|
||||
if _fuzzy_in(rest[0], ("listen",), threshold):
|
||||
return Action("mode", "listen")
|
||||
return None
|
||||
|
||||
if _fuzzy_in(head, ("switch", "target"), threshold) and rest:
|
||||
name = "".join(rest)
|
||||
return Action("switch", name) if name else None
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def parse(transcript: str, wake_phrases: list[str], threshold: float,
|
||||
require_wake: bool) -> Action | None:
|
||||
"""full parse: wake gate then command match. None means discard."""
|
||||
remainder = strip_wake(transcript, wake_phrases, threshold, require_wake)
|
||||
if remainder is None:
|
||||
return None
|
||||
return match_command(remainder, threshold)
|
||||
Loading…
Reference in New Issue
Block a user