ExperimentData

ExperimentData is the central data container that stores inputs, outputs, and metadata for all your experiments. ExperimentSample provides access to individual experiments.

See also

• Tutorial: Working with ExperimentData
• Core Concepts: ExperimentData

f3dasm.ExperimentData

Source code in src/f3dasm/_src/experimentdata.py

class ExperimentData:
    def __init__(
        self,
        domain: Optional[Domain] = None,
        input_data: Optional[
            pd.DataFrame | np.ndarray | list[dict[str, Any]] | str | Path
        ] = None,
        output_data: Optional[
            pd.DataFrame | np.ndarray | list[dict[str, Any]] | str | Path
        ] = None,
        jobs: Optional[pd.Series] = None,
        project_dir: Optional[Path] = None,
    ):
        """
        Main object to store implementations of a design-of-experiments, keep
        track of results, perform optimization and extract data for machine
        learning purposes.

        Parameters
        ----------
        domain : Domain, optional
            The domain of the experiment, by default None.
        input_data : pd.DataFrame | np.ndarray | List[Dict[str, Any]] |
                    str | Path, optional
            The input data of the experiment, by default None.
        output_data : pd.DataFrame | np.ndarray | List[Dict[str, Any]] |
                     str | Path, optional
            The output data of the experiment, by default None.
        jobs : pd.Series, optional
            The status of all the jobs, by default None.
        project_dir : Path, optional
            Directory of the project, by default None.

        Examples
        --------
        Three creation paths cover almost every workflow on the
        current API (post-#331):

        Empty container that a sampler block populates:

        >>> from f3dasm import ExperimentData, create_sampler
        >>> from f3dasm.design import Domain
        >>> domain = Domain()
        >>> domain.add_float("x", 0.0, 1.0)
        >>> data = ExperimentData(domain=domain)
        >>> data = create_sampler("random", seed=0).call(data, n_samples=8)

        From an in-memory numpy array, pandas DataFrame, list of dicts,
        or path to a CSV:

        >>> experiment_data = ExperimentData(
        ...     domain=domain_obj,
        ...     input_data=input_df,
        ...     output_data=output_df,
        ... )

        From a previously stored f3dasm project directory:

        >>> experiment_data = ExperimentData.from_file(
        ...     project_dir="./my_project"
        ... )
        """
        _domain = _domain_factory(domain)
        _project_dir = _project_dir_factory(project_dir)
        _jobs = jobs_factory(jobs)

        # If input_data is a numpy array, create pd.Dataframe to include column
        # names from the domain
        if isinstance(input_data, np.ndarray):
            input_data = convert_numpy_to_dataframe_with_domain(
                array=input_data, names=_domain.input_names, mode="input"
            )

        # Same with output data
        if isinstance(output_data, np.ndarray):
            output_data = convert_numpy_to_dataframe_with_domain(
                array=output_data, names=_domain.output_names, mode="output"
            )

        _input_data = _dict_factory(data=input_data)
        _output_data = _dict_factory(data=output_data)

        # If the domain is empty, try to infer it from the input_data
        # and output_data
        if not _domain:
            _domain = Domain.from_data(
                input_data=_input_data, output_data=_output_data
            )

        _data = data_factory(
            input_data=_input_data,
            output_data=_output_data,
            jobs=_jobs,
            project_dir=_project_dir,
        )

        self.data = _data
        self._domain = _domain
        self._project_dir = _project_dir

        # Store to_disk objects so that the references are kept only
        self.store_objects()

    def __len__(self):
        """
        Returns the number of experiments in the ExperimentData object.

        Returns
        -------
        int
            Number of experiments.

        Examples
        --------
        >>> experimentdata = ExperimentData(input_data=np.array([1, 2, 3]),)
        >>> len(experiment_data)
        3
        """
        return len(self.data)

    def __iter__(self) -> Iterator[tuple[int, ExperimentSample]]:
        """
        Returns an iterator over the ExperimentData object.

        Returns
        -------
        Iterator[Tuple[int, ExperimentSample]]
            Iterator over the ExperimentData object.

        Examples
        --------
        >>> for id, sample in experiment_data:
        ...     print(id, sample)
        0 ExperimentSample(...)
        """
        return iter(self.data.items())

    def __add__(self, __o: ExperimentData) -> ExperimentData:
        """
        Adds two ExperimentData objects.

        Parameters
        ----------
        __o : ExperimentData
            The other ExperimentData object to add.

        Returns
        -------
        ExperimentData
            The combined ExperimentData object.

        Examples
        --------
        >>> combined_data = experiment_data1 + experiment_data2
        """
        copy_self = copy(self).reset_index()
        copy_self._add(__o)
        return copy_self

    def __eq__(self, __o: ExperimentData) -> bool:
        """
        Checks if two ExperimentData objects are equal.

        Parameters
        ----------
        __o : ExperimentData
            The other ExperimentData object to compare.

        Returns
        -------
        bool
            True if the objects are equal, False otherwise.

        Notes
        -----
        Two ExperimentData objects are considered equal if their data, domain
        and project_dir are equal.

        Examples
        --------
        >>> experiment_data1 == experiment_data2
        True
        """
        return (
            self.data == __o.data
            and self._domain == __o._domain
            and self._project_dir == __o._project_dir
        )

    def __getitem__(self, key: int | Iterable[int]) -> ExperimentData:
        """
        Gets a subset of the ExperimentData object.

        Parameters
        ----------
        key : int or Iterable[int]
            The indices to select.

        Returns
        -------
        ExperimentData
            The selected subset of the ExperimentData object.

        """
        if isinstance(key, int):
            key = [key]

        return ExperimentData.from_data(
            data={k: self.data[k] for k in self.index[key]},
            domain=self._domain,
            project_dir=self._project_dir,
        )

    def _repr_html_(self) -> str:
        """
        Returns an HTML representation of the ExperimentData object.

        Returns
        -------
        str
            HTML representation of the ExperimentData object.

        Examples
        --------
        >>> experiment_data._repr_html_()
        '<div>...</div>'
        """
        return self.to_multiindex()._repr_html_()

    def __repr__(self) -> str:
        """
        Returns a string representation of the ExperimentData object.

        Returns
        -------
        str
            String representation of the ExperimentData object.

        Examples
        --------
        >>> repr(experiment_data)
        'ExperimentData(...)'
        """
        return self.to_multiindex().__repr__()

    def __deepcopy__(self) -> ExperimentData:
        """
        Returns a deep copy of the ExperimentData object.

        Returns
        -------
        ExperimentData
            Deep copy of the ExperimentData object.

        Examples
        --------
        >>> from copy import deepcopy
        >>> copied_data = deepcopy(experiment_data)
        """
        return self._copy(in_place=False, deep=True)

    def __copy__(self) -> ExperimentData:
        """
        Returns a shallow copy of the ExperimentData object.

        Returns
        -------
        ExperimentData
            Shallow copy of the ExperimentData object.

        Examples
        --------
        >>> from copy import copy
        >>> copied_data = copy(experiment_data)
        """
        return self._copy(in_place=False, deep=False)

    #                                                                Properties
    # =========================================================================

    @property
    def index(self) -> pd.Index:
        """
        Returns an iterable of the job number of the experiments.

        Returns
        -------
        pd.Index
            The job number of all the experiments in pandas Index format.

        Examples
        --------
        >>> experiment_data.index
        Int64Index([0, 1, 2], dtype='int64')
        """
        return pd.Index(self.data.keys())

    @property
    def jobs(self) -> pd.Series:
        """
        Returns the status of all the jobs.

        Returns
        -------
        pd.Series
            The status of all the jobs.

        Examples
        --------
        >>> experiment_data.jobs
        0    open
        1    finished
        dtype: object
        """
        return pd.Series({id: es.job_status.name for id, es in self})

    @property
    def domain(self) -> Domain:
        """
        Returns the domain of the ExperimentData object.

        Returns
        -------
        Domain
            The domain of the ExperimentData object.
        """
        return self._domain

    @domain.setter
    def domain(self, domain: Domain):
        """
        Sets the domain of the ExperimentData object.

        Parameters
        ----------
        domain : Domain
            The domain to set.
        """
        self._domain = domain

    @property
    def project_dir(self) -> Path:
        """
        Returns the project directory of the ExperimentData object.

        Returns
        -------
        Path
            The project directory.
        """
        return self._project_dir

    @project_dir.setter
    def project_dir(self, project_dir: Path | str):
        """
        Sets the project directory of the ExperimentData object.

        Parameters
        ----------
        project_dir : Path or str
            The project directory to set.
        """
        self._project_dir = _project_dir_factory(project_dir)
        for _, es in self:
            es.project_dir = self._project_dir

    #                                                  Alternative constructors
    # =========================================================================

    @classmethod
    def _from_attributes(
        cls: type[ExperimentData],
        domain: Domain,
        data: dict[int, ExperimentSample],
        project_dir: Path,
    ) -> ExperimentData:
        """
        Create an ExperimentData object from attributes.

        Parameters
        ----------
        domain : Domain
            The domain of the data.
        data : dict of int to ExperimentSample
            The data of the experiment.
        project_dir : Path
            The project directory.

        Returns
        -------
        ExperimentData
            ExperimentData object containing the loaded data.
        """
        experiment_data = cls()
        experiment_data.data = data
        experiment_data._domain = domain
        experiment_data._project_dir = project_dir
        return experiment_data

    @classmethod
    def from_file(
        cls: type[ExperimentData],
        project_dir: Path | str,
        wait_for_creation: bool = False,
        max_tries: int = MAX_TRIES,
    ) -> ExperimentData:
        """
        Create an ExperimentData object from .csv and .json files.

        Parameters
        ----------
        project_dir : Path or str
            User defined path of the experimentdata directory.
        wait_for_creation : bool, optional
            If True, wait for files to be created if not found,
            by default False.
        max_tries : int, optional
            Maximum number of attempts to read the files,
            by default MAX_TRIES.

        Returns
        -------
        ExperimentData
            ExperimentData object containing the loaded data.

        Examples
        --------
        >>> experiment_data = ExperimentData.from_file('path/to/project_dir')
        """
        if isinstance(project_dir, str):
            project_dir = Path(project_dir)

        try:
            return _from_file_attempt(
                project_dir=project_dir,
                wait_for_creation=wait_for_creation,
                max_tries=max_tries,
            )
        except FileNotFoundError:
            try:
                filename_with_path = Path(get_original_cwd()) / project_dir
            except ValueError as exc:  # get_original_cwd() error
                raise FileNotFoundError(
                    f"Cannot find the folder {project_dir} !"
                ) from exc

            return _from_file_attempt(
                project_dir=filename_with_path,
                wait_for_creation=wait_for_creation,
                max_tries=max_tries,
            )

    @classmethod
    def from_yaml(cls, config: DictConfig) -> ExperimentData:
        """
        Create an ExperimentData object from a YAML configuration.

        Parameters
        ----------
        config : DictConfig
            Hydra DictConfig object containing the configuration.

        Returns
        -------
        ExperimentData
            ExperimentData object containing the loaded data.

        Examples
        --------
        >>> experiment_data = ExperimentData.from_yaml(config)
        """
        # Option 1: From existing ExperimentData files
        if "from_file" in config:
            return cls.from_file(config.from_file)

        else:
            return cls(**config)

    @classmethod
    def from_data(
        cls,
        data: Optional[dict[int, ExperimentSample]] = None,
        domain: Optional[Domain] = None,
        project_dir: Optional[Path] = None,
    ) -> ExperimentData:
        """
        Create an ExperimentData object from existing data.

        Parameters
        ----------
        data : dict of int to ExperimentSample, optional
            The existing data, by default None.
        domain : Domain, optional
            The domain of the data, by default None.
        project_dir : Path, optional
            The project directory, by default None.

        Returns
        -------
        ExperimentData
            ExperimentData object containing the loaded data.

        Examples
        --------
        >>> experiment_data = ExperimentData.from_data(data, domain)
        """
        if data is None:
            data = {}

        if domain is None:
            domain = Domain()

        experiment_data = cls()

        experiment_data.data = defaultdict(ExperimentSample, data)
        experiment_data._domain = domain
        experiment_data._project_dir = _project_dir_factory(project_dir)
        return experiment_data

    #                                                         Selecting subsets
    # =========================================================================

    def select(self, indices: int | Iterable[int]) -> ExperimentData:
        """
        Select a subset of the ExperimentData object.

        Parameters
        ----------
        indices : int or Iterable[int]
            The indices to select.

        Returns
        -------
        ExperimentData
            The selected subset of the ExperimentData object.

        Examples
        --------
        >>> subset = experiment_data.select([0, 1, 2])
        """
        return self[indices]

    def select_parameter(self, name: str) -> ExperimentData:
        """Return an ExperimentData containing only the named parameter.

        Parameters
        ----------
        name : str
            Name of the input or output parameter to select.

        Returns
        -------
        ExperimentData
            New ExperimentData with a single-parameter domain and only the
            data for that parameter.

        Raises
        ------
        KeyError
            If ``name`` is not found in either input or output space.

        Examples
        --------
        >>> exp_model = experiment_data.select_parameter('model')
        """
        if name in self._domain.input_space:
            new_domain = Domain(
                input_space={name: self._domain.input_space[name]}
            )
            new_data = {
                idx: ExperimentSample(
                    _input_data={name: es._input_data.get(name)},
                    _output_data={},
                    job_status=es.job_status,
                    project_dir=es.project_dir,
                )
                for idx, es in self
            }
        elif name in self._domain.output_space:
            new_domain = Domain(
                output_space={name: self._domain.output_space[name]}
            )
            new_data = {
                idx: ExperimentSample(
                    _input_data={},
                    _output_data={name: es._output_data.get(name)},
                    job_status=es.job_status,
                    project_dir=es.project_dir,
                )
                for idx, es in self
            }
        else:
            raise KeyError(f"Parameter '{name}' not found in domain.")

        return ExperimentData.from_data(
            data=new_data,
            domain=new_domain,
            project_dir=self._project_dir,
        )

    def move_to_output(self, name: str, in_place: bool = False):
        """Move a parameter from the input space to the output space.

        The parameter entry is removed from the domain's input space and added
        to the output space. For every experiment sample, the corresponding
        value is moved from ``_input_data`` to ``_output_data``.

        Parameters
        ----------
        name : str
            Name of the input parameter to move.
        in_place : bool, optional
            If True, the operation is performed in place and None is returned,
            by default False.

        Returns
        -------
        ExperimentData or None
            A new ExperimentData with the parameter moved, or None if
            ``in_place=True``.

        Raises
        ------
        KeyError
            If ``name`` is not found in the input space.

        Examples
        --------
        >>> new_data = experiment_data.move_to_output('x0')
        >>> experiment_data.move_to_output('x0', in_place=True)
        """
        d = self._copy(in_place=in_place)
        if name not in d._domain.input_space:
            raise KeyError(f"Parameter '{name}' not found in input space.")
        d._domain.output_space[name] = d._domain.input_space.pop(name)
        for _, es in d:
            if name in es._input_data:
                es._output_data[name] = es._input_data.pop(name)
        if in_place:
            return None
        return d

    def move_to_input(self, name: str, in_place: bool = False):
        """Move a parameter from the output space to the input space.

        The parameter entry is removed from the domain's output space and added
        to the input space. For every experiment sample, the corresponding
        value is moved from ``_output_data`` to ``_input_data``.

        Parameters
        ----------
        name : str
            Name of the output parameter to move.
        in_place : bool, optional
            If True, the operation is performed in place and None is returned,
            by default False.

        Returns
        -------
        ExperimentData or None
            A new ExperimentData with the parameter moved, or None if
            ``in_place=True``.

        Raises
        ------
        KeyError
            If ``name`` is not found in the output space.

        Examples
        --------
        >>> new_data = experiment_data.move_to_input('y')
        >>> experiment_data.move_to_input('y', in_place=True)
        """
        d = self._copy(in_place=in_place)
        if name not in d._domain.output_space:
            raise KeyError(f"Parameter '{name}' not found in output space.")
        d._domain.input_space[name] = d._domain.output_space.pop(name)
        for _, es in d:
            if name in es._output_data:
                es._input_data[name] = es._output_data.pop(name)
        if in_place:
            return None
        return d

    def select_with_status(
        self, status: Literal["open", "in_progress", "finished", "error"]
    ) -> ExperimentData:
        """
        Select a subset of the ExperimentData object with a given status.

        Parameters
        ----------
        status : {'open', 'in_progress', 'finished', 'error'}
            The status to select.

        Returns
        -------
        ExperimentData
            The selected subset of the ExperimentData object with the given
            status.

        Examples
        --------
        >>> subset = experiment_data.select_with_status('finished')
        """
        idx = [i for i, es in self if es.is_status(status)]
        return self[idx]

    #                                                                    Export
    # =========================================================================

    def _copy(
        self, in_place: bool = False, deep: bool = True
    ) -> ExperimentData:
        """
        Create a copy of the ExperimentData object.

        Parameters
        ----------
        in_place : bool, optional
            If True, no copy is made and the object itself is returned,
            by default False.
        deep : bool, optional
            If True, a deep copy is made, by default True

        Returns
        -------
        ExperimentData
            A copy of the ExperimentData object or the original object

        Examples
        --------
        >>> copied_data = experiment_data._copy(in_place=False)
        """
        if in_place:
            return self

        if deep:
            data_copy = {k: v._copy() for k, v in self.data.items()}
        else:
            data_copy = self.data

        return ExperimentData._from_attributes(
            data=defaultdict(ExperimentSample, data_copy),
            domain=self._domain._copy(),
            project_dir=self._project_dir,
        )

    def store(
        self,
        project_dir: Optional[Path | str] = None,
        copy_references: bool = False,
    ):
        """
        Write the ExperimentData to disk in the project directory.

        Parameters
        ----------
        project_dir : Optional[Path | str], optional
            The f3dasm project directory to store the
            ExperimentData object to, by default None.
        copy_references : bool, optional
            If True, any :class:`ReferenceValue` objects whose source
            ``project_dir`` differs from the destination are physically
            copied into the destination project directory before writing.
            The in-memory references are updated to point to the new
            locations, by default False.

        Note
        ----
        If no project directory is provided, the ExperimentData object is
        stored in the directory provided by the `.project_dir` attribute that
        is set upon creation of the object.

        The ExperimentData object is stored in a subfolder 'experiment_data'.

        The ExperimentData object is stored in four files:

        * the input data (`input.csv`)
        * the output data (`output.csv`)
        * the jobs (`jobs.csv`)
        * the domain (`domain.json`)

        To avoid the ExperimentData to be written simultaneously by multiple
        processes, a '.lock' file is automatically created
        in the project directory. Concurrent process can only sequentially
        access the lock file. This lock file is removed after the
        ExperimentData object is written to disk.
        """
        old_project_dir = self._project_dir

        if project_dir is not None:
            self.set_project_dir(project_dir, in_place=True)

        if copy_references:
            seen: set[Path] = set()
            for _, es in self:
                for value in list(es._input_data.values()) + list(
                    es._output_data.values()
                ):
                    if (
                        isinstance(value, ReferenceValue)
                        and value.reference not in seen
                    ):
                        seen.add(value.reference)
                        _copy_reference(
                            value.reference, old_project_dir, self._project_dir
                        )

        subdirectory = self._project_dir / EXPERIMENTDATA_SUBFOLDER

        # Create the experimentdata subfolder if it does not exist
        subdirectory.mkdir(parents=True, exist_ok=True)

        # # Store all objects to keep references
        # self.store_objects()

        df_input, df_output = self.to_pandas(keep_references=True)

        df_input.to_csv(
            (subdirectory / INPUT_DATA_FILENAME).with_suffix(".csv")
        )
        df_output.to_csv(
            (subdirectory / OUTPUT_DATA_FILENAME).with_suffix(".csv")
        )
        self._domain.store(subdirectory / DOMAIN_FILENAME)
        self.jobs.to_csv((subdirectory / JOBS_FILENAME).with_suffix(".csv"))

    def to_numpy(self) -> tuple[np.ndarray, np.ndarray]:
        """
        Convert the ExperimentData object to a tuple of numpy arrays.

        Returns
        -------
        tuple of np.ndarray
            A tuple containing two numpy arrays, the first one for input
            columns, and the second for output columns.

        Examples
        --------
        >>> input_array, output_array = experiment_data.to_numpy()
        """
        df_input, df_output = self.to_pandas(keep_references=False)
        return df_input.to_numpy(), df_output.to_numpy()

    def to_pandas(
        self, keep_references: bool = False
    ) -> tuple[pd.DataFrame, pd.DataFrame]:
        """
        Convert the ExperimentData object to pandas DataFrames.

        Parameters
        ----------
        keep_references : bool, optional
            If True, the references to the output data are kept, by default
            False.

        Returns
        -------
        tuple of pd.DataFrame
            A tuple containing two pandas DataFrames, the first one for input
            columns, and the second for output columns.

        Examples
        --------
        >>> df_input, df_output = experiment_data.to_pandas()
        """
        if keep_references:
            return (
                pd.DataFrame(
                    [es._input_data for _, es in self], index=self.index
                ),
                pd.DataFrame(
                    [es._output_data for _, es in self], index=self.index
                ),
            )
        else:
            return (
                pd.DataFrame(
                    [es.input_data for _, es in self], index=self.index
                ),
                pd.DataFrame(
                    [es.output_data for _, es in self], index=self.index
                ),
            )

    def to_xarray(self, keep_references: bool = False) -> xr.Dataset:
        """
        Convert the ExperimentData object to an xarray Dataset.

        Parameters
        ----------
        keep_references : bool, optional
            If True, the references to the output data are kept, by default
            False.

        Returns
        -------
        xr.Dataset
            An xarray Dataset containing the data.

        Examples
        --------
        >>> dataset = experiment_data.to_xarray()
        """
        df_input, df_output = self.to_pandas(keep_references=keep_references)

        da_input = xr.DataArray(
            df_input,
            dims=["iterations", "input_dim"],
            coords={"iterations": self.index, "input_dim": df_input.columns},
        )

        da_output = xr.DataArray(
            df_output,
            dims=["iterations", "output_dim"],
            coords={"iterations": self.index, "output_dim": df_output.columns},
        )

        return xr.Dataset({"input": da_input, "output": da_output})

    def get_n_best_output(
        self, n_samples: int, output_name: Optional[str] = "y"
    ) -> ExperimentData:
        """
        Get the n best samples from the output data. Lower values are better.

        Parameters
        ----------
        n_samples : int
            Number of samples to select.
        output_name : str, optional
            The name of the output column to sort by, by default 'y'.

        Returns
        -------
        ExperimentData
            New ExperimentData object with a selection of the n best samples.

        Examples
        --------
        >>> best_samples = experiment_data.get_n_best_output(5)
        """
        _, df_out = self.to_pandas()
        indices = df_out.nsmallest(n=n_samples, columns=output_name).index
        return self[indices]

    def to_multiindex(self) -> pd.DataFrame:
        """
        Convert the ExperimentData object to a pandas DataFrame with a
        MultiIndex. This is used for visualization purposes in a Jupyter
        notebook environment.

        Returns
        -------
        pd.DataFrame
            A pandas DataFrame with a MultiIndex.

        Examples
        --------
        >>> df_multiindex = experiment_data.to_multiindex()
        """
        list_of_dicts = [sample.to_multiindex() for _, sample in self]
        return pd.DataFrame(merge_dicts(list_of_dicts), index=self.index)

    #                                                     Append or remove data
    # =========================================================================

    def add_experiments(
        self,
        data: ExperimentSample | ExperimentData,
        in_place: bool = False,
    ) -> None:
        """
        Add an ExperimentSample or ExperimentData to the ExperimentData
        attribute.

        Parameters
        ----------
        data : ExperimentSample or ExperimentData
            Experiment(s) to add.
        in_place : bool, optional
            If True, the data is added in place, by default False.

        Raises
        ------
        ValueError
            If the input is not an ExperimentSample or ExperimentData object.

        Examples
        --------
        >>> experiment_data.add_experiments(new_sample)
        >>> experiment_data.add_experiments(new_data)
        """
        d = self._copy(in_place=in_place)

        if isinstance(data, ExperimentSample):
            d._add_experiment_sample(data)

        elif isinstance(data, ExperimentData):
            d._add(data)

        else:
            raise ValueError(
                f"The input to this function should be an ExperimentSample or "
                f"ExperimentData object, not {type(data)} "
            )

        if in_place:
            return None
        else:
            return d

    def remove_rows_bottom(self, number_of_rows: int, in_place: bool = False):
        """
        Remove a number of rows from the end of the ExperimentData object.

        Parameters
        ----------
        number_of_rows : int
            Number of rows to remove from the bottom.
        in_place : bool, optional
            If True, the rows are removed in place, by default False.

        Examples
        --------
        >>> experiment_data.remove_rows_bottom(3)
        """
        d = self._copy(in_place=in_place)

        # remove the last n rows
        for _i in range(number_of_rows):
            d.data.pop(d.index[-1])

        if in_place:
            return None
        else:
            return d

    def reset_index(self) -> ExperimentData:
        """
        Reset the index of the ExperimentData object.
        The index will be reset to a range from 0 to the number of experiments.

        Returns
        -------
        ExperimentData
            ExperimentData object with a reset index.

        Examples
        --------
        >>> reset_data = experiment_data.reset_index()
        """
        return ExperimentData.from_data(
            data={i: v for i, v in enumerate(self.data.values())},
            domain=self._domain,
            project_dir=self._project_dir,
        )

    def join(self, experiment_data: ExperimentData) -> ExperimentData:
        """
        Join two ExperimentData objects.

        Parameters
        ----------
        experiment_data : ExperimentData
            The other ExperimentData object to join with.

        Returns
        -------
        ExperimentData
            The joined ExperimentData object.

        Examples
        --------
        >>> joined_data = experiment_data1.join(experiment_data2)
        """
        copy_self = self.reset_index()
        # TODO: Reset isnt necessary, only copy
        copy_other = experiment_data.reset_index()

        if not copy_self.data:
            copy_other._domain += copy_self._domain
            return copy_other

        for (i, es_self), (_, es_other) in zip(
            copy_self, copy_other, strict=False
        ):
            copy_self.data[i] = es_self + es_other

        copy_self._domain += copy_other._domain

        return copy_self

    def _add(self, experiment_data: ExperimentData):
        # copy and reset self
        copy_other = experiment_data.reset_index()

        # Find the last key in my_dict
        last_key = max(self.index) if self else -1

        # Update keys of other dict
        other_updated_data = {
            last_key + 1 + i: v for i, v in enumerate(copy_other.data.values())
        }

        self.data.update(other_updated_data)
        self._domain += copy_other._domain

    def _add_experiment_sample(self, experiment_sample: ExperimentSample):
        last_key = max(self.index) if self else -1
        self.data[last_key + 1] = experiment_sample

    def replace_nan(self, value: Any, in_place: bool = False):
        """
        Replace all NaN values in the output data with the given value.

        Parameters
        ----------
        value : Any
            The value to replace NaNs with.
        in_place : bool, optional
            If True, the NaN values are replaced in place, by default False.

        Examples
        --------
        >>> experiment_data.replace_nan(0)
        """
        d = self._copy(in_place=in_place)
        for _, es in d:
            es.replace_nan(value)

        if in_place:
            return None
        else:
            return d

    def round(self, decimals: int, in_place: bool = False):
        """
        Round all output data to the given number of decimals.

        Parameters
        ----------
        decimals : int
            Number of decimals to round to.
        in_place : bool, optional
            If True, round values in place, by default False.

        Examples
        --------
        >>> experiment_data.round(2)
        """
        d = self._copy(in_place=in_place)

        for _, es in d:
            es.round(decimals)

        if in_place:
            return None
        else:
            return d

    # TODO: Create tests for this
    def sort(
        self,
        criterion: Callable[[ExperimentSample], Any],
        reverse: bool = False,
    ) -> ExperimentData:
        """
        Sort the ExperimentData object based on a criterion.

        Parameters
        ----------
        criterion : Callable[[ExperimentSample], Any]
            The criterion to sort on. This should be a function that takes an
            ExperimentSample object and returns a value to sort on.
        reverse : bool, optional
            If True, sort in descending order, by default False.

        Returns
        -------
        ExperimentData
            The sorted ExperimentData object.

        Examples
        --------
        >>> sorted_data = experiment_data.sort(lambda x: x.output_data['y'])
        """

        sorted_data = dict(
            sorted(
                self.data.items(),
                key=lambda item: criterion(item[1]),
                reverse=reverse,
            )
        )
        return ExperimentData.from_data(
            data=sorted_data,
            domain=self._domain,
            project_dir=self._project_dir,
        )

    #                                                          ExperimentSample
    # =========================================================================

    def get_experiment_sample(self, id: int) -> ExperimentSample:
        """
        Gets the experiment_sample at the given index.

        Parameters
        ----------
        id : int
            The index of the experiment_sample to retrieve.

        Returns
        -------
        ExperimentSample
            The ExperimentSample at the given index.

        Examples
        --------
        >>> sample = experiment_data.get_experiment_sample(0)
        """
        return self.data[id]

    def store_experimentsample(
        self,
        experiment_sample: ExperimentSample,
        idx: int,
        domain: Domain | None = None,
    ):
        """
        Store an ExperimentSample object in the ExperimentData object and
        update the Domain object.

        Parameters
        ----------
        experiment_sample : ExperimentSample
            The ExperimentSample object to store.
        idx : int
            The index of the ExperimentSample object.

        Examples
        --------
        >>> experiment_data.store_experimentsample(sample, 0)
        """
        experiment_sample, domain = _store(
            experiment_sample=experiment_sample,
            idx=idx,
            domain=domain if domain is not None else self.domain,
        )

        self._domain = domain
        self.data[idx] = experiment_sample

    def store_objects(self):
        self.store_experimentsample_references()
        # Store to_disk objects so that the references are kept only
        for idx, experiment_sample in self:
            self.store_experimentsample(
                experiment_sample=experiment_sample,
                idx=idx,
                domain=self.domain,
            )

    def store_experimentsample_references(self):
        """
        Store references to input and output data in the experiment sample
        based on the domain.

        Notes
        -----
        This method checks the domain for parameters that should be stored
        on disk. If a parameter is marked to be stored on disk, the method
        will store the corresponding value in the experiment sample using
        the `store` method.

        Examples
        --------
        >>> domain = Domain()
        >>> domain.add_float(name='param1', to_disk=True)
        >>> sample = ExperimentSample(
        ...     _input_data={'param1': 1.0, 'param2': 2.0},
        ...     _output_data={'result1': 3.0}
        ... )
        >>> sample.store_experimentsample_references()
        >>> isinstance(sample._input_data['param1'], ToDiskValue)
        True
        """
        for _, es in self:
            for name, value in es._input_data.items():
                input_parameter = self.domain.input_space.get(name, None)
                if input_parameter is not None and input_parameter.to_disk:
                    es.store(
                        name=name,
                        object=value,
                        to_disk=True,
                        store_function=input_parameter.store_function,
                        load_function=input_parameter.load_function,
                        which="input",
                    )

            for name, value in es._output_data.items():
                output_parameter = self.domain.output_space.get(name, None)
                if output_parameter is not None and output_parameter.to_disk:
                    es.store(
                        name=name,
                        object=value,
                        to_disk=True,
                        store_function=output_parameter.store_function,
                        load_function=output_parameter.load_function,
                        which="output",
                    )

    def update_from_experimentssample_json(self, in_place: bool = False):
        """
        Update the ExperimentData from ExperimentSample JSON files.

        Parameters
        ----------
        in_place : bool, optional
            If True, update in place, by default False.

        Returns
        -------
        ExperimentData or None
            Updated ExperimentData object, or None if in_place is True.

        Notes
        -----
        This method loads ExperimentSample objects from JSON files in
        the EXPERIMENTSAMPLE_SUBFOLDER directory. If loading fails for
        any file, a warning is logged and the process continues.
        """
        d = self._copy(in_place=in_place)

        for json_file in (d.project_dir / EXPERIMENTSAMPLE_SUBFOLDER).glob(
            "*.json"
        ):
            try:
                idx = int(json_file.stem)
                es = ExperimentSample.from_json(json_file)
                d.data[idx] = es

            except Exception as exc:
                logger.warning(
                    f"Could not load ExperimentSample from {json_file}: {exc}"
                )

        if in_place:
            return None
        else:
            return d

    def get_open_job(self) -> tuple[int, ExperimentSample, Domain]:
        """
        Get the first open job in the ExperimentData object.

        Returns
        -------
        tuple of int, ExperimentSample and the Domain
            The index, ExperimentSample and Domain of the first open job.

        Notes
        -----
        This function iterates over the ExperimentData object and returns the
        first open job. If no open jobs are found, it returns None.

        The returned open job is marked as 'in_progress'.

        Examples
        --------
        >>> job_id, job_sample = experiment_data.get_open_job()
        """
        for id, es in self:
            if es.is_status("open"):
                es.mark("in_progress")
                return id, es, self.domain

        return None, ExperimentSample(), self.domain

    #                                                                      Jobs
    # =========================================================================

    def is_all_finished(self) -> bool:
        """
        Check if all jobs are finished.

        Returns
        -------
        bool
            True if all jobs are finished, False otherwise.

        Examples
        --------
        >>> experiment_data.is_all_finished()
        True
        """
        return all(es.is_status("finished") for _, es in self)

    def mark(
        self,
        indices: int | Iterable[int],
        status: Literal["open", "in_progress", "finished", "error"],
        in_place: bool = False,
    ):
        """
        Mark the jobs at the given indices with the given status.

        Parameters
        ----------
        indices : int or Iterable[int]
            Indices of the jobs to mark.
        status : {'open', 'in_progress', 'finished', 'error'}
            Status to mark the jobs with.

        Raises
        ------
        ValueError
            If the given status is not valid.

        Examples
        --------
        >>> experiment_data.mark([0, 1], 'finished')
        """
        d = self._copy(in_place=in_place)

        if isinstance(indices, int):
            indices = [indices]
        for i in indices:
            d.data[i].mark(status)

        if in_place:
            return None
        else:
            return d

    def mark_all(
        self,
        status: Literal["open", "in_progress", "finished", "error"],
        in_place: bool = False,
    ):
        """
        Mark all the experiments with the given status.

        Parameters
        ----------
        status : {'open', 'in_progress', 'finished', 'error'}
            Status to mark the jobs with.

        Raises
        ------
        ValueError
            If the given status is not valid.

        Examples
        --------
        >>> experiment_data.mark_all('finished')
        """
        d = self._copy(in_place=in_place)

        for _, es in d:
            es.mark(status)

        if in_place:
            return None
        else:
            return d

    #                                                         Project directory
    # =========================================================================

    def set_project_dir(
        self, project_dir: Path | str, in_place: bool = False
    ) -> ExperimentData:
        """Set the directory of the f3dasm project folder.

        Parameters
        ----------
        project_dir : Path or str
            Path to the project directory
        in_place : bool, optional
            If True, the project directory is set in place, by default False

        Returns
        -------
        ExperimentData
            ExperimentData object with the updated project directory
        """
        d = self._copy(in_place=in_place)
        d._project_dir = _project_dir_factory(project_dir)

        if in_place:
            return None
        else:
            return d

domain property

Returns the domain of the ExperimentData object.

Returns:

Type	Description
Domain	The domain of the ExperimentData object.

index property

Returns an iterable of the job number of the experiments.

Returns:

Type	Description
Index	The job number of all the experiments in pandas Index format.

Examples:

>>> experiment_data.index
Int64Index([0, 1, 2], dtype='int64')

jobs property

Returns the status of all the jobs.

Returns:

Type	Description
Series	The status of all the jobs.

Examples:

>>> experiment_data.jobs
0    open
1    finished
dtype: object

project_dir property

Returns the project directory of the ExperimentData object.

Returns:

Type	Description
Path	The project directory.

_add(experiment_data: ExperimentData)

Source code in src/f3dasm/_src/experimentdata.py

def _add(self, experiment_data: ExperimentData):
    # copy and reset self
    copy_other = experiment_data.reset_index()

    # Find the last key in my_dict
    last_key = max(self.index) if self else -1

    # Update keys of other dict
    other_updated_data = {
        last_key + 1 + i: v for i, v in enumerate(copy_other.data.values())
    }

    self.data.update(other_updated_data)
    self._domain += copy_other._domain

_add_experiment_sample(experiment_sample: ExperimentSample)

Source code in src/f3dasm/_src/experimentdata.py

def _add_experiment_sample(self, experiment_sample: ExperimentSample):
    last_key = max(self.index) if self else -1
    self.data[last_key + 1] = experiment_sample

_copy(in_place: bool = False, deep: bool = True) -> ExperimentData

Create a copy of the ExperimentData object.

Parameters:

Name	Type	Description	Default
in_place	bool	If True, no copy is made and the object itself is returned, by default False.	False
deep	bool	If True, a deep copy is made, by default True	True

Returns:

Type	Description
ExperimentData	A copy of the ExperimentData object or the original object

Examples:

>>> copied_data = experiment_data._copy(in_place=False)

Source code in src/f3dasm/_src/experimentdata.py

def _copy(
    self, in_place: bool = False, deep: bool = True
) -> ExperimentData:
    """
    Create a copy of the ExperimentData object.

    Parameters
    ----------
    in_place : bool, optional
        If True, no copy is made and the object itself is returned,
        by default False.
    deep : bool, optional
        If True, a deep copy is made, by default True

    Returns
    -------
    ExperimentData
        A copy of the ExperimentData object or the original object

    Examples
    --------
    >>> copied_data = experiment_data._copy(in_place=False)
    """
    if in_place:
        return self

    if deep:
        data_copy = {k: v._copy() for k, v in self.data.items()}
    else:
        data_copy = self.data

    return ExperimentData._from_attributes(
        data=defaultdict(ExperimentSample, data_copy),
        domain=self._domain._copy(),
        project_dir=self._project_dir,
    )

_from_attributes(domain: Domain, data: dict[int, ExperimentSample], project_dir: Path) -> ExperimentData classmethod

Create an ExperimentData object from attributes.

Parameters:

Name	Type	Description	Default
domain	Domain	The domain of the data.	required
data	dict of int to ExperimentSample	The data of the experiment.	required
project_dir	Path	The project directory.	required

Returns:

Type	Description
ExperimentData	ExperimentData object containing the loaded data.

Source code in src/f3dasm/_src/experimentdata.py

@classmethod
def _from_attributes(
    cls: type[ExperimentData],
    domain: Domain,
    data: dict[int, ExperimentSample],
    project_dir: Path,
) -> ExperimentData:
    """
    Create an ExperimentData object from attributes.

    Parameters
    ----------
    domain : Domain
        The domain of the data.
    data : dict of int to ExperimentSample
        The data of the experiment.
    project_dir : Path
        The project directory.

    Returns
    -------
    ExperimentData
        ExperimentData object containing the loaded data.
    """
    experiment_data = cls()
    experiment_data.data = data
    experiment_data._domain = domain
    experiment_data._project_dir = project_dir
    return experiment_data

_repr_html_() -> str

Returns an HTML representation of the ExperimentData object.

Returns:

Type	Description
str	HTML representation of the ExperimentData object.

Examples:

>>> experiment_data._repr_html_()
'<div>...</div>'

Source code in src/f3dasm/_src/experimentdata.py

def _repr_html_(self) -> str:
    """
    Returns an HTML representation of the ExperimentData object.

    Returns
    -------
    str
        HTML representation of the ExperimentData object.

    Examples
    --------
    >>> experiment_data._repr_html_()
    '<div>...</div>'
    """
    return self.to_multiindex()._repr_html_()

add_experiments(data: ExperimentSample | ExperimentData, in_place: bool = False) -> None

Add an ExperimentSample or ExperimentData to the ExperimentData attribute.

Parameters:

Name	Type	Description	Default
data	ExperimentSample or ExperimentData	Experiment(s) to add.	required
in_place	bool	If True, the data is added in place, by default False.	False

Raises:

Type	Description
ValueError	If the input is not an ExperimentSample or ExperimentData object.

Examples:

>>> experiment_data.add_experiments(new_sample)
>>> experiment_data.add_experiments(new_data)

Source code in src/f3dasm/_src/experimentdata.py

def add_experiments(
    self,
    data: ExperimentSample | ExperimentData,
    in_place: bool = False,
) -> None:
    """
    Add an ExperimentSample or ExperimentData to the ExperimentData
    attribute.

    Parameters
    ----------
    data : ExperimentSample or ExperimentData
        Experiment(s) to add.
    in_place : bool, optional
        If True, the data is added in place, by default False.

    Raises
    ------
    ValueError
        If the input is not an ExperimentSample or ExperimentData object.

    Examples
    --------
    >>> experiment_data.add_experiments(new_sample)
    >>> experiment_data.add_experiments(new_data)
    """
    d = self._copy(in_place=in_place)

    if isinstance(data, ExperimentSample):
        d._add_experiment_sample(data)

    elif isinstance(data, ExperimentData):
        d._add(data)

    else:
        raise ValueError(
            f"The input to this function should be an ExperimentSample or "
            f"ExperimentData object, not {type(data)} "
        )

    if in_place:
        return None
    else:
        return d

from_data(data: Optional[dict[int, ExperimentSample]] = None, domain: Optional[Domain] = None, project_dir: Optional[Path] = None) -> ExperimentData classmethod

Create an ExperimentData object from existing data.

Parameters:

Name	Type	Description	Default
data	dict of int to ExperimentSample	The existing data, by default None.	None
domain	Domain	The domain of the data, by default None.	None
project_dir	Path	The project directory, by default None.	None

Returns:

Type	Description
ExperimentData	ExperimentData object containing the loaded data.

Examples:

>>> experiment_data = ExperimentData.from_data(data, domain)

Source code in src/f3dasm/_src/experimentdata.py

@classmethod
def from_data(
    cls,
    data: Optional[dict[int, ExperimentSample]] = None,
    domain: Optional[Domain] = None,
    project_dir: Optional[Path] = None,
) -> ExperimentData:
    """
    Create an ExperimentData object from existing data.

    Parameters
    ----------
    data : dict of int to ExperimentSample, optional
        The existing data, by default None.
    domain : Domain, optional
        The domain of the data, by default None.
    project_dir : Path, optional
        The project directory, by default None.

    Returns
    -------
    ExperimentData
        ExperimentData object containing the loaded data.

    Examples
    --------
    >>> experiment_data = ExperimentData.from_data(data, domain)
    """
    if data is None:
        data = {}

    if domain is None:
        domain = Domain()

    experiment_data = cls()

    experiment_data.data = defaultdict(ExperimentSample, data)
    experiment_data._domain = domain
    experiment_data._project_dir = _project_dir_factory(project_dir)
    return experiment_data

from_file(project_dir: Path | str, wait_for_creation: bool = False, max_tries: int = 20) -> ExperimentData classmethod

Create an ExperimentData object from .csv and .json files.

Parameters:

Name	Type	Description	Default
project_dir	Path or str	User defined path of the experimentdata directory.	required
wait_for_creation	bool	If True, wait for files to be created if not found, by default False.	False
max_tries	int	Maximum number of attempts to read the files, by default MAX_TRIES.	20

Returns:

Type	Description
ExperimentData	ExperimentData object containing the loaded data.

Examples:

>>> experiment_data = ExperimentData.from_file('path/to/project_dir')

Source code in src/f3dasm/_src/experimentdata.py

@classmethod
def from_file(
    cls: type[ExperimentData],
    project_dir: Path | str,
    wait_for_creation: bool = False,
    max_tries: int = MAX_TRIES,
) -> ExperimentData:
    """
    Create an ExperimentData object from .csv and .json files.

    Parameters
    ----------
    project_dir : Path or str
        User defined path of the experimentdata directory.
    wait_for_creation : bool, optional
        If True, wait for files to be created if not found,
        by default False.
    max_tries : int, optional
        Maximum number of attempts to read the files,
        by default MAX_TRIES.

    Returns
    -------
    ExperimentData
        ExperimentData object containing the loaded data.

    Examples
    --------
    >>> experiment_data = ExperimentData.from_file('path/to/project_dir')
    """
    if isinstance(project_dir, str):
        project_dir = Path(project_dir)

    try:
        return _from_file_attempt(
            project_dir=project_dir,
            wait_for_creation=wait_for_creation,
            max_tries=max_tries,
        )
    except FileNotFoundError:
        try:
            filename_with_path = Path(get_original_cwd()) / project_dir
        except ValueError as exc:  # get_original_cwd() error
            raise FileNotFoundError(
                f"Cannot find the folder {project_dir} !"
            ) from exc

        return _from_file_attempt(
            project_dir=filename_with_path,
            wait_for_creation=wait_for_creation,
            max_tries=max_tries,
        )

from_yaml(config: DictConfig) -> ExperimentData classmethod

Create an ExperimentData object from a YAML configuration.

Parameters:

Name	Type	Description	Default
config	DictConfig	Hydra DictConfig object containing the configuration.	required

Returns:

Type	Description
ExperimentData	ExperimentData object containing the loaded data.

Examples:

>>> experiment_data = ExperimentData.from_yaml(config)

Source code in src/f3dasm/_src/experimentdata.py

@classmethod
def from_yaml(cls, config: DictConfig) -> ExperimentData:
    """
    Create an ExperimentData object from a YAML configuration.

    Parameters
    ----------
    config : DictConfig
        Hydra DictConfig object containing the configuration.

    Returns
    -------
    ExperimentData
        ExperimentData object containing the loaded data.

    Examples
    --------
    >>> experiment_data = ExperimentData.from_yaml(config)
    """
    # Option 1: From existing ExperimentData files
    if "from_file" in config:
        return cls.from_file(config.from_file)

    else:
        return cls(**config)

get_experiment_sample(id: int) -> ExperimentSample

Gets the experiment_sample at the given index.

Parameters:

Name	Type	Description	Default
id	int	The index of the experiment_sample to retrieve.	required

Returns:

Type	Description
ExperimentSample	The ExperimentSample at the given index.

Examples:

>>> sample = experiment_data.get_experiment_sample(0)

Source code in src/f3dasm/_src/experimentdata.py

def get_experiment_sample(self, id: int) -> ExperimentSample:
    """
    Gets the experiment_sample at the given index.

    Parameters
    ----------
    id : int
        The index of the experiment_sample to retrieve.

    Returns
    -------
    ExperimentSample
        The ExperimentSample at the given index.

    Examples
    --------
    >>> sample = experiment_data.get_experiment_sample(0)
    """
    return self.data[id]

get_n_best_output(n_samples: int, output_name: Optional[str] = 'y') -> ExperimentData

Get the n best samples from the output data. Lower values are better.

Parameters:

Name	Type	Description	Default
n_samples	int	Number of samples to select.	required
output_name	str	The name of the output column to sort by, by default 'y'.	'y'

Returns:

Type	Description
ExperimentData	New ExperimentData object with a selection of the n best samples.

Examples:

>>> best_samples = experiment_data.get_n_best_output(5)

Source code in src/f3dasm/_src/experimentdata.py

def get_n_best_output(
    self, n_samples: int, output_name: Optional[str] = "y"
) -> ExperimentData:
    """
    Get the n best samples from the output data. Lower values are better.

    Parameters
    ----------
    n_samples : int
        Number of samples to select.
    output_name : str, optional
        The name of the output column to sort by, by default 'y'.

    Returns
    -------
    ExperimentData
        New ExperimentData object with a selection of the n best samples.

    Examples
    --------
    >>> best_samples = experiment_data.get_n_best_output(5)
    """
    _, df_out = self.to_pandas()
    indices = df_out.nsmallest(n=n_samples, columns=output_name).index
    return self[indices]

get_open_job() -> tuple[int, ExperimentSample, Domain]

Get the first open job in the ExperimentData object.

Returns:

Type	Description
tuple of int, ExperimentSample and the Domain	The index, ExperimentSample and Domain of the first open job.

Notes

This function iterates over the ExperimentData object and returns the first open job. If no open jobs are found, it returns None.

The returned open job is marked as 'in_progress'.

Examples:

>>> job_id, job_sample = experiment_data.get_open_job()

Source code in src/f3dasm/_src/experimentdata.py

def get_open_job(self) -> tuple[int, ExperimentSample, Domain]:
    """
    Get the first open job in the ExperimentData object.

    Returns
    -------
    tuple of int, ExperimentSample and the Domain
        The index, ExperimentSample and Domain of the first open job.

    Notes
    -----
    This function iterates over the ExperimentData object and returns the
    first open job. If no open jobs are found, it returns None.

    The returned open job is marked as 'in_progress'.

    Examples
    --------
    >>> job_id, job_sample = experiment_data.get_open_job()
    """
    for id, es in self:
        if es.is_status("open"):
            es.mark("in_progress")
            return id, es, self.domain

    return None, ExperimentSample(), self.domain

is_all_finished() -> bool

Check if all jobs are finished.

Returns:

Type	Description
bool	True if all jobs are finished, False otherwise.

Examples:

>>> experiment_data.is_all_finished()
True

Source code in src/f3dasm/_src/experimentdata.py

def is_all_finished(self) -> bool:
    """
    Check if all jobs are finished.

    Returns
    -------
    bool
        True if all jobs are finished, False otherwise.

    Examples
    --------
    >>> experiment_data.is_all_finished()
    True
    """
    return all(es.is_status("finished") for _, es in self)

join(experiment_data: ExperimentData) -> ExperimentData

Join two ExperimentData objects.

Parameters:

Name	Type	Description	Default
experiment_data	ExperimentData	The other ExperimentData object to join with.	required

Returns:

Type	Description
ExperimentData	The joined ExperimentData object.

Examples:

>>> joined_data = experiment_data1.join(experiment_data2)

Source code in src/f3dasm/_src/experimentdata.py

def join(self, experiment_data: ExperimentData) -> ExperimentData:
    """
    Join two ExperimentData objects.

    Parameters
    ----------
    experiment_data : ExperimentData
        The other ExperimentData object to join with.

    Returns
    -------
    ExperimentData
        The joined ExperimentData object.

    Examples
    --------
    >>> joined_data = experiment_data1.join(experiment_data2)
    """
    copy_self = self.reset_index()
    # TODO: Reset isnt necessary, only copy
    copy_other = experiment_data.reset_index()

    if not copy_self.data:
        copy_other._domain += copy_self._domain
        return copy_other

    for (i, es_self), (_, es_other) in zip(
        copy_self, copy_other, strict=False
    ):
        copy_self.data[i] = es_self + es_other

    copy_self._domain += copy_other._domain

    return copy_self

mark(indices: int | Iterable[int], status: Literal[open, in_progress, finished, error], in_place: bool = False)

Mark the jobs at the given indices with the given status.

Parameters:

Name	Type	Description	Default
indices	int or Iterable[int]	Indices of the jobs to mark.	required
status	(open, in_progress, finished, error)	Status to mark the jobs with.	'open'

Raises:

Type	Description
ValueError	If the given status is not valid.

Examples:

>>> experiment_data.mark([0, 1], 'finished')

Source code in src/f3dasm/_src/experimentdata.py

def mark(
    self,
    indices: int | Iterable[int],
    status: Literal["open", "in_progress", "finished", "error"],
    in_place: bool = False,
):
    """
    Mark the jobs at the given indices with the given status.

    Parameters
    ----------
    indices : int or Iterable[int]
        Indices of the jobs to mark.
    status : {'open', 'in_progress', 'finished', 'error'}
        Status to mark the jobs with.

    Raises
    ------
    ValueError
        If the given status is not valid.

    Examples
    --------
    >>> experiment_data.mark([0, 1], 'finished')
    """
    d = self._copy(in_place=in_place)

    if isinstance(indices, int):
        indices = [indices]
    for i in indices:
        d.data[i].mark(status)

    if in_place:
        return None
    else:
        return d

mark_all(status: Literal[open, in_progress, finished, error], in_place: bool = False)

Mark all the experiments with the given status.

Parameters:

Name	Type	Description	Default
status	(open, in_progress, finished, error)	Status to mark the jobs with.	'open'

Raises:

Type	Description
ValueError	If the given status is not valid.

Examples:

>>> experiment_data.mark_all('finished')

Source code in src/f3dasm/_src/experimentdata.py

def mark_all(
    self,
    status: Literal["open", "in_progress", "finished", "error"],
    in_place: bool = False,
):
    """
    Mark all the experiments with the given status.

    Parameters
    ----------
    status : {'open', 'in_progress', 'finished', 'error'}
        Status to mark the jobs with.

    Raises
    ------
    ValueError
        If the given status is not valid.

    Examples
    --------
    >>> experiment_data.mark_all('finished')
    """
    d = self._copy(in_place=in_place)

    for _, es in d:
        es.mark(status)

    if in_place:
        return None
    else:
        return d

move_to_input(name: str, in_place: bool = False)

Move a parameter from the output space to the input space.

The parameter entry is removed from the domain's output space and added to the input space. For every experiment sample, the corresponding value is moved from _output_data to _input_data.

Parameters:

Name	Type	Description	Default
name	str	Name of the output parameter to move.	required
in_place	bool	If True, the operation is performed in place and None is returned, by default False.	False

Returns:

Type	Description
ExperimentData or None	A new ExperimentData with the parameter moved, or None if in_place=True.

Raises:

Type	Description
KeyError	If name is not found in the output space.

Examples:

>>> new_data = experiment_data.move_to_input('y')
>>> experiment_data.move_to_input('y', in_place=True)

Source code in src/f3dasm/_src/experimentdata.py

def move_to_input(self, name: str, in_place: bool = False):
    """Move a parameter from the output space to the input space.

    The parameter entry is removed from the domain's output space and added
    to the input space. For every experiment sample, the corresponding
    value is moved from ``_output_data`` to ``_input_data``.

    Parameters
    ----------
    name : str
        Name of the output parameter to move.
    in_place : bool, optional
        If True, the operation is performed in place and None is returned,
        by default False.

    Returns
    -------
    ExperimentData or None
        A new ExperimentData with the parameter moved, or None if
        ``in_place=True``.

    Raises
    ------
    KeyError
        If ``name`` is not found in the output space.

    Examples
    --------
    >>> new_data = experiment_data.move_to_input('y')
    >>> experiment_data.move_to_input('y', in_place=True)
    """
    d = self._copy(in_place=in_place)
    if name not in d._domain.output_space:
        raise KeyError(f"Parameter '{name}' not found in output space.")
    d._domain.input_space[name] = d._domain.output_space.pop(name)
    for _, es in d:
        if name in es._output_data:
            es._input_data[name] = es._output_data.pop(name)
    if in_place:
        return None
    return d

move_to_output(name: str, in_place: bool = False)

Move a parameter from the input space to the output space.

The parameter entry is removed from the domain's input space and added to the output space. For every experiment sample, the corresponding value is moved from _input_data to _output_data.

Parameters:

Name	Type	Description	Default
name	str	Name of the input parameter to move.	required
in_place	bool	If True, the operation is performed in place and None is returned, by default False.	False

Returns:

Type	Description
ExperimentData or None	A new ExperimentData with the parameter moved, or None if in_place=True.

Raises:

Type	Description
KeyError	If name is not found in the input space.

Examples:

>>> new_data = experiment_data.move_to_output('x0')
>>> experiment_data.move_to_output('x0', in_place=True)

Source code in src/f3dasm/_src/experimentdata.py

def move_to_output(self, name: str, in_place: bool = False):
    """Move a parameter from the input space to the output space.

    The parameter entry is removed from the domain's input space and added
    to the output space. For every experiment sample, the corresponding
    value is moved from ``_input_data`` to ``_output_data``.

    Parameters
    ----------
    name : str
        Name of the input parameter to move.
    in_place : bool, optional
        If True, the operation is performed in place and None is returned,
        by default False.

    Returns
    -------
    ExperimentData or None
        A new ExperimentData with the parameter moved, or None if
        ``in_place=True``.

    Raises
    ------
    KeyError
        If ``name`` is not found in the input space.

    Examples
    --------
    >>> new_data = experiment_data.move_to_output('x0')
    >>> experiment_data.move_to_output('x0', in_place=True)
    """
    d = self._copy(in_place=in_place)
    if name not in d._domain.input_space:
        raise KeyError(f"Parameter '{name}' not found in input space.")
    d._domain.output_space[name] = d._domain.input_space.pop(name)
    for _, es in d:
        if name in es._input_data:
            es._output_data[name] = es._input_data.pop(name)
    if in_place:
        return None
    return d

remove_rows_bottom(number_of_rows: int, in_place: bool = False)

Remove a number of rows from the end of the ExperimentData object.

Parameters:

Name	Type	Description	Default
number_of_rows	int	Number of rows to remove from the bottom.	required
in_place	bool	If True, the rows are removed in place, by default False.	False

Examples:

>>> experiment_data.remove_rows_bottom(3)

Source code in src/f3dasm/_src/experimentdata.py

def remove_rows_bottom(self, number_of_rows: int, in_place: bool = False):
    """
    Remove a number of rows from the end of the ExperimentData object.

    Parameters
    ----------
    number_of_rows : int
        Number of rows to remove from the bottom.
    in_place : bool, optional
        If True, the rows are removed in place, by default False.

    Examples
    --------
    >>> experiment_data.remove_rows_bottom(3)
    """
    d = self._copy(in_place=in_place)

    # remove the last n rows
    for _i in range(number_of_rows):
        d.data.pop(d.index[-1])

    if in_place:
        return None
    else:
        return d

replace_nan(value: Any, in_place: bool = False)

Replace all NaN values in the output data with the given value.

Parameters:

Name	Type	Description	Default
value	Any	The value to replace NaNs with.	required
in_place	bool	If True, the NaN values are replaced in place, by default False.	False

Examples:

>>> experiment_data.replace_nan(0)

Source code in src/f3dasm/_src/experimentdata.py

def replace_nan(self, value: Any, in_place: bool = False):
    """
    Replace all NaN values in the output data with the given value.

    Parameters
    ----------
    value : Any
        The value to replace NaNs with.
    in_place : bool, optional
        If True, the NaN values are replaced in place, by default False.

    Examples
    --------
    >>> experiment_data.replace_nan(0)
    """
    d = self._copy(in_place=in_place)
    for _, es in d:
        es.replace_nan(value)

    if in_place:
        return None
    else:
        return d

reset_index() -> ExperimentData

Reset the index of the ExperimentData object. The index will be reset to a range from 0 to the number of experiments.

Returns:

Type	Description
ExperimentData	ExperimentData object with a reset index.

Examples:

>>> reset_data = experiment_data.reset_index()

Source code in src/f3dasm/_src/experimentdata.py

def reset_index(self) -> ExperimentData:
    """
    Reset the index of the ExperimentData object.
    The index will be reset to a range from 0 to the number of experiments.

    Returns
    -------
    ExperimentData
        ExperimentData object with a reset index.

    Examples
    --------
    >>> reset_data = experiment_data.reset_index()
    """
    return ExperimentData.from_data(
        data={i: v for i, v in enumerate(self.data.values())},
        domain=self._domain,
        project_dir=self._project_dir,
    )

round(decimals: int, in_place: bool = False)

Round all output data to the given number of decimals.

Parameters:

Name	Type	Description	Default
decimals	int	Number of decimals to round to.	required
in_place	bool	If True, round values in place, by default False.	False

Examples:

>>> experiment_data.round(2)

Source code in src/f3dasm/_src/experimentdata.py

def round(self, decimals: int, in_place: bool = False):
    """
    Round all output data to the given number of decimals.

    Parameters
    ----------
    decimals : int
        Number of decimals to round to.
    in_place : bool, optional
        If True, round values in place, by default False.

    Examples
    --------
    >>> experiment_data.round(2)
    """
    d = self._copy(in_place=in_place)

    for _, es in d:
        es.round(decimals)

    if in_place:
        return None
    else:
        return d

select(indices: int | Iterable[int]) -> ExperimentData

Select a subset of the ExperimentData object.

Parameters:

Name	Type	Description	Default
indices	int or Iterable[int]	The indices to select.	required

Returns:

Type	Description
ExperimentData	The selected subset of the ExperimentData object.

Examples:

>>> subset = experiment_data.select([0, 1, 2])

Source code in src/f3dasm/_src/experimentdata.py

def select(self, indices: int | Iterable[int]) -> ExperimentData:
    """
    Select a subset of the ExperimentData object.

    Parameters
    ----------
    indices : int or Iterable[int]
        The indices to select.

    Returns
    -------
    ExperimentData
        The selected subset of the ExperimentData object.

    Examples
    --------
    >>> subset = experiment_data.select([0, 1, 2])
    """
    return self[indices]

select_parameter(name: str) -> ExperimentData

Return an ExperimentData containing only the named parameter.

Parameters:

Name	Type	Description	Default
name	str	Name of the input or output parameter to select.	required

Returns:

Type	Description
ExperimentData	New ExperimentData with a single-parameter domain and only the data for that parameter.

Raises:

Type	Description
KeyError	If name is not found in either input or output space.

Examples:

>>> exp_model = experiment_data.select_parameter('model')

Source code in src/f3dasm/_src/experimentdata.py

def select_parameter(self, name: str) -> ExperimentData:
    """Return an ExperimentData containing only the named parameter.

    Parameters
    ----------
    name : str
        Name of the input or output parameter to select.

    Returns
    -------
    ExperimentData
        New ExperimentData with a single-parameter domain and only the
        data for that parameter.

    Raises
    ------
    KeyError
        If ``name`` is not found in either input or output space.

    Examples
    --------
    >>> exp_model = experiment_data.select_parameter('model')
    """
    if name in self._domain.input_space:
        new_domain = Domain(
            input_space={name: self._domain.input_space[name]}
        )
        new_data = {
            idx: ExperimentSample(
                _input_data={name: es._input_data.get(name)},
                _output_data={},
                job_status=es.job_status,
                project_dir=es.project_dir,
            )
            for idx, es in self
        }
    elif name in self._domain.output_space:
        new_domain = Domain(
            output_space={name: self._domain.output_space[name]}
        )
        new_data = {
            idx: ExperimentSample(
                _input_data={},
                _output_data={name: es._output_data.get(name)},
                job_status=es.job_status,
                project_dir=es.project_dir,
            )
            for idx, es in self
        }
    else:
        raise KeyError(f"Parameter '{name}' not found in domain.")

    return ExperimentData.from_data(
        data=new_data,
        domain=new_domain,
        project_dir=self._project_dir,
    )

select_with_status(status: Literal[open, in_progress, finished, error]) -> ExperimentData

Select a subset of the ExperimentData object with a given status.

Parameters:

Name	Type	Description	Default
status	(open, in_progress, finished, error)	The status to select.	'open'

Returns:

Type	Description
ExperimentData	The selected subset of the ExperimentData object with the given status.

Examples:

>>> subset = experiment_data.select_with_status('finished')

Source code in src/f3dasm/_src/experimentdata.py

def select_with_status(
    self, status: Literal["open", "in_progress", "finished", "error"]
) -> ExperimentData:
    """
    Select a subset of the ExperimentData object with a given status.

    Parameters
    ----------
    status : {'open', 'in_progress', 'finished', 'error'}
        The status to select.

    Returns
    -------
    ExperimentData
        The selected subset of the ExperimentData object with the given
        status.

    Examples
    --------
    >>> subset = experiment_data.select_with_status('finished')
    """
    idx = [i for i, es in self if es.is_status(status)]
    return self[idx]

set_project_dir(project_dir: Path | str, in_place: bool = False) -> ExperimentData

Set the directory of the f3dasm project folder.

Parameters:

Name	Type	Description	Default
project_dir	Path or str	Path to the project directory	required
in_place	bool	If True, the project directory is set in place, by default False	False

Returns:

Type	Description
ExperimentData	ExperimentData object with the updated project directory

Source code in src/f3dasm/_src/experimentdata.py

def set_project_dir(
    self, project_dir: Path | str, in_place: bool = False
) -> ExperimentData:
    """Set the directory of the f3dasm project folder.

    Parameters
    ----------
    project_dir : Path or str
        Path to the project directory
    in_place : bool, optional
        If True, the project directory is set in place, by default False

    Returns
    -------
    ExperimentData
        ExperimentData object with the updated project directory
    """
    d = self._copy(in_place=in_place)
    d._project_dir = _project_dir_factory(project_dir)

    if in_place:
        return None
    else:
        return d

sort(criterion: Callable[[ExperimentSample], Any], reverse: bool = False) -> ExperimentData

Sort the ExperimentData object based on a criterion.

Parameters:

Name	Type	Description	Default
criterion	Callable[[ExperimentSample], Any]	The criterion to sort on. This should be a function that takes an ExperimentSample object and returns a value to sort on.	required
reverse	bool	If True, sort in descending order, by default False.	False

Returns:

Type	Description
ExperimentData	The sorted ExperimentData object.

Examples:

>>> sorted_data = experiment_data.sort(lambda x: x.output_data['y'])

Source code in src/f3dasm/_src/experimentdata.py

def sort(
    self,
    criterion: Callable[[ExperimentSample], Any],
    reverse: bool = False,
) -> ExperimentData:
    """
    Sort the ExperimentData object based on a criterion.

    Parameters
    ----------
    criterion : Callable[[ExperimentSample], Any]
        The criterion to sort on. This should be a function that takes an
        ExperimentSample object and returns a value to sort on.
    reverse : bool, optional
        If True, sort in descending order, by default False.

    Returns
    -------
    ExperimentData
        The sorted ExperimentData object.

    Examples
    --------
    >>> sorted_data = experiment_data.sort(lambda x: x.output_data['y'])
    """

    sorted_data = dict(
        sorted(
            self.data.items(),
            key=lambda item: criterion(item[1]),
            reverse=reverse,
        )
    )
    return ExperimentData.from_data(
        data=sorted_data,
        domain=self._domain,
        project_dir=self._project_dir,
    )

store(project_dir: Optional[Path | str] = None, copy_references: bool = False)

Write the ExperimentData to disk in the project directory.

Parameters:

Name	Type	Description	Default
project_dir	Optional[Path | str]	The f3dasm project directory to store the ExperimentData object to, by default None.	None
copy_references	bool	If True, any :class:ReferenceValue objects whose source project_dir differs from the destination are physically copied into the destination project directory before writing. The in-memory references are updated to point to the new locations, by default False.	False

Note

If no project directory is provided, the ExperimentData object is stored in the directory provided by the .project_dir attribute that is set upon creation of the object.

The ExperimentData object is stored in a subfolder 'experiment_data'.

The ExperimentData object is stored in four files:

• the input data (input.csv)
• the output data (output.csv)
• the jobs (jobs.csv)
• the domain (domain.json)

To avoid the ExperimentData to be written simultaneously by multiple processes, a '.lock' file is automatically created in the project directory. Concurrent process can only sequentially access the lock file. This lock file is removed after the ExperimentData object is written to disk.

Source code in src/f3dasm/_src/experimentdata.py

def store(
    self,
    project_dir: Optional[Path | str] = None,
    copy_references: bool = False,
):
    """
    Write the ExperimentData to disk in the project directory.

    Parameters
    ----------
    project_dir : Optional[Path | str], optional
        The f3dasm project directory to store the
        ExperimentData object to, by default None.
    copy_references : bool, optional
        If True, any :class:`ReferenceValue` objects whose source
        ``project_dir`` differs from the destination are physically
        copied into the destination project directory before writing.
        The in-memory references are updated to point to the new
        locations, by default False.

    Note
    ----
    If no project directory is provided, the ExperimentData object is
    stored in the directory provided by the `.project_dir` attribute that
    is set upon creation of the object.

    The ExperimentData object is stored in a subfolder 'experiment_data'.

    The ExperimentData object is stored in four files:

    * the input data (`input.csv`)
    * the output data (`output.csv`)
    * the jobs (`jobs.csv`)
    * the domain (`domain.json`)

    To avoid the ExperimentData to be written simultaneously by multiple
    processes, a '.lock' file is automatically created
    in the project directory. Concurrent process can only sequentially
    access the lock file. This lock file is removed after the
    ExperimentData object is written to disk.
    """
    old_project_dir = self._project_dir

    if project_dir is not None:
        self.set_project_dir(project_dir, in_place=True)

    if copy_references:
        seen: set[Path] = set()
        for _, es in self:
            for value in list(es._input_data.values()) + list(
                es._output_data.values()
            ):
                if (
                    isinstance(value, ReferenceValue)
                    and value.reference not in seen
                ):
                    seen.add(value.reference)
                    _copy_reference(
                        value.reference, old_project_dir, self._project_dir
                    )

    subdirectory = self._project_dir / EXPERIMENTDATA_SUBFOLDER

    # Create the experimentdata subfolder if it does not exist
    subdirectory.mkdir(parents=True, exist_ok=True)

    # # Store all objects to keep references
    # self.store_objects()

    df_input, df_output = self.to_pandas(keep_references=True)

    df_input.to_csv(
        (subdirectory / INPUT_DATA_FILENAME).with_suffix(".csv")
    )
    df_output.to_csv(
        (subdirectory / OUTPUT_DATA_FILENAME).with_suffix(".csv")
    )
    self._domain.store(subdirectory / DOMAIN_FILENAME)
    self.jobs.to_csv((subdirectory / JOBS_FILENAME).with_suffix(".csv"))

store_experimentsample(experiment_sample: ExperimentSample, idx: int, domain: Domain | None = None)

Store an ExperimentSample object in the ExperimentData object and update the Domain object.

Parameters:

Name	Type	Description	Default
experiment_sample	ExperimentSample	The ExperimentSample object to store.	required
idx	int	The index of the ExperimentSample object.	required

Examples:

>>> experiment_data.store_experimentsample(sample, 0)

Source code in src/f3dasm/_src/experimentdata.py

def store_experimentsample(
    self,
    experiment_sample: ExperimentSample,
    idx: int,
    domain: Domain | None = None,
):
    """
    Store an ExperimentSample object in the ExperimentData object and
    update the Domain object.

    Parameters
    ----------
    experiment_sample : ExperimentSample
        The ExperimentSample object to store.
    idx : int
        The index of the ExperimentSample object.

    Examples
    --------
    >>> experiment_data.store_experimentsample(sample, 0)
    """
    experiment_sample, domain = _store(
        experiment_sample=experiment_sample,
        idx=idx,
        domain=domain if domain is not None else self.domain,
    )

    self._domain = domain
    self.data[idx] = experiment_sample

store_experimentsample_references()

Store references to input and output data in the experiment sample based on the domain.

Notes

This method checks the domain for parameters that should be stored on disk. If a parameter is marked to be stored on disk, the method will store the corresponding value in the experiment sample using the store method.

Examples:

>>> domain = Domain()
>>> domain.add_float(name='param1', to_disk=True)
>>> sample = ExperimentSample(
...     _input_data={'param1': 1.0, 'param2': 2.0},
...     _output_data={'result1': 3.0}
... )
>>> sample.store_experimentsample_references()
>>> isinstance(sample._input_data['param1'], ToDiskValue)
True

Source code in src/f3dasm/_src/experimentdata.py

def store_experimentsample_references(self):
    """
    Store references to input and output data in the experiment sample
    based on the domain.

    Notes
    -----
    This method checks the domain for parameters that should be stored
    on disk. If a parameter is marked to be stored on disk, the method
    will store the corresponding value in the experiment sample using
    the `store` method.

    Examples
    --------
    >>> domain = Domain()
    >>> domain.add_float(name='param1', to_disk=True)
    >>> sample = ExperimentSample(
    ...     _input_data={'param1': 1.0, 'param2': 2.0},
    ...     _output_data={'result1': 3.0}
    ... )
    >>> sample.store_experimentsample_references()
    >>> isinstance(sample._input_data['param1'], ToDiskValue)
    True
    """
    for _, es in self:
        for name, value in es._input_data.items():
            input_parameter = self.domain.input_space.get(name, None)
            if input_parameter is not None and input_parameter.to_disk:
                es.store(
                    name=name,
                    object=value,
                    to_disk=True,
                    store_function=input_parameter.store_function,
                    load_function=input_parameter.load_function,
                    which="input",
                )

        for name, value in es._output_data.items():
            output_parameter = self.domain.output_space.get(name, None)
            if output_parameter is not None and output_parameter.to_disk:
                es.store(
                    name=name,
                    object=value,
                    to_disk=True,
                    store_function=output_parameter.store_function,
                    load_function=output_parameter.load_function,
                    which="output",
                )

store_objects()

Source code in src/f3dasm/_src/experimentdata.py

def store_objects(self):
    self.store_experimentsample_references()
    # Store to_disk objects so that the references are kept only
    for idx, experiment_sample in self:
        self.store_experimentsample(
            experiment_sample=experiment_sample,
            idx=idx,
            domain=self.domain,
        )

to_multiindex() -> pd.DataFrame

Convert the ExperimentData object to a pandas DataFrame with a MultiIndex. This is used for visualization purposes in a Jupyter notebook environment.

Returns:

Type	Description
DataFrame	A pandas DataFrame with a MultiIndex.

Examples:

>>> df_multiindex = experiment_data.to_multiindex()

Source code in src/f3dasm/_src/experimentdata.py

def to_multiindex(self) -> pd.DataFrame:
    """
    Convert the ExperimentData object to a pandas DataFrame with a
    MultiIndex. This is used for visualization purposes in a Jupyter
    notebook environment.

    Returns
    -------
    pd.DataFrame
        A pandas DataFrame with a MultiIndex.

    Examples
    --------
    >>> df_multiindex = experiment_data.to_multiindex()
    """
    list_of_dicts = [sample.to_multiindex() for _, sample in self]
    return pd.DataFrame(merge_dicts(list_of_dicts), index=self.index)

to_numpy() -> tuple[np.ndarray, np.ndarray]

Convert the ExperimentData object to a tuple of numpy arrays.

Returns:

Type	Description
tuple of np.ndarray	A tuple containing two numpy arrays, the first one for input columns, and the second for output columns.

Examples:

>>> input_array, output_array = experiment_data.to_numpy()

Source code in src/f3dasm/_src/experimentdata.py

def to_numpy(self) -> tuple[np.ndarray, np.ndarray]:
    """
    Convert the ExperimentData object to a tuple of numpy arrays.

    Returns
    -------
    tuple of np.ndarray
        A tuple containing two numpy arrays, the first one for input
        columns, and the second for output columns.

    Examples
    --------
    >>> input_array, output_array = experiment_data.to_numpy()
    """
    df_input, df_output = self.to_pandas(keep_references=False)
    return df_input.to_numpy(), df_output.to_numpy()

to_pandas(keep_references: bool = False) -> tuple[pd.DataFrame, pd.DataFrame]

Convert the ExperimentData object to pandas DataFrames.

Parameters:

Name	Type	Description	Default
keep_references	bool	If True, the references to the output data are kept, by default False.	False

Returns:

Type	Description
tuple of pd.DataFrame	A tuple containing two pandas DataFrames, the first one for input columns, and the second for output columns.

Examples:

>>> df_input, df_output = experiment_data.to_pandas()

Source code in src/f3dasm/_src/experimentdata.py

def to_pandas(
    self, keep_references: bool = False
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """
    Convert the ExperimentData object to pandas DataFrames.

    Parameters
    ----------
    keep_references : bool, optional
        If True, the references to the output data are kept, by default
        False.

    Returns
    -------
    tuple of pd.DataFrame
        A tuple containing two pandas DataFrames, the first one for input
        columns, and the second for output columns.

    Examples
    --------
    >>> df_input, df_output = experiment_data.to_pandas()
    """
    if keep_references:
        return (
            pd.DataFrame(
                [es._input_data for _, es in self], index=self.index
            ),
            pd.DataFrame(
                [es._output_data for _, es in self], index=self.index
            ),
        )
    else:
        return (
            pd.DataFrame(
                [es.input_data for _, es in self], index=self.index
            ),
            pd.DataFrame(
                [es.output_data for _, es in self], index=self.index
            ),
        )

to_xarray(keep_references: bool = False) -> xr.Dataset

Convert the ExperimentData object to an xarray Dataset.

Parameters:

Name	Type	Description	Default
keep_references	bool	If True, the references to the output data are kept, by default False.	False

Returns:

Type	Description
Dataset	An xarray Dataset containing the data.

Examples:

>>> dataset = experiment_data.to_xarray()

Source code in src/f3dasm/_src/experimentdata.py

def to_xarray(self, keep_references: bool = False) -> xr.Dataset:
    """
    Convert the ExperimentData object to an xarray Dataset.

    Parameters
    ----------
    keep_references : bool, optional
        If True, the references to the output data are kept, by default
        False.

    Returns
    -------
    xr.Dataset
        An xarray Dataset containing the data.

    Examples
    --------
    >>> dataset = experiment_data.to_xarray()
    """
    df_input, df_output = self.to_pandas(keep_references=keep_references)

    da_input = xr.DataArray(
        df_input,
        dims=["iterations", "input_dim"],
        coords={"iterations": self.index, "input_dim": df_input.columns},
    )

    da_output = xr.DataArray(
        df_output,
        dims=["iterations", "output_dim"],
        coords={"iterations": self.index, "output_dim": df_output.columns},
    )

    return xr.Dataset({"input": da_input, "output": da_output})

update_from_experimentssample_json(in_place: bool = False)

Update the ExperimentData from ExperimentSample JSON files.

Parameters:

Name	Type	Description	Default
in_place	bool	If True, update in place, by default False.	False

Returns:

Type	Description
ExperimentData or None	Updated ExperimentData object, or None if in_place is True.

Notes

This method loads ExperimentSample objects from JSON files in the EXPERIMENTSAMPLE_SUBFOLDER directory. If loading fails for any file, a warning is logged and the process continues.

Source code in src/f3dasm/_src/experimentdata.py

def update_from_experimentssample_json(self, in_place: bool = False):
    """
    Update the ExperimentData from ExperimentSample JSON files.

    Parameters
    ----------
    in_place : bool, optional
        If True, update in place, by default False.

    Returns
    -------
    ExperimentData or None
        Updated ExperimentData object, or None if in_place is True.

    Notes
    -----
    This method loads ExperimentSample objects from JSON files in
    the EXPERIMENTSAMPLE_SUBFOLDER directory. If loading fails for
    any file, a warning is logged and the process continues.
    """
    d = self._copy(in_place=in_place)

    for json_file in (d.project_dir / EXPERIMENTSAMPLE_SUBFOLDER).glob(
        "*.json"
    ):
        try:
            idx = int(json_file.stem)
            es = ExperimentSample.from_json(json_file)
            d.data[idx] = es

        except Exception as exc:
            logger.warning(
                f"Could not load ExperimentSample from {json_file}: {exc}"
            )

    if in_place:
        return None
    else:
        return d

f3dasm.ExperimentSample

ExperimentSample(_input_data: 'dict[str, Any] | None' = , _output_data: 'dict[str, Any] | None' = , job_status: 'JobStatus | None | str' = None, project_dir: 'Path' = )

Source code in src/f3dasm/_src/experimentsample.py

@dataclass
class ExperimentSample:
    _input_data: dict[str, Any] | None = field(default_factory=dict)
    _output_data: dict[str, Any] | None = field(default_factory=dict)
    job_status: JobStatus | None | str = None
    project_dir: Path = field(default_factory=Path.cwd)
    """
    Realization of a single experiment in the design-of-experiment.

    Parameters
    ----------
    _input_data : dict[str, Any] | None
        Input parameters of one experiment.
        The key is the name of the parameter.
    _output_data : dict[str, Any] | None
        Output parameters of one experiment.
        The key is the name of the parameter.
    job_status : JobStatus | None
        Job status of the experiment, by default None.
    project_dir : Optional[Path]
        Directory of the project, by default None.

    Examples
    --------
    >>> sample = ExperimentSample(
    ...     _input_data={'param1': 1.0},
    ...     _output_data={'result1': 2.0}
    ... )
    >>> print(sample)
    ExperimentSample(input_data={'param1': 1.0},
    output_data={'result1': 2.0}, job_status=JobStatus.OPEN)
    """

    def __post_init__(self):
        """Handle defaults and consistency checks after dataclass init."""
        # Infer job_status if not provided
        if self.job_status is None:
            self.job_status = (
                JobStatus.FINISHED if self._output_data else JobStatus.OPEN
            )

        if isinstance(self.job_status, str):
            # Convert string job_status to JobStatus enum
            try:
                self.job_status = JobStatus[self.job_status]
            except KeyError as exc:
                raise DecodeError() from exc

        if self._output_data is None:
            self._output_data = {}
        if self._input_data is None:
            self._input_data = {}

    def __repr__(self):
        """
        Return a string representation of the ExperimentSample instance.

        Returns
        -------
        str
            String representation of the ExperimentSample instance.
        """
        return (
            f"ExperimentSample("
            f"input_data={self.input_data}, "
            f"output_data={self.output_data}, "
            f"job_status={self.job_status})"
        )

    def __add__(self, __o: ExperimentSample) -> ExperimentSample:
        """
        Add two ExperimentSample instances.

        Parameters
        ----------
        __o : ExperimentSample
            Another ExperimentSample instance.

        Returns
        -------
        ExperimentSample
            A new ExperimentSample instance with combined input
            and output data.

        Notes
        -----
        The job status of the new ExperimentSample instance will be
        reconstructed from the absence or presence of output data.
        If output data is present, the job status will be 'FINISHED'.
        Otherwise, the job status will be 'OPEN'.

        Examples
        --------
        >>> sample1 = ExperimentSample(input_data={'param1': 1.0})
        >>> sample2 = ExperimentSample(output_data={'result1': 2.0})
        >>> combined_sample = sample1 + sample2
        >>> print(combined_sample)
        ExperimentSample(input_data={'param1': 1.0},
        output_data={'result1': 2.0}, job_status=JobStatus.FINISHED)
        """
        return ExperimentSample(
            _input_data={**self._input_data, **__o._input_data},
            _output_data={**self._output_data, **__o._output_data},
            project_dir=self.project_dir,
        )

    # TODO: the self.project_dir should also be compared, but it
    # breaks some tests
    def __eq__(self, __o: ExperimentSample) -> bool:
        """
        Check if two ExperimentSample instances are equal.

        Parameters
        ----------
        __o : ExperimentSample
            Another ExperimentSample instance.

        Returns
        -------
        bool
            True if the instances are equal, False otherwise.
        """
        return (
            self._input_data == __o._input_data
            and self._output_data == __o._output_data
            and self.job_status == __o.job_status
        )

    def _copy(self) -> ExperimentSample:
        """
        Create a copy of the ExperimentSample instance.

        Returns
        -------
        ExperimentSample
            A new ExperimentSample instance with the same input and
            output data.
        """
        return ExperimentSample(
            _input_data=deepcopy(self._input_data),
            _output_data=deepcopy(self._output_data),
            job_status=self.job_status,
            project_dir=self.project_dir,
        )

    @property
    def input_data(self) -> dict[str, Any]:
        """
        Get the input data of the experiment.

        Returns
        -------
        Dict[str, Any]
            Input data of the experiment.
        """
        return {
            k: _get_value(value=v, project_dir=self.project_dir)
            for k, v in self._input_data.items()
        }

    @property
    def output_data(self) -> dict[str, Any]:
        """
        Get the output data of the experiment.

        Returns
        -------
        Dict[str, Any]
            Output data of the experiment.
        """
        return {
            k: _get_value(value=v, project_dir=self.project_dir)
            for k, v in self._output_data.items()
        }

    @classmethod
    def from_numpy(
        cls: type[ExperimentSample], input_array: np.ndarray
    ) -> ExperimentSample:
        """
        Create an ExperimentSample instance from a numpy array.

        Parameters
        ----------
        input_array : np.ndarray
            Numpy array containing input data.

        Returns
        -------
        ExperimentSample
            A new ExperimentSample instance.

        Notes
        -----
        The default names will be 'x0', 'x1', etc.

        Examples
        --------
        >>> import numpy as np
        >>> sample = ExperimentSample.from_numpy(np.array([1.0, 2.0]))
        >>> print(sample)
        ExperimentSample(input_data={'x0': 1.0, 'x1': 2.0},
        output_data={}, job_status=JobStatus.OPEN)
        """
        return cls(
            _input_data={
                f"x{i}": v for i, v in enumerate(input_array.flatten())
            },
        )

    @classmethod
    def from_json(cls, path: Path) -> ExperimentSample:
        """
        Create an ExperimentSample instance from a JSON file.

        Parameters
        ----------
        path : Path
            Path to the JSON file.

        Returns
        -------
        ExperimentSample
            A new ExperimentSample instance.

        Examples
        --------
        >>> sample = ExperimentSample.from_json(Path("sample.json"))
        >>> print(sample)
        ExperimentSample(input_data={'param1': 1.0},
        output_data={'result1': 2.0}, job_status=JobStatus.FINISHED)
        """
        with open(path) as f:
            data = json.load(f)

        def restore(obj):
            if (
                isinstance(obj, dict)
                and obj.get("__type__") == "ReferenceValue"
            ):
                return ReferenceValue.from_json(obj)
            return obj

        # Recursively apply restoration
        def walk(d):
            if isinstance(d, dict):
                return {k: walk(restore(v)) for k, v in d.items()}
            else:
                return d

        return cls(
            _input_data=walk(data["input_data"]),
            _output_data=walk(data["output_data"]),
            job_status=data["job_status"],
            project_dir=Path(data["project_dir"]),
        )

    def get(self, name: str) -> Any:
        """
        Get the value of a parameter by name.

        Parameters
        ----------
        name : str
            The name of the parameter.

        Returns
        -------
        Any
            The value of the parameter.

        Raises
        ------
        KeyError
            If the parameter is not found in input or output data.

        Examples
        --------
        >>> sample = ExperimentSample(input_data={'param1': 1.0})
        >>> sample.get('param1')
        1.0
        """
        value = self._input_data.get(name, None)
        if value is None:
            value = self._output_data.get(name, None)

        if value is None:
            raise KeyError(
                f"Parameter '{name}' not found in input or output data."
            )

        return _get_value(value=value, project_dir=self.project_dir)

    def mark(
        self, status: Literal["open", "in_progress", "finished", "error"]
    ):
        """
        Mark the job status of the experiment.

        Parameters
        ----------
        status : Literal['open', 'in_progress', 'finished', 'error']
            The new job status.

        Raises
        ------
        ValueError
            If the status is not valid.

        Examples
        --------
        >>> sample = ExperimentSample()
        >>> sample.mark('finished')
        >>> sample.job_status
        <JobStatus.FINISHED: 2>
        """
        try:
            # Look up enum member
            self.job_status = JobStatus[status.upper()]

        # If the status is invalid, raise ValueError
        except KeyError as exc:
            valid = ", ".join(s.lower() for s in JobStatus.__members__)
            raise ValueError(
                f"Invalid status '{status}'. Must be one of: {valid}"
            ) from exc

    def replace_nan(self, replacement_value: Any):
        """
        Replace NaN values in input_data and output_data with a custom value.

        Parameters
        ----------
        replacement_value : Any
            The value to replace NaN values with.

        Examples
        --------
        >>> sample = ExperimentSample(input_data={'param1': np.nan})
        >>> sample.replace_nan(0)
        >>> sample.input_data['param1']
        0
        """

        def replace_nan_in_dict(data: dict[str, Any]) -> dict[str, Any]:
            return {
                k: (replacement_value if np.isnan(v) else v)
                for k, v in data.items()
            }

        self._input_data = replace_nan_in_dict(self._input_data)
        self._output_data = replace_nan_in_dict(self._output_data)

    def round(self, decimals: int):
        """
        Round the input and output data to a specified number
        of decimal places.

        Parameters
        ----------
        decimals : int
            The number of decimal places to round to.

        Examples
        --------
        >>> sample = ExperimentSample(input_data={'param1': 1.2345})
        >>> sample.round(2)
        >>> sample.input_data['param1']
        1.23
        """

        def round_dict(data: dict[str, Any]) -> dict[str, Any]:
            return {
                k: round(v, decimals) if isinstance(v, int | float) else v
                for k, v in data.items()
            }

        self._input_data = round_dict(self._input_data)
        self._output_data = round_dict(self._output_data)

    def to_multiindex(self) -> dict[tuple[str, str], Any]:
        """
        Convert the experiment sample to a multiindex dictionary.
        Used to display the data prettily as a table in a Jupyter notebook.

        Returns
        -------
        Dict[Tuple[str, str], Any]
            A multiindex dictionary containing the job status, input,
            and output data.

        Examples
        --------
        >>> sample = ExperimentSample(input_data={'param1': 1.0})
        >>> sample.to_multiindex()
        {('jobs', ''): 'open', ('input', 'param1'): 1.0}
        """
        return {
            ("jobs", ""): self.job_status.name.lower(),
            **{("input", k): v for k, v in self._input_data.items()},
            **{("output", k): v for k, v in self._output_data.items()},
        }

    def to_numpy(self) -> tuple[np.ndarray, np.ndarray]:
        """
        Convert the experiment sample to numpy arrays.

        Returns
        -------
        Tuple[np.ndarray, np.ndarray]
            A tuple containing numpy arrays of input and output data.

        Examples
        --------
        >>> sample = ExperimentSample(input_data={'param1': 1.0})
        >>> sample.to_numpy()
        (array([1.]), array([]))
        """
        return (
            np.array(list(self.input_data.values())),
            np.array(list(self.output_data.values())),
        )

    def to_dict(self) -> dict[str, Any]:
        """
        Convert the experiment sample to a dictionary.

        Returns
        -------
        Dict[str, Any]
            A dictionary containing both input and output data.

        Examples
        --------
        >>> sample = ExperimentSample(input_data={'param1': 1.0})
        >>> sample.to_dict()
        {'param1': 1.0}
        """
        return {**self.input_data, **self.output_data}

    def store(
        self,
        name: str,
        object: Any,
        to_disk: bool = False,
        store_function: Callable | None = None,
        load_function: Callable | None = None,
        load_kwargs: dict | None = None,
        which: Literal["input", "output"] = "output",
    ):
        """
        Store an object in the experiment sample.

        Parameters
        ----------
        name : str
            The name of the object to store.
        object : Any
            The object to store.
        to_disk : bool, optional
            If True, the object will be stored on disk, by default False.
        store_function : Optional[Type[Callable]], optional
            The function to use for storing the object on disk
            by default None.
        load_function : Optional[Type[Callable]], optional
            The function to use for loading the object from disk,
            by default None.
        load_kwargs : dict, optional
            Extra keyword arguments forwarded to ``load_function`` when
            the stored object is later loaded (issue #285). Defaults to
            None.
        which : Literal['input', 'output'], optional
            Specify whether to store the object in input or output data,
            by default 'output'.

        Notes
        -----
        The object will be stored in the input data if the name is in the
        input space of the domain. Otherwise, the object will be stored in
        the output data if the name is in the output space of the domain.

        If the object is stored on disk, the path to the stored object will
        be stored in the input or output data, depending on where the object
        is stored.

        The store_function should have the following signature:

        .. code-block:: python

            def store_function(object: Any, path: str) -> Path:
                ...

        The load_function should have the following signature:

        .. code-block:: python

            def load_function(path: str, **kwargs: Any) -> Any:
                ...
        """
        value = (
            object
            if not to_disk
            else ToDiskValue(
                object=object,
                name=name,
                store_function=store_function,
                load_function=load_function,
                load_kwargs=load_kwargs,
            )
        )

        if which == "input":
            self._input_data[name] = value
        elif which == "output":
            self._output_data[name] = value
        else:
            raise ValueError(
                f"Invalid value for 'which': {which}. "
                f"Expected 'input' or 'output'."
            )

    def store_as_json(self, idx: int):
        def default_serializer(obj):
            if isinstance(obj, Path):
                return str(obj)
            if isinstance(obj, ReferenceValue):
                return obj.to_json()
            return str(obj)

        data = {
            "input_data": self._input_data,
            "output_data": self._output_data,
            "job_status": self.job_status.name,
            "project_dir": self.project_dir,
        }

        file_path = (
            self.project_dir / EXPERIMENTSAMPLE_SUBFOLDER / f"{idx}"
        ).with_suffix(".json")
        file_path.parent.mkdir(parents=True, exist_ok=True)

        with open(file_path, "w") as f:
            json.dump(data, f, indent=2, default=default_serializer)

    #                                                                Job status
    # =========================================================================

    def is_status(self, status: str) -> bool:
        """
        Check if the job's current status matches the given status.

        Parameters
        ----------
        status : str
            The status to check against the job's current status.

        Returns
        -------
        bool
            True if the job's current status matches the given status,
            False otherwise.

        Examples
        --------
        >>> sample = ExperimentSample()
        >>> sample.is_status('open')
        True
        """
        return self.job_status == JobStatus[status.upper()]

input_data property

Get the input data of the experiment.

Returns:

Type	Description
Dict[str, Any]	Input data of the experiment.

output_data property

Get the output data of the experiment.

Returns:

Type	Description
Dict[str, Any]	Output data of the experiment.

_copy() -> ExperimentSample

Create a copy of the ExperimentSample instance.

Returns:

Type	Description
ExperimentSample	A new ExperimentSample instance with the same input and output data.

Source code in src/f3dasm/_src/experimentsample.py

def _copy(self) -> ExperimentSample:
    """
    Create a copy of the ExperimentSample instance.

    Returns
    -------
    ExperimentSample
        A new ExperimentSample instance with the same input and
        output data.
    """
    return ExperimentSample(
        _input_data=deepcopy(self._input_data),
        _output_data=deepcopy(self._output_data),
        job_status=self.job_status,
        project_dir=self.project_dir,
    )

from_json(path: Path) -> ExperimentSample classmethod

Create an ExperimentSample instance from a JSON file.

Parameters:

Name	Type	Description	Default
path	Path	Path to the JSON file.	required

Returns:

Type	Description
ExperimentSample	A new ExperimentSample instance.

Examples:

>>> sample = ExperimentSample.from_json(Path("sample.json"))
>>> print(sample)
ExperimentSample(input_data={'param1': 1.0},
output_data={'result1': 2.0}, job_status=JobStatus.FINISHED)

Source code in src/f3dasm/_src/experimentsample.py

@classmethod
def from_json(cls, path: Path) -> ExperimentSample:
    """
    Create an ExperimentSample instance from a JSON file.

    Parameters
    ----------
    path : Path
        Path to the JSON file.

    Returns
    -------
    ExperimentSample
        A new ExperimentSample instance.

    Examples
    --------
    >>> sample = ExperimentSample.from_json(Path("sample.json"))
    >>> print(sample)
    ExperimentSample(input_data={'param1': 1.0},
    output_data={'result1': 2.0}, job_status=JobStatus.FINISHED)
    """
    with open(path) as f:
        data = json.load(f)

    def restore(obj):
        if (
            isinstance(obj, dict)
            and obj.get("__type__") == "ReferenceValue"
        ):
            return ReferenceValue.from_json(obj)
        return obj

    # Recursively apply restoration
    def walk(d):
        if isinstance(d, dict):
            return {k: walk(restore(v)) for k, v in d.items()}
        else:
            return d

    return cls(
        _input_data=walk(data["input_data"]),
        _output_data=walk(data["output_data"]),
        job_status=data["job_status"],
        project_dir=Path(data["project_dir"]),
    )

from_numpy(input_array: np.ndarray) -> ExperimentSample classmethod

Create an ExperimentSample instance from a numpy array.

Parameters:

Name	Type	Description	Default
input_array	ndarray	Numpy array containing input data.	required

Returns:

Type	Description
ExperimentSample	A new ExperimentSample instance.

Notes

The default names will be 'x0', 'x1', etc.

Examples:

>>> import numpy as np
>>> sample = ExperimentSample.from_numpy(np.array([1.0, 2.0]))
>>> print(sample)
ExperimentSample(input_data={'x0': 1.0, 'x1': 2.0},
output_data={}, job_status=JobStatus.OPEN)

Source code in src/f3dasm/_src/experimentsample.py

@classmethod
def from_numpy(
    cls: type[ExperimentSample], input_array: np.ndarray
) -> ExperimentSample:
    """
    Create an ExperimentSample instance from a numpy array.

    Parameters
    ----------
    input_array : np.ndarray
        Numpy array containing input data.

    Returns
    -------
    ExperimentSample
        A new ExperimentSample instance.

    Notes
    -----
    The default names will be 'x0', 'x1', etc.

    Examples
    --------
    >>> import numpy as np
    >>> sample = ExperimentSample.from_numpy(np.array([1.0, 2.0]))
    >>> print(sample)
    ExperimentSample(input_data={'x0': 1.0, 'x1': 2.0},
    output_data={}, job_status=JobStatus.OPEN)
    """
    return cls(
        _input_data={
            f"x{i}": v for i, v in enumerate(input_array.flatten())
        },
    )

get(name: str) -> Any

Get the value of a parameter by name.

Parameters:

Name	Type	Description	Default
name	str	The name of the parameter.	required

Returns:

Type	Description
Any	The value of the parameter.

Raises:

Type	Description
KeyError	If the parameter is not found in input or output data.

Examples:

>>> sample = ExperimentSample(input_data={'param1': 1.0})
>>> sample.get('param1')
1.0

Source code in src/f3dasm/_src/experimentsample.py

def get(self, name: str) -> Any:
    """
    Get the value of a parameter by name.

    Parameters
    ----------
    name : str
        The name of the parameter.

    Returns
    -------
    Any
        The value of the parameter.

    Raises
    ------
    KeyError
        If the parameter is not found in input or output data.

    Examples
    --------
    >>> sample = ExperimentSample(input_data={'param1': 1.0})
    >>> sample.get('param1')
    1.0
    """
    value = self._input_data.get(name, None)
    if value is None:
        value = self._output_data.get(name, None)

    if value is None:
        raise KeyError(
            f"Parameter '{name}' not found in input or output data."
        )

    return _get_value(value=value, project_dir=self.project_dir)

is_status(status: str) -> bool

Check if the job's current status matches the given status.

Parameters:

Name	Type	Description	Default
status	str	The status to check against the job's current status.	required

Returns:

Type	Description
bool	True if the job's current status matches the given status, False otherwise.

Examples:

>>> sample = ExperimentSample()
>>> sample.is_status('open')
True

Source code in src/f3dasm/_src/experimentsample.py

def is_status(self, status: str) -> bool:
    """
    Check if the job's current status matches the given status.

    Parameters
    ----------
    status : str
        The status to check against the job's current status.

    Returns
    -------
    bool
        True if the job's current status matches the given status,
        False otherwise.

    Examples
    --------
    >>> sample = ExperimentSample()
    >>> sample.is_status('open')
    True
    """
    return self.job_status == JobStatus[status.upper()]

mark(status: Literal[open, in_progress, finished, error])

Mark the job status of the experiment.

Parameters:

Name	Type	Description	Default
status	Literal['open', 'in_progress', 'finished', 'error']	The new job status.	required

Raises:

Type	Description
ValueError	If the status is not valid.

Examples:

>>> sample = ExperimentSample()
>>> sample.mark('finished')
>>> sample.job_status
<JobStatus.FINISHED: 2>

Source code in src/f3dasm/_src/experimentsample.py

def mark(
    self, status: Literal["open", "in_progress", "finished", "error"]
):
    """
    Mark the job status of the experiment.

    Parameters
    ----------
    status : Literal['open', 'in_progress', 'finished', 'error']
        The new job status.

    Raises
    ------
    ValueError
        If the status is not valid.

    Examples
    --------
    >>> sample = ExperimentSample()
    >>> sample.mark('finished')
    >>> sample.job_status
    <JobStatus.FINISHED: 2>
    """
    try:
        # Look up enum member
        self.job_status = JobStatus[status.upper()]

    # If the status is invalid, raise ValueError
    except KeyError as exc:
        valid = ", ".join(s.lower() for s in JobStatus.__members__)
        raise ValueError(
            f"Invalid status '{status}'. Must be one of: {valid}"
        ) from exc

replace_nan(replacement_value: Any)

Replace NaN values in input_data and output_data with a custom value.

Parameters:

Name	Type	Description	Default
replacement_value	Any	The value to replace NaN values with.	required

Examples:

>>> sample = ExperimentSample(input_data={'param1': np.nan})
>>> sample.replace_nan(0)
>>> sample.input_data['param1']
0

Source code in src/f3dasm/_src/experimentsample.py

def replace_nan(self, replacement_value: Any):
    """
    Replace NaN values in input_data and output_data with a custom value.

    Parameters
    ----------
    replacement_value : Any
        The value to replace NaN values with.

    Examples
    --------
    >>> sample = ExperimentSample(input_data={'param1': np.nan})
    >>> sample.replace_nan(0)
    >>> sample.input_data['param1']
    0
    """

    def replace_nan_in_dict(data: dict[str, Any]) -> dict[str, Any]:
        return {
            k: (replacement_value if np.isnan(v) else v)
            for k, v in data.items()
        }

    self._input_data = replace_nan_in_dict(self._input_data)
    self._output_data = replace_nan_in_dict(self._output_data)

round(decimals: int)

Round the input and output data to a specified number of decimal places.

Parameters:

Name	Type	Description	Default
decimals	int	The number of decimal places to round to.	required

Examples:

>>> sample = ExperimentSample(input_data={'param1': 1.2345})
>>> sample.round(2)
>>> sample.input_data['param1']
1.23

Source code in src/f3dasm/_src/experimentsample.py

def round(self, decimals: int):
    """
    Round the input and output data to a specified number
    of decimal places.

    Parameters
    ----------
    decimals : int
        The number of decimal places to round to.

    Examples
    --------
    >>> sample = ExperimentSample(input_data={'param1': 1.2345})
    >>> sample.round(2)
    >>> sample.input_data['param1']
    1.23
    """

    def round_dict(data: dict[str, Any]) -> dict[str, Any]:
        return {
            k: round(v, decimals) if isinstance(v, int | float) else v
            for k, v in data.items()
        }

    self._input_data = round_dict(self._input_data)
    self._output_data = round_dict(self._output_data)

store(name: str, object: Any, to_disk: bool = False, store_function: Callable | None = None, load_function: Callable | None = None, load_kwargs: dict | None = None, which: Literal[input, output] = 'output')

Store an object in the experiment sample.

Parameters:

Name	Type	Description	Default
name	str	The name of the object to store.	required
object	Any	The object to store.	required
to_disk	bool	If True, the object will be stored on disk, by default False.	False
store_function	Optional[Type[Callable]]	The function to use for storing the object on disk by default None.	None
load_function	Optional[Type[Callable]]	The function to use for loading the object from disk, by default None.	None
load_kwargs	dict	Extra keyword arguments forwarded to load_function when the stored object is later loaded (issue #285). Defaults to None.	None
which	Literal['input', 'output']	Specify whether to store the object in input or output data, by default 'output'.	'output'

Notes

The object will be stored in the input data if the name is in the input space of the domain. Otherwise, the object will be stored in the output data if the name is in the output space of the domain.

If the object is stored on disk, the path to the stored object will be stored in the input or output data, depending on where the object is stored.

The store_function should have the following signature:

.. code-block:: python

def store_function(object: Any, path: str) -> Path:
    ...

The load_function should have the following signature:

.. code-block:: python

def load_function(path: str, **kwargs: Any) -> Any:
    ...

Source code in src/f3dasm/_src/experimentsample.py

def store(
    self,
    name: str,
    object: Any,
    to_disk: bool = False,
    store_function: Callable | None = None,
    load_function: Callable | None = None,
    load_kwargs: dict | None = None,
    which: Literal["input", "output"] = "output",
):
    """
    Store an object in the experiment sample.

    Parameters
    ----------
    name : str
        The name of the object to store.
    object : Any
        The object to store.
    to_disk : bool, optional
        If True, the object will be stored on disk, by default False.
    store_function : Optional[Type[Callable]], optional
        The function to use for storing the object on disk
        by default None.
    load_function : Optional[Type[Callable]], optional
        The function to use for loading the object from disk,
        by default None.
    load_kwargs : dict, optional
        Extra keyword arguments forwarded to ``load_function`` when
        the stored object is later loaded (issue #285). Defaults to
        None.
    which : Literal['input', 'output'], optional
        Specify whether to store the object in input or output data,
        by default 'output'.

    Notes
    -----
    The object will be stored in the input data if the name is in the
    input space of the domain. Otherwise, the object will be stored in
    the output data if the name is in the output space of the domain.

    If the object is stored on disk, the path to the stored object will
    be stored in the input or output data, depending on where the object
    is stored.

    The store_function should have the following signature:

    .. code-block:: python

        def store_function(object: Any, path: str) -> Path:
            ...

    The load_function should have the following signature:

    .. code-block:: python

        def load_function(path: str, **kwargs: Any) -> Any:
            ...
    """
    value = (
        object
        if not to_disk
        else ToDiskValue(
            object=object,
            name=name,
            store_function=store_function,
            load_function=load_function,
            load_kwargs=load_kwargs,
        )
    )

    if which == "input":
        self._input_data[name] = value
    elif which == "output":
        self._output_data[name] = value
    else:
        raise ValueError(
            f"Invalid value for 'which': {which}. "
            f"Expected 'input' or 'output'."
        )

store_as_json(idx: int)

Source code in src/f3dasm/_src/experimentsample.py

def store_as_json(self, idx: int):
    def default_serializer(obj):
        if isinstance(obj, Path):
            return str(obj)
        if isinstance(obj, ReferenceValue):
            return obj.to_json()
        return str(obj)

    data = {
        "input_data": self._input_data,
        "output_data": self._output_data,
        "job_status": self.job_status.name,
        "project_dir": self.project_dir,
    }

    file_path = (
        self.project_dir / EXPERIMENTSAMPLE_SUBFOLDER / f"{idx}"
    ).with_suffix(".json")
    file_path.parent.mkdir(parents=True, exist_ok=True)

    with open(file_path, "w") as f:
        json.dump(data, f, indent=2, default=default_serializer)

to_dict() -> dict[str, Any]

Convert the experiment sample to a dictionary.

Returns:

Type	Description
Dict[str, Any]	A dictionary containing both input and output data.

Examples:

>>> sample = ExperimentSample(input_data={'param1': 1.0})
>>> sample.to_dict()
{'param1': 1.0}

Source code in src/f3dasm/_src/experimentsample.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert the experiment sample to a dictionary.

    Returns
    -------
    Dict[str, Any]
        A dictionary containing both input and output data.

    Examples
    --------
    >>> sample = ExperimentSample(input_data={'param1': 1.0})
    >>> sample.to_dict()
    {'param1': 1.0}
    """
    return {**self.input_data, **self.output_data}

to_multiindex() -> dict[tuple[str, str], Any]

Convert the experiment sample to a multiindex dictionary. Used to display the data prettily as a table in a Jupyter notebook.

Returns:

Type	Description
Dict[Tuple[str, str], Any]	A multiindex dictionary containing the job status, input, and output data.

Examples:

>>> sample = ExperimentSample(input_data={'param1': 1.0})
>>> sample.to_multiindex()
{('jobs', ''): 'open', ('input', 'param1'): 1.0}

Source code in src/f3dasm/_src/experimentsample.py

def to_multiindex(self) -> dict[tuple[str, str], Any]:
    """
    Convert the experiment sample to a multiindex dictionary.
    Used to display the data prettily as a table in a Jupyter notebook.

    Returns
    -------
    Dict[Tuple[str, str], Any]
        A multiindex dictionary containing the job status, input,
        and output data.

    Examples
    --------
    >>> sample = ExperimentSample(input_data={'param1': 1.0})
    >>> sample.to_multiindex()
    {('jobs', ''): 'open', ('input', 'param1'): 1.0}
    """
    return {
        ("jobs", ""): self.job_status.name.lower(),
        **{("input", k): v for k, v in self._input_data.items()},
        **{("output", k): v for k, v in self._output_data.items()},
    }

to_numpy() -> tuple[np.ndarray, np.ndarray]

Convert the experiment sample to numpy arrays.

Returns:

Type	Description
Tuple[ndarray, ndarray]	A tuple containing numpy arrays of input and output data.

Examples:

>>> sample = ExperimentSample(input_data={'param1': 1.0})
>>> sample.to_numpy()
(array([1.]), array([]))

Source code in src/f3dasm/_src/experimentsample.py

def to_numpy(self) -> tuple[np.ndarray, np.ndarray]:
    """
    Convert the experiment sample to numpy arrays.

    Returns
    -------
    Tuple[np.ndarray, np.ndarray]
        A tuple containing numpy arrays of input and output data.

    Examples
    --------
    >>> sample = ExperimentSample(input_data={'param1': 1.0})
    >>> sample.to_numpy()
    (array([1.]), array([]))
    """
    return (
        np.array(list(self.input_data.values())),
        np.array(list(self.output_data.values())),
    )