storm/fenrir
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Release candidate.

Browse Source
This commit is contained in:
Storm Dragon
2026-07-29 19:12:49 -04:00
parent e180fd53fa 122a6a38b3
commit 312cb46f22
29 changed files with 1269 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+18 -8
View File
@@ -81,13 +81,16 @@ Fenrir is a Linux screen reader. Linux is the only officially supported platform
- espeak or espeak-ng (or any TTS command)
3. **dectalkDriver** - Serial DECtalk-compatible hardware speech driver
- RPITalk gadget mode or a DECtalk-compatible serial device
4. **litetalkDriver** - Serial LiteTalk-compatible hardware speech driver
4. **dectalkUsbDriver** - Native Access Solutions DECtalk USB driver
- USB ID `0dd0:2001`
- PyUSB and a libusb backend
5. **litetalkDriver** - Serial LiteTalk-compatible hardware speech driver
- RPITalk gadget mode or a LiteTalk-compatible serial device
5. **doubletalkDriver** - Serial DoubleTalk LT-compatible hardware speech driver
6. **doubletalkDriver** - Serial DoubleTalk LT-compatible hardware speech driver
- DoubleTalk LT; does not support the internal DoubleTalk PC card
6. **tripletalkDriver** - Serial TripleTalk-compatible hardware speech driver
7. **tripletalkDriver** - Serial TripleTalk-compatible hardware speech driver
- External DB9 serial TripleTalk devices, or USB models that expose a tty serial device
7. **debugDriver** - Debug speech driver for testing
8. **debugDriver** - Debug speech driver for testing
- No dependencies
@@ -459,15 +462,22 @@ setting <action> [parameters]
- `speech#volume=0.1-1.0` - Speech volume
- `speech#voice=voice_name` - Voice selection (e.g., "en-us+f3")
- `speech#module=module_name` - TTS module (e.g., "espeak-ng")
- `speech#driver=driver_name` - Speech driver (speechdDriver/genericDriver/dectalkDriver/litetalkDriver/doubletalkDriver/tripletalkDriver)
- `speech#driver=driver_name` - Speech driver (speechdDriver/genericDriver/dectalkDriver/dectalkUsbDriver/litetalkDriver/doubletalkDriver/tripletalkDriver)
- `speech#hardware_device=/dev/ttyS0` - Hardware synth serial device for dectalkDriver/litetalkDriver
- `speech#hardware_baud_rate=9600` - Hardware synth serial baud rate
- `speech#history_size=50` - Number of spoken items kept in runtime speech history
The native Access Solutions DECtalk USB, USB ID `0dd0:2001`, uses
`dectalkUsbDriver`. It is autodetected and does not use
`speech#hardware_device` or `speech#hardware_baud_rate`. Install PyUSB and a
libusb backend first. From a source checkout, use
`python3 -m pip install '.[dectalk_usb]'`; distributions may package PyUSB as
`python3-usb` or `python-pyusb`. Root-run Fenrir can access the device directly.
A non-root Fenrir process needs a local udev rule granting its account
read/write access to `0dd0:2001`; do not make the device world-writable. Other
USB hardware synths are supported only when Linux exposes them as a serial tty
such as `/dev/ttyACM0` or `/dev/ttyUSB0`. A USB-only TripleTalk with no tty
device would require a separate USB protocol driver. Use an explicit
`speech#hardware_device` path for hardware speech.
such as `/dev/ttyACM0` or `/dev/ttyUSB0`. Use an explicit
`speech#hardware_device` path for serial hardware speech.
- `speech#auto_read_incoming=True/False` - Auto-read new text
*Sound Settings:*
config/settings/settings.conf
+4 -1
View File
@@ -36,10 +36,12 @@ progress_monitoring=True
enabled=True
# Select speech driver, options are speechdDriver, genericDriver,
# dectalkDriver, litetalkDriver, doubletalkDriver, or tripletalkDriver:
# dectalkDriver, dectalkUsbDriver, litetalkDriver, doubletalkDriver,
# or tripletalkDriver:
driver=speechdDriver
#driver=genericDriver
#driver=dectalkDriver
#driver=dectalkUsbDriver
#driver=litetalkDriver
#driver=doubletalkDriver
#driver=tripletalkDriver
@@ -77,6 +79,7 @@ volume=1.0
# Hardware speech synthesizer serial device.
# Used by dectalkDriver, litetalkDriver, doubletalkDriver, and tripletalkDriver.
# dectalkUsbDriver autodetects the Access Solutions device and ignores this.
# USB serial devices are supported if Linux exposes them as /dev/ttyACM*
# or /dev/ttyUSB*. USB-only synths with no tty device need a separate driver.
# Set an explicit device for hardware speech.
docs/fenrir.1
+2
View File
@@ -242,6 +242,8 @@ genericDriver - Command-line TTS (espeak, etc.)
.IP \[bu] 4
dectalkDriver - DECtalk-compatible serial hardware speech
.IP \[bu] 4
dectalkUsbDriver - Native Access Solutions DECtalk USB hardware speech
.IP \[bu] 4
litetalkDriver - LiteTalk-compatible serial hardware speech
.IP \[bu] 4
doubletalkDriver - DoubleTalk LT-compatible serial hardware speech
docs/fenrir.adoc
+12 -5
View File
@@ -1551,9 +1551,11 @@ enabled=True
Values: on=`+True+`, off=`+False+`
# Select speech driver, options are speechdDriver (default),
genericDriver, dectalkDriver, litetalkDriver, doubletalkDriver or tripletalkDriver: driver=speechdDriver
genericDriver, dectalkDriver, dectalkUsbDriver, litetalkDriver,
doubletalkDriver or tripletalkDriver: driver=speechdDriver
#driver=genericDriver
#driver=dectalkDriver
#driver=dectalkUsbDriver
#driver=litetalkDriver
#driver=doubletalkDriver
#driver=tripletalkDriver
@@ -1684,7 +1686,7 @@ the pico module:
language=de-DE
....
Hardware speech drivers use a serial device. Set an explicit path.
Serial hardware speech drivers use a serial device. Set an explicit path.
....
hardware_device=/dev/ttyACM0
@@ -1700,9 +1702,14 @@ hardware_baud_rate=9600
The `+doubletalkDriver+` targets DoubleTalk LT-style serial devices. It does
not support the internal DoubleTalk PC ISA card.
USB hardware speech synthesizers are supported only when Linux exposes them as
a serial tty such as `+/dev/ttyACM0+` or `+/dev/ttyUSB0+`. USB-only TripleTalk
models with no tty device need a separate driver.
The `+dectalkUsbDriver+` autodetects the native Access Solutions DECtalk USB,
USB ID `+0dd0:2001+`. It requires PyUSB and a libusb backend and ignores the
serial device and baud-rate settings. From a source checkout, install the USB
dependencies with `+python3 -m pip install '.[dectalk_usb]'+`. A non-root
Fenrir process needs a local udev rule granting its account read/write access
to `+0dd0:2001+`; do not make the device world-writable. Other USB hardware
speech synthesizers are supported only when Linux exposes them as a serial tty
such as `+/dev/ttyACM0+` or `+/dev/ttyUSB0+`.
Read new text as it occurs auto_read_incoming=True Values: on=`+True+`,
off=`+False+`
docs/user.md
+10 -4
View File
@@ -340,18 +340,24 @@ status formats, which change too frequently to support reliably.
- **speechdDriver** - Speech-dispatcher (recommended)
- **genericDriver** - Command-line TTS (espeak, etc.)
- **dectalkDriver** - Serial DECtalk-compatible hardware speech
- **dectalkUsbDriver** - Native Access Solutions DECtalk USB (`0dd0:2001`)
- **litetalkDriver** - Serial LiteTalk-compatible hardware speech
- **doubletalkDriver** - Serial DoubleTalk LT-compatible hardware speech
- **tripletalkDriver** - Serial TripleTalk-compatible hardware speech
For hardware speech, set `speech#hardware_device` to an explicit serial path.
For serial hardware speech, set `speech#hardware_device` to an explicit path.
RPITalk gadget mode usually appears as `/dev/ttyACM0`; USB serial adapters
usually appear as `/dev/ttyUSB0`; built-in serial ports may be `/dev/ttyS0`.
The default baud rate is `9600`. `doubletalkDriver` targets
DoubleTalk LT-style serial devices, not the internal DoubleTalk PC ISA card.
USB TripleTalk devices work only if Linux exposes them as a serial tty such as
`/dev/ttyACM0` or `/dev/ttyUSB0`; USB-only models with no tty device need a
separate driver.
The `dectalkUsbDriver` autodetects the native Access Solutions DECtalk USB and
ignores both serial settings. It requires PyUSB and a libusb backend. From a
source checkout, install them with
`python3 -m pip install '.[dectalk_usb]'`. A non-root Fenrir process also needs
a local udev rule granting its account read/write access to USB ID
`0dd0:2001`; do not make the device world-writable. USB TripleTalk devices work
only if Linux exposes them as a serial tty such as `/dev/ttyACM0` or
`/dev/ttyUSB0`.
### Sound Drivers
- **genericDriver** - Sox-based (default)
docs/user.txt
+5 -3
View File
@@ -878,10 +878,11 @@ Turn speech on or off:
enabled=True
Values: on=''True'', off=''False''
# Select speech driver, options are speechdDriver (default), genericDriver, dectalkDriver, litetalkDriver, doubletalkDriver or tripletalkDriver:
# Select speech driver, options are speechdDriver (default), genericDriver, dectalkDriver, dectalkUsbDriver, litetalkDriver, doubletalkDriver or tripletalkDriver:
driver=speechdDriver
#driver=genericDriver
#driver=dectalkDriver
#driver=dectalkUsbDriver
#driver=litetalkDriver
#driver=doubletalkDriver
#driver=tripletalkDriver
@@ -894,6 +895,7 @@ Available Drivers:
* ''genericDriver'' using the generic driver, for Fenrir <1.5 this is not available
* ''speechdDriver'' using speech-dispatcher, for Fenrir <1.5 just use ''speechd''
* ''dectalkDriver'' using DECtalk-compatible serial hardware or RPITalk
* ''dectalkUsbDriver'' using the native Access Solutions DECtalk USB
* ''litetalkDriver'' using LiteTalk-compatible serial hardware or RPITalk
* ''doubletalkDriver'' using DoubleTalk LT-compatible serial hardware
* ''tripletalkDriver'' using TripleTalk-compatible serial hardware
@@ -927,7 +929,7 @@ Select the language you want Fenrir to use.
language=english-us
Values: Text, see your TTS synths documentation what is available.
Hardware speech drivers use a serial device. Set an explicit path.
Serial hardware speech drivers use a serial device. Set an explicit path.
hardware_device=/dev/ttyACM0
hardware_device=/dev/ttyUSB0
hardware_device=/dev/ttyS0
@@ -936,7 +938,7 @@ Hardware speech drivers use 9600 baud by default.
hardware_baud_rate=9600
The doubletalkDriver targets DoubleTalk LT-style serial devices. It does not support the internal DoubleTalk PC ISA card.
USB hardware speech synthesizers are supported only when Linux exposes them as a serial tty such as /dev/ttyACM0 or /dev/ttyUSB0. USB-only TripleTalk models with no tty device need a separate driver.
The dectalkUsbDriver autodetects the native Access Solutions DECtalk USB, USB ID 0dd0:2001. It requires PyUSB and a libusb backend and ignores the serial device and baud-rate settings. Other USB hardware speech synthesizers are supported only when Linux exposes them as a serial tty such as /dev/ttyACM0 or /dev/ttyUSB0.
Read new text as it occurs
auto_read_incoming=True
setup.py
+1
View File
@@ -106,6 +106,7 @@ setup(
"pyte>=0.7.0",
],
extras_require={
"dectalk_usb": ["pyusb>=1.2.1"],
"x11": ["python-xlib>=0.33"],
},
)
src/fenrirscreenreader/commands/commands/voice_browser.py
+4 -19
View File
@@ -4,6 +4,7 @@ import subprocess
import time
from fenrirscreenreader.core.i18n import _
from fenrirscreenreader.utils.speechd_utils import get_synthesis_voice_name
class command:
@@ -273,29 +274,13 @@ class command:
for line in lines[1:]:
if not line.strip():
continue
if module.lower() == "espeak-ng":
voice = self.process_espeak_voice(line)
if voice:
voices.append(voice)
else:
voices.append(line.strip())
voice = get_synthesis_voice_name(module, line)
if voice:
voices.append(voice)
return voices
except Exception:
pass
return []
def process_espeak_voice(self, voiceLine):
"""Process espeak voice format"""
parts = [p for p in voiceLine.split() if p]
if len(parts) < 2:
return None
lang_code = parts[-2].lower()
variant = parts[-1].lower()
return (
f"{lang_code}+{variant}"
if variant and variant != "none"
else lang_code
)
def set_callback(self, callback):
pass
src/fenrirscreenreader/commands/commands/voice_browser_safe.py
+4 -22
View File
@@ -5,6 +5,7 @@ import threading
import time
from fenrirscreenreader.core.i18n import _
from fenrirscreenreader.utils.speechd_utils import get_synthesis_voice_name
class command:
@@ -179,12 +180,9 @@ class command:
for line in lines[1:]:
if not line.strip():
continue
if module.lower() == "espeak-ng":
voice = self.process_espeak_voice(line)
if voice:
voices.append(voice)
else:
voices.append(line.strip())
voice = get_synthesis_voice_name(module, line)
if voice:
voices.append(voice)
# Limit voice count to prevent memory issues
if len(voices) > 1000:
@@ -205,21 +203,5 @@ class command:
)
return []
def process_espeak_voice(self, voiceLine):
"""Process espeak voice format"""
try:
parts = [p for p in voiceLine.split() if p]
if len(parts) < 2:
return None
lang_code = parts[-2].lower()
variant = parts[-1].lower()
return (
f"{lang_code}+{variant}"
if variant and variant != "none"
else lang_code
)
except Exception:
return None
def set_callback(self, callback):
pass
src/fenrirscreenreader/commands/onScreenUpdate/65000-progress_detector.py
+44 -8
View File
@@ -79,10 +79,11 @@ class command:
# updates
delta_text = self.env["screen"]["new_delta"]
delta_length = len(delta_text)
explicit_progress_lines = self.get_explicit_progress_lines(delta_text)
if (
delta_length > 200
): # Allow longer progress lines such as terminal status output
if not self.is_explicit_progress_delta(delta_text):
if not explicit_progress_lines:
self.env["runtime"]["DebugManager"].write_debug_out(
f"Progress filter: delta too long ({delta_length})",
debug.DebugLevel.INFO,
@@ -91,7 +92,11 @@ class command:
# If delta contains newlines and is substantial, let incoming handler
# deal with it to avoid interfering with multi-line text output
if '\n' in delta_text and delta_length > 50:
if (
"\n" in delta_text
and delta_length > 50
and not explicit_progress_lines
):
self.env["runtime"]["DebugManager"].write_debug_out(
f"Progress filter: multiline delta ({delta_length} chars)",
debug.DebugLevel.INFO,
@@ -119,13 +124,35 @@ class command:
if not has_percentage:
return self.has_progress_status_fields(text)
return bool(
re.search(
r"[|\[\]#=*>█▉▊▋▌▍▎▏▒▓░]"
r"|\b\d+(?:\.\d+)?\s*[kKmMgGtT](?:i?B)?/s\b",
text,
)
bracketed_bar = re.search(
r"\[(?=[#=*>█▉▊▋▌▍▎▏▒▓░.\s-]{2,}\])"
r"(?=[^\]]*[#=*>█▉▊▋▌▍▎▏▒▓░])"
r"[#=*>█▉▊▋▌▍▎▏▒▓░.\s-]+\]",
text,
)
pipe_bar = re.search(
r"\|(?=[#=*>█▉▊▋▌▍▎▏▒▓░.\s-]{2,}\|)"
r"(?=[^|]*[#=*>█▉▊▋▌▍▎▏▒▓░])"
r"[#=*>█▉▊▋▌▍▎▏▒▓░.\s-]+\|",
text,
)
transfer_rate = re.search(
r"\b\d+(?:\.\d+)?\s*[kKmMgGtT](?:i?B)?/s\b",
text,
)
return bool(
bracketed_bar
or pipe_bar
or transfer_rate
)
def get_explicit_progress_lines(self, text):
"""Return structurally explicit progress rows from a screen delta."""
return [
line
for line in text.splitlines()
if self.is_explicit_progress_delta(line)
]
def reset_progress_state(self):
"""Reset progress state when a prompt is detected, allowing new progress operations to start fresh"""
@@ -140,6 +167,15 @@ class command:
import re
import time
if "\n" in text:
explicit_progress_lines = self.get_explicit_progress_lines(text)
if not explicit_progress_lines:
return
# Parallel transfer tools commonly repaint several package rows,
# followed by an aggregate row. One tone per update is enough, and
# the final explicit row is normally that aggregate.
text = explicit_progress_lines[-1]
current_time = time.time()
# Debug: Print what we're checking
src/fenrirscreenreader/core/dynamicVoiceMenu.py
+4 -23
View File
@@ -6,6 +6,7 @@ import subprocess
import time
from fenrirscreenreader.core import debug
from fenrirscreenreader.utils.speechd_utils import get_synthesis_voice_name
class DynamicVoiceCommand:
@@ -354,30 +355,10 @@ def get_module_voices(module):
for line in lines[1:]:
if not line.strip():
continue
if module.lower() == "espeak-ng":
voice = process_espeak_voice(line)
if voice:
voices.append(voice)
else:
voices.append(line.strip())
voice = get_synthesis_voice_name(module, line)
if voice:
voices.append(voice)
return voices
except Exception:
pass
return []
def process_espeak_voice(voiceLine):
"""Process espeak voice format"""
try:
parts = [p for p in voiceLine.split() if p]
if len(parts) < 2:
return None
lang_code = parts[-2].lower()
variant = parts[-1].lower()
return (
f"{lang_code}+{variant}"
if variant and variant != "none"
else lang_code
)
except Exception:
return None
src/fenrirscreenreader/core/outputManager.py
+8 -8
View File
@@ -133,16 +133,16 @@ class OutputManager:
"""
if not text:
return text
# Handle dots at the beginning of words (like .local, .bashrc, .config)
# Look for non-word character (or start of string), dot, then word characters
text = re.sub(r'(?<!\w)\.(\w+)', r'dot \1', text)
# Replace dots that appear between word characters with spoken form
# Use a loop to handle multiple consecutive dots like www.example.com or a.b.c.d
while re.search(r'\b\w+\.\w+\b', text):
text = re.sub(r'\b(\w+)\.(\w+)\b', r'\1 dot \2', text)
text = re.sub(r"(?<!\w)\.(\w+)", r"dot \1", text)
# Replace dots between word characters in one pass. The old loop
# rescanned the full speech string once per remaining dotted word,
# which made large directory listings block the event loop for seconds.
text = re.sub(r"(?<=\w)\.(?=\w)", " dot ", text)
return text
def speak_text(
src/fenrirscreenreader/core/quickMenuManager.py
+16 -32
View File
@@ -4,11 +4,16 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
import subprocess
import time
from fenrirscreenreader.core import debug
from fenrirscreenreader.core.i18n import _
from fenrirscreenreader.core.settingsData import settings_data
import subprocess
import time
from fenrirscreenreader.utils.speechd_utils import (
get_synthesis_voice_name,
parse_synthesis_voice_line,
)
class SpeechHelperMixin:
@@ -89,20 +94,15 @@ class SpeechHelperMixin:
for line in lines[1:]:
if not line.strip():
continue
if module.lower() == "espeak-ng":
voice = self._process_espeak_voice(line)
if voice:
voices.append(voice)
elif module.lower() == "voxin":
if module.lower() == "voxin":
# For Voxin, store voice name with language
voice_data = self._process_voxin_voice(line)
if voice_data:
voices.append(voice_data)
else:
# For other modules, extract first field (voice name)
parts = line.strip().split()
if parts:
voices.append(parts[0])
voice = get_synthesis_voice_name(module, line)
if voice:
voices.append(voice)
self._voices_cache[module] = voices
return voices
@@ -115,23 +115,6 @@ class SpeechHelperMixin:
return []
def _process_espeak_voice(self, voice_line):
"""Process espeak-ng voice format into usable voice name.
Args:
voice_line (str): Raw line from spd-say -L output
Returns:
str: Processed voice name (e.g., 'en-us' or 'en-us+f3')
"""
parts = [p for p in voice_line.split() if p]
if len(parts) < 2:
return None
lang_code = parts[-2].lower()
variant = parts[-1].lower()
return (f"{lang_code}+{variant}"
if variant and variant != "none" else lang_code)
def _process_voxin_voice(self, voice_line):
"""Process Voxin voice format with language information.
@@ -142,11 +125,12 @@ class SpeechHelperMixin:
Returns:
str: Voice name with language encoded (e.g., 'daniel-embedded-high|en-GB')
"""
parts = [p for p in voice_line.split() if p]
if len(parts) < 2:
voice_data = parse_synthesis_voice_line(voice_line)
if voice_data is None:
return None
voice_name, language, _variant = voice_data
if not language:
return None
voice_name = parts[0]
language = parts[1]
# Encode language with voice for later extraction
return f"{voice_name}|{language}"
src/fenrirscreenreader/core/settingsManager.py
+1
View File
@@ -521,6 +521,7 @@ class SettingsManager:
"speechdDriver",
"genericDriver",
"dectalkDriver",
"dectalkUsbDriver",
"doubletalkDriver",
"litetalkDriver",
"tripletalkDriver",
src/fenrirscreenreader/fenrirVersion.py
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
version = "2026.07.19"
version = "2026.07.29"
code_name = "master"
src/fenrirscreenreader/remoteDriver/unixDriver.py
+42 -36
View File
@@ -329,48 +329,54 @@ class driver(remoteDriver):
def watch_dog(self, active, event_queue):
# echo "command say this is a test" | socat -
# UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
for socket_file, optional in self._get_socket_candidates():
fenrir_sock = self._bind_socket(socket_file, optional)
if fenrir_sock is None:
continue
self.fenrirSocks.append(fenrir_sock)
self.bound_sockets.append((fenrir_sock, socket_file))
try:
for socket_file, optional in self._get_socket_candidates():
fenrir_sock = self._bind_socket(socket_file, optional)
if fenrir_sock is None:
continue
self.fenrirSocks.append(fenrir_sock)
self.bound_sockets.append((fenrir_sock, socket_file))
if not self.fenrirSocks:
return
if not self.fenrirSocks:
return
self._try_register_instance()
last_register = time.time()
while active.value:
if time.time() - last_register > 10.0:
self._try_register_instance()
last_register = time.time()
self._try_register_instance()
last_register = time.time()
while active.value:
if time.time() - last_register > 10.0:
self._try_register_instance()
last_register = time.time()
# Check if the client is still connected and if data is available:
try:
r, _, _ = select.select(self.fenrirSocks, [], [], 0.8)
except select.error:
break
if r == []:
continue
for fenrir_sock in r:
client_sock, client_addr = fenrir_sock.accept()
socket_file = self._socket_file_for_socket(fenrir_sock)
# Ensure client socket is always closed to prevent resource
# leaks
# Check if the client is still connected and data is available.
try:
self._handle_client(client_sock, event_queue, socket_file)
finally:
# Always close client socket, even if data processing fails
r, _, _ = select.select(self.fenrirSocks, [], [], 0.8)
except select.error:
break
if r == []:
continue
for fenrir_sock in r:
client_sock, _client_addr = fenrir_sock.accept()
socket_file = self._socket_file_for_socket(fenrir_sock)
try:
client_sock.close()
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"unixDriver watch_dog: Error closing client socket: "
+ str(e),
debug.DebugLevel.ERROR,
self._handle_client(
client_sock, event_queue, socket_file
)
self._cleanup()
finally:
try:
client_sock.close()
except Exception as e:
self.env["runtime"][
"DebugManager"
].write_debug_out(
"unixDriver watch_dog: Error closing client "
"socket: "
+ str(e),
debug.DebugLevel.ERROR,
)
finally:
# ProcessManager restarts a failed watchdog. Always release the
# old listeners first so each restart cannot leak another socket.
self._cleanup()
def shutdown(self):
self._cleanup()
src/fenrirscreenreader/screenDriver/vcsaDriver.py
+36 -15
View File
@@ -10,6 +10,7 @@
# blink = 5 if attr & 1 else 0
# bold = 1 if attr & 16 else 0
import errno
import fcntl
import glob
import os
@@ -53,6 +54,9 @@ class driver(screenDriver):
fgColorValues: Foreground color value mappings
hichar: High character mask for Unicode support
"""
read_retry_attempts = 10
read_retry_delay = 0.05
def __init__(self):
screenDriver.__init__(self)
self.ListSessions = None
@@ -251,23 +255,40 @@ class driver(screenDriver):
bytes: File content as bytes
"""
d = b""
file.seek(0)
try:
d = file.read()
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"vcsaDriver get_screen_text: Error reading file: " + str(e),
debug.DebugLevel.ERROR,
)
for attempt in range(self.read_retry_attempts + 1):
file.seek(0)
while True:
# Read from file
try:
d += file.readline(1)
if not d:
break
except Exception as e:
try:
return file.read()
except OSError as e:
if (
e.errno == errno.EINVAL
and attempt < self.read_retry_attempts
):
time.sleep(self.read_retry_delay)
continue
self.env["runtime"]["DebugManager"].write_debug_out(
"vcsaDriver get_screen_text: Error reading file: "
+ str(e),
debug.DebugLevel.ERROR,
)
break
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"vcsaDriver get_screen_text: Error reading file: "
+ str(e),
debug.DebugLevel.ERROR,
)
break
file.seek(0)
while True:
try:
chunk = file.readline(1)
if not chunk:
break
d += chunk
except Exception:
break
return d
def update_watchdog(self, active, event_queue):
src/fenrirscreenreader/speechDriver/dectalkDriver.py
+18 -5
View File
@@ -10,19 +10,32 @@ from fenrirscreenreader.speechDriver.hardwareSerialDriver import (
class driver(hardware_serial_driver):
cancel_command = b"\x18"
cancel_command = b"\x03"
use_xon_xoff = True
def _speak_bytes(self, text):
return self._clean_text(text).encode("ascii", errors="replace") + b"\x01"
text = self._neutralize_inline_commands(self._clean_text(text))
return text.encode("ascii", errors="replace") + b"\x0b"
def _neutralize_inline_commands(self, text):
return text.replace("[:", "[ :")
def _rate_command(self, rate):
return self._setting_command("ra", self._scale(rate, 75, 650))
value = self._scale_around_default(rate, 75, 180, 650)
return self._setting_command("ra", value)
def _pitch_command(self, pitch):
return self._setting_command("dv ap", self._scale(pitch, 50, 180))
value = self._scale_around_default(pitch, 50, 122, 350)
return self._setting_command("dv ap", value)
def _volume_command(self, volume):
return self._setting_command("vo", self._scale(volume, 0, 100))
return self._setting_command("dv g5", self._scale(volume, 60, 86))
def _scale_around_default(self, value, minimum, default, maximum):
value = max(0.0, min(1.0, value))
if value <= 0.5:
return self._scale(value * 2, minimum, default)
return self._scale((value - 0.5) * 2, default, maximum)
def _setting_command(self, command, value):
return f"[:{command} {value}]".encode("ascii")
src/fenrirscreenreader/speechDriver/dectalkUsbDriver.py
+288
View File
@@ -0,0 +1,288 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
import threading
from fenrirscreenreader.core import debug
from fenrirscreenreader.speechDriver import dectalkDriver
class driver(dectalkDriver.driver):
vendor_id = 0x0DD0
product_id = 0x2001
interface = 0
speech_endpoint = 0x02
immediate_endpoint = 0x04
packet_size = 64
write_timeout = 1000
buffer_timeout_command = b"[:timeout 1]"
def __init__(self):
super().__init__()
self._speech_cancel_generation = 0
self._speech_cancel_lock = threading.Lock()
def initialize(self, environment):
self.env = environment
self._is_initialized = False
self._reset_setting_cache()
self.usb_device = None
self.usb_core = None
self.usb_util = None
self.immediate_lock = threading.Lock()
self._debug(
"DECtalk USB initialize: requested_device=0dd0:2001",
debug.DebugLevel.INFO,
on_any_level=True,
)
try:
self.usb_core, self.usb_util = self._load_usb_modules()
except ImportError as error:
self._debug(
f"DECtalk USB requires PyUSB: {error}",
debug.DebugLevel.ERROR,
on_any_level=True,
)
raise RuntimeError("DECtalk USB requires PyUSB") from error
try:
usb_device = self.usb_core.find(
idVendor=self.vendor_id,
idProduct=self.product_id,
)
except (
self.usb_core.NoBackendError,
self.usb_core.USBError,
OSError,
) as error:
self._debug(
f"DECtalk USB discovery failed: {error}",
debug.DebugLevel.ERROR,
on_any_level=True,
)
raise RuntimeError(
f"DECtalk USB discovery failed: {error}"
) from error
if usb_device is None:
self._debug(
"DECtalk USB device 0dd0:2001 was not found",
debug.DebugLevel.ERROR,
on_any_level=True,
)
raise RuntimeError("DECtalk USB device 0dd0:2001 was not found")
try:
try:
usb_device.get_active_configuration()
except self.usb_core.USBError:
usb_device.set_configuration()
self.usb_util.claim_interface(usb_device, self.interface)
except (self.usb_core.USBError, OSError) as error:
self._dispose_usb_device(usb_device)
self._debug(
f"DECtalk USB open failed: {error}",
debug.DebugLevel.ERROR,
on_any_level=True,
)
raise RuntimeError(f"DECtalk USB open failed: {error}") from error
self.usb_device = usb_device
self._write_endpoint(
self.speech_endpoint,
self.buffer_timeout_command,
"buffer timeout",
)
self._is_initialized = True
self._stop_worker = False
self.worker_thread = threading.Thread(
target=self._worker,
daemon=True,
)
self.worker_thread.start()
self._debug(
"DECtalk USB device opened: 0dd0:2001",
debug.DebugLevel.INFO,
on_any_level=True,
)
def shutdown(self):
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)
try:
with self.lock, self.immediate_lock:
usb_device = self.usb_device
self.usb_device = None
if usb_device is None:
return
try:
self.usb_util.release_interface(
usb_device,
self.interface,
)
except (self.usb_core.USBError, OSError) as error:
self._debug(
f"DECtalk USB interface release failed: {error}",
debug.DebugLevel.WARNING,
)
self._dispose_usb_device(usb_device)
finally:
self._is_initialized = False
def cancel(self):
if not self._is_initialized:
return
self._advance_speech_cancel_generation()
self.clear_buffer()
self._write_endpoint(
self.immediate_endpoint,
b"\x06",
"cancel",
write_lock=self.immediate_lock,
)
def speak(self, text, queueable=True, ignore_punctuation=False):
if not self._is_initialized:
return
if not queueable:
self.cancel()
if not isinstance(text, str) or text == "":
return
self._debug(
"Hardware speech queued text: "
f"{len(text)} chars, queue_size={self.text_queue.qsize()}",
debug.DebugLevel.INFO,
on_any_level=True,
)
self.text_queue.put(
(self._get_speech_cancel_generation(), text)
)
def _worker(self):
while not self._stop_worker:
queue_item = self.text_queue.get()
if queue_item is None:
return
cancel_generation, text = queue_item
if cancel_generation != self._get_speech_cancel_generation():
continue
try:
data = self._speak_bytes(text)
self._debug(
"Hardware speech worker prepared speech bytes: "
f"{len(data)} bytes",
debug.DebugLevel.INFO,
on_any_level=True,
)
self._write_bytes(data, "speech", cancel_generation)
except Exception as error:
self._debug(
f"Hardware speech worker failed: {error}",
debug.DebugLevel.ERROR,
on_any_level=True,
)
def _write_bytes(self, data, description="data", cancel_generation=None):
if description == "speech" and cancel_generation is None:
cancel_generation = self._get_speech_cancel_generation()
return self._write_endpoint(
self.speech_endpoint,
data,
description,
cancel_generation=cancel_generation,
)
def _write_endpoint(
self,
endpoint,
data,
description,
write_lock=None,
cancel_generation=None,
):
if not data:
return False
if write_lock is None:
write_lock = self.lock
with write_lock:
if self.usb_device is None:
return False
try:
total_written = 0
while total_written < len(data):
if self._speech_cancel_requested(cancel_generation):
self._debug(
"DECtalk USB stopped speech write after "
f"{total_written} bytes because speech was "
"cancelled",
debug.DebugLevel.INFO,
on_any_level=True,
)
return False
chunk = data[
total_written : total_written + self.packet_size
]
bytes_written = self.usb_device.write(
endpoint,
chunk,
timeout=self.write_timeout,
)
if bytes_written != len(chunk):
raise OSError(
"USB write returned "
f"{bytes_written} of {len(chunk)} bytes"
)
total_written += bytes_written
preview = self._format_bytes_preview(data)
self._debug(
"DECtalk USB wrote "
f"{total_written} {description} bytes "
f"to endpoint 0x{endpoint:02x}: {preview}",
debug.DebugLevel.INFO,
on_any_level=True,
)
return True
except (self.usb_core.USBError, OSError) as error:
self._debug(
f"DECtalk USB write failed: {error}",
debug.DebugLevel.ERROR,
on_any_level=True,
)
return False
def _advance_speech_cancel_generation(self):
with self._speech_cancel_lock:
self._speech_cancel_generation += 1
def _get_speech_cancel_generation(self):
with self._speech_cancel_lock:
return self._speech_cancel_generation
def _speech_cancel_requested(self, cancel_generation):
if cancel_generation is None:
return False
return cancel_generation != self._get_speech_cancel_generation()
def _dispose_usb_device(self, usb_device):
try:
self.usb_util.dispose_resources(usb_device)
except Exception as error:
self._debug(
f"DECtalk USB resource disposal failed: {error}",
debug.DebugLevel.WARNING,
)
def _load_usb_modules(self):
import usb.core
import usb.util
return usb.core, usb.util
src/fenrirscreenreader/speechDriver/hardwareSerialDriver.py
+24 -6
View File
@@ -27,6 +27,7 @@ class SpeakQueue(Queue):
class hardware_serial_driver(speech_driver):
cancel_command = b""
default_baud_rate = 9600
use_xon_xoff = False
def __init__(self):
speech_driver.__init__(self)
@@ -37,10 +38,12 @@ class hardware_serial_driver(speech_driver):
self.lock = threading.Lock()
self.worker_thread = None
self._stop_worker = False
self._setting_cache = {}
def initialize(self, environment):
self.env = environment
self._is_initialized = False
self._reset_setting_cache()
settings_manager = self.env["runtime"]["SettingsManager"]
self.device = self._clean_device_setting(
settings_manager.get_setting("speech", "hardware_device")
@@ -114,21 +117,30 @@ class hardware_serial_driver(speech_driver):
return
if not isinstance(rate, float):
return
self._write_bytes(self._rate_command(rate), "rate")
self._write_setting_command("rate", self._rate_command(rate))
def set_pitch(self, pitch):
if not self._is_initialized:
return
if not isinstance(pitch, float):
return
self._write_bytes(self._pitch_command(pitch), "pitch")
self._write_setting_command("pitch", self._pitch_command(pitch))
def set_volume(self, volume):
if not self._is_initialized:
return
if not isinstance(volume, float):
return
self._write_bytes(self._volume_command(volume), "volume")
self._write_setting_command("volume", self._volume_command(volume))
def _reset_setting_cache(self):
self._setting_cache = {}
def _write_setting_command(self, setting, command):
if self._setting_cache.get(setting) == command:
return
if self._write_bytes(command, setting):
self._setting_cache[setting] = command
def _worker(self):
while not self._stop_worker:
@@ -176,7 +188,11 @@ class hardware_serial_driver(speech_driver):
attrs[5] = baud_rate
attrs[6][termios.VMIN] = 0
attrs[6][termios.VTIME] = 0
attrs[0] &= ~(termios.IXON | termios.IXOFF | termios.IXANY)
if self.use_xon_xoff:
attrs[0] |= termios.IXON
attrs[0] &= ~(termios.IXOFF | termios.IXANY)
else:
attrs[0] &= ~(termios.IXON | termios.IXOFF | termios.IXANY)
termios.tcsetattr(port, termios.TCSANOW, attrs)
return port
except (OSError, termios.error) as error:
@@ -218,10 +234,10 @@ class hardware_serial_driver(speech_driver):
def _write_bytes(self, data, description="data"):
if not data:
return
return False
with self.lock:
if self.serial_port is None:
return
return False
try:
total_written = 0
while total_written < len(data):
@@ -238,12 +254,14 @@ class hardware_serial_driver(speech_driver):
debug.DebugLevel.INFO,
on_any_level=True,
)
return True
except OSError as error:
self._debug(
f"Hardware speech write failed: {error}",
debug.DebugLevel.ERROR,
on_any_level=True,
)
return False
def _termios_baud_rate(self, baud_rate):
baud_name = f"B{baud_rate}"
src/fenrirscreenreader/utils/speechd_utils.py
+36
View File
@@ -0,0 +1,36 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
def parse_synthesis_voice_line(voice_line):
"""Parse one column-aligned spd-say synthesis voice row."""
fields = voice_line.rsplit(maxsplit=2)
if len(fields) != 3:
return None
name, language, variant = (field.strip() for field in fields)
if not name:
return None
return name, language, variant
def get_synthesis_voice_name(module, voice_line):
"""Return the value Speech Dispatcher expects for voice selection."""
voice_data = parse_synthesis_voice_line(voice_line)
if voice_data is None:
return None
name, language, variant = voice_data
if module.lower() != "espeak-ng":
return name
language = language.lower()
variant = variant.lower()
if not language:
return None
if variant and variant != "none":
return f"{language}+{variant}"
return language
tests/unit/test_dectalk_usb_driver.py
+351
View File
@@ -0,0 +1,351 @@
import threading
import time
from unittest.mock import Mock
import pytest
from fenrirscreenreader.speechDriver import dectalkUsbDriver
class FakeUsbDevice:
def __init__(self):
self.writes = []
def get_active_configuration(self):
return object()
def write(self, endpoint, data, timeout=None):
payload = bytes(data)
self.writes.append((endpoint, payload, timeout))
return len(payload)
class FakeUsbCore:
USBError = OSError
class NoBackendError(ValueError):
pass
def __init__(self, device, find_error=None):
self.device = device
self.find_error = find_error
self.find_calls = []
def find(self, **kwargs):
self.find_calls.append(kwargs)
if self.find_error:
raise self.find_error
return self.device
class FakeUsbUtil:
def __init__(
self,
claim_error=None,
release_error=None,
dispose_error=None,
):
self.claim_error = claim_error
self.release_error = release_error
self.dispose_error = dispose_error
self.claimed = []
self.released = []
self.disposed = []
def claim_interface(self, device, interface):
if self.claim_error:
raise self.claim_error
self.claimed.append((device, interface))
def release_interface(self, device, interface):
if self.release_error:
raise self.release_error
self.released.append((device, interface))
def dispose_resources(self, device):
if self.dispose_error:
raise self.dispose_error
self.disposed.append(device)
class BlockingUsbDevice(FakeUsbDevice):
def __init__(self):
super().__init__()
self.normal_write_started = threading.Event()
self.release_normal_write = threading.Event()
self.immediate_write_finished = threading.Event()
def write(self, endpoint, data, timeout=None):
if endpoint == 0x02:
self.normal_write_started.set()
self.release_normal_write.wait(timeout=1.0)
result = super().write(endpoint, data, timeout=timeout)
if endpoint == 0x04:
self.immediate_write_finished.set()
return result
class CancelDuringSpeechUsbDevice(FakeUsbDevice):
def __init__(self):
super().__init__()
self.first_normal_write_started = threading.Event()
self.release_first_normal_write = threading.Event()
def write(self, endpoint, data, timeout=None):
if endpoint == 0x02 and not self.first_normal_write_started.is_set():
self.first_normal_write_started.set()
self.release_first_normal_write.wait(timeout=1.0)
return super().write(endpoint, data, timeout=timeout)
def build_environment():
return {
"runtime": {
"SettingsManager": Mock(),
"DebugManager": Mock(),
}
}
def initialized_driver(monkeypatch, clear_startup_writes=True):
device = FakeUsbDevice()
usb_core = FakeUsbCore(device)
usb_util = FakeUsbUtil()
speech_driver = dectalkUsbDriver.driver()
monkeypatch.setattr(
speech_driver,
"_load_usb_modules",
lambda: (usb_core, usb_util),
)
speech_driver.initialize(build_environment())
if clear_startup_writes:
device.writes.clear()
return speech_driver, device, usb_core, usb_util
def wait_for_writes(device, count, timeout=1.0):
deadline = time.monotonic() + timeout
while len(device.writes) < count and time.monotonic() < deadline:
time.sleep(0.01)
def test_dectalk_usb_driver_finds_and_claims_native_device(monkeypatch):
speech_driver, device, usb_core, usb_util = initialized_driver(monkeypatch)
try:
assert usb_core.find_calls == [
{"idVendor": 0x0DD0, "idProduct": 0x2001}
]
assert usb_util.claimed == [(device, 0)]
assert speech_driver._is_initialized
finally:
speech_driver.shutdown()
assert usb_util.released == [(device, 0)]
assert usb_util.disposed == [device]
def test_dectalk_usb_driver_sets_low_buffer_timeout_on_initialize(monkeypatch):
speech_driver, device, _, _ = initialized_driver(
monkeypatch,
clear_startup_writes=False,
)
try:
assert device.writes[0] == (0x02, b"[:timeout 1]", 1000)
finally:
speech_driver.shutdown()
def test_dectalk_usb_driver_writes_commands_and_speech(monkeypatch):
speech_driver, device, _, _ = initialized_driver(monkeypatch)
try:
speech_driver.set_rate(1.0)
speech_driver.set_pitch(1.0)
speech_driver.set_volume(0.5)
speech_driver.speak("Hello\nworld")
wait_for_writes(device, 4)
assert device.writes == [
(0x02, b"[:ra 650]", 1000),
(0x02, b"[:dv ap 350]", 1000),
(0x02, b"[:dv g5 73]", 1000),
(0x02, b"Hello world\x0b", 1000),
]
finally:
speech_driver.shutdown()
def test_dectalk_usb_driver_skips_redundant_setting_writes(monkeypatch):
speech_driver, device, _, _ = initialized_driver(monkeypatch)
try:
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
speech_driver.set_volume(1.0)
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
speech_driver.set_volume(1.0)
assert device.writes == [
(0x02, b"[:ra 180]", 1000),
(0x02, b"[:dv ap 122]", 1000),
(0x02, b"[:dv g5 86]", 1000),
]
finally:
speech_driver.shutdown()
def test_dectalk_usb_driver_chunks_normal_writes(monkeypatch):
speech_driver, device, _, _ = initialized_driver(monkeypatch)
try:
speech_driver._write_bytes(b"x" * 70, "speech")
assert device.writes == [
(0x02, b"x" * 64, 1000),
(0x02, b"x" * 6, 1000),
]
finally:
speech_driver.shutdown()
def test_dectalk_usb_driver_uses_immediate_cancel_endpoint(monkeypatch):
speech_driver, device, _, _ = initialized_driver(monkeypatch)
try:
speech_driver.cancel()
assert device.writes == [(0x04, b"\x06", 1000)]
finally:
speech_driver.shutdown()
def test_dectalk_usb_cancel_bypasses_blocked_normal_write(monkeypatch):
device = BlockingUsbDevice()
usb_core = FakeUsbCore(device)
usb_util = FakeUsbUtil()
speech_driver = dectalkUsbDriver.driver()
monkeypatch.setattr(
speech_driver,
"_load_usb_modules",
lambda: (usb_core, usb_util),
)
speech_driver.initialize(build_environment())
device.writes.clear()
normal_thread = threading.Thread(
target=speech_driver._write_bytes,
args=(b"normal speech", "speech"),
)
cancel_thread = threading.Thread(target=speech_driver.cancel)
try:
normal_thread.start()
assert device.normal_write_started.wait(timeout=0.2)
cancel_thread.start()
assert device.immediate_write_finished.wait(timeout=0.2)
finally:
device.release_normal_write.set()
normal_thread.join(timeout=1.0)
cancel_thread.join(timeout=1.0)
speech_driver.shutdown()
def test_dectalk_usb_cancel_stops_current_speech_upload(monkeypatch):
device = CancelDuringSpeechUsbDevice()
usb_core = FakeUsbCore(device)
usb_util = FakeUsbUtil()
speech_driver = dectalkUsbDriver.driver()
monkeypatch.setattr(
speech_driver,
"_load_usb_modules",
lambda: (usb_core, usb_util),
)
speech_driver.initialize(build_environment())
device.writes.clear()
device.first_normal_write_started.clear()
normal_thread = threading.Thread(
target=speech_driver._write_bytes,
args=(b"x" * 192, "speech"),
)
try:
normal_thread.start()
assert device.first_normal_write_started.wait(timeout=0.2)
speech_driver.cancel()
finally:
device.release_first_normal_write.set()
normal_thread.join(timeout=1.0)
speech_driver.shutdown()
normal_writes = [
payload for endpoint, payload, _ in device.writes if endpoint == 0x02
]
assert normal_writes == [b"x" * 64]
def test_dectalk_usb_driver_reports_missing_pyusb(monkeypatch):
speech_driver = dectalkUsbDriver.driver()
def missing_pyusb():
raise ImportError("No module named usb")
monkeypatch.setattr(speech_driver, "_load_usb_modules", missing_pyusb)
with pytest.raises(RuntimeError, match="PyUSB"):
speech_driver.initialize(build_environment())
def test_dectalk_usb_driver_reports_missing_libusb_backend(monkeypatch):
speech_driver = dectalkUsbDriver.driver()
backend_error = FakeUsbCore.NoBackendError("No backend available")
usb_core = FakeUsbCore(None, find_error=backend_error)
monkeypatch.setattr(
speech_driver,
"_load_usb_modules",
lambda: (usb_core, FakeUsbUtil()),
)
with pytest.raises(RuntimeError, match="discovery failed.*backend"):
speech_driver.initialize(build_environment())
def test_dectalk_usb_driver_reports_missing_device(monkeypatch):
speech_driver = dectalkUsbDriver.driver()
usb_core = FakeUsbCore(None)
usb_util = FakeUsbUtil()
monkeypatch.setattr(
speech_driver,
"_load_usb_modules",
lambda: (usb_core, usb_util),
)
with pytest.raises(RuntimeError, match="0dd0:2001"):
speech_driver.initialize(build_environment())
def test_dectalk_usb_driver_reports_interface_permission_error(monkeypatch):
speech_driver = dectalkUsbDriver.driver()
device = FakeUsbDevice()
usb_core = FakeUsbCore(device)
usb_util = FakeUsbUtil(claim_error=PermissionError("access denied"))
monkeypatch.setattr(
speech_driver,
"_load_usb_modules",
lambda: (usb_core, usb_util),
)
with pytest.raises(RuntimeError, match="open failed.*access denied"):
speech_driver.initialize(build_environment())
def test_dectalk_usb_shutdown_survives_cleanup_errors(monkeypatch):
device = FakeUsbDevice()
usb_core = FakeUsbCore(device)
usb_util = FakeUsbUtil(
release_error=OSError("release failed"),
dispose_error=OSError("dispose failed"),
)
speech_driver = dectalkUsbDriver.driver()
monkeypatch.setattr(
speech_driver,
"_load_usb_modules",
lambda: (usb_core, usb_util),
)
speech_driver.initialize(build_environment())
speech_driver.shutdown()
assert not speech_driver._is_initialized
assert speech_driver.usb_device is None
tests/unit/test_hardware_speech_drivers.py
+76 -5
View File
@@ -1,5 +1,6 @@
import os
import select
import termios
import time
from unittest.mock import ANY
from unittest.mock import Mock
@@ -64,7 +65,19 @@ def test_dectalk_driver_speaks_printable_text(serial_pair):
speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
try:
speech_driver.speak("Hello\nworld ☃")
assert read_available(master_fd, 13) == b"Hello world \x01"
assert read_available(master_fd, 13) == b"Hello world \x0b"
finally:
speech_driver.shutdown()
def test_dectalk_driver_neutralizes_inline_commands_in_speech(
serial_pair,
):
speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
try:
speech_driver.speak("status [:dv g5 10] done")
expected = b"status [ :dv g5 10] done\x0b"
assert read_available(master_fd, len(expected)) == expected
finally:
speech_driver.shutdown()
@@ -73,12 +86,49 @@ def test_dectalk_driver_writes_settings_and_cancel(serial_pair):
speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
try:
speech_driver.set_rate(1.0)
speech_driver.set_pitch(0.0)
speech_driver.set_pitch(1.0)
speech_driver.set_volume(0.5)
speech_driver.cancel()
assert read_available(master_fd, 33) == (
b"[:ra 650][:dv ap 50][:vo 50]\x18"
)
expected = b"[:ra 650][:dv ap 350][:dv g5 73]\x03"
assert read_available(master_fd, len(expected)) == expected
finally:
speech_driver.shutdown()
def test_dectalk_driver_skips_redundant_setting_commands(serial_pair):
speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
try:
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
speech_driver.set_volume(1.0)
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
speech_driver.set_volume(1.0)
expected = b"[:ra 180][:dv ap 122][:dv g5 86]"
assert read_available(master_fd, len(expected)) == expected
assert read_available(master_fd, 1, timeout=0.2) == b""
finally:
speech_driver.shutdown()
def test_dectalk_driver_uses_native_defaults_at_fenrir_midpoint(serial_pair):
speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
try:
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
expected = b"[:ra 180][:dv ap 122]"
assert read_available(master_fd, len(expected)) == expected
finally:
speech_driver.shutdown()
def test_dectalk_driver_enables_xon_xoff_flow_control(serial_pair):
speech_driver, _ = initialized_driver(dectalkDriver, serial_pair)
try:
input_flags = termios.tcgetattr(speech_driver.serial_port)[0]
assert input_flags & termios.IXON
assert not input_flags & termios.IXOFF
assert not input_flags & termios.IXANY
finally:
speech_driver.shutdown()
@@ -165,6 +215,27 @@ def test_hardware_driver_retries_partial_serial_writes(monkeypatch):
assert written_chunks == [b"ab", b"cd", b"ef"]
def test_hardware_driver_retries_setting_after_failed_write(monkeypatch):
speech_driver = dectalkDriver.driver()
speech_driver._is_initialized = True
write_results = [False, True]
written_commands = []
def fake_write(data, description):
written_commands.append((data, description))
return write_results.pop(0)
monkeypatch.setattr(speech_driver, "_write_bytes", fake_write)
speech_driver.set_rate(0.5)
speech_driver.set_rate(0.5)
assert written_commands == [
(b"[:ra 180]", "rate"),
(b"[:ra 180]", "rate"),
]
@pytest.mark.parametrize("driver_class", [doubletalkDriver, tripletalkDriver])
def test_litetalk_compatible_alias_drivers(driver_class, serial_pair):
speech_driver, master_fd = initialized_driver(driver_class, serial_pair)
tests/unit/test_output_manager.py
+13
View File
@@ -178,6 +178,19 @@ def test_present_text_records_speech_history_when_enabled():
speech_driver.speak.assert_called_once()
@pytest.mark.unit
def test_process_mid_word_punctuation_handles_chained_dotted_words():
output_manager, _sound_driver, _speech_driver = build_output_manager()
result = output_manager.process_mid_word_punctuation(
".config archive.tar.gz settings.conf."
)
assert result == (
"dot config archive dot tar dot gz settings dot conf."
)
@pytest.mark.unit
def test_present_text_does_not_record_when_speech_disabled():
output_manager, _sound_driver, speech_driver = build_output_manager()
tests/unit/test_progress_detector.py
+52
View File
@@ -238,3 +238,55 @@ def test_progress_detector_ignores_generic_key_value_text():
command.play_activity_beep.assert_not_called()
command.play_progress_tone.assert_not_called()
@pytest.mark.unit
def test_progress_detector_beeps_for_pacman_total_in_multiline_delta():
progress_module = _load_progress_module()
command = progress_module.command()
total_line = (
"Total ( 3/749) "
"830.7 MiB 4.70 MiB/s 04:05 "
"[#################################################"
"-----------------------------------------------------------------------] 41%"
)
sample = (
"package-one 125.0 MiB 2.10 MiB/s 00:12 "
"[####################--------------------] 50%\n"
+ total_line
)
command.env = {
"commandBuffer": {
"progress_monitoring": True,
"lastProgressValue": -1,
"lastProgressTime": 0,
},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
},
"screen": {
"new_delta": sample,
"new_delta_is_typing": False,
"new_content_text": sample,
"old_cursor": {"x": 0, "y": 0},
"new_cursor": {"x": 0, "y": 0},
},
}
command.play_progress_tone = Mock()
command.run()
command.play_progress_tone.assert_called_once_with(41.0)
assert command.env["commandBuffer"]["lastProgressValue"] == 41.0
@pytest.mark.unit
def test_explicit_progress_delta_ignores_percentage_in_bracketed_prose():
progress_module = _load_progress_module()
command = progress_module.command()
assert not command.is_explicit_progress_delta(
"Release notes [issue #749] are 41% complete"
)
tests/unit/test_settings_validation.py
+3
View File
@@ -74,6 +74,9 @@ class TestSpeechSettingsValidation:
self.manager._validate_setting_value("speech", "driver", "speechdDriver")
self.manager._validate_setting_value("speech", "driver", "genericDriver")
self.manager._validate_setting_value("speech", "driver", "dectalkDriver")
self.manager._validate_setting_value(
"speech", "driver", "dectalkUsbDriver"
)
self.manager._validate_setting_value("speech", "driver", "doubletalkDriver")
self.manager._validate_setting_value("speech", "driver", "litetalkDriver")
self.manager._validate_setting_value("speech", "driver", "tripletalkDriver")
tests/unit/test_speechd_voice_selection.py
+90
View File
@@ -0,0 +1,90 @@
from types import SimpleNamespace
from unittest.mock import Mock
from fenrirscreenreader.commands.commands import voice_browser
from fenrirscreenreader.commands.commands import voice_browser_safe
from fenrirscreenreader.core import dynamicVoiceMenu
from fenrirscreenreader.core.quickMenuManager import SpeechHelperMixin
from fenrirscreenreader.utils.speechd_utils import get_synthesis_voice_name
VOICE_LIST = (
" NAME LANGUAGE VARIANT\n"
" Perfect Paul en-US none\n"
" Big Bob en-US none\n"
)
def completed_voice_list():
return SimpleNamespace(returncode=0, stdout=VOICE_LIST)
def test_quick_menu_preserves_multiword_synthesis_voice_names(monkeypatch):
monkeypatch.setattr(
"fenrirscreenreader.core.quickMenuManager.subprocess.run",
Mock(return_value=completed_voice_list()),
)
helper = SpeechHelperMixin()
helper.env = {"runtime": {"DebugManager": Mock()}}
assert helper.get_module_voices("doubletalk") == [
"Perfect Paul",
"Big Bob",
]
def test_safe_voice_browser_preserves_multiword_synthesis_voice_names(
monkeypatch,
):
monkeypatch.setattr(
voice_browser_safe.subprocess,
"run",
Mock(return_value=completed_voice_list()),
)
browser = voice_browser_safe.command()
browser.initialize({"runtime": {}})
assert browser.get_module_voices_with_timeout("doubletalk") == [
"Perfect Paul",
"Big Bob",
]
def test_interactive_voice_browser_preserves_multiword_synthesis_voice_names(
monkeypatch,
):
monkeypatch.setattr(
voice_browser.subprocess,
"run",
Mock(return_value=completed_voice_list()),
)
browser = voice_browser.command()
assert browser.get_module_voices("doubletalk") == [
"Perfect Paul",
"Big Bob",
]
def test_dynamic_voice_menu_preserves_multiword_synthesis_voice_names(
monkeypatch,
):
monkeypatch.setattr(
dynamicVoiceMenu.subprocess,
"run",
Mock(return_value=completed_voice_list()),
)
assert dynamicVoiceMenu.get_module_voices("doubletalk") == [
"Perfect Paul",
"Big Bob",
]
def test_espeak_voice_selection_keeps_language_and_variant_behavior():
voice = get_synthesis_voice_name(
"espeak-ng",
"English (America)+female3 en-US female3",
)
assert voice == "en-us+female3"
tests/unit/test_unix_remote_driver.py
+36
View File
@@ -1,5 +1,7 @@
from unittest.mock import Mock, mock_open, patch
import pytest
from fenrirscreenreader.remoteDriver import unixDriver
@@ -27,6 +29,40 @@ def test_watchdog_keeps_serving_when_instance_registration_fails(
select_call.assert_called_once()
def test_watchdog_closes_listeners_when_unexpected_error_escapes(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
listener = Mock()
active = Mock(value=True)
with patch.object(
driver,
"_get_socket_candidates",
return_value=[("/tmp/fenrir-test.sock", False)],
), patch.object(
driver, "_bind_socket", return_value=listener
), patch.object(
driver, "_try_register_instance"
), patch(
"fenrirscreenreader.remoteDriver.unixDriver.select.select",
side_effect=RuntimeError("unexpected failure"),
), patch(
"fenrirscreenreader.remoteDriver.unixDriver.os.path.exists",
return_value=False,
), patch(
"fenrirscreenreader.remoteDriver.unixDriver.remoteInstanceRegistry.remove_instance"
) as remove_instance:
with pytest.raises(RuntimeError, match="unexpected failure"):
driver.watch_dog(active, Mock())
listener.close.assert_called_once_with()
assert driver.fenrirSocks == []
assert driver.bound_sockets == []
remove_instance.assert_called_once_with()
class FakeClientSocket:
def __init__(self, data):
self.data = data
tests/unit/test_vcsa_driver.py
+70
View File
@@ -0,0 +1,70 @@
from unittest.mock import Mock
from fenrirscreenreader.screenDriver.vcsaDriver import driver as VcsaDriver
class BulkReadFailure:
def __init__(self, chunks):
self.chunks = iter(chunks)
self.readline_calls = 0
def seek(self, offset):
assert offset == 0
def read(self):
raise OSError(22, "Invalid argument")
def readline(self, size):
assert size == 1
self.readline_calls += 1
if self.readline_calls > 3:
raise RuntimeError("read continued after EOF")
return next(self.chunks)
class TransientInvalidArgumentRead:
def __init__(self, payload):
self.payload = payload
self.read_calls = 0
self.readline_calls = 0
def seek(self, offset):
assert offset == 0
def read(self):
self.read_calls += 1
if self.read_calls == 1:
raise OSError(22, "Invalid argument")
return self.payload
def readline(self, size):
self.readline_calls += 1
return b""
def build_vcsa_driver():
vcsa_driver = VcsaDriver.__new__(VcsaDriver)
vcsa_driver.env = {
"runtime": {
"DebugManager": Mock(),
}
}
vcsa_driver.read_retry_delay = 0
return vcsa_driver
def test_vcsa_read_fallback_stops_at_eof():
vcsa_driver = build_vcsa_driver()
screen_file = BulkReadFailure([b"a", b"b", b""])
assert vcsa_driver.read_file(screen_file) == b"ab"
assert screen_file.readline_calls == 3
def test_vcsa_read_retries_transient_invalid_argument():
vcsa_driver = build_vcsa_driver()
screen_file = TransientInvalidArgumentRead(b"\x01\x02\x00\x00x")
assert vcsa_driver.read_file(screen_file) == b"\x01\x02\x00\x00x"
assert screen_file.read_calls == 2
assert screen_file.readline_calls == 0