Core API

The core package holds the voice-control engine: wake-word detection, transcription, command routing, speech output, the panel indicator, and the GNOME Shell extension lifecycle. Plugins receive an EasySpeak instance and use its small public API (speak, host_run, transcribe, wait_for_speech, record_until_silence, deactivate, ...) to act on commands.

core.main

core.main

EasySpeak Core - Voice Control for Linux.

Loads plugins from plugins/ folder automatically. Uses pyopen-wakeword for fast wake detection.

EasySpeak

EasySpeak()

The voice-control daemon: wake detection, transcription, and routing.

Owns the audio pipeline (wake word -> speech recognizer), the loaded plugins, the text-to-speech pipeline, and the panel indicator, and exposes the small plugin-facing API (speak, host_run, transcribe, ...) that plugins use to act on commands.

Initialise daemon state; models and audio are loaded later in run().

Source code in src/core/main.py

def __init__(self):
    """Initialise daemon state; models and audio are loaded later in run()."""
    self.plugins = []
    self.whisper = None
    self.parakeet = None
    self.wakeword = None
    self.audio = None
    self.stream = None
    self.last_wake_time = 0
    self.misunderstand_count = 0
    self.help_shown = False
    self.needs_wake_word = False
    self.unrecognized = False
    self.spoke = False
    # Set by listen_modal when the tray asks to quit from inside a plugin's
    # modal mode, so the request survives the unwind back to the main loop.
    self.exit_requested = False
    self.last_misunderstand_time = 0
    self.require_wake_word = REQUIRE_WAKE_WORD
    # Persistent text-to-speech pipeline (piper -> audio player) so the
    # voice model is loaded only once.
    self.speech = SpeechPipeline()
    # Panel indicator: owns the icon and the asleep lifecycle.
    self.tray = Tray(speak=self.speak)
    # Hold-to-dictate keyboard activation; the dictation plugin registers
    # the session to run while the combo is held (see register_push_to_talk).
    self.hotkey = HotkeyListener(HOTKEY_COMBO)
    self._push_to_talk = None
    # Remembered after the first sound plays, so later ones cost one process.
    self._sound_player = None
    # Raised to suit the room by calibrate_silence() once the mic is open.
    self.silence_threshold = SILENCE_THRESHOLD

host_run

host_run(cmd, background=False, clean_env=False)

Run a shell command.

With clean_env, EasySpeak's injected LD_LIBRARY_PATH and GI_TYPELIB_PATH are stripped from the child's environment. The dev flake prepends its own libraries (glib among them) to those paths for EasySpeak's own dependencies; left in place they leak into spawned desktop apps, which then load EasySpeak's flake-pinned libraries instead of their own rpath ones. A glib/GIO build mismatch there wedges GTK apps such as the file manager. Pass clean_env=True when launching external GUI programs so they run in the plain host environment.

A background program gets its own session, so pressing Ctrl+C in the terminal EasySpeak runs in quits EasySpeak only, not the apps it opened.

Source code in src/core/main.py

def host_run(self, cmd, background=False, clean_env=False):
    """Run a shell command.

    With clean_env, EasySpeak's injected LD_LIBRARY_PATH and GI_TYPELIB_PATH
    are stripped from the child's environment. The dev flake prepends its own
    libraries (glib among them) to those paths for EasySpeak's own
    dependencies; left in place they leak into spawned desktop apps, which
    then load EasySpeak's flake-pinned libraries instead of their own rpath
    ones. A glib/GIO build mismatch there wedges GTK apps such as the file
    manager. Pass clean_env=True when launching external GUI programs so they
    run in the plain host environment.

    A background program gets its own session, so pressing Ctrl+C in the terminal
    EasySpeak runs in quits EasySpeak only, not the apps it opened.
    """
    env = None
    if clean_env:
        env = {
            k: v
            for k, v in os.environ.items()
            if k not in ("LD_LIBRARY_PATH", "GI_TYPELIB_PATH")
        }
    if background:
        return subprocess.Popen(
            cmd,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            env=env,
            start_new_session=True,
        )
    return subprocess.run(cmd, capture_output=True, text=True, env=env)

speak

speak(text)

Speak a phrase.

Stable plugin-facing API; delegates to the pipeline.

Source code in src/core/main.py

def speak(self, text):
    """Speak a phrase.

    Stable plugin-facing API; delegates to the pipeline.
    """
    self.spoke = True
    self.speech.speak(text)

tap_key

tap_key(keycode)

Replay a multimedia key so the desktop renders its native feedback.

Returns True if the key was injected, False if unavailable (e.g. a non-GNOME session) so the caller can fall back to a silent change. jeepney is imported lazily so the dependency isn't needed to load.

Source code in src/core/main.py

def tap_key(self, keycode):
    """Replay a multimedia key so the desktop renders its native feedback.

    Returns True if the key was injected, False if unavailable (e.g. a non-GNOME
    session) so the caller can fall back to a silent change. jeepney is imported
    lazily so the dependency isn't needed to load.
    """
    try:
        from . import mediakeys

        mediakeys.tap_key(keycode)
        return True
    except Exception:
        return False

deactivate

deactivate()

Request the assistant go to sleep (plugin-facing).

Releases the mic and stops wake detection until reactivated from the tray. The actual release happens at the listening loop's next tray poll (handled by the tray controller), right after the triggering command has finished and its reply has been drained, so the follow-up window does not stay open on it.

Source code in src/core/main.py

def deactivate(self):
    """Request the assistant go to sleep (plugin-facing).

    Releases the mic and stops wake detection until reactivated from the tray. The
    actual release happens at the listening loop's next tray poll (handled by the
    tray controller), right after the triggering command has finished and its
    reply has been drained, so the follow-up window does not stay open on it.
    """
    self.tray.request_sleep()

register_push_to_talk

register_push_to_talk(handler)

Register the dictation session the hotkey runs while its combo is held.

Plugin-facing: the dictation plugin registers here in its setup() so core can drive keyboard (silent) activation without importing a plugin directly. handler takes one should_continue predicate and runs until it returns False (the keys are released).

Source code in src/core/main.py

def register_push_to_talk(self, handler):
    """Register the dictation session the hotkey runs while its combo is held.

    Plugin-facing: the dictation plugin registers here in its setup() so core
    can drive keyboard (silent) activation without importing a plugin
    directly. `handler` takes one `should_continue` predicate and runs
    until it returns False (the keys are released).
    """
    self._push_to_talk = handler

load_plugins

load_plugins()

Discover and import every plugin from the plugins/ dir.

A plugin is a package (a directory with __init__.py, and a locale/ for its translations) or a plain module; names starting with _ are skipped. A plugin is registered only if it exposes NAME and handle; its optional setup hook runs once. Import or setup failures are logged and skipped, never fatal.

Commands are routed to the plugins in order of their PRIORITY (default 50, lower first; equal ones by name), so a mode that must see a command first declares a low one, and the catch-all for help and exit a high one.

Source code in src/core/main.py

def load_plugins(self):
    """Discover and import every plugin from the `plugins/` dir.

    A plugin is a package (a directory with `__init__.py`, and a `locale/` for
    its translations) or a plain module; names starting with `_` are skipped. A
    plugin is registered only if it exposes `NAME` and `handle`; its optional
    `setup` hook runs once. Import or setup failures are logged and skipped,
    never fatal.

    Commands are routed to the plugins in order of their `PRIORITY` (default
    50, lower first; equal ones by name), so a mode that must see a command
    first declares a low one, and the catch-all for help and exit a high one.
    """
    plugins_dir = Path(__file__).parent.parent / "plugins"
    if not plugins_dir.exists():
        logger.warning("No plugins directory found")
        return

    sys.path.insert(0, str(plugins_dir.parent))

    loaded = []
    for file in sorted(plugins_dir.iterdir()):
        is_package = file.is_dir() and (file / "__init__.py").is_file()
        if file.name.startswith("_") or not (is_package or file.suffix == ".py"):
            continue

        module_name = f"plugins.{file.stem}"
        try:
            module = importlib.import_module(module_name)
        except Exception as e:
            logger.warning("  ✗ Failed to load %s: %s", file.name, e)
            continue
        if hasattr(module, "NAME") and hasattr(module, "handle"):
            loaded.append(module)
        else:
            logger.warning(
                "  ✗ Invalid plugin: %s (missing NAME or handle)", file.name
            )

    loaded.sort(key=lambda m: (getattr(m, "PRIORITY", 50), m.NAME))
    for module in loaded:
        try:
            if hasattr(module, "setup"):
                module.setup(self)
        except Exception as e:
            logger.warning("  ✗ Failed to load %s: %s", module.NAME, e)
            continue
        self.plugins.append(module)
        logger.info("  ✓ Loaded: %s", module.NAME)

get_all_commands

get_all_commands()

Get all commands from all plugins for help text.

Source code in src/core/main.py

def get_all_commands(self):
    """Get all commands from all plugins for help text."""
    commands = []
    for plugin in self.plugins:
        if hasattr(plugin, "COMMANDS"):
            commands.extend(plugin.COMMANDS)
    return commands

route_command

route_command(cmd)

Route command to appropriate plugin.

A plugin that raises doesn't stop the others from trying the command. If none takes it, the command was understood but carrying it out broke, so the user hears an apology rather than "I didn't understand". Returns False to exit.

Source code in src/core/main.py

def route_command(self, cmd):
    """Route command to appropriate plugin.

    A plugin that raises doesn't stop the others from trying the command. If none
    takes it, the command was understood but carrying it out broke, so the user
    hears an apology rather than "I didn't understand". Returns False to exit.
    """
    cmd = strip_wake_words(cmd).lower()
    self.unrecognized = False

    if not cmd:
        return True

    failed = False
    for plugin in self.plugins:
        try:
            result = plugin.handle(cmd, self)
            if result is True:
                self.misunderstand_count = 0
                self.help_shown = False
                return True
            if result is False:
                return False
        except Exception:
            logger.exception("Plugin error (%s)", plugin.NAME)
            failed = True

    if failed:
        self.misunderstand_count = 0
        self.help_shown = False
        self.speak(_("Sorry, something went wrong with that command."))
        return True
    self._report_not_understood()
    return True

flush_stream

flush_stream()

Flush any remaining audio data from the stream buffer.

Source code in src/core/main.py

def flush_stream(self):
    """Flush any remaining audio data from the stream buffer."""
    # intentionally suppress everything to prevent cleanup failures
    with contextlib.suppress(Exception):
        self.stream.read(
            self.stream.get_read_available(),
            exception_on_overflow=False,
        )

calibrate_silence

calibrate_silence()

Measure the room's noise floor and set the silence threshold above it.

A fixed threshold assumes a quiet room. Where the ambient level sits above it -- a fan, a desktop, an air conditioner -- is_silence never returns True, so nothing ever stops recording early: a three-word command took the whole five-second cap and a dictated sentence took twenty, which is most of the wait between speaking and seeing text.

Uses the median chunk level so a cough or a passing car doesn't skew the measurement, and clamps the result: never below the configured floor, and never so high that speech itself would read as silence.

Source code in src/core/main.py

def calibrate_silence(self):
    """Measure the room's noise floor and set the silence threshold above it.

    A fixed threshold assumes a quiet room. Where the ambient level sits above
    it -- a fan, a desktop, an air conditioner -- `is_silence` never returns
    True, so nothing ever stops recording early: a three-word command took the
    whole five-second cap and a dictated sentence took twenty, which is most of
    the wait between speaking and seeing text.

    Uses the median chunk level so a cough or a passing car doesn't skew the
    measurement, and clamps the result: never below the configured floor, and
    never so high that speech itself would read as silence.
    """
    override = os.environ.get("EASYSPEAK_SILENCE_THRESHOLD")
    if override:
        with contextlib.suppress(ValueError):
            self.silence_threshold = int(override)
            logger.info(
                "Silence threshold %d (from environment)", self.silence_threshold
            )
            return

    chunks = []
    for _tick in range(int(SILENCE_CALIBRATION_SECONDS * 16000 / 1600)):
        with contextlib.suppress(Exception):
            pcm = self.stream.read(1600, exception_on_overflow=False)
            chunks.append(np.abs(np.frombuffer(pcm, dtype=np.int16)).mean())

    if not chunks:
        return  # no audio to measure; the floor stands

    floor = float(np.median(chunks))
    self.silence_threshold = int(
        min(
            max(floor * SILENCE_NOISE_MARGIN, SILENCE_THRESHOLD),
            SILENCE_THRESHOLD_MAX,
        )
    )
    logger.info(
        "Room noise floor %.0f; silence threshold %d",
        floor,
        self.silence_threshold,
    )

is_silence

is_silence(audio_chunk)

Return True if the audio chunk's mean amplitude is below the threshold.

Source code in src/core/main.py

def is_silence(self, audio_chunk):
    """Return True if the audio chunk's mean amplitude is below the threshold."""
    return np.abs(audio_chunk).mean() < self.silence_threshold

record_until_silence

record_until_silence(should_continue=None, max_seconds=None, silence_duration=None)

Record mic audio until a short silence, or the cap. Plugin-facing.

Returns the captured PCM bytes. should_continue (used by push-to-talk) lets a key release cut the recording short instead of waiting out the silence window.

Both limits are adjustable because a command and a dictated sentence need very different ones: the command defaults stopped recording during the pause in the middle of a sentence, and truncated anything past five seconds.

Source code in src/core/main.py

def record_until_silence(
    self, should_continue=None, max_seconds=None, silence_duration=None
):
    """Record mic audio until a short silence, or the cap. Plugin-facing.

    Returns the captured PCM bytes. `should_continue` (used by push-to-talk) lets
    a key release cut the recording short instead of waiting out the silence
    window.

    Both limits are adjustable because a command and a dictated sentence need
    very different ones: the command defaults stopped recording during the pause
    in the middle of a sentence, and truncated anything past five seconds.
    """
    max_seconds = MAX_RECORD_SECONDS if max_seconds is None else max_seconds
    if silence_duration is None:
        silence_duration = SILENCE_DURATION

    frames = []
    silent_chunks = 0
    chunks_needed = int(silence_duration * 16000 / 1600)

    for i in range(int(max_seconds * 16000 / 1600)):
        if should_continue is not None and not should_continue():
            break
        pcm = self.stream.read(1600, exception_on_overflow=False)
        frames.append(pcm)

        if i >= 5:
            if self.is_silence(np.frombuffer(pcm, dtype=np.int16)):
                silent_chunks += 1
                if silent_chunks >= chunks_needed:
                    break
            else:
                silent_chunks = 0

    return b"".join(frames)

wait_for_speech

wait_for_speech(timeout=5, should_continue=None)

Block until speech is heard, returning its first PCM chunk.

Returns None if nothing is heard within timeout seconds. Plugin-facing. should_continue (used by push-to-talk) returns None early once it goes False, so a key release ends the wait.

Source code in src/core/main.py

def wait_for_speech(self, timeout=5, should_continue=None):
    """Block until speech is heard, returning its first PCM chunk.

    Returns None if nothing is heard within `timeout` seconds. Plugin-facing.
    `should_continue` (used by push-to-talk) returns None early once it goes False,
    so a key release ends the wait.
    """
    for _tick in range(int(timeout * 16000 / 1600)):
        if should_continue is not None and not should_continue():
            return None
        pcm = self.stream.read(1600, exception_on_overflow=False)
        if not self.is_silence(np.frombuffer(pcm, dtype=np.int16)):
            return pcm
    return None

transcribe

transcribe(audio_data, prompt=None, language='en')

Transcribe raw PCM audio to text with the configured speech model.

prompt biases recognition (defaults to the command vocabulary), and language is the Whisper language code to transcribe as. The main loop passes COMMAND_LANGUAGE and dictation the user's LANGUAGE; the modes, whose words are still English, take the English default. Plugin-facing. An echo of that prompt is dropped as silence: Whisper hands its own initial_prompt back when given near-silence, and grid mode was executing that as a command. Parakeet (EASYSPEAK_STT=parakeet) takes neither a prompt nor a language: it tells the language apart itself.

Source code in src/core/main.py

def transcribe(self, audio_data, prompt=None, language="en"):
    """Transcribe raw PCM audio to text with the configured speech model.

    `prompt` biases recognition (defaults to the command vocabulary), and
    `language` is the Whisper language code to transcribe as. The main loop
    passes `COMMAND_LANGUAGE` and dictation the user's `LANGUAGE`; the modes,
    whose words are still English, take the English default. Plugin-facing.
    An echo of that prompt is dropped as silence: Whisper hands its own
    `initial_prompt` back when given near-silence, and grid mode was executing
    that as a command. Parakeet (`EASYSPEAK_STT=parakeet`) takes neither a
    prompt nor a language: it tells the language apart itself.
    """
    samples = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0

    # The default prompt names the commands in the language being decoded,
    # so a mode still decoding English is not primed with German words.
    use_prompt = prompt or COMMAND_PROMPTS.get(language, COMMAND_PROMPT)
    started = time.monotonic()
    if self.parakeet is not None:
        text = self.parakeet.recognize(samples).strip()
    else:
        segments, _ = self.whisper.transcribe(
            samples,
            initial_prompt=use_prompt,
            beam_size=1,
            vad_filter=True,
            language=language,
            # Nothing here reads timestamps, and generating them costs tokens.
            without_timestamps=True,
            condition_on_previous_text=False,
        )
        text = " ".join([s.text for s in segments]).strip()
    logger.debug(
        "transcribed %.1fs of audio in %.2fs",
        len(audio_data) / 32000,
        time.monotonic() - started,
    )

    if text and self.parakeet is None and self._is_prompt_echo(text, use_prompt):
        logger.debug("Ignoring prompt echo: %s", text)
        return ""
    return text

listen_modal

listen_modal(label, prompt=None, timeout=10, idle_timeout=30, max_record_seconds=None, silence_duration=None, wake_gated=True, language='en')

Yield transcribed commands for a plugin's modal mode. Plugin-facing.

Plugins that take over the microphone (grid, browser, dictation, head tracking) each ran their own while loop, which starved everything the main loop owns: the tray was never polled, so its Quit and Mute entries and the Quick Settings toggle did nothing; and with no idle limit a mode never ended on its own, so an unattended session stayed in it indefinitely with the wake word unreachable. Driving the loop from here keeps those alive — the tray is polled between utterances, and the mode ends idle_timeout seconds after the last command it understood.

The deadline is wall-clock and only a recognised command pushes it back. Counting silent rounds instead doesn't work in a real room: any sound over the threshold ends the wait early, so ambient noise both resets a round-based counter forever and makes each round take an unpredictable amount of time.

label names the mode in the exit log line and the spoken notice, and prompt biases Whisper towards that mode's vocabulary. timeout is how long a single listen waits for speech, and idle_timeout how long the mode survives without a recognised command. max_record_seconds and silence_duration fall back to the command defaults when None; dictation passes larger values, since a sentence has pauses in it and runs longer than a command. wake_gated is False for modes that capture continuous speech rather than commands, so require_wake_word does not ask for the wake word before every dictated sentence. language is passed on to transcribe.

Yields each recognised command stripped of the wake word and surrounding punctuation, in the case Whisper wrote it: dictation inserts it as is, the modes lowercase it to match their words. The generator simply stops when the mode should end, so a caller's for loop falls through to its own cleanup.

Source code in src/core/main.py

def listen_modal(
    self,
    label,
    prompt=None,
    timeout=10,
    idle_timeout=30,
    max_record_seconds=None,
    silence_duration=None,
    wake_gated=True,
    language="en",
):
    """Yield transcribed commands for a plugin's modal mode. Plugin-facing.

    Plugins that take over the microphone (grid, browser, dictation, head
    tracking) each ran their own `while` loop, which starved everything the main
    loop owns: the tray was never polled, so its Quit and Mute entries and the
    Quick Settings toggle did nothing; and with no idle limit a mode never ended
    on its own, so an unattended session stayed in it indefinitely with the wake
    word unreachable. Driving the loop from here keeps those alive — the tray is
    polled between utterances, and the mode ends `idle_timeout` seconds after the
    last command it understood.

    The deadline is wall-clock and only a recognised command pushes it back.
    Counting silent *rounds* instead doesn't work in a real room: any sound over
    the threshold ends the wait early, so ambient noise both resets a round-based
    counter forever and makes each round take an unpredictable amount of time.

    `label` names the mode in the exit log line and the spoken notice, and
    `prompt` biases Whisper towards that mode's vocabulary. `timeout` is how
    long a single listen waits for speech, and `idle_timeout` how long the mode
    survives without a recognised command. `max_record_seconds` and
    `silence_duration` fall back to the command defaults when None; dictation
    passes larger values, since a sentence has pauses in it and runs longer
    than a command. `wake_gated` is False for modes that capture continuous
    speech rather than commands, so `require_wake_word` does not ask for the
    wake word before every dictated sentence. `language` is passed on to
    [`transcribe`][core.main.EasySpeak.transcribe].

    Yields each recognised command stripped of the wake word and surrounding
    punctuation, in the case Whisper wrote it: dictation inserts it as is,
    the modes lowercase it to match their words. The generator simply stops
    when the mode should end, so a caller's `for` loop falls through to its
    own cleanup.
    """
    if self.spoke:
        self._drain_feedback()
        self.spoke = False

    deadline = time.monotonic() + idle_timeout
    while True:
        # The same poll the wake-word loop runs, so sleep and quit still work
        # while a mode holds the microphone.
        action = self.tray.poll(self._close_stream, self._open_stream)
        if action is TrayAction.QUIT:
            self.exit_requested = True
            return
        if action is TrayAction.RESUME:
            logger.info("%s mode ended: reactivated", label.capitalize())
            self._leave_unattended(label)
            return

        remaining = deadline - time.monotonic()
        if remaining <= 0:
            logger.info("%s mode ended: idle", label.capitalize())
            self._leave_unattended(label)
            return

        if (
            wake_gated
            and self.require_wake_word
            and not self.wait_for_wake(timeout=min(timeout, remaining))
        ):
            continue

        self.flush_stream()
        # Never wait past the deadline, so the mode ends on time even when the
        # room is quiet enough that every listen runs its full length.
        first = self.wait_for_speech(timeout=min(timeout, remaining))
        if first is None:
            continue

        rest = self.record_until_silence(
            max_seconds=max_record_seconds, silence_duration=silence_duration
        )
        text = self.transcribe(first + rest, prompt=prompt, language=language)
        if not text:
            # Noise, or an echo of our own prompt. Not a command, so the
            # deadline stands.
            continue

        spoken = strip_wake_words(text)
        if not spoken:
            continue  # the wake word on its own is not a command

        deadline = time.monotonic() + idle_timeout
        self.spoke = False
        yield spoken

        if self.spoke:
            self._drain_feedback()
            self.spoke = False
            deadline = time.monotonic() + idle_timeout

wait_for_wake

wait_for_wake(timeout)

Block until the wake word is heard, then chime. False on timeout.

The same detector, cooldown and chime the main loop uses, so a mode that asks for the wake word behaves exactly like the wake-word loop does.

Source code in src/core/main.py

def wait_for_wake(self, timeout):
    """Block until the wake word is heard, then chime. False on timeout.

    The same detector, cooldown and chime the main loop uses, so a mode that
    asks for the wake word behaves exactly like the wake-word loop does.
    """
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        pcm = self.stream.read(1280, exception_on_overflow=False)
        score = self.wakeword.predict(pcm)
        if score <= WAKE_THRESHOLD:
            continue
        now = time.time()
        if now - self.last_wake_time < WAKE_COOLDOWN:
            continue
        self.last_wake_time = now
        logger.info("🎤 Wake! (confidence: %.2f)", score)
        self._reset_detector()
        self._play_wake_chime()
        return True
    return False

play_sound

play_sound(sound)

Play a short desktop sound, best-effort. Plugin-facing.

The daemon is voice-first, so an audible cue is often the only signal a user gets -- especially with no Piper voice installed, where every spoken reply is dropped. Failures are never fatal: this used to be a bare paplay call, so a machine without pulseaudio-utils took the whole daemon down on the first wake word. The working player is remembered after the first success, so a mode change costs one process, not three.

Source code in src/core/main.py

def play_sound(self, sound):
    """Play a short desktop sound, best-effort. Plugin-facing.

    The daemon is voice-first, so an audible cue is often the only signal a
    user gets -- especially with no Piper voice installed, where every spoken
    reply is dropped. Failures are never fatal: this used to be a bare
    `paplay` call, so a machine without pulseaudio-utils took the whole daemon
    down on the first wake word. The working player is remembered after the
    first success, so a mode change costs one process, not three.
    """
    if not Path(sound).exists():
        logger.debug("sound file not found: %s", sound)
        return False
    players = [self._sound_player] if self._sound_player else SOUND_PLAYERS
    for player in players:
        try:
            result = subprocess.run(
                [*player, str(sound)], capture_output=True, timeout=SOUND_TIMEOUT
            )
        except (OSError, subprocess.TimeoutExpired):
            continue
        if result.returncode == 0:
            self._sound_player = player
            return True
    self._sound_player = None
    logger.debug("no working audio player for %s", sound)
    return False

run

run()

Load models and plugins, then run the wake-word listen loop forever.

Blocks until the user quits (voice command, tray, or Ctrl-C), always releasing the microphone and draining speech on the way out.

Source code in src/core/main.py

    def run(self):
        """Load models and plugins, then run the wake-word listen loop forever.

        Blocks until the user quits (voice command, tray, or Ctrl-C), always releasing
        the microphone and draining speech on the way out.
        """
        logger.info("Loading OpenWakeWord...")
        self.wakeword = WakeWordModel()

        try:
            if STT == "parakeet":
                logger.info(
                    "Loading Parakeet (%s, %s, language=%s)...",
                    PARAKEET_MODEL,
                    PARAKEET_QUANTIZATION,
                    LANGUAGE,
                )
                self.parakeet = load_parakeet_model()
            else:
                logger.info(
                    "Loading Whisper (%s, %s, language=%s, cpu_threads=%s)...",
                    WHISPER_MODEL,
                    WHISPER_COMPUTE_TYPE,
                    LANGUAGE,
                    WHISPER_CPU_THREADS or "auto",
                )
                self.whisper = load_whisper_model()
        except RuntimeError as exc:
            logger.error("Cannot start EasySpeak: %s", exc)  # noqa: TRY400
            raise SystemExit(1) from exc
        for warning in (*STT_WARNINGS, *LANGUAGE_WARNINGS):
            logger.warning(warning)

        ensure_extension()

        logger.info("\nLoading plugins...")
        self.load_plugins()

        if not self.plugins:
            logger.error("No plugins loaded. Exiting.")
            return

        # Warm up the text-to-speech pipeline now so the piper model is loaded
        # during startup, not on the first spoken response.
        logger.info("Warming up speech...")
        try:
            self.speech.ensure()
        except OSError:
            logger.warning("Speech unavailable; continuing without it.")

        logger.info("""
╔══════════════════════════════════════════╗
║            EasySpeak                     ║
╠══════════════════════════════════════════╣
║  Wake word: "Hey Jarvis"                 ║
║  Say "help" for available commands       ║
╚══════════════════════════════════════════╝
""")

        try:
            with suppressed_c_stderr():
                self.audio = pyaudio.PyAudio()
            self._open_stream()

            # Measure the room before listening, so silence detection matches it.
            self.calibrate_silence()

            self.tray.started()
            # Start keyboard (silent) activation; no-op if disabled or no
            # /dev/input access. Plugins have registered their handlers by now.
            self.hotkey.start()
            logger.info("Listening for wake word...")
            audio_buffer = []

            while True:
                # The tray controller owns sleep/quit; it releases and reopens
                # the mic via these callbacks so this loop stays about audio.
                action = self.tray.poll(self._close_stream, self._open_stream)
                if action is TrayAction.QUIT:
                    break
                if action is TrayAction.RESUME:
                    self._reset_detector()
                    audio_buffer = []
                    logger.info("Listening for wake word...")
                    continue

                # Keyboard activation bypasses the wake word: dictate while held.
                if self.hotkey.take_activation():
                    self._run_push_to_talk()
                    self._reset_detector()
                    audio_buffer = []
                    logger.info("Listening for wake word...")
                    continue

                pcm = self.stream.read(1280, exception_on_overflow=False)

                audio_buffer.append(pcm)
                if len(audio_buffer) > 50:
                    audio_buffer.pop(0)

                score = self.wakeword.predict(pcm)

                if score > WAKE_THRESHOLD:
                    now = time.time()
                    if now - self.last_wake_time < WAKE_COOLDOWN:
                        continue
                    self.last_wake_time = now

                    logger.info("🎤 Wake! (confidence: %.2f)", score)

                    self._reset_detector()
                    audio_buffer = []
                    self._play_wake_chime()

                    if self._capture_command_session():
                        break

                    audio_buffer = []
                    logger.info("Listening for wake word...")

        except KeyboardInterrupt:
            logger.info("\nStopped (Ctrl+C).")
        finally:
            # Hide the indicator so the daemon's exit doesn't leave a stale icon.
            self.tray.stopped()
            self.hotkey.stop()
            self.speech.drain()
            if self.stream is not None:
                self.stream.stop_stream()
                self.stream.close()
            if self.audio is not None:
                self.audio.terminate()

strip_wake_words

strip_wake_words(cmd)

Remove a leading spoken wake word and surrounding punctuation.

Only a prefix is removed. Replacing every occurrence anywhere in the utterance -- which is what this did -- eats the word out of the middle of a command, so "search jarvis" became "search" and a URL containing it lost part of itself. Longest prefix first, so "hey jarvis," is matched before "jarvis". The case of what remains is kept: dictation inserts it as spoken, and every other consumer lowercases for itself.

Source code in src/core/main.py

def strip_wake_words(cmd):
    """Remove a leading spoken wake word and surrounding punctuation.

    Only a prefix is removed. Replacing every occurrence anywhere in the utterance
    -- which is what this did -- eats the word out of the middle of a command, so
    "search jarvis" became "search" and a URL containing it lost part of itself.
    Longest prefix first, so "hey jarvis," is matched before "jarvis". The case
    of what remains is kept: dictation inserts it as spoken, and every other
    consumer lowercases for itself.
    """
    cmd = cmd.strip()
    lowered = cmd.lower()
    for wake in sorted(WAKE_PREFIXES, key=len, reverse=True):
        if lowered == wake:
            return ""
        if lowered.startswith(wake):
            rest = cmd[len(wake) :]
            # Only a prefix if a word actually ends here.
            if not rest or rest[0] in " ,.!?":
                cmd = rest.strip()
                break
    return cmd.strip(".,!? ")

core.cli

core.cli

Command-line entry point: parse arguments, set up logging, run the app.

Verbosity comes from the mutually exclusive -v/--verbose and -q/--quiet flags, or — with neither — the EASYSPEAK_LOG_LEVEL environment variable, which the flags override (see resolve_level). --language is the command-line form of EASYSPEAK_LANGUAGE and overrides it. The one-shot --configure and --preview subcommands set up or print the desktop-integration files and exit. All EASYSPEAK_* variables are listed in core.config.

parse_args

parse_args(argv=None)

Parse EasySpeak's command-line arguments.

Source code in src/core/cli.py

def parse_args(argv=None):
    """Parse EasySpeak's command-line arguments."""
    config_items = {"extension", "service", "autostart", "desktop"}
    config_defaults = {"extension", "service"}

    parser = argparse.ArgumentParser(
        prog="easyspeak", description="Voice control for Linux desktops."
    )
    verbosity = parser.add_mutually_exclusive_group()
    verbosity.add_argument(
        "-q", "--quiet", action="store_true", help="show only warnings and errors"
    )
    verbosity.add_argument(
        "-v", "--verbose", action="store_true", help="show debug output"
    )
    parser.add_argument(
        "--configure",
        nargs="*",
        choices=config_items,
        help=(
            "set up the listed activities and exit "
            f"(default: {' '.join(config_defaults)})"
        ),
    )
    parser.add_argument(
        "--preview",
        choices=config_items,
        help="print the file content that would be configured and exit",
    )
    parser.add_argument(
        "--language",
        metavar="CODE",
        help=(
            "language to dictate and be answered in, e.g. de; "
            "alternatively set EASYSPEAK_LANGUAGE (default: en)"
        ),
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {app_version()}",
        help="show the installed version and exit",
    )
    args = parser.parse_args(argv)

    if args.configure is not None:
        args.configure = set(args.configure) or config_defaults

    return args

run

run(argv=None)

Start the application, or run a one-shot subcommand and exit.

EasySpeak is imported lazily so the one-shot subcommands work without the audio stack (pyopen-wakeword, pyaudio, …) that core.main pulls in. Ctrl+C ends EasySpeak quietly even while that stack or its models are still loading.

Source code in src/core/cli.py

def run(argv=None):
    """Start the application, or run a one-shot subcommand and exit.

    `EasySpeak` is imported lazily so the one-shot subcommands work without the
    audio stack (pyopen-wakeword, pyaudio, …) that `core.main` pulls in. Ctrl+C
    ends EasySpeak quietly even while that stack or its models are still loading.
    """
    args = parse_args(argv)

    if args.preview is not None:
        from .desktop_integration import preview

        preview(args.preview)
        return

    log.configure(log.resolve_level(verbose=args.verbose, quiet=args.quiet))

    if args.language:
        # The config reads the variable when it is imported, below.
        os.environ["EASYSPEAK_LANGUAGE"] = args.language

    if args.configure is not None:
        from .desktop_integration import configure

        configure(args.configure)
        return

    try:
        from .main import EasySpeak

        EasySpeak().run()
    except KeyboardInterrupt:
        logger.info("\nStopped (Ctrl+C).")

core.config

core.config

Central tuning constants and the speech-model factory for EasySpeak.

Holds the wake-word, audio, silent-hotkey, speech-model, and desktop-sound settings the daemon reads at import, alongside load_parakeet_model() and load_whisper_model(), which build the speech model from them. Most are plain constants; these honor an EASYSPEAK_* environment variable:

• EASYSPEAK_HOTKEY
• EASYSPEAK_LANGUAGE
• EASYSPEAK_MODELS_DIR
• EASYSPEAK_OFFLINE
• EASYSPEAK_PIPER_BIN
• EASYSPEAK_PIPER_MODEL
• EASYSPEAK_SOUNDS_DIR
• EASYSPEAK_STT
• EASYSPEAK_WHISPER_COMPUTE_TYPE
• EASYSPEAK_WHISPER_CPU_THREADS
• EASYSPEAK_WHISPER_MODEL

See the Configuration guide for what each does, its default, and EasySpeak's other EASYSPEAK_* variables.

A plugin that needs host-environment setup does it in its own setup() hook — see Writing Plugins.

parakeet_installed

parakeet_installed(model_dir=PARAKEET_MODEL_DIR)

Whether every file of the Parakeet model is in model_dir.

A directory left behind by an interrupted download has some of them, and counts as not installed.

Source code in src/core/config.py

def parakeet_installed(model_dir: Path = PARAKEET_MODEL_DIR) -> bool:
    """Whether every file of the Parakeet model is in `model_dir`.

    A directory left behind by an interrupted download has some of them, and
    counts as not installed.
    """
    return all((model_dir / name).is_file() for name in PARAKEET_FILES)

load_whisper_model

load_whisper_model(model_name=WHISPER_MODEL, compute_type=WHISPER_COMPUTE_TYPE, cpu_threads=WHISPER_CPU_THREADS)

Build a faster-whisper model from the configured (or given) settings.

A model already on disk — bundled in a language pack or cached from an earlier run — loads without any network access. When it is missing, EasySpeak stays offline by default and raises an actionable message; setting EASYSPEAK_OFFLINE=relaxed (see NETWORK_ALLOWED) lets it fetch a bare name like base.en from Hugging Face instead.

Source code in src/core/config.py

def load_whisper_model(
    model_name: str = WHISPER_MODEL,
    compute_type: str = WHISPER_COMPUTE_TYPE,
    cpu_threads: int = WHISPER_CPU_THREADS,
) -> WhisperModel:
    """Build a faster-whisper model from the configured (or given) settings.

    A model already on disk — bundled in a language pack or cached from an earlier
    run — loads without any network access. When it is missing, EasySpeak stays
    offline by default and raises an actionable message; setting
    `EASYSPEAK_OFFLINE=relaxed` (see `NETWORK_ALLOWED`) lets it fetch a bare name
    like `base.en` from Hugging Face instead.
    """
    kwargs = {"compute_type": compute_type, "cpu_threads": cpu_threads}
    try:
        return WhisperModel(model_name, local_files_only=True, **kwargs)
    except FileNotFoundError:
        if not NETWORK_ALLOWED:
            msg = (
                f"speech model {model_name!r} is not installed; set "
                "EASYSPEAK_OFFLINE=relaxed to download it, or install a language pack"
            )
            raise RuntimeError(msg) from None
        logger.warning(
            "Speech model %r is not installed; downloading it from Hugging Face. "
            "Install a language pack to avoid this, or set EASYSPEAK_OFFLINE=strict "
            "to keep EasySpeak offline.",
            model_name,
        )
        return WhisperModel(model_name, local_files_only=False, **kwargs)

load_parakeet_model

load_parakeet_model(model_dir=PARAKEET_MODEL_DIR)

Build the Parakeet TDT v3 recognizer from the ONNX files in model_dir.

A directory holding the whole model loads offline; a missing one is fetched from Hugging Face into it when EASYSPEAK_OFFLINE=relaxed, and refused otherwise, like the Whisper models. An incomplete one, left by an interrupted download, is reported for the user to delete: onnx-asr never downloads into a directory that exists. Whatever else onnx-asr refuses to load is reported the same way, as a RuntimeError the daemon turns into a clean exit.

Source code in src/core/config.py

def load_parakeet_model(model_dir: Path = PARAKEET_MODEL_DIR) -> Any:
    """Build the Parakeet TDT v3 recognizer from the ONNX files in `model_dir`.

    A directory holding the whole model loads offline; a missing one is fetched
    from Hugging Face into it when `EASYSPEAK_OFFLINE=relaxed`, and refused
    otherwise, like the Whisper models. An incomplete one, left by an interrupted
    download, is reported for the user to delete: onnx-asr never downloads into
    a directory that exists. Whatever else onnx-asr refuses to load is reported
    the same way, as a `RuntimeError` the daemon turns into a clean exit.
    """
    import onnx_asr

    if not parakeet_installed(model_dir):
        if model_dir.exists():
            msg = (
                f"speech model {PARAKEET_MODEL!r} is incomplete in {model_dir}; "
                "delete that directory to download it again"
            )
            raise RuntimeError(msg)
        if not NETWORK_ALLOWED:
            msg = (
                f"speech model {PARAKEET_MODEL!r} is not installed in {model_dir}; "
                "set EASYSPEAK_OFFLINE=relaxed to download it"
            )
            raise RuntimeError(msg)
        logger.warning(
            "Speech model %r is not installed; downloading it from Hugging Face "
            "into %s. Set EASYSPEAK_OFFLINE=strict to keep EasySpeak offline.",
            PARAKEET_MODEL,
            model_dir,
        )
    try:
        return onnx_asr.load_model(
            PARAKEET_MODEL, model_dir, quantization=PARAKEET_QUANTIZATION
        )
    except (onnx_asr.utils.ModelLoadingError, ValueError) as exc:
        msg = f"speech model {PARAKEET_MODEL!r} in {model_dir} cannot be loaded: {exc}"
        raise RuntimeError(msg) from exc

core.log

core.log

Logging setup for EasySpeak's terminal output.

Output is message-only (no level or timestamp prefixes) so the CLI reads nicely for humans; the level only gates which lines appear. Only the easyspeak and plugins logger hierarchies are configured, so importing the package as a library leaves the root logger untouched.

resolve_level

resolve_level(*, verbose=False, quiet=False)

Pick the log level: CLI flags win, then EASYSPEAK_LOG_LEVEL, else INFO.

Source code in src/core/log.py

def resolve_level(*, verbose=False, quiet=False):
    """Pick the log level: CLI flags win, then EASYSPEAK_LOG_LEVEL, else INFO."""
    if verbose:
        return logging.DEBUG
    if quiet:
        return logging.WARNING
    name = os.environ.get("EASYSPEAK_LOG_LEVEL", "INFO").upper()
    level = logging.getLevelName(name)
    return level if isinstance(level, int) else logging.INFO

configure

configure(level)

Attach a message-only stderr handler to the package loggers.

The level is also passed on to the processes EasySpeak starts, such as the tray helper, through EASYSPEAK_LOG_LEVEL, so they log like the daemon, even when the level came from --verbose or --quiet.

Source code in src/core/log.py

def configure(level):
    """Attach a message-only stderr handler to the package loggers.

    The level is also passed on to the processes EasySpeak starts, such as the tray
    helper, through `EASYSPEAK_LOG_LEVEL`, so they log like the daemon, even when
    the level came from `--verbose` or `--quiet`.
    """
    os.environ["EASYSPEAK_LOG_LEVEL"] = logging.getLevelName(level)
    # Plain message-only output keeps the normal CLI clean; at DEBUG, prefix
    # each line with its level since that mode is for diagnosing.
    fmt = "%(levelname)s %(message)s" if level <= logging.DEBUG else "%(message)s"
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(fmt))
    for name in _PACKAGE_LOGGERS:
        log = logging.getLogger(name)
        log.handlers.clear()
        log.addHandler(handler)
        log.setLevel(level)
        log.propagate = False
    logger.debug(
        "logging configured: level=%s, stderr, format=%r, loggers=%s",
        logging.getLevelName(level),
        fmt,
        ", ".join(_PACKAGE_LOGGERS),
    )

core.speech

core.speech

Text-to-speech playback pipeline and audio-device noise suppression.

Extracted from core.main so the orchestration loop isn't tangled up with the subprocess plumbing that keeps a piper voice model warm. SpeechPipeline owns a persistent piper -> player pair; suppressed_c_stderr hides the unrelated ALSA/JACK probe spew PortAudio emits when the input side (PyAudio) starts up.

SpeechPipeline

SpeechPipeline()

A persistent piper -> audio player pipeline for low-latency speech.

Keeping piper alive avoids reloading its voice model (~2s) on every phrase, and streaming raw PCM straight to the player means playback starts as soon as synthesis does instead of after a temp WAV is fully written.

Create an idle pipeline; the subprocesses are spawned on first use.

Source code in src/core/speech.py

def __init__(self):
    """Create an idle pipeline; the subprocesses are spawned on first use."""
    self._piper = None
    self._player = None

ensure

ensure()

Spawn the persistent piper -> player pipeline if not already running.

Raises OSError if the pipeline can't be brought up (missing binary, bad model), so callers can warn once rather than silently failing per phrase.

Source code in src/core/speech.py

def ensure(self):
    """Spawn the persistent piper -> player pipeline if not already running.

    Raises OSError if the pipeline can't be brought up (missing binary, bad model),
    so callers can warn once rather than silently failing per phrase.
    """
    if self._alive(self._piper) and self._alive(self._player):
        return
    self._kill_speech()  # clear out any half-dead pipeline first
    rate = self._piper_sample_rate()
    try:
        self._player = subprocess.Popen(
            self._player_cmd(rate),
            stdin=subprocess.PIPE,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        self._piper = subprocess.Popen(
            [PIPER_BIN, "--model", PIPER_MODEL, "--output-raw"],
            stdin=subprocess.PIPE,
            stdout=self._player.stdin,
            stderr=subprocess.DEVNULL,
        )
        self._player.stdin.close()
        time.sleep(SPEECH_SPAWN_GRACE)
        if not (self._alive(self._piper) and self._alive(self._player)):
            msg = "speech pipeline exited immediately after start"
            raise OSError(msg)
    except Exception:
        self._kill_speech()  # don't leak a half-spawned pipeline
        raise

speak

speak(text)

Text-to-speech output.

Non-blocking: the phrase is handed to a warm piper process that streams audio to the player, so this returns immediately and the speech plays concurrently with whatever the caller does next.

Source code in src/core/speech.py

def speak(self, text):
    """Text-to-speech output.

    Non-blocking: the phrase is handed to a warm piper process that streams
    audio to the player, so this returns immediately and the speech plays
    concurrently with whatever the caller does next.
    """
    text = text.strip()
    if not text:
        return
    logger.info("💬 %s", text)
    for _ in range(2):
        try:
            self.ensure()
            self._piper.stdin.write((text + "\n").encode())
            self._piper.stdin.flush()
            return
        except (BrokenPipeError, OSError, ValueError):
            self._kill_speech()  # tear down the broken pipeline, then retry
    logger.debug("speech pipeline unavailable; dropped phrase: %s", text)

drain

drain()

Flush pending speech, wait for playback, then stop the pipeline.

Used on shutdown so a final phrase (e.g. "Goodbye.") is heard in full before the process exits.

Source code in src/core/speech.py

def drain(self):
    """Flush pending speech, wait for playback, then stop the pipeline.

    Used on shutdown so a final phrase (e.g. "Goodbye.") is heard in full before the
    process exits.
    """
    piper, player = self._piper, self._player
    self._piper = self._player = None
    if piper is not None:
        self._close(piper)
    if player is not None:
        self._close(player)

suppressed_c_stderr

suppressed_c_stderr()

Silence the device-probe spew PortAudio dumps when PyAudio starts up.

Initializing PyAudio makes PortAudio enumerate every ALSA PCM named in the system alsa.conf (surround*, hdmi, iec958, ...) and ping JACK. Devices the machine doesn't have each print a harmless error, and libasound/libjack write them straight to file descriptor 2 from C — Python's sys.stderr never sees them, so only redirecting the fd itself can hide them.

Suppression is best-effort: if stderr has no real fd (e.g. captured in tests), the fd can't be snapshotted (e.g. fd exhaustion), or the fd swap itself fails (e.g. fd 2 closed/invalid), there's nothing safe to redirect, so pass through rather than abort the caller.

Source code in src/core/speech.py

@contextlib.contextmanager
def suppressed_c_stderr():
    """Silence the device-probe spew PortAudio dumps when PyAudio starts up.

    Initializing PyAudio makes PortAudio enumerate every ALSA PCM named in the
    system alsa.conf (surround*, hdmi, iec958, ...) and ping JACK. Devices the
    machine doesn't have each print a harmless error, and libasound/libjack
    write them straight to file descriptor 2 from C — Python's sys.stderr never
    sees them, so only redirecting the fd itself can hide them.

    Suppression is best-effort: if stderr has no real fd (e.g. captured in
    tests), the fd can't be snapshotted (e.g. fd exhaustion), or the fd swap
    itself fails (e.g. fd 2 closed/invalid), there's nothing safe to redirect,
    so pass through rather than abort the caller.
    """
    try:
        stderr_fd = sys.stderr.fileno()
        # Flush first so already-buffered Python output lands on the real
        # stderr and not the /dev/null we're about to swap in.
        sys.stderr.flush()
        saved_fd = os.dup(stderr_fd)
    except (AttributeError, OSError, ValueError):
        yield
        return
    try:
        with open(os.devnull, "w") as devnull:  # noqa: PTH123
            try:
                os.dup2(devnull.fileno(), stderr_fd)
            except OSError:
                # Couldn't point fd 2 at /dev/null; run the body unsuppressed
                # rather than abort the caller, as the docstring promises.
                yield
                return
            try:
                yield
            finally:
                # Best-effort restore: a failed swap-back must not mask the
                # outcome of the body we just ran.
                with contextlib.suppress(OSError):
                    os.dup2(saved_fd, stderr_fd)
    finally:
        os.close(saved_fd)

core.i18n

core.i18n

Translations of the spoken replies, read straight from gettext .po files.

The core and every plugin are each a gettext domain with a locale/ directory next to their code, laid out the conventional way: locale/<language>/LC_MESSAGES/<domain>.po, the domain being the directory's name. The catalog for REPLY_LANGUAGE is built in memory from the .po at startup, so no compiled .mo files exist, and a missing catalog or a missing string falls back to the English source text.

A plugin opts in with one line, _ = translator(__file__), and wraps what it speaks: core.speak(_("Opening {app}.").format(app=app)). just translations keeps the .po files in step with the code.

_ module-attribute

_ = translator(__file__)

The core's own translator, for the replies spoken by core.main and core.tray.

translator

translator(file, language=None)

Return the gettext function for the package that file belongs to.

The package directory's name is the domain and its locale/ directory holds the catalogs; language defaults to config.REPLY_LANGUAGE.

Source code in src/core/i18n.py

def translator(file, language=None):
    """Return the `gettext` function for the package that `file` belongs to.

    The package directory's name is the domain and its `locale/` directory
    holds the catalogs; `language` defaults to `config.REPLY_LANGUAGE`.
    """
    package = Path(file).parent
    language = language or config.REPLY_LANGUAGE
    po = package / "locale" / language / "LC_MESSAGES" / f"{package.name}.po"
    if not po.is_file():
        return gettext.NullTranslations().gettext
    catalog = gettext.GNUTranslations(io.BytesIO(polib.pofile(str(po)).to_binary()))
    return catalog.gettext

core.wakeword

core.wakeword

Wake-word detection backed by pyopen-wakeword.

Wraps rhasspy's pyopen-wakeword (an alternative openWakeWord implementation that ships a bundled TensorFlow Lite and depends only on numpy). pyopen-wakeword splits detection into a feature extractor and a per-word model and streams probabilities; this thin wrapper just feeds one through the other and reports the strongest score for the chunk. It is glue for that two-stage streaming API, not a compatibility shim — there is no extra copy or conversion on the hot path.

WakeWordModel

WakeWordModel(wake_word=WAKE_WORD)

Single-wake-word detector: scores 16 kHz PCM chunks for one wake word.

Load the shared feature extractor and the model for one wake word.

Source code in src/core/wakeword.py

def __init__(self, wake_word: str = WAKE_WORD):
    """Load the shared feature extractor and the model for one wake word."""
    self._features = OpenWakeWordFeatures.from_builtin()
    self._model = OpenWakeWord.from_builtin(Model(wake_word))

predict

predict(pcm)

Return the wake-word probability for a chunk of 16-bit mono 16 kHz PCM.

pyopen-wakeword yields zero or more probabilities per chunk as its internal buffers fill; the strongest stands in for "latest score for this frame", so the caller's threshold check is unchanged.

Source code in src/core/wakeword.py

def predict(self, pcm: bytes) -> float:
    """Return the wake-word probability for a chunk of 16-bit mono 16 kHz PCM.

    pyopen-wakeword yields zero or more probabilities per chunk as its
    internal buffers fill; the strongest stands in for "latest score for
    this frame", so the caller's threshold check is unchanged.
    """
    score = 0.0
    for embeddings in self._features.process_streaming(pcm):
        for prob in self._model.process_streaming(embeddings):
            score = max(score, prob)
    return score

reset

reset()

Drop buffered audio and embeddings so stale state can't fire a wake.

Source code in src/core/wakeword.py

def reset(self) -> None:
    """Drop buffered audio and embeddings so stale state can't fire a wake."""
    self._features.reset()
    self._model.reset()

core.tray

core.tray

Panel-indicator bridge and asleep-state lifecycle for the daemon.

The daemon is voice-first and otherwise headless; this module gives it a top-panel microphone icon and owns everything about it so the audio loop in core.main stays about audio. On GNOME the bundled gnome@easyspeak.dev Shell extension serves the icon; on other desktops, such as KDE Plasma, a StatusNotifierItem helper process (core.statusnotifier) does. Either way the daemon runs without a D-Bus server of its own, via two one-directional channels:

• daemon -> icon: state is pushed for display via a one-shot gdbus call (the same path the mouse-grid plugin uses to drive the extension).
• icon -> daemon: the extension's menu writes a single command to a control file; the controller consumes it. The audio loop already wakes every ~80ms on a read, so a cheap file probe per iteration is enough — no GLib loop needed.

The indicator is shown only while the assistant is asleep (deactivated); while running it stays hidden because GNOME's own microphone privacy icon already signals the open mic.

TrayAction

Bases: Enum

              flowchart TD
              core.tray.TrayAction[TrayAction]

              

              click core.tray.TrayAction href "" "core.tray.TrayAction"
            

What Tray.poll is telling the audio loop to do next.

Tray

Tray(control_file=CONTROL_FILE, speak=None)

Owns the EasySpeak panel indicator and the asleep (deactivated) lifecycle.

The audio loop calls poll once per iteration; the controller pushes display state, consumes menu commands, and — when deactivated — runs the idle loop itself, using caller-supplied callbacks to release and reacquire the microphone. It never touches audio directly.

Best-effort throughout: without a tray (no gdbus, extension or panel to show the icon) the state pushes simply fail and the daemon carries on, mirroring how the mouse-grid plugin tolerates a missing extension.

Set up the controller, optionally with a spoken-feedback callback.

speak (core.speak) voices the asleep/awake lifecycle: the deactivation confirmation for a button mute, the reactivation and startup greetings, and the explanation when sleep can't engage. A voice "go to sleep" is already spoken by the sleep plugin, hence the announce flag on sleep. Defaults to a no-op so the tray works headless and in tests.

Source code in src/core/tray.py

def __init__(self, control_file=CONTROL_FILE, speak=None):
    """Set up the controller, optionally with a spoken-feedback callback.

    `speak` (core.speak) voices the asleep/awake lifecycle: the deactivation
    confirmation for a button mute, the reactivation and startup greetings,
    and the explanation when sleep can't engage. A voice "go to sleep" is
    already spoken by the sleep plugin, hence the `announce` flag on
    [`sleep`][core.tray.Tray.sleep]. Defaults to a no-op so the tray works
    headless and in tests.
    """
    self._control_file = Path(control_file)
    self._service = SHELL_SERVICE if on_gnome() else TRAY_SERVICE
    self._helper = None
    self._about = None
    self._sleep_requested = False
    self._speak = speak or (lambda _text: None)

started

started()

Daemon is up and listening; ensure the indicator is hidden.

Also drop any command left in the control file from before startup. It's a one-shot channel for live menu clicks, so a stale 'about'/'help'/ 'quit' written during a previous session (or before the daemon was running to consume it) must not fire the moment we begin polling — which otherwise pops the About window open on launch.

Greets the user once startup reaches a listening state, the spoken counterpart of the wake chime — and the same greeting reactivation gives.

Outside GNOME it starts the tray helper first.

Source code in src/core/tray.py

def started(self):
    """Daemon is up and listening; ensure the indicator is hidden.

    Also drop any command left in the control file from before startup. It's a
    one-shot channel for *live* menu clicks, so a stale 'about'/'help'/ 'quit'
    written during a previous session (or before the daemon was running to consume
    it) must not fire the moment we begin polling — which otherwise pops the About
    window open on launch.

    Greets the user once startup reaches a listening state, the spoken
    counterpart of the wake chime — and the same greeting reactivation gives.

    Outside GNOME it starts the tray helper first.
    """
    self.take_command()
    if self._service == TRAY_SERVICE:
        self._start_helper()
    self.set_state(STATE_LISTENING)
    self._speak(_("Welcome! I'm ready."))

stopped

stopped()

Daemon is exiting; hide the indicator so no stale icon is left.

Source code in src/core/tray.py

def stopped(self):
    """Daemon is exiting; hide the indicator so no stale icon is left."""
    self.set_state(STATE_LISTENING)
    if self._helper:
        self._helper.terminate()
        self._helper = None

request_sleep

request_sleep()

Queue a deactivate (e.g. the "go to sleep" voice command).

The mic is released at the audio loop's next poll, after the current command finishes.

Source code in src/core/tray.py

def request_sleep(self):
    """Queue a deactivate (e.g. the "go to sleep" voice command).

    The mic is released at the audio loop's next [`poll`][core.tray.Tray.poll],
    after the current command finishes.
    """
    self._sleep_requested = True

poll

poll(release_mic, acquire_mic)

Act on any pending menu command or queued sleep request.

release_mic / acquire_mic are zero-arg callbacks that close and reopen the input stream; the controller calls them around the asleep idle loop so this module stays out of the audio internals. Returns a TrayAction for the audio loop to dispatch on.

Source code in src/core/tray.py

def poll(self, release_mic, acquire_mic):
    """Act on any pending menu command or queued sleep request.

    `release_mic` / `acquire_mic` are zero-arg callbacks that close and reopen the
    input stream; the controller calls them around the asleep idle loop so this
    module stays out of the audio internals. Returns a
    [`TrayAction`][core.tray.TrayAction] for the audio loop to dispatch on.
    """
    command = self.take_command()
    if command == COMMAND_QUIT:
        return TrayAction.QUIT
    if self._run_menu_action(command):
        return TrayAction.CONTINUE
    if command == COMMAND_MUTE or self._sleep_requested:
        announce = command == COMMAND_MUTE
        self._sleep_requested = False
        return self.sleep(release_mic, acquire_mic, announce=announce)
    return TrayAction.CONTINUE

open_about

open_about()

Open the About dialog, from the menu or by voice.

The dialog needs PyGObject + GTK4/libadwaita, which aren't in the daemon's own (uv) venv. Reuse the PyGObject-equipped interpreter the dictation AT-SPI helper already runs in, falling back to our own interpreter when it isn't set (e.g. a plain pip install). GTK 4 and libadwaita are optional, so without them the user still hears who they're talking to, with the details in a notification.

Outside GNOME, KDE's kdialog shows the information first where it's installed, in the desktop's own look. An About dialog that is already showing is closed and opened again, which brings it to the front: Wayland desktops don't let one program raise another one's window.

Source code in src/core/tray.py

def open_about(self):
    """Open the About dialog, from the menu or by voice.

    The dialog needs PyGObject + GTK4/libadwaita, which aren't in the daemon's
    own (uv) venv. Reuse the PyGObject-equipped interpreter the dictation AT-SPI
    helper already runs in, falling back to our own interpreter when it isn't set
    (e.g. a plain pip install). GTK 4 and libadwaita are optional, so without them
    the user still hears who they're talking to, with the details in a notification.

    Outside GNOME, KDE's `kdialog` shows the information first where it's
    installed, in the desktop's own look. An About dialog that is already
    showing is closed and opened again, which brings it to the front: Wayland
    desktops don't let one program raise another one's window.
    """
    self.close_about()
    if not on_gnome() and shutil.which("kdialog"):
        kdialog = ["kdialog", "--title", "About EasySpeak"]
        kdialog += ["--icon", APPLICATION_ICON, "--msgbox", about_html()]
        self._about = self._spawn(kdialog, "About dialog (kdialog)")
        if self._about:
            return
    python = os.environ.get("EASYSPEAK_ATSPI_PYTHON", sys.executable)
    self._about = self._spawn([python, ABOUT_HELPER], "About dialog")
    if not self._about:
        self._tell_about()

about_open

about_open()

Whether an About dialog EasySpeak opened is still showing.

Source code in src/core/tray.py

def about_open(self):
    """Whether an About dialog EasySpeak opened is still showing."""
    return self._about is not None and self._about.poll() is None

close_about

close_about()

Close the About dialog EasySpeak opened; False if none is showing.

Source code in src/core/tray.py

def close_about(self):
    """Close the About dialog EasySpeak opened; False if none is showing."""
    if not self.about_open():
        return False
    self._about.terminate()
    self._about = None
    return True

sleep

sleep(release_mic, acquire_mic, announce=True)

Release the mic and idle until reactivated, then greet and resume.

Pushes the muted state first and refuses to sleep unless it lands: reactivation arrives via the tray (its menu or the Quick Settings toggle), so releasing the mic while the indicator is missing (no extension, or no panel to show the icon) would strand the daemon with no way back. While asleep the muted state is re-asserted every MUTED_REPUSH_INTERVAL seconds so the icon recovers if the panel or the extension reloads (it would otherwise come back hidden).

announce says whether to speak the deactivation confirmation: True for a button mute, False for a voice "go to sleep" (the sleep plugin already spoke, so speaking again would just repeat it). Reactivation always greets, matching the startup greeting.

Freeing the stream also clears GNOME's privacy microphone indicator, an OS-level 'not listening' cue alongside our own muted glyph.

Source code in src/core/tray.py

def sleep(self, release_mic, acquire_mic, announce=True):
    """Release the mic and idle until reactivated, then greet and resume.

    Pushes the muted state first and refuses to sleep unless it lands:
    reactivation arrives via the tray (its menu or the Quick Settings
    toggle), so releasing the mic while the indicator is missing (no
    extension, or no panel to show the icon) would strand the daemon with no
    way back. While asleep the muted state is re-asserted every
    `MUTED_REPUSH_INTERVAL` seconds so the icon recovers if the panel or the
    extension reloads (it would otherwise come back hidden).

    `announce` says whether to speak the deactivation confirmation: True for
    a button mute, False for a voice "go to sleep" (the sleep plugin already
    spoke, so speaking again would just repeat it). Reactivation always
    greets, matching the startup greeting.

    Freeing the stream also clears GNOME's privacy microphone indicator, an
    OS-level 'not listening' cue alongside our own muted glyph.
    """
    if not self.set_state(STATE_MUTED):
        logger.warning(
            "Tray indicator unavailable; staying awake so you keep a way to "
            "reactivate."
        )
        self._speak(_("I couldn't turn voice control off, so I'll keep listening."))
        return TrayAction.CONTINUE
    release_mic()
    logger.info("Muted; microphone released. Waiting for reactivation...")
    if announce:
        self._speak(_("Voice control turned off."))
    next_repush = time.monotonic() + MUTED_REPUSH_INTERVAL
    while True:
        command = self.take_command()
        if command == COMMAND_QUIT:
            return TrayAction.QUIT
        if command == COMMAND_UNMUTE:
            acquire_mic()
            self.set_state(STATE_LISTENING)
            self._speak(_("Welcome! I'm ready."))
            return TrayAction.RESUME
        self._run_menu_action(command)
        if time.monotonic() >= next_repush:
            self.set_state(STATE_MUTED)
            next_repush = time.monotonic() + MUTED_REPUSH_INTERVAL
        time.sleep(SLEEP_POLL_INTERVAL)

set_state

set_state(state)

Push the current display state to the panel indicator (best-effort).

Source code in src/core/tray.py

def set_state(self, state):
    """Push the current display state to the panel indicator (best-effort)."""
    cmd = [
        "gdbus",
        "call",
        "--session",
        "--dest",
        self._service,
        "--object-path",
        "/org/easyspeak/Desktop",
        "--method",
        "org.easyspeak.Desktop.SetState",
        str(state),
    ]
    try:
        result = subprocess.run(
            cmd, capture_output=True, text=True, timeout=GDBUS_TIMEOUT
        )
        return result.returncode == 0
    except (OSError, subprocess.TimeoutExpired):
        return False  # gdbus missing or a wedged bus

take_command

take_command()

Return and clear the latest menu command, or None if none is pending.

Reads the one-shot control file the extension writes, then deletes it so each menu click fires exactly once. A missing file is the common case (every idle iteration), so it's checked cheaply first. The delete is best-effort and kept separate from the read: a command that was read successfully is returned even if the unlink fails (e.g. the file vanished in a race), rather than being silently dropped.

Source code in src/core/tray.py

def take_command(self):
    """Return and clear the latest menu command, or None if none is pending.

    Reads the one-shot control file the extension writes, then deletes it so
    each menu click fires exactly once. A missing file is the common case
    (every idle iteration), so it's checked cheaply first. The delete is
    best-effort and kept separate from the read: a command that was read
    successfully is returned even if the unlink fails (e.g. the file vanished
    in a race), rather than being silently dropped.
    """
    try:
        if not self._control_file.exists():
            return None
        command = self._control_file.read_text().strip()
    except OSError:
        return None
    # already gone or a transient error; the command still stands
    with contextlib.suppress(OSError):
        self._control_file.unlink()
    return command or None

on_gnome

on_gnome()

Return whether the session is GNOME, where the Shell extension is the tray.

Source code in src/core/tray.py

def on_gnome():
    """Return whether the session is GNOME, where the Shell extension is the tray."""
    return "GNOME" in os.environ.get("XDG_CURRENT_DESKTOP", "").split(":")

notify

notify(summary, body)

Show a desktop notification; False if no notification service took it.

Speaks the freedesktop Notifications protocol over D-Bus, which every desktop's notification service implements. jeepney is imported lazily so the dependency isn't needed to load this module.

Source code in src/core/tray.py

def notify(summary, body):
    """Show a desktop notification; False if no notification service took it.

    Speaks the freedesktop Notifications protocol over D-Bus, which every desktop's
    notification service implements. jeepney is imported lazily so the dependency
    isn't needed to load this module.
    """
    from jeepney import DBusAddress, MessageType, new_method_call
    from jeepney.io.blocking import open_dbus_connection

    service = "org.freedesktop.Notifications"
    address = DBusAddress(
        "/org/freedesktop/Notifications", bus_name=service, interface=service
    )
    message = new_method_call(
        address,
        "Notify",
        "susssasa{sv}i",
        (APPLICATION_NAME, 0, APPLICATION_ICON, summary, body, [], {}, -1),
    )
    try:
        with open_dbus_connection(bus="SESSION") as conn:
            reply = conn.send_and_get_reply(message, timeout=GDBUS_TIMEOUT)
    except Exception as exc:
        logger.debug("No desktop notification: %s", exc)
        return False
    return reply.header.message_type is MessageType.method_return

core.statusnotifier

core.statusnotifier

Tray icon for desktops other than GNOME, such as KDE Plasma.

Implements the freedesktop StatusNotifierItem specification and its D-Bus menu, so any panel with a StatusNotifierItem host shows it: KDE Plasma, Xfce, Cinnamon, Budgie, waybar, and GNOME with the AppIndicator extension. On GNOME itself the bundled Shell extension draws the icon instead.

Runs as its own process, started by Tray with the control file as its argument, because serving D-Bus means waiting on the bus, which the audio loop can't do. It talks to the daemon over the same two channels as the GNOME extension: the daemon pushes its state through SetState on /org/easyspeak/Desktop, and the menu writes its commands into the control file. The process exits when the daemon does.

The icon sits in the panel's status area while EasySpeak is asleep ("Active") and retreats into the hidden icons while it listens ("Passive"), mirroring the GNOME indicator, which shows only while asleep. Clicking it toggles between the two.

StatusNotifier

StatusNotifier(connection, control_file)

The tray item, its menu and the daemon's SetState on one D-Bus connection.

Serve on connection, writing menu commands to control_file.

Source code in src/core/statusnotifier.py

def __init__(self, connection, control_file):
    """Serve on `connection`, writing menu commands to `control_file`."""
    self._bus = connection
    self._control_file = control_file
    self._state = None
    self._revision = 1
    self._registered = False
    self._backlog = deque()
    self._handlers = {
        "SetState": self._on_set_state,
        "Activate": self._on_activate,
        "ContextMenu": self._on_context_menu,
        "SecondaryActivate": self._on_context_menu,
        "Scroll": self._on_context_menu,
        "Get": self._on_get,
        "GetAll": self._on_get_all,
        "Introspect": self._on_introspect,
        "GetLayout": self._on_get_layout,
        "GetGroupProperties": self._on_get_group_properties,
        "GetProperty": self._on_get_property,
        "Event": self._on_event,
        "EventGroup": self._on_event_group,
        "AboutToShow": self._on_about_to_show,
        "AboutToShowGroup": self._on_about_to_show_group,
    }
    self._watcher = DBusAddress(WATCHER_PATH, WATCHER_NAME, WATCHER_NAME)
    self._properties = DBusAddress(
        WATCHER_PATH, WATCHER_NAME, "org.freedesktop.DBus.Properties"
    )

start

start()

Claim the bus name and register with the panel, if there is one.

Returns False when another tray process already holds the name.

Source code in src/core/statusnotifier.py

def start(self):
    """Claim the bus name and register with the panel, if there is one.

    Returns False when another tray process already holds the name.
    """
    reply = self._call(message_bus.RequestName(BUS_NAME, 4))  # do not queue
    if reply is None or reply.body[0] != 1:  # not the primary owner
        return False
    rule = MatchRule(
        type="signal",
        interface="org.freedesktop.DBus",
        member="NameOwnerChanged",
    )
    rule.add_arg_condition(0, WATCHER_NAME)
    self._call(message_bus.AddMatch(rule))
    self._register()
    return True

next_message

next_message(timeout)

Return the next incoming message, or None when timeout passes first.

Source code in src/core/statusnotifier.py

def next_message(self, timeout):
    """Return the next incoming message, or None when `timeout` passes first."""
    if self._backlog:
        return self._backlog.popleft()
    try:
        return self._bus.receive(timeout=timeout)
    except TimeoutError:
        return None

handle

handle(msg)

Answer a method call, or react to the panel (re)appearing.

Source code in src/core/statusnotifier.py

def handle(self, msg):
    """Answer a method call, or react to the panel (re)appearing."""
    fields = msg.header.fields
    if msg.header.message_type == MessageType.signal:
        if fields.get(HeaderFields.member) == "NameOwnerChanged" and msg.body[2]:
            self._register()
        return
    if msg.header.message_type != MessageType.method_call:
        return
    method = fields.get(HeaderFields.member)
    handler = self._handlers.get(method)
    if handler is None:
        self._bus.send(
            new_error(
                msg, "org.freedesktop.DBus.Error.UnknownMethod", "s", (method,)
            )
        )
        return
    handler(msg, *msg.body)

is_muted

is_muted(state)

Return whether the daemon state means EasySpeak is asleep.

Source code in src/core/statusnotifier.py

def is_muted(state):
    """Return whether the daemon state means EasySpeak is asleep."""
    return state == STATE_MUTED

status

status(state)

Return the item's StatusNotifierItem status for a daemon state.

Source code in src/core/statusnotifier.py

def status(state):
    """Return the item's StatusNotifierItem status for a daemon state."""
    return "Active" if is_muted(state) else "Passive"

icon

icon(state)

Return the icon name for a daemon state.

Source code in src/core/statusnotifier.py

def icon(state):
    """Return the icon name for a daemon state."""
    return ICON_MUTED if is_muted(state) else ICON_LISTENING

tooltip

tooltip(state)

Return the tooltip's subtitle for a daemon state.

Source code in src/core/statusnotifier.py

def tooltip(state):
    """Return the tooltip's subtitle for a daemon state."""
    return "Asleep — click to reactivate" if is_muted(state) else "Listening"

toggle_command

toggle_command(state)

Return the command that flips between asleep and listening.

Source code in src/core/statusnotifier.py

def toggle_command(state):
    """Return the command that flips between asleep and listening."""
    return "unmute" if is_muted(state) else "mute"

menu_items

menu_items(state, starts_at_login=False)

Return the menu entries as (id, properties) pairs, in display order.

"Start at Login" is a checkmark entry, ticked when starts_at_login.

Source code in src/core/statusnotifier.py

def menu_items(state, starts_at_login=False):
    """Return the menu entries as `(id, properties)` pairs, in display order.

    "Start at Login" is a checkmark entry, ticked when `starts_at_login`.
    """
    toggle = "Reactivate EasySpeak" if is_muted(state) else "Deactivate EasySpeak"
    return [
        (MENU_TOGGLE, {"label": toggle}),
        (2, {"type": "separator"}),
        (
            MENU_AUTOSTART,
            {
                "label": "Start at Login",
                "toggle-type": "checkmark",
                "toggle-state": int(starts_at_login),
            },
        ),
        (MENU_HELP, {"label": "Help"}),
        (MENU_ABOUT, {"label": "About EasySpeak"}),
        (5, {"type": "separator"}),
        (MENU_QUIT, {"label": "Quit EasySpeak"}),
    ]

menu_command

menu_command(item_id, state)

Return the command a click on a menu entry sends, or None.

Source code in src/core/statusnotifier.py

def menu_command(item_id, state):
    """Return the command a click on a menu entry sends, or None."""
    return {
        MENU_TOGGLE: toggle_command(state),
        MENU_HELP: "help",
        MENU_ABOUT: "about",
        MENU_QUIT: "quit",
    }.get(item_id)

write_command

write_command(control_file, command)

Hand a command to the daemon through the one-shot control file.

Source code in src/core/statusnotifier.py

def write_command(control_file, command):
    """Hand a command to the daemon through the one-shot control file."""
    control_file = Path(control_file)
    control_file.parent.mkdir(parents=True, exist_ok=True)
    control_file.write_text(command)

main

main(argv=None)

Serve the tray icon until the daemon that started it goes away.

Logs at the daemon's level, which it passes on through EASYSPEAK_LOG_LEVEL.

Source code in src/core/statusnotifier.py

def main(argv=None):  # pragma: no cover - needs a session bus; run live
    """Serve the tray icon until the daemon that started it goes away.

    Logs at the daemon's level, which it passes on through `EASYSPEAK_LOG_LEVEL`.
    """
    log.configure(log.resolve_level())
    argv = sys.argv[1:] if argv is None else argv
    daemon = os.getppid()
    with open_dbus_connection("SESSION") as connection:
        tray = StatusNotifier(connection, Path(argv[0]))
        if not tray.start():
            return f"EasySpeak: {BUS_NAME} is taken by another tray."
        with contextlib.suppress(ConnectionError):  # the session ended
            while os.getppid() == daemon:
                msg = tray.next_message(PARENT_CHECK_INTERVAL)
                if msg is not None:
                    tray.handle(msg)
    return None

core.about

core.about

Standalone libadwaita "About EasySpeak" dialog.

The daemon is headless and the panel indicator lives in a GNOME Shell extension (Clutter/St), which can't host GTK widgets; so the About dialog runs as its own short-lived process that the tray launches via python -m easyspeak.core.about on the menu's "About" command. It uses libadwaita's AboutDialog (the modern, GNOME-recommended widget), falling back to the older AboutWindow on pre-1.5 libadwaita.

Everything GUI is imported lazily inside main so this module stays importable (for its metadata constants and version lookup) without PyGObject or a display present — which is how the unit suite exercises it.

app_version

app_version()

Return the installed package version, or "" if it can't be determined.

Read at runtime (rather than hard-coded) so the dialog always shows the version actually installed; an editable/source checkout without dist metadata simply shows no version rather than failing.

Source code in src/core/about.py

def app_version():
    """Return the installed package version, or "" if it can't be determined.

    Read at runtime (rather than hard-coded) so the dialog always shows the version
    actually installed; an editable/source checkout without dist metadata simply shows
    no version rather than failing.
    """
    try:
        return importlib.metadata.version("easyspeak-linux")
    except importlib.metadata.PackageNotFoundError:
        return ""

about_html

about_html()

Return the About information as rich text, for desktops without libadwaita.

kdialog shows it on KDE Plasma. It carries what the dialog's legal section does, as some models' licenses require crediting them.

Source code in src/core/about.py

def about_html():
    """Return the About information as rich text, for desktops without libadwaita.

    `kdialog` shows it on KDE Plasma. It carries what the dialog's legal section
    does, as some models' licenses require crediting them.
    """
    name = html.escape(f"{APPLICATION_NAME} {app_version()}".strip())
    models = "<br>".join(
        html.escape(f"{title}, {holder}, {license_id}")
        for title, holder, license_id in LEGAL
    )
    return (
        f"<h3>{name}</h3>"
        f"<p>{html.escape(COMMENTS)}</p>"
        f'<p><a href="{DOCS_URL}">{html.escape(DOCS_URL)}</a></p>'
        f"<p>{html.escape(COPYRIGHT)}<br>GNU General Public License v3.0</p>"
        f"<p><b>Speech models and voices</b><br>{models}</p>"
    )

main

main()

Show the About dialog and run until the user closes it.

Source code in src/core/about.py

def main():  # pragma: no cover - needs libadwaita and a display; run live
    """Show the About dialog and run until the user closes it."""
    import gi

    gi.require_version("Gtk", "4.0")
    gi.require_version("Adw", "1")
    from gi.repository import Adw, Gio

    app = Adw.Application(application_id=APP_ID, flags=Gio.ApplicationFlags.FLAGS_NONE)
    app.connect("activate", _present)
    return app.run(None)

core.hotkey

core.hotkey

Hold-to-dictate keyboard activation via evdev (the silent-activation path).

On Wayland an ordinary process can't grab global keys, and only raw evdev events from /dev/input expose key release — which "hold the keys to dictate, release to stop" needs. This listener reads every keyboard device in a background thread and tracks whether a configured modifier combo (default Ctrl+Shift) is fully held, so the audio loop can start dictation on press and end it the moment the keys come up.

It is best-effort: if python-evdev isn't installed or /dev/input isn't readable (the user isn't in the input group), the listener logs how to enable it and stays inert, so the daemon runs normally without the hotkey.

HotkeyListener

HotkeyListener(combo='ctrl+shift', enabled=True)

Track whether a key combo is held, off a background evdev reader.

The audio loop calls take_activation once per iteration to learn when the combo was just pressed, then drives a dictation session gated on is_held so releasing the keys ends it. All evdev/threading state is kept behind a lock; the event-processing logic in _process is pure and the I/O lives in _grab_devices/_drain, so the behaviour is unit-testable without a real keyboard.

Set up the listener for combo (e.g. "ctrl+shift").

An empty, unparseable, or unknown-key combo, or enabled=False, leaves the listener disabled so start is a no-op.

Source code in src/core/hotkey.py

def __init__(self, combo="ctrl+shift", enabled=True):
    """Set up the listener for `combo` (e.g. `"ctrl+shift"`).

    An empty, unparseable, or unknown-key combo, or `enabled=False`, leaves the
    listener disabled so [`start`][core.hotkey.HotkeyListener.start] is a no-op.
    """
    self._spec = combo
    self._groups = parse_combo(combo)
    unknown = unknown_keys(combo)
    if unknown:
        logger.warning(
            "Ignoring hotkey combo %r: unknown key(s) %s.",
            combo,
            ", ".join(unknown),
        )
    self._enabled = enabled and bool(self._groups) and not unknown
    self._pressed = set()  # combo keys currently held
    self._held = False  # whole combo satisfied
    self._activation = False  # rising edge, consumed by take_activation()
    self._lock = threading.Lock()
    self._stop = threading.Event()
    self._thread = None
    self._devices = []

is_held

is_held()

Return whether the full combo is currently held.

Source code in src/core/hotkey.py

def is_held(self):
    """Return whether the full combo is currently held."""
    with self._lock:
        return self._held

take_activation

take_activation()

Return True once per press of the combo, clearing the edge.

The audio loop polls this each iteration; a True means "the user just pressed the combo — start dictation now".

Source code in src/core/hotkey.py

def take_activation(self):
    """Return True once per press of the combo, clearing the edge.

    The audio loop polls this each iteration; a True means "the user just pressed
    the combo — start dictation now".
    """
    with self._lock:
        fired = self._activation
        self._activation = False
        return fired

start

start()

Begin watching the keyboard, unless disabled or no device is readable.

Logs and stays inert (no thread) when the feature is turned off, evdev is missing, or /dev/input can't be read, so the daemon runs normally either way.

Source code in src/core/hotkey.py

def start(self):
    """Begin watching the keyboard, unless disabled or no device is readable.

    Logs and stays inert (no thread) when the feature is turned off, evdev is
    missing, or `/dev/input` can't be read, so the daemon runs normally either way.
    """
    if not self._enabled:
        logger.debug("Keyboard activation disabled.")
        return
    self._devices = self._grab_devices()
    if not self._devices:
        return
    logger.info("Hold %s to dictate (silent activation).", self._spec)
    self._thread = threading.Thread(
        target=self._run, name="easyspeak-hotkey", daemon=True
    )
    self._thread.start()

stop

stop()

Stop the reader thread, then release the keyboard devices.

Joins the thread first so it leaves its select() loop before the fds close under it — closing a fd mid-select can raise in the thread. The loop wakes within one POLL_TIMEOUT, so the join returns promptly.

Source code in src/core/hotkey.py

def stop(self):
    """Stop the reader thread, then release the keyboard devices.

    Joins the thread first so it leaves its `select()` loop before the fds close
    under it — closing a fd mid-select can raise in the thread. The loop wakes
    within one POLL_TIMEOUT, so the join returns promptly.
    """
    self._stop.set()
    thread, self._thread = self._thread, None
    if thread is not None:
        thread.join(timeout=POLL_TIMEOUT + 1.0)
    for device in self._devices:
        with contextlib.suppress(Exception):
            device.close()
    self._devices = []

parse_combo

parse_combo(spec)

Parse a '+'-separated combo string into groups of accepted key names.

Each element becomes a group of evdev key names of which any one satisfies that part of the combo, so "ctrl+shift" matches either Ctrl together with either Shift. Names that aren't known aliases fall through as a single literal evdev key name ("rightctrl" -> "KEY_RIGHTCTRL") so less common combos still work.

Parameters:

Name	Type	Description	Default
spec	str	A combo such as "ctrl+shift" or "super+space".	required

Returns:

Type	Description
frozenset[str]	A tuple of frozensets, one per combo element; the combo is held when
...	every group has at least one of its keys down.

Source code in src/core/hotkey.py

def parse_combo(spec: str) -> tuple[frozenset[str], ...]:
    """Parse a `'+'`-separated combo string into groups of accepted key names.

    Each element becomes a group of evdev key names of which any one satisfies
    that part of the combo, so `"ctrl+shift"` matches either Ctrl together
    with either Shift. Names that aren't known aliases fall through as a single
    literal evdev key name (`"rightctrl"` -> `"KEY_RIGHTCTRL"`) so less
    common combos still work.

    Args:
        spec: A combo such as `"ctrl+shift"` or `"super+space"`.

    Returns:
        A tuple of frozensets, one per combo element; the combo is held when
        every group has at least one of its keys down.
    """
    groups = []
    for part in spec.split("+"):
        name = part.strip().lower()
        if not name:
            continue
        if name in _MODIFIER_ALIASES:
            groups.append(frozenset(_MODIFIER_ALIASES[name]))
        else:
            key = name.upper()
            if not key.startswith("KEY_"):
                key = "KEY_" + key
            groups.append(frozenset((key,)))
    return tuple(groups)

unknown_keys

unknown_keys(spec)

Return combo tokens that are neither a modifier alias nor a real evdev key.

Validates against evdev.ecodes, so a typo (ctrl+spcae) is caught instead of silently becoming a dead key. Returns [] when python-evdev can't be imported, since the listener is inert without it anyway. Only evdev key names are checked; the portal-based activation in issue #92 would use a different trigger grammar.

Source code in src/core/hotkey.py

def unknown_keys(spec: str) -> list[str]:
    """Return combo tokens that are neither a modifier alias nor a real evdev key.

    Validates against `evdev.ecodes`, so a typo (`ctrl+spcae`) is caught instead of
    silently becoming a dead key. Returns `[]` when python-evdev can't be imported,
    since the listener is inert without it anyway. Only evdev key names are checked;
    the portal-based activation in issue #92 would use a different trigger grammar.
    """
    try:
        from evdev import ecodes
    except ImportError:
        return []
    unknown = []
    for part in spec.split("+"):
        name = part.strip().lower()
        if not name or name in _MODIFIER_ALIASES:
            continue
        if f"KEY_{name.upper()}" not in ecodes.ecodes:
            unknown.append(part.strip())
    return unknown

core.mediakeys

core.mediakeys

Replay multimedia keys through GNOME (Mutter) for native desktop feedback.

This lets the desktop render its own volume OSD and chime rather than imitating them. Pressing a volume key does not run any command: gnome-settings-daemon grabs the raw evdev key and handles the volume change, OSD and chime itself. The only way to reproduce that exactly is to replay the key. We inject it through Mutter's RemoteDesktop interface, which needs no special privileges. A RemoteDesktop session lives only as long as the D-Bus connection that created it, so the whole CreateSession -> Start -> NotifyKeyboardKeycode -> Stop sequence runs on one connection.

RemoteDesktopUnavailableError

Bases: RuntimeError

              flowchart TD
              core.mediakeys.RemoteDesktopUnavailableError[RemoteDesktopUnavailableError]

              

              click core.mediakeys.RemoteDesktopUnavailableError href "" "core.mediakeys.RemoteDesktopUnavailableError"
            

GNOME's RemoteDesktop interface refused or is missing, e.g. on KDE Plasma.

tap_key

tap_key(keycode)

Press and release one evdev keycode via Mutter RemoteDesktop.

Raises if RemoteDesktop is unavailable (e.g. a non-GNOME session) so the caller can fall back to a silent change.

Source code in src/core/mediakeys.py

def tap_key(keycode):
    """Press and release one evdev keycode via Mutter RemoteDesktop.

    Raises if RemoteDesktop is unavailable (e.g. a non-GNOME session) so the caller can
    fall back to a silent change.
    """
    tap_chord([keycode])

tap_chord

tap_chord(keycodes)

Hold a chord of evdev keycodes together, then release it.

Keys go down in the order given and come up in reverse, so [CTRL, V] is the Ctrl+V an application expects rather than two unrelated keypresses. Used for pasting dictated text, which is how text reaches an application: every toolkit implements paste, whereas accessibility-level insertion is widely stubbed out (Chromium-based apps accept it and discard it).

A desktop without RemoteDesktop makes this raise RemoteDesktopUnavailableError, so the caller can report why.

Source code in src/core/mediakeys.py

def tap_chord(keycodes):
    """Hold a chord of evdev keycodes together, then release it.

    Keys go down in the order given and come up in reverse, so `[CTRL, V]` is the
    Ctrl+V an application expects rather than two unrelated keypresses. Used for
    pasting dictated text, which is how text reaches an application: every toolkit
    implements paste, whereas accessibility-level insertion is widely stubbed out
    (Chromium-based apps accept it and discard it).

    A desktop without RemoteDesktop makes this raise
    [`RemoteDesktopUnavailableError`][core.mediakeys.RemoteDesktopUnavailableError],
    so the caller can report why.
    """
    conn = open_dbus_connection(bus="SESSION")
    try:
        reply = _call(conn, new_method_call(_REMOTE_DESKTOP, "CreateSession"))
        session = DBusAddress(
            reply.body[0],
            bus_name=_BUS,
            interface=f"{_BUS}.Session",
        )
        _call(conn, new_method_call(session, "Start"))
        # Synchronous replies guarantee Mutter has processed each event before
        # we move on, so no settling delay is needed before Stop.
        for keycode in keycodes:
            _call(
                conn,
                new_method_call(
                    session, "NotifyKeyboardKeycode", "ub", (keycode, True)
                ),
            )
        for keycode in reversed(keycodes):
            _call(
                conn,
                new_method_call(
                    session, "NotifyKeyboardKeycode", "ub", (keycode, False)
                ),
            )
        _call(conn, new_method_call(session, "Stop"))
    finally:
        conn.close()

parse_key_request

parse_key_request(words, bare_allowed, *, names=None, counts=None, prefixes=None)

Return (keycode, repeats) for a spoken keystroke command, else None.

Accepts an optional "press" prefix, key names of up to three words, and a trailing count as digits or a number word. bare_allowed names the keys a caller will accept without the prefix; every other key needs it, so that words already meaning something else in that mode are left alone. The vocabulary is English unless names (spoken name -> key in KEYS), counts (number word -> int) and prefixes say otherwise, as dictation does for the user's language.

Source code in src/core/mediakeys.py

def parse_key_request(words, bare_allowed, *, names=None, counts=None, prefixes=None):
    """Return (keycode, repeats) for a spoken keystroke command, else None.

    Accepts an optional "press" prefix, key names of up to three words, and a trailing
    count as digits or a number word. `bare_allowed` names the keys a caller will
    accept without the prefix; every other key needs it, so that words already
    meaning something else in that mode are left alone. The vocabulary is English
    unless `names` (spoken name -> key in `KEYS`), `counts` (number word -> int)
    and `prefixes` say otherwise, as dictation does for the user's language.
    """
    names = KEY_NAMES if names is None else names
    counts = SPOKEN_COUNTS if counts is None else counts
    prefixes = PRESS_PREFIXES if prefixes is None else prefixes
    explicit = bool(words) and words[0] in prefixes
    if explicit:
        words = words[1:]
    if not words:
        return None

    for length in (3, 2, 1):
        name = names.get(" ".join(words[:length]))
        if name is None or (not explicit and name not in bare_allowed):
            continue
        tail = words[length:]
        if not tail:
            return KEYS[name], 1
        if len(tail) > 1:
            return None
        count = int(tail[0]) if tail[0].isdigit() else counts.get(tail[0])
        if count is None:
            return None
        return KEYS[name], min(count, MAX_KEY_REPEATS)
    return None

press_key

press_key(keycode, repeats=1)

Tap keycode repeats times; True when the keystrokes were delivered.

Source code in src/core/mediakeys.py

def press_key(keycode, repeats=1):
    """Tap `keycode` `repeats` times; True when the keystrokes were delivered."""
    try:
        for _ in range(repeats):
            tap_chord([keycode])
    except (RuntimeError, OSError) as exc:
        logger.warning("Keystrokes need GNOME's RemoteDesktop interface: %s", exc)
        return False
    return True

core.gnome_extension

core.gnome_extension

Install, refresh, and enable the bundled GNOME Shell extension.

Core owns the extension's lifecycle because both the mousegrid plugin and core.tray drive it over D-Bus; ensure_extension runs once at startup. The extension ships as package data — extension.js, the extension-helpers.js it imports, and metadata.json — copied into the user's extensions dir as a set. On Wayland GNOME only loads extension code at login, so the startup copy is always one login behind; a oneshot systemd user unit ordered before the shell re-copies it at the start of every login to close that gap. The unit runs as the user, writes only to $HOME, and needs no privileges.

RefreshResult

Bases: Enum

              flowchart TD
              core.gnome_extension.RefreshResult[RefreshResult]

              

              click core.gnome_extension.RefreshResult href "" "core.gnome_extension.RefreshResult"
            

Outcome of a single refresh attempt.

See refresh_extension_files.

extension_source_dir

extension_source_dir()

Return the directory holding the bundled extension assets.

The extension's sources live in the src/gnome@easyspeak.dev/ folder, shipped in the wheel as the easyspeak.gnome package. Resolving by package name works in both editable and wheel installs, where that folder sits in different places relative to this module.

Source code in src/core/gnome_extension.py

def extension_source_dir():
    """Return the directory holding the bundled extension assets.

    The extension's sources live in the `src/gnome@easyspeak.dev/` folder,
    shipped in the wheel as the `easyspeak.gnome` package. Resolving by package
    name works in both editable and wheel installs, where that folder sits in
    different places relative to this module.
    """
    return Path(resources.files("easyspeak.gnome"))

extensions_root

extensions_root()

User-local directory GNOME Shell reads installed extensions from.

Source code in src/core/gnome_extension.py

def extensions_root():
    """User-local directory GNOME Shell reads installed extensions from."""
    return Path.home() / ".local" / "share" / "gnome-shell" / "extensions"

extension_dest_dir

extension_dest_dir()

User-local install location for the bundled extension.

Source code in src/core/gnome_extension.py

def extension_dest_dir():
    """User-local install location for the bundled extension."""
    return extensions_root() / EXTENSION_UUID

refresh_extension_files

refresh_extension_files(src_dir, dest_dir)

Copy the bundled assets into dest_dir when they differ from it.

Best-effort: returns REFRESHED if written, UNCHANGED if already current, or ERROR (noted on stderr) when a source is missing or a copy fails. Assets are staged then moved into place (extension.js last), so a failure leaves any working install untouched and never installs extension.js without the helper it imports.

Source code in src/core/gnome_extension.py

def refresh_extension_files(src_dir, dest_dir):
    """Copy the bundled assets into `dest_dir` when they differ from it.

    Best-effort: returns `REFRESHED` if written, `UNCHANGED` if already
    current, or `ERROR` (noted on stderr) when a source is missing or a copy
    fails. Assets are staged then moved into place (extension.js last), so a
    failure leaves any working install untouched and never installs extension.js
    without the helper it imports.
    """
    srcs = {name: src_dir / name for name in EXTENSION_ASSETS}
    if not all(s.is_file() for s in srcs.values()):
        logger.warning("bundled GNOME extension sources missing at %s", src_dir)
        return RefreshResult.ERROR
    # Clear temps an interrupted run may have left, so the dir self-heals even on
    # the UNCHANGED path below.
    _discard_staged(dest_dir)
    try:
        unchanged = all(
            (dest_dir / name).is_file()
            and filecmp.cmp(src, dest_dir / name, shallow=False)
            for name, src in srcs.items()
        )
        if unchanged:
            return RefreshResult.UNCHANGED
        dest_dir.mkdir(parents=True, exist_ok=True)
        staged = []
        for name, src in srcs.items():
            tmp = _staged_tmp(dest_dir, name)
            tmp.parent.mkdir(parents=True, exist_ok=True)  # subdir assets (schemas/)
            shutil.copy2(src, tmp)
            staged.append((tmp, dest_dir / name))
        for tmp, dest in reversed(staged):  # extension.js leads EXTENSION_ASSETS
            tmp.replace(dest)
        return RefreshResult.REFRESHED
    except OSError as e:
        _discard_staged(dest_dir)
        logger.warning("could not write GNOME extension to %s (%s)", dest_dir, e)
        return RefreshResult.ERROR

refresh_installed_extension

refresh_installed_extension()

Refresh the installed extension from the bundled copy.

Run by the systemd user unit at each login.

Source code in src/core/gnome_extension.py

def refresh_installed_extension():
    """Refresh the installed extension from the bundled copy.

    Run by the systemd user unit at each login.
    """
    return refresh_extension_files(extension_source_dir(), extension_dest_dir())

unit_path

unit_path()

User-local path of the extension-refresh unit.

Source code in src/core/gnome_extension.py

def unit_path():
    """User-local path of the extension-refresh unit."""
    return Path.home() / ".config" / "systemd" / "user" / REFRESH_UNIT_NAME

unit_text

unit_text()

Render the systemd user unit, baking in the interpreter and module path.

The unit body lives in a template shipped as package data (easyspeak.data) so it reads as a service file, not a Python string. A missing template is a broken install, so let the read error surface rather than masking it.

Two non-obvious choices in the rendered unit: it's ordered before gnome-session-pre.target and the org.gnome.Shell@* services (missing ones are ignored) so a changed extension takes effect on this login, not the next — on Wayland the shell only loads extensions at login. And both ExecStart arguments are quoted, since systemd splits ExecStart on whitespace.

Source code in src/core/gnome_extension.py

def unit_text():
    """Render the systemd user unit, baking in the interpreter and module path.

    The unit body lives in a template shipped as package data (`easyspeak.data`)
    so it reads as a service file, not a Python string. A missing template is a
    broken install, so let the read error surface rather than masking it.

    Two non-obvious choices in the rendered unit: it's ordered before
    `gnome-session-pre.target` and the `org.gnome.Shell@*` services (missing ones
    are ignored) so a changed extension takes effect on this login, not the next
    — on Wayland the shell only loads extensions at login. And both ExecStart
    arguments are quoted, since systemd splits ExecStart on whitespace.
    """
    return data_text(
        REFRESH_UNIT_TEMPLATE,
        target=PRE_SHELL_TARGET,
        python=_stable_interpreter(),
        script=Path(__file__).resolve(),
    )

install_refresh_unit

install_refresh_unit()

Install and enable the pre-shell refresh unit, idempotently.

Returns a short status string, or None when nothing changed or it isn't applicable (no systemd, a packaged system unit already present, write/enable failure).

Source code in src/core/gnome_extension.py

def install_refresh_unit():
    """Install and enable the pre-shell refresh unit, idempotently.

    Returns a short status string, or None when nothing changed or it isn't applicable
    (no systemd, a packaged system unit already present, write/enable failure).
    """
    if shutil.which("systemctl") is None:
        return None

    path = unit_path()

    if _system_unit_exists():
        if path.is_file():
            _run_systemctl("disable", REFRESH_UNIT_NAME)
            with contextlib.suppress(OSError):
                path.unlink()
        return None

    desired = unit_text()
    try:
        current = path.read_text() if path.is_file() else None
    except OSError:
        current = None

    changed = current != desired
    if changed:
        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(desired)
        except OSError:
            return None
        _run_systemctl("daemon-reload")
    elif _is_enabled():
        return None

    enabled = _run_systemctl("enable", REFRESH_UNIT_NAME)
    if enabled is None or enabled.returncode != 0:
        return None
    return (
        f"{'installed' if changed else 'enabled'} systemd user unit {REFRESH_UNIT_NAME}"
    )

migrate_legacy_extensions

migrate_legacy_extensions()

Remove any pre-rename extension install so it can't double-load.

The extension's UUID changed over time — easyspeak-grid@local (when it was just a mouse grid), then easyspeak@local, now gnome@easyspeak.dev. A leftover copy under an old UUID would stay enabled and add a second panel indicator and grid that the daemon no longer drives, so clean each up: disable it (so GNOME drops it from the enabled set) then delete its directory. Best-effort throughout — returns True if any legacy install was found and removed.

One-off migration shim: once users have had a release or two to upgrade past the old UUIDs this can be removed (call site and tests included).

Source code in src/core/gnome_extension.py

def migrate_legacy_extensions():
    """Remove any pre-rename extension install so it can't double-load.

    The extension's UUID changed over time — `easyspeak-grid@local` (when it was
    just a mouse grid), then `easyspeak@local`, now `gnome@easyspeak.dev`. A
    leftover copy under an old UUID would stay enabled and add a second panel
    indicator and grid that the daemon no longer drives, so clean each up:
    disable it (so GNOME drops it from the enabled set) then delete its
    directory. Best-effort throughout — returns True if any legacy install was
    found and removed.

    One-off migration shim: once users have had a release or two to upgrade past
    the old UUIDs this can be removed (call site and tests included).
    """
    removed = False
    for uuid in LEGACY_EXTENSION_UUIDS:
        legacy_dir = extensions_root() / uuid
        if not legacy_dir.is_dir():
            continue
        with contextlib.suppress(OSError, subprocess.TimeoutExpired):
            subprocess.run(
                ["gnome-extensions", "disable", uuid],
                capture_output=True,
                text=True,
                check=False,
                timeout=SUBPROCESS_TIMEOUT,
            )
        shutil.rmtree(legacy_dir, ignore_errors=True)
        logger.info(
            "removed the legacy %s extension (renamed to %s) — log out and back "
            "in to load the new one",
            uuid,
            EXTENSION_UUID,
        )
        removed = True
    return removed

ensure_extension

ensure_extension()

Startup hook: install the refresh service and activate the extension.

Composes the two independently-runnable activities core.desktop_integration also exposes. Skipped on non-GNOME desktops.

Source code in src/core/gnome_extension.py

def ensure_extension():
    """Startup hook: install the refresh service and activate the extension.

    Composes the two independently-runnable activities `core.desktop_integration`
    also exposes. Skipped on non-GNOME desktops.
    """
    if shutil.which("gnome-extensions") is None:
        return
    unit_status = install_refresh_unit()
    if unit_status:
        logger.info("%s", unit_status)
    activate_extension()

activate_extension

activate_extension()

Install, refresh, and enable the bundled extension, reporting what it did.

Installs it if missing, keeps the installed copy current, and enables it. Skipped on non-GNOME desktops; non-fatal on missing sources or write failures.

Source code in src/core/gnome_extension.py

def activate_extension():
    """Install, refresh, and enable the bundled extension, reporting what it did.

    Installs it if missing, keeps the installed copy current, and enables it. Skipped on
    non-GNOME desktops; non-fatal on missing sources or write failures.
    """
    if shutil.which("gnome-extensions") is None:
        return

    migrate_legacy_extensions()

    try:
        listed = subprocess.run(
            ["gnome-extensions", "list"],
            capture_output=True,
            text=True,
            check=False,
            timeout=SUBPROCESS_TIMEOUT,
        )
        listed_enabled = subprocess.run(
            ["gnome-extensions", "list", "--enabled"],
            capture_output=True,
            text=True,
            check=False,
            timeout=SUBPROCESS_TIMEOUT,
        )
    except (OSError, subprocess.TimeoutExpired):
        return
    if listed.returncode != 0:
        return

    installed = EXTENSION_UUID in listed.stdout.split()
    # A failed --enabled probe leaves this False, so we fall through and try
    # enabling (a no-op when it's already on).
    already_enabled = (
        listed_enabled.returncode == 0
        and EXTENSION_UUID in listed_enabled.stdout.split()
    )
    dest_dir = extension_dest_dir()
    root = extension_source_dir()

    if installed and already_enabled:
        result = refresh_extension_files(root, dest_dir)
        if result is RefreshResult.REFRESHED:
            logger.info(
                "updated GNOME extension %s to the bundled version — "
                "log out and back in to load it",
                EXTENSION_UUID,
            )
        elif result is RefreshResult.UNCHANGED:
            logger.info(
                "GNOME extension %s already installed and enabled",
                EXTENSION_UUID,
            )
        # ERROR was already noted by refresh_extension_files; stay quiet rather
        # than claim a healthy install.
        return

    if not installed:
        if not all((root / name).is_file() for name in EXTENSION_ASSETS):
            logger.warning(
                "could not auto-install GNOME extension — source files not "
                "found at %s. The panel indicator and grid commands will not "
                "work until the extension is installed manually.",
                root,
            )
            return

        if refresh_extension_files(root, dest_dir) is RefreshResult.ERROR:
            logger.warning(
                "the panel indicator and grid commands will not work until "
                "the GNOME extension is installed manually."
            )
            return

    # On Wayland GNOME only rescans at login, so enable fails with "Unknown
    # extension" until then — treated as "installed, log back in to use".
    try:
        enabled = subprocess.run(
            ["gnome-extensions", "enable", EXTENSION_UUID],
            capture_output=True,
            text=True,
            check=False,
            timeout=SUBPROCESS_TIMEOUT,
        )
        ok = enabled.returncode == 0
    except (OSError, subprocess.TimeoutExpired):
        ok = False

    if installed:
        if ok:
            logger.info(
                "enabled GNOME extension %s (was installed but disabled)",
                EXTENSION_UUID,
            )
        else:
            logger.warning(
                "GNOME extension %s is installed but disabled, and could not "
                "be enabled. Try: gnome-extensions enable %s",
                EXTENSION_UUID,
                EXTENSION_UUID,
            )
        return

    if ok:
        logger.info(
            "installed and enabled GNOME extension %s at %s",
            EXTENSION_UUID,
            dest_dir,
        )
    else:
        logger.info(
            "installed GNOME extension to %s — "
            "log out and back in to finish enabling it.",
            dest_dir,
        )

main

main()

Refresh the installed extension; entry point for the systemd unit.

Source code in src/core/gnome_extension.py

def main():
    """Refresh the installed extension; entry point for the systemd unit."""
    refresh_installed_extension()
    return 0

core.autostart

core.autostart

Whether EasySpeak starts at login, and the switch to change that.

Follows the freedesktop autostart specification: the packages install a system-wide entry, and a user overrides it with their own copy of the same name, marked Hidden=true to turn it off. X-GNOME-Autostart-enabled is kept in step, as GNOME's own settings and the GNOME extension's settings read that key. This is the tray menu's "Start at Login" on desktops other than GNOME; the extension's settings do the same in JavaScript.

user_entry

user_entry()

The user's own autostart entry, which overrides the system-wide one.

It lives in the XDG user config directory, ~/.config unless XDG_CONFIG_HOME says otherwise, where the desktop looks for it too. As the XDG specification demands, a relative XDG_CONFIG_HOME is ignored.

Source code in src/core/autostart.py

def user_entry():
    """The user's own autostart entry, which overrides the system-wide one.

    It lives in the XDG user config directory, `~/.config` unless
    `XDG_CONFIG_HOME` says otherwise, where the desktop looks for it too. As the
    XDG specification demands, a relative `XDG_CONFIG_HOME` is ignored.
    """
    config = os.environ.get("XDG_CONFIG_HOME", "")
    if not Path(config).is_absolute():
        config = Path.home() / ".config"
    return Path(config) / "autostart" / ENTRY_NAME

system_entries

system_entries()

Where packages put the system-wide autostart entry, most important first.

Relative entries in XDG_CONFIG_DIRS are ignored, as the XDG specification demands.

Source code in src/core/autostart.py

def system_entries():
    """Where packages put the system-wide autostart entry, most important first.

    Relative entries in `XDG_CONFIG_DIRS` are ignored, as the XDG specification
    demands.
    """
    dirs = os.environ.get("XDG_CONFIG_DIRS", "").split(":")
    absolute = [Path(d) for d in dirs if d and Path(d).is_absolute()]
    return [d / "autostart" / ENTRY_NAME for d in absolute or [Path("/etc/xdg")]]

read

read(path)

Return a file's text, or None if it's missing or unreadable.

Source code in src/core/autostart.py

def read(path):
    """Return a file's text, or None if it's missing or unreadable."""
    try:
        return path.read_text(encoding="utf-8")
    except OSError:
        return None

value

value(text, key)

Return a key's value in a desktop entry, or None if it's not set.

Source code in src/core/autostart.py

def value(text, key):
    """Return a key's value in a desktop entry, or None if it's not set."""
    match = re.search(rf"^{re.escape(key)}\s*=\s*(\S+)", text, re.MULTILINE)
    return match.group(1).lower() if match else None

enabled_in

enabled_in(text)

Whether a desktop entry says to start at login.

Source code in src/core/autostart.py

def enabled_in(text):
    """Whether a desktop entry says to start at login."""
    return value(text, "Hidden") != "true" and (
        value(text, "X-GNOME-Autostart-enabled") != "false"
    )

with_value

with_value(text, key, new)

Return a desktop entry with key set to new, appended if it was missing.

Source code in src/core/autostart.py

def with_value(text, key, new):
    """Return a desktop entry with `key` set to `new`, appended if it was missing."""
    line = f"{key}={new}"
    pattern = rf"^{re.escape(key)}\s*=.*$"
    if re.search(pattern, text, re.MULTILINE):
        return re.sub(pattern, line, text, flags=re.MULTILINE)
    return (text if text.endswith("\n") else text + "\n") + line + "\n"

is_enabled

is_enabled()

Whether EasySpeak starts at login: the user's entry decides, else the system's.

With neither, nothing starts it, e.g. after a pip install without easyspeak --configure autostart.

Source code in src/core/autostart.py

def is_enabled():
    """Whether EasySpeak starts at login: the user's entry decides, else the system's.

    With neither, nothing starts it, e.g. after a pip install without
    `easyspeak --configure autostart`.
    """
    for path in (user_entry(), *system_entries()):
        text = read(path)
        if text is not None:
            return enabled_in(text)
    return False

set_enabled

set_enabled(enabled)

Turn starting at login on or off, through the user's own entry.

The entry is based on the user's existing one, so hand edits survive, else on the system-wide one, else on the template EasySpeak ships.

Source code in src/core/autostart.py

def set_enabled(enabled):
    """Turn starting at login on or off, through the user's own entry.

    The entry is based on the user's existing one, so hand edits survive, else
    on the system-wide one, else on the template EasySpeak ships.
    """
    sources = (read(path) for path in (user_entry(), *system_entries()))
    text = next((t for t in sources if t is not None), None) or data_text(TEMPLATE)
    flag = "true" if enabled else "false"
    text = with_value(text, "Hidden", "false" if enabled else "true")
    text = with_value(text, "X-GNOME-Autostart-enabled", flag)
    entry = user_entry()
    entry.parent.mkdir(parents=True, exist_ok=True)
    entry.write_text(text, encoding="utf-8")

core.desktop_integration

core.desktop_integration

Optional desktop integration for non-packaged (pip / dev) installs.

Debian/RPM packages ship the launcher, autostart entry, and refresh service as system files, so users there need do nothing. A pip or uv install has no such hooks, so easyspeak --configure writes the per-user equivalents under $HOME.

The GNOME extension and its refresh service are always set up — they're what makes the panel/tray icon work — while the launcher and autostart entries are opt-in, since not everyone wants EasySpeak in their menu or starting at login.

autostart_path

autostart_path()

User-local autostart entry, read by the session at login.

Source code in src/core/desktop_integration.py

def autostart_path():
    """User-local autostart entry, read by the session at login."""
    return user_entry()

desktop_path

desktop_path()

User-local application launcher (app grid / menu).

Source code in src/core/desktop_integration.py

def desktop_path():
    """User-local application launcher (app grid / menu)."""
    return Path.home() / ".local" / "share" / "applications" / INSTALLED_DESKTOP_NAME

install_autostart_file

install_autostart_file()

Write the per-user autostart entry so EasySpeak starts at login.

Source code in src/core/desktop_integration.py

def install_autostart_file():
    """Write the per-user autostart entry so EasySpeak starts at login."""
    return _install_data_file(AUTOSTART_TEMPLATE, autostart_path())

install_desktop_file

install_desktop_file()

Write the per-user application launcher.

Source code in src/core/desktop_integration.py

def install_desktop_file():
    """Write the per-user application launcher."""
    return _install_data_file(DESKTOP_TEMPLATE, desktop_path())

configure

configure(items)

Run the requested setup activities, then return.

Each activity in items is independent, so callers get exactly what they ask for and nothing implicitly. The CLI fills in its own default set, so an empty items here simply does nothing.

Source code in src/core/desktop_integration.py

def configure(items: set[str]):
    """Run the requested setup activities, then return.

    Each activity in `items` is independent, so callers get exactly what they ask
    for and nothing implicitly. The CLI fills in its own default set, so an empty
    `items` here simply does nothing.
    """
    if "extension" in items:
        activate_extension()
    if "service" in items:
        # install_refresh_unit only reports when it changes something; say so
        # either way, so this item is never silent.
        logger.info(install_refresh_unit() or "refresh service already configured")
    if "desktop" in items:
        install_desktop_file()
    if "autostart" in items:
        install_autostart_file()

preview

preview(item)

Write an item's content to stdout and its target install path to stderr.

Backs --preview. The TARGET: line goes to stderr so packaging can capture just the content off stdout; the content is written verbatim, since adding a trailing newline is packaging's concern, not this function's.

Source code in src/core/desktop_integration.py

def preview(item):
    """Write an item's content to stdout and its target install path to stderr.

    Backs `--preview`. The `TARGET:` line goes to stderr so packaging can capture
    just the content off stdout; the content is written verbatim, since adding a
    trailing newline is packaging's concern, not this function's.
    """
    render, path = {
        "extension": (_manifest_text, extension_dest_dir),
        "service": (unit_text, unit_path),
        "desktop": (partial(data_text, DESKTOP_TEMPLATE), desktop_path),
        "autostart": (partial(data_text, AUTOSTART_TEMPLATE), autostart_path),
    }[item]
    sys.stderr.write(f"TARGET: {path()}\n")
    sys.stdout.write(render())