From 947b30c22e122946588b0ee6548df2016adce206 Mon Sep 17 00:00:00 2001 From: disqualifier Date: Thu, 25 Jun 2026 17:55:21 -0400 Subject: [PATCH] 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 --- src/claudedo/grammar.py | 159 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 src/claudedo/grammar.py diff --git a/src/claudedo/grammar.py b/src/claudedo/grammar.py new file mode 100644 index 0000000..f66e46e --- /dev/null +++ b/src/claudedo/grammar.py @@ -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)