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 -> Whisper), 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.wakeword = None
    self.audio = None
    self.stream = None
    self.last_wake_time = 0
    self.misunderstand_count = 0
    self.help_shown = False
    self.keep_listening = 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()
    # GNOME 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.

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.
    """
    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
        )
    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 next main-loop iteration (handled by the tray controller) so the triggering command can finish, and speak, first.

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 next main-loop iteration (handled by the tray
    controller) so the triggering command can finish, and speak, first.
    """
    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 module from the plugins/ dir.

Files are loaded in sorted order (numeric prefixes set load order); names starting with _ are skipped. A module 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.

Source code in src/core/main.py

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

    Files are loaded in sorted order (numeric prefixes set load order); names
    starting with `_` are skipped. A module 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.
    """
    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))

    for file in sorted(plugins_dir.glob("*.py")):
        if file.name.startswith("_"):
            continue

        module_name = f"plugins.{file.stem}"
        try:
            module = importlib.import_module(module_name)

            if hasattr(module, "NAME") and hasattr(module, "handle"):
                if hasattr(module, "setup"):
                    module.setup(self)

                self.plugins.append(module)
                logger.info("  ✓ Loaded: %s", module.NAME)
            else:
                logger.warning(
                    "  ✗ Invalid plugin: %s (missing NAME or handle)", file.name
                )
        except Exception as e:
            logger.warning("  ✗ Failed to load %s: %s", file.name, e)

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.

Returns False to exit.

Source code in src/core/main.py

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

    Returns False to exit.
    """
    cmd = strip_wake_words(cmd)
    self.unrecognized = False

    if not cmd:
        return True

    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)

    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 _ 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 _ 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)

Transcribe raw PCM audio to text with Whisper.

prompt biases recognition (defaults to the command vocabulary). 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.

Source code in src/core/main.py

def transcribe(self, audio_data, prompt=None):
    """Transcribe raw PCM audio to text with Whisper.

    `prompt` biases recognition (defaults to the command vocabulary). 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.
    """
    samples = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0

    use_prompt = prompt or COMMAND_PROMPT
    started = time.monotonic()
    segments, _ = self.whisper.transcribe(
        samples,
        initial_prompt=use_prompt,
        beam_size=1,
        vad_filter=True,
        language="en",
        # 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._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)

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.

Yields each recognised command, lowercased and stripped of surrounding punctuation. 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,
):
    """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.

    Yields each recognised command, lowercased and stripped of surrounding
    punctuation. 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.speak(f"Leaving {label}. Say {WAKE_WORD_SPOKEN} to continue.")
            return

        remaining = deadline - time.monotonic()
        if remaining <= 0:
            logger.info("%s mode ended: idle", label.capitalize())
            self.speak(f"Leaving {label}. Say {WAKE_WORD_SPOKEN} to continue.")
            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)
        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()

        logger.info(
            "Loading Whisper (%s, %s, cpu_threads=%s)...",
            WHISPER_MODEL,
            WHISPER_COMPUTE_TYPE,
            WHISPER_CPU_THREADS or "auto",
        )
        try:
            self.whisper = load_whisper_model()
        except RuntimeError as exc:
            logger.error("Cannot start EasySpeak: %s", exc)  # noqa: TRY400
            raise SystemExit(1) from exc

        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("\nBye!")
        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".

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".
    """
    cmd = cmd.lower().strip()
    for wake in sorted(WAKE_PREFIXES, key=len, reverse=True):
        if cmd == wake:
            return ""
        if cmd.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). 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(
        "--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.

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.
    """
    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.configure is not None:
        from .desktop_integration import configure

        configure(args.configure)
        return

    from .main import EasySpeak

    EasySpeak().run()

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_whisper_model(), which builds the faster-whisper model from them. Most are plain constants; these honor an EASYSPEAK_* environment variable:

• EASYSPEAK_HOTKEY
• EASYSPEAK_OFFLINE
• EASYSPEAK_PIPER_BIN
• EASYSPEAK_PIPER_MODEL
• EASYSPEAK_SOUNDS_DIR
• 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.

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)

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.

Source code in src/core/log.py

def configure(level):
    """Attach a message-only stderr handler to the package loggers."""
    # 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.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

GNOME 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 (served by the bundled gnome@easyspeak.dev GNOME Shell extension) and owns everything about it so the audio loop in core.main stays about audio. It 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: on a non-GNOME desktop (no gdbus/extension) 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._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.

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.
    """
    self.take_command()
    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)

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

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 extension (tray menu or Quick Settings toggle), so releasing the mic while the indicator is missing (no GNOME/extension) 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 GNOME Shell 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 extension (tray menu or Quick Settings
    toggle), so releasing the mic while the indicator is missing (no
    GNOME/extension) 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 GNOME Shell 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",
        "org.gnome.Shell",
        "--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 (non-GNOME) 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

core.about

core.about

Standalone libadwaita "About EasySpeak" window.

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 ""

main

main()

Show the About window 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 window 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.

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).

Raises if RemoteDesktop is unavailable, 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).

    Raises if RemoteDesktop is unavailable, so the caller can report why.
    """
    conn = open_dbus_connection(bus="SESSION")
    try:
        reply = conn.send_and_get_reply(
            new_method_call(_REMOTE_DESKTOP, "CreateSession")
        )
        session = DBusAddress(
            reply.body[0],
            bus_name=_BUS,
            interface=f"{_BUS}.Session",
        )
        conn.send_and_get_reply(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:
            conn.send_and_get_reply(
                new_method_call(session, "NotifyKeyboardKeycode", "ub", (keycode, True))
            )
        for keycode in reversed(keycodes):
            conn.send_and_get_reply(
                new_method_call(
                    session, "NotifyKeyboardKeycode", "ub", (keycode, False)
                )
            )
        conn.send_and_get_reply(new_method_call(session, "Stop"))
    finally:
        conn.close()

parse_key_request

parse_key_request(words, bare_allowed)

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

Accepts an optional "press" prefix, one- or two-word key names, 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.

Source code in src/core/mediakeys.py

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

    Accepts an optional "press" prefix, one- or two-word key names, 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.
    """
    explicit = bool(words) and words[0] == "press"
    if explicit:
        words = words[1:]
    if not words:
        return None

    for length in (2, 1):
        name = " ".join(words[:length])
        name = KEY_ALIASES.get(name, name)
        if name not in KEYS 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 SPOKEN_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.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 Path.home() / ".config" / "autostart" / INSTALLED_DESKTOP_NAME

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())