Source code for alab_management.builders.samplebuilder

"""Build the sample object."""

from typing import TYPE_CHECKING, Any

from bson import ObjectId  # type: ignore

if TYPE_CHECKING:
    from alab_management.builders.experimentbuilder import ExperimentBuilder


[docs] class SampleBuilder: """ Build a sample. Each sample has a name, tags, and metadata. Each sample also has a list of tasks which are binded to it. Each sample is a node in a directed graph of tasks. Each task has a list of samples which are binded to it. Each sample also has a list of tags and metadata. The tags and metadata are used to filter the samples and tasks. The tags and metadata are also used to group the samples and tasks. """ def __init__( self, name: str, experiment: "ExperimentBuilder", tags: list[str] | None = None, **metadata, ): self.name = name self._tasks: list[str] = [] # type: ignore self.experiment = experiment self.metadata = metadata self._id = str(ObjectId()) self.tags = tags or []
[docs] def add_task( self, task_id: str, ) -> None: """ Add a task to the sample. You should use this function only for special cases which are not handled by the `add_sample` function. Args: task_id (str): The object id of the task in mongodb Returns: None. """ if task_id not in self._tasks: self._tasks.append(task_id)
[docs] def to_dict(self) -> dict[str, Any]: """Return Sample as a dictionary. This looks like: .. code-block:: { "_id": str(ObjectId), "name": str, "tags": List[str], "metadata": Dict[str, Any], } Returns ------- Dict[str, Any]: sample as a dictionary """ return { "_id": str(self._id), "name": self.name, "tags": self.tags, "metadata": self.metadata, }
@property def tasks(self): """Return the tasks binded to this sample.""" return self._tasks def __eq__(self, other): """Check if two samples are equal.""" return self._id == other._id def __repr__(self): """Return a string representation of the sample.""" return f"<Sample: {self.name}>"