Lazy load EStop (#3810)

Author: Andrewyx

src/software/embedded/robot_diagnostics_cli/embedded_communication.py

@@ -37,19 +37,29 @@ def __init__(self):
             True,
         )
 
-        # initialising the estop
         self.estop_reader = None
         self.should_send_stop = False
-        self.estop_mode, self.estop_path = get_estop_config(
-            keyboard_estop=False, disable_communication=False
-        )
+        self.estop_mode = EstopMode.PHYSICAL_ESTOP
+        self.estop_path = None
+
+    def setup_estop(self) -> bool:
+        """Lazily sets up the physical estop reader the first time a command
+        that actuates the robot's electrical/mechanical components is run.
+
+        :return: True if a physical estop is connected and ready, False otherwise
+        """
+        if self.estop_reader is not None:
+            return True
 
-        # Always in PHYSICAL_ESTOP mode within CLI
         try:
+            self.estop_mode, self.estop_path = get_estop_config(
+                keyboard_estop=False, disable_communication=False
+            )
             self.estop_reader = tbots_cpp.ThreadedEstopReader(self.estop_path, 115200)
             self.__update_estop_state()
-        except Exception as e:
-            raise Exception(f"Invalid Estop found at location {self.estop_path} as {e}")
+            return True
+        except Exception:
+            return False
 
     def __receive_robot_status(self, robot_status: Message) -> None:
         """Updates the dynamic information with the new RobotStatus message.

src/software/embedded/robot_diagnostics_cli/embedded_data.py

@@ -64,14 +64,9 @@ def get_kick_coeff(self) -> str:
     def get_chip_pulse_width(self) -> str:
         return self._get_value(ROBOT_CHIP_PULSE_WIDTH_CONFIG_KEY)
 
-    def get_current_draw(self) -> str:
-        return self._get_value(ROBOT_CURRENT_DRAW_CONFIG_KEY)
-
-    def get_battery_volt(self) -> str:
-        return self._get_value(ROBOT_BATTERY_VOLTAGE_CONFIG_KEY)
-
-    def get_cap_volt(self) -> str:
-        return self._get_value(ROBOT_CAPACITOR_VOLTAGE_CONFIG_KEY)
+    # TODO: #3809
+    # def get_cap_volt(self) -> str:
+    #     return self._get_value(ROBOT_CAPACITOR_VOLTAGE_CONFIG_KEY)
 
     def __clamp(self, val: float, min_val: float, max_val: float) -> float:
         """Simple Math Clamp function (Faster than numpy & fewer dependencies)

src/software/embedded/robot_diagnostics_cli/robot_diagnostics_cli.py

@@ -1,6 +1,7 @@
 import subprocess
 import logging
 import typer as Typer
+from typer.main import get_command
 from rich import print
 from rich.live import Live
 from rich.table import Table
@@ -52,7 +53,9 @@ def __init__(self, embedded_communication: EmbeddedCommunication) -> None:
         self.app.command(short_help="Shows TOML Config Values")(self.config)
         self.app.command(short_help="Prints Thunderloop Logs")(self.log)
         self.app.command(short_help="Prints Thunderloop Status")(self.status)
-        self.app.command(short_help="Restarts Thunderloop")(self.restart_thunderloop)
+        self.app.command(short_help="Restarts Thunderloop")(self.restart)
+        self.app.command(short_help="Starts Thunderloop")(self.start)
+        self.app.command(short_help="Stops Thunderloop")(self.stop)
         # Communication object responsible for proto execution/transmission
         self.embedded_communication = embedded_communication
         # Data handler responsible for disk information
@@ -101,6 +104,24 @@ def wrapper(self, *args, **kwargs):
 
         return decorator
 
+    def requires_estop(func):
+        """Decorator that ensures a physical estop is plugged in before running a
+        command that actuates the robot's electrical/mechanical components.
+        """
+
+        @wraps(func)
+        def wrapper(self, *args, **kwargs):
+            if not self.embedded_communication.setup_estop():
+                logging.warning(
+                    "[bold yellow]No estop plugged in: this command controls the "
+                    "robot's electrical/mechanical components and requires an estop. "
+                    "Plug one in and try again. Command not executed.[/bold yellow]"
+                )
+                return
+            return func(self, *args, **kwargs)
+
+        return wrapper
+
     def __generate_stats_table(self) -> Table:
         """Make a new table with robot status information."""
         table = Table()
@@ -161,21 +182,12 @@ def __generate_config_table(self) -> Table:
             f"{ROBOT_CHIP_PULSE_WIDTH_CONFIG_KEY}",
             self.embedded_data.get_chip_pulse_width(),
         )
-        table.add_row(
-            "Battery Voltage",
-            f"{ROBOT_BATTERY_VOLTAGE_CONFIG_KEY}",
-            self.embedded_data.get_battery_volt(),
-        )
-        table.add_row(
-            "Battery Current Draw",
-            f"{ROBOT_CURRENT_DRAW_CONFIG_KEY}",
-            self.embedded_data.get_current_draw(),
-        )
-        table.add_row(
-            "Capacitor Voltage",
-            f"{ROBOT_CAPACITOR_VOLTAGE_CONFIG_KEY}",
-            self.embedded_data.get_cap_volt(),
-        )
+        # TODO: #3809
+        # table.add_row(
+        #     "Capacitor Voltage",
+        #     f"{ROBOT_CAPACITOR_VOLTAGE_CONFIG_KEY}",
+        #     self.embedded_data.get_cap_volt(),
+        # )
         return table
 
     def stats(self) -> None:
@@ -193,16 +205,27 @@ def status(self):
         log = subprocess.call(["service", "thunderloop", "status"])
         print(log)
 
-    def restart_thunderloop(self):
+    def restart(self):
         """CLI Command to restart Thunderloop service and print status"""
-        subprocess.run(["service", "thunderloop", "restart"])
+        subprocess.run(["sudo", "service", "thunderloop", "restart"])
+        self.status()
+
+    def start(self):
+        """CLI Command to start Thunderloop service and print status"""
+        subprocess.run(["sudo", "service", "thunderloop", "start"])
+        self.status()
+
+    def stop(self):
+        """CLI Command to stop Thunderloop service and print status"""
+        subprocess.run(["sudo", "service", "thunderloop", "stop"])
         self.status()
 
     def log(self):
         """CLI Command to print device logs"""
         log = subprocess.call(["sudo", "journalctl", "-f", "-n", "100"])
         print(log)
 
+    @requires_estop
     @catch_interrupt_exception()
     def rotate(
         self,
@@ -227,6 +250,7 @@ def rotate(
             description,
         )
 
+    @requires_estop
     @catch_interrupt_exception()
     def move(
         self,
@@ -256,6 +280,7 @@ def move(
             description,
         )
 
+    @requires_estop
     @catch_interrupt_exception()
     def chip(
         self,
@@ -289,6 +314,7 @@ def chip(
             self.embedded_communication.run_primitive(primitive)
             print(description)
 
+    @requires_estop
     @catch_interrupt_exception()
     def kick(
         self,
@@ -331,6 +357,7 @@ def kick(
                 "Recharging...",
             )
 
+    @requires_estop
     @catch_interrupt_exception()
     def dribble(
         self,
@@ -358,6 +385,7 @@ def dribble(
             description,
         )
 
+    @requires_estop
     @catch_interrupt_exception()
     def move_wheel(
         self,
@@ -395,7 +423,14 @@ def emote(self):
         # TODO (#3434): Add an emote function!
         return
 
+    def run(self) -> None:
+        """Launch the interactive Diagnostics CLI shell."""
+        self.app._add_completion = False
+        command = get_command(self.app)
+        command.add_help_option = False
+        command()
+
 
 if __name__ == "__main__":
     with EmbeddedCommunication() as embedded_communication:
-        RobotDiagnosticsCLI(embedded_communication).app()
+        RobotDiagnosticsCLI(embedded_communication).run()