feat(controller): 8 - add command line interface for running on OSCAR

autora/controller/__main__.py

@@ -0,0 +1,60 @@
+import logging
+import os
+import pathlib
+from typing import Optional
+
+import typer
+import yaml
+
+from autora.controller import Controller
+from autora.theorist.__main__ import _configure_logger
+
+_logger = logging.getLogger(__name__)
+
+
+def main(
+    manager: pathlib.Path=typer.Argument(..., help="Manager path"),
+    directory: pathlib.Path=typer.Argument(..., help="Directory path"),
+    step_name: Optional[str] = typer.Argument(None, help="Name of step"),
+    verbose: bool = typer.Option(False, help="Turns on info logging level."),
+    debug: bool = typer.Option(False, help="Turns on debug logging level."),
+):
+    _logger.debug("initializing")
+    _configure_logger(debug, verbose)
+    controller_ = _load_manager(manager)
+
+    _logger.debug(f"loading manager state from {directory=}")
+    controller_.load(directory)
+
+    if step_name is not None:
+        controller_ = _set_next_step_name(controller_, step_name)
+
+    _logger.info("running next step")
+    next(controller_)
+
+    _logger.debug(f"last result: {controller_.state.history[-1]}")
+
+    _logger.info("writing out results")
+    controller_.dump(directory)
+
+    return
+
+
+def _load_manager(path: pathlib.Path) -> Controller:
+    _logger.debug(f"_load_manager: loading from {path=} (currently in {os.getcwd()})")
+    with open(path, "r") as f:
+        controller_ = yaml.load(f, yaml.Loader)
+    assert isinstance(
+        controller_, Controller
+    ), f"controller type {type(controller_)=} unsupported"
+    return controller_
+
+
+def _set_next_step_name(controller: Controller, step_name: str):
+    _logger.info(f"setting next {step_name=}")
+    controller.planner = lambda _: step_name
+    return controller
+
+
+if __name__ == "__main__":
+    typer.run(main)

autora/controller/base.py

@@ -2,16 +2,14 @@
 from __future__ import annotations
 
 import logging
-from typing import Callable, Mapping, Optional, TypeVar
-
-_logger = logging.getLogger(__name__)
+from typing import Callable, Generic, Mapping, Optional
 
+from autora.controller.protocol import State
 
-State = TypeVar("State")
-ExecutorName = TypeVar("ExecutorName", bound=str)
+_logger = logging.getLogger(__name__)
 
 
-class BaseController:
+class BaseController(Generic[State]):
     """
     Runs an experimentalist, theorist and experiment runner in a loop.
 
@@ -36,8 +34,8 @@ class BaseController:
     def __init__(
         self,
         state: State,
-        planner: Callable[[State], ExecutorName],
-        executor_collection: Mapping[ExecutorName, Callable[[State], State]],
+        planner: Callable[[State], str],
+        executor_collection: Mapping[str, Callable[[State], State]],
         monitor: Optional[Callable[[State], None]] = None,
     ):
         """

autora/controller/controller.py

@@ -2,21 +2,23 @@
 from __future__ import annotations
 
 import logging
+import pathlib
 from typing import Callable, Dict, Optional
 
 from sklearn.base import BaseEstimator
 
-from autora.controller.base import BaseController, ExecutorName
+from autora.controller.base import BaseController
 from autora.controller.executor import make_online_executor_collection
 from autora.controller.planner import last_result_kind_planner
+from autora.controller.serializer import HistorySerializer
 from autora.controller.state import History
 from autora.experimentalist.pipeline import Pipeline
 from autora.variable import VariableCollection
 
 _logger = logging.getLogger(__name__)
 
 
-class Controller(BaseController):
+class Controller(BaseController[History]):
     """
     Runs an experimentalist, experiment runner, and theorist in order.
 
@@ -33,13 +35,13 @@ class Controller(BaseController):
 
     def __init__(
         self,
-        metadata: Optional[VariableCollection],
+        metadata: Optional[VariableCollection] = None,
         theorist: Optional[BaseEstimator] = None,
         experimentalist: Optional[Pipeline] = None,
         experiment_runner: Optional[Callable] = None,
         params: Optional[Dict] = None,
         monitor: Optional[Callable[[History], None]] = None,
-        planner: Callable[[History], ExecutorName] = last_result_kind_planner,
+        planner: Callable[[History], str] = last_result_kind_planner,
     ):
         """
         Args:
@@ -102,3 +104,14 @@ def __init__(
     def seed(self, **kwargs):
         for key, value in kwargs.items():
             self.state = self.state.update(**{key: value})
+
+    def load(self, directory: pathlib.Path):
+        serializer = HistorySerializer(directory)
+        state = serializer.load()
+        self.state = state
+        return
+
+    def dump(self, directory: pathlib.Path):
+        serializer = HistorySerializer(directory)
+        serializer.dump(self.state)
+        return

autora/controller/executor.py

@@ -6,11 +6,13 @@
 
 import copy
 import logging
+import pprint
 from functools import partial
 from types import MappingProxyType
 from typing import Callable, Dict, Iterable, Literal, Optional, Tuple, Union
 
 import numpy as np
+import pandas as pd
 from sklearn.base import BaseEstimator
 
 from autora.controller.protocol import SupportsControllerState
@@ -27,15 +29,27 @@ def experimentalist_wrapper(
     params_ = resolve_state_params(params, state)
     new_conditions = pipeline(**params_)
 
-    assert isinstance(new_conditions, Iterable)
-    # If the pipeline gives us an iterable, we need to make it into a concrete array.
-    # We can't move this logic to the Pipeline, because the pipeline doesn't know whether
-    # it's within another pipeline and whether it should convert the iterable to a
-    # concrete array.
-    new_conditions_values = list(new_conditions)
-    new_conditions_array = np.array(new_conditions_values)
+    if isinstance(new_conditions, pd.DataFrame):
+        new_conditions_array = new_conditions
+    elif isinstance(new_conditions, np.ndarray):
+        _logger.warning(
+            f"{new_conditions=} is an ndarray, so variable confusion is a possibility"
+        )
+        new_conditions_array = new_conditions
+    elif isinstance(new_conditions, np.recarray):
+        new_conditions_array = new_conditions
+    elif isinstance(new_conditions, Iterable):
+        # If the pipeline gives us an iterable, we need to make it into a concrete array.
+        # We can't move this logic to the Pipeline, because the pipeline doesn't know whether
+        # it's within another pipeline and whether it should convert the iterable to a
+        # concrete array.
+        new_conditions_values = list(new_conditions)
+        new_conditions_array = np.array(new_conditions_values)
+    else:
+        raise NotImplementedError(
+            f"Can't handle experimentalist output {new_conditions=}"
+        )
 
-    assert isinstance(new_conditions_array, np.ndarray)  # Check the object is bounded
     new_state = state.update(conditions=[new_conditions_array])
     return new_state
 
@@ -46,8 +60,15 @@ def experiment_runner_wrapper(
     """Interface for running the experiment runner callable."""
     params_ = resolve_state_params(params, state)
     x = state.conditions[-1]
-    y = callable(x, **params_)
-    new_observations = np.column_stack([x, y])
+    output = callable(x, **params_)
+
+    if isinstance(x, pd.DataFrame):
+        new_observations = output
+    elif isinstance(x, np.ndarray):
+        new_observations = np.column_stack([x, output])
+    else:
+        raise NotImplementedError(f"type {x=} not supported")
+
     new_state = state.update(observations=[new_observations])
     return new_state
 
@@ -59,13 +80,35 @@ def theorist_wrapper(
     params_ = resolve_state_params(params, state)
     metadata = state.metadata
     observations = state.observations
-    all_observations = np.row_stack(observations)
-    n_xs = len(metadata.independent_variables)
-    x, y = all_observations[:, :n_xs], all_observations[:, n_xs:]
-    if y.shape[1] == 1:
-        y = y.ravel()
+
+    if isinstance(observations[-1], pd.DataFrame):
+        all_observations = pd.concat(observations)
+        iv_names = [iv.name for iv in metadata.independent_variables]
+        dv_names = [dv.name for dv in metadata.dependent_variables]
+        x, y = all_observations[iv_names], all_observations[dv_names]
+    elif isinstance(observations[-1], np.ndarray):
+        all_observations = np.row_stack(observations)
+        n_xs = len(metadata.independent_variables)
+        x, y = all_observations[:, :n_xs], all_observations[:, n_xs:]
+        if y.shape[1] == 1:
+            y = y.ravel()
+    else:
+        raise NotImplementedError(f"type {observations[-1]=} not supported")
+
     new_theorist = copy.deepcopy(estimator)
     new_theorist.fit(x, y, **params_)
+
+    try:
+        _logger.debug(
+            f"fitted {new_theorist=}\nnew_theorist.__dict__:"
+            f"\n{pprint.pformat(new_theorist.__dict__)}"
+        )
+    except AttributeError:
+        _logger.debug(
+            f"fitted {new_theorist=} "
+            f"new_theorist has no __dict__ attribute, so no results are shown"
+        )
+
     new_state = state.update(theories=[new_theorist])
     return new_state
 

autora/controller/protocol.py

@@ -128,9 +128,7 @@ def __call__(self, __state: State, __params: Dict) -> State:
         ...
 
 
-ExecutorName = TypeVar("ExecutorName", bound=str)
-
-ExecutorCollection = Mapping[ExecutorName, Executor]
+ExecutorCollection = Mapping[str, Executor]
 
 
 @runtime_checkable

autora/controller/serializer/__init__.py

@@ -1,15 +1,12 @@
 import pickle
 import tempfile
+from abc import abstractmethod
 from pathlib import Path
-from typing import Mapping, NamedTuple, Optional, Type, Union
+from typing import Generic, Mapping, NamedTuple, Union
 
 import numpy as np
 
-from autora.controller.protocol import (
-    ResultKind,
-    SupportsControllerStateHistory,
-    SupportsLoadDump,
-)
+from autora.controller.protocol import ResultKind, State, SupportsLoadDump
 from autora.controller.serializer import yaml_ as YAMLSerializer
 from autora.controller.state import History
 
@@ -25,7 +22,17 @@ class _LoadSpec(NamedTuple):
     mode: str
 
 
-class HistorySerializer:
+class StateSerializer(Generic[State]):
+    @abstractmethod
+    def load(self) -> State:
+        ...
+
+    @abstractmethod
+    def dump(self, ___state: State):
+        ...
+
+
+class HistorySerializer(StateSerializer[History]):
     """Serializes and deserializes History objects."""
 
     def __init__(
@@ -51,7 +58,7 @@ def __init__(
             ".pickle": _LoadSpec(pickle, "rb"),
         }
 
-    def dump(self, data_collection: SupportsControllerStateHistory):
+    def dump(self, data_collection: History):
         """
 
         Args:
@@ -129,7 +136,7 @@ def dump(self, data_collection: SupportsControllerStateHistory):
             with open(Path(path, filename), mode) as f:
                 serializer.dump(container, f)
 
-    def load(self, cls: Type[SupportsControllerStateHistory] = History):
+    def load(self) -> History:
         """
 
         Examples:
@@ -185,7 +192,7 @@ def load(self, cls: Type[SupportsControllerStateHistory] = History):
                 loaded_object = serializer.load(f)
                 data.append(loaded_object)
 
-        data_collection = cls(history=data)
+        data_collection = History(history=data)
 
         return data_collection