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 OpenWakeWord 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, HOTKEY_ENABLED)
    self._push_to_talk = None

host_run

host_run(cmd, background=False)

Run a shell command.

Source code in src/core/main.py

def host_run(self, cmd, background=False):
    """Run a shell command."""
    if background:
        return subprocess.Popen(
            cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
        )
    return subprocess.run(cmd, capture_output=True, text=True)

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()
        os.remove(f.name)
        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",
        )
        self.whisper = load_whisper_model()

        # 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_data = np.frombuffer(pcm, dtype=np.int16)

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

                prediction = self.wakeword.predict(audio_data)
                score = prediction.get(WAKE_WORD, 0)

                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.

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."""
    parser = argparse.ArgumentParser(
        prog="easyspeak", description="Voice control for Linux desktops."
    )
    verbosity = parser.add_mutually_exclusive_group()
    verbosity.add_argument(
        "-v", "--verbose", action="store_true", help="show debug output"
    )
    verbosity.add_argument(
        "-q", "--quiet", action="store_true", help="show only warnings and errors"
    )
    return parser.parse_args(argv)

run

run(argv=None)

Start the application.

Source code in src/core/cli.py

def run(argv=None):
    """Start the application."""
    args = parse_args(argv)
    log.configure(log.resolve_level(verbose=args.verbose, quiet=args.quiet))
    EasySpeak().run()

core.config

core.config

Tuning constants and model factory for EasySpeak.

Override the EASYSPEAK_* environment variables to customise behaviour without editing source. Plugin-specific host-environment setup lives in each plugin's own setup() hook, not here.

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.

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."""
    return WhisperModel(model_name, compute_type=compute_type, cpu_threads=cpu_threads)

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:
            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 easyspeak@local 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 defaults to a no-op so the tray works headless and in tests.

For the spoken feedback callback (core.speak) the plugin only announces the attempt ("Going to sleep."); the tray confirms or, when it can't actually sleep, explains — since only it knows whether sleep engaged. 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` defaults to a no-op so the tray works headless and in tests.

    For the spoken feedback callback (core.speak) the plugin only announces the
    *attempt* ("Going to sleep."); the tray confirms or, when it can't actually
    sleep, explains — since only it knows whether sleep engaged. 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.

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

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:
        self._sleep_requested = False
        return self._sleep(release_mic, acquire_mic)
    return TrayAction.CONTINUE

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 unparseable or empty 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 unparseable or empty 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)
    self._enabled = enabled and bool(self._groups)
    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)

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 package dir holding the bundled assets.

Resolved relative to this module (src/) so it works in both editable and wheel installs.

Source code in src/core/gnome_extension.py

def extension_source_dir():
    """Return the package dir holding the bundled assets.

    Resolved relative to this module (`src/`) so it works in both editable and wheel
    installs.
    """
    return Path(__file__).resolve().parents[1]

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

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, 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, write/enable failure).
    """
    if shutil.which("systemctl") is None:
        return None

    unit_path = _unit_path()
    desired = _unit_text()
    try:
        current = unit_path.read_text() if unit_path.is_file() else None
    except OSError:
        current = None

    changed = current != desired
    if changed:
        try:
            unit_path.parent.mkdir(parents=True, exist_ok=True)
            unit_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_extension

migrate_legacy_extension()

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

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

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

Source code in src/core/gnome_extension.py

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

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

    One-off migration shim: once users have had a release or two to upgrade past
    `easyspeak-grid@local` this can be removed (call site and tests included).
    """
    legacy_dir = extensions_root() / LEGACY_EXTENSION_UUID
    if not legacy_dir.is_dir():
        return False
    with contextlib.suppress(OSError, subprocess.TimeoutExpired):
        subprocess.run(
            ["gnome-extensions", "disable", LEGACY_EXTENSION_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",
        LEGACY_EXTENSION_UUID,
        EXTENSION_UUID,
    )
    return True

ensure_extension

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

    unit_status = install_refresh_unit()
    if unit_status:
        logger.info("%s", unit_status)

    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