4 Commits

17 changed files with 1209 additions and 57 deletions
+60 -10
View File
@@ -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`.
+6 -6
View File
@@ -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"
```
+29
View File
@@ -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>
+133
View File
@@ -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"] = \
+13
View File
@@ -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.
+571
View File
@@ -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
)
+1
View File
@@ -101,6 +101,7 @@ cthulhu_python_sources = files([
'speech.py',
'spellcheck.py',
'speechdispatcherfactory.py',
'hardwarefactory.py',
'speech_generator.py',
'speechserver.py',
'piperfactory.py',
+8 -6
View File
@@ -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(
+100 -6
View File
@@ -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
+13 -4
View File
@@ -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:
-24
View File
@@ -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
+5 -1
View File
@@ -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
+24
View File
@@ -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."""
+100
View File
@@ -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()
+30
View File
@@ -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()