9 Commits

18 changed files with 326 additions and 372 deletions
+3 -2
View File
@@ -460,13 +460,14 @@ setting <action> [parameters]
- `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#hardware_device=auto` - Hardware synth serial device for dectalkDriver/litetalkDriver
- `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
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.
device would require a separate USB protocol driver. Use an explicit
`speech#hardware_device` path for hardware speech.
- `speech#auto_read_incoming=True/False` - Auto-read new text
*Sound Settings:*
+2 -2
View File
@@ -79,12 +79,12 @@ volume=1.0
# 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*.
# Set an explicit device for hardware speech.
# 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
hardware_device=/dev/ttyS0
# Serial baud rate for hardware speech synthesizers.
hardware_baud_rate=9600
+1 -4
View File
@@ -1684,12 +1684,9 @@ 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 speech drivers use a serial device. Set an explicit path.
....
hardware_device=auto
hardware_device=/dev/ttyACM0
hardware_device=/dev/ttyUSB0
hardware_device=/dev/ttyS0
+5 -5
View File
@@ -101,7 +101,7 @@ driver=speechdDriver
rate=0.5
pitch=0.5
volume=1.0
hardware_device=auto
hardware_device=/dev/ttyS0
hardware_baud_rate=9600
history_size=50
@@ -341,10 +341,10 @@ Fenrir automatically detects and provides audio feedback for progress indicators
- **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
For hardware speech, set `speech#hardware_device` to 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
+1 -2
View File
@@ -927,8 +927,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. The default ''auto'' checks /dev/ttyACM* first, then /dev/ttyUSB*. Set an explicit path for stable systems.
hardware_device=auto
Hardware speech drivers use a serial device. Set an explicit path.
hardware_device=/dev/ttyACM0
hardware_device=/dev/ttyUSB0
hardware_device=/dev/ttyS0
@@ -82,11 +82,12 @@ class command:
if (
delta_length > 200
): # Allow longer progress lines like Claude Code's status
self.env["runtime"]["DebugManager"].write_debug_out(
f"Progress filter: delta too long ({delta_length})",
debug.DebugLevel.INFO,
)
return False
if not self.is_explicit_progress_delta(delta_text):
self.env["runtime"]["DebugManager"].write_debug_out(
f"Progress filter: delta too long ({delta_length})",
debug.DebugLevel.INFO,
)
return False
# If delta contains newlines and is substantial, let incoming handler
# deal with it to avoid interfering with multi-line text output
@@ -107,6 +108,25 @@ class command:
return True
def is_explicit_progress_delta(self, text):
"""Allow long single-line deltas that still look like progress output."""
import re
if "\n" in text or self.contains_url(text):
return False
has_percentage = re.search(r"(^|\s)\d+(?:\.\d+)?\s*%", text)
if not has_percentage:
return False
return bool(
re.search(
r"[|\[\]#=*>█▉▊▋▌▍▎▏▒▓░]"
r"|\b\d+(?:\.\d+)?\s*[kKmMgGtT](?:i?B)?/s\b",
text,
)
)
def reset_progress_state(self):
"""Reset progress state when a prompt is detected, allowing new progress operations to start fresh"""
self.env["runtime"]["DebugManager"].write_debug_out(
@@ -127,7 +127,7 @@ 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_device", "/dev/ttyS0")
self.config.set("speech", "hardware_baud_rate", "9600")
self.config.add_section("sound")
@@ -108,7 +108,7 @@ class command(config_command):
"rate": "0.5",
"pitch": "0.5",
"volume": "1.0",
"hardware_device": "auto",
"hardware_device": "/dev/ttyS0",
"hardware_baud_rate": "9600",
"auto_read_incoming": "True",
}
+4 -1
View File
@@ -311,7 +311,10 @@ class FenrirManager:
self.singleKeyCommand = True
elif (
(
self.environment["runtime"]["DiffReviewManager"].is_active()
self.environment["runtime"]["VmenuManager"].get_active()
or self.environment["runtime"][
"DiffReviewManager"
].is_active()
or self.environment["runtime"][
"SpeechHistoryManager"
].is_active()
+1 -1
View File
@@ -28,7 +28,7 @@ settings_data = {
"module": "",
"voice": "en-us",
"language": "",
"hardware_device": "auto",
"hardware_device": "/dev/ttyS0",
"hardware_baud_rate": 9600,
"auto_read_incoming": True,
"read_numbers_as_digits": False,
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
version = "2026.05.23"
version = "2026.05.24"
code_name = "master"
@@ -333,10 +333,79 @@ class driver(screenDriver):
def handle_stdin_input(self, msg_bytes, event_queue):
if self.synthesize_backspace_shortcut(msg_bytes, event_queue):
return
if self.handle_vmenu_stdin_input(msg_bytes, event_queue):
return
self.record_stdin_keypress(msg_bytes)
self.interrupt_output_on_stdin_input(msg_bytes)
self.inject_text_to_screen(msg_bytes)
def handle_vmenu_stdin_input(self, msg_bytes, event_queue):
if not self.is_vmenu_active():
return False
key_name = self.vmenu_stdin_key_name(msg_bytes)
if key_name and not self.vmenu_key_already_handled(key_name):
self.queue_keypress(key_name, event_queue)
return True
def is_vmenu_active(self):
try:
return self.env["runtime"]["VmenuManager"].get_active()
except Exception:
return False
def vmenu_stdin_key_name(self, msg_bytes):
key_map = {
b"\x1b": "KEY_ESC",
b"\x1b[A": "KEY_UP",
b"\x1b[B": "KEY_DOWN",
b"\x1b[C": "KEY_RIGHT",
b"\x1b[D": "KEY_LEFT",
b"\x1b[5~": "KEY_PAGEUP",
b"\x1b[6~": "KEY_PAGEDOWN",
b"\r": "KEY_ENTER",
b"\n": "KEY_ENTER",
b" ": "KEY_SPACE",
}
if msg_bytes in key_map:
return key_map[msg_bytes]
if len(msg_bytes) != 1:
return None
char = chr(msg_bytes[0])
if "a" <= char <= "z" or "A" <= char <= "Z":
return "KEY_" + char.upper()
return None
def vmenu_key_already_handled(self, key_name):
try:
return key_name in self.env["input"]["curr_input"]
except Exception:
return False
def queue_keypress(self, key_name, event_queue):
event_time = time.time()
for event_state in [1, 0]:
try:
event_queue.put(
{
"Type": FenrirEventType.keyboard_input,
"data": {
"event_name": key_name,
"event_value": 0,
"event_sec": int(event_time),
"event_usec": int((event_time % 1) * 1000000),
"event_state": event_state,
"event_type": 0,
},
},
block=False,
)
except Full:
self.env["runtime"]["DebugManager"].write_debug_out(
"ptyDriver queue_keypress: Event queue full, dropping "
+ key_name,
debug.DebugLevel.WARNING,
)
def record_stdin_keypress(self, msg_bytes):
if msg_bytes != b"\t":
return
@@ -4,9 +4,7 @@
# Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors.
import glob
import os
import select
import termios
import threading
import tty
@@ -29,8 +27,6 @@ class SpeakQueue(Queue):
class hardware_serial_driver(speech_driver):
cancel_command = b""
default_baud_rate = 9600
hardware_probe_command = b""
hardware_probe_timeout = 0.2
def __init__(self):
speech_driver.__init__(self)
@@ -71,9 +67,9 @@ class hardware_serial_driver(speech_driver):
def _clean_device_setting(self, device):
if not isinstance(device, str):
return "auto"
return ""
device = device.split("#", 1)[0].split(";", 1)[0].strip()
return device or "auto"
return device
def shutdown(self):
if not self._is_initialized:
@@ -156,50 +152,20 @@ class hardware_serial_driver(speech_driver):
)
def _open_serial_port(self):
auto_detect = self.device == "auto"
devices = self._resolve_devices(self.device)
if not devices:
if not self.device or self.device == "auto":
self._debug(
"Hardware speech device not found",
"Hardware speech requires an explicit serial device",
debug.DebugLevel.ERROR,
on_any_level=True,
)
return
fallback_device = None
fallback_port = None
for device in devices:
port = self._open_configured_serial_port(device)
if port is None:
continue
if not auto_detect:
self._activate_serial_port(
device, port, probe_matched=False
)
return
if self._probe_serial_port(port):
if fallback_port is not None:
self._close_port(fallback_port)
self._activate_serial_port(device, port, probe_matched=True)
return
if fallback_port is None:
fallback_device = device
fallback_port = port
else:
self._close_port(port)
if fallback_port is not None:
self._debug(
"Hardware speech probe did not identify a synth; "
f"falling back to {fallback_device}",
debug.DebugLevel.WARNING,
on_any_level=True,
)
self._activate_serial_port(
fallback_device, fallback_port, probe_matched=False
)
port = self._open_configured_serial_port(self.device)
if port is not None:
self._activate_serial_port(self.device, port)
def _open_configured_serial_port(self, device):
port = None
try:
port = os.open(device, os.O_RDWR | os.O_NOCTTY)
tty.setraw(port)
@@ -214,6 +180,7 @@ class hardware_serial_driver(speech_driver):
termios.tcsetattr(port, termios.TCSANOW, attrs)
return port
except (OSError, termios.error) as error:
self._close_port(port)
self._debug(
f"Hardware speech device open failed: {device}: {error}",
debug.DebugLevel.ERROR,
@@ -221,60 +188,16 @@ class hardware_serial_driver(speech_driver):
)
return None
def _activate_serial_port(self, device, port, probe_matched=False):
def _activate_serial_port(self, device, port):
self.serial_port = port
self.device = device
probe_status = "matched" if probe_matched else "not matched"
self._debug(
"Hardware speech device opened: "
f"{device}, baud_rate={self.baud_rate}, probe={probe_status}",
f"{device}, baud_rate={self.baud_rate}",
debug.DebugLevel.INFO,
on_any_level=True,
)
def _probe_serial_port(self, port):
if not self.hardware_probe_command:
return True
try:
self._drain_serial_input(port)
os.write(port, self.hardware_probe_command)
readable, _, _ = select.select(
[port], [], [], self.hardware_probe_timeout
)
if not readable:
self._debug(
"Hardware speech probe got no response",
debug.DebugLevel.INFO,
on_any_level=True,
)
return False
response = os.read(port, 256)
self._debug(
"Hardware speech probe response: "
f"{self._format_bytes_preview(response)}",
debug.DebugLevel.INFO,
on_any_level=True,
)
return self._is_hardware_probe_response(response)
except OSError as error:
self._debug(
f"Hardware speech probe failed: {error}",
debug.DebugLevel.WARNING,
on_any_level=True,
)
return False
def _drain_serial_input(self, port):
while True:
readable, _, _ = select.select([port], [], [], 0)
if not readable:
return
if not os.read(port, 256):
return
def _is_hardware_probe_response(self, response):
return bool(response)
def _close_serial_port(self):
with self.lock:
if self.serial_port is None:
@@ -322,38 +245,6 @@ class hardware_serial_driver(speech_driver):
on_any_level=True,
)
def _resolve_devices(self, device):
if device and device != "auto":
self._debug(
f"Hardware speech using configured device: {device}",
debug.DebugLevel.INFO,
on_any_level=True,
)
return [device]
devices = []
for pattern in ("/dev/ttyACM*", "/dev/ttyUSB*", "/dev/ttyS*"):
matches = sorted(glob.glob(pattern))
self._debug(
f"Hardware speech auto scan {pattern}: {matches}",
debug.DebugLevel.INFO,
on_any_level=True,
)
if matches:
if len(matches) > 1:
self._debug(
"Hardware speech auto found multiple devices for "
f"{pattern}: {matches}; probing in order",
debug.DebugLevel.WARNING,
on_any_level=True,
)
self._debug(
f"Hardware speech auto candidate devices: {matches}",
debug.DebugLevel.INFO,
on_any_level=True,
)
devices.extend(matches)
return devices
def _termios_baud_rate(self, baud_rate):
baud_name = f"B{baud_rate}"
if hasattr(termios, baud_name):
@@ -11,7 +11,6 @@ from fenrirscreenreader.speechDriver.hardwareSerialDriver import (
class driver(hardware_serial_driver):
cancel_command = b"\x18"
hardware_probe_command = b"\x01I"
def _speak_bytes(self, text):
return self._clean_text(text).encode("ascii", errors="replace") + b"\r"
@@ -27,10 +26,3 @@ class driver(hardware_serial_driver):
def _setting_command(self, value, command):
return b"\x01" + str(value).encode("ascii") + command
def _is_hardware_probe_response(self, response):
response_text = response.decode("ascii", errors="ignore").lower()
return any(
token in response_text
for token in ("litetalk", "lite talk", "rpitalk", "doubletalk")
)
@@ -22,6 +22,7 @@ def test_speech_history_plain_key_modal_command_is_dispatched():
)
speech_history_manager = Mock(is_active=Mock(return_value=True))
diff_review_manager = Mock(is_active=Mock(return_value=False))
vmenu_manager = Mock(get_active=Mock(return_value=False))
manager.environment = {
"input": {
@@ -32,6 +33,7 @@ def test_speech_history_plain_key_modal_command_is_dispatched():
"runtime": {
"InputManager": input_manager,
"EventManager": event_manager,
"VmenuManager": vmenu_manager,
"DiffReviewManager": diff_review_manager,
"SpeechHistoryManager": speech_history_manager,
},
@@ -42,3 +44,43 @@ def test_speech_history_plain_key_modal_command_is_dispatched():
event_manager.put_to_event_queue.assert_called_once_with(
FenrirEventType.execute_command, "SPEECH_HISTORY_PREV"
)
@pytest.mark.unit
def test_vmenu_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="PREV_VMENU_ENTRY"),
)
vmenu_manager = Mock(get_active=Mock(return_value=True))
speech_history_manager = Mock(is_active=Mock(return_value=False))
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,
"VmenuManager": vmenu_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, "PREV_VMENU_ENTRY"
)
+5 -219
View File
@@ -1,6 +1,5 @@
import os
import select
import termios
import time
from unittest.mock import ANY
from unittest.mock import Mock
@@ -105,26 +104,14 @@ def test_litetalk_driver_writes_settings_and_cancel(serial_pair):
speech_driver.shutdown()
def test_auto_device_detection_includes_classic_serial(
monkeypatch, serial_pair
):
def test_configured_device_supports_classic_serial(serial_pair):
master_fd, slave_name = serial_pair
def fake_glob(pattern):
if pattern == "/dev/ttyS*":
return [slave_name]
return []
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.glob.glob",
fake_glob,
)
speech_driver = litetalkDriver.driver()
speech_driver.initialize(build_environment("auto"))
speech_driver.initialize(build_environment(slave_name))
try:
assert speech_driver.device == slave_name
speech_driver.speak("Serial")
assert read_available(master_fd, 9) == b"\x01ISerial\r"
assert read_available(master_fd, 7) == b"Serial\r"
finally:
speech_driver.shutdown()
@@ -142,208 +129,7 @@ def test_configured_device_strips_inline_comment(serial_pair):
speech_driver.shutdown()
def test_auto_device_detection_prefers_probe_response(monkeypatch):
opened_ports = []
closed_ports = []
writes = {}
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.glob.glob",
lambda pattern: ["/dev/ttyUSB0", "/dev/ttyUSB1"]
if pattern == "/dev/ttyUSB*"
else [],
)
def fake_open(device, flags):
port = 100 + len(opened_ports)
opened_ports.append((device, port))
return port
def fake_write(port, data):
writes.setdefault(port, b"")
writes[port] += data
return len(data)
def fake_select(readable, writable, exceptional, timeout):
port = readable[0]
if port == 101 and writes.get(port) == b"\x01I":
return readable, writable, exceptional
return [], writable, exceptional
def fake_read(port, size):
if port == 101:
return b"RPItalk 1.3\r"
return b""
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.open",
fake_open,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.close",
lambda port: closed_ports.append(port),
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.write",
fake_write,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.read",
fake_read,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.select.select",
fake_select,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.termios.tcgetattr",
lambda port: [0, 0, 0, 0, 0, 0, [0] * 32],
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.termios.tcsetattr",
lambda port, when, attrs: None,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.tty.setraw",
lambda port: None,
)
speech_driver = litetalkDriver.driver()
speech_driver.initialize(build_environment("auto"))
try:
assert opened_ports == [("/dev/ttyUSB0", 100), ("/dev/ttyUSB1", 101)]
assert writes == {100: b"\x01I", 101: b"\x01I"}
assert closed_ports == [100]
assert speech_driver.device == "/dev/ttyUSB1"
finally:
speech_driver.shutdown()
def test_auto_device_detection_falls_back_without_probe_response(
monkeypatch
):
opened_ports = []
closed_ports = []
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.glob.glob",
lambda pattern: ["/dev/ttyUSB0", "/dev/ttyUSB1"]
if pattern == "/dev/ttyUSB*"
else [],
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.open",
lambda device, flags: opened_ports.append(device) or len(opened_ports),
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.close",
lambda port: closed_ports.append(port),
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.write",
lambda port, data: len(data),
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.select.select",
lambda readable, writable, exceptional, timeout: (
[],
writable,
exceptional,
),
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.termios.tcgetattr",
lambda port: [0, 0, 0, 0, 0, 0, [0] * 32],
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.termios.tcsetattr",
lambda port, when, attrs: None,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.tty.setraw",
lambda port: None,
)
speech_driver = litetalkDriver.driver()
speech_driver.initialize(build_environment("auto"))
try:
assert opened_ports == ["/dev/ttyUSB0", "/dev/ttyUSB1"]
assert closed_ports == [2]
assert speech_driver.device == "/dev/ttyUSB0"
finally:
speech_driver.shutdown()
def test_auto_device_detection_skips_termios_failures(monkeypatch):
opened_ports = []
closed_ports = []
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.glob.glob",
lambda pattern: ["/dev/ttyUSB0"]
if pattern == "/dev/ttyUSB*"
else ["/dev/ttyS0"]
if pattern == "/dev/ttyS*"
else [],
)
def fake_open(device, flags):
port = 100 + len(opened_ports)
opened_ports.append((device, port))
return port
def fake_tcgetattr(port):
if port == 101:
raise termios.error(5, "Input/output error")
return [0, 0, 0, 0, 0, 0, [0] * 32]
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.open",
fake_open,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.close",
lambda port: closed_ports.append(port),
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.os.write",
lambda port, data: len(data),
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.select.select",
lambda readable, writable, exceptional, timeout: (
[],
writable,
exceptional,
),
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.termios.tcgetattr",
fake_tcgetattr,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.termios.tcsetattr",
lambda port, when, attrs: None,
)
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.tty.setraw",
lambda port: None,
)
speech_driver = litetalkDriver.driver()
speech_driver.initialize(build_environment("auto"))
try:
assert opened_ports == [("/dev/ttyUSB0", 100), ("/dev/ttyS0", 101)]
assert speech_driver.device == "/dev/ttyUSB0"
finally:
speech_driver.shutdown()
def test_auto_device_detection_fails_when_no_serial_device(monkeypatch):
monkeypatch.setattr(
"fenrirscreenreader.speechDriver.hardwareSerialDriver.glob.glob",
lambda pattern: [],
)
def test_auto_device_is_rejected():
speech_driver = litetalkDriver.driver()
with pytest.raises(RuntimeError, match="hardware speech device"):
@@ -351,7 +137,7 @@ def test_auto_device_detection_fails_when_no_serial_device(monkeypatch):
debug_manager = speech_driver.env["runtime"]["DebugManager"]
debug_manager.write_debug_out.assert_called_with(
"Hardware speech device not found",
"Hardware speech requires an explicit serial device",
ANY,
on_any_level=True,
)
+70
View File
@@ -39,3 +39,73 @@ def test_progress_detector_skips_typing_delta():
command.is_real_progress_update.assert_not_called()
command.detect_progress.assert_not_called()
@pytest.mark.unit
def test_progress_detector_allows_long_tqdm_transfer_delta():
progress_module = _load_progress_module()
command = progress_module.command()
sample = (
"88%|"
"████████████████████████████████████████████████████████████████"
"████████████████████████████████████████████████████████████████"
"████████████████████████████████████████████████████████████████"
"█████████████████████████▊ "
"| 843M/954M [00:54<00:07, 15.2MB/s]"
)
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_long_tqdm_transfer_delta():
progress_module = _load_progress_module()
command = progress_module.command()
sample = (
"90%|"
"████████████████████████████████████████████████████████████████"
"████████████████████████████████████████████████████████████████"
"████████████████████████████████████████████████████████████████"
"█████████████████████████████████████▍ "
"| 856M/954M [00:56<00:14, 6.78MB/s]"
)
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(90.0)
assert command.env["commandBuffer"]["lastProgressValue"] == 90.0
+84
View File
@@ -160,6 +160,90 @@ def test_pty_plain_stdin_does_not_record_tab_keypress():
pty_driver.inject_text_to_screen.assert_called_once_with(b"a")
@pytest.mark.unit
@pytest.mark.parametrize(
("sequence", "key_name"),
[
(b"\x1b[A", "KEY_UP"),
(b"\x1b[B", "KEY_DOWN"),
(b"\x1b[C", "KEY_RIGHT"),
(b"\x1b[D", "KEY_LEFT"),
(b"\x1b[5~", "KEY_PAGEUP"),
(b"\x1b[6~", "KEY_PAGEDOWN"),
(b"\x1b", "KEY_ESC"),
(b"\r", "KEY_ENTER"),
(b" ", "KEY_SPACE"),
(b"a", "KEY_A"),
(b"Z", "KEY_Z"),
],
)
def test_pty_vmenu_stdin_is_consumed_and_synthesizes_key_events(
sequence,
key_name,
):
pty_driver = PtyDriver()
event_queue = Mock()
settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False
pty_driver.env = {
"input": {"curr_input": []},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"SettingsManager": settings_manager,
"VmenuManager": Mock(get_active=Mock(return_value=True)),
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(sequence, event_queue)
pty_driver.inject_text_to_screen.assert_not_called()
assert event_queue.put.call_count == 2
first_event = event_queue.put.call_args_list[0].args[0]
second_event = event_queue.put.call_args_list[1].args[0]
assert first_event["Type"] == FenrirEventType.keyboard_input
assert first_event["data"]["event_name"] == key_name
assert first_event["data"]["event_state"] == 1
assert second_event["data"]["event_name"] == key_name
assert second_event["data"]["event_state"] == 0
@pytest.mark.unit
def test_pty_vmenu_unknown_stdin_is_consumed_without_injection():
pty_driver = PtyDriver()
event_queue = Mock()
pty_driver.env = {
"input": {"curr_input": []},
"runtime": {
"VmenuManager": Mock(get_active=Mock(return_value=True)),
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"\x1b[1;5A", event_queue)
pty_driver.inject_text_to_screen.assert_not_called()
event_queue.put.assert_not_called()
@pytest.mark.unit
def test_pty_vmenu_stdin_does_not_duplicate_current_x11_key():
pty_driver = PtyDriver()
event_queue = Mock()
pty_driver.env = {
"input": {"curr_input": ["KEY_RIGHT"]},
"runtime": {
"VmenuManager": Mock(get_active=Mock(return_value=True)),
},
}
pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"\x1b[C", event_queue)
pty_driver.inject_text_to_screen.assert_not_called()
event_queue.put.assert_not_called()
@pytest.mark.unit
def test_pty_stdin_input_honors_interrupt_disabled():
pty_driver = PtyDriver()