[py314-and-log-dir-patch] Making QMI Python 3.14 compatible and fixed a bug in log directory location for services.

Author: Henri Ervasti

.gitignore

@@ -129,4 +129,7 @@ dmypy.json
 .pyre/
 
 # VS code settings
-.vscode
+.vscode/
+
+# PyCharm
+.idea

CHANGELOG.md

@@ -5,6 +5,15 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 
+## [0.51.2] - 2026-02-20
+
+### Added
+- `run` method in `QMI_Context` class as this is now required for Python 3.14, after changes in `threading` module.
+
+### Fixed
+- Log file location config for services in `qmi_proc.py`. Location string starting with tilde (~) is now interpreted correctly to be the user's home directory.
+- Minor edits in docstrings, tutorial and .gitignore file.
+
 ## [0.51.1] - 2026-01-26
 
 ### Fixed

documentation/sphinx/source/tutorial.rst

@@ -860,16 +860,15 @@ Connecting with USBTMC devices on Windows can be tricky. Make sure you have libu
 https://pypi.org/project/libusb1/ and https://pypi.org/project/PyVISA/ (and perhaps pyvisa-py).
 
 Then you'll need to have the backend set-up correctly, in case the ``libusb-1.0.dll`` is not found in your path.
-An example script to set-up and test the backend is
-```python
-import usb.core
-from usb.backend import libusb1
+An example script to set-up and test the backend is::
 
-backend = libusb1.get_backend(
-    find_library=lambda x: "<path_to_your_env>\\Lib\\site-packages\\usb1\\libusb-1.0.dll")
+  import usb.core
+  from usb.backend import libusb1
 
-dev = list(usb.core.find(find_all=True))
-```
+  backend = libusb1.get_backend(
+      find_library=lambda x: "<path_to_your_env>\\Lib\\site-packages\\usb1\\libusb-1.0.dll")
+
+  dev = list(usb.core.find(find_all=True))
 
 If you can now find devices, the backend is set correctly. There are of course other ways to set-up your backend
 as well, but as said, it can be tricky...

qmi/core/context.py

@@ -461,6 +461,12 @@ def resolve_file_name(self, file_name: str) -> str:
 
         return os.path.normpath(string.Template(file_name).substitute(mapping))
 
+    def run(self, _rpc_thread_run: Callable) -> None:
+        """Run thread. From Python 3.14 onwards objects that are run within a context, need a `run` method with
+        a single input where the callable to run is given. Addition of this function should have no effect when
+        using earlier Python versions."""
+        _rpc_thread_run()
+
     def start(self) -> None:
         """Start the context.
 

qmi/core/transport.py

@@ -1321,7 +1321,7 @@ def create_transport(
       - "port" (for UDP and TCP transports) specifies the UDP/TCP port number of the server/client.
       - "connect_timeout" is TCP connection timeout.
 
-     Serial:
+    Serial:
       - "device" is the name of the serial port, for example "COM3" or "/dev/ttyUSB0".
       - "baudrate" specifies the number of bits per second.
         This attribute is only required for instruments with a configurable baud rate.

qmi/tools/proc.py

@@ -326,6 +326,8 @@ def start_local_process(context_name: str) -> int:
     else:
         output_dir = qmi.context().get_qmi_home_dir()
 
+    # Replace tilde with full path as otherwise it might be literally used
+    output_dir = output_dir.replace("~", os.path.expanduser("~")) if output_dir.startswith("~") else output_dir
     # Create output log file.
     datetime = time.strftime("%Y%m%d_%H%M%S", time.gmtime())
     output_file_name = os.path.join(output_dir, context_name + "_" + datetime + ".out")

tests/core/test_transport.py

@@ -118,18 +118,11 @@ async def the_call():
         with open_close(QMI_UdpTransport("localhost", int(self.server_port) - 1)) as trans:
 
             # Send some bytes from server to transport.
-            loop = asyncio.get_event_loop()
-            if loop.is_closed():
-                asyncio.set_event_loop(loop)
-
-            loop.run_until_complete(the_call())
+            asyncio.run(the_call())
             # Receive message through transport.
             data = trans.read_until(b"\n", timeout=1.0)
             self.assertEqual(data, b"aap noot\n")
 
-            if loop.is_running():
-                loop.close()
-
             w_data = b"aap noot\n"
             trans.write(w_data)
             data = trans.read(len(w_data))
@@ -603,10 +596,10 @@ def test_udp_large_packages(self):
                 with self.assertRaises(qmi.core.exceptions.QMI_RuntimeException):
                     read = trans.read(100, timeout=1.0)  # The `read` will set the size to 4096 in any case
 
-            except AssertionError as ass:
+            except AssertionError as a_err:
                 # Catch this as some servers apparently fragment the message to be max of 4096 bytes, so it does not crash
                 if len(read) != 100:
-                    raise AssertionError from ass
+                    raise AssertionError from a_err
 
                 trans.discard_read()
 
@@ -618,26 +611,19 @@ async def the_call():
                 l.stop()
 
             # Send some bytes from server to transport.
-            loop = asyncio.get_event_loop()
-            if loop.is_closed():
-                asyncio.set_event_loop(loop)
-
-            loop.run_until_complete(the_call())
+            asyncio.run(the_call())
             try:
                 # The same should happen with read_until (triggers exception).
                 with self.assertRaises(qmi.core.exceptions.QMI_RuntimeException):
                     trans.read_until(b"\n", timeout=1.0)
 
-            except qmi.core.exceptions.QMI_TimeoutException as tim:
+            except qmi.core.exceptions.QMI_TimeoutException as t_exc:
                 # Catch this as some servers apparently fragment the message to be max of 4096 bytes, so it does not crash
                 if len(trans._read_buffer) != 4096:
-                    raise AssertionError from tim
+                    raise AssertionError from t_exc
 
                 trans.discard_read()
 
-            if loop.is_running():
-                loop.close()
-
             # Send some bytes from server to transport.
             self.server_sock.sendto(s.encode(), trans._address)