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
    self.last_misunderstand_time = 0
    # 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

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 = cmd.lower()
    for wake in [
        "hey jarvis",
        "hey jarvis,",
        "hey, jarvis",
        "hey, jarvis,",
        "hey jarvis.",
        "jarvis",
        "jarvis,",
    ]:
        cmd = cmd.replace(wake, "").strip()
    cmd = cmd.strip(".,!? ")
    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,
        )

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

record_until_silence

record_until_silence(should_continue=None)

Record mic audio until a short silence, capped at five seconds.

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

Source code in src/core/main.py

def record_until_silence(self, should_continue=None):
    """Record mic audio until a short silence, capped at five seconds.

    Returns the captured PCM bytes. Plugin-facing. `should_continue` (used by
    push-to-talk) lets a key release cut the recording short instead of waiting out
    the silence window.
    """
    frames = []
    silent_chunks = 0
    chunks_needed = int(SILENCE_DURATION * 16000 / 1600)

    for i in range(int(5 * 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.

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.
    """
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        with wave.open(f.name, "wb") as wf:
            wf.setnchannels(1)
            wf.setsampwidth(2)
            wf.setframerate(16000)
            wf.writeframes(audio_data)

        use_prompt = prompt or COMMAND_PROMPT
        segments, _ = self.whisper.transcribe(
            f.name, initial_prompt=use_prompt, beam_size=1, vad_filter=True
        )
        text = " ".join([s.text for s in segments]).strip()
        Path(f.name).unlink()
        return text

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

        # The GNOME Shell extension powers both the panel indicator and the
        # mouse grid, so core (not a plugin) owns installing/refreshing/enabling
        # it before anything starts driving it over D-Bus.
        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:
            # PortAudio probes every ALSA/JACK device on init, spamming stderr
            # about hardware this machine lacks; redirect fd 2 to silence that
            # C-level noise. PyAudio init emits no Python stderr of its own; the
            # stream is opened separately by _open_stream().
            with suppressed_c_stderr():
                self.audio = pyaudio.PyAudio()
            self._open_stream()

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

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",
    )
    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,
        )
        # Hand the write end of the player's input pipe entirely to piper.
        # If the parent kept its own copy open, the player would never see
        # EOF when piper exits and would hang forever in read() (a dead
        # piper leaving an immortal pw-play behind), wedging the pipeline.
        self._player.stdin.close()
        # A bad model path or unsupported player flag lets Popen succeed but
        # the process dies milliseconds later. Give it a brief grace period
        # and confirm both are still alive, so warmup reports the failure
        # once instead of speak() rebuilding a dead pipeline on every phrase.
        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.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:
        # A voice "go to sleep" reaches here via _sleep_requested, and the
        # sleep plugin has already spoken; a button mute (COMMAND_MUTE) has
        # not, so only then does the tray announce the deactivation itself.
        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.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.
    """
    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 pressed in (True, False):
            conn.send_and_get_reply(
                new_method_call(
                    session, "NotifyKeyboardKeycode", "ub", (keycode, pressed)
                )
            )
        conn.send_and_get_reply(new_method_call(session, "Stop"))
    finally:
        conn.close()

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():
        # The package owns the unit; don't shadow it with a user copy. Clear out a
        # stale one a prior pip/dev install left — `~/.config/systemd/user` takes
        # precedence over the packaged path and would point at a vanished
        # interpreter, so it must go for the packaged unit to take effect.
        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:
            # refresh_extension_files already reported the write error (with the
            # OSError detail) on stderr; add only the consequence, not a second
            # description of the same failure.
            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