More attempts at getting this usb dectalk working.

2026-07-26 20:18:17 -04:00
parent 45bd33f756
commit c863245c24
11 changed files with 531 additions and 21 deletions
@@ -0,0 +1,273 @@
+import threading
+import time
+from unittest.mock import Mock
+
+import pytest
+
+from fenrirscreenreader.speechDriver import dectalkUsbDriver
+
+
+class FakeUsbDevice:
+    def __init__(self):
+        self.writes = []
+
+    def get_active_configuration(self):
+        return object()
+
+    def write(self, endpoint, data, timeout=None):
+        payload = bytes(data)
+        self.writes.append((endpoint, payload, timeout))
+        return len(payload)
+
+
+class FakeUsbCore:
+    USBError = OSError
+
+    class NoBackendError(ValueError):
+        pass
+
+    def __init__(self, device, find_error=None):
+        self.device = device
+        self.find_error = find_error
+        self.find_calls = []
+
+    def find(self, **kwargs):
+        self.find_calls.append(kwargs)
+        if self.find_error:
+            raise self.find_error
+        return self.device
+
+
+class FakeUsbUtil:
+    def __init__(
+        self,
+        claim_error=None,
+        release_error=None,
+        dispose_error=None,
+    ):
+        self.claim_error = claim_error
+        self.release_error = release_error
+        self.dispose_error = dispose_error
+        self.claimed = []
+        self.released = []
+        self.disposed = []
+
+    def claim_interface(self, device, interface):
+        if self.claim_error:
+            raise self.claim_error
+        self.claimed.append((device, interface))
+
+    def release_interface(self, device, interface):
+        if self.release_error:
+            raise self.release_error
+        self.released.append((device, interface))
+
+    def dispose_resources(self, device):
+        if self.dispose_error:
+            raise self.dispose_error
+        self.disposed.append(device)
+
+
+class BlockingUsbDevice(FakeUsbDevice):
+    def __init__(self):
+        super().__init__()
+        self.normal_write_started = threading.Event()
+        self.release_normal_write = threading.Event()
+        self.immediate_write_finished = threading.Event()
+
+    def write(self, endpoint, data, timeout=None):
+        if endpoint == 0x02:
+            self.normal_write_started.set()
+            self.release_normal_write.wait(timeout=1.0)
+        result = super().write(endpoint, data, timeout=timeout)
+        if endpoint == 0x04:
+            self.immediate_write_finished.set()
+        return result
+
+
+def build_environment():
+    return {
+        "runtime": {
+            "SettingsManager": Mock(),
+            "DebugManager": Mock(),
+        }
+    }
+
+
+def initialized_driver(monkeypatch):
+    device = FakeUsbDevice()
+    usb_core = FakeUsbCore(device)
+    usb_util = FakeUsbUtil()
+    speech_driver = dectalkUsbDriver.driver()
+    monkeypatch.setattr(
+        speech_driver,
+        "_load_usb_modules",
+        lambda: (usb_core, usb_util),
+    )
+    speech_driver.initialize(build_environment())
+    return speech_driver, device, usb_core, usb_util
+
+
+def wait_for_writes(device, count, timeout=1.0):
+    deadline = time.monotonic() + timeout
+    while len(device.writes) < count and time.monotonic() < deadline:
+        time.sleep(0.01)
+
+
+def test_dectalk_usb_driver_finds_and_claims_native_device(monkeypatch):
+    speech_driver, device, usb_core, usb_util = initialized_driver(monkeypatch)
+    try:
+        assert usb_core.find_calls == [
+            {"idVendor": 0x0DD0, "idProduct": 0x2001}
+        ]
+        assert usb_util.claimed == [(device, 0)]
+        assert speech_driver._is_initialized
+    finally:
+        speech_driver.shutdown()
+
+    assert usb_util.released == [(device, 0)]
+    assert usb_util.disposed == [device]
+
+
+def test_dectalk_usb_driver_writes_commands_and_speech(monkeypatch):
+    speech_driver, device, _, _ = initialized_driver(monkeypatch)
+    try:
+        speech_driver.set_rate(1.0)
+        speech_driver.set_pitch(1.0)
+        speech_driver.set_volume(0.5)
+        speech_driver.speak("Hello\nworld")
+        wait_for_writes(device, 4)
+
+        assert device.writes == [
+            (0x02, b"[:ra 650]", 1000),
+            (0x02, b"[:dv ap 350]", 1000),
+            (0x02, b"[:vo 50]", 1000),
+            (0x02, b"Hello world\x0b", 1000),
+        ]
+    finally:
+        speech_driver.shutdown()
+
+
+def test_dectalk_usb_driver_chunks_normal_writes(monkeypatch):
+    speech_driver, device, _, _ = initialized_driver(monkeypatch)
+    try:
+        speech_driver._write_bytes(b"x" * 70, "speech")
+        assert device.writes == [
+            (0x02, b"x" * 64, 1000),
+            (0x02, b"x" * 6, 1000),
+        ]
+    finally:
+        speech_driver.shutdown()
+
+
+def test_dectalk_usb_driver_uses_immediate_cancel_endpoint(monkeypatch):
+    speech_driver, device, _, _ = initialized_driver(monkeypatch)
+    try:
+        speech_driver.cancel()
+        assert device.writes == [(0x04, b"\x06", 1000)]
+    finally:
+        speech_driver.shutdown()
+
+
+def test_dectalk_usb_cancel_bypasses_blocked_normal_write(monkeypatch):
+    device = BlockingUsbDevice()
+    usb_core = FakeUsbCore(device)
+    usb_util = FakeUsbUtil()
+    speech_driver = dectalkUsbDriver.driver()
+    monkeypatch.setattr(
+        speech_driver,
+        "_load_usb_modules",
+        lambda: (usb_core, usb_util),
+    )
+    speech_driver.initialize(build_environment())
+    normal_thread = threading.Thread(
+        target=speech_driver._write_bytes,
+        args=(b"normal speech", "speech"),
+    )
+    cancel_thread = threading.Thread(target=speech_driver.cancel)
+    try:
+        normal_thread.start()
+        assert device.normal_write_started.wait(timeout=0.2)
+        cancel_thread.start()
+        assert device.immediate_write_finished.wait(timeout=0.2)
+    finally:
+        device.release_normal_write.set()
+        normal_thread.join(timeout=1.0)
+        cancel_thread.join(timeout=1.0)
+        speech_driver.shutdown()
+
+
+def test_dectalk_usb_driver_reports_missing_pyusb(monkeypatch):
+    speech_driver = dectalkUsbDriver.driver()
+
+    def missing_pyusb():
+        raise ImportError("No module named usb")
+
+    monkeypatch.setattr(speech_driver, "_load_usb_modules", missing_pyusb)
+
+    with pytest.raises(RuntimeError, match="PyUSB"):
+        speech_driver.initialize(build_environment())
+
+
+def test_dectalk_usb_driver_reports_missing_libusb_backend(monkeypatch):
+    speech_driver = dectalkUsbDriver.driver()
+    backend_error = FakeUsbCore.NoBackendError("No backend available")
+    usb_core = FakeUsbCore(None, find_error=backend_error)
+    monkeypatch.setattr(
+        speech_driver,
+        "_load_usb_modules",
+        lambda: (usb_core, FakeUsbUtil()),
+    )
+
+    with pytest.raises(RuntimeError, match="discovery failed.*backend"):
+        speech_driver.initialize(build_environment())
+
+
+def test_dectalk_usb_driver_reports_missing_device(monkeypatch):
+    speech_driver = dectalkUsbDriver.driver()
+    usb_core = FakeUsbCore(None)
+    usb_util = FakeUsbUtil()
+    monkeypatch.setattr(
+        speech_driver,
+        "_load_usb_modules",
+        lambda: (usb_core, usb_util),
+    )
+
+    with pytest.raises(RuntimeError, match="0dd0:2001"):
+        speech_driver.initialize(build_environment())
+
+
+def test_dectalk_usb_driver_reports_interface_permission_error(monkeypatch):
+    speech_driver = dectalkUsbDriver.driver()
+    device = FakeUsbDevice()
+    usb_core = FakeUsbCore(device)
+    usb_util = FakeUsbUtil(claim_error=PermissionError("access denied"))
+    monkeypatch.setattr(
+        speech_driver,
+        "_load_usb_modules",
+        lambda: (usb_core, usb_util),
+    )
+
+    with pytest.raises(RuntimeError, match="open failed.*access denied"):
+        speech_driver.initialize(build_environment())
+
+
+def test_dectalk_usb_shutdown_survives_cleanup_errors(monkeypatch):
+    device = FakeUsbDevice()
+    usb_core = FakeUsbCore(device)
+    usb_util = FakeUsbUtil(
+        release_error=OSError("release failed"),
+        dispose_error=OSError("dispose failed"),
+    )
+    speech_driver = dectalkUsbDriver.driver()
+    monkeypatch.setattr(
+        speech_driver,
+        "_load_usb_modules",
+        lambda: (usb_core, usb_util),
+    )
+    speech_driver.initialize(build_environment())
+
+    speech_driver.shutdown()
+
+    assert not speech_driver._is_initialized
+    assert speech_driver.usb_device is None