diff --git a/shared/examples/config.example.json b/shared/examples/config.example.json index 9295dd5..a4bfe08 100644 --- a/shared/examples/config.example.json +++ b/shared/examples/config.example.json @@ -1,5 +1,10 @@ { "schema_version": 1, + "bindings": { + "mouse.side_front.press": "toggle_recording", + "mouse.side_rear.press": "trigger_secondary_action", + "hotkey.record_toggle": "toggle_recording" + }, "transcriber": { "backend": "funasr_onnx", "model_name": "iic/SenseVoiceSmall", diff --git a/shared/schema/config.schema.json b/shared/schema/config.schema.json index eb4215a..6ec7865 100644 --- a/shared/schema/config.schema.json +++ b/shared/schema/config.schema.json @@ -16,8 +16,23 @@ "bindings": { "type": "object", "default": {}, + "propertyNames": { + "type": "string", + "pattern": "^[a-z][a-z0-9_]*(\\.[a-z][a-z0-9_]*)*$" + }, "additionalProperties": { - "type": "string" + "type": "string", + "enum": [ + "noop", + "reload_config", + "send_enter", + "shutdown", + "submit_recording", + "toggle_recording", + "trigger_secondary_action", + "workspace_left", + "workspace_right" + ] } }, "transcriber": { diff --git a/tests/bindings/test_resolver.py b/tests/bindings/test_resolver.py new file mode 100644 index 0000000..6fa3202 --- /dev/null +++ b/tests/bindings/test_resolver.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace + +from vibemouse.bindings.actions import ( + build_default_bindings, + build_resolved_bindings, + command_for_legacy_gesture_action, +) +from vibemouse.bindings.resolver import BindingResolver +from vibemouse.core.commands import ( + COMMAND_NOOP, + COMMAND_SEND_ENTER, + COMMAND_SUBMIT_RECORDING, + COMMAND_TOGGLE_RECORDING, + COMMAND_TRIGGER_SECONDARY_ACTION, + COMMAND_WORKSPACE_LEFT, + COMMAND_WORKSPACE_RIGHT, + EVENT_GESTURE_DOWN, + EVENT_GESTURE_LEFT, + EVENT_GESTURE_RIGHT, + EVENT_GESTURE_UP, + EVENT_HOTKEY_RECORDING_SUBMIT, + EVENT_HOTKEY_RECORD_TOGGLE, + EVENT_MOUSE_SIDE_FRONT_PRESS, + EVENT_MOUSE_SIDE_REAR_PRESS, +) + + +class BindingActionsTests(unittest.TestCase): + @staticmethod + def _make_config(**overrides: object) -> SimpleNamespace: + values = { + "bindings": {}, + "gesture_up_action": "record_toggle", + "gesture_down_action": "noop", + "gesture_left_action": "workspace_left", + "gesture_right_action": "workspace_right", + "recording_submit_keycode": 28, + } + values.update(overrides) + return SimpleNamespace(**values) + + def test_default_bindings_follow_runtime_defaults(self) -> None: + config = self._make_config() + + bindings = build_default_bindings(config) + + self.assertEqual(bindings[EVENT_MOUSE_SIDE_FRONT_PRESS], COMMAND_TOGGLE_RECORDING) + self.assertEqual( + bindings[EVENT_MOUSE_SIDE_REAR_PRESS], + COMMAND_TRIGGER_SECONDARY_ACTION, + ) + self.assertEqual(bindings[EVENT_HOTKEY_RECORD_TOGGLE], COMMAND_TOGGLE_RECORDING) + self.assertEqual( + bindings[EVENT_HOTKEY_RECORDING_SUBMIT], + COMMAND_SUBMIT_RECORDING, + ) + self.assertEqual(bindings[EVENT_GESTURE_UP], COMMAND_TOGGLE_RECORDING) + self.assertEqual(bindings[EVENT_GESTURE_DOWN], COMMAND_NOOP) + self.assertEqual(bindings[EVENT_GESTURE_LEFT], COMMAND_WORKSPACE_LEFT) + self.assertEqual(bindings[EVENT_GESTURE_RIGHT], COMMAND_WORKSPACE_RIGHT) + + def test_custom_bindings_override_defaults(self) -> None: + config = self._make_config( + bindings={EVENT_MOUSE_SIDE_FRONT_PRESS: COMMAND_SEND_ENTER} + ) + + bindings = build_resolved_bindings(config) + + self.assertEqual(bindings[EVENT_MOUSE_SIDE_FRONT_PRESS], COMMAND_SEND_ENTER) + + def test_legacy_gesture_action_names_translate_to_commands(self) -> None: + self.assertEqual( + command_for_legacy_gesture_action("record_toggle"), + COMMAND_TOGGLE_RECORDING, + ) + self.assertEqual( + command_for_legacy_gesture_action("workspace_right"), + COMMAND_WORKSPACE_RIGHT, + ) + self.assertEqual(command_for_legacy_gesture_action("noop"), COMMAND_NOOP) + + +class BindingResolverTests(unittest.TestCase): + def test_resolve_returns_bound_command(self) -> None: + resolver = BindingResolver({EVENT_MOUSE_SIDE_FRONT_PRESS: COMMAND_SEND_ENTER}) + + self.assertEqual( + resolver.resolve(EVENT_MOUSE_SIDE_FRONT_PRESS), + COMMAND_SEND_ENTER, + ) + + def test_resolve_returns_none_for_unbound_event(self) -> None: + resolver = BindingResolver({EVENT_MOUSE_SIDE_FRONT_PRESS: COMMAND_SEND_ENTER}) + + self.assertIsNone(resolver.resolve("mouse.middle.press")) diff --git a/tests/core/test_app.py b/tests/core/test_app.py index 818653f..de80c3c 100644 --- a/tests/core/test_app.py +++ b/tests/core/test_app.py @@ -11,6 +11,7 @@ from typing import cast from unittest.mock import patch +from vibemouse.core.commands import COMMAND_SEND_ENTER, EVENT_MOUSE_SIDE_FRONT_PRESS from vibemouse.app import VoiceMouseApp @@ -276,6 +277,33 @@ def test_recording_submit_press_is_ignored_when_idle(self) -> None: self.assertEqual(rear_calls, []) + def test_handle_input_event_routes_through_binding_resolver(self) -> None: + subject = self._make_subject() + send_enter_calls: list[str] = [] + setattr( + subject, + "_binding_resolver", + SimpleNamespace( + resolve=lambda event_name: COMMAND_SEND_ENTER + if event_name == EVENT_MOUSE_SIDE_FRONT_PRESS + else None + ), + ) + setattr( + subject, + "_output", + SimpleNamespace(send_enter=lambda mode: send_enter_calls.append(mode)), + ) + setattr(subject, "_config", SimpleNamespace(enter_mode="enter")) + + handle_event = cast( + Callable[[str], None], + getattr(subject, "_handle_input_event"), + ) + handle_event(EVENT_MOUSE_SIDE_FRONT_PRESS) + + self.assertEqual(send_enter_calls, ["enter"]) + def test_transcribe_and_output_openclaw_uses_openclaw_sender(self) -> None: subject = self._make_subject() recording = SimpleNamespace(duration_s=1.0, path=Path("/tmp/transcribe.wav")) diff --git a/tests/listener/test_keyboard_listener.py b/tests/listener/test_keyboard_listener.py index 186f043..a1bd506 100644 --- a/tests/listener/test_keyboard_listener.py +++ b/tests/listener/test_keyboard_listener.py @@ -4,6 +4,7 @@ from collections.abc import Callable from typing import cast +from vibemouse.core.commands import EVENT_HOTKEY_RECORD_TOGGLE from vibemouse.keyboard_listener import KeyboardHotkeyListener @@ -16,6 +17,13 @@ def test_constructor_rejects_empty_combo(self) -> None: with self.assertRaisesRegex(ValueError, "keycodes must not be empty"): _ = KeyboardHotkeyListener(on_hotkey=_noop, keycodes=()) + def test_constructor_requires_callback_or_event(self) -> None: + with self.assertRaisesRegex( + ValueError, + "on_hotkey or on_event/event_name must be configured", + ): + _ = KeyboardHotkeyListener(keycodes=(42, 125, 193)) + def test_combo_fires_once_until_released(self) -> None: listener = KeyboardHotkeyListener( on_hotkey=_noop, keycodes=(42, 125, 193), debounce_s=0.0 @@ -60,3 +68,17 @@ def test_reset_pressed_state_clears_latched_combo(self) -> None: self.assertFalse(process(42, 1)) self.assertFalse(process(125, 1)) self.assertTrue(process(193, 1)) + + def test_dispatch_hotkey_emits_configured_event(self) -> None: + seen: list[str] = [] + listener = KeyboardHotkeyListener( + on_event=seen.append, + event_name=EVENT_HOTKEY_RECORD_TOGGLE, + keycodes=(42, 125, 193), + debounce_s=0.0, + ) + + dispatch = cast(Callable[[], None], getattr(listener, "_dispatch_hotkey")) + dispatch() + + self.assertEqual(seen, [EVENT_HOTKEY_RECORD_TOGGLE]) diff --git a/tests/listener/test_mouse_listener.py b/tests/listener/test_mouse_listener.py index 121c3fc..83709fa 100644 --- a/tests/listener/test_mouse_listener.py +++ b/tests/listener/test_mouse_listener.py @@ -6,6 +6,7 @@ from typing import cast from unittest.mock import patch +from vibemouse.core.commands import EVENT_GESTURE_UP, EVENT_MOUSE_SIDE_FRONT_PRESS from vibemouse.mouse_listener import SideButtonListener @@ -98,6 +99,13 @@ def test_constructor_accepts_right_trigger_button(self) -> None: self.assertIsNotNone(listener) + def test_constructor_requires_event_or_button_callbacks(self) -> None: + with self.assertRaisesRegex( + ValueError, + "on_event or on_front_press/on_rear_press must be configured", + ): + _ = SideButtonListener(front_button="x1", rear_button="x2") + def test_constructor_clamps_rescan_interval_to_minimum(self) -> None: listener = SideButtonListener( on_front_press=_noop_button, @@ -131,6 +139,40 @@ def on_gesture(direction: str) -> None: dispatch_gesture("up") self.assertEqual(seen, ["up"]) + def test_dispatch_front_press_emits_normalized_event_when_configured(self) -> None: + seen: list[str] = [] + listener = SideButtonListener( + on_event=seen.append, + front_button="x1", + rear_button="x2", + debounce_s=0.0, + ) + + dispatch_front = cast( + Callable[[], None], getattr(listener, "_dispatch_front_press") + ) + dispatch_front() + + self.assertEqual(seen, [EVENT_MOUSE_SIDE_FRONT_PRESS]) + + def test_dispatch_gesture_emits_normalized_event_when_on_event_is_used( + self, + ) -> None: + seen: list[str] = [] + listener = SideButtonListener( + on_event=seen.append, + front_button="x1", + rear_button="x2", + ) + + dispatch_gesture = cast( + Callable[[str], None], + getattr(listener, "_dispatch_gesture"), + ) + dispatch_gesture("up") + + self.assertEqual(seen, [EVENT_GESTURE_UP]) + def test_finish_gesture_restores_cursor_after_direction_action(self) -> None: seen: list[str] = [] restored: list[tuple[int, int]] = [] diff --git a/tests/test_config.py b/tests/test_config.py index 39117e8..d4fbf07 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -37,6 +37,7 @@ def test_defaults_disable_trust_remote_code(self) -> None: self.assertEqual(config.rear_button, "x2") self.assertEqual(config.record_hotkey_keycodes, (42, 125, 193)) self.assertIsNone(config.recording_submit_keycode) + self.assertEqual(config.bindings, {}) def test_record_hotkey_keycodes_can_be_configured(self) -> None: with patch.dict( diff --git a/tests/test_config_store.py b/tests/test_config_store.py index 712b4d7..74bb4fb 100644 --- a/tests/test_config_store.py +++ b/tests/test_config_store.py @@ -30,6 +30,9 @@ def test_json_config_values_are_loaded(self) -> None: json.dumps( { "schema_version": 1, + "bindings": { + "mouse.side_front.press": "send_enter", + }, "transcriber": { "backend": "funasr", "model_name": "custom/model", @@ -64,6 +67,10 @@ def test_json_config_values_are_loaded(self) -> None: self.assertTrue(config.trust_remote_code) self.assertEqual(config.gesture_trigger_button, "right") self.assertEqual(config.record_hotkey_keycodes, (30, 31, 32)) + self.assertEqual( + config.bindings, + {"mouse.side_front.press": "send_enter"}, + ) self.assertTrue(config.auto_paste) self.assertEqual(config.enter_mode, "ctrl_enter") self.assertEqual(config.log_level, "ERROR") @@ -121,6 +128,9 @@ def test_save_document_writes_normalized_config(self) -> None: store.save_document( { "schema_version": 1, + "bindings": { + "mouse.side_front.press": "send_enter", + }, "input": { "front_button": "x2", "rear_button": "x1", @@ -132,12 +142,37 @@ def test_save_document_writes_normalized_config(self) -> None: payload = json.loads(config_path.read_text(encoding="utf-8")) self.assertEqual(payload["schema_version"], 1) + self.assertEqual( + payload["bindings"], + {"mouse.side_front.press": "send_enter"}, + ) self.assertEqual(payload["input"]["front_button"], "x2") self.assertEqual(payload["input"]["rear_button"], "x1") self.assertEqual(payload["input"]["record_hotkey_keycodes"], [10, 20, 30]) self.assertIn("transcriber", payload) self.assertIn("runtime", payload) + def test_invalid_binding_command_is_rejected(self) -> None: + with tempfile.TemporaryDirectory(prefix="vibemouse-config-") as tmp: + config_path = Path(tmp) / "config.json" + config_path.write_text( + json.dumps( + { + "schema_version": 1, + "bindings": { + "mouse.side_front.press": "paste_now", + }, + } + ), + encoding="utf-8", + ) + + with self.assertRaisesRegex( + ValueError, + "bindings\\['mouse\\.side_front\\.press'\\] must be one of", + ): + _ = load_config(config_path, env={}) + class StatusStoreTests(unittest.TestCase): def test_write_persists_normalized_status_payload(self) -> None: diff --git a/vibemouse/bindings/__init__.py b/vibemouse/bindings/__init__.py new file mode 100644 index 0000000..a9a2c5b --- /dev/null +++ b/vibemouse/bindings/__init__.py @@ -0,0 +1 @@ +__all__ = [] diff --git a/vibemouse/bindings/actions.py b/vibemouse/bindings/actions.py new file mode 100644 index 0000000..c965906 --- /dev/null +++ b/vibemouse/bindings/actions.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from vibemouse.config.schema import AppConfig +from vibemouse.core.commands import ( + COMMAND_NOOP, + COMMAND_SEND_ENTER, + COMMAND_SUBMIT_RECORDING, + COMMAND_TOGGLE_RECORDING, + COMMAND_TRIGGER_SECONDARY_ACTION, + COMMAND_WORKSPACE_LEFT, + COMMAND_WORKSPACE_RIGHT, + EVENT_GESTURE_DOWN, + EVENT_GESTURE_LEFT, + EVENT_GESTURE_RIGHT, + EVENT_GESTURE_UP, + EVENT_HOTKEY_RECORDING_SUBMIT, + EVENT_HOTKEY_RECORD_TOGGLE, + EVENT_MOUSE_SIDE_FRONT_PRESS, + EVENT_MOUSE_SIDE_REAR_PRESS, +) + + +_LEGACY_GESTURE_ACTION_TO_COMMAND: dict[str, str] = { + "noop": COMMAND_NOOP, + "record_toggle": COMMAND_TOGGLE_RECORDING, + "send_enter": COMMAND_SEND_ENTER, + "workspace_left": COMMAND_WORKSPACE_LEFT, + "workspace_right": COMMAND_WORKSPACE_RIGHT, +} + + +def command_for_legacy_gesture_action(action: str) -> str: + return _LEGACY_GESTURE_ACTION_TO_COMMAND[action.strip().lower()] + + +def build_default_bindings(config: AppConfig) -> dict[str, str]: + bindings = { + EVENT_MOUSE_SIDE_FRONT_PRESS: COMMAND_TOGGLE_RECORDING, + EVENT_MOUSE_SIDE_REAR_PRESS: COMMAND_TRIGGER_SECONDARY_ACTION, + EVENT_HOTKEY_RECORD_TOGGLE: COMMAND_TOGGLE_RECORDING, + EVENT_GESTURE_UP: command_for_legacy_gesture_action(config.gesture_up_action), + EVENT_GESTURE_DOWN: command_for_legacy_gesture_action( + config.gesture_down_action + ), + EVENT_GESTURE_LEFT: command_for_legacy_gesture_action( + config.gesture_left_action + ), + EVENT_GESTURE_RIGHT: command_for_legacy_gesture_action( + config.gesture_right_action + ), + } + if config.recording_submit_keycode is not None: + bindings[EVENT_HOTKEY_RECORDING_SUBMIT] = COMMAND_SUBMIT_RECORDING + return bindings + + +def build_resolved_bindings(config: AppConfig) -> dict[str, str]: + bindings = build_default_bindings(config) + bindings.update(config.bindings) + return bindings diff --git a/vibemouse/bindings/resolver.py b/vibemouse/bindings/resolver.py new file mode 100644 index 0000000..ab17dab --- /dev/null +++ b/vibemouse/bindings/resolver.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from collections.abc import Mapping + +from vibemouse.bindings.actions import build_resolved_bindings +from vibemouse.config.schema import AppConfig + + +class BindingResolver: + def __init__(self, bindings: Mapping[str, str]) -> None: + self._bindings: dict[str, str] = { + str(event_name).strip().lower(): str(command_name).strip().lower() + for event_name, command_name in bindings.items() + } + + @classmethod + def from_config(cls, config: AppConfig) -> "BindingResolver": + return cls(build_resolved_bindings(config)) + + def resolve(self, event_name: str) -> str | None: + normalized = event_name.strip().lower() + if not normalized: + return None + return self._bindings.get(normalized) + + def snapshot(self) -> dict[str, str]: + return dict(self._bindings) diff --git a/vibemouse/config/schema.py b/vibemouse/config/schema.py index 08d7923..370986e 100644 --- a/vibemouse/config/schema.py +++ b/vibemouse/config/schema.py @@ -8,6 +8,8 @@ from pathlib import Path from typing import Mapping +from vibemouse.core.commands import KNOWN_COMMAND_NAMES, is_valid_event_name + LATEST_CONFIG_SCHEMA_VERSION = 1 @@ -28,6 +30,7 @@ @dataclass(frozen=True) class AppConfig: + bindings: dict[str, str] sample_rate: int channels: int dtype: str @@ -155,6 +158,7 @@ def build_default_config_document() -> dict[str, object]: def config_document_to_app_config(document: Mapping[str, object]) -> AppConfig: normalized = normalize_config_document(document) + bindings = _expect_mapping(normalized, "bindings") transcriber = _expect_mapping(normalized, "transcriber") input_section = _expect_mapping(normalized, "input") output = _expect_mapping(normalized, "output") @@ -164,6 +168,7 @@ def config_document_to_app_config(document: Mapping[str, object]) -> AppConfig: runtime = _expect_mapping(normalized, "runtime") return AppConfig( + bindings={str(key): str(value) for key, value in bindings.items()}, sample_rate=int(transcriber["sample_rate"]), channels=int(transcriber["channels"]), dtype=str(transcriber["dtype"]), @@ -346,8 +351,21 @@ def _normalize_bindings(raw: object) -> dict[str, str]: section = _expect_mapping_value(raw, "bindings") normalized: dict[str, str] = {} for key, value in section.items(): - event_name = _coerce_string(key, "bindings key") - command_name = _coerce_string(value, f"bindings[{event_name!r}]") + event_name = _coerce_non_empty_string(key, "bindings key").lower() + if not is_valid_event_name(event_name): + raise ValueError( + "bindings key must use normalized event segments separated by dots; " + + f"got {event_name!r}" + ) + command_name = _coerce_non_empty_string( + value, + f"bindings[{event_name!r}]", + ).lower() + if command_name not in KNOWN_COMMAND_NAMES: + options = ", ".join(sorted(KNOWN_COMMAND_NAMES)) + raise ValueError( + f"bindings[{event_name!r}] must be one of: {options}; got {command_name!r}" + ) normalized[event_name] = command_name return normalized diff --git a/vibemouse/core/app.py b/vibemouse/core/app.py index a7ba9de..735e15c 100644 --- a/vibemouse/core/app.py +++ b/vibemouse/core/app.py @@ -6,7 +6,21 @@ from pathlib import Path from typing import Literal +from vibemouse.bindings.actions import command_for_legacy_gesture_action +from vibemouse.bindings.resolver import BindingResolver from vibemouse.core.audio import AudioRecorder, AudioRecording +from vibemouse.core.commands import ( + COMMAND_NOOP, + COMMAND_SEND_ENTER, + COMMAND_SUBMIT_RECORDING, + COMMAND_TOGGLE_RECORDING, + COMMAND_TRIGGER_SECONDARY_ACTION, + COMMAND_WORKSPACE_LEFT, + COMMAND_WORKSPACE_RIGHT, + EVENT_HOTKEY_RECORDING_SUBMIT, + EVENT_HOTKEY_RECORD_TOGGLE, + gesture_direction_to_event, +) from vibemouse.config import AppConfig, write_status from vibemouse.core.output import TextOutput from vibemouse.core.transcriber import SenseVoiceTranscriber @@ -43,10 +57,9 @@ def __init__(self, config: AppConfig) -> None: openclaw_timeout_s=config.openclaw_timeout_s, openclaw_retries=config.openclaw_retries, ) + self._binding_resolver: BindingResolver = BindingResolver.from_config(config) self._listener: SideButtonListener = SideButtonListener( - on_front_press=self._on_front_press, - on_rear_press=self._on_rear_press, - on_gesture=self._on_gesture, + on_event=self._handle_input_event, front_button=config.front_button, rear_button=config.rear_button, debounce_s=config.button_debounce_ms / 1000.0, @@ -58,14 +71,16 @@ def __init__(self, config: AppConfig) -> None: system_integration=self._system_integration, ) self._keyboard_listener: KeyboardHotkeyListener = KeyboardHotkeyListener( - on_hotkey=self._on_front_press, + on_event=self._handle_input_event, + event_name=EVENT_HOTKEY_RECORD_TOGGLE, keycodes=config.record_hotkey_keycodes, debounce_s=config.button_debounce_ms / 1000.0, ) self._recording_submit_listener: KeyboardHotkeyListener | None = None if config.recording_submit_keycode is not None: self._recording_submit_listener = KeyboardHotkeyListener( - on_hotkey=self._on_recording_submit_press, + on_event=self._handle_input_event, + event_name=EVENT_HOTKEY_RECORDING_SUBMIT, keycodes=(config.recording_submit_keycode,), debounce_s=config.button_debounce_ms / 1000.0, ) @@ -127,6 +142,75 @@ def shutdown(self) -> None: ) def _on_front_press(self) -> None: + self._execute_command(COMMAND_TOGGLE_RECORDING) + + def _on_rear_press(self) -> None: + self._execute_command(COMMAND_TRIGGER_SECONDARY_ACTION) + + def _on_recording_submit_press(self) -> None: + self._execute_command(COMMAND_SUBMIT_RECORDING) + + def _on_gesture(self, direction: str) -> None: + event_name = gesture_direction_to_event(direction) + if event_name is None: + _LOG.warning("Gesture '%s' mapped to unknown direction", direction) + return + + try: + binding_resolver = self._binding_resolver + except AttributeError: + action = self._resolve_gesture_action(direction) + self._execute_command(command_for_legacy_gesture_action(action)) + return + + command_name = binding_resolver.resolve(event_name) + if command_name is None: + _LOG.info("Gesture '%s' recognized with no action configured", direction) + return + self._execute_command(command_name, source_event=event_name) + + def _handle_input_event(self, event_name: str) -> None: + command_name = self._binding_resolver.resolve(event_name) + if command_name is None: + _LOG.debug("Ignoring unbound input event: %s", event_name) + return + self._execute_command(command_name, source_event=event_name) + + def _execute_command( + self, + command_name: str, + *, + source_event: str | None = None, + ) -> None: + if source_event is not None: + _LOG.debug("Resolved input event '%s' -> '%s'", source_event, command_name) + + if command_name == COMMAND_NOOP: + if source_event is not None: + _LOG.info("Input event '%s' resolved to noop", source_event) + return + if command_name == COMMAND_TOGGLE_RECORDING: + self._toggle_recording() + return + if command_name == COMMAND_TRIGGER_SECONDARY_ACTION: + self._trigger_secondary_action() + return + if command_name == COMMAND_SUBMIT_RECORDING: + self._submit_recording() + return + if command_name == COMMAND_SEND_ENTER: + self._send_enter_command(force_when_disabled=True) + return + if command_name == COMMAND_WORKSPACE_LEFT: + self._dispatch_workspace_command("left") + return + if command_name == COMMAND_WORKSPACE_RIGHT: + self._dispatch_workspace_command("right") + return + + _LOG.warning("Ignoring unsupported command '%s'", command_name) + + def _toggle_recording(self) -> None: if not self._recorder.is_recording: try: self._recorder.start() @@ -148,77 +232,41 @@ def _on_front_press(self) -> None: self._start_transcription_worker(recording, output_target="default") - def _on_rear_press(self) -> None: + def _trigger_secondary_action(self) -> None: if self._recorder.is_recording: - try: - recording = self._stop_recording() - except Exception as error: - _LOG.exception("Failed to stop recording from rear button: %s", error) - return - - if recording is None: - return - - _LOG.info( - "Recording stopped by rear button, sending transcript to OpenClaw" + self._stop_recording_for_output( + output_target="openclaw", + error_prefix="Failed to stop recording from secondary action", + success_message=( + "Recording stopped by secondary action, sending transcript to OpenClaw" + ), ) - self._start_transcription_worker(recording, output_target="openclaw") return - try: - self._output.send_enter(mode=self._config.enter_mode) - if self._config.enter_mode == "none": - _LOG.info("Enter key handling disabled (enter_mode=none)") - else: - _LOG.info("Enter key sent") - except Exception as error: - _LOG.exception("Failed to send Enter: %s", error) + self._send_enter_command(force_when_disabled=False) - def _on_recording_submit_press(self) -> None: + def _submit_recording(self) -> None: if not self._recorder.is_recording: return - _LOG.info("Recording submit hotkey pressed, routing to rear-button logic") - self._on_rear_press() - def _on_gesture(self, direction: str) -> None: - action = self._resolve_gesture_action(direction) - if action == "noop": - _LOG.info("Gesture '%s' recognized with no action configured", direction) - return - - if action == "record_toggle": - _LOG.info("Gesture '%s' -> toggle recording", direction) - self._on_front_press() - return + self._stop_recording_for_output( + output_target="openclaw", + error_prefix="Failed to stop recording from submit command", + success_message="Recording submit command received, sending transcript to OpenClaw", + ) - if action == "send_enter": + def _send_enter_command(self, *, force_when_disabled: bool) -> None: + try: mode = self._config.enter_mode - if mode == "none": + if force_when_disabled and mode == "none": mode = "enter" - try: - self._output.send_enter(mode=mode) - _LOG.info("Gesture '%s' -> send enter (%s)", direction, mode) - except Exception as error: - _LOG.exception( - "Gesture '%s' failed to send enter: %s", direction, error - ) - return - - if action == "workspace_left": - if self._switch_workspace("left"): - _LOG.info("Gesture '%s' -> switch workspace left", direction) - else: - _LOG.warning("Gesture '%s' failed to switch workspace left", direction) - return - - if action == "workspace_right": - if self._switch_workspace("right"): - _LOG.info("Gesture '%s' -> switch workspace right", direction) + self._output.send_enter(mode=mode) + if mode == "none": + _LOG.info("Enter key handling disabled (enter_mode=none)") else: - _LOG.warning("Gesture '%s' failed to switch workspace right", direction) - return - - _LOG.warning("Gesture '%s' mapped to unknown action '%s'", direction, action) + _LOG.info("Enter key sent") + except Exception as error: + _LOG.exception("Failed to send Enter: %s", error) def _resolve_gesture_action(self, direction: str) -> str: mapping = { @@ -229,6 +277,12 @@ def _resolve_gesture_action(self, direction: str) -> str: } return mapping.get(direction, "noop") + def _dispatch_workspace_command(self, direction: str) -> None: + if self._switch_workspace(direction): + _LOG.info("Workspace switch command succeeded: %s", direction) + return + _LOG.warning("Workspace switch command failed: %s", direction) + def _switch_workspace(self, direction: str) -> bool: try: system_integration = self._system_integration @@ -268,6 +322,26 @@ def _stop_recording(self) -> AudioRecording | None: return None return recording + def _stop_recording_for_output( + self, + *, + output_target: TranscriptionTarget, + error_prefix: str, + success_message: str | None = None, + ) -> None: + try: + recording = self._stop_recording() + except Exception as error: + _LOG.exception("%s: %s", error_prefix, error) + return + + if recording is None: + return + + if success_message is not None: + _LOG.info(success_message) + self._start_transcription_worker(recording, output_target=output_target) + def _start_transcription_worker( self, recording: AudioRecording, diff --git a/vibemouse/core/commands.py b/vibemouse/core/commands.py new file mode 100644 index 0000000..99e2f58 --- /dev/null +++ b/vibemouse/core/commands.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import re +from typing import Final + + +COMMAND_NOOP: Final = "noop" +COMMAND_RELOAD_CONFIG: Final = "reload_config" +COMMAND_SEND_ENTER: Final = "send_enter" +COMMAND_SHUTDOWN: Final = "shutdown" +COMMAND_SUBMIT_RECORDING: Final = "submit_recording" +COMMAND_TOGGLE_RECORDING: Final = "toggle_recording" +COMMAND_TRIGGER_SECONDARY_ACTION: Final = "trigger_secondary_action" +COMMAND_WORKSPACE_LEFT: Final = "workspace_left" +COMMAND_WORKSPACE_RIGHT: Final = "workspace_right" + +KNOWN_COMMAND_NAMES: Final[frozenset[str]] = frozenset( + { + COMMAND_NOOP, + COMMAND_RELOAD_CONFIG, + COMMAND_SEND_ENTER, + COMMAND_SHUTDOWN, + COMMAND_SUBMIT_RECORDING, + COMMAND_TOGGLE_RECORDING, + COMMAND_TRIGGER_SECONDARY_ACTION, + COMMAND_WORKSPACE_LEFT, + COMMAND_WORKSPACE_RIGHT, + } +) + +EVENT_GESTURE_DOWN: Final = "gesture.down" +EVENT_GESTURE_LEFT: Final = "gesture.left" +EVENT_GESTURE_RIGHT: Final = "gesture.right" +EVENT_GESTURE_UP: Final = "gesture.up" +EVENT_HOTKEY_RECORDING_SUBMIT: Final = "hotkey.recording_submit" +EVENT_HOTKEY_RECORD_TOGGLE: Final = "hotkey.record_toggle" +EVENT_MOUSE_SIDE_FRONT_PRESS: Final = "mouse.side_front.press" +EVENT_MOUSE_SIDE_REAR_PRESS: Final = "mouse.side_rear.press" + +KNOWN_INPUT_EVENTS: Final[frozenset[str]] = frozenset( + { + EVENT_GESTURE_DOWN, + EVENT_GESTURE_LEFT, + EVENT_GESTURE_RIGHT, + EVENT_GESTURE_UP, + EVENT_HOTKEY_RECORDING_SUBMIT, + EVENT_HOTKEY_RECORD_TOGGLE, + EVENT_MOUSE_SIDE_FRONT_PRESS, + EVENT_MOUSE_SIDE_REAR_PRESS, + } +) + +_EVENT_NAME_RE = re.compile(r"^[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*)*$") + + +def gesture_direction_to_event(direction: str) -> str | None: + mapping = { + "up": EVENT_GESTURE_UP, + "down": EVENT_GESTURE_DOWN, + "left": EVENT_GESTURE_LEFT, + "right": EVENT_GESTURE_RIGHT, + } + return mapping.get(direction.strip().lower()) + + +def is_valid_event_name(value: str) -> bool: + return bool(_EVENT_NAME_RE.fullmatch(value.strip().lower())) diff --git a/vibemouse/listener/keyboard_listener.py b/vibemouse/listener/keyboard_listener.py index 117bb7a..c90a0dd 100644 --- a/vibemouse/listener/keyboard_listener.py +++ b/vibemouse/listener/keyboard_listener.py @@ -9,20 +9,29 @@ HotkeyCallback = Callable[[], None] +EventCallback = Callable[[str], None] class KeyboardHotkeyListener: def __init__( self, *, - on_hotkey: HotkeyCallback, keycodes: tuple[int, ...], + on_hotkey: HotkeyCallback | None = None, + on_event: EventCallback | None = None, + event_name: str | None = None, debounce_s: float = 0.15, rescan_interval_s: float = 2.0, ) -> None: if not keycodes: raise ValueError("keycodes must not be empty") - self._on_hotkey: HotkeyCallback = on_hotkey + if on_hotkey is None and (on_event is None or not event_name): + raise ValueError( + "on_hotkey or on_event/event_name must be configured" + ) + self._on_hotkey: HotkeyCallback | None = on_hotkey + self._on_event: EventCallback | None = on_event + self._event_name: str | None = event_name self._combo: frozenset[int] = frozenset(keycodes) self._debounce_s: float = max(0.0, debounce_s) self._rescan_interval_s: float = max(0.2, rescan_interval_s) @@ -118,7 +127,7 @@ def _run_evdev(self) -> None: if event.type != ecodes.EV_KEY: continue if self._process_key_event(event.code, event.value): - self._on_hotkey() + self._dispatch_hotkey() finally: for dev in devices: dev.close() @@ -151,6 +160,15 @@ def _process_key_event(self, keycode: int, value: int) -> bool: return True return False + def _dispatch_hotkey(self) -> None: + event_name = self._event_name + if self._on_event is not None and isinstance(event_name, str) and event_name: + self._on_event(event_name) + return + callback = self._on_hotkey + if callback is not None: + callback() + class _EvdevEvent(Protocol): type: int diff --git a/vibemouse/listener/mouse_listener.py b/vibemouse/listener/mouse_listener.py index a632418..f85a912 100644 --- a/vibemouse/listener/mouse_listener.py +++ b/vibemouse/listener/mouse_listener.py @@ -9,6 +9,11 @@ from collections.abc import Callable from typing import Protocol, cast +from vibemouse.core.commands import ( + EVENT_MOUSE_SIDE_FRONT_PRESS, + EVENT_MOUSE_SIDE_REAR_PRESS, + gesture_direction_to_event, +) from vibemouse.platform.system_integration import ( SystemIntegration, create_system_integration, @@ -18,18 +23,21 @@ ButtonCallback = Callable[[], None] GestureCallback = Callable[[str], None] +EventCallback = Callable[[str], None] _LOG = logging.getLogger(__name__) class SideButtonListener: def __init__( self, - on_front_press: ButtonCallback, - on_rear_press: ButtonCallback, + *, front_button: str, rear_button: str, + on_front_press: ButtonCallback | None = None, + on_rear_press: ButtonCallback | None = None, debounce_s: float = 0.15, on_gesture: GestureCallback | None = None, + on_event: EventCallback | None = None, gestures_enabled: bool = False, gesture_trigger_button: str = "rear", gesture_threshold_px: int = 120, @@ -42,9 +50,14 @@ def __init__( raise ValueError( "gesture_trigger_button must be one of: front, rear, right" ) - self._on_front_press: ButtonCallback = on_front_press - self._on_rear_press: ButtonCallback = on_rear_press + if on_event is None and (on_front_press is None or on_rear_press is None): + raise ValueError( + "on_event or on_front_press/on_rear_press must be configured" + ) + self._on_front_press: ButtonCallback | None = on_front_press + self._on_rear_press: ButtonCallback | None = on_rear_press self._on_gesture: GestureCallback | None = on_gesture + self._on_event: EventCallback | None = on_event self._front_button: str = front_button self._rear_button: str = rear_button self._debounce_s: float = max(0.0, debounce_s) @@ -622,6 +635,11 @@ def _finish_gesture_capture(self, button_label: str) -> None: self._restore_cursor_position(anchor_cursor) def _dispatch_gesture(self, direction: str) -> None: + if self._on_event is not None: + event_name = gesture_direction_to_event(direction) + if event_name is not None: + self._on_event(event_name) + return callback = self._on_gesture if callback is None: return @@ -853,11 +871,21 @@ def _classify_gesture(dx: int, dy: int, threshold_px: int) -> str | None: def _dispatch_front_press(self) -> None: if self._should_fire_front(): - self._on_front_press() + callback = self._on_front_press + if self._on_event is not None: + self._on_event(EVENT_MOUSE_SIDE_FRONT_PRESS) + return + if callback is not None: + callback() def _dispatch_rear_press(self) -> None: if self._should_fire_rear(): - self._on_rear_press() + callback = self._on_rear_press + if self._on_event is not None: + self._on_event(EVENT_MOUSE_SIDE_REAR_PRESS) + return + if callback is not None: + callback() def _should_fire_front(self) -> bool: now = time.monotonic()