Accessing "State-Dependent Properties"¶

Some experimentalists, experiment runners and theorists require access to the values created during the cycle execution, e.g. experimentalists which require access to the current best model or the observed data. These data update each cycle, and so cannot easily be set using simple params.

For this case, it is possible to use "state-dependent properties" in the params dictionary. These are the following strings, which will be replaced during execution by their respective current values:

• "%experiment_data.conditions[-1]%": the last observed independent variables
• "%experiment_data.observations[-1]%": the last observed dependent variables
• "%experiment_data.conditions%": all the observed independent variables, concatenated into a single array
• "%experiment_data.observations%": all the observed dependent variables, concatenated into a single array
• "%models[-1]%": the last fitted theorist
• "%models%": all the fitted theorists

In the following example, we use the "observations.ivs" cycle property for an experimentalist which excludes those conditions which have already been seen.

In [ ]:

# Uncomment the following line when running on Google Colab
# !pip install autora

# Uncomment the following line when running on Google Colab # !pip install autora

In [ ]:

import numpy as np
from autora.experimentalist.pipeline import make_pipeline
from autora.variable import VariableCollection, Variable
from sklearn.linear_model import LinearRegression
import pandas as pd
from functools import partial

from autora.workflow import Cycle

import numpy as np from autora.experimentalist.pipeline import make_pipeline from autora.variable import VariableCollection, Variable from sklearn.linear_model import LinearRegression import pandas as pd from functools import partial from autora.workflow import Cycle

In [ ]:

def ground_truth(x):
    return x + 1
variables = VariableCollection(
   independent_variables=[Variable(name="x", allowed_values=range(10))],
   dependent_variables=[Variable(name="y")],
   )
random_sampler_rng = np.random.default_rng(seed=180)
def custom_random_sampler(conditions, n):
    sampled_conditions = random_sampler_rng.choice(conditions, size=n, replace=False)
    return sampled_conditions
def exclude_conditions(conditions, excluded_conditions):
    remaining_conditions = list(set(conditions) - set(excluded_conditions.flatten()))
    return remaining_conditions
unobserved_data_experimentalist = make_pipeline([
    variables.independent_variables[0].allowed_values,
    exclude_conditions,
    custom_random_sampler,
    partial(pd.DataFrame, columns=["x"])
    ]
)
example_theorist = LinearRegression()
def get_example_synthetic_experiment_runner():
    rng = np.random.default_rng(seed=180)
    def runner(x):
        return ground_truth(x) + rng.normal(0, 0.1, x.shape)
    def dataframe_runner(conditions_df: pd.DataFrame):
        observations_df = conditions_df.copy()
        observations_df["y"] = runner(conditions_df["x"])
        return observations_df
    return dataframe_runner

example_synthetic_experiment_runner = get_example_synthetic_experiment_runner()

cycle_with_state_dep_properties = Cycle(
    variables=variables,
    theorist=example_theorist,
    experimentalist=unobserved_data_experimentalist,
    experiment_runner=example_synthetic_experiment_runner,
    params={
        "experimentalist": {
            "exclude_conditions": {"excluded_conditions": "%experiment_data.conditions%"},
            "custom_random_sampler": {"n": 1}
        }
    }
)

def ground_truth(x): return x + 1 variables = VariableCollection( independent_variables=[Variable(name="x", allowed_values=range(10))], dependent_variables=[Variable(name="y")], ) random_sampler_rng = np.random.default_rng(seed=180) def custom_random_sampler(conditions, n): sampled_conditions = random_sampler_rng.choice(conditions, size=n, replace=False) return sampled_conditions def exclude_conditions(conditions, excluded_conditions): remaining_conditions = list(set(conditions) - set(excluded_conditions.flatten())) return remaining_conditions unobserved_data_experimentalist = make_pipeline([ variables.independent_variables[0].allowed_values, exclude_conditions, custom_random_sampler, partial(pd.DataFrame, columns=["x"]) ] ) example_theorist = LinearRegression() def get_example_synthetic_experiment_runner(): rng = np.random.default_rng(seed=180) def runner(x): return ground_truth(x) + rng.normal(0, 0.1, x.shape) def dataframe_runner(conditions_df: pd.DataFrame): observations_df = conditions_df.copy() observations_df["y"] = runner(conditions_df["x"]) return observations_df return dataframe_runner example_synthetic_experiment_runner = get_example_synthetic_experiment_runner() cycle_with_state_dep_properties = Cycle( variables=variables, theorist=example_theorist, experimentalist=unobserved_data_experimentalist, experiment_runner=example_synthetic_experiment_runner, params={ "experimentalist": { "exclude_conditions": {"excluded_conditions": "%experiment_data.conditions%"}, "custom_random_sampler": {"n": 1} } } )

Now we can run the cycler to generate conditions and run experiments. The first time round, we have the full set of 10 possible conditions to select from, and we select "2" at random:

In [ ]:

cycle_with_state_dep_properties.run().data.conditions[-1]

cycle_with_state_dep_properties.run().data.conditions[-1]

Out[ ]:

	x
0	2

We can continue to run the cycler, each time we add more to the list of "excluded" options:

In [ ]:

cycle_with_state_dep_properties.run(num_cycles=9).data.conditions[-1]

cycle_with_state_dep_properties.run(num_cycles=9).data.conditions[-1]

Out[ ]:

	x
0	1

If we try to evaluate it again, the experimentalist fails, as there aren't any more conditions which are available:

In [ ]:

cycle_with_state_dep_properties.run()

cycle_with_state_dep_properties.run()

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
Cell In[6], line 1
----> 1 cycle_with_state_dep_properties.run()

File ~/Developer/autora-workflow/src/autora/workflow/cycle.py:82, in Cycle.run(self, num_cycles)
     80 def run(self, num_cycles: int = 1):
     81     """Execute the next step in the cycle."""
---> 82     super().run(num_steps=num_cycles)
     83     return self

File ~/Developer/autora-workflow/src/autora/workflow/base.py:85, in BaseController.run(self, num_steps)
     83 """Run the next num_steps planned steps in the workflow."""
     84 for i in range(num_steps):
---> 85     self.run_once()
     86 return self

File ~/Developer/autora-workflow/src/autora/workflow/base.py:71, in BaseController.run_once(self, step_name)
     68 _logger.debug(f"{next_params=}")
     70 # Execute
---> 71 result = next_function(self.state, params=next_params)
     72 _logger.debug(f"{result=}")
     74 # Update

File ~/Developer/autora-workflow/src/autora/workflow/executor.py:174, in full_cycle_wrapper.<locals>._executor_full_cycle(state, params)
    169 experimentalist_params = params.get("experimentalist", {})
    171 experimentalist_executor = from_experimentalist_pipeline(
    172     experimentalist_pipeline
    173 )
--> 174 experimentalist_result = experimentalist_executor(state, experimentalist_params)
    176 experiment_runner_params = params.get("experiment_runner", {})
    177 experiment_runner_executor = from_experiment_runner_callable(
    178     experiment_runner_callable
    179 )

File ~/Developer/autora-workflow/src/autora/workflow/executor.py:64, in from_experimentalist_pipeline.<locals>._executor_experimentalist(state, params)
     62 def _executor_experimentalist(state: SupportsControllerState, params: Dict):
     63     params_ = resolve_state_params(params, state)
---> 64     new_conditions = pipeline(**params_)
     66     if isinstance(new_conditions, pd.DataFrame):
     67         new_conditions_array = new_conditions

File ~/Developer/autora-workflow/.venv/lib/python3.8/site-packages/autora/experimentalist/pipeline.py:171, in Pipeline.__call__(self, ex, **params)
    169     assert isinstance(pipe, Pipe)
    170     all_params_for_pipe = merged_params.get(name, dict())
--> 171     results.append(pipe(results[-1], **all_params_for_pipe))
    173 return results[-1]

Cell In[3], line 9, in custom_random_sampler(conditions, n)
      8 def custom_random_sampler(conditions, n):
----> 9     sampled_conditions = random_sampler_rng.choice(conditions, size=n, replace=False)
     10     return sampled_conditions

File _generator.pyx:729, in numpy.random._generator.Generator.choice()

ValueError: a cannot be empty unless no samples are taken