diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/parley/config.py | 22 | ||||
| -rw-r--r-- | src/parley/controller.py | 173 | ||||
| -rw-r--r-- | src/parley/ctl.py | 78 | ||||
| -rw-r--r-- | src/parley/daemon.py | 212 | ||||
| -rw-r--r-- | src/parley/desktop.py | 53 | ||||
| -rw-r--r-- | src/parley/interfaces/__init__.py | 1 | ||||
| -rw-r--r-- | src/parley/interfaces/org.parley.Transcription1.xml | 36 | ||||
| -rw-r--r-- | src/parley/state.py | 62 | ||||
| -rw-r--r-- | src/parley/transcription.py | 35 |
9 files changed, 659 insertions, 13 deletions
diff --git a/src/parley/config.py b/src/parley/config.py index 88640b6..9dd953e 100644 --- a/src/parley/config.py +++ b/src/parley/config.py @@ -35,21 +35,27 @@ class Config: @classmethod def from_environment(cls) -> "Config": prototype = _prototype_root() + data_home = _default_data_home() ffmpeg = os.environ.get("PARLEY_FFMPEG") or shutil.which("ffmpeg") or "ffmpeg" + installed_model = ( + data_home / "parley/models/parakeet-unified-en-0.6b-Q8_0.gguf" + ) + default_model = ( + installed_model + if installed_model.is_file() + else prototype / "models/parakeet-unified-en-0.6b-Q8_0.gguf" + ) return cls( ffmpeg=Path(ffmpeg).expanduser(), transcribe_cli=Path( - os.environ.get("PARLEY_TRANSCRIBE_CLI", prototype / "bin/transcribe-cli") - ).expanduser(), - model=Path( - os.environ.get( - "PARLEY_MODEL", - prototype / "models/parakeet-unified-en-0.6b-Q8_0.gguf", - ) + os.environ.get("PARLEY_TRANSCRIBE_CLI") + or shutil.which("transcribe-cli") + or prototype / "bin/transcribe-cli" ).expanduser(), + model=Path(os.environ.get("PARLEY_MODEL") or default_model).expanduser(), transcript_dir=Path( os.environ.get( - "PARLEY_TRANSCRIPT_DIR", _default_data_home() / "parley/transcripts" + "PARLEY_TRANSCRIPT_DIR", data_home / "parley/transcripts" ) ).expanduser(), device=os.environ.get("PARLEY_DEVICE", "default"), diff --git a/src/parley/controller.py b/src/parley/controller.py new file mode 100644 index 0000000..58091cf --- /dev/null +++ b/src/parley/controller.py @@ -0,0 +1,173 @@ +"""Asynchronous orchestration shared by the D-Bus daemon and tests.""" + +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +import tempfile +import threading + +from .config import Config +from .persistence import save_transcript +from .recording import Recorder +from .state import State, StateMachine +from .transcription import Transcriber + + +Dispatch = Callable[[Callable[[], None]], None] + + +class Controller: + """Serialize one recording/transcription operation without blocking its caller.""" + + def __init__( + self, + config: Config, + *, + dispatch: Dispatch | None = None, + on_state: Callable[[State], None] | None = None, + on_transcript: Callable[[str, Path], None] | None = None, + on_error: Callable[[str, str], None] | None = None, + ) -> None: + self.config = config + self.machine = StateMachine() + self.recorder = Recorder(config) + self.transcriber = Transcriber(config) + self.dispatch = dispatch or (lambda callback: callback()) + self.on_state = on_state or (lambda state: None) + self.on_transcript = on_transcript or (lambda text, path: None) + self.on_error = on_error or (lambda code, message: None) + self.last_transcript = "" + self.last_transcript_path: Path | None = None + self.last_error = "" + self._workspace: tempfile.TemporaryDirectory[str] | None = None + self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="parley") + self._lock = threading.Lock() + self._generation = 0 + + @property + def state(self) -> State: + return self.machine.state + + def start(self) -> None: + self.machine.start_recording() + self.last_error = "" + try: + self.config.validate() + self._workspace = tempfile.TemporaryDirectory(prefix="parley-") + wav_path = Path(self._workspace.name) / "recording.wav" + self.recorder.start(wav_path) + except Exception as error: + self._cleanup_workspace() + self._fail("recording-start-failed", str(error)) + return + self._generation += 1 + self._emit_state() + + def stop_and_transcribe(self) -> None: + self.machine.begin_transcription() + token = self._generation + workspace = self._workspace + if workspace is None: + self._fail("transcription-failed", "recording workspace is unavailable") + return + self._emit_state() + self._executor.submit(self._transcribe_worker, token, workspace) + + def toggle(self) -> None: + if self.state in (State.IDLE, State.ERROR): + self.start() + elif self.state is State.RECORDING: + self.stop_and_transcribe() + + def cancel(self) -> None: + previous = self.state + self.machine.cancel() + self._generation += 1 + if previous is State.RECORDING: + self.recorder.cancel() + self._cleanup_workspace() + else: + self.transcriber.cancel() + self._emit_state() + + def close(self) -> None: + if self.state is State.RECORDING: + self.recorder.cancel() + elif self.state is State.TRANSCRIBING: + self.transcriber.cancel() + self._generation += 1 + self._cleanup_workspace() + self._executor.shutdown(wait=False, cancel_futures=True) + + def _transcribe_worker( + self, token: int, workspace: tempfile.TemporaryDirectory[str] + ) -> None: + try: + wav_path = self.recorder.stop() + if token != self._generation: + self.dispatch(lambda: self._discard_worker(workspace)) + return + text = self.transcriber.transcribe(wav_path, Path(workspace.name)) + if token != self._generation: + self.dispatch(lambda: self._discard_worker(workspace)) + return + path = save_transcript(text, self.config.transcript_dir) + except Exception as error: + self.dispatch( + lambda error=error: self._worker_failed(token, workspace, error) + ) + else: + self.dispatch( + lambda: self._worker_succeeded(token, workspace, text, path) + ) + + def _worker_succeeded( + self, + token: int, + workspace: tempfile.TemporaryDirectory[str], + text: str, + path: Path, + ) -> None: + if token != self._generation: + self._cleanup_workspace(workspace) + return + self.last_transcript = text + self.last_transcript_path = path + self.machine.complete() + self._cleanup_workspace(workspace) + self.on_transcript(text, path) + self._emit_state() + + def _worker_failed( + self, + token: int, + workspace: tempfile.TemporaryDirectory[str], + error: Exception, + ) -> None: + if token != self._generation: + self._cleanup_workspace(workspace) + return + self._cleanup_workspace(workspace) + self._fail("transcription-failed", str(error)) + + def _discard_worker(self, workspace: tempfile.TemporaryDirectory[str]) -> None: + self._cleanup_workspace(workspace) + + def _fail(self, code: str, message: str) -> None: + self.machine.fail() + self.last_error = message + self.on_error(code, message) + self._emit_state() + + def _emit_state(self) -> None: + self.on_state(self.state) + + def _cleanup_workspace( + self, workspace: tempfile.TemporaryDirectory[str] | None = None + ) -> None: + with self._lock: + target = workspace or self._workspace + if target is self._workspace: + self._workspace = None + if target is not None: + target.cleanup() diff --git a/src/parley/ctl.py b/src/parley/ctl.py new file mode 100644 index 0000000..8ae7450 --- /dev/null +++ b/src/parley/ctl.py @@ -0,0 +1,78 @@ +"""Thin command-line client for the Parley D-Bus daemon.""" + +import argparse +import sys + +import gi + +gi.require_version("Gio", "2.0") +from gi.repository import Gio, GLib + +from .daemon import BUS_NAME, INTERFACE, OBJECT_PATH + + +METHODS = { + "toggle": "Toggle", + "start": "StartRecording", + "stop": "StopAndTranscribe", + "cancel": "Cancel", + "copy": "CopyLastTranscript", + "insert": "InsertLastTranscript", + "open-folder": "OpenTranscriptFolder", +} + + +def _proxy() -> Gio.DBusProxy: + return Gio.DBusProxy.new_for_bus_sync( + Gio.BusType.SESSION, + Gio.DBusProxyFlags.NONE, + None, + BUS_NAME, + OBJECT_PATH, + INTERFACE, + None, + ) + + +def _status(proxy: Gio.DBusProxy) -> None: + result = proxy.call_sync( + "org.freedesktop.DBus.Properties.GetAll", + GLib.Variant("(s)", (INTERFACE,)), + Gio.DBusCallFlags.NONE, + -1, + None, + ) + properties = result.unpack()[0] + print(f"state: {properties['State']}") + print(f"insertion mode: {properties['InsertionMode']}") + print(f"auto insert: {'yes' if properties['AutoInsert'] else 'no'}") + if properties["LastTranscriptPath"]: + print(f"last transcript: {properties['LastTranscriptPath']}") + if properties["LastError"]: + print(f"last error: {properties['LastError']}") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="parleyctl") + parser.add_argument("command", choices=[*METHODS, "status"]) + arguments = parser.parse_args(argv) + try: + proxy = _proxy() + if arguments.command == "status": + _status(proxy) + else: + proxy.call_sync( + METHODS[arguments.command], + None, + Gio.DBusCallFlags.NONE, + -1, + None, + ) + return 0 + except GLib.Error as error: + print(f"parleyctl: {error.message}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/parley/daemon.py b/src/parley/daemon.py new file mode 100644 index 0000000..09fb566 --- /dev/null +++ b/src/parley/daemon.py @@ -0,0 +1,212 @@ +"""Per-user GObject/D-Bus daemon for Parley.""" + +from importlib.resources import files +import signal +import sys + +import gi + +gi.require_version("Gio", "2.0") +gi.require_version("GLib", "2.0") +gi.require_version("GLibUnix", "2.0") +from gi.repository import Gio, GLib, GLibUnix + +from .config import Config +from .controller import Controller +from .desktop import copy_text, notify, open_folder +from .errors import ParleyError +from .state import State + + +BUS_NAME = "org.parley.Transcription1" +OBJECT_PATH = "/org/parley/Transcription1" +INTERFACE = BUS_NAME + + +class Daemon: + def __init__(self, config: Config | None = None) -> None: + xml = files("parley.interfaces").joinpath( + "org.parley.Transcription1.xml" + ).read_text(encoding="utf-8") + self.node_info = Gio.DBusNodeInfo.new_for_xml(xml) + self.interface_info = self.node_info.interfaces[0] + self.loop = GLib.MainLoop() + self.connection: Gio.DBusConnection | None = None + self.registration_id = 0 + self.controller = Controller( + config or Config.from_environment(), + dispatch=lambda callback: GLib.idle_add(callback), + on_state=self._on_state, + on_transcript=self._on_transcript, + on_error=self._on_error, + ) + + def run(self) -> int: + owner_id = Gio.bus_own_name( + Gio.BusType.SESSION, + BUS_NAME, + Gio.BusNameOwnerFlags.NONE, + self._on_bus_acquired, + self._on_name_acquired, + self._on_name_lost, + ) + GLibUnix.signal_add(GLib.PRIORITY_DEFAULT, signal.SIGTERM, self._quit) + GLibUnix.signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, self._quit) + try: + self.loop.run() + finally: + self.controller.close() + if self.connection is not None and self.registration_id: + self.connection.unregister_object(self.registration_id) + Gio.bus_unown_name(owner_id) + return 0 + + def _on_bus_acquired(self, connection: Gio.DBusConnection, name: str) -> None: + self.connection = connection + self.registration_id = connection.register_object( + OBJECT_PATH, + self.interface_info, + self._method_call, + self._get_property, + None, + ) + + def _on_name_acquired(self, connection: Gio.DBusConnection, name: str) -> None: + print(f"Parley daemon owns {name}", file=sys.stderr) + + def _on_name_lost( + self, connection: Gio.DBusConnection | None, name: str + ) -> None: + print(f"error: could not own D-Bus name {name}", file=sys.stderr) + self.loop.quit() + + def _method_call( + self, + connection: Gio.DBusConnection, + sender: str, + object_path: str, + interface_name: str, + method_name: str, + parameters: GLib.Variant, + invocation: Gio.DBusMethodInvocation, + ) -> None: + try: + methods = { + "StartRecording": self.controller.start, + "StopAndTranscribe": self.controller.stop_and_transcribe, + "Toggle": self.controller.toggle, + "Cancel": self.controller.cancel, + "CopyLastTranscript": self._copy_last, + "InsertLastTranscript": self._insert_last, + "OpenTranscriptFolder": self._open_transcript_folder, + } + method = methods.get(method_name) + if method is None: + invocation.return_dbus_error( + f"{INTERFACE}.Error.UnknownMethod", f"unknown method: {method_name}" + ) + return + method() + invocation.return_value(None) + except (OSError, ParleyError) as error: + invocation.return_dbus_error(f"{INTERFACE}.Error.Failed", str(error)) + except Exception as error: + print(f"unexpected method failure: {error}", file=sys.stderr) + invocation.return_dbus_error( + f"{INTERFACE}.Error.Internal", "internal daemon error; see user journal" + ) + + def _get_property( + self, + connection: Gio.DBusConnection, + sender: str, + object_path: str, + interface_name: str, + property_name: str, + ) -> GLib.Variant: + values = { + "State": GLib.Variant("s", self.controller.state.value), + "LastTranscript": GLib.Variant("s", self.controller.last_transcript), + "LastTranscriptPath": GLib.Variant( + "s", + str(self.controller.last_transcript_path or ""), + ), + "LastError": GLib.Variant("s", self.controller.last_error), + "InsertionMode": GLib.Variant("s", "clipboard"), + "AutoInsert": GLib.Variant("b", False), + } + return values[property_name] + + def _copy_last(self) -> None: + if not self.controller.last_transcript_path: + raise ParleyError("no transcript is available") + copy_text(self.controller.last_transcript) + + def _insert_last(self) -> None: + try: + self._copy_last() + except ParleyError as error: + self._emit("InsertionFinished", GLib.Variant("(sbs)", ("clipboard", False, str(error)))) + raise + message = "copied; press Ctrl+V to insert" + self._emit( + "InsertionFinished", GLib.Variant("(sbs)", ("clipboard", True, message)) + ) + + def _open_transcript_folder(self) -> None: + directory = self.controller.config.transcript_dir + directory.mkdir(parents=True, exist_ok=True) + open_folder(directory) + + def _on_state(self, state: State) -> None: + self._emit("StateChanged", GLib.Variant("(s)", (state.value,))) + self._properties_changed("State", GLib.Variant("s", state.value)) + + def _on_transcript(self, text, path) -> None: + self._emit("TranscriptReady", GLib.Variant("(ss)", (text, str(path)))) + self._properties_changed( + "LastTranscript", GLib.Variant("s", text), + "LastTranscriptPath", GLib.Variant("s", str(path)), + ) + try: + copy_text(text) + except ParleyError as error: + self.controller.last_error = str(error) + self._on_error("clipboard-failed", str(error)) + notify("Transcription saved", f"Clipboard failed: {error}") + else: + notify("Transcription ready", "Copied to the clipboard") + + def _on_error(self, code: str, message: str) -> None: + self._emit("Error", GLib.Variant("(ss)", (code, message))) + self._properties_changed("LastError", GLib.Variant("s", message)) + notify("Parley error", message) + + def _properties_changed(self, *items) -> None: + changed = {items[index]: items[index + 1] for index in range(0, len(items), 2)} + if self.connection is not None: + self.connection.emit_signal( + None, + OBJECT_PATH, + "org.freedesktop.DBus.Properties", + "PropertiesChanged", + GLib.Variant("(sa{sv}as)", (INTERFACE, changed, [])), + ) + + def _emit(self, name: str, parameters: GLib.Variant) -> None: + if self.connection is not None: + self.connection.emit_signal( + None, OBJECT_PATH, INTERFACE, name, parameters + ) + + def _quit(self) -> bool: + self.loop.quit() + return GLib.SOURCE_REMOVE + + +def main() -> int: + return Daemon().run() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/parley/desktop.py b/src/parley/desktop.py new file mode 100644 index 0000000..f95e537 --- /dev/null +++ b/src/parley/desktop.py @@ -0,0 +1,53 @@ +"""Small desktop integrations used by the prototype daemon.""" + +from pathlib import Path +import shutil +import subprocess + +from .errors import ParleyError + + +class DesktopIntegrationError(ParleyError): + pass + + +def copy_text(text: str) -> None: + """Copy text with GDK, retaining ownership in the long-running daemon.""" + try: + import gi + + gi.require_version("Gdk", "4.0") + from gi.repository import Gdk + + display = Gdk.Display.get_default() or Gdk.Display.open(None) + if display is None: + raise DesktopIntegrationError("could not connect to the graphical display") + display.get_clipboard().set(text) + except DesktopIntegrationError: + raise + except Exception as error: + raise DesktopIntegrationError(f"GDK clipboard is unavailable: {error}") from error + + +def notify(summary: str, body: str) -> None: + executable = shutil.which("notify-send") + if executable is not None: + subprocess.Popen( + [executable, "--app-name", "Parley", summary, body], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def open_folder(path: Path) -> None: + executable = shutil.which("xdg-open") + if executable is None: + raise DesktopIntegrationError("xdg-open is not installed") + try: + subprocess.Popen( + [executable, str(path)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except OSError as error: + raise DesktopIntegrationError(f"could not open transcript folder: {error}") from error diff --git a/src/parley/interfaces/__init__.py b/src/parley/interfaces/__init__.py new file mode 100644 index 0000000..bded51d --- /dev/null +++ b/src/parley/interfaces/__init__.py @@ -0,0 +1 @@ +"""Packaged D-Bus interface definitions.""" diff --git a/src/parley/interfaces/org.parley.Transcription1.xml b/src/parley/interfaces/org.parley.Transcription1.xml new file mode 100644 index 0000000..12b7960 --- /dev/null +++ b/src/parley/interfaces/org.parley.Transcription1.xml @@ -0,0 +1,36 @@ +<?xml version="1.0" encoding="UTF-8"?> +<node> + <interface name="org.parley.Transcription1"> + <method name="StartRecording"/> + <method name="StopAndTranscribe"/> + <method name="Toggle"/> + <method name="Cancel"/> + <method name="CopyLastTranscript"/> + <method name="InsertLastTranscript"/> + <method name="OpenTranscriptFolder"/> + + <property name="State" type="s" access="read"/> + <property name="LastTranscript" type="s" access="read"/> + <property name="LastTranscriptPath" type="s" access="read"/> + <property name="LastError" type="s" access="read"/> + <property name="InsertionMode" type="s" access="read"/> + <property name="AutoInsert" type="b" access="read"/> + + <signal name="StateChanged"> + <arg name="state" type="s"/> + </signal> + <signal name="TranscriptReady"> + <arg name="text" type="s"/> + <arg name="path" type="s"/> + </signal> + <signal name="Error"> + <arg name="code" type="s"/> + <arg name="message" type="s"/> + </signal> + <signal name="InsertionFinished"> + <arg name="backend" type="s"/> + <arg name="success" type="b"/> + <arg name="message" type="s"/> + </signal> + </interface> +</node> diff --git a/src/parley/state.py b/src/parley/state.py new file mode 100644 index 0000000..8ec7a35 --- /dev/null +++ b/src/parley/state.py @@ -0,0 +1,62 @@ +"""The daemon's UI-independent operation state machine.""" + +from enum import StrEnum + +from .errors import ParleyError + + +class State(StrEnum): + IDLE = "idle" + RECORDING = "recording" + TRANSCRIBING = "transcribing" + INSERTING = "inserting" + ERROR = "error" + + +class InvalidStateError(ParleyError): + """An operation is not valid in the daemon's current state.""" + + +class StateMachine: + """Validate state transitions independently of D-Bus and subprocesses.""" + + def __init__(self) -> None: + self.state = State.IDLE + + def start_recording(self) -> State: + self._require(State.IDLE, State.ERROR) + return self._set(State.RECORDING) + + def begin_transcription(self) -> State: + self._require(State.RECORDING) + return self._set(State.TRANSCRIBING) + + def begin_insertion(self) -> State: + self._require(State.TRANSCRIBING) + return self._set(State.INSERTING) + + def complete(self) -> State: + self._require(State.TRANSCRIBING, State.INSERTING) + return self._set(State.IDLE) + + def cancel(self) -> State: + self._require(State.RECORDING, State.TRANSCRIBING) + return self._set(State.IDLE) + + def fail(self) -> State: + return self._set(State.ERROR) + + def reset(self) -> State: + self._require(State.ERROR) + return self._set(State.IDLE) + + def _require(self, *allowed: State) -> None: + if self.state not in allowed: + expected = ", ".join(value.value for value in allowed) + raise InvalidStateError( + f"operation requires state {expected}; current state is {self.state.value}" + ) + + def _set(self, state: State) -> State: + self.state = state + return state diff --git a/src/parley/transcription.py b/src/parley/transcription.py index eea0497..87ce66a 100644 --- a/src/parley/transcription.py +++ b/src/parley/transcription.py @@ -3,6 +3,7 @@ import json from pathlib import Path import subprocess +import threading from .config import Config from .errors import TranscriptionError @@ -28,8 +29,11 @@ def parse_jsonl(output: str) -> str: class Transcriber: - def __init__(self, config: Config) -> None: + def __init__(self, config: Config, *, cancel_timeout: float = 5.0) -> None: self.config = config + self.cancel_timeout = cancel_timeout + self._lock = threading.Lock() + self._process: subprocess.Popen[str] | None = None def transcribe(self, wav_path: Path, workspace: Path) -> str: batch_file = workspace / "input.txt" @@ -50,10 +54,31 @@ class Transcriber: "--batch-jsonl", ] try: - result = subprocess.run(command, text=True, capture_output=True) + process = subprocess.Popen(command, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except OSError as error: raise TranscriptionError(f"could not start transcription: {error}") from error - if result.returncode != 0: - detail = result.stderr.strip() or "transcription runtime failed" + with self._lock: + self._process = process + try: + stdout, stderr = process.communicate() + finally: + with self._lock: + if self._process is process: + self._process = None + if process.returncode != 0: + detail = stderr.strip() or "transcription runtime failed" raise TranscriptionError(detail) - return parse_jsonl(result.stdout) + return parse_jsonl(stdout) + + def cancel(self) -> None: + """Terminate an active runtime, escalating if it does not exit.""" + with self._lock: + process = self._process + if process is None or process.poll() is not None: + return + process.terminate() + try: + process.wait(timeout=self.cancel_timeout) + except subprocess.TimeoutExpired: + process.kill() + process.wait() |