Initial attempt at porting over fenrir's hardware synth support. Probably buggy.

tests/test_hardwarefactory_regressions.py

@@ -0,0 +1,100 @@
+import os
+import select
+import sys
+import time
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from cthulhu import hardwarefactory
+from cthulhu import settings
+
+
+def read_available(fd, expectedLength, timeout=1.0):
+    deadline = time.monotonic() + timeout
+    data = b""
+    while len(data) < expectedLength and time.monotonic() < deadline:
+        readable, _, _ = select.select([fd], [], [], 0.05)
+        if readable:
+            data += os.read(fd, 1024)
+    return data
+
+
+class HardwareFactoryRegressionTests(unittest.TestCase):
+    def setUp(self):
+        self._oldDevice = settings.hardwareSpeechDevice
+        self._oldBaudRate = settings.hardwareSpeechBaudRate
+        hardwarefactory.SpeechServer.shutdownActiveServers()
+
+    def tearDown(self):
+        hardwarefactory.SpeechServer.shutdownActiveServers()
+        settings.hardwareSpeechDevice = self._oldDevice
+        settings.hardwareSpeechBaudRate = self._oldBaudRate
+
+    def test_lists_explicit_synth_choices_without_opening_serial_device(self):
+        settings.hardwareSpeechDevice = ""
+
+        servers = hardwarefactory.SpeechServer.getSpeechServers()
+
+        self.assertEqual(
+            ["litetalk", "doubletalk", "tripletalk", "dectalk"],
+            [server.getInfo()[1] for server in servers],
+        )
+        self.assertEqual({}, hardwarefactory.SpeechServer._active_servers)
+        self.assertTrue(all(server._driver is None for server in servers))
+
+    def test_failed_initialization_is_not_cached(self):
+        settings.hardwareSpeechDevice = ""
+
+        self.assertIsNone(
+            hardwarefactory.SpeechServer.getSpeechServer(["LiteTalk", "litetalk"])
+        )
+        self.assertEqual({}, hardwarefactory.SpeechServer._active_servers)
+
+        masterFd, slaveFd = os.openpty()
+        try:
+            settings.hardwareSpeechDevice = os.ttyname(slaveFd)
+            server = hardwarefactory.SpeechServer.getSpeechServer(
+                ["LiteTalk", "litetalk"]
+            )
+
+            self.assertIsNotNone(server)
+            self.assertIsNotNone(server._driver)
+            self.assertIs(
+                server,
+                hardwarefactory.SpeechServer._active_servers.get("litetalk"),
+            )
+        finally:
+            os.close(masterFd)
+            os.close(slaveFd)
+
+    def test_explicit_synth_choices_write_expected_serial_bytes(self):
+        expectedBytes = {
+            "litetalk": b"Alias\r",
+            "doubletalk": b"Alias\r",
+            "tripletalk": b"Alias\r",
+            "dectalk": b"Alias\x01",
+        }
+
+        for synthId, expected in expectedBytes.items():
+            with self.subTest(synthId=synthId):
+                hardwarefactory.SpeechServer.shutdownActiveServers()
+                masterFd, slaveFd = os.openpty()
+                try:
+                    settings.hardwareSpeechDevice = os.ttyname(slaveFd)
+                    server = hardwarefactory.SpeechServer.getSpeechServer(
+                        ["", synthId]
+                    )
+
+                    self.assertIsNotNone(server)
+                    server.speak("Alias", interrupt=False)
+                    self.assertEqual(expected, read_available(masterFd, len(expected)))
+                finally:
+                    hardwarefactory.SpeechServer.shutdownActiveServers()
+                    os.close(masterFd)
+                    os.close(slaveFd)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_piperfactory_rate_mapping.py

@@ -0,0 +1,30 @@
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from cthulhu import piperfactory
+
+
+class PiperFactoryRateMappingTests(unittest.TestCase):
+    def setUp(self):
+        self.server = piperfactory.SpeechServer.__new__(piperfactory.SpeechServer)
+
+    def test_default_rate_maps_to_native_piper_speed(self):
+        self.assertEqual(1.0, self.server._mapRate(50))
+
+    def test_rate_scale_uses_full_cthulhu_range(self):
+        self.assertEqual(2.0, self.server._mapRate(0))
+        self.assertEqual(0.25, self.server._mapRate(100))
+
+    def test_high_screen_reader_rate_is_substantially_faster(self):
+        self.assertAlmostEqual(0.415, self.server._mapRate(89), places=3)
+
+    def test_rate_values_are_clamped(self):
+        self.assertEqual(2.0, self.server._mapRate(-1))
+        self.assertEqual(0.25, self.server._mapRate(101))
+
+
+if __name__ == "__main__":
+    unittest.main()