Improve type hints (#65)

setup.py

@@ -130,7 +130,7 @@ class SDist(cmdclass["sdist"]):
         ("freeze-first=", None, ""),
     ]
 
-    def initialize_options(self):
+    def initialize_options(self) -> None:
         super().initialize_options()
         self.freeze_first = None
 
@@ -144,11 +144,11 @@ class BDistWheel(bdist_wheel):
         ),
     ]
 
-    def initialize_options(self):
+    def initialize_options(self) -> None:
         super().initialize_options()
         self.freeze_first = None
 
-    def finalize_options(self):
+    def finalize_options(self) -> None:
         if self.freeze_first:
             self.distribution.install_requires = freeze_requirements(
                 list(self.distribution.install_requires)

src/amulet_editor/__main__.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 
 
-def main():
+def main() -> None:
     try:
         # Verify the python version
         import sys

src/amulet_editor/__pyinstaller/__init__.py

@@ -1,2 +1,2 @@
-def get_hook_dirs():
+def get_hook_dirs() -> None:
     return __path__

src/amulet_editor/application/_app.py

@@ -18,7 +18,7 @@
 
 
 class AmuletApp(QApplication):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__()
         self.setApplicationName("Amulet Editor")
         self.setApplicationVersion(__version__)
@@ -39,7 +39,7 @@ def instance() -> Optional[AmuletApp]:
         return QApplication.instance()
 
     @Slot()
-    def _last_window_closed(self):
+    def _last_window_closed(self) -> None:
         # The unload method opens a window and then closes it.
         # We must unbind this signal so that it does not end in a loop.
         self.lastWindowClosed.disconnect(self._last_window_closed)
@@ -49,7 +49,7 @@ def _last_window_closed(self):
         self.quit()
 
     @Slot()
-    def _locale_changed(self):
+    def _locale_changed(self) -> None:
         self._translator.load_lang(
             QLocale(),
             "",

src/amulet_editor/application/_cli.py

@@ -114,7 +114,7 @@ def parse_args() -> Args:
 
 
 # TODO: move this somewhere more sensible
-def spawn_process(path: str = None):
+def spawn_process(path: str = None) -> None:
     """Spawn the broker process passing over the input CLI values."""
     this_args = parse_args()
     new_args = [sys.executable, sys.argv[0]]

src/amulet_editor/application/_invoke.py

@@ -11,7 +11,7 @@
 
 @final
 class InvokeMethod(QObject):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__()
         self.__method = None
         self.__return = None
@@ -67,7 +67,7 @@ def invoke(cls, method: Callable[[], T], parent: QObject = None) -> T:
     start_signal = Signal()
 
     @Slot()
-    def execute(self):
+    def execute(self) -> None:
         try:
             self.__return = self.__method()
         except BaseException as e:

src/amulet_editor/application/_main.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import threading
-from typing import Optional
+from typing import Optional, Callable, TypeAlias, Any, Union
+from types import FrameType
 import sys
 import os
 import logging
@@ -15,6 +16,7 @@
     qInstallMessageHandler,
     QtMsgType,
     QLocale,
+    QMessageLogContext
 )
 from PySide6.QtWidgets import QApplication
 from PySide6.QtGui import QSurfaceFormat
@@ -30,11 +32,13 @@
 from ._app import AmuletApp
 from amulet_editor.data.paths._application import _init_paths, logging_directory
 
+TraceFunction: TypeAlias = Callable[[FrameType, str, Any], Union["TraceFunction", None]]
+
 log = logging.getLogger(__name__)
 qt_log = logging.getLogger("Qt")
 
 
-def _qt_log(msg_type, context, msg):
+def _qt_log(msg_type: QtMsgType, context: QMessageLogContext, msg: str) -> None:
     if msg_type == QtMsgType.QtDebugMsg:
         qt_log.debug(msg)
     if msg_type == QtMsgType.QtInfoMsg:
@@ -47,7 +51,7 @@ def _qt_log(msg_type, context, msg):
         qt_log.fatal(msg)
 
 
-def app_main():
+def app_main() -> None:
     args = parse_args()
     _init_paths(args.data_dir, args.config_dir, args.cache_dir, args.log_dir)
 
@@ -71,13 +75,15 @@ def app_main():
     # TODO: remove old log files
 
     class StdCapture(TextIOWrapper):
-        def __init__(self, logger):
-            super().__init__(log_file)
+        def __init__(self, logger: Callable[[str], None]) -> None:
+            super().__init__(log_file)  # type: ignore
             self._logger = logger
 
-        def write(self, msg):
+        def write(self, msg: str) -> int:
             if msg != "\n":
                 self._logger(msg)
+                return len(msg)
+            return 0
 
     # Convert all direct stdout calls (eg print) to info log calls
     sys.stdout = StdCapture(logging.getLogger("Python stdout").info)
@@ -92,7 +98,7 @@ def write(self, msg):
 
     if args.trace:
 
-        def trace_calls(frame, event, arg):
+        def trace_calls(frame: FrameType, event: str, arg: Any) -> TraceFunction:
             if event == "call":
                 try:
                     qual_name = frame.f_code.co_qualname

src/amulet_editor/application/appearance/_theme.py

@@ -77,7 +77,7 @@ def load_style_sheet(self, file_path: str) -> str:
             style_sheet = fh.read().replace('"', "")
 
         icons: set[str] = set(
-            [icon for icon in re.findall("url\((.*?)\)", style_sheet)]
+            [icon for icon in re.findall(r"url\((.*?)\)", style_sheet)]
         )
         for icon in icons:
             style_sheet = style_sheet.replace(

src/amulet_editor/data/_localisation/__init__.py

@@ -9,7 +9,7 @@ class LocaleObject(QObject):
 locale_changed = _obj.locale_changed
 
 
-def set_locale(locale: QLocale):
+def set_locale(locale: QLocale) -> None:
     """Set the application locale"""
     QLocale.setDefault(locale)
     locale_changed.emit(locale)

src/amulet_editor/data/_rpc.py

@@ -101,7 +101,7 @@ def is_landing_process() -> Optional[bool]:
 
 
 @register_remote_procedure
-def call_global(address, args, kwargs):
+def call_global(address, args, kwargs) -> None:
     """
     Call a procedure in all child processes.
     The return values are discarded.
@@ -118,7 +118,7 @@ def call_global(address, args, kwargs):
 
 # class register_global_remote_procedure:
 #     """A decorator that enables a function to be called in all processes."""
-#     def __init__(self, func: Callable):
+#     def __init__(self, func: Callable) -> None:
 #         if not callable(func):
 #             raise TypeError("func must be callable")
 #         self._func = register_remote_procedure(func)
@@ -138,7 +138,7 @@ def call_global(address, args, kwargs):
 _listener_connections: dict[RemoteProcedureCallingConnection, Optional[bool]] = {}
 
 
-def _on_listener_connect():
+def _on_listener_connect() -> None:
     with DisplayException("Error initialising socket", suppress=True, log=log):
         socket = _remote_call_listener.nextPendingConnection()
         connection = RemoteProcedureCallingConnection(socket)
@@ -147,11 +147,11 @@ def _on_listener_connect():
 
         if _is_broker:
 
-            def is_landing_success(response: bool):
+            def is_landing_success(response: bool) -> None:
                 log.debug(f"New connection is_landing {response}")
                 _listener_connections[connection] = response
 
-            def is_landing_err(tb_str):
+            def is_landing_err(tb_str) -> None:
                 logging.exception(tb_str)
 
             connection.call(
@@ -160,7 +160,7 @@ def is_landing_err(tb_str):
                 is_landing_process,
             )
 
-        def on_disconnect():
+        def on_disconnect() -> None:
             with DisplayException("Error on socket disconnect", suppress=True, log=log):
                 is_landing = _listener_connections.pop(connection, None)
                 log.debug(f"Listener connection disconnected {socket}. {is_landing}")
@@ -194,7 +194,7 @@ def on_disconnect():
 class RemoteProcedureCallingConnection:
     """A subclass of QLocalSocket to facilitate calling a procedure in a remote process and getting the return value."""
 
-    def __init__(self, socket: QLocalSocket = None):
+    def __init__(self, socket: QLocalSocket = None) -> None:
         self.socket = socket or QLocalSocket()
         # A dictionary mapping the UUID for the call to the callback functions.
         self._calls: CallDataStorage = {}
@@ -240,7 +240,7 @@ def call(
         payload = self.encode_request(identifier, address, args, kwargs)
         self.socket.write(payload)
 
-    def _process_msg(self):
+    def _process_msg(self) -> None:
         with DisplayException(
             "Exception processing remote procedure call.", suppress=True, log=log
         ):
@@ -310,7 +310,7 @@ def _add_size(payload: bytes) -> bytes:
         return struct.pack(">I", payload_size) + payload
 
     @classmethod
-    def encode_request(cls, identifier: bytes, address: str, args, kwargs):
+    def encode_request(cls, identifier: bytes, address: str, args, kwargs) -> bytes:
         """
         Encode an RPC request.
 
@@ -337,7 +337,7 @@ def encode_error_response(cls, identifier: bytes, msg: str) -> bytes:
 _broker_connection = RemoteProcedureCallingConnection()
 
 
-def init_rpc(broker=False):
+def init_rpc(broker=False) -> None:
     """Init the messaging state"""
     global _is_broker
     log.debug("Initialising RPC.")
@@ -354,23 +354,23 @@ def init_rpc(broker=False):
         else:
             raise Exception(msg)
 
-    def on_connect():
+    def on_connect() -> None:
         with DisplayException("Error on socket connect", suppress=True, log=log):
             nonlocal failed_connections
             failed_connections = 0
             log.debug("Connected to broker process.")
 
             if _is_broker:
 
-                def on_success_response(result):
+                def on_success_response(result) -> None:
                     if result == server_uuid:
                         log.debug("I am the broker")
                         _broker_connection.socket.close()
                     else:
                         log.debug("Exiting because a broker already exists.")
                         QApplication.quit()
 
-                def on_error_response(tb_str):
+                def on_error_response(tb_str) -> None:
                     log.exception(tb_str)
                     display_exception(
                         title="Broker exception",
@@ -382,7 +382,7 @@ def on_error_response(tb_str):
                     on_success_response, on_error_response, get_server_uuid
                 )
 
-    def on_error():
+    def on_error() -> None:
         with DisplayException(
             "Error on socket connection error", suppress=True, log=log
         ):

src/amulet_editor/data/dev/_debug.py

@@ -1,4 +1,4 @@
-def enable_trace():
+def enable_trace() -> None:
     """
     Enable debugging support.
     This must be called to allow debugging python function run in a QThreadPool.

src/amulet_editor/data/packages/_manage.py

@@ -4,7 +4,7 @@
 from amulet_editor.models.package import AmuletPackage, AmuletTool
 
 
-def install_builtins():
+def install_builtins() -> None:
     pass
 
 

src/amulet_editor/data/paths/_plugin.py

@@ -4,11 +4,11 @@
 from ._application import data_directory
 
 
-def first_party_plugin_directory():
+def first_party_plugin_directory() -> str:
     return os.path.abspath(os.path.join(amulet_editor.__path__[0], "plugins"))
 
 
-def third_party_plugin_directory():
+def third_party_plugin_directory() -> str:
     """Returns the path within which dynamic plugins are stored."""
     path = os.path.abspath(os.path.join(data_directory(), "plugins"))
     os.makedirs(path, exist_ok=True)