Added python style check.

Author: mklomp

.github/workflows/python-check.yml

@@ -0,0 +1,11 @@
+name: python style check
+on: [push, pull_request]
+jobs:
+  check-python-style:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Black Code Formatter
+        uses: lgeiger/black-action@master
+        with:
+          args: ". --check --diff"

README.md

@@ -1,3 +1,12 @@
-# Mirte Python API
+# MIRTE Python API
+This package provides the API for the MIRTE Robot. Please 
+read the MIRTE documentation.
 
-This package provides the API for the Mirte Robot. Please read the Mirte documentation.
+# Test code style
+To check the Python code style run:
+```sh
+pip install black
+black --check **/**.py
+# Fix by using
+black **/**.py
+```

mirte_robot/linetrace.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-#TODO: for debugging purposes we could *also* listen to keyboard events
+# TODO: for debugging purposes we could *also* listen to keyboard events
 
 import sys
 import os
@@ -14,93 +14,102 @@
 from mirte_robot import robot
 
 # Global shared memory objects (TODO: check if we need shared memory, why is server working?)
-stepper = multiprocessing.Value('b', True)
-do_step = multiprocessing.Value('b', False)
-running = multiprocessing.Value('b', False)
+stepper = multiprocessing.Value("b", True)
+do_step = multiprocessing.Value("b", False)
+running = multiprocessing.Value("b", False)
+
 
 # the stop be stopped when:
 # 1) the code finishes
 # 2) the user stopped the process
 # 3) the websocket connection is closed
 def stop_mirte():
-     global running
-     if running.value:
+    global running
+    if running.value:
         process.terminate()
-     running.value = False
+    running.value = False
+
 
 def load_mirte_module(stepper, do_step, running):
 
     def trace_lines(frame, event, arg):
-       global do_step
-       if event != 'line':
-           return
-       # Only return line number to websocket when in step mode
-       if stepper.value:
-          server.send_message_to_all(str(frame.f_lineno))
-       while stepper.value and not do_step.value:
-          time.sleep(.01)
-       do_step.value = False
+        global do_step
+        if event != "line":
+            return
+        # Only return line number to websocket when in step mode
+        if stepper.value:
+            server.send_message_to_all(str(frame.f_lineno))
+        while stepper.value and not do_step.value:
+            time.sleep(0.01)
+        do_step.value = False
 
     def traceit(frame, event, arg):
-       co = frame.f_code
-       filename = co.co_filename
-       if not filename.endswith('mirte.py'):
-          return
-       return trace_lines
+        co = frame.f_code
+        filename = co.co_filename
+        if not filename.endswith("mirte.py"):
+            return
+        return trace_lines
 
     # Send the PID to the web interface and give it some time to call strace on this process
     # to see the output of this script
     server.send_message_to_all("pid:" + str(os.getpid()))
-    time.sleep(0.2) #TODO: let client send signal when strace is started
+    time.sleep(0.2)  # TODO: let client send signal when strace is started
 
     sys.settrace(traceit)
     # rospy.init_node() for some reason needs to be called from __main__ when importing in the regular way.
     # We thereofe need to load teh module from source instead of importing it.
     # https://answers.ros.org/question/266612/rospy-init_node-inside-imported-file
     test = SourceFileLoader("mirte", "/home/mirte/workdir/mirte.py").load_module()
 
-    # Stop the motors. The atexit call in robot.py does not work when running from a subprocess: 
+    # Stop the motors. The atexit call in robot.py does not work when running from a subprocess:
     # https://stackoverflow.com/questions/34506638/how-to-register-atexit-function-in-pythons-multiprocessing-subprocess
     # TODO: this assumes we have the robot initlized under variable 'mirte'. As soon as we let them create their own python,
     # this might not work anymore.
-    if hasattr(test, 'mirte'):
-       test.mirte.stop()
+    if hasattr(test, "mirte"):
+        test.mirte.stop()
 
     # Sending the linetrace 0 to the client
     server.send_message_to_all("0")
     running.value = False
     stop_mirte()
 
-process = multiprocessing.Process(target = load_mirte_module, args=(stepper, do_step))
+
+process = multiprocessing.Process(target=load_mirte_module, args=(stepper, do_step))
+
 
 def start_mirte():
     global process
     # process should already have been killed after stop, or disconnect
     # but, just in case make sure to stop this
     if process.is_alive():
-      process.terminate()
-    process = multiprocessing.Process(target = load_mirte_module, args=(stepper, do_step, running))
+        process.terminate()
+    process = multiprocessing.Process(
+        target=load_mirte_module, args=(stepper, do_step, running)
+    )
     process.start()
 
+
 def client_left(client, server):
     stop_mirte()
 
+
 def message_received(client, server, message):
-   global stepper, do_step, running
-
-   if message == "b": #break (pause)
-      stepper.value = True
-   if message == "c": #continue (play)
-      stepper.value = False
-      if not running.value:
-         start_mirte()
-         running.value = True
-   if message == "s": #step (step)
-      do_step.value = True
-   if message == "e": #exit (stop)
-      stepper.value = True
-      do_step.value = False
-      stop_mirte()
+    global stepper, do_step, running
+
+    if message == "b":  # break (pause)
+        stepper.value = True
+    if message == "c":  # continue (play)
+        stepper.value = False
+        if not running.value:
+            start_mirte()
+            running.value = True
+    if message == "s":  # step (step)
+        do_step.value = True
+    if message == "e":  # exit (stop)
+        stepper.value = True
+        do_step.value = False
+        stop_mirte()
+
 
 server = WebsocketServer(host="0.0.0.0", port=8001, loglevel=logging.CRITICAL)
 server.set_fn_message_received(message_received)

mirte_robot/robot.py

@@ -51,6 +51,7 @@
 
 mirte = {}
 
+
 class Robot:
     """Robot API
 
@@ -69,6 +70,7 @@ class Robot:
     # function. Therefor the sphynx documentation is not able to process
     # this class anymore.
     _instance = None
+
     def __new__(cls, *args, **kwargs):
         if cls._instance is None:
             cls._instance = super(Robot, cls).__new__(cls)
@@ -84,24 +86,24 @@ def __init__(
         if getattr(self, "_initialized", False):
             return
 
-#        Parameters:
-#            machine_namespace (Optional[str], optional): The Namespace from '/' to the ROS namespace for the specific Mirte. Defaults to "/{HOSTNAME}". (This only has to be changed when running the Robot API from a different machine directly. It is configured correctly for the Web interface)
-#            hardware_namespace (str, optional): The namespace for the hardware peripherals. Defaults to "io".
-
-        self._machine_namespace = "" #(
-#            machine_namespace
-#            if machine_namespace and validate_namespace(machine_namespace)
-#            else "/" + platform.node().replace("-", "_").lower()
-#       )
-        self._hardware_namespace = "/io" #(
-#            hardware_namespace
-#            if validate_namespace(
-#                hardware_namespace
-#                if hardware_namespace.startswith("/")
-#                else (self._machine_namespace + "/" + hardware_namespace)
-#            )
-#            else "io"
-#        )
+        #        Parameters:
+        #            machine_namespace (Optional[str], optional): The Namespace from '/' to the ROS namespace for the specific Mirte. Defaults to "/{HOSTNAME}". (This only has to be changed when running the Robot API from a different machine directly. It is configured correctly for the Web interface)
+        #            hardware_namespace (str, optional): The namespace for the hardware peripherals. Defaults to "io".
+
+        self._machine_namespace = ""  # (
+        #            machine_namespace
+        #            if machine_namespace and validate_namespace(machine_namespace)
+        #            else "/" + platform.node().replace("-", "_").lower()
+        #       )
+        self._hardware_namespace = "/io"  # (
+        #            hardware_namespace
+        #            if validate_namespace(
+        #                hardware_namespace
+        #                if hardware_namespace.startswith("/")
+        #                else (self._machine_namespace + "/" + hardware_namespace)
+        #            )
+        #            else "io"
+        #        )
 
         ROS_DISTRO = os.getenv("ROS_DISTRO")
 
@@ -415,7 +417,6 @@ def finalize(node: rclpy.node.Node, motors: dict[str, rclpy.client.Client]):
                 self.color_services[sensor]["RGBW"].wait_for_service()
                 self.color_services[sensor]["HSL"].wait_for_service()
 
-
         self._node.destroy_client(list_parameters)
 
         self._get_digital_pin_value_service = self._node.create_client(
@@ -444,14 +445,14 @@ def _call_service(
     ) -> rclpy.client.SrvTypeResponse:
 
         with self._lock:
-          future_response = client.call_async(request)
-          while not future_response.done() and not self._stopping:
-            rclpy.spin_once(self._node, timeout_sec=0.1)
+            future_response = client.call_async(request)
+            while not future_response.done() and not self._stopping:
+                rclpy.spin_once(self._node, timeout_sec=0.1)
 
         if self._stopping:
-          with self._lock:
-             self._stopping = False
-          self._at_exit()
+            with self._lock:
+                self._stopping = False
+            self._at_exit()
 
         return future_response.result()
 
@@ -863,13 +864,11 @@ def stop(self) -> None:
     def getROSNode(self):
         return self._node
 
-
     def _signal_handler(self, sig, frame):
         self._stopping = True
 
-        if (not self._lock.locked()):
-          self._at_exit()
-
+        if not self._lock.locked():
+            self._at_exit()
 
     def _at_exit(self) -> None:
         self.stop()