feat(core): enable actuators and operators to use Rich progress indicators

orchestrator/modules/operators/_explore_orchestration.py

@@ -31,6 +31,10 @@
     _run_operation_harness,
     log_space_details,
 )
+from orchestrator.modules.operators.console_output import (
+    RichConsoleQueue,
+    run_operation_live_updates,
+)
 from orchestrator.modules.operators.discovery_space_manager import DiscoverySpaceManager
 
 moduleLog = logging.getLogger("explore_orchestration")
@@ -144,103 +148,29 @@ def run_explore_operation_core_closure(
 ) -> typing.Callable[[], OperationOutput | None]:
 
     def _run_explore_operation_core() -> OperationOutput:
-        import numpy as np
-        import pandas as pd
-        from rich.console import Console
-        from rich.live import Live
-        from rich.table import Table
+        import ray
+
+        # Create RichConsoleQueue
+        # this needs to be created before operation starts
+        # so operators and actuators can put messages
+        queue_handle = RichConsoleQueue.options(
+            name="RichConsoleQueue", lifetime="detached", get_if_exists=True
+        ).remote()
 
         discovery_space = ray.get(state.discoverySpace.remote())
         operation_id = ray.get(operator.operationIdentifier.remote())
 
-        def output_operation_results(row_limit: int | None) -> Table:
-            df: pd.DataFrame = (
-                discovery_space.complete_measurement_request_with_results_timeseries(
-                    operation_id=operation_id,
-                    output_format="target",
-                )
-            )
-
-            table_title = (
-                f"Latest measurements - {operation_id}"
-                if row_limit
-                else f"Measurements - {operation_id}"
-            )
-            table = Table(title=table_title)
-
-            if df.empty:
-                return table
-
-            # Remove the columns result_index, generatorid and entityIdentifier
-            # We have the constitutive properties in the df, so we don't need to show them
-            df = df.drop(
-                columns=["result_index", "generatorid", "identifier"], errors="ignore"
-            )
-            df.insert(0, "index", np.arange(len(df)))
-
-            # If there is only one experiment drop the experiment column
-            if len(discovery_space.measurementSpace.experiments) == 1:
-                df = df.drop(columns=["experiment_id"], errors="ignore")
-            else:
-                # Convert the experiment column - which is ExperimentReference instances
-                # to experiment identifiers
-                df["experiment_id"] = df["experiment_id"].apply(
-                    lambda x: x.experimentIdentifier
-                )
-
-            # Dynamically determine how many columns can fit the screen
-            console = Console()
-            terminal_width = console.width
-            min_col_width = 12  # Minimum width per column (estimate)
-            max_columns = max(1, terminal_width // min_col_width)
-
-            visible_columns = list(df.columns[:max_columns])
-            hidden_columns = len(df.columns) - max_columns
-
-            if hidden_columns > 0:
-                visible_columns.append(f"... (+{hidden_columns} more)")
-
-            # Add columns manually setting overflow="fold" - this will cause text to wrap
-            # It can't be set at table level
-            for col in visible_columns:
-                table.add_column(col, overflow="fold")
-
-            for row_number, (_, row) in enumerate(df[::-1].iterrows()):
-
-                if row_limit and row_number == row_limit:
-                    break
-
-                # Format numbers to 2 significant figures
-                # Add the row index from the DataFrame to the first column
-                row_data = [
-                    (
-                        f"{row[col]:.2f}"
-                        if isinstance(row[col], float)
-                        else str(row[col])
-                    )
-                    for col in df.columns[:max_columns]
-                ]
-
-                if hidden_columns > 0:
-                    row_data.append("...")
-
-                table.add_row(*row_data)
-
-            return table
-
         state.startMonitoring.remote()
         future = operator.run.remote()
-        finished = []
 
-        # Try to make the table be more or less half of the terminal height
-        table_height = max(int(Console().height / 2) - 4, 4)
-        with Live(output_operation_results(row_limit=table_height)) as live:
-            while not finished:
-                live.update(output_operation_results(row_limit=table_height))
-                finished, _ = ray.wait(ray_waitables=[future], timeout=2)
+        # Start the rich live updates
+        run_operation_live_updates(
+            discovery_space=discovery_space,
+            operation_id=operation_id,
+            console_queue=queue_handle,
+            operation_future=future,
+        )
 
-            # Output the whole table before exiting
-            live.update(output_operation_results(row_limit=None))
         return ray.get(future)  # type: OperationOutput
 
     return _run_explore_operation_core

orchestrator/modules/operators/_orchestrate_core.py

@@ -34,15 +34,6 @@ def log_space_details(discovery_space: "DiscoverySpace"):
 
     print("=========== Discovery Space ===========\n")
     print(pretty.pretty(discovery_space))
-    numberEntities = discovery_space.sample_store.numberOfEntities
-    if numberEntities > 0:
-        e = discovery_space.sample_store.entities[0]
-
-        print("Example entity (first retrieved from sample store):\n")
-        print(
-            orchestrator.utilities.output.pydantic_model_as_yaml(e, exclude_unset=True)
-        )
-        print("\n")
 
 
 def _run_operation_harness(
@@ -72,7 +63,9 @@ def _run_operation_harness(
     # START THE OPERATION
     #
 
-    print("\n=========== Starting Discovery Operation ===========\n")
+    print(
+        f"\n=========== Starting Operation {operation_resource.identifier} ===========\n"
+    )
 
     operation_output = None
     operationStatus = OperationResourceStatus(