Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f1ed2ce085 | |||
| 43996e7b8c | |||
| 6cd745dc57 | |||
| 6ecc775c6d |
@@ -24,6 +24,54 @@ This repository is a screen reader. Prioritize accessibility, correctness, and s
|
||||
- If repo and installed behavior differ, prefer rebuilding with `./build-local.sh` over patching the installed package directly.
|
||||
- Treat direct edits under `~/.local/.../cthulhu/` as an exception path that requires explicit user approval.
|
||||
|
||||
## Bug intake
|
||||
- Assume reporters may have little or no development experience. Ask for user-visible steps and guide them through logs or diagnostic commands without expecting them to identify the subsystem or use technical terminology.
|
||||
- Capture the application or browser, page URL when relevant, desktop/session type, exact steps, expected result, actual result, and whether the problem is reproducible elsewhere.
|
||||
- Separate keyboard focus from speech output: ask whether focus fails to move, moves but is not announced, or is reachable through another navigation method.
|
||||
- For web issues, determine whether the behavior is site-specific, browser-specific, or reproducible across similar controls before changing code.
|
||||
- When a log is needed, guide the reporter through launching the locally installed build with `~/.local/bin/cthulhu --debug-file /tmp/cthulhu.log --debug`, reproducing the problem once, exiting Cthulhu, and providing `/tmp/cthulhu.log` for review.
|
||||
- If a reporter still sees the same problem after a fix, verify that the refreshed build is actually installed and running before assuming the patch failed. Guide the reporter through the check rather than assuming they know how the local install works.
|
||||
|
||||
## Contribution workflow (including generated patches)
|
||||
- Assume contributors may use code-generation tools without understanding every changed line. Review the resulting code, not the contributor's confidence in it.
|
||||
- Start from a reproducible user-visible problem. Complete the relevant bug-intake steps before changing code.
|
||||
- Investigate the full confirmed behavior class before implementing a fix. For example, a stale-focus bug seen in one browser may also affect other applications or empty workspaces.
|
||||
- Prefer the smallest root-cause fix that covers the full confirmed behavior class. Do not add app-specific exceptions, desktop-specific branches, compatibility fallbacks, or broad refactors unless the evidence requires them.
|
||||
- Add or update automated regression tests for the general behavior and the originally reported workflow whenever practical.
|
||||
- Read every generated diff before committing. Remove unrelated rewrites, speculative cleanup, dead code, debug leftovers, and generated files that are not required by the fix.
|
||||
- Never commit a behavior change solely because it compiles or because an automated tool says it works. Verify the real user-facing workflow after rebuilding the installed copy.
|
||||
|
||||
## Verification checklist before commit
|
||||
Run the narrowest relevant checks first, then broaden testing based on the affected behavior:
|
||||
- For documentation-only changes, diff inspection is sufficient unless the edited documentation includes commands or generated content that need validation.
|
||||
|
||||
1. Inspect the diff:
|
||||
- `git diff --check`
|
||||
- `git status --short`
|
||||
- `git diff --stat`
|
||||
- `git diff -- <changed files>`
|
||||
2. Run syntax checks for each changed Python file:
|
||||
- `python -m py_compile <changed .py files>`
|
||||
3. Run focused automated regression tests:
|
||||
- `python -m unittest <relevant test modules>`
|
||||
- For shared input, focus, script lifecycle, settings, plugin loading, or installation changes, run `./test-local.sh` after the focused tests pass.
|
||||
4. For code or behavior changes, rebuild the local installed copy:
|
||||
- `./build-local.sh`
|
||||
5. For code or behavior changes, confirm the runtime import resolves to the refreshed local install:
|
||||
- `python - <<'PY'`
|
||||
- `import importlib.util`
|
||||
- `print(importlib.util.find_spec("cthulhu").origin)`
|
||||
- `PY`
|
||||
6. For code or behavior changes, reproduce the original user-visible workflow against the rebuilt copy.
|
||||
7. For focus, keyboard, or window-tracking changes, manually test closely related regressions:
|
||||
- Xorg and the user's active window manager or desktop.
|
||||
- Switching among browser content, terminal windows, GTK applications, dialogs, and empty workspaces.
|
||||
- Returning to the original application after each switch.
|
||||
- Cthulhu shortcuts, structural navigation, flat review, and any delegated key handling such as Fenrir in XTerm.
|
||||
- Both key press and key release behavior for modifiers, NumLock, and keypad keys when relevant.
|
||||
- Clean shutdown without crashing the browser or leaving grabs behind.
|
||||
8. Record what was tested, what could not be tested locally, and any remaining uncertainty in the commit message or review notes.
|
||||
|
||||
## Platform support stance
|
||||
- **critical** Robust Xorg support is required and is a merge gate for Cthulhu.
|
||||
- Wayland support is desirable, but it is secondary to keeping Xorg stable and usable.
|
||||
@@ -45,7 +93,7 @@ This repository is a screen reader. Prioritize accessibility, correctness, and s
|
||||
- Screen-reader-first UX: assume non-visual navigation.
|
||||
- No keyboard traps; Tab/Shift+Tab must move through all controls.
|
||||
- Use clear, complete labels (e.g., “Confirm Password”, not “Confirm”).
|
||||
- **No Speech-Dispatcher usage in GUI apps** (no `spd-say`); rely on the accessibility API.
|
||||
- Do not invoke `spd-say` or add direct speech output to GUI dialogs. Expose accessible UI state and let the active screen reader announce it.
|
||||
|
||||
### Python GTK (Gtk 3)
|
||||
- Associate labels with controls (mnemonics + buddy widget):
|
||||
@@ -54,11 +102,6 @@ This repository is a screen reader. Prioritize accessibility, correctness, and s
|
||||
- For `Gtk.TextView`, call `set_accepts_tab(False)` so Tab moves focus.
|
||||
- For custom widgets, set accessible name/role via ATK where applicable.
|
||||
|
||||
### PySide6 / Qt
|
||||
- Set `setAccessibleName()` on all widgets.
|
||||
- Associate labels via `setBuddy()`.
|
||||
- Use `setAccessibleDescription()` when extra context is needed.
|
||||
|
||||
## Shell script rules
|
||||
- **Bash variables must be `camelCase`** (except system env vars like `ACCESSIBILITY_ENABLED`).
|
||||
- If you edit a `#!/usr/bin/env bash`, `#!/bin/bash`, or POSIX `sh` script, run `shellcheck` and fix issues.
|
||||
@@ -67,13 +110,20 @@ This repository is a screen reader. Prioritize accessibility, correctness, and s
|
||||
## Plugins (Cthulhu)
|
||||
- System plugins live in `src/cthulhu/plugins/`.
|
||||
- Follow existing plugin patterns (keybinding registration, GTK dialog/window patterns, debug logging).
|
||||
- Ensure plugin install integration is wired via Meson (`src/cthulhu/plugins/meson.build` + plugin subdir `meson.build`).
|
||||
|
||||
## Meson install reminder (important)
|
||||
- If you add new Python modules under `src/cthulhu/`, update `src/cthulhu/meson.build` so they get installed (otherwise imports can fail after install).
|
||||
- If you add a new plugin directory, update `src/cthulhu/plugins/meson.build` and add a `meson.build` in the plugin directory.
|
||||
|
||||
## Settings policy (important)
|
||||
- Keep persistent settings in Cthulhu TOML files through the existing settings manager and TOML backend.
|
||||
- Do not add `gsettings`, `dconf`, or `Gio.Settings` dependencies or fallback paths.
|
||||
|
||||
## Dependency synchronization (important)
|
||||
- This repository does not use `requirements.txt`. When adding, removing, or changing a runtime, optional, or build dependency, keep the relevant dependency declarations and documentation synchronized.
|
||||
- Check `pyproject.toml`, `meson.build`, the dependency sections in `README.md` and `README-DEVELOPMENT.md`, and each applicable distro package recipe under `distro-packages/`.
|
||||
- In particular, update both Arch Linux `PKGBUILD` files and, when relevant, Slint's `cthulhu-info` and `README`. Review the Slackware and Slint build files for any corresponding build-dependency changes.
|
||||
- Preserve the distinction between required, optional, and build-only dependencies so package builds install what Cthulhu actually needs without forcing optional features on every user.
|
||||
|
||||
## Common Cthulhu agent mistakes
|
||||
- Checking the import origin, seeing `~/.local/...`, and then editing the installed package instead of the repo.
|
||||
- Forgetting that `./build-local.sh` is the normal way to apply repo changes into the installed copy for testing.
|
||||
- Making repo fixes and then diagnosing the old installed copy without rebuilding.
|
||||
- Do not edit or diagnose a stale installed copy under `~/.local/...`; update the repo and rebuild with `./build-local.sh`.
|
||||
|
||||
@@ -57,9 +57,9 @@ toolkit, OpenOffice/LibreOffice, Gecko, WebKitGtk, and KDE Qt toolkit.
|
||||
- **Preserves exit key**: Only sleep toggle remains active
|
||||
|
||||
### Self-Voicing
|
||||
- **Unix socket interface**: Direct speech output via `/tmp/cthulhu.sock`
|
||||
- **Unix socket interface**: Direct speech output via `${XDG_RUNTIME_DIR}/cthulhu.sock`
|
||||
- **External integration**: Other applications can speak through Cthulhu
|
||||
- **Simple protocol**: `echo "text" | socat - UNIX-CLIENT:/tmp/cthulhu.sock`
|
||||
- **Simple protocol**: `echo "text" | socat - UNIX-CLIENT:"${XDG_RUNTIME_DIR}/cthulhu.sock"`
|
||||
|
||||
## D-Bus Remote Controller
|
||||
|
||||
@@ -275,9 +275,9 @@ Cthulhu offers a mechanism through which messages may be spoken directly by the
|
||||
|
||||
```bash
|
||||
# Speak hello world.
|
||||
echo "Hello world." | socat - UNIX-CLIENT:/tmp/cthulhu.sock
|
||||
echo "Hello world." | socat - UNIX-CLIENT:"${XDG_RUNTIME_DIR}/cthulhu.sock"
|
||||
# Speak Hello world without interrupting the previous speech.
|
||||
echo "<!#APPEND#!>Hello world." | socat - UNIX-CLIENT:/tmp/cthulhu.sock
|
||||
# Make hello world persistant in Braille.
|
||||
echo "Hello world.<#APPEND#>" | socat - UNIX-CLIENT:/tmp/cthulhu.sock
|
||||
echo "<#APPEND#>Hello world." | socat - UNIX-CLIENT:"${XDG_RUNTIME_DIR}/cthulhu.sock"
|
||||
# Make hello world persistent in Braille.
|
||||
echo "Hello world.<#PERSISTENT#>" | socat - UNIX-CLIENT:"${XDG_RUNTIME_DIR}/cthulhu.sock"
|
||||
```
|
||||
|
||||
@@ -1787,6 +1787,35 @@
|
||||
<property name="top_attach">5</property>
|
||||
</packing>
|
||||
</child>
|
||||
<child>
|
||||
<object class="GtkLabel" id="hardwareDeviceLabel">
|
||||
<property name="visible">False</property>
|
||||
<property name="can_focus">False</property>
|
||||
<property name="xalign">1</property>
|
||||
<property name="label" translatable="yes">Serial _device:</property>
|
||||
<property name="use_underline">True</property>
|
||||
<property name="justify">right</property>
|
||||
<property name="mnemonic_widget">hardwareDeviceCombo</property>
|
||||
<accessibility>
|
||||
<relation type="label-for" target="hardwareDeviceCombo"/>
|
||||
</accessibility>
|
||||
</object>
|
||||
<packing>
|
||||
<property name="left_attach">0</property>
|
||||
<property name="top_attach">9</property>
|
||||
</packing>
|
||||
</child>
|
||||
<child>
|
||||
<object class="GtkComboBox" id="hardwareDeviceCombo">
|
||||
<property name="visible">False</property>
|
||||
<property name="can_focus">False</property>
|
||||
<signal name="changed" handler="hardwareDeviceChanged" swapped="no"/>
|
||||
</object>
|
||||
<packing>
|
||||
<property name="left_attach">1</property>
|
||||
<property name="top_attach">9</property>
|
||||
</packing>
|
||||
</child>
|
||||
</object>
|
||||
</child>
|
||||
</object>
|
||||
|
||||
@@ -168,6 +168,9 @@ class CthulhuSetupGUI(cthulhu_gtkbuilder.GtkBuilderWrapper):
|
||||
self.speechFamiliesChoice = None
|
||||
self.speechFamiliesChoices = None
|
||||
self.speechFamiliesModel = None
|
||||
self.hardwareDeviceChoice = None
|
||||
self.hardwareDeviceChoices = None
|
||||
self.hardwareDeviceModel = None
|
||||
self.speechLanguagesChoice = None
|
||||
self.speechLanguagesChoices = None
|
||||
self.speechLanguagesModel = None
|
||||
@@ -405,6 +408,11 @@ class CthulhuSetupGUI(cthulhu_gtkbuilder.GtkBuilderWrapper):
|
||||
self._initComboBox(self.get_widget("speechLanguages"))
|
||||
self.speechFamiliesModel = \
|
||||
self._initComboBox(self.get_widget("speechFamilies"))
|
||||
try:
|
||||
self.hardwareDeviceModel = \
|
||||
self._initComboBox(self.get_widget("hardwareDeviceCombo"))
|
||||
except AttributeError:
|
||||
self.hardwareDeviceModel = None
|
||||
self.echoSpeechServersModel = \
|
||||
self._initComboBox(self.get_widget("echoSpeechServers"))
|
||||
self.echoSpeechFamiliesModel = \
|
||||
@@ -1703,6 +1711,8 @@ class CthulhuSetupGUI(cthulhu_gtkbuilder.GtkBuilderWrapper):
|
||||
#
|
||||
self.initializingSpeech = True
|
||||
self._setupSpeechSystems(factories)
|
||||
self._setupHardwareDevice()
|
||||
self._updateHardwareDeviceVisibility()
|
||||
self.initializingSpeech = False
|
||||
|
||||
def _getSpeechDispatcherFactory(self):
|
||||
@@ -3847,6 +3857,118 @@ print(json.dumps(result))
|
||||
self.prefsDict["onlySpeakDisplayedText"] = enable
|
||||
self.get_widget("contextOptionsGrid").set_sensitive(not enable)
|
||||
|
||||
|
||||
def _scanSerialDevices(self):
|
||||
"""Scan for available serial devices and return a list of paths."""
|
||||
import glob
|
||||
devices = []
|
||||
patterns = [
|
||||
"/dev/ttyUSB*",
|
||||
"/dev/ttyACM*",
|
||||
"/dev/ttyS*",
|
||||
"/dev/ttyAMA*",
|
||||
"/dev/rfcomm*",
|
||||
"/dev/serial/by-id/*",
|
||||
]
|
||||
for pattern in patterns:
|
||||
devices.extend(glob.glob(pattern))
|
||||
devices = sorted(set(devices))
|
||||
return devices
|
||||
|
||||
def _setupHardwareDevice(self):
|
||||
"""Sets up the hardware device combo box with available serial ports.
|
||||
|
||||
Populates the combo with scanned serial devices and restores the
|
||||
previously saved selection if still available.
|
||||
"""
|
||||
if self.hardwareDeviceModel is None:
|
||||
return
|
||||
combobox = self.get_widget("hardwareDeviceCombo")
|
||||
combobox.set_model(None)
|
||||
self.hardwareDeviceModel.clear()
|
||||
self.hardwareDeviceChoices = []
|
||||
|
||||
devices = self._scanSerialDevices()
|
||||
saved_device = self.prefsDict.get("hardwareSpeechDevice",
|
||||
settings.hardwareSpeechDevice)
|
||||
|
||||
# Always include a "(none)" option so the user can clear the device
|
||||
self.hardwareDeviceChoices.append("")
|
||||
self.hardwareDeviceModel.append((0, "(none)"))
|
||||
i = 1
|
||||
for device in devices:
|
||||
self.hardwareDeviceChoices.append(device)
|
||||
self.hardwareDeviceModel.append((i, device))
|
||||
i += 1
|
||||
|
||||
# If the saved device is not in the scanned list but is non-empty,
|
||||
# append it so the user still sees their configured device.
|
||||
if saved_device and saved_device not in devices:
|
||||
self.hardwareDeviceChoices.append(saved_device)
|
||||
self.hardwareDeviceModel.append((i, saved_device))
|
||||
i += 1
|
||||
|
||||
combobox.set_model(self.hardwareDeviceModel)
|
||||
self._setHardwareDeviceChoice(saved_device)
|
||||
|
||||
def _setHardwareDeviceChoice(self, device_name):
|
||||
"""Set the active item in the hardware device combo box.
|
||||
|
||||
Arguments:
|
||||
- device_name: the device path to select.
|
||||
"""
|
||||
if not self.hardwareDeviceChoices:
|
||||
self.hardwareDeviceChoice = None
|
||||
return
|
||||
|
||||
for i, choice in enumerate(self.hardwareDeviceChoices):
|
||||
if choice == device_name:
|
||||
self.get_widget("hardwareDeviceCombo").set_active(i)
|
||||
self.hardwareDeviceChoice = choice
|
||||
return
|
||||
|
||||
self.get_widget("hardwareDeviceCombo").set_active(0)
|
||||
self.hardwareDeviceChoice = self.hardwareDeviceChoices[0]
|
||||
|
||||
def _updateHardwareDeviceVisibility(self):
|
||||
"""Show or hide the hardware device combo based on speech system.
|
||||
|
||||
The hardware device selector is only visible when the hardware
|
||||
speech synthesizer factory is active.
|
||||
"""
|
||||
if self.hardwareDeviceModel is None:
|
||||
return
|
||||
is_hardware = False
|
||||
if self.speechSystemsChoice:
|
||||
try:
|
||||
is_hardware = (
|
||||
self.speechSystemsChoice.__name__ == "hardwarefactory"
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
self.get_widget("hardwareDeviceLabel").set_visible(is_hardware)
|
||||
self.get_widget("hardwareDeviceCombo").set_visible(is_hardware)
|
||||
|
||||
def hardwareDeviceChanged(self, widget):
|
||||
"""Signal handler for the hardware device combo box changed signal.
|
||||
|
||||
Arguments:
|
||||
- widget: the component that generated the signal.
|
||||
"""
|
||||
if self.initializingSpeech:
|
||||
return
|
||||
|
||||
selected_index = widget.get_active()
|
||||
if selected_index >= 0 and selected_index < len(self.hardwareDeviceChoices):
|
||||
self.hardwareDeviceChoice = self.hardwareDeviceChoices[selected_index]
|
||||
else:
|
||||
self.hardwareDeviceChoice = None
|
||||
|
||||
# Update runtime settings so the factory sees the new device
|
||||
if self.hardwareDeviceChoice is not None:
|
||||
settings.hardwareSpeechDevice = self.hardwareDeviceChoice
|
||||
|
||||
def speechSystemsChanged(self, widget):
|
||||
"""Signal handler for the "changed" signal for the speechSystems
|
||||
GtkComboBox widget. The user has selected a different speech
|
||||
@@ -3866,6 +3988,7 @@ print(json.dumps(result))
|
||||
self._setupSpeechServers()
|
||||
self._setupEchoSpeechServers()
|
||||
self._setEchoVoiceItems()
|
||||
self._updateHardwareDeviceVisibility()
|
||||
|
||||
def speechServersChanged(self, widget):
|
||||
"""Signal handler for the "changed" signal for the speechServers
|
||||
@@ -4927,6 +5050,16 @@ print(json.dumps(result))
|
||||
self.prefsDict["speechServerFactory"] = \
|
||||
self.speechSystemsChoice.__name__
|
||||
|
||||
# Save hardware speech device setting when hardware factory is active
|
||||
if self.speechSystemsChoice and \
|
||||
self.speechSystemsChoice.__name__ == "hardwarefactory":
|
||||
if self.hardwareDeviceChoice is not None:
|
||||
self.prefsDict["hardwareSpeechDevice"] = self.hardwareDeviceChoice
|
||||
else:
|
||||
self.prefsDict["hardwareSpeechDevice"] = ""
|
||||
else:
|
||||
self.prefsDict["hardwareSpeechDevice"] = settings.hardwareSpeechDevice
|
||||
|
||||
speechServerChoice = self._getSpeechServerChoiceForSave()
|
||||
if speechServerChoice:
|
||||
self.prefsDict["speechServerInfo"] = \
|
||||
|
||||
@@ -870,6 +870,19 @@ SPEECH_DISPATCHER = _("Speech Dispatcher")
|
||||
# Translators: This label refers to the Piper neural text-to-speech system.
|
||||
# (https://github.com/rhasspy/piper)
|
||||
PIPER_TTS = _("Piper Neural TTS")
|
||||
# Translators: This label refers to external hardware serial speech synthesizers.
|
||||
HARDWARE_SPEECH = _("Hardware Speech Synthesizer")
|
||||
# Translators: This label refers to the LiteTalk hardware speech synthesizer.
|
||||
HARDWARE_LITETALK = _("LiteTalk")
|
||||
# Translators: This label refers to the DoubleTalk LT hardware speech synthesizer.
|
||||
HARDWARE_DOUBLETALK = _("DoubleTalk LT")
|
||||
# Translators: This label refers to the TripleTalk hardware speech synthesizer.
|
||||
HARDWARE_TRIPLETALK = _("TripleTalk")
|
||||
# Translators: This label refers to the Dectalk hardware synthesizer.
|
||||
HARDWARE_DECTALK = _("Dectalk")
|
||||
# Translators: This is the label for the combo box that lets the user choose
|
||||
# the serial device used by a hardware speech synthesizer.
|
||||
HARDWARE_SERIAL_DEVICE = _("Serial _device:")
|
||||
|
||||
# Translators: This is a label for a group of options related to Cthulhu's behavior
|
||||
# when presenting an application's spell check dialog.
|
||||
|
||||
@@ -0,0 +1,571 @@
|
||||
#!/usr/bin/env python3
|
||||
#
|
||||
# Copyright (c) 2024 Stormux
|
||||
#
|
||||
# This library is free software; you can redistribute it and/or
|
||||
# modify it under the terms of the GNU Lesser General Public
|
||||
# License as published by the Free Software Foundation; either
|
||||
# version 2.1 of the License, or (at your option) any later version.
|
||||
#
|
||||
# This library is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
||||
# Lesser General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Lesser General Public
|
||||
# License along with this library; if not, write to the
|
||||
# Free Software Foundation, Inc., Franklin Street, Fifth Floor,
|
||||
# Boston MA 02110-1301 USA.
|
||||
#
|
||||
# Cthulhu project: https://git.stormux.org/storm/cthulhu
|
||||
|
||||
"""Provides a Cthulhu speech server for hardware serial synthesizers.
|
||||
|
||||
Ports Fenrir's hardware serial drivers (LiteTalk/DoubleTalk/TripleTalk,
|
||||
Dectalk) to Cthulhu's SpeechServer interface.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import termios
|
||||
import threading
|
||||
import tty
|
||||
from queue import Empty, Queue
|
||||
|
||||
from . import debug
|
||||
from . import guilabels
|
||||
from . import messages
|
||||
from . import settings
|
||||
from . import speechserver
|
||||
from .acss import ACSS
|
||||
|
||||
|
||||
class _SpeakQueue(Queue):
|
||||
"""Queue with a clear() method."""
|
||||
|
||||
def clear(self):
|
||||
try:
|
||||
while True:
|
||||
self.get_nowait()
|
||||
except Empty:
|
||||
pass
|
||||
|
||||
|
||||
class _HardwareSerialDriver:
|
||||
"""Base class for hardware serial speech synthesizers.
|
||||
|
||||
Ported from Fenrir's hardwareSerialDriver.py.
|
||||
"""
|
||||
|
||||
cancel_command = b""
|
||||
default_baud_rate = 9600
|
||||
|
||||
def __init__(self, device: str, baud_rate: int):
|
||||
self.device = device
|
||||
self.baud_rate = baud_rate
|
||||
self.serial_port: int | None = None
|
||||
self.text_queue = _SpeakQueue()
|
||||
self.lock = threading.Lock()
|
||||
self.worker_thread: threading.Thread | None = None
|
||||
self._stop_worker = False
|
||||
self._is_initialized = False
|
||||
|
||||
def initialize(self) -> bool:
|
||||
self._open_serial_port()
|
||||
self._is_initialized = self.serial_port is not None
|
||||
if self._is_initialized:
|
||||
self._stop_worker = False
|
||||
self.worker_thread = threading.Thread(target=self._worker, daemon=True)
|
||||
self.worker_thread.start()
|
||||
return self._is_initialized
|
||||
|
||||
def shutdown(self) -> None:
|
||||
if not self._is_initialized:
|
||||
return
|
||||
self._stop_worker = True
|
||||
self.clear_buffer()
|
||||
self.text_queue.put(None)
|
||||
if self.worker_thread:
|
||||
self.worker_thread.join(timeout=0.5)
|
||||
self._close_serial_port()
|
||||
self._is_initialized = False
|
||||
|
||||
def speak(self, text: str, interrupt: bool = True) -> None:
|
||||
if not self._is_initialized:
|
||||
return
|
||||
if interrupt:
|
||||
self.stop()
|
||||
if not isinstance(text, str) or text == "":
|
||||
return
|
||||
self.text_queue.put(text)
|
||||
|
||||
def stop(self) -> None:
|
||||
if not self._is_initialized:
|
||||
return
|
||||
self.clear_buffer()
|
||||
if self.cancel_command:
|
||||
self._write_bytes(self.cancel_command, "cancel")
|
||||
|
||||
def clear_buffer(self) -> None:
|
||||
if not self._is_initialized:
|
||||
return
|
||||
self.text_queue.clear()
|
||||
|
||||
def set_rate(self, rate: float) -> None:
|
||||
if not self._is_initialized:
|
||||
return
|
||||
self._write_bytes(self._rate_command(rate), "rate")
|
||||
|
||||
def set_pitch(self, pitch: float) -> None:
|
||||
if not self._is_initialized:
|
||||
return
|
||||
self._write_bytes(self._pitch_command(pitch), "pitch")
|
||||
|
||||
def set_volume(self, volume: float) -> None:
|
||||
if not self._is_initialized:
|
||||
return
|
||||
self._write_bytes(self._volume_command(volume), "volume")
|
||||
|
||||
def _worker(self) -> None:
|
||||
while not self._stop_worker:
|
||||
text = self.text_queue.get()
|
||||
if text is None:
|
||||
return
|
||||
try:
|
||||
data = self._speak_bytes(text)
|
||||
self._write_bytes(data, "speech")
|
||||
except Exception as error:
|
||||
msg = f"HARDWARE SPEECH: worker failed: {error}"
|
||||
debug.printMessage(debug.LEVEL_ERROR, msg, True)
|
||||
|
||||
def _open_serial_port(self) -> None:
|
||||
if not self.device or self.device == "auto":
|
||||
msg = "HARDWARE SPEECH: requires an explicit serial device"
|
||||
debug.printMessage(debug.LEVEL_WARNING, msg, True)
|
||||
return
|
||||
|
||||
port = self._open_configured_serial_port(self.device)
|
||||
if port is not None:
|
||||
self._activate_serial_port(self.device, port)
|
||||
|
||||
def _open_configured_serial_port(self, device: str) -> int | None:
|
||||
port = None
|
||||
try:
|
||||
port = os.open(device, os.O_RDWR | os.O_NOCTTY)
|
||||
tty.setraw(port)
|
||||
attrs = termios.tcgetattr(port)
|
||||
attrs[2] |= termios.CLOCAL | termios.CREAD
|
||||
baud_rate = self._termios_baud_rate(self.baud_rate)
|
||||
attrs[4] = baud_rate
|
||||
attrs[5] = baud_rate
|
||||
attrs[6][termios.VMIN] = 0
|
||||
attrs[6][termios.VTIME] = 0
|
||||
attrs[0] &= ~(termios.IXON | termios.IXOFF | termios.IXANY)
|
||||
termios.tcsetattr(port, termios.TCSANOW, attrs)
|
||||
return port
|
||||
except (OSError, termios.error) as error:
|
||||
self._close_port(port)
|
||||
msg = f"HARDWARE SPEECH: device open failed: {device}: {error}"
|
||||
debug.printMessage(debug.LEVEL_WARNING, msg, True)
|
||||
return None
|
||||
|
||||
def _activate_serial_port(self, device: str, port: int) -> None:
|
||||
self.serial_port = port
|
||||
self.device = device
|
||||
msg = f"HARDWARE SPEECH: device opened: {device}, baud_rate={self.baud_rate}"
|
||||
debug.printMessage(debug.LEVEL_INFO, msg, True)
|
||||
|
||||
def _close_serial_port(self) -> None:
|
||||
with self.lock:
|
||||
if self.serial_port is None:
|
||||
return
|
||||
self._close_port(self.serial_port)
|
||||
self.serial_port = None
|
||||
|
||||
def _close_port(self, port: int | None) -> None:
|
||||
if port is None:
|
||||
return
|
||||
try:
|
||||
os.close(port)
|
||||
except OSError as error:
|
||||
msg = f"HARDWARE SPEECH: device close failed: {error}"
|
||||
debug.printMessage(debug.LEVEL_WARNING, msg, True)
|
||||
|
||||
def _write_bytes(self, data: bytes, description: str = "data") -> None:
|
||||
if not data:
|
||||
return
|
||||
with self.lock:
|
||||
if self.serial_port is None:
|
||||
return
|
||||
try:
|
||||
total_written = 0
|
||||
while total_written < len(data):
|
||||
bytes_written = os.write(self.serial_port, data[total_written:])
|
||||
if bytes_written == 0:
|
||||
raise OSError("serial write returned 0 bytes")
|
||||
total_written += bytes_written
|
||||
preview = self._format_bytes_preview(data)
|
||||
msg = f"HARDWARE SPEECH: wrote {total_written} {description} bytes: {preview}"
|
||||
debug.printMessage(debug.LEVEL_INFO, msg, True)
|
||||
except OSError as error:
|
||||
msg = f"HARDWARE SPEECH: write failed: {error}"
|
||||
debug.printMessage(debug.LEVEL_ERROR, msg, True)
|
||||
|
||||
def _termios_baud_rate(self, baud_rate: int) -> int:
|
||||
baud_name = f"B{baud_rate}"
|
||||
if hasattr(termios, baud_name):
|
||||
return getattr(termios, baud_name)
|
||||
msg = f"HARDWARE SPEECH: unsupported baud rate {baud_rate}; using 9600"
|
||||
debug.printMessage(debug.LEVEL_WARNING, msg, True)
|
||||
return termios.B9600
|
||||
|
||||
@staticmethod
|
||||
def _clean_text(text: str) -> str:
|
||||
text = text.replace("\r", " ").replace("\n", " ")
|
||||
return "".join(char if 0x20 <= ord(char) <= 0x7E else " " for char in text)
|
||||
|
||||
@staticmethod
|
||||
def _scale(value: float, minimum: int, maximum: int) -> int:
|
||||
value = max(0.0, min(1.0, value))
|
||||
return int(round(minimum + value * (maximum - minimum)))
|
||||
|
||||
@staticmethod
|
||||
def _format_bytes_preview(data: bytes, limit: int = 32) -> str:
|
||||
preview = data[:limit]
|
||||
hex_preview = " ".join(f"{byte:02x}" for byte in preview)
|
||||
ascii_preview = "".join(
|
||||
chr(byte) if 0x20 <= byte <= 0x7E else "." for byte in preview
|
||||
)
|
||||
suffix = "" if len(data) <= limit else " ..."
|
||||
return f"hex=[{hex_preview}{suffix}] ascii=[{ascii_preview}{suffix}]"
|
||||
|
||||
def _speak_bytes(self, text: str) -> bytes:
|
||||
raise NotImplementedError
|
||||
|
||||
def _rate_command(self, rate: float) -> bytes:
|
||||
return b""
|
||||
|
||||
def _pitch_command(self, pitch: float) -> bytes:
|
||||
return b""
|
||||
|
||||
def _volume_command(self, volume: float) -> bytes:
|
||||
return b""
|
||||
|
||||
|
||||
class _LiteTalkDriver(_HardwareSerialDriver):
|
||||
"""LiteTalk-compatible serial driver."""
|
||||
|
||||
cancel_command = b"\x18"
|
||||
|
||||
def _speak_bytes(self, text: str) -> bytes:
|
||||
return self._clean_text(text).encode("ascii", errors="replace") + b"\r"
|
||||
|
||||
def _rate_command(self, rate: float) -> bytes:
|
||||
return self._setting_command(self._scale(rate, 0, 9), b"S")
|
||||
|
||||
def _pitch_command(self, pitch: float) -> bytes:
|
||||
return self._setting_command(self._scale(pitch, 0, 99), b"P")
|
||||
|
||||
def _volume_command(self, volume: float) -> bytes:
|
||||
return self._setting_command(self._scale(volume, 0, 9), b"V")
|
||||
|
||||
@staticmethod
|
||||
def _setting_command(value: int, command: bytes) -> bytes:
|
||||
return b"\x01" + str(value).encode("ascii") + command
|
||||
|
||||
|
||||
class _DectalkDriver(_HardwareSerialDriver):
|
||||
"""Dectalk serial driver."""
|
||||
|
||||
cancel_command = b"\x18"
|
||||
|
||||
def _speak_bytes(self, text: str) -> bytes:
|
||||
return self._clean_text(text).encode("ascii", errors="replace") + b"\x01"
|
||||
|
||||
def _rate_command(self, rate: float) -> bytes:
|
||||
return self._setting_command("ra", self._scale(rate, 75, 650))
|
||||
|
||||
def _pitch_command(self, pitch: float) -> bytes:
|
||||
return self._setting_command("dv ap", self._scale(pitch, 50, 180))
|
||||
|
||||
def _volume_command(self, volume: float) -> bytes:
|
||||
return self._setting_command("vo", self._scale(volume, 0, 100))
|
||||
|
||||
@staticmethod
|
||||
def _setting_command(command: str, value: int) -> bytes:
|
||||
return f"[:{command} {value}]".encode("ascii")
|
||||
|
||||
|
||||
_DRIVER_MAP: dict[str, type[_HardwareSerialDriver]] = {
|
||||
"litetalk": _LiteTalkDriver,
|
||||
"doubletalk": _LiteTalkDriver,
|
||||
"tripletalk": _LiteTalkDriver,
|
||||
"dectalk": _DectalkDriver,
|
||||
}
|
||||
|
||||
_SYNTH_DISPLAY_NAMES = {
|
||||
"litetalk": guilabels.HARDWARE_LITETALK,
|
||||
"doubletalk": guilabels.HARDWARE_DOUBLETALK,
|
||||
"tripletalk": guilabels.HARDWARE_TRIPLETALK,
|
||||
"dectalk": guilabels.HARDWARE_DECTALK,
|
||||
}
|
||||
|
||||
|
||||
class SpeechServer(speechserver.SpeechServer):
|
||||
"""Hardware serial speech server implementation for Cthulhu."""
|
||||
|
||||
_active_servers: dict[str, SpeechServer] = {}
|
||||
|
||||
@staticmethod
|
||||
def getFactoryName() -> str:
|
||||
"""Returns a localized name describing this factory."""
|
||||
return guilabels.HARDWARE_SPEECH
|
||||
|
||||
@staticmethod
|
||||
def getSpeechServers() -> list[SpeechServer]:
|
||||
"""Gets available speech servers as a list."""
|
||||
return [
|
||||
SpeechServer(server_id, initialize=False, register=False)
|
||||
for server_id in _DRIVER_MAP
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def _getSpeechServer(cls, server_id: str) -> SpeechServer | None:
|
||||
"""Return an active server for the given id."""
|
||||
active_server = cls._active_servers.get(server_id)
|
||||
if active_server is not None:
|
||||
if active_server._matches_current_settings():
|
||||
return active_server
|
||||
active_server.shutdown()
|
||||
|
||||
cls(server_id)
|
||||
return cls._active_servers.get(server_id)
|
||||
|
||||
@staticmethod
|
||||
def getSpeechServer(info: list[str] | None = None) -> SpeechServer | None:
|
||||
"""Gets a given SpeechServer based upon the info."""
|
||||
if info and len(info) >= 2:
|
||||
server_id = info[1]
|
||||
else:
|
||||
server_id = "litetalk"
|
||||
return SpeechServer._getSpeechServer(server_id)
|
||||
|
||||
@staticmethod
|
||||
def shutdownActiveServers() -> None:
|
||||
"""Cleans up and shuts down this factory."""
|
||||
servers = list(SpeechServer._active_servers.values())
|
||||
for server in servers:
|
||||
server.shutdown()
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
server_id: str,
|
||||
initialize: bool = True,
|
||||
register: bool = True,
|
||||
):
|
||||
super().__init__()
|
||||
self._id = server_id
|
||||
self._driver: _HardwareSerialDriver | None = None
|
||||
self._info: list[str] = []
|
||||
self._device = ""
|
||||
self._baud_rate = settings.hardwareSpeechBaudRate
|
||||
|
||||
driver_class = _DRIVER_MAP.get(server_id)
|
||||
if driver_class is None:
|
||||
msg = f"HARDWARE SPEECH: unknown synth type: {server_id}"
|
||||
debug.printMessage(debug.LEVEL_WARNING, msg, True)
|
||||
return
|
||||
|
||||
display_name = _SYNTH_DISPLAY_NAMES.get(server_id, server_id)
|
||||
self._info = [display_name, server_id]
|
||||
|
||||
if not initialize:
|
||||
return
|
||||
|
||||
self._device = settings.hardwareSpeechDevice
|
||||
self._baud_rate = settings.hardwareSpeechBaudRate
|
||||
self._driver = driver_class(self._device, self._baud_rate)
|
||||
if self._driver.initialize():
|
||||
if register:
|
||||
SpeechServer._active_servers[server_id] = self
|
||||
msg = f"HARDWARE SPEECH: server initialized: {server_id} on {self._device}"
|
||||
debug.printMessage(debug.LEVEL_INFO, msg, True)
|
||||
else:
|
||||
msg = f"HARDWARE SPEECH: server initialization failed: {server_id}"
|
||||
debug.printMessage(debug.LEVEL_WARNING, msg, True)
|
||||
self._driver = None
|
||||
|
||||
def _matches_current_settings(self) -> bool:
|
||||
return (
|
||||
self._driver is not None
|
||||
and self._device == settings.hardwareSpeechDevice
|
||||
and self._baud_rate == settings.hardwareSpeechBaudRate
|
||||
)
|
||||
|
||||
def getInfo(self) -> list[str]:
|
||||
"""Returns [name, id]."""
|
||||
return self._info
|
||||
|
||||
def getVoiceFamilies(self) -> list[dict[str, str]]:
|
||||
"""Returns a list of VoiceFamily instances."""
|
||||
return []
|
||||
|
||||
def speakCharacter(self, character: str, acss: dict | None = None) -> None:
|
||||
"""Speaks a single character immediately."""
|
||||
if self._driver:
|
||||
self._apply_acss(acss)
|
||||
self._driver.speak(character, interrupt=True)
|
||||
|
||||
def speakKeyEvent(self, event, acss: dict | None = None) -> None:
|
||||
"""Speaks a key event immediately."""
|
||||
event_string = event.getKeyName()
|
||||
locking_state_string = event.getLockingStateString()
|
||||
text = f"{event_string} {locking_state_string}".strip()
|
||||
self.speak(text, acss=acss)
|
||||
|
||||
def speak(
|
||||
self,
|
||||
text: str | None = None,
|
||||
acss: dict | None = None,
|
||||
interrupt: bool = True,
|
||||
) -> None:
|
||||
"""Speaks all queued text immediately."""
|
||||
if not self._driver or text is None:
|
||||
return
|
||||
self._apply_acss(acss)
|
||||
self._driver.speak(text, interrupt=interrupt)
|
||||
|
||||
def sayAll(self, utteranceIterator, progressCallback) -> None:
|
||||
"""Iterates through the given utteranceIterator, speaking each utterance."""
|
||||
for context, acss in utteranceIterator:
|
||||
self.speak(context.utterance, acss=acss, interrupt=False)
|
||||
|
||||
def increaseSpeechRate(self, step: int = 5) -> None:
|
||||
self._change_default_speech_rate(step)
|
||||
|
||||
def decreaseSpeechRate(self, step: int = 5) -> None:
|
||||
self._change_default_speech_rate(step, decrease=True)
|
||||
|
||||
def increaseSpeechPitch(self, step: float = 0.5) -> None:
|
||||
self._change_default_speech_pitch(step)
|
||||
|
||||
def decreaseSpeechPitch(self, step: float = 0.5) -> None:
|
||||
self._change_default_speech_pitch(step, decrease=True)
|
||||
|
||||
def increaseSpeechVolume(self, step: float = 0.5) -> None:
|
||||
self._change_default_speech_volume(step)
|
||||
|
||||
def decreaseSpeechVolume(self, step: float = 0.5) -> None:
|
||||
self._change_default_speech_volume(step, decrease=True)
|
||||
|
||||
def updateCapitalizationStyle(self) -> None:
|
||||
pass
|
||||
|
||||
def updatePunctuationLevel(self) -> None:
|
||||
pass
|
||||
|
||||
def stop(self) -> None:
|
||||
if self._driver:
|
||||
self._driver.stop()
|
||||
|
||||
def shutdown(self) -> None:
|
||||
if self._driver:
|
||||
self._driver.shutdown()
|
||||
self._driver = None
|
||||
if self._id in SpeechServer._active_servers:
|
||||
del SpeechServer._active_servers[self._id]
|
||||
|
||||
def reset(self, text: str | None = None, acss: dict | None = None) -> None:
|
||||
if self._driver:
|
||||
self._driver.shutdown()
|
||||
self._driver = None
|
||||
|
||||
driver_class = _DRIVER_MAP.get(self._id)
|
||||
if driver_class is None:
|
||||
return
|
||||
|
||||
self._device = settings.hardwareSpeechDevice
|
||||
self._baud_rate = settings.hardwareSpeechBaudRate
|
||||
self._driver = driver_class(self._device, self._baud_rate)
|
||||
if not self._driver.initialize():
|
||||
self._driver = None
|
||||
|
||||
def _apply_acss(self, acss: dict | None) -> None:
|
||||
if not self._driver or not acss:
|
||||
return
|
||||
try:
|
||||
rate = acss.get(ACSS.RATE)
|
||||
if rate is not None:
|
||||
normalized = max(0.0, min(99.0, float(rate))) / 99.0
|
||||
self._driver.set_rate(normalized)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
pitch = acss.get(ACSS.AVERAGE_PITCH)
|
||||
if pitch is not None:
|
||||
normalized = max(0.0, min(9.0, float(pitch))) / 9.0
|
||||
self._driver.set_pitch(normalized)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
volume = acss.get(ACSS.GAIN)
|
||||
if volume is not None:
|
||||
normalized = max(0.0, min(9.0, float(volume))) / 9.0
|
||||
self._driver.set_volume(normalized)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _change_default_speech_rate(self, step: float, decrease: bool = False) -> None:
|
||||
acss = settings.voices[settings.DEFAULT_VOICE]
|
||||
delta = step * (-1 if decrease else 1)
|
||||
try:
|
||||
rate = acss[ACSS.RATE]
|
||||
except KeyError:
|
||||
rate = 50.0
|
||||
acss[ACSS.RATE] = max(0, min(99, rate + delta))
|
||||
msg = f"HARDWARE SPEECH: rate set to {acss[ACSS.RATE]}"
|
||||
debug.printMessage(debug.LEVEL_INFO, msg, True)
|
||||
if self._driver:
|
||||
normalized = acss[ACSS.RATE] / 99.0
|
||||
self._driver.set_rate(normalized)
|
||||
self.speak(
|
||||
messages.SPEECH_SLOWER if decrease else messages.SPEECH_FASTER,
|
||||
acss=acss
|
||||
)
|
||||
|
||||
def _change_default_speech_pitch(self, step: float, decrease: bool = False) -> None:
|
||||
acss = settings.voices[settings.DEFAULT_VOICE]
|
||||
delta = step * (-1 if decrease else 1)
|
||||
try:
|
||||
pitch = acss[ACSS.AVERAGE_PITCH]
|
||||
except KeyError:
|
||||
pitch = 5.0
|
||||
acss[ACSS.AVERAGE_PITCH] = max(0, min(9, pitch + delta))
|
||||
msg = f"HARDWARE SPEECH: pitch set to {acss[ACSS.AVERAGE_PITCH]}"
|
||||
debug.printMessage(debug.LEVEL_INFO, msg, True)
|
||||
if self._driver:
|
||||
normalized = acss[ACSS.AVERAGE_PITCH] / 9.0
|
||||
self._driver.set_pitch(normalized)
|
||||
self.speak(
|
||||
messages.SPEECH_LOWER if decrease else messages.SPEECH_HIGHER,
|
||||
acss=acss
|
||||
)
|
||||
|
||||
def _change_default_speech_volume(self, step: float, decrease: bool = False) -> None:
|
||||
acss = settings.voices[settings.DEFAULT_VOICE]
|
||||
delta = step * (-1 if decrease else 1)
|
||||
try:
|
||||
volume = acss[ACSS.GAIN]
|
||||
except KeyError:
|
||||
volume = 10.0
|
||||
acss[ACSS.GAIN] = max(0, min(9, volume + delta))
|
||||
msg = f"HARDWARE SPEECH: volume set to {acss[ACSS.GAIN]}"
|
||||
debug.printMessage(debug.LEVEL_INFO, msg, True)
|
||||
if self._driver:
|
||||
normalized = acss[ACSS.GAIN] / 9.0
|
||||
self._driver.set_volume(normalized)
|
||||
self.speak(
|
||||
messages.SPEECH_SOFTER if decrease else messages.SPEECH_LOUDER,
|
||||
acss=acss
|
||||
)
|
||||
@@ -101,6 +101,7 @@ cthulhu_python_sources = files([
|
||||
'speech.py',
|
||||
'spellcheck.py',
|
||||
'speechdispatcherfactory.py',
|
||||
'hardwarefactory.py',
|
||||
'speech_generator.py',
|
||||
'speechserver.py',
|
||||
'piperfactory.py',
|
||||
|
||||
@@ -300,19 +300,21 @@ class SpeechServer(speechserver.SpeechServer):
|
||||
return voiceInfo.sampleRate if voiceInfo else None
|
||||
|
||||
def _mapRate(self, acssRate):
|
||||
"""Map ACSS rate (0-99) to Piper length_scale.
|
||||
"""Map ACSS rate (0-100) to Piper length_scale.
|
||||
|
||||
ACSS rate 50 (default) = length_scale 1.0
|
||||
Higher ACSS rate = lower length_scale (faster)
|
||||
Lower ACSS rate = higher length_scale (slower)
|
||||
|
||||
Arguments:
|
||||
- acssRate: Rate value from 0-99
|
||||
- acssRate: Rate value from 0-100
|
||||
"""
|
||||
rate = acssRate if acssRate is not None else 50
|
||||
rate = max(0, min(99, rate))
|
||||
lengthScale = 2.0 - (rate / 99.0) * 1.5
|
||||
return max(0.5, min(2.0, lengthScale))
|
||||
rate = max(0.0, min(100.0, float(rate)))
|
||||
if rate <= 50.0:
|
||||
return 2.0 - (rate / 50.0)
|
||||
|
||||
return 1.0 - ((rate - 50.0) / 50.0) * 0.75
|
||||
|
||||
def _mapPitch(self, acssPitch):
|
||||
"""Map ACSS pitch (0-9) to pitch adjustment factor.
|
||||
@@ -614,7 +616,7 @@ class SpeechServer(speechserver.SpeechServer):
|
||||
rate = acss[ACSS.RATE]
|
||||
except KeyError:
|
||||
rate = 50
|
||||
acss[ACSS.RATE] = max(0, min(99, rate + delta))
|
||||
acss[ACSS.RATE] = max(0, min(100, rate + delta))
|
||||
msg = f"PIPER: Rate set to {acss[ACSS.RATE]}"
|
||||
debug.printMessage(debug.LEVEL_INFO, msg, True)
|
||||
self.speak(
|
||||
|
||||
@@ -24,6 +24,7 @@ import logging
|
||||
|
||||
from gi.repository import GLib
|
||||
|
||||
from cthulhu.ax_object import AXObject
|
||||
from cthulhu import debug
|
||||
from cthulhu import dbus_service
|
||||
from cthulhu.plugin import Plugin, cthulhu_hookimpl
|
||||
@@ -53,6 +54,9 @@ class WindowTitleReader(Plugin):
|
||||
self._enabled = False
|
||||
self._pollSourceId = None
|
||||
self._pollIntervalMs = 100
|
||||
self._fallbackDelayMs = 250
|
||||
self._pendingFallbackSourceId = None
|
||||
self._lastActiveWindowId = None
|
||||
self._lastTitle = None
|
||||
self._display = None
|
||||
self._root = None
|
||||
@@ -71,6 +75,13 @@ class WindowTitleReader(Plugin):
|
||||
return True
|
||||
|
||||
self._register_keybinding()
|
||||
self.app.getDynamicApiManager().registerAPI(
|
||||
"WindowTitleReader",
|
||||
self,
|
||||
overwrite=True,
|
||||
)
|
||||
if xlibAvailable:
|
||||
self._start_tracking()
|
||||
self._activated = True
|
||||
debug.printMessage(debug.LEVEL_INFO, "WindowTitleReader: Activated", True)
|
||||
return True
|
||||
@@ -81,6 +92,8 @@ class WindowTitleReader(Plugin):
|
||||
return
|
||||
|
||||
self._stop_tracking()
|
||||
if self.app:
|
||||
self.app.getDynamicApiManager().unregisterAPI("WindowTitleReader")
|
||||
self._activated = False
|
||||
debug.printMessage(debug.LEVEL_INFO, "WindowTitleReader: Deactivated", True)
|
||||
return True
|
||||
@@ -118,7 +131,8 @@ class WindowTitleReader(Plugin):
|
||||
|
||||
def _toggle_tracking(self, script=None, inputEvent=None):
|
||||
if self._enabled:
|
||||
self._stop_tracking()
|
||||
self._enabled = False
|
||||
self._lastTitle = None
|
||||
self._present_message("Window title reader off")
|
||||
return True
|
||||
|
||||
@@ -132,6 +146,8 @@ class WindowTitleReader(Plugin):
|
||||
return True
|
||||
|
||||
if self._start_tracking():
|
||||
self._enabled = True
|
||||
self._poll_window_title()
|
||||
self._present_message("Window title reader on")
|
||||
else:
|
||||
self._present_message("Window title reader unavailable")
|
||||
@@ -162,6 +178,8 @@ class WindowTitleReader(Plugin):
|
||||
return False
|
||||
|
||||
if self._start_tracking():
|
||||
self._enabled = True
|
||||
self._poll_window_title()
|
||||
if notify_user:
|
||||
self._present_message("Window title reader on")
|
||||
return True
|
||||
@@ -171,7 +189,8 @@ class WindowTitleReader(Plugin):
|
||||
return False
|
||||
|
||||
if self._enabled:
|
||||
self._stop_tracking()
|
||||
self._enabled = False
|
||||
self._lastTitle = None
|
||||
|
||||
if notify_user:
|
||||
self._present_message("Window title reader off")
|
||||
@@ -195,7 +214,7 @@ class WindowTitleReader(Plugin):
|
||||
pluginLogger.exception("WindowTitleReader: Failed to present message")
|
||||
|
||||
def _start_tracking(self):
|
||||
if self._enabled:
|
||||
if self._pollSourceId is not None:
|
||||
return True
|
||||
|
||||
try:
|
||||
@@ -203,7 +222,6 @@ class WindowTitleReader(Plugin):
|
||||
self._root = self._display.screen().root
|
||||
self._init_atoms()
|
||||
self._pollSourceId = GLib.timeout_add(self._pollIntervalMs, self._poll_window_title)
|
||||
self._enabled = True
|
||||
self._poll_window_title()
|
||||
return True
|
||||
except Exception as error:
|
||||
@@ -217,11 +235,16 @@ class WindowTitleReader(Plugin):
|
||||
return False
|
||||
|
||||
def _stop_tracking(self):
|
||||
if self._pendingFallbackSourceId is not None:
|
||||
GLib.source_remove(self._pendingFallbackSourceId)
|
||||
self._pendingFallbackSourceId = None
|
||||
|
||||
if self._pollSourceId is not None:
|
||||
GLib.source_remove(self._pollSourceId)
|
||||
self._pollSourceId = None
|
||||
|
||||
self._enabled = False
|
||||
self._lastActiveWindowId = None
|
||||
self._lastTitle = None
|
||||
self._cleanup_display()
|
||||
|
||||
@@ -246,7 +269,7 @@ class WindowTitleReader(Plugin):
|
||||
}
|
||||
|
||||
def _poll_window_title(self):
|
||||
if not self._enabled:
|
||||
if self._pollSourceId is None:
|
||||
return False
|
||||
|
||||
try:
|
||||
@@ -260,7 +283,14 @@ class WindowTitleReader(Plugin):
|
||||
self._lastTitle = None
|
||||
return True
|
||||
|
||||
if windowTitle != self._lastTitle:
|
||||
activeWindowId = activeWindow.id
|
||||
if self._lastActiveWindowId is None:
|
||||
self._lastActiveWindowId = activeWindowId
|
||||
elif activeWindowId != self._lastActiveWindowId:
|
||||
self._lastActiveWindowId = activeWindowId
|
||||
self._schedule_fallback_title()
|
||||
|
||||
if self._enabled and windowTitle != self._lastTitle:
|
||||
self._present_title(windowTitle)
|
||||
self._lastTitle = windowTitle
|
||||
except Exception as error:
|
||||
@@ -269,6 +299,70 @@ class WindowTitleReader(Plugin):
|
||||
|
||||
return True
|
||||
|
||||
def _schedule_fallback_title(self):
|
||||
if self._enabled:
|
||||
return
|
||||
|
||||
if self._pendingFallbackSourceId is not None:
|
||||
GLib.source_remove(self._pendingFallbackSourceId)
|
||||
|
||||
self._pendingFallbackSourceId = GLib.timeout_add(
|
||||
self._fallbackDelayMs,
|
||||
self._present_pending_fallback_title,
|
||||
)
|
||||
|
||||
def _present_pending_fallback_title(self):
|
||||
self._pendingFallbackSourceId = None
|
||||
titleText = self.get_fallback_title()
|
||||
if titleText:
|
||||
self._present_title(titleText)
|
||||
return False
|
||||
|
||||
def get_fallback_title(self, atspiTitle=None):
|
||||
"""Returns the X11 title when AT-SPI has not exposed an equivalent title."""
|
||||
|
||||
if not xlibAvailable and self._display is None:
|
||||
return ""
|
||||
|
||||
startedTracking = self._pollSourceId is not None
|
||||
if not startedTracking and not self._start_tracking():
|
||||
return ""
|
||||
|
||||
activeWindow = self._get_active_window()
|
||||
titleText = self._get_current_title(activeWindow) if activeWindow else ""
|
||||
if not titleText:
|
||||
return ""
|
||||
|
||||
if atspiTitle is None:
|
||||
atspiTitle = self._get_atspi_title()
|
||||
|
||||
if self._titles_match(atspiTitle, titleText):
|
||||
return ""
|
||||
|
||||
return titleText
|
||||
|
||||
def _get_atspi_title(self):
|
||||
if not self.app:
|
||||
return ""
|
||||
|
||||
appState = self.app.getDynamicApiManager().getAPI("CthulhuState")
|
||||
if not appState:
|
||||
return ""
|
||||
|
||||
return AXObject.get_name(appState.activeWindow) or ""
|
||||
|
||||
def _titles_match(self, atspiTitle, fallbackTitle):
|
||||
if not atspiTitle or not fallbackTitle:
|
||||
return False
|
||||
|
||||
if self._is_wine_desktop_title(atspiTitle):
|
||||
return False
|
||||
|
||||
normalizedAtspiTitle = " ".join(atspiTitle.casefold().split())
|
||||
normalizedFallbackTitle = " ".join(fallbackTitle.casefold().split())
|
||||
return normalizedAtspiTitle in normalizedFallbackTitle \
|
||||
or normalizedFallbackTitle in normalizedAtspiTitle
|
||||
|
||||
def _present_title(self, titleText):
|
||||
if not self.app:
|
||||
return
|
||||
|
||||
@@ -26,6 +26,7 @@ import select
|
||||
import logging
|
||||
import threading
|
||||
from threading import Thread, Lock
|
||||
from typing import Optional
|
||||
from cthulhu.plugin import Plugin, cthulhu_hookimpl
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -34,6 +35,13 @@ logger = logging.getLogger(__name__)
|
||||
APPEND_CODE = '<#APPEND#>'
|
||||
PERSISTENT_CODE = '<#PERSISTENT#>'
|
||||
|
||||
def _get_socket_file() -> Optional[str]:
|
||||
runtimeDir = os.environ.get('XDG_RUNTIME_DIR')
|
||||
if not runtimeDir:
|
||||
return None
|
||||
|
||||
return os.path.join(runtimeDir, 'cthulhu.sock')
|
||||
|
||||
class SelfVoice(Plugin):
|
||||
"""Plugin that provides a socket interface for external applications to send text to Cthulhu."""
|
||||
|
||||
@@ -115,10 +123,11 @@ class SelfVoice(Plugin):
|
||||
|
||||
def voiceWorker(self):
|
||||
"""Worker thread that listens on a socket for messages to speak."""
|
||||
socketFile = '/tmp/cthulhu.sock'
|
||||
# For testing purposes
|
||||
# socketFile = '/tmp/cthulhu-plugin.sock'
|
||||
|
||||
socketFile = _get_socket_file()
|
||||
if not socketFile:
|
||||
logger.error("XDG_RUNTIME_DIR is not set; Self Voice plugin cannot create its socket")
|
||||
return
|
||||
|
||||
# Clean up any existing socket file
|
||||
if os.path.exists(socketFile):
|
||||
try:
|
||||
|
||||
@@ -393,8 +393,6 @@ class Script(script.Script):
|
||||
return keyBindings
|
||||
|
||||
def getExtensionBindings(self):
|
||||
with open('/tmp/extension_bindings_debug.log', 'a') as f:
|
||||
f.write(f"=== getExtensionBindings() called ===\n")
|
||||
keyBindings = keybindings.KeyBindings()
|
||||
|
||||
bindings = self.notificationPresenter.get_bindings()
|
||||
@@ -443,35 +441,13 @@ class Script(script.Script):
|
||||
try:
|
||||
if hasattr(cthulhu, 'cthulhuApp') and cthulhu.cthulhuApp:
|
||||
api_helper = cthulhu.cthulhuApp.getAPIHelper()
|
||||
with open('/tmp/extension_bindings_debug.log', 'a') as f:
|
||||
f.write(f"=== Checking for plugin bindings ===\n")
|
||||
f.write(f"api_helper exists: {api_helper is not None}\n")
|
||||
if api_helper:
|
||||
f.write(f"api_helper has _gestureBindings: {hasattr(api_helper, '_gestureBindings')}\n")
|
||||
if hasattr(api_helper, '_gestureBindings'):
|
||||
f.write(f"_gestureBindings content: {api_helper._gestureBindings}\n")
|
||||
f.write(f"Available contexts: {list(api_helper._gestureBindings.keys())}\n")
|
||||
|
||||
if api_helper and hasattr(api_helper, '_gestureBindings'):
|
||||
with open('/tmp/extension_bindings_debug.log', 'a') as f:
|
||||
f.write(f"=== Adding plugin bindings in getExtensionBindings() ===\n")
|
||||
|
||||
for context_name, context_bindings in api_helper._gestureBindings.items():
|
||||
for binding in context_bindings:
|
||||
keyBindings.add(binding)
|
||||
with open('/tmp/extension_bindings_debug.log', 'a') as f:
|
||||
f.write(f"Added plugin binding: {binding.keysymstring} modifiers={binding.modifiers} desc={binding.handler.description}\n")
|
||||
else:
|
||||
with open('/tmp/extension_bindings_debug.log', 'a') as f:
|
||||
f.write(f"=== No plugin bindings available ===\n")
|
||||
else:
|
||||
with open('/tmp/extension_bindings_debug.log', 'a') as f:
|
||||
f.write(f"=== cthulhuApp not available ===\n")
|
||||
except Exception as e:
|
||||
import cthulhu.debug as debug
|
||||
debug.printMessage(debug.LEVEL_WARNING, f"Failed to add plugin bindings: {e}", True)
|
||||
with open('/tmp/extension_bindings_debug.log', 'a') as f:
|
||||
f.write(f"Exception in plugin binding addition: {e}\n")
|
||||
|
||||
return keyBindings
|
||||
|
||||
|
||||
@@ -42,6 +42,8 @@ userCustomizableSettings = [
|
||||
"onlySpeakDisplayedText",
|
||||
"speechServerFactory",
|
||||
"speechServerInfo",
|
||||
"hardwareSpeechDevice",
|
||||
"hardwareSpeechBaudRate",
|
||||
"voices",
|
||||
"speechVerbosityLevel",
|
||||
"readFullRowInGUITable",
|
||||
@@ -265,9 +267,11 @@ activeProfile = ['Default', 'default']
|
||||
profile = ['Default', 'default']
|
||||
|
||||
# Speech
|
||||
speechFactoryModules = ["speechdispatcherfactory", "piperfactory"]
|
||||
speechFactoryModules = ["speechdispatcherfactory", "piperfactory", "hardwarefactory"]
|
||||
speechServerFactory = "speechdispatcherfactory"
|
||||
speechServerInfo = None # None means let the factory decide.
|
||||
hardwareSpeechDevice = ""
|
||||
hardwareSpeechBaudRate = 9600
|
||||
enableSpeech = True
|
||||
silenceSpeech = False
|
||||
enableTutorialMessages = False
|
||||
|
||||
@@ -33,6 +33,7 @@ __copyright__ = "Copyright (c) 2005-2008 Sun Microsystems Inc." \
|
||||
__license__ = "LGPL"
|
||||
|
||||
from . import cmdnames
|
||||
from . import cthulhu
|
||||
from . import debug
|
||||
from . import input_event
|
||||
from . import keybindings
|
||||
@@ -319,10 +320,33 @@ class WhereAmIPresenter:
|
||||
return True
|
||||
|
||||
title = script.speechGenerator.generateTitle(obj)
|
||||
fallbackTitle = self._get_fallback_title(title)
|
||||
if fallbackTitle:
|
||||
script.presentMessage(fallbackTitle)
|
||||
return True
|
||||
|
||||
for (string, voice) in title:
|
||||
script.presentMessage(string, voice=voice)
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _get_fallback_title(generatedTitle):
|
||||
"""Returns a non-AT-SPI title when the active title reader can provide one."""
|
||||
|
||||
try:
|
||||
reader = cthulhu.getManager().getDynamicApiManager().getAPI("WindowTitleReader")
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
if reader is None:
|
||||
return ""
|
||||
|
||||
atspiTitle = " ".join(
|
||||
str(item[0]) for item in generatedTitle
|
||||
if isinstance(item, (list, tuple)) and item
|
||||
)
|
||||
return reader.get_fallback_title(atspiTitle)
|
||||
|
||||
def _present_default_button(self, script, event=None, dialog=None, error_messages=True):
|
||||
"""Presents the default button of the current dialog."""
|
||||
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
import os
|
||||
import select
|
||||
import sys
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
|
||||
from cthulhu import hardwarefactory
|
||||
from cthulhu import settings
|
||||
|
||||
|
||||
def read_available(fd, expectedLength, timeout=1.0):
|
||||
deadline = time.monotonic() + timeout
|
||||
data = b""
|
||||
while len(data) < expectedLength and time.monotonic() < deadline:
|
||||
readable, _, _ = select.select([fd], [], [], 0.05)
|
||||
if readable:
|
||||
data += os.read(fd, 1024)
|
||||
return data
|
||||
|
||||
|
||||
class HardwareFactoryRegressionTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._oldDevice = settings.hardwareSpeechDevice
|
||||
self._oldBaudRate = settings.hardwareSpeechBaudRate
|
||||
hardwarefactory.SpeechServer.shutdownActiveServers()
|
||||
|
||||
def tearDown(self):
|
||||
hardwarefactory.SpeechServer.shutdownActiveServers()
|
||||
settings.hardwareSpeechDevice = self._oldDevice
|
||||
settings.hardwareSpeechBaudRate = self._oldBaudRate
|
||||
|
||||
def test_lists_explicit_synth_choices_without_opening_serial_device(self):
|
||||
settings.hardwareSpeechDevice = ""
|
||||
|
||||
servers = hardwarefactory.SpeechServer.getSpeechServers()
|
||||
|
||||
self.assertEqual(
|
||||
["litetalk", "doubletalk", "tripletalk", "dectalk"],
|
||||
[server.getInfo()[1] for server in servers],
|
||||
)
|
||||
self.assertEqual({}, hardwarefactory.SpeechServer._active_servers)
|
||||
self.assertTrue(all(server._driver is None for server in servers))
|
||||
|
||||
def test_failed_initialization_is_not_cached(self):
|
||||
settings.hardwareSpeechDevice = ""
|
||||
|
||||
self.assertIsNone(
|
||||
hardwarefactory.SpeechServer.getSpeechServer(["LiteTalk", "litetalk"])
|
||||
)
|
||||
self.assertEqual({}, hardwarefactory.SpeechServer._active_servers)
|
||||
|
||||
masterFd, slaveFd = os.openpty()
|
||||
try:
|
||||
settings.hardwareSpeechDevice = os.ttyname(slaveFd)
|
||||
server = hardwarefactory.SpeechServer.getSpeechServer(
|
||||
["LiteTalk", "litetalk"]
|
||||
)
|
||||
|
||||
self.assertIsNotNone(server)
|
||||
self.assertIsNotNone(server._driver)
|
||||
self.assertIs(
|
||||
server,
|
||||
hardwarefactory.SpeechServer._active_servers.get("litetalk"),
|
||||
)
|
||||
finally:
|
||||
os.close(masterFd)
|
||||
os.close(slaveFd)
|
||||
|
||||
def test_explicit_synth_choices_write_expected_serial_bytes(self):
|
||||
expectedBytes = {
|
||||
"litetalk": b"Alias\r",
|
||||
"doubletalk": b"Alias\r",
|
||||
"tripletalk": b"Alias\r",
|
||||
"dectalk": b"Alias\x01",
|
||||
}
|
||||
|
||||
for synthId, expected in expectedBytes.items():
|
||||
with self.subTest(synthId=synthId):
|
||||
hardwarefactory.SpeechServer.shutdownActiveServers()
|
||||
masterFd, slaveFd = os.openpty()
|
||||
try:
|
||||
settings.hardwareSpeechDevice = os.ttyname(slaveFd)
|
||||
server = hardwarefactory.SpeechServer.getSpeechServer(
|
||||
["", synthId]
|
||||
)
|
||||
|
||||
self.assertIsNotNone(server)
|
||||
server.speak("Alias", interrupt=False)
|
||||
self.assertEqual(expected, read_available(masterFd, len(expected)))
|
||||
finally:
|
||||
hardwarefactory.SpeechServer.shutdownActiveServers()
|
||||
os.close(masterFd)
|
||||
os.close(slaveFd)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,30 @@
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
|
||||
from cthulhu import piperfactory
|
||||
|
||||
|
||||
class PiperFactoryRateMappingTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.server = piperfactory.SpeechServer.__new__(piperfactory.SpeechServer)
|
||||
|
||||
def test_default_rate_maps_to_native_piper_speed(self):
|
||||
self.assertEqual(1.0, self.server._mapRate(50))
|
||||
|
||||
def test_rate_scale_uses_full_cthulhu_range(self):
|
||||
self.assertEqual(2.0, self.server._mapRate(0))
|
||||
self.assertEqual(0.25, self.server._mapRate(100))
|
||||
|
||||
def test_high_screen_reader_rate_is_substantially_faster(self):
|
||||
self.assertAlmostEqual(0.415, self.server._mapRate(89), places=3)
|
||||
|
||||
def test_rate_values_are_clamped(self):
|
||||
self.assertEqual(2.0, self.server._mapRate(-1))
|
||||
self.assertEqual(0.25, self.server._mapRate(101))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,19 @@
|
||||
import os
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from cthulhu.plugins.self_voice import plugin
|
||||
|
||||
|
||||
class SelfVoicePluginRegressionTests(unittest.TestCase):
|
||||
def test_socket_file_is_scoped_to_user_runtime_directory(self) -> None:
|
||||
with mock.patch.dict(os.environ, {"XDG_RUNTIME_DIR": "/run/user/1000"}, clear=True):
|
||||
self.assertEqual(plugin._get_socket_file(), "/run/user/1000/cthulhu.sock")
|
||||
|
||||
def test_socket_file_is_unavailable_without_user_runtime_directory(self) -> None:
|
||||
with mock.patch.dict(os.environ, {}, clear=True):
|
||||
self.assertIsNone(plugin._get_socket_file())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,97 @@
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
|
||||
from cthulhu import where_am_i_presenter
|
||||
from cthulhu.plugins.WindowTitleReader.plugin import WindowTitleReader
|
||||
|
||||
|
||||
class WindowTitleFallbackRegressionTests(unittest.TestCase):
|
||||
def test_activate_registers_reader_api_and_starts_fallback_tracking(self):
|
||||
plugin = WindowTitleReader()
|
||||
plugin.app = mock.Mock()
|
||||
|
||||
with (
|
||||
mock.patch.object(plugin, "_register_keybinding"),
|
||||
mock.patch.object(plugin, "_start_tracking") as startTracking,
|
||||
):
|
||||
plugin.activate(plugin)
|
||||
|
||||
plugin.app.getDynamicApiManager.return_value.registerAPI.assert_called_once_with(
|
||||
"WindowTitleReader",
|
||||
plugin,
|
||||
overwrite=True,
|
||||
)
|
||||
startTracking.assert_called_once_with()
|
||||
|
||||
def test_poll_schedules_fallback_after_active_window_changes(self):
|
||||
plugin = WindowTitleReader()
|
||||
plugin._pollSourceId = 1
|
||||
plugin._lastActiveWindowId = 100
|
||||
activeWindow = mock.Mock(id=200)
|
||||
|
||||
with (
|
||||
mock.patch.object(plugin, "_get_active_window", return_value=activeWindow),
|
||||
mock.patch.object(plugin, "_get_current_title", return_value="XTerm"),
|
||||
mock.patch.object(plugin, "_schedule_fallback_title") as scheduleFallback,
|
||||
):
|
||||
self.assertTrue(plugin._poll_window_title())
|
||||
|
||||
scheduleFallback.assert_called_once_with()
|
||||
|
||||
def test_poll_keeps_previous_window_across_transient_missing_active_window(self):
|
||||
plugin = WindowTitleReader()
|
||||
plugin._pollSourceId = 1
|
||||
plugin._lastActiveWindowId = 100
|
||||
|
||||
with mock.patch.object(plugin, "_get_active_window", return_value=None):
|
||||
self.assertTrue(plugin._poll_window_title())
|
||||
|
||||
self.assertEqual(plugin._lastActiveWindowId, 100)
|
||||
|
||||
def test_fallback_is_empty_when_atspi_exposes_same_title(self):
|
||||
plugin = WindowTitleReader()
|
||||
plugin._pollSourceId = 1
|
||||
activeWindow = mock.Mock()
|
||||
|
||||
with (
|
||||
mock.patch.object(plugin, "_get_active_window", return_value=activeWindow),
|
||||
mock.patch.object(plugin, "_get_current_title", return_value="Terminal"),
|
||||
):
|
||||
self.assertEqual(plugin.get_fallback_title("Terminal"), "")
|
||||
|
||||
def test_fallback_replaces_wine_desktop_title(self):
|
||||
plugin = WindowTitleReader()
|
||||
plugin._pollSourceId = 1
|
||||
activeWindow = mock.Mock()
|
||||
|
||||
with (
|
||||
mock.patch.object(plugin, "_get_active_window", return_value=activeWindow),
|
||||
mock.patch.object(plugin, "_get_current_title", return_value="Game Window"),
|
||||
):
|
||||
self.assertEqual(plugin.get_fallback_title("Wine Desktop"), "Game Window")
|
||||
|
||||
def test_present_title_uses_fallback_instead_of_atspi_title(self):
|
||||
presenter = where_am_i_presenter.WhereAmIPresenter()
|
||||
script = mock.Mock()
|
||||
script.speechGenerator.generateTitle.return_value = [("Wine Desktop", None)]
|
||||
|
||||
with (
|
||||
mock.patch.object(where_am_i_presenter.cthulhu_state, "locusOfFocus", object()),
|
||||
mock.patch.object(where_am_i_presenter.AXObject, "is_dead", return_value=False),
|
||||
mock.patch.object(
|
||||
presenter,
|
||||
"_get_fallback_title",
|
||||
return_value="Game Window",
|
||||
),
|
||||
):
|
||||
self.assertTrue(presenter.present_title(script))
|
||||
|
||||
script.presentMessage.assert_called_once_with("Game Window")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user