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

Attempt to decrease dectalk lag. Hoping you don't have to send settings over every time you want it to say something.

Browse Source
This commit is contained in:
Storm Dragon
2026-07-26 21:34:04 -04:00
parent fac3065c6a
commit 17b6e6f909
4 changed files with 81 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fenrirscreenreader/speechDriver/dectalkUsbDriver.py
+7 -4
View File
@@ -27,6 +27,7 @@ class driver(dectalkDriver.driver):
def initialize(self, environment):
self.env = environment
self._is_initialized = False
self._reset_setting_cache()
self.usb_device = None
self.usb_core = None
self.usb_util = None
@@ -187,7 +188,7 @@ class driver(dectalkDriver.driver):
def _write_bytes(self, data, description="data", cancel_generation=None):
if description == "speech" and cancel_generation is None:
cancel_generation = self._get_speech_cancel_generation()
self._write_endpoint(
return self._write_endpoint(
self.speech_endpoint,
data,
description,
@@ -203,12 +204,12 @@ class driver(dectalkDriver.driver):
cancel_generation=None,
):
if not data:
return
return False
if write_lock is None:
write_lock = self.lock
with write_lock:
if self.usb_device is None:
return
return False
try:
total_written = 0
while total_written < len(data):
@@ -220,7 +221,7 @@ class driver(dectalkDriver.driver):
debug.DebugLevel.INFO,
on_any_level=True,
)
return
return False
chunk = data[
total_written : total_written + self.packet_size
]
@@ -243,12 +244,14 @@ class driver(dectalkDriver.driver):
debug.DebugLevel.INFO,
on_any_level=True,
)
return True
except (self.usb_core.USBError, OSError) as error:
self._debug(
f"DECtalk USB write failed: {error}",
debug.DebugLevel.ERROR,
on_any_level=True,
)
return False
def _advance_speech_cancel_generation(self):
with self._speech_cancel_lock:
src/fenrirscreenreader/speechDriver/hardwareSerialDriver.py
+18 -5
View File
@@ -38,10 +38,12 @@ class hardware_serial_driver(speech_driver):
self.lock = threading.Lock()
self.worker_thread = None
self._stop_worker = False
self._setting_cache = {}
def initialize(self, environment):
self.env = environment
self._is_initialized = False
self._reset_setting_cache()
settings_manager = self.env["runtime"]["SettingsManager"]
self.device = self._clean_device_setting(
settings_manager.get_setting("speech", "hardware_device")
@@ -115,21 +117,30 @@ class hardware_serial_driver(speech_driver):
return
if not isinstance(rate, float):
return
self._write_bytes(self._rate_command(rate), "rate")
self._write_setting_command("rate", self._rate_command(rate))
def set_pitch(self, pitch):
if not self._is_initialized:
return
if not isinstance(pitch, float):
return
self._write_bytes(self._pitch_command(pitch), "pitch")
self._write_setting_command("pitch", self._pitch_command(pitch))
def set_volume(self, volume):
if not self._is_initialized:
return
if not isinstance(volume, float):
return
self._write_bytes(self._volume_command(volume), "volume")
self._write_setting_command("volume", self._volume_command(volume))
def _reset_setting_cache(self):
self._setting_cache = {}
def _write_setting_command(self, setting, command):
if self._setting_cache.get(setting) == command:
return
if self._write_bytes(command, setting):
self._setting_cache[setting] = command
def _worker(self):
while not self._stop_worker:
@@ -223,10 +234,10 @@ class hardware_serial_driver(speech_driver):
def _write_bytes(self, data, description="data"):
if not data:
return
return False
with self.lock:
if self.serial_port is None:
return
return False
try:
total_written = 0
while total_written < len(data):
@@ -243,12 +254,14 @@ class hardware_serial_driver(speech_driver):
debug.DebugLevel.INFO,
on_any_level=True,
)
return True
except OSError as error:
self._debug(
f"Hardware speech write failed: {error}",
debug.DebugLevel.ERROR,
on_any_level=True,
)
return False
def _termios_baud_rate(self, baud_rate):
baud_name = f"B{baud_rate}"
tests/unit/test_dectalk_usb_driver.py
+19
View File
@@ -161,6 +161,25 @@ def test_dectalk_usb_driver_writes_commands_and_speech(monkeypatch):
speech_driver.shutdown()
def test_dectalk_usb_driver_skips_redundant_setting_writes(monkeypatch):
speech_driver, device, _, _ = initialized_driver(monkeypatch)
try:
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
speech_driver.set_volume(1.0)
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
speech_driver.set_volume(1.0)
assert device.writes == [
(0x02, b"[:ra 180]", 1000),
(0x02, b"[:dv ap 122]", 1000),
(0x02, b"[:dv g5 86]", 1000),
]
finally:
speech_driver.shutdown()
def test_dectalk_usb_driver_chunks_normal_writes(monkeypatch):
speech_driver, device, _, _ = initialized_driver(monkeypatch)
try:
tests/unit/test_hardware_speech_drivers.py
+37
View File
@@ -95,6 +95,22 @@ def test_dectalk_driver_writes_settings_and_cancel(serial_pair):
speech_driver.shutdown()
def test_dectalk_driver_skips_redundant_setting_commands(serial_pair):
speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
try:
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
speech_driver.set_volume(1.0)
speech_driver.set_rate(0.5)
speech_driver.set_pitch(0.5)
speech_driver.set_volume(1.0)
expected = b"[:ra 180][:dv ap 122][:dv g5 86]"
assert read_available(master_fd, len(expected)) == expected
assert read_available(master_fd, 1, timeout=0.2) == b""
finally:
speech_driver.shutdown()
def test_dectalk_driver_uses_native_defaults_at_fenrir_midpoint(serial_pair):
speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
try:
@@ -199,6 +215,27 @@ def test_hardware_driver_retries_partial_serial_writes(monkeypatch):
assert written_chunks == [b"ab", b"cd", b"ef"]
def test_hardware_driver_retries_setting_after_failed_write(monkeypatch):
speech_driver = dectalkDriver.driver()
speech_driver._is_initialized = True
write_results = [False, True]
written_commands = []
def fake_write(data, description):
written_commands.append((data, description))
return write_results.pop(0)
monkeypatch.setattr(speech_driver, "_write_bytes", fake_write)
speech_driver.set_rate(0.5)
speech_driver.set_rate(0.5)
assert written_commands == [
(b"[:ra 180]", "rate"),
(b"[:ra 180]", "rate"),
]
@pytest.mark.parametrize("driver_class", [doubletalkDriver, tripletalkDriver])
def test_litetalk_compatible_alias_drivers(driver_class, serial_pair):
speech_driver, master_fd = initialized_driver(driver_class, serial_pair)