7 Commits

45 changed files with 1985 additions and 238 deletions
+3 -3
View File
@@ -88,12 +88,12 @@ Application-specific menus in `vmenu-profiles/KEY/{app}/`:
## Remote Control
Unix socket: `/tmp/fenrirscreenreader-deamon.sock`
Unix socket: `/tmp/fenrirscreenreader-daemon.sock`
TCP: localhost:22447
```bash
echo "command say Hello" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command say Hello" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set speech#rate=0.8" | nc localhost 22447
```
+58 -39
View File
@@ -79,7 +79,15 @@ Fenrir is a Linux screen reader. Linux is the only officially supported platform
- python-speechd
2. **genericDriver** - Generic subprocess speech driver
- espeak or espeak-ng (or any TTS command)
3. **debugDriver** - Debug speech driver for testing
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
- RPITalk gadget mode or a LiteTalk-compatible serial device
5. **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
- External DB9 serial TripleTalk devices, or USB models that expose a tty serial device
7. **debugDriver** - Debug speech driver for testing
- No dependencies
@@ -208,6 +216,7 @@ sudo /usr/share/fenrirscreenreader/tools/configure_pipewire.sh
2. **Basic Navigation**:
- **Fenrir Key**: By default `Insert`, `Keypad Insert`, or `Meta/Super` key
- **Tutorial Mode**: `Fenrir + H` to learn all commands interactively
- **Speech History**: `Fenrir + Ctrl + H` to review recent speech
- **Quit Fenrir**: `Fenrir + Q`
3. **Essential Commands**:
@@ -273,7 +282,7 @@ enable_command_remote=True # allow command execution
### Remote Drivers
1. **unixDriver** (recommended): Uses Unix domain sockets
- Socket location: `/tmp/fenrirscreenreader-deamon.sock` for the standard control socket
- Socket location: `/tmp/fenrirscreenreader-daemon.sock` for the standard control socket
- `fenrir -x` instances also create private sockets: `/tmp/fenrirscreenreader-<pid>.sock`
- More secure, local-only access
- Works with `socat`
@@ -290,99 +299,102 @@ The `socat` command provides the easiest way to send commands to Fenrir:
#### Instance Discovery
```bash
# List registered Fenrir instances and their socket paths
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
In X terminal mode (`fenrir -x`), multiple Fenrir instances can run at the same
time. Each instance has a private socket, and one instance may also own the
standard control socket. Use `ls` or `command ls` on the standard socket to find
the private socket for a specific instance. Untargeted commands sent through a
shared or broadcast path are claimed by one instance so duplicate instances do
not all perform the same action.
the private socket for a specific instance. Commands sent to the standard socket
are handled by its owner when possible; otherwise they are forwarded to a
registered private socket, preferring the sender's Fenrir ancestor when one can
be found. Untargeted commands sent through a shared or broadcast path are
claimed by one instance so duplicate instances do not all perform the same
action.
#### Basic Speech Control
```bash
# Interrupt current speech
echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Speak custom text
echo "command say Hello, this is a test message" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command say Hello, this is a test message" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Temporarily disable speech (until next keystroke)
echo "command tempdisablespeech" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command tempdisablespeech" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
#### Settings Control
```bash
# Enable highlight tracking mode
echo "setting set focus#highlight=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set focus#highlight=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Change speech parameters
echo "setting set speech#rate=0.8" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#pitch=0.6" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#volume=0.9" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#rate=0.8" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set speech#pitch=0.6" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set speech#volume=0.9" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Change punctuation level (none/some/most/all)
echo "setting set general#punctuation_level=all" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set general#punctuation_level=none" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set general#punctuation_level=all" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set general#punctuation_level=none" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Voice and TTS engine control
echo "setting set speech#voice=en-us+f3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#module=espeak-ng" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#voice=en-us+f3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set speech#module=espeak-ng" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Disable sound temporarily
echo "setting set sound#enabled=False" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set sound#volume=0.5" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set sound#enabled=False" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set sound#volume=0.5" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Keyboard and input settings
echo "setting set keyboard#char_echo_mode=1" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set keyboard#word_echo=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set keyboard#char_echo_mode=1" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set keyboard#word_echo=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Screen control (ignore specific TTYs)
echo "setting set screen#ignore_screen=1,2,3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set screen#ignore_screen=1,2,3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Multiple settings at once
echo "setting set speech#rate=0.8;sound#volume=0.7;general#punctuation_level=most" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#rate=0.8;sound#volume=0.7;general#punctuation_level=most" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Reset all settings to defaults
echo "setting reset" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting reset" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Save current settings
echo "setting save" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting saveas /tmp/my-fenrir-settings.conf" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting save" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting saveas /tmp/my-fenrir-settings.conf" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
#### Clipboard Operations
```bash
# Place text into clipboard
echo "command clipboard This text will be copied to clipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command clipboard This text will be copied to clipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Export clipboard to file
echo "command exportclipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command exportclipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
#### Window Management
```bash
# Define a window area (x1 y1 x2 y2)
echo "command window 0 0 80 24" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command window 0 0 80 24" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Reset window to full screen
echo "command resetwindow" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command resetwindow" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
#### VMenu Control
```bash
# Set virtual menu context
echo "command vmenu nano/file" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command vmenu nano/file" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Reset virtual menu
echo "command resetvmenu" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command resetvmenu" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
#### Application Control
```bash
# Quit Fenrir
echo "command quitapplication" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command quitapplication" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
### Using TCP Driver
@@ -447,7 +459,14 @@ 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)
- `speech#driver=driver_name` - Speech driver (speechdDriver/genericDriver/dectalkDriver/litetalkDriver/doubletalkDriver/tripletalkDriver)
- `speech#hardware_device=auto` - 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
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.
- `speech#auto_read_incoming=True/False` - Auto-read new text
*Sound Settings:*
@@ -576,7 +595,7 @@ Fenrir provides intelligent progress bar detection and audio feedback for variou
To enable progress monitoring:
1. Add a key binding in your keyboard layout file
2. Or use the remote control system: `echo "command progress_bar_monitor" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock`
2. Or use the remote control system: `echo "command progress_bar_monitor" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock`
### Progress Detection Patterns
@@ -637,7 +656,7 @@ Building...
```bash
# Enable progress monitoring
echo "command progress_bar_monitor" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command progress_bar_monitor" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Common scenarios where progress monitoring is useful:
wget https://example.com/large-file.zip # Download progress
@@ -661,7 +680,7 @@ Progress monitoring can be configured through settings:
#!/bin/bash
# notify_fenrir.sh - Send notifications to Fenrir
SOCKET="/tmp/fenrirscreenreader-deamon.sock"
SOCKET="/tmp/fenrirscreenreader-daemon.sock"
fenrir_say() {
echo "command say $1" | socat - UNIX-CLIENT:$SOCKET
@@ -684,7 +703,7 @@ import os
def send_fenrir_command(command):
"""Send command to Fenrir via Unix socket"""
socket_path = "/tmp/fenrirscreenreader-deamon.sock"
socket_path = "/tmp/fenrirscreenreader-daemon.sock"
if os.path.exists(socket_path):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
@@ -717,7 +736,7 @@ send_fenrir_command("setting set speech#rate=0.9")
**Commands not working:**
- Verify `enable_command_remote=True` in settings
- Check Fenrir debug logs: `/var/log/fenrir.log`
- Test with simple command: `echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock`
- Test with simple command: `echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock`
## Command Line Options
+6
View File
@@ -26,6 +26,12 @@ This directory contains keyboard layout files for Fenrir screen reader.
- **Exit review**: `Fenrir + Keypad .`
- **Screen reading**: `Fenrir + Keypad 5` (current screen)
### Speech History
- **Open speech history**: `Fenrir + Ctrl + H`
- **Navigate history**: `Up` and `Down`
- **Copy current item**: `Enter`
- **Exit speech history**: `Escape`
## Configuration
To change keyboard layout, edit `/etc/fenrirscreenreader/settings/settings.conf`
+1
View File
@@ -1,4 +1,5 @@
KEY_FENRIR,KEY_H=toggle_tutorial_mode
KEY_FENRIR,KEY_CTRL,KEY_H=speech_history
KEY_CTRL=shut_up
KEY_FENRIR,KEY_KP9=review_bottom
KEY_FENRIR,KEY_KP7=review_top
+1
View File
@@ -1,4 +1,5 @@
KEY_FENRIR,KEY_H=toggle_tutorial_mode
KEY_FENRIR,KEY_CTRL,KEY_H=speech_history
KEY_CTRL=shut_up
KEY_FENRIR,KEY_SHIFT,KEY_O=review_bottom
KEY_FENRIR,KEY_SHIFT,KEY_U=review_top
+23 -1
View File
@@ -35,9 +35,14 @@ progress_monitoring=True
# Turn speech on or off:
enabled=True
# Select speech driver, options are speechdDriver or genericDriver:
# Select speech driver, options are speechdDriver, genericDriver,
# dectalkDriver, litetalkDriver, doubletalkDriver, or tripletalkDriver:
driver=speechdDriver
#driver=genericDriver
#driver=dectalkDriver
#driver=litetalkDriver
#driver=doubletalkDriver
#driver=tripletalkDriver
# The rate selects how fast Fenrir will speak. Options range from 0, slowest, to 1.0, fastest.
rate=0.5
@@ -70,12 +75,29 @@ volume=1.0
# Select the language you want Fenrir to use.
#language=en
# Hardware speech synthesizer serial device.
# Used by dectalkDriver, litetalkDriver, doubletalkDriver, and tripletalkDriver.
# 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.
# auto checks /dev/ttyACM* first, then /dev/ttyUSB*.
# Examples:
# hardware_device=/dev/ttyACM0 # RPITalk USB gadget mode
# hardware_device=/dev/ttyUSB0 # USB serial adapter
# hardware_device=/dev/ttyS0 # built-in serial port
hardware_device=auto
# Serial baud rate for hardware speech synthesizers.
hardware_baud_rate=9600
# Read new text as it happens?
auto_read_incoming=True
# Speak individual numbers instead of whole string.
read_numbers_as_digits = False
# Number of spoken items kept in runtime speech history.
history_size=50
# Flood control: batch rapid updates instead of speaking each one
# Number of updates within rapid_update_window to trigger batching
rapid_update_threshold=5
+24 -13
View File
@@ -238,6 +238,14 @@ speechdDriver - Speech-dispatcher (recommended)
.IP \[bu] 4
genericDriver - Command-line TTS (espeak, etc.)
.IP \[bu] 4
dectalkDriver - DECtalk-compatible serial hardware speech
.IP \[bu] 4
litetalkDriver - LiteTalk-compatible serial hardware speech
.IP \[bu] 4
doubletalkDriver - DoubleTalk LT-compatible serial hardware speech
.IP \[bu] 4
tripletalkDriver - TripleTalk-compatible serial hardware speech
.IP \[bu] 4
debugDriver - Debug/testing
.TP
@@ -313,58 +321,61 @@ enable_command_remote=True
.B Instance Discovery:
.EX
# List registered Fenrir instances and their socket paths
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
.EE
In X terminal mode (fenrir -x), multiple Fenrir instances can run at the
same time. Each instance has a private socket at
/tmp/fenrirscreenreader-<pid>.sock, and one instance may also own the
standard control socket. Use ls or "command ls" on the standard socket to
find the private socket for a specific instance.
find the private socket for a specific instance. Commands sent to the standard
socket are handled by its owner when possible; otherwise they are forwarded to a
registered private socket, preferring the sender's Fenrir ancestor when one can
be found.
.TP
.B Basic Speech Control:
.EX
# Interrupt current speech
echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Speak custom text
echo "command say Hello, this is a test" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command say Hello, this is a test" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Temporarily disable speech
echo "command tempdisablespeech" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command tempdisablespeech" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
.EE
.TP
.B Settings Control:
.EX
# Enable highlight tracking
echo "setting set focus#highlight=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set focus#highlight=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Change speech rate
echo "setting set speech#rate=0.8" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#rate=0.8" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Change punctuation level (none/some/most/all)
echo "setting set general#punctuation_level=all" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set general#punctuation_level=all" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Voice and TTS control
echo "setting set speech#voice=en-us+f3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#voice=en-us+f3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Multiple settings at once
echo "setting set speech#rate=0.8;sound#volume=0.7;general#punctuation_level=most" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#rate=0.8;sound#volume=0.7;general#punctuation_level=most" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Reset all settings
echo "setting reset" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting reset" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
.EE
.TP
.B Clipboard Operations:
.EX
# Add text to clipboard
echo "command clipboard Text to copy" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command clipboard Text to copy" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Export clipboard to file
echo "command exportclipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command exportclipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
.EE
.SS Command Reference
+45 -15
View File
@@ -1278,65 +1278,68 @@ The `+socat+` command provides the easiest way to send commands to Fenrir:
....
# List registered Fenrir instances and their socket paths
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
....
In X terminal mode (`+fenrir -x+`), multiple Fenrir instances can run at the
same time. Each instance has a private socket at
`+/tmp/fenrirscreenreader-<pid>.sock+`, and one instance may also own the
standard control socket. Use `+ls+` or `+command ls+` on the standard socket to
find the private socket for a specific instance.
find the private socket for a specific instance. Commands sent to the standard
socket are handled by its owner when possible; otherwise they are forwarded to a
registered private socket, preferring the sender's Fenrir ancestor when one can
be found.
===== Basic Speech Control
....
# Interrupt current speech
echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Speak custom text
echo "command say Hello, this is a test message" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command say Hello, this is a test message" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Temporarily disable speech (until next keystroke)
echo "command tempdisablespeech" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command tempdisablespeech" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
....
===== Settings Control
....
# Enable highlight tracking mode
echo "setting set focus#highlight=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set focus#highlight=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Change speech rate
echo "setting set speech#rate=0.8" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#rate=0.8" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Change punctuation level (none/some/most/all)
echo "setting set general#punctuation_level=all" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set general#punctuation_level=all" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Voice and TTS control
echo "setting set speech#voice=en-us+f3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#voice=en-us+f3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Multiple settings at once
echo "setting set speech#rate=0.8;sound#volume=0.7;general#punctuation_level=most" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#rate=0.8;sound#volume=0.7;general#punctuation_level=most" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Reset all settings to defaults
echo "setting reset" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting reset" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
....
===== Clipboard Operations
....
# Place text into clipboard
echo "command clipboard This text will be copied to clipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command clipboard This text will be copied to clipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Export clipboard to file
echo "command exportclipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command exportclipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
....
===== Application Control
....
# Quit Fenrir
echo "command quitapplication" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command quitapplication" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
....
==== Command Reference
@@ -1548,8 +1551,12 @@ enabled=True
Values: on=`+True+`, off=`+False+`
# Select speech driver, options are speechdDriver (default),
genericDriver or espeakDriver: driver=speechdDriver #driver=espeakDriver
genericDriver, dectalkDriver, litetalkDriver, doubletalkDriver or tripletalkDriver: driver=speechdDriver
#driver=genericDriver
#driver=dectalkDriver
#driver=litetalkDriver
#driver=doubletalkDriver
#driver=tripletalkDriver
This Selects the driver used to generate speech output.
@@ -1677,6 +1684,29 @@ the pico module:
language=de-DE
....
Hardware speech drivers use a serial device. The default `+auto+` checks
`+/dev/ttyACM*+` first, then `+/dev/ttyUSB*+`. Set an explicit path for
stable systems.
....
hardware_device=auto
hardware_device=/dev/ttyACM0
hardware_device=/dev/ttyUSB0
hardware_device=/dev/ttyS0
....
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.
Read new text as it occurs auto_read_incoming=True Values: on=`+True+`,
off=`+False+`
+40 -21
View File
@@ -38,6 +38,7 @@ Navigate the screen without moving the text cursor. Essential for examining cont
### Navigation (Desktop Layout)
- `Ctrl` - Stop speech (shut up)
- `Fenrir + H` - Tutorial mode
- `Fenrir + Ctrl + H` - Speech history
- `Fenrir + Q` - Quit Fenrir
- `Fenrir + Keypad 5` - Read current screen
- `Keypad 8` - Read current line
@@ -100,6 +101,9 @@ driver=speechdDriver
rate=0.5
pitch=0.5
volume=1.0
hardware_device=auto
hardware_baud_rate=9600
history_size=50
[sound]
enabled=True
@@ -141,68 +145,70 @@ enable_command_remote=True # allow command execution
#### Instance Discovery
```bash
# List registered Fenrir instances and their socket paths
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "ls" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
In X terminal mode (`fenrir -x`), multiple Fenrir instances can run at the same
time. Each instance has a private socket at `/tmp/fenrirscreenreader-<pid>.sock`,
and one instance may also own the standard control socket. Use `ls` or
`command ls` on the standard socket to find the private socket for a specific
instance.
instance. Commands sent to the standard socket are handled by its owner when
possible; otherwise they are forwarded to a registered private socket,
preferring the sender's Fenrir ancestor when one can be found.
#### Speech Control
```bash
# Interrupt current speech
echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command interrupt" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Speak custom text
echo "command say Hello, this is a test" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command say Hello, this is a test" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Temporarily disable speech
echo "command tempdisablespeech" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command tempdisablespeech" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
#### Settings Control
```bash
# Enable highlight tracking
echo "setting set focus#highlight=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set focus#highlight=True" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Change speech parameters
echo "setting set speech#rate=0.8" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#pitch=0.6" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#volume=0.9" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#rate=0.8" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set speech#pitch=0.6" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set speech#volume=0.9" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Change punctuation level (none/some/most/all)
echo "setting set general#punctuation_level=all" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set general#punctuation_level=all" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Voice and TTS control
echo "setting set speech#voice=en-us+f3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#module=espeak-ng" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#voice=en-us+f3" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting set speech#module=espeak-ng" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Multiple settings at once
echo "setting set speech#rate=0.8;sound#volume=0.7;general#punctuation_level=most" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting set speech#rate=0.8;sound#volume=0.7;general#punctuation_level=most" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Reset all settings
echo "setting reset" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting reset" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Save settings
echo "setting save" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting saveas /tmp/my-settings.conf" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "setting save" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
echo "setting saveas /tmp/my-settings.conf" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
#### Clipboard Operations
```bash
# Add text to clipboard
echo "command clipboard Text to copy" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command clipboard Text to copy" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Export clipboard to file
echo "command exportclipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command exportclipboard" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
#### Application Control
```bash
# Quit Fenrir
echo "command quitapplication" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command quitapplication" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
```
### Command Reference
@@ -238,7 +244,7 @@ echo "command quitapplication" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-dea
#### Bash Helper Function
```bash
fenrir_say() {
echo "command say $1" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo "command say $1" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
}
# Usage
@@ -251,7 +257,7 @@ import socket
import os
def send_fenrir_command(command):
socket_path = "/tmp/fenrirscreenreader-deamon.sock"
socket_path = "/tmp/fenrirscreenreader-daemon.sock"
if os.path.exists(socket_path):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
@@ -330,6 +336,19 @@ Fenrir automatically detects and provides audio feedback for progress indicators
### Speech Drivers
- **speechdDriver** - Speech-dispatcher (recommended)
- **genericDriver** - Command-line TTS (espeak, etc.)
- **dectalkDriver** - Serial DECtalk-compatible hardware speech
- **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 `auto` or an explicit
serial 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.
### Sound Drivers
- **genericDriver** - Sox-based (default)
+21 -3
View File
@@ -878,10 +878,13 @@ Turn speech on or off:
enabled=True
Values: on=''True'', off=''False''
# Select speech driver, options are speechdDriver (default), genericDriver or espeakDriver:
# Select speech driver, options are speechdDriver (default), genericDriver, dectalkDriver, litetalkDriver, doubletalkDriver or tripletalkDriver:
driver=speechdDriver
#driver=espeakDriver
#driver=genericDriver
#driver=dectalkDriver
#driver=litetalkDriver
#driver=doubletalkDriver
#driver=tripletalkDriver
Select the driver used to generate speech output.
@@ -890,7 +893,10 @@ Select the driver used to generate speech output.
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''
* ''espeakDriver'' using the espeak directly, for Fenrir <1.5 just use ''espeak''
* ''dectalkDriver'' using DECtalk-compatible serial hardware or RPITalk
* ''litetalkDriver'' using LiteTalk-compatible serial hardware or RPITalk
* ''doubletalkDriver'' using DoubleTalk LT-compatible serial hardware
* ''tripletalkDriver'' using TripleTalk-compatible serial hardware
The rate selects how fast Fenrir will speak.
rate=0.65
@@ -921,6 +927,18 @@ 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. The default ''auto'' checks /dev/ttyACM* first, then /dev/ttyUSB*. Set an explicit path for stable systems.
hardware_device=auto
hardware_device=/dev/ttyACM0
hardware_device=/dev/ttyUSB0
hardware_device=/dev/ttyS0
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.
Read new text as it occurs
auto_read_incoming=True
Values: on=''True'', off=''False''
+5 -2
View File
@@ -106,14 +106,17 @@ def run_fenrir():
fenrirApp.proceed()
except Exception as e:
print(f"Error starting Fenrir: {e}", file=sys.stderr)
sys.exit(1)
except KeyboardInterrupt:
print("Interrupted", file=sys.stderr)
sys.exit(1)
finally:
if fenrirApp and hasattr(fenrirApp, 'cleanup_on_error'):
try:
fenrirApp.cleanup_on_error()
except Exception as cleanup_error:
print(
f"Error during cleanup: {cleanup_error}", file=sys.stderr)
sys.exit(1)
finally:
if fenrirApp:
del fenrirApp
# Clean up PID file if it exists
@@ -0,0 +1,27 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.core.i18n import _
class command:
def __init__(self):
pass
def initialize(self, environment):
self.env = environment
def shutdown(self):
pass
def get_description(self):
return _("opens speech history")
def run(self):
self.env["runtime"]["SpeechHistoryManager"].open_history()
def set_callback(self, callback):
pass
@@ -0,0 +1,27 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.core.i18n import _
class command:
def __init__(self):
pass
def initialize(self, environment):
self.env = environment
def shutdown(self):
pass
def get_description(self):
return _("closes speech history")
def run(self):
self.env["runtime"]["SpeechHistoryManager"].close_history()
def set_callback(self, callback):
pass
@@ -0,0 +1,27 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.core.i18n import _
class command:
def __init__(self):
pass
def initialize(self, environment):
self.env = environment
def shutdown(self):
pass
def get_description(self):
return _("copies current speech history item to the clipboard")
def run(self):
self.env["runtime"]["SpeechHistoryManager"].copy_current_to_clipboard()
def set_callback(self, callback):
pass
@@ -0,0 +1,27 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.core.i18n import _
class command:
def __init__(self):
pass
def initialize(self, environment):
self.env = environment
def shutdown(self):
pass
def get_description(self):
return _("speaks current speech history item")
def run(self):
self.env["runtime"]["SpeechHistoryManager"].present_current()
def set_callback(self, callback):
pass
@@ -0,0 +1,27 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.core.i18n import _
class command:
def __init__(self):
pass
def initialize(self, environment):
self.env = environment
def shutdown(self):
pass
def get_description(self):
return _("selects the next speech history item")
def run(self):
self.env["runtime"]["SpeechHistoryManager"].next_entry()
def set_callback(self, callback):
pass
@@ -0,0 +1,27 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.core.i18n import _
class command:
def __init__(self):
pass
def initialize(self, environment):
self.env = environment
def shutdown(self):
pass
def get_description(self):
return _("selects the previous speech history item")
def run(self):
self.env["runtime"]["SpeechHistoryManager"].prev_entry()
def set_callback(self, callback):
pass
@@ -127,6 +127,8 @@ class config_command:
self.config.set("speech", "rate", "0.75")
self.config.set("speech", "pitch", "0.5")
self.config.set("speech", "volume", "1.0")
self.config.set("speech", "hardware_device", "auto")
self.config.set("speech", "hardware_baud_rate", "9600")
self.config.add_section("sound")
self.config.set("sound", "driver", "genericDriver")
@@ -108,6 +108,8 @@ class command(config_command):
"rate": "0.5",
"pitch": "0.5",
"volume": "1.0",
"hardware_device": "auto",
"hardware_baud_rate": "9600",
"auto_read_incoming": "True",
}
+48 -18
View File
@@ -5,6 +5,7 @@
# By Chrys, Storm Dragon, and contributors.
import os
import socket
import signal
import sys
import time
@@ -40,6 +41,7 @@ class FenrirManager:
# Set signal handlers after successful initialization
signal.signal(signal.SIGINT, self.capture_signal)
signal.signal(signal.SIGTERM, self.capture_signal)
signal.signal(signal.SIGHUP, self.capture_signal)
self.signal_handlers_set = True
self.is_initialized = True
@@ -101,6 +103,10 @@ class FenrirManager:
self.environment["runtime"][
"InputManager"
].clear_event_buffer()
if self.environment["runtime"]["SpeechHistoryManager"].is_active():
self.environment["runtime"][
"InputManager"
].clear_event_buffer()
self.detect_shortcut_command()
@@ -157,6 +163,14 @@ class FenrirManager:
current_command, "vmenu-navigation"
)
return
elif self.environment["runtime"]["SpeechHistoryManager"].is_active():
if self.environment["runtime"]["CommandManager"].command_exists(
current_command, "speech-history"
):
self.environment["runtime"]["CommandManager"].execute_command(
current_command, "speech-history"
)
return
# default
self.environment["runtime"]["CommandManager"].execute_command(
@@ -296,12 +310,18 @@ class FenrirManager:
if self.command != "":
self.singleKeyCommand = True
elif (
self.environment["runtime"]["DiffReviewManager"].is_active()
(
self.environment["runtime"]["DiffReviewManager"].is_active()
or self.environment["runtime"][
"SpeechHistoryManager"
].is_active()
)
and self.command != ""
):
# Diff mode uses non-Fenrir modified bindings (Shift/Ctrl).
# Modal modes use non-Fenrir modified bindings.
# Promote resolved shortcuts to executable commands so
# combinations like Shift+H and Ctrl+Right are dispatched.
# combinations like Shift+H, Ctrl+Right, and plain arrows
# are dispatched.
self.singleKeyCommand = True
if not (self.singleKeyCommand or self.modifierInput):
@@ -381,9 +401,15 @@ class FenrirManager:
time.sleep(0.6)
for currentManager in self.environment["general"]["managerList"]:
if self.environment["runtime"][currentManager]:
self.environment["runtime"][currentManager].shutdown()
del self.environment["runtime"][currentManager]
try:
if (
currentManager in self.environment["runtime"]
and self.environment["runtime"][currentManager]
):
self.environment["runtime"][currentManager].shutdown()
del self.environment["runtime"][currentManager]
except Exception:
pass # Ignore errors during individual manager shutdown
self.environment = None
@@ -394,6 +420,7 @@ class FenrirManager:
if self.signal_handlers_set:
signal.signal(signal.SIGINT, signal.SIG_DFL)
signal.signal(signal.SIGTERM, signal.SIG_DFL)
signal.signal(signal.SIGHUP, signal.SIG_DFL)
self.signal_handlers_set = False
# Clean up any initialized managers
@@ -454,7 +481,6 @@ class FenrirManager:
# Clean up socket files that might not be removed by the driver
try:
socket_file = None
screen_driver = None
if (
"runtime" in self.environment
and "SettingsManager" in self.environment["runtime"]
@@ -465,19 +491,9 @@ class FenrirManager:
].get_setting("remote", "socket_file")
except Exception:
pass # Use default socket file path
try:
screen_driver = self.environment["runtime"][
"SettingsManager"
].get_setting("screen", "driver")
except Exception:
pass
if not socket_file:
if screen_driver == "vcsaDriver":
socket_file = "/tmp/fenrirscreenreader-deamon.sock"
if os.path.exists(socket_file):
os.unlink(socket_file)
# Always clean up PID-specific socket
pid_socket_file = (
"/tmp/fenrirscreenreader-"
+ str(os.getpid())
@@ -485,6 +501,20 @@ class FenrirManager:
)
if os.path.exists(pid_socket_file):
os.unlink(pid_socket_file)
# Clean up main socket only if it is stale (not active)
main_socket_file = "/tmp/fenrirscreenreader-daemon.sock"
if os.path.exists(main_socket_file):
try:
test_sock = socket.socket(
socket.AF_UNIX, socket.SOCK_STREAM
)
test_sock.settimeout(0.2)
test_sock.connect(main_socket_file)
except OSError:
os.unlink(main_socket_file)
finally:
test_sock.close()
elif os.path.exists(socket_file):
os.unlink(socket_file)
remoteInstanceRegistry.remove_instance()
@@ -20,6 +20,7 @@ general_data = {
"ScreenManager",
"InputManager",
"OutputManager",
"SpeechHistoryManager",
"HelpManager",
"MemoryManager",
"EventManager",
@@ -48,5 +49,6 @@ general_data = {
"onSwitchApplicationProfile",
"help",
"vmenu-navigation",
"speech-history",
],
}
+145 -112
View File
@@ -22,6 +22,8 @@ class OutputManager:
self.interrupt_thread = None
self.interrupt_done = None
self.interrupt_wait_timeout = 0.1
self.speech_driver_lock = threading.Lock()
self.speech_driver_lock_timeout = 0.05
def initialize(self, environment):
self.env = environment
@@ -73,6 +75,14 @@ class OutputManager:
return
if (len(text) > 1) and (text.strip(string.whitespace) == ""):
return
if self.env["runtime"]["SettingsManager"].get_setting_as_bool(
"speech", "enabled"
):
speech_history_manager = self.env["runtime"].get(
"SpeechHistoryManager"
)
if speech_history_manager:
speech_history_manager.add_text(text)
is_capital = self._should_announce_capital(text, announce_capital)
use_pitch_for_capital = False
@@ -159,132 +169,144 @@ class OutputManager:
return
if interrupt or flush:
self.interrupt_output()
if not self.speech_driver_lock.acquire(
timeout=self.speech_driver_lock_timeout
):
self.env["runtime"]["DebugManager"].write_debug_out(
"OutputManager.speak_text: Speech driver busy, dropping speech",
debug.DebugLevel.WARNING,
)
return
try:
self.env["runtime"]["SpeechDriver"].set_language(
self.env["runtime"]["SettingsManager"].get_setting(
"speech", "language"
try:
self.env["runtime"]["SpeechDriver"].set_language(
self.env["runtime"]["SettingsManager"].get_setting(
"speech", "language"
)
)
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech language in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
self.env["runtime"]["SpeechDriver"].set_voice(
self.env["runtime"]["SettingsManager"].get_setting(
"speech", "voice"
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech language in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"Error while setting speech voice in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
if announce_capital:
self.env["runtime"]["SpeechDriver"].set_pitch(
try:
self.env["runtime"]["SpeechDriver"].set_voice(
self.env["runtime"]["SettingsManager"].get_setting(
"speech", "voice"
)
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"Error while setting speech voice in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
if announce_capital:
self.env["runtime"]["SpeechDriver"].set_pitch(
self.env["runtime"][
"SettingsManager"
].get_setting_as_float("speech", "capital_pitch")
)
else:
self.env["runtime"]["SpeechDriver"].set_pitch(
self.env["runtime"][
"SettingsManager"
].get_setting_as_float("speech", "pitch")
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech pitch in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
self.env["runtime"]["SpeechDriver"].set_rate(
self.env["runtime"][
"SettingsManager"
].get_setting_as_float("speech", "capital_pitch")
].get_setting_as_float("speech", "rate")
)
else:
self.env["runtime"]["SpeechDriver"].set_pitch(
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech rate in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
self.env["runtime"]["SpeechDriver"].set_module(
self.env["runtime"]["SettingsManager"].get_setting(
"speech", "module"
)
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech module in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
self.env["runtime"]["SpeechDriver"].set_volume(
self.env["runtime"][
"SettingsManager"
].get_setting_as_float("speech", "pitch")
].get_setting_as_float("speech", "volume")
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech pitch in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
self.env["runtime"]["SpeechDriver"].set_rate(
self.env["runtime"]["SettingsManager"].get_setting_as_float(
"speech", "rate"
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech volume in OutputManager.speak_text ",
debug.DebugLevel.ERROR,
)
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech rate in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
self.env["runtime"]["SpeechDriver"].set_module(
self.env["runtime"]["SettingsManager"].get_setting(
"speech", "module"
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech module in OutputManager.speak_text",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
self.env["runtime"]["SpeechDriver"].set_volume(
self.env["runtime"]["SettingsManager"].get_setting_as_float(
"speech", "volume"
try:
if self.env["runtime"]["SettingsManager"].get_setting_as_bool(
"general", "new_line_pause"
):
clean_text = text.replace("\n", " , ")
else:
clean_text = text.replace("\n", " ")
clean_text = self.env["runtime"][
"TextManager"
].replace_head_lines(clean_text)
clean_text = self.process_mid_word_punctuation(clean_text)
clean_text = self.env["runtime"][
"PunctuationManager"
].proceed_punctuation(clean_text, ignore_punctuation)
clean_text = re.sub(" +$", " ", clean_text)
self.env["runtime"]["SpeechDriver"].speak(
clean_text, True, ignore_punctuation
)
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"setting speech volume in OutputManager.speak_text ",
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
try:
if self.env["runtime"]["SettingsManager"].get_setting_as_bool(
"general", "new_line_pause"
):
clean_text = text.replace("\n", " , ")
else:
clean_text = text.replace("\n", " ")
clean_text = self.env["runtime"]["TextManager"].replace_head_lines(
clean_text
)
clean_text = self.process_mid_word_punctuation(clean_text)
clean_text = self.env["runtime"][
"PunctuationManager"
].proceed_punctuation(clean_text, ignore_punctuation)
clean_text = re.sub(" +$", " ", clean_text)
self.env["runtime"]["SpeechDriver"].speak(
clean_text, True, ignore_punctuation
)
self.env["runtime"]["DebugManager"].write_debug_out(
"Speak: " + clean_text, debug.DebugLevel.INFO
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
'"speak" in OutputManager.speak_text ', debug.DebugLevel.ERROR
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
self.env["runtime"]["DebugManager"].write_debug_out(
"Speak: " + clean_text, debug.DebugLevel.INFO
)
except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out(
'"speak" in OutputManager.speak_text ',
debug.DebugLevel.ERROR,
)
self.env["runtime"]["DebugManager"].write_debug_out(
str(e), debug.DebugLevel.ERROR
)
finally:
self.speech_driver_lock.release()
def interrupt_output(self, wait=True):
interrupt_done, started = self.start_interrupt_output()
@@ -320,6 +342,15 @@ class OutputManager:
self.interrupt_running = False
def cancel_speech(self):
if not self.speech_driver_lock.acquire(
timeout=self.speech_driver_lock_timeout
):
self.env["runtime"]["DebugManager"].write_debug_out(
"OutputManager interrupt_output: Speech driver busy, "
"skipping interrupt",
debug.DebugLevel.WARNING,
)
return
try:
self.env["runtime"]["SpeechDriver"].cancel()
self.env["runtime"]["DebugManager"].write_debug_out(
@@ -331,6 +362,8 @@ class OutputManager:
+ str(e),
debug.DebugLevel.ERROR,
)
finally:
self.speech_driver_lock.release()
def play_sound_icon(self, sound_icon="", interrupt=True):
if sound_icon == "":
@@ -28,6 +28,7 @@ def write_instance(instance_data):
registry_dir = get_registry_dir()
os.makedirs(registry_dir, mode=0o755, exist_ok=True)
os.chmod(registry_dir, 0o755)
prune_stale_instances()
instance_data["updated_at"] = time.time()
instance_path = get_instance_file(instance_data.get("pid"))
with open(instance_path, "w", encoding="utf-8") as instance_file:
@@ -53,6 +54,39 @@ def process_exists(pid):
return False
def prune_stale_instances():
registry_dir = get_registry_dir()
try:
instance_files = os.listdir(registry_dir)
except FileNotFoundError:
return
now = time.time()
for instance_name in instance_files:
instance_path = os.path.join(registry_dir, instance_name)
try:
with open(instance_path, "r", encoding="utf-8") as instance_file:
instance_data = json.load(instance_file)
pid = int(instance_data.get("pid", 0))
updated_at = float(instance_data.get("updated_at", 0))
except (OSError, ValueError, TypeError, json.JSONDecodeError):
try:
os.unlink(instance_path)
except OSError:
pass
continue
if (
not pid
or not process_exists(pid)
or now - updated_at > INSTANCE_TIMEOUT_SEC
):
try:
os.unlink(instance_path)
except OSError:
pass
def list_instances():
registry_dir = get_registry_dir()
instances = []
@@ -72,7 +106,11 @@ def list_instances():
except (OSError, ValueError, TypeError, json.JSONDecodeError):
continue
if not pid or not process_exists(pid) or now - updated_at > INSTANCE_TIMEOUT_SEC:
if (
not pid
or not process_exists(pid)
or now - updated_at > INSTANCE_TIMEOUT_SEC
):
try:
os.unlink(instance_path)
except OSError:
@@ -16,6 +16,7 @@ runtime_data = {
"CommandManager": None,
"ScreenManager": None,
"OutputManager": None,
"SpeechHistoryManager": None,
"DebugManager": None,
"SettingsManager": None,
"FenrirManager": None,
@@ -28,8 +28,11 @@ settings_data = {
"module": "",
"voice": "en-us",
"language": "",
"hardware_device": "auto",
"hardware_baud_rate": 9600,
"auto_read_incoming": True,
"read_numbers_as_digits": False,
"history_size": 50,
"rapid_update_threshold": 5,
"rapid_update_window": 0.3,
"batch_flush_interval": 0.5,
@@ -29,6 +29,7 @@ from fenrirscreenreader.core import readAllManager
from fenrirscreenreader.core import remoteManager
from fenrirscreenreader.core import sayAllManager
from fenrirscreenreader.core import screenManager
from fenrirscreenreader.core import speechHistoryManager
from fenrirscreenreader.core import tableManager
from fenrirscreenreader.core import textManager
from fenrirscreenreader.core import vmenuManager
@@ -508,6 +509,10 @@ class SettingsManager:
valid_drivers = [
"speechdDriver",
"genericDriver",
"dectalkDriver",
"doubletalkDriver",
"litetalkDriver",
"tripletalkDriver",
"dummyDriver",
]
if value not in valid_drivers:
@@ -734,6 +739,11 @@ class SettingsManager:
environment["runtime"]["OutputManager"] = outputManager.OutputManager()
environment["runtime"]["OutputManager"].initialize(environment)
environment["runtime"][
"SpeechHistoryManager"
] = speechHistoryManager.SpeechHistoryManager()
environment["runtime"]["SpeechHistoryManager"].initialize(environment)
environment["runtime"]["InputManager"] = inputManager.InputManager()
environment["runtime"]["InputManager"].initialize(environment)
self.load_keyboard_layout(environment)
@@ -0,0 +1,182 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.core.i18n import _
class SpeechHistoryManager:
def __init__(self):
self.env = None
self.history = []
self.curr_index = -1
self.active = False
self.bindings_backup = None
self.raw_bindings_backup = None
def initialize(self, environment):
self.env = environment
def shutdown(self):
self.set_active(False)
def is_active(self):
return self.active
def add_text(self, text):
if self.active:
return False
if not isinstance(text, str):
return False
if text == "":
return False
text_key = self._get_history_key(text)
if text_key == "":
return False
if text_key in [self._get_history_key(item) for item in self.history]:
return False
history_size = self._get_history_size()
if history_size <= 0:
return False
self.history.insert(0, text)
del self.history[history_size:]
if self.curr_index >= len(self.history):
self.curr_index = len(self.history) - 1
return True
def open_history(self):
if not self.history:
self.env["runtime"]["OutputManager"].present_text(
_("speech history empty"), interrupt=True
)
return False
self.curr_index = -1
self.set_active(True)
self.env["runtime"]["OutputManager"].present_text(
_("Speech history"), interrupt=True
)
return True
def close_history(self, announce=True):
if announce:
self.env["runtime"]["OutputManager"].present_text(
_("speech history closed"), interrupt=True
)
self.set_active(False)
def next_entry(self):
if not self._has_history():
return
if self.curr_index == -1:
self.curr_index = 0
self.present_current()
return
if self.curr_index <= 0:
self.curr_index = 0
self.env["runtime"]["OutputManager"].present_text(
_("First speech history item"), interrupt=True
)
self.present_current(interrupt=False)
return
self.curr_index -= 1
self.present_current()
def prev_entry(self):
if not self._has_history():
return
if self.curr_index == -1:
self.curr_index = 0
self.present_current()
return
if self.curr_index >= len(self.history) - 1:
self.curr_index = len(self.history) - 1
self.env["runtime"]["OutputManager"].present_text(
_("Last speech history item"), interrupt=True
)
self.present_current(interrupt=False)
return
self.curr_index += 1
self.present_current()
def present_current(self, interrupt=True):
if not self._has_history():
self.env["runtime"]["OutputManager"].present_text(
_("speech history empty"), interrupt=True
)
return
self.env["runtime"]["OutputManager"].present_text(
self.history[self.curr_index], interrupt=interrupt
)
def copy_current_to_clipboard(self):
if not self._has_history():
self.close_history()
return
text = self.history[self.curr_index]
self.env["runtime"]["MemoryManager"].add_value_to_first_index(
"clipboardHistory", text
)
self.env["runtime"]["OutputManager"].present_text(
_("copied to clipboard"),
sound_icon="CopyToClipboard",
interrupt=True,
)
self.set_active(False)
def set_active(self, active):
if active == self.active:
return
self.active = active
if self.active:
self._install_bindings()
else:
self._restore_bindings()
def _has_history(self):
if not self.history:
self.curr_index = -1
return False
if self.curr_index >= len(self.history):
self.curr_index = len(self.history) - 1
return True
def _get_history_size(self):
try:
return self.env["runtime"]["SettingsManager"].get_setting_as_int(
"speech", "history_size"
)
except Exception:
return 50
def _get_history_key(self, text):
return " ".join(text.split())
def _install_bindings(self):
self.bindings_backup = self.env["bindings"].copy()
self.raw_bindings_backup = self.env["rawBindings"].copy()
self.env["bindings"] = {
str([1, ["KEY_UP"]]): "SPEECH_HISTORY_PREV",
str([1, ["KEY_DOWN"]]): "SPEECH_HISTORY_NEXT",
str([1, ["KEY_SPACE"]]): "SPEECH_HISTORY_CURRENT",
str([1, ["KEY_ENTER"]]): "SPEECH_HISTORY_COPY",
str([1, ["KEY_KPENTER"]]): "SPEECH_HISTORY_COPY",
str([1, ["KEY_ESC"]]): "SPEECH_HISTORY_CLOSE",
}
self.env["rawBindings"] = {
str([1, ["KEY_UP"]]): [1, ["KEY_UP"]],
str([1, ["KEY_DOWN"]]): [1, ["KEY_DOWN"]],
str([1, ["KEY_SPACE"]]): [1, ["KEY_SPACE"]],
str([1, ["KEY_ENTER"]]): [1, ["KEY_ENTER"]],
str([1, ["KEY_KPENTER"]]): [1, ["KEY_KPENTER"]],
str([1, ["KEY_ESC"]]): [1, ["KEY_ESC"]],
}
def _restore_bindings(self):
if self.bindings_backup is not None:
self.env["bindings"] = self.bindings_backup
if self.raw_bindings_backup is not None:
self.env["rawBindings"] = self.raw_bindings_backup
self.bindings_backup = None
self.raw_bindings_backup = None
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
version = "2026.05.14"
version = "2026.05.21"
code_name = "testing"
@@ -368,6 +368,7 @@ class driver(inputDriver):
"event_state": 1 if event.type == X.KeyPress else 0,
"event_type": event.type,
"event_raw_state": getattr(event, "state", 0),
"event_x_time": getattr(event, "time", X.CurrentTime),
}
def refresh_modifier_state(self):
@@ -650,8 +651,28 @@ class driver(inputDriver):
self.clear_event_buffer()
def write_event_buffer(self):
if self.display:
for event in self.env["input"]["event_buffer"]:
self.replay_key_event(event)
self.clear_event_buffer()
def replay_key_event(self, event):
if not isinstance(event, dict):
return
if event.get("event_type") != X.KeyPress:
return
try:
self.display.allow_events(
X.ReplayKeyboard,
event.get("event_x_time", X.CurrentTime),
)
self.display.flush()
except Exception as e:
self.write_debug(
"x11Driver replay key event failed: " + str(e),
debug.DebugLevel.ERROR,
)
def clear_event_buffer(self):
if not self._initialized:
return
@@ -8,6 +8,7 @@ import os
import os.path
import select
import socket
import struct
import time
from fenrirscreenreader.core import debug
@@ -16,7 +17,7 @@ from fenrirscreenreader.core.eventData import FenrirEventType
from fenrirscreenreader.core.remoteDriver import RemoteDriver as remoteDriver
MAIN_SOCKET_FILE = "/tmp/fenrirscreenreader-deamon.sock"
MAIN_SOCKET_FILE = "/tmp/fenrirscreenreader-daemon.sock"
class driver(remoteDriver):
@@ -132,6 +133,131 @@ class driver(remoteDriver):
return False
return False
def _socket_file_for_socket(self, fenrir_sock):
for bound_sock, socket_file in self.bound_sockets:
if bound_sock == fenrir_sock:
return socket_file
return ""
def _get_peer_pid(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 pid
except OSError:
return 0
def _get_parent_pid(self, pid):
try:
with open(f"/proc/{pid}/stat", "r", encoding="utf-8") as proc_file:
stat_text = proc_file.read()
except OSError:
return 0
end_command = stat_text.rfind(")")
if end_command == -1:
return 0
stat_fields = stat_text[end_command + 2 :].split()
if len(stat_fields) < 2:
return 0
try:
return int(stat_fields[1])
except ValueError:
return 0
def _find_ancestor_private_socket(self, pid):
seen_pids = set()
while pid > 1 and pid not in seen_pids:
seen_pids.add(pid)
socket_file = f"/tmp/fenrirscreenreader-{pid}.sock"
if self._is_registered_private_socket(socket_file):
return socket_file
pid = self._get_parent_pid(pid)
return ""
def _is_registered_private_socket(self, socket_file):
for instance in remoteInstanceRegistry.list_instances():
if socket_file in instance.get("socket_files", []):
return True
return False
def _get_registered_private_sockets(self):
socket_files = []
for instance in remoteInstanceRegistry.list_instances():
for socket_file in instance.get("socket_files", []):
if socket_file == MAIN_SOCKET_FILE:
continue
if socket_file in socket_files:
continue
socket_files.append(socket_file)
return socket_files
def _find_available_private_socket(self, preferred_socket=""):
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 self._is_own_socket_file(socket_file):
return socket_file
if self._is_socket_active(socket_file):
return socket_file
return ""
def _is_own_socket_file(self, socket_file):
return any(
socket_file == bound_socket_file
for _bound_sock, bound_socket_file in self.bound_sockets
)
def _has_own_private_socket(self):
return any(
bound_socket_file != MAIN_SOCKET_FILE
for _bound_sock, bound_socket_file in self.bound_sockets
)
def _forward_remote_to_socket(self, data, socket_file):
forward_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
try:
forward_sock.settimeout(0.2)
forward_sock.connect(socket_file)
forward_sock.sendall((data + "\n").encode("utf-8"))
return True
except OSError as e:
self.env["runtime"]["DebugManager"].write_debug_out(
"unixDriver watch_dog: Error forwarding remote data to "
+ socket_file
+ ": "
+ str(e),
debug.DebugLevel.ERROR,
)
return False
finally:
forward_sock.close()
def _route_main_socket_command(self, data, client_sock, socket_file):
if socket_file != MAIN_SOCKET_FILE:
return False
if not self._has_own_private_socket():
return False
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 not target_socket:
return False
if self._is_own_socket_file(target_socket):
return False
return self._forward_remote_to_socket(data, target_socket)
def _cleanup(self):
for fenrir_sock in self.fenrirSocks:
try:
@@ -152,7 +278,7 @@ class driver(remoteDriver):
self.bound_sockets = []
remoteInstanceRegistry.remove_instance()
def _handle_client(self, client_sock, event_queue):
def _handle_client(self, client_sock, event_queue, socket_file=""):
try:
rawdata = client_sock.recv(8129)
except Exception as e:
@@ -173,6 +299,9 @@ class driver(remoteDriver):
client_sock.sendall((response["message"] + "\n").encode("utf-8"))
return
if self._route_main_socket_command(data, client_sock, socket_file):
return
event_queue.put(
{
"Type": FenrirEventType.remote_incomming,
@@ -188,7 +317,7 @@ class driver(remoteDriver):
def watch_dog(self, active, event_queue):
# echo "command say this is a test" | socat -
# UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
# 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:
@@ -215,10 +344,11 @@ class driver(remoteDriver):
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
try:
self._handle_client(client_sock, event_queue)
self._handle_client(client_sock, event_queue, socket_file)
finally:
# Always close client socket, even if data processing fails
try:
@@ -0,0 +1,28 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.speechDriver.hardwareSerialDriver import (
hardware_serial_driver,
)
class driver(hardware_serial_driver):
cancel_command = b"\x18"
def _speak_bytes(self, text):
return self._clean_text(text).encode("ascii", errors="replace") + b"\x01"
def _rate_command(self, rate):
return self._setting_command("ra", self._scale(rate, 75, 650))
def _pitch_command(self, pitch):
return self._setting_command("dv ap", self._scale(pitch, 50, 180))
def _volume_command(self, volume):
return self._setting_command("vo", self._scale(volume, 0, 100))
def _setting_command(self, command, value):
return f"[:{command} {value}]".encode("ascii")
@@ -0,0 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.speechDriver.litetalkDriver import driver
@@ -0,0 +1,226 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
import glob
import os
import termios
import threading
import tty
from queue import Empty
from queue import Queue
from fenrirscreenreader.core import debug
from fenrirscreenreader.core.speechDriver import speech_driver
class SpeakQueue(Queue):
def clear(self):
try:
while True:
self.get_nowait()
except Empty:
pass
class hardware_serial_driver(speech_driver):
cancel_command = b""
default_baud_rate = 9600
def __init__(self):
speech_driver.__init__(self)
self.device = ""
self.baud_rate = self.default_baud_rate
self.serial_port = None
self.text_queue = SpeakQueue()
self.lock = threading.Lock()
self.worker_thread = None
self._stop_worker = False
def initialize(self, environment):
self.env = environment
self._is_initialized = False
settings_manager = self.env["runtime"]["SettingsManager"]
self.device = settings_manager.get_setting(
"speech", "hardware_device"
)
self.baud_rate = settings_manager.get_setting_as_int(
"speech", "hardware_baud_rate"
)
self._open_serial_port()
self._is_initialized = self.serial_port is not None
if self._is_initialized:
self._stop_worker = False
self.worker_thread = threading.Thread(
target=self._worker, daemon=True
)
self.worker_thread.start()
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)
self._close_serial_port()
self._is_initialized = False
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.text_queue.put(text)
def cancel(self):
if not self._is_initialized:
return
self.clear_buffer()
if self.cancel_command:
self._write_bytes(self.cancel_command)
def clear_buffer(self):
if not self._is_initialized:
return
self.text_queue.clear()
def set_rate(self, rate):
if not self._is_initialized:
return
if not isinstance(rate, float):
return
self._write_bytes(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))
def set_volume(self, volume):
if not self._is_initialized:
return
if not isinstance(volume, float):
return
self._write_bytes(self._volume_command(volume))
def _worker(self):
while not self._stop_worker:
text = self.text_queue.get()
if text is None:
return
self._write_bytes(self._speak_bytes(text))
def _open_serial_port(self):
device = self._resolve_device(self.device)
if not device:
self._debug(
"Hardware speech device not found",
debug.DebugLevel.ERROR,
)
return
try:
port = os.open(device, os.O_RDWR | os.O_NOCTTY)
tty.setraw(port)
attrs = termios.tcgetattr(port)
attrs[2] |= termios.CLOCAL | termios.CREAD
baud_rate = self._termios_baud_rate(self.baud_rate)
attrs[4] = baud_rate
attrs[5] = baud_rate
attrs[6][termios.VMIN] = 0
attrs[6][termios.VTIME] = 0
attrs[0] &= ~(termios.IXON | termios.IXOFF | termios.IXANY)
termios.tcsetattr(port, termios.TCSANOW, attrs)
self.serial_port = port
self.device = device
except OSError as error:
self._debug(
f"Hardware speech device open failed: {device}: {error}",
debug.DebugLevel.ERROR,
)
self.serial_port = None
def _close_serial_port(self):
with self.lock:
if self.serial_port is None:
return
try:
os.close(self.serial_port)
except OSError as error:
self._debug(
f"Hardware speech device close failed: {error}",
debug.DebugLevel.WARNING,
)
finally:
self.serial_port = None
def _write_bytes(self, data):
if not data:
return
with self.lock:
if self.serial_port is None:
return
try:
os.write(self.serial_port, data)
except OSError as error:
self._debug(
f"Hardware speech write failed: {error}",
debug.DebugLevel.ERROR,
)
def _resolve_device(self, device):
if device and device != "auto":
return device
for pattern in ("/dev/ttyACM*", "/dev/ttyUSB*"):
matches = sorted(glob.glob(pattern))
if matches:
return matches[0]
return ""
def _termios_baud_rate(self, baud_rate):
baud_name = f"B{baud_rate}"
if hasattr(termios, baud_name):
return getattr(termios, baud_name)
self._debug(
f"Unsupported hardware speech baud rate {baud_rate}; using 9600",
debug.DebugLevel.WARNING,
)
return termios.B9600
def _clean_text(self, text):
text = text.replace("\r", " ").replace("\n", " ")
return "".join(
char if 0x20 <= ord(char) <= 0x7E else " "
for char in text
)
def _scale(self, value, minimum, maximum):
value = max(0.0, min(1.0, value))
return int(round(minimum + value * (maximum - minimum)))
def _debug(self, message, level):
try:
self.env["runtime"]["DebugManager"].write_debug_out(
message, level
)
except Exception:
pass
def _speak_bytes(self, text):
raise NotImplementedError
def _rate_command(self, rate):
return b""
def _pitch_command(self, pitch):
return b""
def _volume_command(self, volume):
return b""
@@ -0,0 +1,28 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.speechDriver.hardwareSerialDriver import (
hardware_serial_driver,
)
class driver(hardware_serial_driver):
cancel_command = b"\x18"
def _speak_bytes(self, text):
return self._clean_text(text).encode("ascii", errors="replace") + b"\r"
def _rate_command(self, rate):
return self._setting_command(self._scale(rate, 0, 9), b"S")
def _pitch_command(self, pitch):
return self._setting_command(self._scale(pitch, 0, 99), b"P")
def _volume_command(self, volume):
return self._setting_command(self._scale(volume, 0, 9), b"V")
def _setting_command(self, value, command):
return b"\x01" + str(value).encode("ascii") + command
@@ -0,0 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
from fenrirscreenreader.speechDriver.litetalkDriver import driver
+1 -1
View File
@@ -232,7 +232,7 @@ class TestRemoteDataFormat:
"x11_window_id": "0x123",
"socket_files": [
"/tmp/fenrirscreenreader-123.sock",
"/tmp/fenrirscreenreader-deamon.sock",
"/tmp/fenrirscreenreader-daemon.sock",
],
}
],
@@ -0,0 +1,44 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.eventData import FenrirEventType
from fenrirscreenreader.core.fenrirManager import FenrirManager
@pytest.mark.unit
def test_speech_history_plain_key_modal_command_is_dispatched():
manager = FenrirManager.__new__(FenrirManager)
manager.modifierInput = False
manager.singleKeyCommand = False
manager.command = ""
event_manager = Mock(put_to_event_queue=Mock())
input_manager = Mock(
is_key_press=Mock(return_value=False),
no_key_pressed=Mock(return_value=False),
get_curr_shortcut=Mock(return_value=str([1, ["KEY_UP"]])),
get_command_for_shortcut=Mock(return_value="SPEECH_HISTORY_PREV"),
)
speech_history_manager = Mock(is_active=Mock(return_value=True))
diff_review_manager = Mock(is_active=Mock(return_value=False))
manager.environment = {
"input": {
"key_forward": 0,
"prev_input": ["KEY_UP"],
"curr_input": ["KEY_UP"],
},
"runtime": {
"InputManager": input_manager,
"EventManager": event_manager,
"DiffReviewManager": diff_review_manager,
"SpeechHistoryManager": speech_history_manager,
},
}
manager.detect_shortcut_command()
event_manager.put_to_event_queue.assert_called_once_with(
FenrirEventType.execute_command, "SPEECH_HISTORY_PREV"
)
+124
View File
@@ -0,0 +1,124 @@
import os
import select
import time
from unittest.mock import Mock
import pytest
from fenrirscreenreader.speechDriver import dectalkDriver
from fenrirscreenreader.speechDriver import doubletalkDriver
from fenrirscreenreader.speechDriver import litetalkDriver
from fenrirscreenreader.speechDriver import tripletalkDriver
def build_environment(device):
settings_manager = Mock()
settings_manager.get_setting.side_effect = (
lambda section, setting: device
if (section, setting) == ("speech", "hardware_device")
else ""
)
settings_manager.get_setting_as_int.side_effect = (
lambda section, setting: 9600
if (section, setting) == ("speech", "hardware_baud_rate")
else 0
)
return {
"runtime": {
"SettingsManager": settings_manager,
"DebugManager": Mock(),
}
}
def read_available(fd, expected_length, timeout=1.0):
deadline = time.monotonic() + timeout
data = b""
while len(data) < expected_length and time.monotonic() < deadline:
readable, _, _ = select.select([fd], [], [], 0.05)
if readable:
data += os.read(fd, 1024)
return data
@pytest.fixture
def serial_pair():
master_fd, slave_fd = os.openpty()
try:
yield master_fd, os.ttyname(slave_fd)
finally:
os.close(master_fd)
os.close(slave_fd)
def initialized_driver(driver_class, serial_pair):
master_fd, slave_name = serial_pair
speech_driver = driver_class.driver()
speech_driver.initialize(build_environment(slave_name))
assert speech_driver._is_initialized
return speech_driver, master_fd
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"
finally:
speech_driver.shutdown()
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_volume(0.5)
speech_driver.cancel()
assert read_available(master_fd, 33) == (
b"[:ra 650][:dv ap 50][:vo 50]\x18"
)
finally:
speech_driver.shutdown()
def test_litetalk_driver_speaks_printable_text(serial_pair):
speech_driver, master_fd = initialized_driver(litetalkDriver, serial_pair)
try:
speech_driver.speak("Ready")
assert read_available(master_fd, 6) == b"Ready\r"
finally:
speech_driver.shutdown()
def test_litetalk_driver_writes_settings_and_cancel(serial_pair):
speech_driver, master_fd = initialized_driver(litetalkDriver, serial_pair)
try:
speech_driver.set_rate(1.0)
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"
finally:
speech_driver.shutdown()
@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)
try:
speech_driver.speak("Alias")
speech_driver.set_rate(1.0)
assert read_available(master_fd, 10) == b"\x019SAlias\r"
finally:
speech_driver.shutdown()
def test_hardware_driver_ignores_empty_and_non_string_text(serial_pair):
speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
try:
speech_driver.speak("")
speech_driver.speak(None)
assert read_available(master_fd, 1, timeout=0.2) == b""
finally:
speech_driver.shutdown()
+79
View File
@@ -26,6 +26,14 @@ def build_output_manager():
"SoundDriver": sound_driver,
"SpeechDriver": speech_driver,
"DebugManager": Mock(write_debug_out=Mock()),
"TextManager": Mock(
replace_head_lines=Mock(side_effect=lambda text: text)
),
"PunctuationManager": Mock(
proceed_punctuation=Mock(
side_effect=lambda text, _ignore_punctuation: text
)
),
},
}
return output_manager, sound_driver, speech_driver
@@ -122,6 +130,77 @@ def test_interrupt_output_waits_only_briefly_for_slow_cancel():
output_manager.interrupt_thread.join(timeout=1.0)
@pytest.mark.unit
def test_cancel_speech_skips_when_speech_driver_is_busy():
output_manager, _sound_driver, speech_driver = build_output_manager()
output_manager.speech_driver_lock_timeout = 0.01
output_manager.speech_driver_lock.acquire()
try:
start_time = time.monotonic()
output_manager.cancel_speech()
elapsed = time.monotonic() - start_time
finally:
output_manager.speech_driver_lock.release()
assert elapsed < 0.2
speech_driver.cancel.assert_not_called()
@pytest.mark.unit
def test_speak_text_drops_speech_when_cancel_holds_driver_lock():
output_manager, _sound_driver, speech_driver = build_output_manager()
output_manager.speech_driver_lock_timeout = 0.01
output_manager.speech_driver_lock.acquire()
try:
start_time = time.monotonic()
output_manager.speak_text("hello", interrupt=False, flush=False)
elapsed = time.monotonic() - start_time
finally:
output_manager.speech_driver_lock.release()
assert elapsed < 0.2
speech_driver.speak.assert_not_called()
@pytest.mark.unit
def test_present_text_records_speech_history_when_enabled():
output_manager, _sound_driver, speech_driver = build_output_manager()
speech_history_manager = Mock(add_text=Mock())
output_manager.env["runtime"]["SpeechHistoryManager"] = (
speech_history_manager
)
output_manager.present_text("hello history", interrupt=False)
speech_history_manager.add_text.assert_called_once_with("hello history")
speech_driver.speak.assert_called_once()
@pytest.mark.unit
def test_present_text_does_not_record_when_speech_disabled():
output_manager, _sound_driver, speech_driver = build_output_manager()
speech_history_manager = Mock(add_text=Mock())
output_manager.env["runtime"]["SpeechHistoryManager"] = (
speech_history_manager
)
def _get_setting_as_bool(section, setting):
if (section, setting) == ("speech", "enabled"):
return False
return True
output_manager.env["runtime"][
"SettingsManager"
].get_setting_as_bool.side_effect = _get_setting_as_bool
output_manager.present_text("hello history", interrupt=False)
speech_history_manager.add_text.assert_not_called()
speech_driver.speak.assert_not_called()
@pytest.mark.unit
def test_key_interrupt_command_uses_nonblocking_interrupt():
module = load_key_interrupt_module()
@@ -0,0 +1,35 @@
import json
import time
from fenrirscreenreader.core import remoteInstanceRegistry
def test_write_instance_prunes_stale_registry_files(tmp_path, monkeypatch):
monkeypatch.setattr(
remoteInstanceRegistry, "get_registry_dir", lambda: str(tmp_path)
)
monkeypatch.setattr(
remoteInstanceRegistry,
"process_exists",
lambda pid: pid == 456,
)
stale_instance = tmp_path / "123.json"
stale_instance.write_text(
json.dumps(
{
"pid": 123,
"updated_at": time.time(),
}
)
+ "\n",
encoding="utf-8",
)
invalid_instance = tmp_path / "invalid.json"
invalid_instance.write_text("not json\n", encoding="utf-8")
remoteInstanceRegistry.write_instance({"pid": 456})
assert not stale_instance.exists()
assert not invalid_instance.exists()
assert (tmp_path / "456.json").exists()
+4
View File
@@ -71,6 +71,10 @@ class TestSpeechSettingsValidation:
# Valid drivers
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", "doubletalkDriver")
self.manager._validate_setting_value("speech", "driver", "litetalkDriver")
self.manager._validate_setting_value("speech", "driver", "tripletalkDriver")
self.manager._validate_setting_value("speech", "driver", "dummyDriver")
# Invalid driver
+147
View File
@@ -0,0 +1,147 @@
from unittest.mock import Mock
import pytest
from fenrirscreenreader.core.speechHistoryManager import SpeechHistoryManager
def build_speech_history_manager(history_size=3):
spoken_messages = []
output_manager = Mock()
def _capture_message(message, **_kwargs):
spoken_messages.append(message)
output_manager.present_text.side_effect = _capture_message
settings_manager = Mock()
settings_manager.get_setting_as_int.return_value = history_size
memory_manager = Mock(add_value_to_first_index=Mock())
env = {
"runtime": {
"OutputManager": output_manager,
"SettingsManager": settings_manager,
"MemoryManager": memory_manager,
},
"bindings": {"original": "COMMAND"},
"rawBindings": {"original": [1, ["KEY_FENRIR"]]},
}
manager = SpeechHistoryManager()
manager.initialize(env)
return manager, env, spoken_messages, memory_manager
@pytest.mark.unit
def test_speech_history_keeps_configured_number_of_items():
manager, _env, _spoken_messages, _memory_manager = (
build_speech_history_manager(history_size=2)
)
assert manager.add_text("one")
assert manager.add_text("two")
assert manager.add_text("three")
assert manager.history == ["three", "two"]
@pytest.mark.unit
def test_speech_history_suppresses_exact_duplicates_until_item_drops():
manager, _env, _spoken_messages, _memory_manager = (
build_speech_history_manager(history_size=2)
)
assert manager.add_text("hello world")
assert not manager.add_text("hello world")
assert manager.add_text("other")
assert manager.add_text("third")
assert manager.add_text("hello world")
assert manager.history == ["hello world", "third"]
@pytest.mark.unit
def test_speech_history_dedupe_keeps_case_and_suppresses_whitespace_variants():
manager, _env, _spoken_messages, _memory_manager = (
build_speech_history_manager()
)
assert manager.add_text("hello")
assert manager.add_text("Hello")
assert not manager.add_text("hello ")
assert not manager.add_text("hello ")
assert manager.history == ["Hello", "hello"]
@pytest.mark.unit
def test_open_empty_history_announces_empty_without_modal_bindings():
manager, env, spoken_messages, _memory_manager = (
build_speech_history_manager()
)
assert not manager.open_history()
assert not manager.is_active()
assert spoken_messages == ["speech history empty"]
assert env["bindings"] == {"original": "COMMAND"}
@pytest.mark.unit
def test_open_history_installs_modal_bindings_and_replay_is_not_recorded():
manager, env, spoken_messages, _memory_manager = (
build_speech_history_manager()
)
manager.add_text("first")
manager.add_text("second")
assert manager.open_history()
manager.add_text("replayed")
assert manager.is_active()
assert spoken_messages == ["Speech history"]
assert manager.curr_index == -1
assert manager.history == ["second", "first"]
assert "original" not in env["bindings"]
assert "original" not in env["rawBindings"]
assert env["bindings"][str([1, ["KEY_UP"]])] == "SPEECH_HISTORY_PREV"
assert env["bindings"][str([1, ["KEY_ENTER"]])] == "SPEECH_HISTORY_COPY"
assert env["bindings"][str([1, ["KEY_ESC"]])] == "SPEECH_HISTORY_CLOSE"
@pytest.mark.unit
def test_navigation_moves_between_newer_and_older_items():
manager, _env, spoken_messages, _memory_manager = (
build_speech_history_manager()
)
manager.add_text("oldest")
manager.add_text("middle")
manager.add_text("newest")
manager.open_history()
manager.prev_entry()
manager.prev_entry()
manager.prev_entry()
manager.next_entry()
assert spoken_messages[-4:] == ["newest", "middle", "oldest", "middle"]
@pytest.mark.unit
def test_copy_current_adds_clipboard_and_restores_bindings():
manager, env, spoken_messages, memory_manager = (
build_speech_history_manager()
)
manager.add_text("first")
manager.add_text("second")
manager.open_history()
manager.prev_entry()
manager.prev_entry()
manager.copy_current_to_clipboard()
memory_manager.add_value_to_first_index.assert_called_once_with(
"clipboardHistory", "first"
)
assert spoken_messages[-1] == "copied to clipboard"
assert not manager.is_active()
assert env["bindings"] == {"original": "COMMAND"}
assert env["rawBindings"] == {"original": [1, ["KEY_FENRIR"]]}
+223
View File
@@ -0,0 +1,223 @@
from unittest.mock import Mock, mock_open, patch
from fenrirscreenreader.remoteDriver import unixDriver
class FakeClientSocket:
def __init__(self, data):
self.data = data
self.sent = b""
def recv(self, _size):
return self.data
def sendall(self, data):
self.sent += data
def test_main_socket_routes_to_ancestor_private_socket(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
driver.bound_sockets = [
(Mock(), "/tmp/fenrirscreenreader-999.sock"),
(Mock(), unixDriver.MAIN_SOCKET_FILE),
]
client_sock = FakeClientSocket(b"command say routed")
event_queue = Mock()
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
driver,
"_find_ancestor_private_socket",
return_value="/tmp/fenrirscreenreader-222.sock",
), patch.object(
driver,
"_find_available_private_socket",
return_value="/tmp/fenrirscreenreader-222.sock",
), patch.object(
driver, "_forward_remote_to_socket", return_value=True
) as forward:
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
forward.assert_called_once_with(
"command say routed", "/tmp/fenrirscreenreader-222.sock"
)
event_queue.put.assert_not_called()
def test_main_socket_routes_to_first_available_private_socket(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
driver.bound_sockets = [
(Mock(), "/tmp/fenrirscreenreader-999.sock"),
(Mock(), unixDriver.MAIN_SOCKET_FILE),
]
client_sock = FakeClientSocket(b"command say fallback")
event_queue = Mock()
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
driver,
"_find_ancestor_private_socket",
return_value="",
), patch.object(
driver,
"_find_available_private_socket",
return_value="/tmp/fenrirscreenreader-333.sock",
), patch.object(
driver, "_forward_remote_to_socket", return_value=True
) as forward:
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
forward.assert_called_once_with(
"command say fallback", "/tmp/fenrirscreenreader-333.sock"
)
event_queue.put.assert_not_called()
def test_main_socket_handles_command_locally_without_available_target(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
client_sock = FakeClientSocket(b"command say local")
event_queue = Mock()
with patch.object(driver, "_get_peer_pid", return_value=1234), patch.object(
driver,
"_find_available_private_socket",
return_value="",
):
driver._handle_client(
client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE
)
event_queue.put.assert_called_once_with(
{
"Type": unixDriver.FenrirEventType.remote_incomming,
"data": "command say local",
}
)
def test_vcsa_main_socket_owner_handles_command_locally(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
driver.bound_sockets = [(Mock(), unixDriver.MAIN_SOCKET_FILE)]
client_sock = FakeClientSocket(b"command say root")
event_queue = Mock()
with 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 root",
}
)
def test_private_socket_handles_command_locally(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
client_sock = FakeClientSocket(b"command say private")
event_queue = Mock()
driver._handle_client(
client_sock, event_queue, "/tmp/fenrirscreenreader-222.sock"
)
event_queue.put.assert_called_once_with(
{
"Type": unixDriver.FenrirEventType.remote_incomming,
"data": "command say private",
}
)
def test_list_command_still_returns_response_from_main_socket(mock_environment):
driver = unixDriver.driver()
driver.env = mock_environment
mock_environment["runtime"]["RemoteManager"] = Mock()
client_sock = FakeClientSocket(b"command list")
event_queue = Mock()
mock_environment["runtime"][
"RemoteManager"
].handle_remote_incomming_with_response.return_value = {
"success": True,
"message": "pid=1",
}
driver._handle_client(client_sock, event_queue, unixDriver.MAIN_SOCKET_FILE)
assert client_sock.sent == b"pid=1\n"
event_queue.put.assert_not_called()
def test_get_parent_pid_parses_process_names_with_spaces():
driver = unixDriver.driver()
stat_file = mock_open(
read_data="1234 (name with spaces) S 567 1 1 0 -1 0\n"
)
with patch("builtins.open", stat_file):
assert driver._get_parent_pid(1234) == 567
def test_find_available_private_socket_prefers_ancestor_socket(
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, "_is_socket_active", return_value=True):
assert (
driver._find_available_private_socket(
"/tmp/fenrirscreenreader-222.sock"
)
== "/tmp/fenrirscreenreader-222.sock"
)
def test_find_available_private_socket_skips_main_socket(
mock_environment,
):
driver = unixDriver.driver()
driver.env = mock_environment
with patch(
"fenrirscreenreader.remoteDriver.unixDriver.remoteInstanceRegistry.list_instances",
return_value=[
{
"pid": 111,
"socket_files": [
unixDriver.MAIN_SOCKET_FILE,
"/tmp/fenrirscreenreader-111.sock",
],
}
],
), patch.object(driver, "_is_socket_active", return_value=True):
assert (
driver._find_available_private_socket()
== "/tmp/fenrirscreenreader-111.sock"
)
+50 -2
View File
@@ -284,16 +284,64 @@ def test_x11_parse_window_id_accepts_decimal_and_hex():
@pytest.mark.unit
def test_x11_write_event_buffer_clears_buffer():
def test_x11_write_event_buffer_replays_grabbed_key_press():
x11 = X11Driver()
x11._initialized = True
x11.env = {"input": {"event_buffer": ["KEY_KP0"]}}
x11.display = Mock()
x11.env = {
"input": {
"event_buffer": [
{
"event_name": "KEY_LEFTMETA",
"event_state": 1,
"event_type": X.KeyPress,
"event_x_time": 1234,
}
]
}
}
x11.write_event_buffer()
x11.display.allow_events.assert_called_once_with(X.ReplayKeyboard, 1234)
assert x11.env["input"]["event_buffer"] == []
@pytest.mark.unit
def test_x11_write_event_buffer_does_not_replay_key_release():
x11 = X11Driver()
x11._initialized = True
x11.display = Mock()
x11.env = {
"input": {
"event_buffer": [
{
"event_name": "KEY_LEFTMETA",
"event_state": 0,
"event_type": X.KeyRelease,
"event_x_time": 1235,
}
]
}
}
x11.write_event_buffer()
x11.display.allow_events.assert_not_called()
assert x11.env["input"]["event_buffer"] == []
@pytest.mark.unit
def test_x11_map_event_keeps_x_event_time_for_replay():
x11 = X11Driver()
x11.keycode_to_key_name = Mock(return_value="KEY_LEFTMETA")
event = Mock(type=X.KeyPress, detail=133, state=X.Mod4Mask, time=5678)
input_event = x11.map_event(event)
assert input_event["event_x_time"] == 5678
@pytest.mark.unit
def test_x11_find_num_lock_mask_uses_modifier_mapping():
x11 = X11Driver()
+3 -2
View File
@@ -1,8 +1,9 @@
#!/bin/bash
# shellcheck disable=SC2329
cleanup() {
# Make sure Fenrir is restored on exit of this script
echo -n "setting set screen#suspendingScreen=" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo -n "setting set screen#suspendingScreen=" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
}
# Call the cleanup function on exit of this script
@@ -20,7 +21,7 @@ if ! [[ "$term" =~ ^[1-9]+$ ]]; then
fi
# Suspend the current terminal for Fenrir
echo -n "setting set screen#suspendingScreen=$term" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-deamon.sock
echo -n "setting set screen#suspendingScreen=$term" | socat - UNIX-CLIENT:/tmp/fenrirscreenreader-daemon.sock
# Start the x session
command startx