Add Python output modules

src/modules/Makefile.am

@@ -30,6 +30,7 @@ if HAVE_PYTHON
 speechd_python_modules_pythondir = $(pythondir)/speechd_python_modules
 dist_speechd_python_modules_python_PYTHON = \
 	speechd_python_modules/__init__.py \
+	speechd_python_modules/module_readline.py \
 	speechd_python_modules/module_utils.py \
 	speechd_python_modules/speechd_types.py
 endif

src/modules/speechd_python_modules/module_readline.py

@@ -0,0 +1,131 @@
+#
+# module_readline.py - Input buffering for Python output modules.
+#
+# Copyright (C) 2020 Samuel Thibault <samuel.thibault@ens-lyon.org>
+# Copyright (C) 2026 Jean-François David <jeanfrancoismanutea@gmail.com>
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+# 1. Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS ``AS IS'' AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+# SUCH DAMAGE.
+#
+
+import os
+import select
+import sys
+
+
+READ_CHUNK = 4096
+
+_fd_buffers = {}
+
+
+class _ReadBuffer:
+    def __init__(self):
+        self.data = bytearray()
+        self.no_lf = 0
+
+def module_readline(source=None, block=True):
+    if source is None:
+        source = sys.stdin
+    if isinstance(source, int):
+        return _readline_fd(source, block)
+
+    fd = _source_fd(source)
+    if fd is not None:
+        return _readline_fd(fd, block)
+
+    if not block:
+        return None
+
+    line = source.readline()
+    return _decode_complete_line(line)
+
+
+def _readline_fd(fd, block):
+    state = _fd_buffers.get(fd)
+
+    while True:
+        if state is not None:
+            newline = state.data.find(b"\n", state.no_lf)
+            if newline != -1:
+                line = bytes(state.data[: newline + 1])
+                del state.data[: newline + 1]
+                state.no_lf = 0
+                if not state.data:
+                    _fd_buffers.pop(fd, None)
+                return _decode_bytes(line)
+
+            state.no_lf = len(state.data)
+
+        try:
+            readable, _, _ = select.select([fd], [], [], None if block else 0)
+        except (InterruptedError, BlockingIOError):
+            if not block:
+                return None
+            continue
+        except OSError:
+            _fd_buffers.pop(fd, None)
+            return None
+
+        if not readable:
+            return None
+
+        try:
+            chunk = os.read(fd, READ_CHUNK)
+        except (InterruptedError, BlockingIOError):
+            if not block:
+                return None
+            continue
+        except OSError:
+            _fd_buffers.pop(fd, None)
+            return None
+
+        if not chunk:
+            if state is not None:
+                _fd_buffers.pop(fd, None)
+            return None
+
+        if state is None:
+            state = _ReadBuffer()
+            _fd_buffers[fd] = state
+
+        state.data.extend(chunk)
+
+
+def _source_fd(source):
+    try:
+        return source.fileno()
+    except (AttributeError, OSError, ValueError):
+        return None
+
+
+def _decode_complete_line(line):
+    if not line:
+        return None
+    if not line.endswith(b"\n" if isinstance(line, bytes) else "\n"):
+        return None
+    if isinstance(line, bytes):
+        return _decode_bytes(line)
+    return line
+
+
+def _decode_bytes(data):
+    return data.decode("utf-8", "surrogateescape")

src/modules/speechd_python_modules/module_utils.py

@@ -27,6 +27,58 @@
 #
 
 
+import re
+
+
+log_level = 0
+Debug = 0
+CustomDebugFile = None
+
+
+def module_loglevel_set(cur_item, cur_value):
+    global log_level
+
+    if cur_item != "log_level":
+        return -1
+
+    match = re.match(r"\s*([+-]?[0-9]+)", cur_value)
+    if match is None:
+        return -1
+
+    log_level = int(match.group(1), 10)
+    return 0
+
+
+# TODO Add an equivalent of C MSG() that writes to CustomDebugFile.
+def module_debug(enable, filename):
+    global CustomDebugFile, Debug
+
+    if enable:
+        try:
+            new_custom_debug_file = open(filename, "w+")
+        except OSError:
+            return -1
+
+        if CustomDebugFile is not None:
+            CustomDebugFile.close()
+        CustomDebugFile = new_custom_debug_file
+        if Debug == 1:
+            Debug = 3
+        else:
+            Debug = 2
+    else:
+        if Debug == 3:
+            Debug = 1
+        else:
+            Debug = 0
+
+        if CustomDebugFile is not None:
+            CustomDebugFile.close()
+        CustomDebugFile = None
+
+    return 0
+
+
 def module_strip_ssml(message: str) -> str:
     out = []
     append = out.append

src/tests/Makefile.am

@@ -43,6 +43,11 @@ AUTOM4TE = autom4te
 AUTOTEST = $(AUTOM4TE) --language=autotest
 
 TESTSUITE_AT = c_api.at python_module.at
+PYTHON_TESTS = \
+	python/test_imports.py \
+	python/test_module_readline.py \
+	python/test_module_utils.py \
+	python/test_speechd_types.py
 TESTSUITE = ./testsuite
 $(TESTSUITE): package.m4 testsuite.at $(TESTSUITE_AT)
 	$(AUTOTEST) -I '$(srcdir)' -o $@.tmp $@.at
@@ -79,7 +84,7 @@ run_test_LDADD = $(c_api)/libspeechd.la $(GLIB_LIBS) $(EXTRA_SOCKET_LIBS)
 EXTRA_DIST= basic.test general.test keys.test priority_progress.test \
             pronunciation.test punctuation.test sound_icons.test spelling.test \
             ssml.test stop_and_pause.test voices.test yo.wav \
-            atlocal.in testsuite.at $(TESTSUITE_AT) sayfortune.sh
+            atlocal.in testsuite.at $(TESTSUITE_AT) $(PYTHON_TESTS) sayfortune.sh
 
 clean-local:
 	test ! -f $(TESTSUITE) || $(SHELL) $(TESTSUITE) --clean

src/tests/python/test_imports.py

@@ -0,0 +1,42 @@
+#
+# test_imports.py - Python module import tests
+#
+# Copyright (C) 2026 Jean-François David
+#
+# This is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2, or (at your option)
+# any later version.
+#
+# This software is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import os
+import unittest
+
+from speechd_python_modules import module_readline, module_utils, speechd_types
+
+
+class ImportsTest(unittest.TestCase):
+    def test_modules_are_imported_from_package_directory(self):
+        module_root = os.environ.get(
+            "TEST_PYTHONPATH",
+            os.path.join(os.path.dirname(__file__), "..", "..", "modules"),
+        )
+        package_dir = os.path.join(module_root, "speechd_python_modules")
+        expected_dir = os.path.realpath(package_dir)
+        modules = [module_readline, module_utils, speechd_types]
+
+        for module in modules:
+            with self.subTest(module=module.__name__):
+                module_file = os.path.realpath(module.__file__)
+                self.assertEqual(os.path.dirname(module_file), expected_dir)
+
+
+if __name__ == "__main__":
+    unittest.main()

src/tests/python/test_module_readline.py

@@ -0,0 +1,105 @@
+#
+# test_module_readline.py - Python module_readline unit tests
+#
+# Copyright (C) 2026 Jean-François David
+#
+# This is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2, or (at your option)
+# any later version.
+#
+# This software is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import io
+import os
+import unittest
+
+import speechd_python_modules.module_readline as module_readline
+
+
+class ModuleReadlineTest(unittest.TestCase):
+    def setUp(self):
+        self._fds = []
+
+    def tearDown(self):
+        for fd in self._fds:
+            module_readline._fd_buffers.pop(fd, None)
+        for fd in reversed(self._fds):
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+
+    def pipe(self):
+        read_fd, write_fd = os.pipe()
+        self._fds.extend((read_fd, write_fd))
+        return read_fd, write_fd
+
+    def close_fd(self, fd):
+        os.close(fd)
+        self._fds.remove(fd)
+        module_readline._fd_buffers.pop(fd, None)
+
+    def test_nonblocking_empty_fd_returns_none(self):
+        read_fd, _write_fd = self.pipe()
+
+        self.assertIsNone(module_readline.module_readline(read_fd, block=False))
+
+    def test_nonblocking_partial_line_is_buffered(self):
+        read_fd, write_fd = self.pipe()
+
+        os.write(write_fd, b"partial")
+        self.assertIsNone(module_readline.module_readline(read_fd, block=False))
+
+        os.write(write_fd, b"\nnext\n")
+        self.assertEqual(
+            module_readline.module_readline(read_fd, block=False),
+            "partial\n",
+        )
+        self.assertEqual(
+            module_readline.module_readline(read_fd, block=False),
+            "next\n",
+        )
+
+    def test_eof_with_partial_line_returns_none(self):
+        read_fd, write_fd = self.pipe()
+
+        os.write(write_fd, b"partial")
+        self.close_fd(write_fd)
+
+        self.assertIsNone(module_readline.module_readline(read_fd, block=True))
+        self.assertNotIn(read_fd, module_readline._fd_buffers)
+
+    def test_invalid_utf8_round_trips_with_surrogateescape(self):
+        read_fd, write_fd = self.pipe()
+
+        os.write(write_fd, b"bad\xff\n")
+
+        line = module_readline.module_readline(read_fd, block=True)
+        self.assertEqual(line, "bad\udcff\n")
+        self.assertEqual(line.encode("utf-8", "surrogateescape"), b"bad\xff\n")
+
+    def test_file_like_source_uses_readline_fallback(self):
+        source = io.StringIO("hello\n")
+
+        self.assertEqual(module_readline.module_readline(source), "hello\n")
+
+    def test_incomplete_file_like_line_returns_none(self):
+        source = io.StringIO("partial")
+
+        self.assertIsNone(module_readline.module_readline(source))
+
+    def test_nonblocking_file_like_source_without_fd_returns_none(self):
+        source = io.StringIO("hello\n")
+
+        self.assertIsNone(module_readline.module_readline(source, block=False))
+
+
+if __name__ == "__main__":
+    unittest.main()