mopipe.core.data.experiment

Experiment structure classes.

LDType Objects

class LDType(StrEnum)

Type of data associated with a level.

ExperimentLevel Objects

class ExperimentLevel()

Base class for experiment structure classes.

This class can be used to drop in data that applies to specific levels of an experiment, e.g. experiment-wide data, group-level, condition-level, trial-level etc.

__init__

def __init__(level_name: str,
             level_id: t.Optional[str] = None,
             level_metadata: t.Optional[MetaData] = None) -> None

Initialize an ExperimentLevel.

level_name

@property
def level_name() -> str

Name of the level.

level_id

@property
def level_id() -> str

ID of the level.

parent

@property
def parent() -> t.Optional["ExperimentLevel"]

Parent level.

parent

@parent.setter
def parent(parent: "ExperimentLevel") -> None

Set the parent level.

child

@property
def child() -> t.Optional["ExperimentLevel"]

Child level.

child

@child.setter
def child(child: "ExperimentLevel") -> None

Set the child level.

depth

@property
def depth() -> int

Depth of the level.

leveldata

@property
def leveldata() -> list["EmpiricalData"]

Level data.

leveldata

@leveldata.setter
def leveldata(leveldata: t.Iterable["EmpiricalData"]) -> None

Set the level data.

add_leveldata

def add_leveldata(leveldata: "EmpiricalData") -> None

Add level data to the level.

timeseries

@property
def timeseries() -> list["EmpiricalData"]

Timeseries data.

timeseries

@timeseries.setter
def timeseries(timeseries: t.Iterable["EmpiricalData"]) -> None

Set the timeseries data.

add_timeseries

def add_timeseries(timeseries: "EmpiricalData") -> None

Add a timeseries to the level.

level_metadata

@property
def level_metadata() -> "MetaData"

Level metadata.

level_metadata

@level_metadata.setter
def level_metadata(level_metadata: "MetaData") -> None

Set the level metadata.

get_timeseries_by_name

def get_timeseries_by_name(name: str) -> t.Optional["EmpiricalData"]

Look up a timeseries by name.

Parameters

name : str The name of the timeseries to find.

Returns

EmpiricalData or None The first matching timeseries, or None if not found.

get_timeseries_by_id

def get_timeseries_by_id(data_id: str) -> t.Optional["EmpiricalData"]

Look up a timeseries by data ID.

Parameters

data_id : str The ID of the timeseries to find.

Returns

EmpiricalData or None The first matching timeseries, or None if not found.

run_pipeline

def run_pipeline(pipeline: "Pipeline",
                 data_name: t.Optional[str] = None,
                 result_name: t.Optional[str] = None,
                 *,
                 store_result: bool = True,
                 **kwargs) -> "EmpiricalData"

Run a pipeline on timeseries data at this level.

Parameters

pipeline : Pipeline The pipeline to run. data_name : str, optional Name of the timeseries to use as input. If None, uses the first available. result_name : str, optional Name for the result timeseries. Defaults to "{source_name}_processed". store_result : bool, optional Whether to store the result back on this level. Defaults to True. **kwargs Additional keyword arguments passed to pipeline.run().

Returns

EmpiricalData The result wrapped as a TimeseriesData.

Raises

ValueError If no timeseries data is available or the named timeseries is not found.

run_pipeline_on_descendants

def run_pipeline_on_descendants(pipeline: "Pipeline",
                                target_depth: t.Optional[int] = None,
                                data_name: t.Optional[str] = None,
                                result_name: t.Optional[str] = None,
                                **kwargs) -> list["EmpiricalData"]

Run a pipeline on descendant levels that have timeseries data.

Parameters

pipeline : Pipeline The pipeline to run. target_depth : int, optional Only run on levels at this depth. If None, runs on all descendants with data. data_name : str, optional Name of the timeseries to use as input on each level. result_name : str, optional Name for the result timeseries on each level. **kwargs Additional keyword arguments passed to pipeline.run().

Returns

list[EmpiricalData] List of result EmpiricalData from each level that was processed.

climb

def climb() -> t.Iterator["ExperimentLevel"]

Climb the experiment structure.

descend

def descend() -> t.Iterator["ExperimentLevel"]

Descend the experiment structure.

top

def top() -> "ExperimentLevel"

Get the top level.

bottom

def bottom() -> "ExperimentLevel"

Get the bottom level.

relevel_stack

def relevel_stack()

Relevel the experiment structure.

This method will relevel the experiment structure, so that the top level is at depth 0, and each level below it is at depth 1. If the top level is not an Experiment, then the depth of the top level will be 1.

relevel_children

def relevel_children() -> None

Relevel the children of the experiment structure.

Experiment Objects

class Experiment(ExperimentLevel)

__init__

def __init__(experiment_id: str) -> None

Initialize an Experiment.

parent

@property
def parent() -> ExperimentLevel | None

Parent level.

parent

@parent.setter
def parent(parent: t.Any) -> None

Set the parent level.

Trial Objects

class Trial(ExperimentLevel)

__init__

def __init__(trial_id: str) -> None

Initialize a Trial.

child

@property
def child() -> ExperimentLevel | None

Child level.

child

@child.setter
def child(child: t.Any) -> None

Set the child level.