26 Commits

Author SHA1 Message Date
Storm Dragon 14227981cb Release 2026.08.06 2026-08-06 13:51:18 -04:00
Storm Dragon ee441c319a Improved progress bar detections. 2026-08-06 13:49:53 -04:00
Storm Dragon 5cf7a11260 Release candidate after bug fixes. 2026-07-30 04:57:29 -04:00
Storm Dragon 7593c8d0b3 Fix to socket routing for self-voicing. 2026-07-30 04:52:52 -04:00
Storm Dragon 312cb46f22 Release candidate. 2026-07-29 19:12:49 -04:00
Storm Dragon 122a6a38b3 Fix Unix remote watchdog socket leak 2026-07-29 14:13:53 -04:00
Storm Dragon 76b8f2f2cf Very scary test here. Trying to see if we can boost speed on very large terminals. 2026-07-26 22:01:11 -04:00
Storm Dragon d81565fbf5 Still attempting to kill off this delay. 2026-07-26 21:49:20 -04:00
Storm Dragon 186512f79a Another try for lag. 2026-07-26 21:42:53 -04:00
Storm Dragon 17b6e6f909 Attempt to decrease dectalk lag. Hoping you don't have to send settings over every time you want it to say something. 2026-07-26 21:34:04 -04:00
Storm Dragon fac3065c6a More tweaks to dectalk driver. Think we're nearly there. Hopefully. 2026-07-26 21:24:28 -04:00
Storm Dragon e93aba109a Fixes with the dectalk driver with rate, etc. 2026-07-26 21:10:39 -04:00
Storm Dragon 11a4244cc3 Fixed some errors with my dectalk protocol implementation. Thanks Codex. :) 2026-07-26 20:59:14 -04:00
Storm Dragon 9ac2e9e844 Found and hopefullyfixed a bug with fenrir watchdog. 2026-07-26 20:44:50 -04:00
Storm Dragon c863245c24 More attempts at getting this usb dectalk working. 2026-07-26 20:18:17 -04:00
Storm Dragon 45bd33f756 Improvements to Dectalk driver. 2026-07-26 19:44:10 -04:00
Storm Dragon 6d9260e79b Fixed 2 speech-dispatcher bugs. First, multiword voice names were not available. Next, when the module changed the old voice name was carried in causing some modules to fail. 2026-07-25 04:36:39 -04:00
Storm Dragon e180fd53fa Release candidate. 2026-07-19 16:02:24 -04:00
Storm Dragon 385356e745 Prevent incoming output from bypassing auto-read toggle 2026-07-19 15:59:51 -04:00
Storm Dragon 81f1757357 Release candidate. 2026-07-15 00:20:57 -04:00
Storm Dragon 28e4f8ab0c Progress bar beeps now work on ffmpeg output. 2026-07-12 15:06:44 -04:00
lilmike 4d94b56ee9 Fix very high cpu usage when a console fenrir under root and an x fenrir under your user are running at the same time. 2026-07-01 23:44:12 -07:00
Storm Dragon eb93a8aa62 Fixed a typo in tests/unit/test_hardware_speech_drivers.py. It was reading 9 bytes while comparing against a 10-byte payload. Thanks Samuel Thibault for finding and pointing this out. 2026-06-27 16:59:05 -04:00
Storm Dragon ad863ad706 Add ability for X clibpard and Fenrir's clipboard to stay in sync. 2026-06-27 16:55:00 -04:00
Storm Dragon 447e2cc896 Missed one commented setting. 2026-06-18 10:51:44 -04:00
Storm Dragon 2b7d205f06 Fixed some half-completed work in generic drivers. Minor -x bug fixes. 2026-06-18 10:35:48 -04:00
56 changed files with 2958 additions and 428 deletions
+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:*
+32 -20
View File
@@ -19,14 +19,14 @@ volume=0.7
# shell commands for generic sound driver
# the folowing variable are substituted
# fenrirVolume = the current volume setting
# fenrirSoundFile = the soundfile for an soundicon
# fenrirFrequence = the frequency to play
# fenrirDuration = the duration of the frequency
# fenrir_volume = the current volume setting
# fenrir_sound_file = the soundfile for an soundicon
# fenrir_frequency = the frequency to play
# fenrir_duration = the duration of the frequency
# the following command is used to play a soundfile
generic_play_file_command=play -q -v fenrirVolume fenrirSoundFile
generic_play_file_command=play -q -v fenrir_volume fenrir_sound_file
#the following command is used to generate a frequency beep
generic_frequency_command=play -q -v fenrirVolume -n -c1 synth fenrirDuration sine fenrirFrequence
generic_frequency_command=play -q -v fenrir_volume -n -c1 synth fenrir_duration sine fenrir_frequency
# Enable progress bar monitoring with ascending tones by default
progress_monitoring=True
@@ -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.
@@ -114,17 +117,17 @@ max_batch_lines=100
# Only enable flood control if this many new lines appear in the window
flood_line_threshold=500
# genericSpeechCommand is the command that is executed for talking
# generic_speech_command is the command that is executed for talking
# the following variables are replaced with values
# fenrirText = is the text that should be spoken
# fenrirModule = may be the speech module like used in speech-dispatcher, not every TTY need this
# fenrirLanguage = the language
# fenrirVoice = is the current voice that should be used. Set the voice variable above.
# fenrir_text = is the text that should be spoken
# fenrir_module = may be the speech module like used in speech-dispatcher, not every TTY need this
# fenrir_language = the language
# fenrir_voice = is the current voice that should be used. Set the voice variable above.
# the current volume, pitch and rate is calculated like this
# value = min + settingValue * (min - max )
# fenrirVolume = is replaced with the current volume
# fenrirPitch = is replaced with the current pitch
# fenrirRate = is replaced with the current speed (speech rate)
# fenrir_volume = is replaced with the current volume
# fenrir_pitch = is replaced with the current pitch
# fenrir_rate = is replaced with the current speed (speech rate)
generic_speech_command=espeak-ng -a fenrir_volume -s fenrir_rate -p fenrir_pitch -v fenrir_voice -- "fenrir_text"
# min and max values of the TTS system that is used in generic_speech_command
@@ -198,11 +201,6 @@ respect_punctuation_pause=True
replace_undefined_punctuation_with_space=True
# Pause speech briefly at newline characters for better readability
new_line_pause=True
number_of_clipboards=50
# used path for "export_clipboard_to_file"
# $user is replaced by username
#clipboardExportPath=/home/$user/fenrirClipboard
clipboard_export_path=/tmp/fenrirClipboard
# Convert text emoticons like :) to descriptive text (e.g., "smiling face")
emoticons=True
# Define the Fenrir modifier key(s) - used to trigger Fenrir commands
@@ -263,6 +261,20 @@ diff_presentation=both
# verbose = include diff line content during navigation
diff_verbosity=compact
[clipboard]
# Number of clipboard history entries Fenrir keeps.
number_of_clipboards=50
# used path for "export_clipboard_to_file"
# $user is replaced by username
#clipboard_export_path=/home/$user/fenrirClipboard
clipboard_export_path=/tmp/fenrirClipboard
# Keep Fenrir's clipboard history and the X clipboard synchronized.
# In fenrir -x, an empty sync_display uses the current DISPLAY.
# In console/TTY mode, set sync_display explicitly, for example :0.
sync_enabled=False
sync_display=
sync_interval=0.5
[focus]
# Follow and announce text cursor position changes
cursor=True
+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
+27 -18
View File
@@ -1506,38 +1506,38 @@ Values: `+0.0+` is quietest, `+1.0+` is loudest.
The generic sound driver uses shell commands for play sound and
frequencies.
`+genericPlayFileCommand+` defines the command that is used to play a
`+generic_play_file_command+` defines the command that is used to play a
sound file.
....
generic_play_file_command=<your command for playing a file>
....
`+genericFrequencyCommand+` defines the command that is used playing
`+generic_frequency_command+` defines the command that is used playing
frequencies.
....
generic_frequency_command=<your command for playing a frequence>
....
The following variables are substituted in `+genericPlayFileCommand+`
and `+genericFrequencyCommand+`:
The following variables are substituted in `+generic_play_file_command+`
and `+generic_frequency_command+`:
* `+fenrirVolume+` = the current volume setting
* `+fenrirSoundFile+` = the sound file for an sound icon
* `+fenrirFrequence+` = the frequency to play
* `+fenrirDuration+` = the duration of the frequency
* `+fenrir_volume+` = the current volume setting
* `+fenrir_sound_file+` = the sound file for an sound icon
* `+fenrir_frequency+` = the frequency to play
* `+fenrir_duration+` = the duration of the frequency
Example genericPlayFileCommand (default)
Example generic_play_file_command (default)
....
generic_play_file_command=play -q -v fenrirVolume fenrirSoundFile
generic_play_file_command=play -q -v fenrir_volume fenrir_sound_file
....
Example genericFrequencyCommand (default)
Example generic_frequency_command (default)
....
generic_frequency_command=play -q -v fenrirVolume -n -c1 synth fenrirDuration sine fenrirFrequence
generic_frequency_command=play -q -v fenrir_volume -n -c1 synth fenrir_duration sine fenrir_frequency
....
==== Speech
@@ -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+`
@@ -1921,7 +1928,8 @@ link:#export clipboard to file[export clipboard to file]. The variable
`+$user+` is replaced by the current logged username.
....
clipboardExportPath=/tmp/fenrirClipboard
[clipboard]
clipboard_export_path=/tmp/fenrirClipboard
....
Values: Text, Systemfilepath
@@ -1929,6 +1937,7 @@ Values: Text, Systemfilepath
The number of available clipboards:
....
[clipboard]
number_of_clipboards=10
....
+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)
+32 -28
View File
@@ -857,31 +857,32 @@ Values: ''0.0'' is quietest, ''1.0'' is loudest.
=== Generic Driver ===
The generic sound driver uses shell commands for play sound and frequencies.
''genericPlayFileCommand'' defines the command that is used to play a sound file.
''generic_play_file_command'' defines the command that is used to play a sound file.
generic_play_file_command=<your command for playing a file>
''genericFrequencyCommand'' defines the command that is used playing frequencies.
''generic_frequency_command'' defines the command that is used playing frequencies.
generic_frequency_command=<your command for playing a frequence>
The following variables are substituted in ''genericPlayFileCommand'' and ''genericFrequencyCommand'':
* ''fenrirVolume'' = the current volume setting
* ''fenrirSoundFile'' = the sound file for an sound icon
* ''fenrirFrequence'' = the frequency to play
* ''fenrirDuration'' = the duration of the frequency
The following variables are substituted in ''generic_play_file_command'' and ''generic_frequency_command'':
* ''fenrir_volume'' = the current volume setting
* ''fenrir_sound_file'' = the sound file for an sound icon
* ''fenrir_frequency'' = the frequency to play
* ''fenrir_duration'' = the duration of the frequency
Example genericPlayFileCommand (default)
generic_play_file_command=play -q -v fenrirVolume fenrirSoundFile
Example genericFrequencyCommand (default)
generic_frequency_command=play -q -v fenrirVolume -n -c1 synth fenrirDuration sine fenrirFrequence
Example generic_play_file_command (default)
generic_play_file_command=play -q -v fenrir_volume fenrir_sound_file
Example generic_frequency_command (default)
generic_frequency_command=play -q -v fenrir_volume -n -c1 synth fenrir_duration sine fenrir_frequency
==== Speech ====
Speech is configured in section ''[speech]''.
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
@@ -945,22 +947,22 @@ Values: on=''True'', off=''False''
=== Generic Driver ===
The generic speech driver uses shell commands for speech synthisus.
''genericSpeechCommand'' defines the command that is executed for creating speech
The following variables are substituted in ''genericSpeechCommand'':
* ''FenrirText'' = is the text that should be spoken
* ''fenrirModule'' = may be the speech module like used in speech-dispatcher, not every TTY needs this
* ''fenrirLanguage'' = the language to speak in
* ''fenrirVoice'' = is the current voice that should be used
* ''fenrirVolume'' = is replaced with the current volume
* ''fenrirPitch'' = is replaced with the current pitch
* ''fenrirRate'' = is replaced with the current speed (speech rate)
''generic_speech_command'' defines the command that is executed for creating speech
The following variables are substituted in ''generic_speech_command'':
* ''fenrir_text'' = is the text that should be spoken
* ''fenrir_module'' = may be the speech module like used in speech-dispatcher, not every TTY needs this
* ''fenrir_language'' = the language to speak in
* ''fenrir_voice'' = is the current voice that should be used
* ''fenrir_volume'' = is replaced with the current volume
* ''fenrir_pitch'' = is replaced with the current pitch
* ''fenrir_rate'' = is replaced with the current speed (speech rate)
Example genericSpeechCommand (default):
generic_speech_command=espeak -a fenrirVolume -s fenrirRate -p fenrirPitch -v fenrirVoice "fenrirText"
Example generic_speech_command (default):
generic_speech_command=espeak -a fenrir_volume -s fenrir_rate -p fenrir_pitch -v fenrir_voice "fenrir_text"
These are the minimum and maximum values of the TTS system used in genericSpeechCommand. They are needed to calculate the abstract range in volume, rate and pitch 0.0 - 1.0.
These are the minimum and maximum values of the TTS system used in generic_speech_command. They are needed to calculate the abstract range in volume, rate and pitch 0.0 - 1.0.
FenrirMinVolume=0
fenrir_min_volume=0
fenrir_max_volume=200
fenrir_min_pitch=0
fenrir_max_pitch=99
@@ -1133,10 +1135,12 @@ Values: on=''True'', off=''False''
Specify the path where the clipboard should be exported to.
See [[#export clipboard to file|export clipboard to file]].
The variable ''$user'' is replaced by the current logged username.
clipboardExportPath=/tmp/fenrirClipboard
[clipboard]
clipboard_export_path=/tmp/fenrirClipboard
Values: Text, Systemfilepath
The number of available clipboards:
[clipboard]
number_of_clipboards=10
Values: Integer, 1 - 999
+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"],
},
)
@@ -22,7 +22,7 @@ class command:
self.env["runtime"]["MemoryManager"].add_index_list(
"clipboardHistory",
self.env["runtime"]["SettingsManager"].get_setting_as_int(
"general", "number_of_clipboards"
"clipboard", "number_of_clipboards"
),
)
@@ -26,7 +26,7 @@ class command:
def run(self):
clipboard_file_path = self.env["runtime"][
"SettingsManager"
].get_setting("general", "clipboard_export_path")
].get_setting("clipboard", "clipboard_export_path")
clipboard_file_path = clipboard_file_path.replace(
"$user", self.env["general"]["curr_user"]
)
@@ -5,12 +5,9 @@
# By Chrys, Storm Dragon, and contributors.
import _thread
import importlib
import os
import pyperclip
from fenrirscreenreader.core.i18n import _
from fenrirscreenreader.utils import x_clipboard
class command:
@@ -46,36 +43,20 @@ class command:
"MemoryManager"
].get_index_list_element("clipboardHistory")
# Remember original display environment variable if it exists
original_display = os.environ.get("DISPLAY", "")
success = False
# Try different display options
for i in range(10):
display = f":{i}"
try:
# Set display environment variable
os.environ["DISPLAY"] = display
# Attempt to set clipboard content
# Weird workaround for some distros
importlib.reload(pyperclip)
pyperclip.copy(clipboard)
# If we get here without exception, we found a working
# display
success = True
break
except Exception:
# Failed for this display, try next one
continue
# Restore original display setting
if original_display:
os.environ["DISPLAY"] = original_display
else:
os.environ.pop("DISPLAY", None)
try:
success = x_clipboard.write_text(
clipboard, scan_displays=True
)
except Exception:
success = False
# Notify the user of the result
if success:
sync_manager = self.env["runtime"].get(
"ClipboardSyncManager"
)
if sync_manager:
sync_manager.mark_written_to_x(clipboard)
self.env["runtime"]["OutputManager"].present_text(
_("exported to the X session."), interrupt=True
)
@@ -26,7 +26,7 @@ class command:
def run(self):
clipboard_file_path = self.env["runtime"][
"SettingsManager"
].get_setting("general", "clipboard_export_path")
].get_setting("clipboard", "clipboard_export_path")
clipboard_file_path = clipboard_file_path.replace(
"$user", self.env["general"]["curr_user"]
)
@@ -5,12 +5,9 @@
# By Chrys, Storm Dragon, and contributors.
import _thread
import importlib
import os
import pyperclip
from fenrirscreenreader.core.i18n import _
from fenrirscreenreader.utils import x_clipboard
class command:
@@ -32,33 +29,12 @@ class command:
def _thread_run(self):
try:
# Remember original display environment variable if it exists
original_display = os.environ.get("DISPLAY", "")
clipboard_content = None
# Try different display options
for i in range(10):
display = f":{i}"
try:
# Set display environment variable
os.environ["DISPLAY"] = display
# Attempt to get clipboard content
# Weird workaround for some distros
importlib.reload(pyperclip)
clipboard_content = pyperclip.paste()
# If we get here without exception, we found a working
# display
if clipboard_content:
break
except Exception:
# Failed for this display, try next one
continue
# Restore original display setting
if original_display:
os.environ["DISPLAY"] = original_display
else:
os.environ.pop("DISPLAY", None)
try:
clipboard_content = x_clipboard.read_text(
scan_displays=True
)
except Exception:
clipboard_content = None
# Process the clipboard content if we found any
if clipboard_content and isinstance(clipboard_content, str):
@@ -18,7 +18,7 @@ class command:
self.env["runtime"]["MemoryManager"].add_index_list(
"clipboardHistory",
self.env["runtime"]["SettingsManager"].get_setting_as_int(
"general", "number_of_clipboards"
"clipboard", "number_of_clipboards"
),
)
@@ -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
@@ -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
@@ -27,6 +27,8 @@ class command:
return
if not self.env["runtime"]["ScreenManager"].is_delta():
return
if not self.env["screen"].get("new_delta_is_typing", False):
return
x_move = (
self.env["screen"]["new_cursor"]["x"]
@@ -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,
@@ -108,6 +113,40 @@ class command:
return True
def get_curl_classic_percentage(self, text):
"""Return progress from a structurally valid curl classic row."""
import re
size_field = r"\d+(?:\.\d+)?(?:[kKmMgGtT](?:i?[bB])?)?"
time_field = r"(?:(?:\d+:)?\d{2}:\d{2}|--:--:--)"
curl_classic_match = re.match(
rf"^\s*(\d+)\s+{size_field}\s+"
rf"(\d+)\s+{size_field}\s+"
rf"(\d+)\s+{size_field}\s+"
rf"{size_field}\s+{size_field}\s+"
rf"{time_field}\s+{time_field}\s+{time_field}\s+"
rf"{size_field}(?=\s|$)",
text,
)
if not curl_classic_match:
return None
total_percentage = int(curl_classic_match.group(1))
received_percentage = int(curl_classic_match.group(2))
uploaded_percentage = int(curl_classic_match.group(3))
if not all(
0 <= percentage <= 100
for percentage in (
total_percentage,
received_percentage,
uploaded_percentage,
)
):
return None
if total_percentage != received_percentage:
return None
return float(total_percentage)
def is_explicit_progress_delta(self, text):
"""Allow long single-line deltas that still look like progress output."""
import re
@@ -115,17 +154,42 @@ class command:
if "\n" in text or self.contains_url(text):
return False
if self.get_curl_classic_percentage(text) is not None:
return True
has_percentage = re.search(r"(^|\s)\d+(?:\.\d+)?\s*%", text)
if not has_percentage:
return False
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 +204,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
@@ -193,34 +266,21 @@ class command:
return
# Pattern 1a2: Curl classic progress format (percentage without % symbol)
# Extract percentage from curl's classic format
curl_classic_match = re.search(
r"^\s*(\d+)\s+\d+[kMGT]?\s+(\d+)\s+\d+[kMGT]?\s+\d+\s+\d+\s+\d+[kMGT]?\s+\d+\s+\d+:\d+:\d+\s+\d+:\d+:\d+\s+\d+:\d+:\d+\s+\d+[kMGT]?\s*$", text
)
if curl_classic_match:
# Use the first percentage (total progress)
percentage = float(curl_classic_match.group(1))
if 0 <= percentage <= 100:
percentage = self.get_curl_classic_percentage(text)
if percentage is not None:
self.env["runtime"]["DebugManager"].write_debug_out(
"found curl classic percentage: " + str(percentage),
debug.DebugLevel.INFO,
)
if percentage != self.env["commandBuffer"]["lastProgressValue"]:
self.env["runtime"]["DebugManager"].write_debug_out(
"found curl classic percentage: " + str(percentage),
"Playing tone for curl: " + str(percentage),
debug.DebugLevel.INFO,
)
if (
percentage
!= self.env["commandBuffer"]["lastProgressValue"]
):
self.env["runtime"]["DebugManager"].write_debug_out(
"Playing tone for curl: " + str(percentage),
debug.DebugLevel.INFO,
)
self.play_progress_tone(percentage)
self.env["commandBuffer"][
"lastProgressValue"
] = percentage
self.env["commandBuffer"][
"lastProgressTime"
] = current_time
return
self.play_progress_tone(percentage)
self.env["commandBuffer"]["lastProgressValue"] = percentage
self.env["commandBuffer"]["lastProgressTime"] = current_time
return
# Pattern 1b: Time/token activity (not percentage-based, so use single
# beep)
@@ -240,8 +300,20 @@ class command:
pacman_match = re.search(
r"\d+(?:\.\d+)?\s+[kKmMgGtT]iB\s+\d+(?:\.\d+)?\s+[kKmMgGtT]iB/s\s+\d+:\d+", text
)
# Pattern 1g: Key/value status lines from ffmpeg-like terminal tools.
# These do not expose a percentage in the status line, so treat them as
# activity rather than guessing progress from elapsed time.
status_fields_match = self.has_progress_status_fields(text)
if time_match or token_match or dd_match or curl_match or transfer_match or pacman_match:
if (
time_match
or token_match
or dd_match
or curl_match
or transfer_match
or pacman_match
or status_fields_match
):
# For non-percentage progress, use a single activity beep every 2
# seconds
if (
@@ -577,5 +649,48 @@ class command:
return True
return False
def has_progress_status_fields(self, text):
"""Recognize dense key=value status lines from terminal tools."""
import re
if "\n" in text or self.contains_url(text):
return False
fields = re.findall(r"\b([a-zA-Z][a-zA-Z0-9_-]*)=\s*\S+", text)
if len(fields) < 3:
return False
field_names = {field.lower() for field in fields}
progress_fields = {
"frame",
"fps",
"size",
"time",
"bitrate",
"speed",
"elapsed",
"dup",
"drop",
}
if len(field_names & progress_fields) < 3:
return False
has_time_field = bool(
re.search(
r"\b(?:time|elapsed|eta)=\s*\d+(?::\d{2}){1,2}(?:\.\d+)?\b",
text,
re.IGNORECASE,
)
)
has_rate_field = bool(
re.search(
r"\b(?:speed|fps|bitrate)=\s*[\d.]+[a-zA-Z/]*\b",
text,
re.IGNORECASE,
)
)
return has_time_field and has_rate_field
def set_callback(self, callback):
pass
@@ -141,7 +141,12 @@ class command(config_command):
self.config["general"] = {
"punctuation_level": "some",
"debug_level": "0",
}
# Basic clipboard defaults
self.config["clipboard"] = {
"number_of_clipboards": "50",
"clipboard_export_path": "/tmp/fenrirClipboard",
}
# Write the configuration
@@ -0,0 +1,171 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
import os
import threading
import time
from fenrirscreenreader.core import debug
from fenrirscreenreader.utils import x_clipboard
class ClipboardSyncManager:
def __init__(self):
self.env = None
self.enabled = False
self.display = ""
self.interval = 0.5
self.running = False
self.thread = None
self.last_written_to_x = None
self.last_imported_from_x = None
self.last_observed_fenrir = None
self.last_observed_x = None
def initialize(self, environment):
self.env = environment
self.enabled, self.display = self._resolve_startup()
self.interval = max(
0.1,
self.env["runtime"]["SettingsManager"].get_setting_as_float(
"clipboard", "sync_interval"
),
)
if self.enabled:
self._debug(
"Clipboard sync enabled for display "
+ self.display
+ " at interval "
+ str(self.interval)
)
self.start()
else:
self._debug("Clipboard sync disabled at startup")
def shutdown(self):
self.stop()
def mark_written_to_x(self, text):
if not isinstance(text, str) or not text:
return
self.last_written_to_x = text
self.last_observed_fenrir = text
self.last_observed_x = text
def _resolve_startup(self):
settings_manager = self.env["runtime"]["SettingsManager"]
if not settings_manager.get_setting_as_bool(
"clipboard", "sync_enabled"
):
return False, ""
configured_display = settings_manager.get_setting(
"clipboard", "sync_display"
)
screen_driver = settings_manager.get_setting("screen", "driver")
keyboard_driver = settings_manager.get_setting("keyboard", "driver")
is_x_mode = (
screen_driver == "ptyDriver" and keyboard_driver == "x11Driver"
)
if configured_display:
return True, configured_display
if is_x_mode:
current_display = os.environ.get("DISPLAY", "")
return bool(current_display), current_display
return False, ""
def start(self):
if self.running:
return
self.running = True
self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start()
def stop(self):
self.running = False
if self.thread:
self.thread.join(timeout=1.0)
self.thread = None
def _run(self):
while self.running:
self.poll_once()
time.sleep(self.interval)
def poll_once(self):
fenrir_text = self._get_fenrir_clipboard_text()
x_text = self._get_x_clipboard_text()
if fenrir_text and x_text and fenrir_text == x_text:
self.last_observed_fenrir = fenrir_text
self.last_observed_x = x_text
return
fenrir_changed = (
fenrir_text
and fenrir_text != self.last_observed_fenrir
)
x_changed = (
x_text
and x_text != self.last_observed_x
)
if fenrir_changed:
if self._write_x_clipboard_text(fenrir_text):
self.last_written_to_x = fenrir_text
self.last_observed_x = fenrir_text
self.last_observed_fenrir = fenrir_text
return
if x_changed:
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
"clipboardHistory", x_text
)
self.last_imported_from_x = x_text
self.last_observed_fenrir = x_text
self.last_observed_x = x_text
return
if fenrir_text:
self.last_observed_fenrir = fenrir_text
if x_text:
self.last_observed_x = x_text
def _get_fenrir_clipboard_text(self):
memory_manager = self.env["runtime"]["MemoryManager"]
if memory_manager.is_index_list_empty("clipboardHistory"):
return None
text = memory_manager.get_index_list_element("clipboardHistory")
if isinstance(text, str) and text:
return text
return None
def _get_x_clipboard_text(self):
try:
text = x_clipboard.read_text(self.display)
if isinstance(text, str) and text:
return text
return None
except Exception as error:
self.env["runtime"]["DebugManager"].write_debug_out(
"ClipboardSyncManager paste failed: " + str(error),
debug.DebugLevel.INFO,
)
return None
def _write_x_clipboard_text(self, text):
try:
return x_clipboard.write_text(text, self.display)
except Exception as error:
self._debug("ClipboardSyncManager copy failed: " + str(error))
return False
def _debug(self, message):
self.env["runtime"]["DebugManager"].write_debug_out(
message,
debug.DebugLevel.INFO,
)
@@ -22,4 +22,7 @@ command_info = {
# 'curr_command': '',
"lastCommandExecutionTime": time.time(),
"lastCommandRequestTime": time.time(),
"lastCommand": "",
"lastCommandSection": "",
"lastCommandRunTime": 0,
}
@@ -474,6 +474,10 @@ class CommandManager:
def run_command(self, command, section="commands"):
if self.command_exists(command, section):
try:
command_time = time.time()
self.env["commandInfo"]["lastCommand"] = command
self.env["commandInfo"]["lastCommandSection"] = section
self.env["commandInfo"]["lastCommandRunTime"] = command_time
self.env["runtime"]["DebugManager"].write_debug_out(
"run_command command:" + section + "." + command,
debug.DebugLevel.INFO,
@@ -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
@@ -17,6 +17,7 @@ general_data = {
"CursorManager",
"ApplicationManager",
"CommandManager",
"ClipboardSyncManager",
"ScreenManager",
"InputManager",
"OutputManager",
+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(
+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}"
@@ -12,6 +12,7 @@ import time
INSTANCE_TIMEOUT_SEC = 30.0
REGISTRY_DIR_MODE = 0o1777
def get_registry_dir():
@@ -26,8 +27,8 @@ def get_instance_file(pid=None):
def write_instance(instance_data):
registry_dir = get_registry_dir()
os.makedirs(registry_dir, mode=0o755, exist_ok=True)
os.chmod(registry_dir, 0o755)
os.makedirs(registry_dir, mode=REGISTRY_DIR_MODE, exist_ok=True)
os.chmod(registry_dir, REGISTRY_DIR_MODE)
prune_stale_instances()
instance_data["updated_at"] = time.time()
instance_path = get_instance_file(instance_data.get("pid"))
+1 -1
View File
@@ -422,7 +422,7 @@ class RemoteManager:
def export_clipboard(self):
clipboard_file_path = self.env["runtime"][
"SettingsManager"
].get_setting("general", "clipboard_export_path")
].get_setting("clipboard", "clipboard_export_path")
clipboard_file_path = clipboard_file_path.replace(
"$user", self.env["general"]["curr_user"]
)
@@ -14,6 +14,7 @@ runtime_data = {
"RemoteDriver": None,
"InputManager": None,
"CommandManager": None,
"ClipboardSyncManager": None,
"ScreenManager": None,
"OutputManager": None,
"SpeechHistoryManager": None,
+18 -5
View File
@@ -12,8 +12,13 @@ settings_data = {
"driver": "genericDriver",
"theme": "default",
"volume": 1.0,
"generic_play_file_command": "play -q -v fenrirVolume fenrirSoundFile",
"generic_frequency_command": "play -q -v fenrirVolume -n -c1 synth fenrirDuration sine fenrirFrequence",
"generic_play_file_command": (
"play -q -v fenrir_volume fenrir_sound_file"
),
"generic_frequency_command": (
"play -q -v fenrir_volume -n -c1 synth "
"fenrir_duration sine fenrir_frequency"
),
"progress_monitoring": True,
},
"speech": {
@@ -38,7 +43,10 @@ settings_data = {
"batch_flush_interval": 0.5,
"max_batch_lines": 100,
"flood_line_threshold": 500,
"generic_speech_command": 'espeak -a fenrirVolume -s fenrirRate -p fenrirPitch -v fenrirVoice "fenrirText"',
"generic_speech_command": (
"espeak -a fenrir_volume -s fenrir_rate -p fenrir_pitch "
'-v fenrir_voice "fenrir_text"'
),
"fenrir_min_volume": 0,
"fenrir_max_volume": 200,
"fenrir_min_pitch": 0,
@@ -62,8 +70,6 @@ settings_data = {
"respect_punctuation_pause": True,
"replace_undefined_punctuation_with_space": True,
"new_line_pause": True,
"number_of_clipboards": 10,
"clipboard_export_path": "/tmp/fenrirClipboard",
"emoticons": True,
"fenrir_keys": "KEY_KP0,KEY_META",
"script_keys": "KEY_COMPOSE",
@@ -81,6 +87,13 @@ settings_data = {
"diff_presentation": "both",
"diff_verbosity": "compact",
},
"clipboard": {
"number_of_clipboards": 10,
"clipboard_export_path": "/tmp/fenrirClipboard",
"sync_enabled": False,
"sync_display": "",
"sync_interval": 0.5,
},
"focus": {
"cursor": True,
"highlight": False,
@@ -12,6 +12,7 @@ from configparser import ConfigParser
from fenrirscreenreader.core import applicationManager
from fenrirscreenreader.core import attributeManager
from fenrirscreenreader.core import barrierManager
from fenrirscreenreader.core import clipboardSyncManager
from fenrirscreenreader.core import commandManager
from fenrirscreenreader.core import cursorManager
from fenrirscreenreader.core import debug
@@ -520,6 +521,7 @@ class SettingsManager:
"speechdDriver",
"genericDriver",
"dectalkDriver",
"dectalkUsbDriver",
"doubletalkDriver",
"litetalkDriver",
"tripletalkDriver",
@@ -771,6 +773,13 @@ class SettingsManager:
] = commandManager.CommandManager()
environment["runtime"]["CommandManager"].initialize(environment)
environment["runtime"][
"ClipboardSyncManager"
] = clipboardSyncManager.ClipboardSyncManager()
environment["runtime"]["ClipboardSyncManager"].initialize(
environment
)
environment["runtime"]["HelpManager"] = helpManager.HelpManager()
environment["runtime"]["HelpManager"].initialize(environment)
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
version = "2026.06.01"
version = "2026.08.06"
code_name = "master"
@@ -114,6 +114,17 @@ class driver(remoteDriver):
}
remoteInstanceRegistry.write_instance(instance_data)
def _try_register_instance(self):
try:
self._register_instance()
return True
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"unixDriver could not register instance: " + str(e),
debug.DebugLevel.ERROR,
)
return False
def _socket_file_matches_socket(self, fenrir_sock, socket_file):
try:
if os.stat(socket_file).st_uid != os.getuid():
@@ -150,6 +161,23 @@ class driver(remoteDriver):
except OSError:
return 0
def _get_peer_uid(self, client_sock):
so_peercred = getattr(socket, "SO_PEERCRED", 17)
try:
creds = client_sock.getsockopt(
socket.SOL_SOCKET, so_peercred, struct.calcsize("3i")
)
_pid, uid, _gid = struct.unpack("3i", creds)
return uid
except OSError:
return -1
def _get_socket_owner_uid(self, socket_file):
try:
return os.stat(socket_file).st_uid
except OSError:
return -1
def _get_parent_pid(self, pid):
try:
with open(f"/proc/{pid}/stat", "r", encoding="utf-8") as proc_file:
@@ -195,13 +223,20 @@ class driver(remoteDriver):
socket_files.append(socket_file)
return socket_files
def _find_available_private_socket(self, preferred_socket=""):
def _find_available_private_socket(
self, preferred_socket="", owner_uid=None
):
socket_files = self._get_registered_private_sockets()
if preferred_socket and preferred_socket in socket_files:
socket_files.remove(preferred_socket)
socket_files.insert(0, preferred_socket)
for socket_file in socket_files:
if (
owner_uid is not None
and self._get_socket_owner_uid(socket_file) != owner_uid
):
continue
if self._is_own_socket_file(socket_file):
return socket_file
if self._is_socket_active(socket_file):
@@ -242,14 +277,27 @@ class driver(remoteDriver):
def _route_main_socket_command(self, data, client_sock, socket_file):
if socket_file != MAIN_SOCKET_FILE:
return False
owner_uid = None
if not self._has_own_private_socket():
return False
peer_uid = self._get_peer_uid(client_sock)
main_socket_uid = self._get_socket_owner_uid(MAIN_SOCKET_FILE)
if peer_uid < 0 or peer_uid == main_socket_uid:
return False
owner_uid = peer_uid
peer_pid = self._get_peer_pid(client_sock)
ancestor_socket = ""
if peer_pid > 1:
ancestor_socket = self._find_ancestor_private_socket(peer_pid)
target_socket = self._find_available_private_socket(ancestor_socket)
if owner_uid is None:
target_socket = self._find_available_private_socket(
ancestor_socket
)
else:
target_socket = self._find_available_private_socket(
ancestor_socket, owner_uid=owner_uid
)
if not target_socket:
return False
@@ -318,48 +366,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._register_instance()
last_register = time.time()
while active.value:
if time.time() - last_register > 10.0:
self._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()
@@ -304,6 +304,11 @@ class driver(screenDriver):
"keyboard", "interrupt_on_key_press_filter"
).strip():
return
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver interrupt_output_on_stdin_input: "
+ repr(msg_bytes),
debug.DebugLevel.INFO,
)
self.start_stdin_interrupt_thread()
def start_stdin_interrupt_thread(self):
@@ -335,10 +340,43 @@ class driver(screenDriver):
return
if self.handle_vmenu_stdin_input(msg_bytes, event_queue):
return
if self.suppress_recent_review_stdin_tail(msg_bytes):
return
self.record_stdin_keypress(msg_bytes)
self.interrupt_output_on_stdin_input(msg_bytes)
self.inject_text_to_screen(msg_bytes)
def suppress_recent_review_stdin_tail(self, msg_bytes):
if not self.is_keyboard_control_sequence(msg_bytes):
return False
try:
command_info = self.env["commandInfo"]
last_command = command_info.get("lastCommand", "")
last_section = command_info.get("lastCommandSection", "")
last_run_time = command_info.get("lastCommandRunTime", 0)
except Exception:
return False
if last_section != "commands" or not last_command.startswith("REVIEW_"):
return False
if time.time() - last_run_time > 0.8:
return False
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver suppressing recent review stdin tail: "
+ repr(msg_bytes),
debug.DebugLevel.INFO,
)
return True
def is_keyboard_control_sequence(self, msg_bytes):
if not msg_bytes:
return False
if msg_bytes.startswith(b"\x1b"):
return True
if len(msg_bytes) == 1:
value = msg_bytes[0]
return value < 32 and msg_bytes not in [b"\t", b"\n", b"\r"]
return False
def handle_vmenu_stdin_input(self, msg_bytes, event_queue):
if not self.is_vmenu_active():
return False
@@ -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):
@@ -26,15 +26,15 @@ class driver(sound_driver):
Attributes:
proc: Currently running subprocess for sound playback
soundFileCommand (str): Command template for playing sound files
frequenceCommand (str): Command template for generating frequencies
sound_file_command (str): Command template for playing sound files
frequency_command (str): Command template for generating frequencies
"""
def __init__(self):
sound_driver.__init__(self)
self.proc = None
self.soundType = ""
self.soundFileCommand = ""
self.frequenceCommand = ""
self.sound_type = ""
self.sound_file_command = ""
self.frequency_command = ""
def initialize(self, environment):
"""Initialize the generic sound driver.
@@ -46,17 +46,20 @@ class driver(sound_driver):
environment: Fenrir environment dictionary with settings
"""
self.env = environment
self.soundFileCommand = self.env["runtime"][
self.sound_file_command = self.env["runtime"][
"SettingsManager"
].get_setting("sound", "generic_play_file_command")
self.frequenceCommand = self.env["runtime"][
self.frequency_command = self.env["runtime"][
"SettingsManager"
].get_setting("sound", "generic_frequency_command")
if self.soundFileCommand == "":
self.soundFileCommand = "play -q -v fenrirVolume fenrirSoundFile"
if self.frequenceCommand == "":
self.frequenceCommand = (
"play -q -v fenrirVolume -n -c1 synth fenrirDuration sine fenrirFrequence"
if self.sound_file_command == "":
self.sound_file_command = (
"play -q -v fenrir_volume fenrir_sound_file"
)
if self.frequency_command == "":
self.frequency_command = (
"play -q -v fenrir_volume -n -c1 synth "
"fenrir_duration sine fenrir_frequency"
)
self._initialized = True
@@ -75,22 +78,22 @@ class driver(sound_driver):
return
if interrupt:
self.cancel()
popen_frequence_command = shlex.split(self.frequenceCommand)
for idx, word in enumerate(popen_frequence_command):
popen_frequency_command = shlex.split(self.frequency_command)
for idx, word in enumerate(popen_frequency_command):
word = word.replace(
"fenrirVolume", str(self.volume * adjust_volume)
"fenrir_volume", str(self.volume * adjust_volume)
)
word = word.replace("fenrirDuration", str(duration))
word = word.replace("fenrirFrequence", str(frequence))
popen_frequence_command[idx] = word
word = word.replace("fenrir_duration", str(duration))
word = word.replace("fenrir_frequency", str(frequence))
popen_frequency_command[idx] = word
self.proc = subprocess.Popen(
popen_frequence_command,
popen_frequency_command,
stdin=None,
stdout=None,
stderr=None,
shell=False,
)
self.soundType = "frequence"
self.sound_type = "frequence"
def play_sound_file(self, file_path, interrupt=True):
"""Play a sound file.
@@ -108,13 +111,13 @@ class driver(sound_driver):
if not os.path.isfile(file_path) or ".." in file_path:
return
popen_sound_file_command = shlex.split(self.soundFileCommand)
popen_sound_file_command = shlex.split(self.sound_file_command)
for idx, word in enumerate(popen_sound_file_command):
word = word.replace("fenrirVolume", str(self.volume))
word = word.replace("fenrirSoundFile", shlex.quote(str(file_path)))
word = word.replace("fenrir_volume", str(self.volume))
word = word.replace("fenrir_sound_file", str(file_path))
popen_sound_file_command[idx] = word
self.proc = subprocess.Popen(popen_sound_file_command, shell=False)
self.soundType = "file"
self.sound_type = "file"
def cancel(self):
"""Cancel currently playing sound.
@@ -123,9 +126,9 @@ class driver(sound_driver):
"""
if not self._initialized:
return
if self.soundType == "":
if self.sound_type == "":
return
if self.soundType == "file":
if self.sound_type == "file":
self.proc.kill()
try:
# Wait for process to finish to prevent zombies
@@ -134,7 +137,7 @@ class driver(sound_driver):
pass # Process already terminated
except Exception as e:
pass # Handle any other wait errors
if self.soundType == "frequence":
if self.sound_type == "frequence":
self.proc.kill()
try:
# Wait for process to finish to prevent zombies
@@ -143,4 +146,4 @@ class driver(sound_driver):
pass # Process already terminated
except Exception as e:
pass # Handle any other wait errors
self.soundType = ""
self.sound_type = ""
@@ -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")
@@ -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
@@ -30,49 +30,52 @@ class driver(speech_driver):
def __init__(self):
speech_driver.__init__(self)
self.proc = None
self.speechThread = Thread(target=self.worker)
self.speech_thread = Thread(target=self.worker)
self.lock = Lock()
self.textQueue = SpeakQueue()
self.text_queue = SpeakQueue()
def initialize(self, environment):
self.env = environment
self.minVolume = self.env["runtime"][
self.min_volume = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_min_volume")
self.maxVolume = self.env["runtime"][
self.max_volume = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_max_volume")
self.minPitch = self.env["runtime"][
self.min_pitch = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_min_pitch")
self.maxPitch = self.env["runtime"][
self.max_pitch = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_max_pitch")
self.minRate = self.env["runtime"][
self.min_rate = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_min_rate")
self.maxRate = self.env["runtime"][
self.max_rate = self.env["runtime"][
"SettingsManager"
].get_setting_as_int("speech", "fenrir_max_rate")
self.speechCommand = self.env["runtime"][
self.speech_command = self.env["runtime"][
"SettingsManager"
].get_setting("speech", "generic_speech_command")
if self.speechCommand == "":
self.speechCommand = 'espeak -a fenrirVolume -s fenrirRate -p fenrirPitch -v fenrirVoice -- "fenrirText"'
if self.speech_command == "":
self.speech_command = (
"espeak -a fenrir_volume -s fenrir_rate -p fenrir_pitch "
'-v fenrir_voice -- "fenrir_text"'
)
if False: # for debugging overwrite here
# self.speechCommand = 'spd-say --wait -r 100 -i 100 "fenrirText"'
self.speechCommand = 'flite -t "fenrirText"'
# self.speech_command = 'spd-say --wait -r 100 -i 100 "fenrir_text"'
self.speech_command = 'flite -t "fenrir_text"'
self._is_initialized = True
if self._is_initialized:
self.speechThread.start()
self.speech_thread.start()
def shutdown(self):
if not self._is_initialized:
return
self.cancel()
self.textQueue.put(-1)
self.text_queue.put(-1)
def speak(self, text, queueable=True, ignore_punctuation=False):
if not self._is_initialized:
@@ -88,7 +91,7 @@ class driver(speech_driver):
"language": self.language,
"voice": self.voice,
}
self.textQueue.put(utterance.copy())
self.text_queue.put(utterance.copy())
def cancel(self):
if not self._is_initialized:
@@ -129,7 +132,7 @@ class driver(speech_driver):
def clear_buffer(self):
if not self._is_initialized:
return
self.textQueue.clear()
self.text_queue.clear()
def set_voice(self, voice):
if not self._is_initialized:
@@ -140,13 +143,13 @@ class driver(speech_driver):
if not self._is_initialized:
return
self.pitch = str(
self.minPitch + pitch * (self.maxPitch - self.minPitch)
self.min_pitch + pitch * (self.max_pitch - self.min_pitch)
)
def set_rate(self, rate):
if not self._is_initialized:
return
self.rate = str(self.minRate + rate * (self.maxRate - self.minRate))
self.rate = str(self.min_rate + rate * (self.max_rate - self.min_rate))
def set_module(self, module):
if not self._is_initialized:
@@ -162,12 +165,29 @@ class driver(speech_driver):
if not self._is_initialized:
return
self.volume = str(
self.minVolume + volume * (self.maxVolume - self.minVolume)
self.min_volume + volume * (self.max_volume - self.min_volume)
)
def _build_speech_command(self, utterance):
replacements = {
"fenrir_volume": str(utterance["volume"]),
"fenrir_module": str(utterance["module"]),
"fenrir_language": str(utterance["language"]),
"fenrir_voice": str(utterance["voice"]),
"fenrir_pitch": str(utterance["pitch"]),
"fenrir_rate": str(utterance["rate"]),
"fenrir_text": shlex.quote(str(utterance["text"])),
}
speech_command = shlex.split(self.speech_command)
for idx, word in enumerate(speech_command):
for placeholder, value in replacements.items():
word = word.replace(placeholder, value)
speech_command[idx] = word
return speech_command
def worker(self):
while True:
utterance = self.textQueue.get()
utterance = self.text_queue.get()
if isinstance(utterance, int):
if utterance == -1:
@@ -209,21 +229,7 @@ class driver(speech_driver):
if not isinstance(utterance["rate"], str):
utterance["rate"] = ""
popen_speech_command = shlex.split(self.speechCommand)
for idx, word in enumerate(popen_speech_command):
word = word.replace("fenrirVolume", str(utterance["volume"]))
word = word.replace("fenrirModule", str(utterance["module"]))
word = word.replace(
"fenrirLanguage", str(utterance["language"])
)
word = word.replace("fenrirVoice", str(utterance["voice"]))
word = word.replace("fenrirPitch", str(utterance["pitch"]))
word = word.replace("fenrirRate", str(utterance["rate"]))
# Properly quote text to prevent command injection
word = word.replace(
"fenrirText", shlex.quote(str(utterance["text"]))
)
popen_speech_command[idx] = word
popen_speech_command = self._build_speech_command(utterance)
try:
self.env["runtime"]["DebugManager"].write_debug_out(
@@ -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}"
@@ -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
+140
View File
@@ -0,0 +1,140 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
import os
import shutil
import subprocess
import threading
_display_lock = threading.RLock()
_CLIPBOARD_TIMEOUT = 2.0
_ENCODING = "utf-8"
def _command_exists(command):
return shutil.which(command) is not None
def _env_for_display(display):
env = os.environ.copy()
if display:
env["DISPLAY"] = display
return env
def _display_candidates(display="", scan_displays=False):
if display:
return [display]
current_display = os.environ.get("DISPLAY", "")
candidates = []
if current_display:
candidates.append(current_display)
if scan_displays:
candidates.extend(f":{index}" for index in range(10))
if not candidates:
candidates.append("")
return list(dict.fromkeys(candidates))
def _read_commands(display):
commands = []
if display:
if _command_exists("xclip"):
commands.append(["xclip", "-selection", "clipboard", "-o"])
if _command_exists("xsel"):
commands.append(["xsel", "-b", "-o"])
elif os.environ.get("WAYLAND_DISPLAY") and _command_exists("wl-paste"):
commands.append(["wl-paste", "-n", "-t", "text"])
return commands
def _write_commands(display):
commands = []
if display:
if _command_exists("xclip"):
commands.append(["xclip", "-selection", "clipboard"])
if _command_exists("xsel"):
commands.append(["xsel", "-b", "-i"])
elif os.environ.get("WAYLAND_DISPLAY") and _command_exists("wl-copy"):
commands.append(["wl-copy"])
return commands
def _run_command(command, display, input_text=None):
input_bytes = None
if input_text is not None:
input_bytes = input_text.encode(_ENCODING)
return subprocess.run(
command,
input=input_bytes,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=_env_for_display(display),
timeout=_CLIPBOARD_TIMEOUT,
check=False,
)
def _command_error(command, result):
stderr = result.stderr.decode(_ENCODING, errors="replace").strip()
if stderr:
return RuntimeError(stderr)
return RuntimeError(
"clipboard command failed: "
+ " ".join(command)
+ " exited with "
+ str(result.returncode)
)
def read_text(display="", scan_displays=False):
last_error = None
command_succeeded = False
with _display_lock:
for candidate in _display_candidates(display, scan_displays):
commands = _read_commands(candidate)
if not commands:
last_error = RuntimeError("no supported clipboard reader found")
for command in commands:
try:
result = _run_command(command, candidate)
if result.returncode != 0:
last_error = _command_error(command, result)
continue
command_succeeded = True
text = result.stdout.decode(_ENCODING)
if text:
return text
except Exception as error:
last_error = error
if last_error:
if command_succeeded:
return None
raise last_error
return None
def write_text(text, display="", scan_displays=False):
if not isinstance(text, str) or not text:
return False
last_error = None
with _display_lock:
for candidate in _display_candidates(display, scan_displays):
commands = _write_commands(candidate)
if not commands:
last_error = RuntimeError("no supported clipboard writer found")
for command in commands:
try:
result = _run_command(command, candidate, text)
if result.returncode == 0:
return True
last_error = _command_error(command, result)
except Exception as error:
last_error = error
if last_error:
raise last_error
return False
+272
View File
@@ -0,0 +1,272 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.clipboardSyncManager import ClipboardSyncManager
def build_env(
sync_enabled=False,
sync_display="",
screen_driver="vcsaDriver",
keyboard_driver="evdevDriver",
fenrir_text=None,
):
def get_setting(section, setting):
values = {
("clipboard", "sync_display"): sync_display,
("screen", "driver"): screen_driver,
("keyboard", "driver"): keyboard_driver,
}
return values[(section, setting)]
settings_manager = Mock(
get_setting_as_bool=Mock(return_value=sync_enabled),
get_setting_as_float=Mock(return_value=0.5),
get_setting=Mock(side_effect=get_setting),
)
memory_manager = Mock(
is_index_list_empty=Mock(return_value=fenrir_text is None),
get_index_list_element=Mock(return_value=fenrir_text),
add_value_to_first_index=Mock(),
)
return {
"runtime": {
"SettingsManager": settings_manager,
"MemoryManager": memory_manager,
"DebugManager": Mock(write_debug_out=Mock()),
}
}
@pytest.mark.unit
def test_clipboard_sync_disabled_by_default():
manager = ClipboardSyncManager()
manager.start = Mock()
manager.initialize(build_env(sync_enabled=False))
assert manager.enabled is False
manager.start.assert_not_called()
@pytest.mark.unit
def test_clipboard_sync_enabled_in_x_mode_with_current_display(monkeypatch):
monkeypatch.setenv("DISPLAY", ":1")
manager = ClipboardSyncManager()
manager.start = Mock()
manager.initialize(
build_env(
sync_enabled=True,
screen_driver="ptyDriver",
keyboard_driver="x11Driver",
)
)
assert manager.enabled is True
assert manager.display == ":1"
manager.start.assert_called_once_with()
@pytest.mark.unit
def test_clipboard_sync_console_skips_empty_display(monkeypatch):
monkeypatch.delenv("DISPLAY", raising=False)
manager = ClipboardSyncManager()
manager.start = Mock()
manager.initialize(build_env(sync_enabled=True))
assert manager.enabled is False
manager.start.assert_not_called()
@pytest.mark.unit
def test_clipboard_sync_console_allows_configured_display(monkeypatch):
monkeypatch.delenv("DISPLAY", raising=False)
manager = ClipboardSyncManager()
manager.start = Mock()
manager.initialize(build_env(sync_enabled=True, sync_display=":0"))
assert manager.enabled is True
assert manager.display == ":0"
manager.start.assert_called_once_with()
@pytest.mark.unit
def test_mark_written_to_x_updates_observed_state():
manager = ClipboardSyncManager()
manager.mark_written_to_x("manual export")
assert manager.last_written_to_x == "manual export"
assert manager.last_observed_fenrir == "manual export"
assert manager.last_observed_x == "manual export"
@pytest.mark.unit
def test_fenrir_to_x_write_is_not_reimported(monkeypatch):
env = build_env(fenrir_text="from fenrir")
manager = ClipboardSyncManager()
manager.env = env
manager.display = ":1"
write_text = Mock(return_value=True)
read_text = Mock(side_effect=[None, "from fenrir"])
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.write_text",
write_text,
)
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text",
read_text,
)
manager.poll_once()
manager.poll_once()
write_text.assert_called_once_with("from fenrir", ":1")
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_not_called()
@pytest.mark.unit
def test_conflicting_clipboards_do_not_swap_values(monkeypatch):
env = build_env(fenrir_text="from fenrir")
manager = ClipboardSyncManager()
manager.env = env
manager.display = ":1"
write_text = Mock(return_value=True)
read_text = Mock(return_value="from x")
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.write_text",
write_text,
)
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text",
read_text,
)
manager.poll_once()
write_text.assert_called_once_with("from fenrir", ":1")
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_not_called()
assert manager.last_written_to_x == "from fenrir"
assert manager.last_observed_x == "from fenrir"
assert manager.last_observed_fenrir == "from fenrir"
@pytest.mark.unit
def test_matching_clipboards_are_observed_without_duplicate_import(monkeypatch):
env = build_env(fenrir_text="same text")
manager = ClipboardSyncManager()
manager.env = env
manager.display = ":1"
write_text = Mock(return_value=True)
read_text = Mock(return_value="same text")
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.write_text",
write_text,
)
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text",
read_text,
)
manager.poll_once()
write_text.assert_not_called()
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_not_called()
assert manager.last_observed_x == "same text"
assert manager.last_observed_fenrir == "same text"
@pytest.mark.unit
def test_x_to_fenrir_import_is_not_reexported(monkeypatch):
env = build_env(fenrir_text=None)
manager = ClipboardSyncManager()
manager.env = env
manager.display = ":1"
write_text = Mock(return_value=True)
read_text = Mock(return_value="from x")
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.write_text",
write_text,
)
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text",
read_text,
)
manager.poll_once()
env["runtime"]["MemoryManager"].is_index_list_empty.return_value = False
env["runtime"]["MemoryManager"].get_index_list_element.return_value = (
"from x"
)
manager.poll_once()
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_called_once_with(
"clipboardHistory", "from x"
)
write_text.assert_not_called()
@pytest.mark.unit
def test_selected_imported_text_can_export_after_x_changes(monkeypatch):
env = build_env(fenrir_text=None)
manager = ClipboardSyncManager()
manager.env = env
manager.display = ":1"
write_text = Mock(return_value=True)
read_text = Mock(side_effect=["from x", "other x"])
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.write_text",
write_text,
)
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text",
read_text,
)
manager.poll_once()
env["runtime"]["MemoryManager"].is_index_list_empty.return_value = False
env["runtime"]["MemoryManager"].get_index_list_element.return_value = (
"from x"
)
manager.last_observed_fenrir = "different fenrir entry"
manager.poll_once()
write_text.assert_called_once_with("from x", ":1")
@pytest.mark.unit
@pytest.mark.parametrize("x_value", [None, "", object()])
def test_x_clipboard_non_text_values_are_ignored(monkeypatch, x_value):
env = build_env(fenrir_text=None)
manager = ClipboardSyncManager()
manager.env = env
manager.display = ":1"
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text",
Mock(return_value=x_value),
)
manager.poll_once()
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_not_called()
@pytest.mark.unit
def test_x_clipboard_paste_exception_is_ignored(monkeypatch):
env = build_env(fenrir_text=None)
manager = ClipboardSyncManager()
manager.env = env
manager.display = ":1"
monkeypatch.setattr(
"fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text",
Mock(side_effect=RuntimeError("no text target")),
)
manager.poll_once()
env["runtime"]["MemoryManager"].add_value_to_first_index.assert_not_called()
env["runtime"]["DebugManager"].write_debug_out.assert_called_once()
+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
+114
View File
@@ -0,0 +1,114 @@
from configparser import ConfigParser
from pathlib import Path
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.settingsData import settings_data
from fenrirscreenreader.soundDriver import genericDriver
class SettingsManager:
def __init__(self, settings):
self.settings = settings
def get_setting(self, section, setting):
return self.settings[setting]
def _sound_driver(settings):
sound_driver = genericDriver.driver()
sound_driver.initialize(
{"runtime": {"SettingsManager": SettingsManager(settings)}}
)
sound_driver.set_volume(0.75)
return sound_driver
def _configured_sound_settings():
config = ConfigParser(interpolation=None)
config.read(Path(__file__).parents[2] / "config/settings/settings.conf")
return {
"generic_play_file_command": config.get(
"sound", "generic_play_file_command"
),
"generic_frequency_command": config.get(
"sound", "generic_frequency_command"
),
}
def test_play_frequency_replaces_snake_case_placeholders(monkeypatch):
popen = Mock()
monkeypatch.setattr(genericDriver.subprocess, "Popen", popen)
sound_driver = _sound_driver(
{
"generic_play_file_command": "player fenrir_sound_file",
"generic_frequency_command": (
"tone --volume fenrir_volume --duration fenrir_duration "
"--frequency fenrir_frequency"
),
}
)
sound_driver.play_frequence(
440, 0.25, adjust_volume=0.5, interrupt=False
)
assert popen.call_args.args[0] == [
"tone",
"--volume",
"0.375",
"--duration",
"0.25",
"--frequency",
"440",
]
def test_play_sound_file_replaces_snake_case_placeholders(
monkeypatch, tmp_path
):
popen = Mock()
monkeypatch.setattr(genericDriver.subprocess, "Popen", popen)
sound_file = tmp_path / "sound file.ogg"
sound_file.write_bytes(b"")
sound_driver = _sound_driver(
{
"generic_play_file_command": (
"player --volume fenrir_volume fenrir_sound_file"
),
"generic_frequency_command": "tone fenrir_frequency",
}
)
sound_driver.play_sound_file(str(sound_file), interrupt=False)
assert popen.call_args.args[0] == [
"player",
"--volume",
"0.75",
str(sound_file),
]
@pytest.mark.parametrize(
"settings",
[settings_data["sound"], _configured_sound_settings()],
)
def test_default_sound_commands_use_supported_placeholders(
monkeypatch, tmp_path, settings
):
popen = Mock()
monkeypatch.setattr(genericDriver.subprocess, "Popen", popen)
sound_file = tmp_path / "sound.ogg"
sound_file.write_bytes(b"")
sound_driver = _sound_driver(settings)
sound_driver.play_frequence(
440, 0.25, adjust_volume=0.5, interrupt=False
)
sound_driver.play_sound_file(str(sound_file), interrupt=False)
for call in popen.call_args_list:
assert not any("fenrir_" in argument for argument in call.args[0])
+73
View File
@@ -0,0 +1,73 @@
from configparser import ConfigParser
from pathlib import Path
import pytest
from fenrirscreenreader.core.settingsData import settings_data
from fenrirscreenreader.speechDriver import genericDriver
def test_build_speech_command_replaces_snake_case_placeholders():
speech_driver = genericDriver.driver()
speech_driver.speech_command = (
"synth --volume fenrir_volume --module fenrir_module "
"--language fenrir_language --voice fenrir_voice "
"--pitch fenrir_pitch --rate fenrir_rate fenrir_text"
)
utterance = {
"volume": "150",
"module": "module-name",
"language": "en-US",
"voice": "voice-name",
"pitch": "50",
"rate": "240",
"text": "hello",
}
assert speech_driver._build_speech_command(utterance) == [
"synth",
"--volume",
"150",
"--module",
"module-name",
"--language",
"en-US",
"--voice",
"voice-name",
"--pitch",
"50",
"--rate",
"240",
"hello",
]
def _configured_speech_command():
config = ConfigParser(interpolation=None)
config.read(Path(__file__).parents[2] / "config/settings/settings.conf")
return config.get("speech", "generic_speech_command")
@pytest.mark.parametrize(
"speech_command",
[
settings_data["speech"]["generic_speech_command"],
_configured_speech_command(),
],
)
def test_default_speech_commands_use_supported_placeholders(speech_command):
speech_driver = genericDriver.driver()
speech_driver.speech_command = speech_command
utterance = {
"volume": "150",
"module": "",
"language": "en-US",
"voice": "voice-name",
"pitch": "50",
"rate": "240",
"text": "hello",
}
built_command = speech_driver._build_speech_command(utterance)
assert not any("fenrir_" in argument for argument in built_command)
+77 -6
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()
@@ -99,7 +149,7 @@ def test_litetalk_driver_writes_settings_and_cancel(serial_pair):
speech_driver.set_pitch(0.0)
speech_driver.set_volume(0.5)
speech_driver.cancel()
assert read_available(master_fd, 9) == b"\x019S\x010P\x014V\x18"
assert read_available(master_fd, 10) == b"\x019S\x010P\x014V\x18"
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)
+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()
+229
View File
@@ -140,3 +140,232 @@ def test_progress_detector_beeps_for_interruptible_status_without_ellipsis():
command.run()
command.play_activity_beep.assert_called_once_with()
@pytest.mark.unit
def test_progress_detector_allows_long_key_value_status_delta():
progress_module = _load_progress_module()
command = progress_module.command()
sample = (
"frame=33480 fps=429 q=28.0 size= 195328KiB "
"time=00:23:16.26 bitrate=1146.0kbits/s dup=0 drop=1 "
"speed=17.9x elapsed=0:01:18.01 "
"filter=scale,format output=archive-video-final-render-pass.mkv "
"metadata=preserve-chapters-and-stream-layout"
)
command.env = {
"commandBuffer": {"progress_monitoring": True},
"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_content_text": sample,
"old_cursor": {"x": 0, "y": 0},
"new_cursor": {"x": 0, "y": 0},
},
}
assert len(sample) > 200
assert command.is_real_progress_update()
@pytest.mark.unit
def test_progress_detector_beeps_for_ffmpeg_status_line():
progress_module = _load_progress_module()
command = progress_module.command()
sample = (
"frame=33480 fps=429 q=28.0 size= 195328KiB "
"time=00:23:16.26 bitrate=1146.0kbits/s dup=0 drop=1 "
"speed=17.9x elapsed=0:01:18.01"
)
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_activity_beep = Mock()
command.run()
command.play_activity_beep.assert_called_once_with()
@pytest.mark.unit
def test_progress_detector_ignores_generic_key_value_text():
progress_module = _load_progress_module()
command = progress_module.command()
sample = "host=server mode=debug user=Username retry=3 status=ready"
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_activity_beep = Mock()
command.play_progress_tone = Mock()
command.run()
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"
)
@pytest.mark.unit
def test_progress_detector_beeps_for_curl_classic_repaint_with_trailing_text():
progress_module = _load_progress_module()
command = progress_module.command()
sample = (
"53 537.8M 53 285.8M 0 0 9.48M 0 00:56 00:30 00:26 9.94M "
"Executing cabextract -q -d /home/Username/.local/wine64/dosdevices/"
"c:/windows/syswow64 -F dx8vb.dll /home/Username/.cache/winetricks/"
"dx8vb/DX81NTger.exe"
)
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()
assert len(sample) > 200
command.run()
command.play_progress_tone.assert_called_once_with(53.0)
assert command.env["commandBuffer"]["lastProgressValue"] == 53.0
@pytest.mark.unit
def test_progress_detector_ignores_curl_like_fields_with_mismatched_percentages():
progress_module = _load_progress_module()
command = progress_module.command()
sample = (
"53 537.8M 52 285.8M 0 0 9.48M 0 00:56 00:30 00:26 9.94M "
"ordinary numeric report with enough trailing text to cross the long "
"delta filter without representing a coherent download progress row "
"or providing trustworthy percentage information"
)
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_activity_beep = Mock()
command.play_progress_tone = Mock()
assert len(sample) > 200
command.run()
command.play_activity_beep.assert_not_called()
command.play_progress_tone.assert_not_called()
+57
View File
@@ -70,6 +70,7 @@ def test_pty_stdin_input_interrupts_output_when_all_keys_interrupt_enabled():
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
"DebugManager": Mock(write_debug_out=Mock()),
}
}
@@ -254,6 +255,7 @@ def test_pty_stdin_input_honors_interrupt_disabled():
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
"DebugManager": Mock(write_debug_out=Mock()),
}
}
@@ -273,6 +275,7 @@ def test_pty_stdin_input_leaves_filtered_interrupts_to_key_events():
"runtime": {
"SettingsManager": settings_manager,
"OutputManager": output_manager,
"DebugManager": Mock(write_debug_out=Mock()),
}
}
@@ -281,6 +284,60 @@ def test_pty_stdin_input_leaves_filtered_interrupts_to_key_events():
output_manager.interrupt_output.assert_not_called()
@pytest.mark.unit
def test_pty_recent_review_escape_tail_is_consumed_without_interrupt_or_injection():
pty_driver = PtyDriver()
event_queue = Mock()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = True
settings_manager.get_setting.return_value = ""
output_manager = Mock()
pty_driver.env = {
"commandInfo": {
"lastCommand": "REVIEW_PREV_LINE",
"lastCommandSection": "commands",
"lastCommandRunTime": time.time(),
},
"input": {"curr_input": []},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"OutputManager": output_manager,
"SettingsManager": settings_manager,
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"\x1b[7~", event_queue)
output_manager.interrupt_output.assert_not_called()
pty_driver.inject_text_to_screen.assert_not_called()
@pytest.mark.unit
def test_pty_recent_review_does_not_consume_plain_text_input():
pty_driver = PtyDriver()
event_queue = Mock()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
pty_driver.env = {
"commandInfo": {
"lastCommand": "REVIEW_PREV_LINE",
"lastCommandSection": "commands",
"lastCommandRunTime": time.time(),
},
"input": {"curr_input": []},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"SettingsManager": settings_manager,
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"a", event_queue)
pty_driver.inject_text_to_screen.assert_called_once_with(b"a")
@pytest.mark.unit
def test_pty_backspace_with_fenrir_key_synthesizes_shortcut_events():
pty_driver = PtyDriver()
@@ -1,9 +1,24 @@
import json
import stat
import time
from fenrirscreenreader.core import remoteInstanceRegistry
def test_write_instance_creates_shared_sticky_registry_directory(
tmp_path, monkeypatch
):
registry_dir = tmp_path / "instances"
monkeypatch.setattr(
remoteInstanceRegistry, "get_registry_dir", lambda: str(registry_dir)
)
remoteInstanceRegistry.write_instance({"pid": 456})
mode = stat.S_IMODE(registry_dir.stat().st_mode)
assert mode == 0o1777
def test_write_instance_prunes_stale_registry_files(tmp_path, monkeypatch):
monkeypatch.setattr(
remoteInstanceRegistry, "get_registry_dir", lambda: str(tmp_path)
+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")
@@ -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"
+36
View File
@@ -344,6 +344,7 @@ def test_large_insertion_echo_speaks_pasted_cursor_text():
"old_cursor": {"x": 2, "y": 0},
"new_cursor": {"x": 13, "y": 0},
"new_delta": "hello world",
"new_delta_is_typing": True,
},
}
command = large_insertion_module.command()
@@ -359,6 +360,41 @@ def test_large_insertion_echo_speaks_pasted_cursor_text():
)
@pytest.mark.unit
def test_large_insertion_echo_ignores_incoming_cursor_progress():
large_insertion_module = _load_large_insertion_module()
output_manager = Mock(present_text=Mock())
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
input_manager = Mock()
input_manager.get_last_deepest_input.return_value = []
input_manager.get_last_event.return_value = None
screen_manager = Mock(
is_screen_change=Mock(return_value=False),
is_delta=Mock(return_value=True),
)
env = {
"runtime": {
"InputManager": input_manager,
"OutputManager": output_manager,
"ScreenManager": screen_manager,
"SettingsManager": settings_manager,
},
"screen": {
"old_cursor": {"x": 0, "y": 0},
"new_cursor": {"x": 12, "y": 0},
"new_delta": "upgrading 1/3",
"new_delta_is_typing": False,
},
}
command = large_insertion_module.command()
command.initialize(env)
command.run()
output_manager.present_text.assert_not_called()
@pytest.mark.unit
def test_large_insertion_echo_defers_recent_tab_to_tab_completion():
large_insertion_module = _load_large_insertion_module()
+172 -1
View File
@@ -1,8 +1,70 @@
import socket
import struct
from unittest.mock import Mock, mock_open, patch
import pytest
from fenrirscreenreader.remoteDriver import unixDriver
def test_watchdog_keeps_serving_when_instance_registration_fails(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
driver.fenrirSocks = [Mock()]
active = Mock(value=True)
event_queue = Mock()
def stop_after_select(*_args):
active.value = False
return ([], [], [])
with patch.object(driver, "_get_socket_candidates", return_value=[]), patch.object(
driver, "_register_instance", side_effect=PermissionError("denied")
), patch(
"fenrirscreenreader.remoteDriver.unixDriver.select.select",
side_effect=stop_after_select,
) as select_call:
driver.watch_dog(active, event_queue)
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
@@ -15,6 +77,19 @@ class FakeClientSocket:
self.sent += data
def test_get_peer_uid_from_unix_socket():
driver = unixDriver.driver()
client_sock = Mock()
client_sock.getsockopt.return_value = struct.pack("3i", 4321, 1000, 100)
assert driver._get_peer_uid(client_sock) == 1000
client_sock.getsockopt.assert_called_once_with(
socket.SOL_SOCKET,
socket.SO_PEERCRED,
struct.calcsize("3i"),
)
def test_main_socket_routes_to_ancestor_private_socket(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
@@ -88,6 +163,10 @@ def test_main_socket_handles_command_locally_without_available_target(
event_queue = Mock()
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
driver, "_get_peer_uid", return_value=1000
), patch.object(
driver, "_get_socket_owner_uid", return_value=65534
), patch.object(
driver,
"_find_available_private_socket",
return_value="",
@@ -111,7 +190,11 @@ def test_vcsa_main_socket_owner_handles_command_locally(mock_environment):
client_sock = FakeClientSocket(b"command say root")
event_queue = Mock()
with patch.object(driver, "_find_available_private_socket") as find_available:
with patch.object(driver, "_get_peer_uid", return_value=1000), patch.object(
driver, "_get_socket_owner_uid", return_value=1000
), patch.object(
driver, "_find_available_private_socket"
) as find_available:
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
@@ -125,6 +208,64 @@ def test_vcsa_main_socket_owner_handles_command_locally(mock_environment):
)
def test_vcsa_main_socket_routes_to_same_user_private_socket(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
driver.bound_sockets = [(Mock(), unixDriver.MAIN_SOCKET_FILE)]
client_sock = FakeClientSocket(b"command say x session")
event_queue = Mock()
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
driver, "_get_peer_uid", return_value=1000
), patch.object(
driver, "_get_socket_owner_uid", return_value=65534
), patch.object(
driver, "_find_ancestor_private_socket", return_value=""
), patch.object(
driver,
"_find_available_private_socket",
return_value="/tmp/fenrirscreenreader-222.sock",
) as find_available, patch.object(
driver, "_forward_remote_to_socket", return_value=True
) as forward:
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
find_available.assert_called_once_with("", owner_uid=1000)
forward.assert_called_once_with(
"command say x session", "/tmp/fenrirscreenreader-222.sock"
)
event_queue.put.assert_not_called()
def test_vcsa_main_socket_stays_local_without_peer_credentials(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
driver.bound_sockets = [(Mock(), unixDriver.MAIN_SOCKET_FILE)]
client_sock = FakeClientSocket(b"command say unknown")
event_queue = Mock()
with patch.object(driver, "_get_peer_uid", return_value=-1), patch.object(
driver, "_find_available_private_socket"
) as find_available:
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
find_available.assert_not_called()
event_queue.put.assert_called_once_with(
{
"Type": unixDriver.FenrirEventType.remote_incomming,
"data": "command say unknown",
}
)
def test_private_socket_handles_command_locally(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
@@ -221,3 +362,33 @@ def test_find_available_private_socket_skips_main_socket(
driver._find_available_private_socket()
== "/tmp/fenrirscreenreader-111.sock"
)
def test_find_available_private_socket_filters_by_owner(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
with patch(
"fenrirscreenreader.remoteDriver.unixDriver.remoteInstanceRegistry.list_instances",
return_value=[
{
"pid": 111,
"socket_files": ["/tmp/fenrirscreenreader-111.sock"],
},
{
"pid": 222,
"socket_files": ["/tmp/fenrirscreenreader-222.sock"],
},
],
), patch.object(
driver,
"_get_socket_owner_uid",
side_effect=lambda socket_file: {
"/tmp/fenrirscreenreader-111.sock": 1001,
"/tmp/fenrirscreenreader-222.sock": 1000,
}[socket_file],
), patch.object(driver, "_is_socket_active", return_value=True):
assert (
driver._find_available_private_socket(owner_uid=1000)
== "/tmp/fenrirscreenreader-222.sock"
)
+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
+90
View File
@@ -0,0 +1,90 @@
import os
import subprocess
from unittest.mock import Mock
import pytest
from fenrirscreenreader.utils import x_clipboard
def command_exists(command):
if command == "xclip":
return "/usr/bin/xclip"
return None
@pytest.mark.unit
def test_write_text_uses_display_env_without_mutating_process_env(
monkeypatch,
):
monkeypatch.delenv("DISPLAY", raising=False)
monkeypatch.setattr(x_clipboard.shutil, "which", command_exists)
run_command = Mock(
return_value=subprocess.CompletedProcess(
["xclip"], 0, stdout=b"", stderr=b""
)
)
monkeypatch.setattr(x_clipboard.subprocess, "run", run_command)
assert x_clipboard.write_text("clipboard text", ":3") is True
command = run_command.call_args.args[0]
kwargs = run_command.call_args.kwargs
assert command == ["xclip", "-selection", "clipboard"]
assert kwargs["input"] == b"clipboard text"
assert kwargs["env"]["DISPLAY"] == ":3"
assert kwargs["timeout"] == 2.0
assert os.environ.get("DISPLAY") is None
@pytest.mark.unit
def test_read_text_scans_displays_until_text_is_found(monkeypatch):
monkeypatch.delenv("DISPLAY", raising=False)
monkeypatch.setattr(x_clipboard.shutil, "which", command_exists)
displays = []
def run_command(command, **kwargs):
displays.append(kwargs["env"].get("DISPLAY"))
if kwargs["env"].get("DISPLAY") == ":1":
return subprocess.CompletedProcess(
command, 0, stdout=b"from x", stderr=b""
)
return subprocess.CompletedProcess(
command, 1, stdout=b"", stderr=b"missing display"
)
monkeypatch.setattr(x_clipboard.subprocess, "run", run_command)
assert x_clipboard.read_text(scan_displays=True) == "from x"
assert displays == [":0", ":1"]
@pytest.mark.unit
def test_read_text_empty_success_returns_none(monkeypatch):
monkeypatch.setenv("DISPLAY", ":2")
monkeypatch.setattr(x_clipboard.shutil, "which", command_exists)
monkeypatch.setattr(
x_clipboard.subprocess,
"run",
Mock(
return_value=subprocess.CompletedProcess(
["xclip"], 0, stdout=b"", stderr=b""
)
),
)
assert x_clipboard.read_text() is None
@pytest.mark.unit
def test_write_text_propagates_clipboard_timeout(monkeypatch):
monkeypatch.setattr(x_clipboard.shutil, "which", command_exists)
timeout = subprocess.TimeoutExpired(["xclip"], 2.0)
monkeypatch.setattr(
x_clipboard.subprocess,
"run",
Mock(side_effect=timeout),
)
with pytest.raises(subprocess.TimeoutExpired):
x_clipboard.write_text("clipboard text", ":3")