ad-sdl/madsci/madsci-experiments

ExperimentBase (MadsciClientMixin)
  ├── ExperimentScript     # run() -> manage_experiment() -> run_experiment()
  ├── ExperimentNotebook   # start() / end() with Rich display
  ├── ExperimentTUI        # Textual TUI with threading.Event pause/cancel
  └── ExperimentNode       # Wraps RestNode internally, exposes run_experiment as action

from madsci.common.types.experiment_types import ExperimentDesign
design = ExperimentDesign(
    experiment_name="Synthesis Optimization",
    experiment_description="Optimize reaction conditions for compound X",
)
# Or load from YAML
design = ExperimentDesign.from_yaml("experiment_design.yaml")

from madsci.experiment_application.experiment_script import ExperimentScript
from madsci.common.types.experiment_types import ExperimentDesign
class SynthesisExperiment(ExperimentScript):
    experiment_design = ExperimentDesign(
        experiment_name="Synthesis Run",
        experiment_description="Automated synthesis workflow",
    )
    def run_experiment(self, sample_id: str = "default", cycles: int = 3) -> dict:
        """Core experiment logic. Called within manage_experiment() context."""
        results = []
        for i in range(cycles):
            result = self.workcell_client.run_workflow(
                "synthesis", parameters={"sample_id": sample_id, "cycle": i}
            )
            results.append(result)
            self.logger.info("Cycle completed", cycle=i, result=result)
        return {"sample_id": sample_id, "results": results}
if __name__ == "__main__":
    SynthesisExperiment.main(sample_id="ABC123", cycles=5)

# Cell 1: Setup
from my_experiment import MyNotebookExperiment
exp = MyNotebookExperiment()
# Cell 2: Start
exp.start(run_name="Exploration Run 1")
# Cell 3: Execute
result = exp.run_workflow("characterize", parameters={"sample": "S1"})
# Cell 4: Visualize
exp.display(result, title="Characterization Results")
# Cell 5: End
exp.end()

with MyNotebookExperiment() as exp:
    result = exp.run_workflow("synthesis")
    exp.display(result, title="Results")

from madsci.experiment_application.experiment_tui import ExperimentTUI
from madsci.common.types.experiment_types import ExperimentDesign
class OperatorExperiment(ExperimentTUI):
    experiment_design = ExperimentDesign(
        experiment_name="Operator-Assisted Synthesis",
    )
    def run_experiment(self) -> dict:
        results = []
        for step in range(10):
            self.check_experiment_status()  # Handles pause/cancel locally
            result = self.workcell_client.run_workflow("step", parameters={"n": step})
            results.append(result)
        return {"steps_completed": len(results)}
if __name__ == "__main__":
    OperatorExperiment().run_tui()

from madsci.experiment_application.experiment_node import ExperimentNode
from madsci.common.types.experiment_types import ExperimentDesign
class RemoteExperiment(ExperimentNode):
    experiment_design = ExperimentDesign(
        experiment_name="Remote Synthesis",
    )
    def run_experiment(self, sample_id: str, temperature: float = 25.0) -> dict:
        return self.workcell_client.run_workflow(
            "process", parameters={"sample_id": sample_id, "temp": temperature}
        )
if __name__ == "__main__":
    RemoteExperiment().start_server()

with self.manage_experiment(run_name="Run 1") as exp:
    # Experiment started, logging context established
    result = exp.workcell_client.run_workflow("my_workflow")
    # On success: end_experiment(COMPLETED) called automatically
    # On exception: handle_exception() called, then re-raised

exp.start_experiment_run(run_name="Manual Run")
try:
    exp.run_experiment()
    exp.end_experiment(status=ExperimentStatus.COMPLETED)
except Exception as e:
    exp.handle_exception(e)
    raise

def handle_exception(self, exception: Exception) -> None:
    if isinstance(exception, RecoverableError):
        self.logger.warning("Recoverable error, retrying", error=str(exception))
        return  # Don't end experiment
    super().handle_exception(exception)  # Logs error + ends with FAILED

class ExperimentStatus(str, Enum):
    IN_PROGRESS = "in_progress"
    PAUSED = "paused"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    UNKNOWN = "unknown"

from madsci.common.exceptions import (
    ExperimentCancelledError,    # Raised when experiment cancelled externally
    ExperimentFailedError,       # Raised when experiment fails externally
    ExperimentPauseTimeoutError, # Raised when pause exceeds max_pause_wait
)

self.event_client       # EventClient - logging and events
self.logger             # Alias for event_client
self.resource_client    # ResourceClient - inventory
self.data_client        # DataClient - data storage
self.experiment_client  # ExperimentClient - experiment lifecycle
self.workcell_client    # WorkcellClient - workflow execution
self.location_client    # LocationClient - locations
self.lab_client         # LabClient - lab coordination

# Common fields (ExperimentBaseConfig):
lab_server_url: Optional[AnyUrl]         # Lab manager for service discovery
event_server_url: Optional[AnyUrl]       # Override Event Manager URL
experiment_server_url: Optional[AnyUrl]  # Override Experiment Manager URL
workcell_server_url: Optional[AnyUrl]    # Override Workcell Manager URL
data_server_url: Optional[AnyUrl]        # Override Data Manager URL
resource_server_url: Optional[AnyUrl]    # Override Resource Manager URL
location_server_url: Optional[AnyUrl]    # Override Location Manager URL
max_pause_wait: Optional[float]          # Max seconds to wait while paused (None = forever)

madsci new experiment
# Choose modality: script, notebook, tui, node
# Provide: experiment_name, description

def run_experiment(self):
    for batch in batches:
        self.check_experiment_status()  # Blocks if paused, raises if cancelled
        process(batch)