python: Add typing to RPC server and Messenger

python/grass/pygrass/messages/__init__.py

@@ -6,34 +6,45 @@
 Fast and exit-safe interface to GRASS C-library message functions
 
 
-(C) 2013 by the GRASS Development Team
+(C) 2013-2024 by the GRASS Development Team
 This program is free software under the GNU General Public
 License (>=v2). Read the file COPYING that comes with GRASS
 for details.
 
-@author Soeren Gebbert
+@author Soeren Gebbert, Edouard Choinière
 """
 
+from __future__ import annotations
+
 import sys
-from multiprocessing import Process, Lock, Pipe
+from multiprocessing import Lock, Pipe, Process
+from typing import TYPE_CHECKING, Literal, NoReturn
 
 import grass.lib.gis as libgis
-
 from grass.exceptions import FatalError
 
+if TYPE_CHECKING:
+    from multiprocessing.connection import Connection
+    from multiprocessing.context import _LockLike
+
+    _MessagesLiteral = Literal[
+        "INFO", "IMPORTANT", "VERBOSE", "WARNING", "ERROR", "FATAL"
+    ]
+
 
-def message_server(lock, conn):
+def message_server(lock: _LockLike, conn: Connection) -> NoReturn:
     """The GRASS message server function designed to be a target for
     multiprocessing.Process
 
 
     :param lock: A multiprocessing.Lock
-    :param conn: A multiprocessing.Pipe
+    :param conn: A multiprocessing.connection.Connection object obtained from
+                 multiprocessing.Pipe
 
     This function will use the G_* message C-functions from grass.lib.gis
     to provide an interface to the GRASS C-library messaging system.
 
-    The data that is send through the pipe must provide an
+    The data that is sent through the pipe must provide an
     identifier string to specify which C-function should be called.
 
     The following identifiers are supported:
@@ -52,9 +63,9 @@ def message_server(lock, conn):
     - "FATAL"      Calls G_fatal_error(), this functions is only for
                    testing purpose
 
-    The that is end through the pipe must be a list of values:
+    The data that is sent through the pipe must be a list of values:
 
-    - Messages: ["INFO|VERBOSE|WARNING|ERROR|FATAL", "MESSAGE"]
+    - Messages: ["INFO|IMPORTANT|VERBOSE|WARNING|ERROR|FATAL", "MESSAGE"]
     - Debug:    ["DEBUG", level, "MESSAGE"]
     - Percent:  ["PERCENT", n, d, s]
 
@@ -65,44 +76,42 @@ def message_server(lock, conn):
         # Avoid busy waiting
         conn.poll(None)
         data = conn.recv()
-        message_type = data[0]
+        message_type: Literal[_MessagesLiteral, "DEBUG", "PERCENT", "STOP"] = data[0]
 
         # Only one process is allowed to write to stderr
-        lock.acquire()
-
-        # Stop the pipe and the infinite loop
-        if message_type == "STOP":
-            conn.close()
-            lock.release()
-            libgis.G_debug(1, "Stop messenger server")
-            sys.exit()
-
-        message = data[1]
-
-        if message_type == "PERCENT":
-            n = int(data[1])
-            d = int(data[2])
-            s = int(data[3])
-            libgis.G_percent(n, d, s)
-        elif message_type == "DEBUG":
-            level = data[1]
-            message = data[2]
-            libgis.G_debug(level, message)
-        elif message_type == "VERBOSE":
-            libgis.G_verbose_message(message)
-        elif message_type == "INFO":
-            libgis.G_message(message)
-        elif message_type == "IMPORTANT":
-            libgis.G_important_message(message)
-        elif message_type == "WARNING":
-            libgis.G_warning(message)
-        elif message_type == "ERROR":
-            libgis.G_important_message("ERROR: %s" % message)
-        # This is for testing only
-        elif message_type == "FATAL":
-            libgis.G_fatal_error(message)
-
-        lock.release()
+        with lock:
+            # Stop the pipe and the infinite loop
+            if message_type == "STOP":
+                conn.close()
+                libgis.G_debug(1, "Stop messenger server")
+                sys.exit()
+
+            if message_type == "PERCENT":
+                n = int(data[1])
+                d = int(data[2])
+                s = int(data[3])
+                libgis.G_percent(n, d, s)
+                continue
+            if message_type == "DEBUG":
+                level = int(data[1])
+                message_debug = data[2]
+                libgis.G_debug(level, message_debug)
+                continue
+
+            message: str = data[1]
+            if message_type == "VERBOSE":
+                libgis.G_verbose_message(message)
+            elif message_type == "INFO":
+                libgis.G_message(message)
+            elif message_type == "IMPORTANT":
+                libgis.G_important_message(message)
+            elif message_type == "WARNING":
+                libgis.G_warning(message)
+            elif message_type == "ERROR":
+                libgis.G_important_message("ERROR: %s" % message)
+            # This is for testing only
+            elif message_type == "FATAL":
+                libgis.G_fatal_error(message)
 
 
 class Messenger:
@@ -165,22 +174,27 @@ class Messenger:
 
     """
 
-    def __init__(self, raise_on_error=False):
-        self.client_conn = None
-        self.server_conn = None
-        self.server = None
+    client_conn: Connection
+    server_conn: Connection
+    server: Process
+
+    def __init__(self, raise_on_error: bool = False) -> None:
         self.raise_on_error = raise_on_error
-        self.start_server()
+        self.client_conn, self.server_conn = Pipe()
+        self.lock = Lock()
+        self.server = Process(target=message_server, args=(self.lock, self.server_conn))
+        self.server.daemon = True
+        self.server.start()
 
-    def start_server(self):
+    def start_server(self) -> None:
         """Start the messenger server and open the pipe"""
         self.client_conn, self.server_conn = Pipe()
         self.lock = Lock()
         self.server = Process(target=message_server, args=(self.lock, self.server_conn))
         self.server.daemon = True
         self.server.start()
 
-    def _check_restart_server(self):
+    def _check_restart_server(self) -> None:
         """Restart the server if it was terminated"""
         if self.server.is_alive() is True:
             return
@@ -189,67 +203,61 @@ def _check_restart_server(self):
         self.start_server()
         self.warning("Needed to restart the messenger server")
 
-    def message(self, message):
+    def message(self, message: str) -> None:
         """Send a message to stderr
 
         :param message: the text of message
-        :type message: str
 
            G_message() will be called in the messenger server process
         """
         self._check_restart_server()
         self.client_conn.send(["INFO", message])
 
-    def verbose(self, message):
+    def verbose(self, message: str) -> None:
         """Send a verbose message to stderr
 
         :param message: the text of message
-        :type message: str
 
            G_verbose_message() will be called in the messenger server process
         """
         self._check_restart_server()
         self.client_conn.send(["VERBOSE", message])
 
-    def important(self, message):
+    def important(self, message: str) -> None:
         """Send an important message to stderr
 
         :param message: the text of message
-        :type message: str
 
            G_important_message() will be called in the messenger server process
         """
         self._check_restart_server()
         self.client_conn.send(["IMPORTANT", message])
 
-    def warning(self, message):
+    def warning(self, message: str) -> None:
         """Send a warning message to stderr
 
         :param message: the text of message
-        :type message: str
 
            G_warning() will be called in the messenger server process
         """
         self._check_restart_server()
         self.client_conn.send(["WARNING", message])
 
-    def error(self, message):
+    def error(self, message: str) -> None:
         """Send an error message to stderr
 
         :param message: the text of message
-        :type message: str
 
            G_important_message() with an additional "ERROR:" string at
            the start will be called in the messenger server process
         """
         self._check_restart_server()
         self.client_conn.send(["ERROR", message])
 
-    def fatal(self, message):
+    def fatal(self, message: str) -> NoReturn:
         """Send an error message to stderr, call sys.exit(1) or raise FatalError
 
         :param message: the text of message
-        :type message: str
 
            This function emulates the behavior of G_fatal_error(). It prints
            an error message to stderr and calls sys.exit(1). If raise_on_error
@@ -264,30 +272,29 @@ def fatal(self, message):
             raise FatalError(message)
         sys.exit(1)
 
-    def debug(self, level, message):
+    def debug(self, level: int, message: str) -> None:
         """Send a debug message to stderr
 
         :param message: the text of message
-        :type message: str
 
            G_debug() will be called in the messenger server process
         """
         self._check_restart_server()
         self.client_conn.send(["DEBUG", level, message])
 
-    def percent(self, n, d, s):
+    def percent(self, n: int, d: int, s: int) -> None:
         """Send a percentage to stderr
 
-        :param message: the text of message
-        :type message: str
-
+        :param n: The current element
+        :param d: Total number of elements
+        :param s: Increment size
 
            G_percent() will be called in the messenger server process
         """
         self._check_restart_server()
         self.client_conn.send(["PERCENT", n, d, s])
 
-    def stop(self):
+    def stop(self) -> None:
         """Stop the messenger server and close the pipe"""
         if self.server is not None and self.server.is_alive():
             self.client_conn.send(
@@ -300,12 +307,11 @@ def stop(self):
         if self.client_conn is not None:
             self.client_conn.close()
 
-    def set_raise_on_error(self, raise_on_error=True):
+    def set_raise_on_error(self, raise_on_error: bool = True) -> None:
         """Set the fatal error behavior
 
         :param raise_on_error: if True a FatalError exception will be
                                raised instead of calling sys.exit(1)
-        :type raise_on_error: bool
 
         - If raise_on_error == True, a FatalError exception will be raised
           if fatal() is called
@@ -315,15 +321,15 @@ def set_raise_on_error(self, raise_on_error=True):
         """
         self.raise_on_error = raise_on_error
 
-    def get_raise_on_error(self):
+    def get_raise_on_error(self) -> bool:
         """Get the fatal error behavior
 
         :returns: True if a FatalError exception will be raised or False if
                   sys.exit(1) will be called in case of invoking fatal()
         """
         return self.raise_on_error
 
-    def test_fatal_error(self, message):
+    def test_fatal_error(self, message: str) -> None:
         """Force the messenger server to call G_fatal_error()"""
         import time
 
@@ -338,7 +344,7 @@ def get_msgr(
     ],
     *args,
     **kwargs,
-):
+) -> Messenger:
     """Return a Messenger instance.
 
        :returns: the Messenger instance.

python/grass/pygrass/messages/testsuite/test_pygrass_messages_doctests.py

@@ -28,7 +28,7 @@
 
 def load_tests(loader, tests, ignore):
     # TODO: this must be somewhere when doctest is called, not here
-    # TODO: ultimate solution is not to use _ as a buildin in lib/python
+    # TODO: ultimate solution is not to use _ as a builtin in lib/python
     # for now it is the only place where it works
     grass.gunittest.utils.do_doctest_gettext_workaround()
     # this should be called at some top level