Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bd54ec0edb | |||
| b9518f52ec | |||
| c143c9a561 | |||
| 7e2f927596 | |||
| 788e678ed6 | |||
| ea89e90c2f | |||
| ce43d64e77 | |||
| 618987546a | |||
| 604221a29d |
@@ -460,13 +460,14 @@ setting <action> [parameters]
|
|||||||
- `speech#voice=voice_name` - Voice selection (e.g., "en-us+f3")
|
- `speech#voice=voice_name` - Voice selection (e.g., "en-us+f3")
|
||||||
- `speech#module=module_name` - TTS module (e.g., "espeak-ng")
|
- `speech#module=module_name` - TTS module (e.g., "espeak-ng")
|
||||||
- `speech#driver=driver_name` - Speech driver (speechdDriver/genericDriver/dectalkDriver/litetalkDriver/doubletalkDriver/tripletalkDriver)
|
- `speech#driver=driver_name` - Speech driver (speechdDriver/genericDriver/dectalkDriver/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#hardware_baud_rate=9600` - Hardware synth serial baud rate
|
||||||
- `speech#history_size=50` - Number of spoken items kept in runtime speech history
|
- `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
|
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
|
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
|
- `speech#auto_read_incoming=True/False` - Auto-read new text
|
||||||
|
|
||||||
*Sound Settings:*
|
*Sound Settings:*
|
||||||
|
|||||||
@@ -79,12 +79,12 @@ volume=1.0
|
|||||||
# Used by dectalkDriver, litetalkDriver, doubletalkDriver, and tripletalkDriver.
|
# Used by dectalkDriver, litetalkDriver, doubletalkDriver, and tripletalkDriver.
|
||||||
# USB serial devices are supported if Linux exposes them as /dev/ttyACM*
|
# 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.
|
# 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:
|
# Examples:
|
||||||
# hardware_device=/dev/ttyACM0 # RPITalk USB gadget mode
|
# hardware_device=/dev/ttyACM0 # RPITalk USB gadget mode
|
||||||
# hardware_device=/dev/ttyUSB0 # USB serial adapter
|
# hardware_device=/dev/ttyUSB0 # USB serial adapter
|
||||||
# hardware_device=/dev/ttyS0 # built-in serial port
|
# hardware_device=/dev/ttyS0 # built-in serial port
|
||||||
hardware_device=auto
|
hardware_device=/dev/ttyS0
|
||||||
|
|
||||||
# Serial baud rate for hardware speech synthesizers.
|
# Serial baud rate for hardware speech synthesizers.
|
||||||
hardware_baud_rate=9600
|
hardware_baud_rate=9600
|
||||||
|
|||||||
+1
-4
@@ -1684,12 +1684,9 @@ the pico module:
|
|||||||
language=de-DE
|
language=de-DE
|
||||||
....
|
....
|
||||||
|
|
||||||
Hardware speech drivers use a serial device. The default `+auto+` checks
|
Hardware speech drivers use a serial device. Set an explicit path.
|
||||||
`+/dev/ttyACM*+` first, then `+/dev/ttyUSB*+`. Set an explicit path for
|
|
||||||
stable systems.
|
|
||||||
|
|
||||||
....
|
....
|
||||||
hardware_device=auto
|
|
||||||
hardware_device=/dev/ttyACM0
|
hardware_device=/dev/ttyACM0
|
||||||
hardware_device=/dev/ttyUSB0
|
hardware_device=/dev/ttyUSB0
|
||||||
hardware_device=/dev/ttyS0
|
hardware_device=/dev/ttyS0
|
||||||
|
|||||||
+5
-5
@@ -101,7 +101,7 @@ driver=speechdDriver
|
|||||||
rate=0.5
|
rate=0.5
|
||||||
pitch=0.5
|
pitch=0.5
|
||||||
volume=1.0
|
volume=1.0
|
||||||
hardware_device=auto
|
hardware_device=/dev/ttyS0
|
||||||
hardware_baud_rate=9600
|
hardware_baud_rate=9600
|
||||||
history_size=50
|
history_size=50
|
||||||
|
|
||||||
@@ -341,10 +341,10 @@ Fenrir automatically detects and provides audio feedback for progress indicators
|
|||||||
- **doubletalkDriver** - Serial DoubleTalk LT-compatible hardware speech
|
- **doubletalkDriver** - Serial DoubleTalk LT-compatible hardware speech
|
||||||
- **tripletalkDriver** - Serial TripleTalk-compatible hardware speech
|
- **tripletalkDriver** - Serial TripleTalk-compatible hardware speech
|
||||||
|
|
||||||
For hardware speech, set `speech#hardware_device` to `auto` or an explicit
|
For hardware speech, set `speech#hardware_device` to an explicit serial path.
|
||||||
serial path. RPITalk gadget mode usually appears as `/dev/ttyACM0`; USB serial
|
RPITalk gadget mode usually appears as `/dev/ttyACM0`; USB serial adapters
|
||||||
adapters usually appear as `/dev/ttyUSB0`; built-in serial ports may be
|
usually appear as `/dev/ttyUSB0`; built-in serial ports may be `/dev/ttyS0`.
|
||||||
`/dev/ttyS0`. The default baud rate is `9600`. `doubletalkDriver` targets
|
The default baud rate is `9600`. `doubletalkDriver` targets
|
||||||
DoubleTalk LT-style serial devices, not the internal DoubleTalk PC ISA card.
|
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
|
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
|
`/dev/ttyACM0` or `/dev/ttyUSB0`; USB-only models with no tty device need a
|
||||||
|
|||||||
+1
-2
@@ -927,8 +927,7 @@ Select the language you want Fenrir to use.
|
|||||||
language=english-us
|
language=english-us
|
||||||
Values: Text, see your TTS synths documentation what is available.
|
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 speech drivers use a serial device. Set an explicit path.
|
||||||
hardware_device=auto
|
|
||||||
hardware_device=/dev/ttyACM0
|
hardware_device=/dev/ttyACM0
|
||||||
hardware_device=/dev/ttyUSB0
|
hardware_device=/dev/ttyUSB0
|
||||||
hardware_device=/dev/ttyS0
|
hardware_device=/dev/ttyS0
|
||||||
|
|||||||
@@ -82,11 +82,12 @@ class command:
|
|||||||
if (
|
if (
|
||||||
delta_length > 200
|
delta_length > 200
|
||||||
): # Allow longer progress lines like Claude Code's status
|
): # Allow longer progress lines like Claude Code's status
|
||||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
if not self.is_explicit_progress_delta(delta_text):
|
||||||
f"Progress filter: delta too long ({delta_length})",
|
self.env["runtime"]["DebugManager"].write_debug_out(
|
||||||
debug.DebugLevel.INFO,
|
f"Progress filter: delta too long ({delta_length})",
|
||||||
)
|
debug.DebugLevel.INFO,
|
||||||
return False
|
)
|
||||||
|
return False
|
||||||
|
|
||||||
# If delta contains newlines and is substantial, let incoming handler
|
# If delta contains newlines and is substantial, let incoming handler
|
||||||
# deal with it to avoid interfering with multi-line text output
|
# deal with it to avoid interfering with multi-line text output
|
||||||
@@ -107,6 +108,25 @@ class command:
|
|||||||
|
|
||||||
return True
|
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):
|
def reset_progress_state(self):
|
||||||
"""Reset progress state when a prompt is detected, allowing new progress operations to start fresh"""
|
"""Reset progress state when a prompt is detected, allowing new progress operations to start fresh"""
|
||||||
self.env["runtime"]["DebugManager"].write_debug_out(
|
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", "rate", "0.75")
|
||||||
self.config.set("speech", "pitch", "0.5")
|
self.config.set("speech", "pitch", "0.5")
|
||||||
self.config.set("speech", "volume", "1.0")
|
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.set("speech", "hardware_baud_rate", "9600")
|
||||||
|
|
||||||
self.config.add_section("sound")
|
self.config.add_section("sound")
|
||||||
|
|||||||
+1
-1
@@ -108,7 +108,7 @@ class command(config_command):
|
|||||||
"rate": "0.5",
|
"rate": "0.5",
|
||||||
"pitch": "0.5",
|
"pitch": "0.5",
|
||||||
"volume": "1.0",
|
"volume": "1.0",
|
||||||
"hardware_device": "auto",
|
"hardware_device": "/dev/ttyS0",
|
||||||
"hardware_baud_rate": "9600",
|
"hardware_baud_rate": "9600",
|
||||||
"auto_read_incoming": "True",
|
"auto_read_incoming": "True",
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -311,7 +311,10 @@ class FenrirManager:
|
|||||||
self.singleKeyCommand = True
|
self.singleKeyCommand = True
|
||||||
elif (
|
elif (
|
||||||
(
|
(
|
||||||
self.environment["runtime"]["DiffReviewManager"].is_active()
|
self.environment["runtime"]["VmenuManager"].get_active()
|
||||||
|
or self.environment["runtime"][
|
||||||
|
"DiffReviewManager"
|
||||||
|
].is_active()
|
||||||
or self.environment["runtime"][
|
or self.environment["runtime"][
|
||||||
"SpeechHistoryManager"
|
"SpeechHistoryManager"
|
||||||
].is_active()
|
].is_active()
|
||||||
|
|||||||
@@ -28,7 +28,7 @@ settings_data = {
|
|||||||
"module": "",
|
"module": "",
|
||||||
"voice": "en-us",
|
"voice": "en-us",
|
||||||
"language": "",
|
"language": "",
|
||||||
"hardware_device": "auto",
|
"hardware_device": "/dev/ttyS0",
|
||||||
"hardware_baud_rate": 9600,
|
"hardware_baud_rate": 9600,
|
||||||
"auto_read_incoming": True,
|
"auto_read_incoming": True,
|
||||||
"read_numbers_as_digits": False,
|
"read_numbers_as_digits": False,
|
||||||
|
|||||||
@@ -4,5 +4,5 @@
|
|||||||
# Fenrir TTY screen reader
|
# Fenrir TTY screen reader
|
||||||
# By Chrys, Storm Dragon, and contributors.
|
# By Chrys, Storm Dragon, and contributors.
|
||||||
|
|
||||||
version = "2026.05.23"
|
version = "2026.05.24"
|
||||||
code_name = "master"
|
code_name = "master"
|
||||||
|
|||||||
@@ -333,10 +333,79 @@ class driver(screenDriver):
|
|||||||
def handle_stdin_input(self, msg_bytes, event_queue):
|
def handle_stdin_input(self, msg_bytes, event_queue):
|
||||||
if self.synthesize_backspace_shortcut(msg_bytes, event_queue):
|
if self.synthesize_backspace_shortcut(msg_bytes, event_queue):
|
||||||
return
|
return
|
||||||
|
if self.handle_vmenu_stdin_input(msg_bytes, event_queue):
|
||||||
|
return
|
||||||
self.record_stdin_keypress(msg_bytes)
|
self.record_stdin_keypress(msg_bytes)
|
||||||
self.interrupt_output_on_stdin_input(msg_bytes)
|
self.interrupt_output_on_stdin_input(msg_bytes)
|
||||||
self.inject_text_to_screen(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):
|
def record_stdin_keypress(self, msg_bytes):
|
||||||
if msg_bytes != b"\t":
|
if msg_bytes != b"\t":
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -4,9 +4,7 @@
|
|||||||
# Fenrir TTY screen reader
|
# Fenrir TTY screen reader
|
||||||
# By Chrys, Storm Dragon, and contributors.
|
# By Chrys, Storm Dragon, and contributors.
|
||||||
|
|
||||||
import glob
|
|
||||||
import os
|
import os
|
||||||
import select
|
|
||||||
import termios
|
import termios
|
||||||
import threading
|
import threading
|
||||||
import tty
|
import tty
|
||||||
@@ -29,8 +27,6 @@ class SpeakQueue(Queue):
|
|||||||
class hardware_serial_driver(speech_driver):
|
class hardware_serial_driver(speech_driver):
|
||||||
cancel_command = b""
|
cancel_command = b""
|
||||||
default_baud_rate = 9600
|
default_baud_rate = 9600
|
||||||
hardware_probe_command = b""
|
|
||||||
hardware_probe_timeout = 0.2
|
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
speech_driver.__init__(self)
|
speech_driver.__init__(self)
|
||||||
@@ -71,9 +67,9 @@ class hardware_serial_driver(speech_driver):
|
|||||||
|
|
||||||
def _clean_device_setting(self, device):
|
def _clean_device_setting(self, device):
|
||||||
if not isinstance(device, str):
|
if not isinstance(device, str):
|
||||||
return "auto"
|
return ""
|
||||||
device = device.split("#", 1)[0].split(";", 1)[0].strip()
|
device = device.split("#", 1)[0].split(";", 1)[0].strip()
|
||||||
return device or "auto"
|
return device
|
||||||
|
|
||||||
def shutdown(self):
|
def shutdown(self):
|
||||||
if not self._is_initialized:
|
if not self._is_initialized:
|
||||||
@@ -156,50 +152,20 @@ class hardware_serial_driver(speech_driver):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def _open_serial_port(self):
|
def _open_serial_port(self):
|
||||||
auto_detect = self.device == "auto"
|
if not self.device or self.device == "auto":
|
||||||
devices = self._resolve_devices(self.device)
|
|
||||||
if not devices:
|
|
||||||
self._debug(
|
self._debug(
|
||||||
"Hardware speech device not found",
|
"Hardware speech requires an explicit serial device",
|
||||||
debug.DebugLevel.ERROR,
|
debug.DebugLevel.ERROR,
|
||||||
on_any_level=True,
|
on_any_level=True,
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
|
|
||||||
fallback_device = None
|
port = self._open_configured_serial_port(self.device)
|
||||||
fallback_port = None
|
if port is not None:
|
||||||
for device in devices:
|
self._activate_serial_port(self.device, port)
|
||||||
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
|
|
||||||
)
|
|
||||||
|
|
||||||
def _open_configured_serial_port(self, device):
|
def _open_configured_serial_port(self, device):
|
||||||
|
port = None
|
||||||
try:
|
try:
|
||||||
port = os.open(device, os.O_RDWR | os.O_NOCTTY)
|
port = os.open(device, os.O_RDWR | os.O_NOCTTY)
|
||||||
tty.setraw(port)
|
tty.setraw(port)
|
||||||
@@ -214,6 +180,7 @@ class hardware_serial_driver(speech_driver):
|
|||||||
termios.tcsetattr(port, termios.TCSANOW, attrs)
|
termios.tcsetattr(port, termios.TCSANOW, attrs)
|
||||||
return port
|
return port
|
||||||
except (OSError, termios.error) as error:
|
except (OSError, termios.error) as error:
|
||||||
|
self._close_port(port)
|
||||||
self._debug(
|
self._debug(
|
||||||
f"Hardware speech device open failed: {device}: {error}",
|
f"Hardware speech device open failed: {device}: {error}",
|
||||||
debug.DebugLevel.ERROR,
|
debug.DebugLevel.ERROR,
|
||||||
@@ -221,60 +188,16 @@ class hardware_serial_driver(speech_driver):
|
|||||||
)
|
)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def _activate_serial_port(self, device, port, probe_matched=False):
|
def _activate_serial_port(self, device, port):
|
||||||
self.serial_port = port
|
self.serial_port = port
|
||||||
self.device = device
|
self.device = device
|
||||||
probe_status = "matched" if probe_matched else "not matched"
|
|
||||||
self._debug(
|
self._debug(
|
||||||
"Hardware speech device opened: "
|
"Hardware speech device opened: "
|
||||||
f"{device}, baud_rate={self.baud_rate}, probe={probe_status}",
|
f"{device}, baud_rate={self.baud_rate}",
|
||||||
debug.DebugLevel.INFO,
|
debug.DebugLevel.INFO,
|
||||||
on_any_level=True,
|
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):
|
def _close_serial_port(self):
|
||||||
with self.lock:
|
with self.lock:
|
||||||
if self.serial_port is None:
|
if self.serial_port is None:
|
||||||
@@ -322,38 +245,6 @@ class hardware_serial_driver(speech_driver):
|
|||||||
on_any_level=True,
|
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):
|
def _termios_baud_rate(self, baud_rate):
|
||||||
baud_name = f"B{baud_rate}"
|
baud_name = f"B{baud_rate}"
|
||||||
if hasattr(termios, baud_name):
|
if hasattr(termios, baud_name):
|
||||||
|
|||||||
@@ -11,7 +11,6 @@ from fenrirscreenreader.speechDriver.hardwareSerialDriver import (
|
|||||||
|
|
||||||
class driver(hardware_serial_driver):
|
class driver(hardware_serial_driver):
|
||||||
cancel_command = b"\x18"
|
cancel_command = b"\x18"
|
||||||
hardware_probe_command = b"\x01I"
|
|
||||||
|
|
||||||
def _speak_bytes(self, text):
|
def _speak_bytes(self, text):
|
||||||
return self._clean_text(text).encode("ascii", errors="replace") + b"\r"
|
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):
|
def _setting_command(self, value, command):
|
||||||
return b"\x01" + str(value).encode("ascii") + 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))
|
speech_history_manager = Mock(is_active=Mock(return_value=True))
|
||||||
diff_review_manager = Mock(is_active=Mock(return_value=False))
|
diff_review_manager = Mock(is_active=Mock(return_value=False))
|
||||||
|
vmenu_manager = Mock(get_active=Mock(return_value=False))
|
||||||
|
|
||||||
manager.environment = {
|
manager.environment = {
|
||||||
"input": {
|
"input": {
|
||||||
@@ -32,6 +33,7 @@ def test_speech_history_plain_key_modal_command_is_dispatched():
|
|||||||
"runtime": {
|
"runtime": {
|
||||||
"InputManager": input_manager,
|
"InputManager": input_manager,
|
||||||
"EventManager": event_manager,
|
"EventManager": event_manager,
|
||||||
|
"VmenuManager": vmenu_manager,
|
||||||
"DiffReviewManager": diff_review_manager,
|
"DiffReviewManager": diff_review_manager,
|
||||||
"SpeechHistoryManager": speech_history_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(
|
event_manager.put_to_event_queue.assert_called_once_with(
|
||||||
FenrirEventType.execute_command, "SPEECH_HISTORY_PREV"
|
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"
|
||||||
|
)
|
||||||
|
|||||||
@@ -1,6 +1,5 @@
|
|||||||
import os
|
import os
|
||||||
import select
|
import select
|
||||||
import termios
|
|
||||||
import time
|
import time
|
||||||
from unittest.mock import ANY
|
from unittest.mock import ANY
|
||||||
from unittest.mock import Mock
|
from unittest.mock import Mock
|
||||||
@@ -105,26 +104,14 @@ def test_litetalk_driver_writes_settings_and_cancel(serial_pair):
|
|||||||
speech_driver.shutdown()
|
speech_driver.shutdown()
|
||||||
|
|
||||||
|
|
||||||
def test_auto_device_detection_includes_classic_serial(
|
def test_configured_device_supports_classic_serial(serial_pair):
|
||||||
monkeypatch, serial_pair
|
|
||||||
):
|
|
||||||
master_fd, slave_name = 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 = litetalkDriver.driver()
|
||||||
speech_driver.initialize(build_environment("auto"))
|
speech_driver.initialize(build_environment(slave_name))
|
||||||
try:
|
try:
|
||||||
assert speech_driver.device == slave_name
|
assert speech_driver.device == slave_name
|
||||||
speech_driver.speak("Serial")
|
speech_driver.speak("Serial")
|
||||||
assert read_available(master_fd, 9) == b"\x01ISerial\r"
|
assert read_available(master_fd, 7) == b"Serial\r"
|
||||||
finally:
|
finally:
|
||||||
speech_driver.shutdown()
|
speech_driver.shutdown()
|
||||||
|
|
||||||
@@ -142,208 +129,7 @@ def test_configured_device_strips_inline_comment(serial_pair):
|
|||||||
speech_driver.shutdown()
|
speech_driver.shutdown()
|
||||||
|
|
||||||
|
|
||||||
def test_auto_device_detection_prefers_probe_response(monkeypatch):
|
def test_auto_device_is_rejected():
|
||||||
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: [],
|
|
||||||
)
|
|
||||||
speech_driver = litetalkDriver.driver()
|
speech_driver = litetalkDriver.driver()
|
||||||
|
|
||||||
with pytest.raises(RuntimeError, match="hardware speech device"):
|
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 = speech_driver.env["runtime"]["DebugManager"]
|
||||||
debug_manager.write_debug_out.assert_called_with(
|
debug_manager.write_debug_out.assert_called_with(
|
||||||
"Hardware speech device not found",
|
"Hardware speech requires an explicit serial device",
|
||||||
ANY,
|
ANY,
|
||||||
on_any_level=True,
|
on_any_level=True,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -39,3 +39,73 @@ def test_progress_detector_skips_typing_delta():
|
|||||||
|
|
||||||
command.is_real_progress_update.assert_not_called()
|
command.is_real_progress_update.assert_not_called()
|
||||||
command.detect_progress.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
|
||||||
|
|||||||
@@ -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")
|
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
|
@pytest.mark.unit
|
||||||
def test_pty_stdin_input_honors_interrupt_disabled():
|
def test_pty_stdin_input_honors_interrupt_disabled():
|
||||||
pty_driver = PtyDriver()
|
pty_driver = PtyDriver()
|
||||||
|
|||||||
Reference in New Issue
Block a user