adding pio rm

pioreactor/automations/dosing/base.py

@@ -464,7 +464,7 @@ def execute_io_action(
 
     @property
     def most_stale_time(self) -> datetime:
-        return min(self.latest_normalized_od_at, self.latest_growth_rate_at)
+        return min(self.latest_normalized_od_at, self.latest_growth_rate_at, self.latest_od_at)
 
     @property
     def latest_growth_rate(self) -> float:
@@ -514,9 +514,9 @@ def latest_od(self) -> dict[pt.PdChannel, float]:
                 raise exc.JobRequiredError("`od_reading` should be Ready.")
 
         # check most stale time
-        if (current_utc_datetime() - self.most_stale_time).seconds > 5 * 60:
+        if (current_utc_datetime() - self.latest_od_at).seconds > 5 * 60:
             raise exc.JobRequiredError(
-                f"readings are too stale (over 5 minutes old) - is `od_reading` running?. Last reading occurred at {self.most_stale_time}."
+                f"readings are too stale (over 5 minutes old) - is `od_reading` running?. Last reading occurred at {self.latest_od_at}."
             )
 
         assert self._latest_od is not None

pioreactor/cli/pios.py

@@ -121,14 +121,14 @@ def sync_config_files(unit: str, shared: bool, specific: bool) -> None:
                 raise e
         return
 
-    @pios.command("cp", short_help="cp a file across clusters")
+    @pios.command("cp", short_help="cp a file across the cluster")
     @click.argument("filepath", type=click.Path(exists=True))
     @click.option(
         "--units",
         multiple=True,
         default=(UNIVERSAL_IDENTIFIER,),
         type=click.STRING,
-        help="specify a Pioreactor name, default is all active units",
+        help="specify a Pioreactor name, default is all active non-leader units",
     )
     def cp(
         filepath: str,
@@ -150,6 +150,46 @@ def _thread_function(unit: str) -> bool:
         for unit in units:
             _thread_function(unit)
 
+    @pios.command("rm", short_help="rm a file across the cluster")
+    @click.argument("filepath", type=click.Path(exists=True))
+    @click.option(
+        "--units",
+        multiple=True,
+        default=(UNIVERSAL_IDENTIFIER,),
+        type=click.STRING,
+        help="specify a Pioreactor name, default is all active non-leader units",
+    )
+    def rm(
+        filepath: str,
+        units: tuple[str, ...],
+    ) -> None:
+        logger = create_logger("rm", unit=get_unit_name(), experiment=UNIVERSAL_EXPERIMENT)
+        units = remove_leader(universal_identifier_to_all_workers(units))
+
+        from sh import ssh  # type: ignore
+        from sh import ErrorReturnCode_255  # type: ignore
+        from sh import ErrorReturnCode_1
+        from shlex import join  # https://docs.python.org/3/library/shlex.html#shlex.quote
+
+        command = join(["rm", filepath])
+
+        def _thread_function(unit: str) -> bool:
+            logger.debug(f"Removing {unit}:{filepath}...")
+            try:
+                ssh(add_local(unit), command)
+                return True
+            except ErrorReturnCode_255 as e:
+                logger.error(f"Unable to connect to unit {unit}. {e.stderr.decode()}")
+                logger.debug(e, exc_info=True)
+                return False
+            except ErrorReturnCode_1 as e:
+                logger.error(f"Error occurred: {e}. See logs for more.")
+                logger.debug(e.stderr, exc_info=True)
+                return False
+
+        for unit in units:
+            _thread_function(unit)
+
     @pios.command("update", short_help="update PioreactorApp on workers")
     @click.option(
         "--units",