RunCard

RunCards are used to store metrics and artifacts related to DataCards and ModelCards. While a RunCard can be used as a object itself, it's best when used as part of a Project run.

Creating A Run

Runs are unique context-managed executions associated with a Project that record all created cards and their associated metrics, params, and artifacts to a single card called a RunCard.

The following example shows how to create a simple run as well as use CardInfo to store helper info

from sklearn.linear_model import LinearRegression

from opsml import (
    CardInfo,
    DataCard,
    DataSplit,
    ModelCard,
    OpsmlProject,
    PandasData,
    ProjectInfo,
    SklearnModel,
)
from opsml.helpers.data import create_fake_data

info = ProjectInfo(name="opsml-project", repository="opsml", contact="user@email.com")

# create card info and set NAME, REPOSITORY, and CONTACT as environment variables
card_info = CardInfo(name="linear-reg", repository="opsml", contact="user@email.com").set_env()

# create project
project = OpsmlProject(info=info)

with project.run() as run:
    # create fake data
    X, y = create_fake_data(n_samples=1000, task_type="regression")
    X["target"] = y

    # Create data interface
    data_interface = PandasData(
        data=X,
        data_splits=[
            DataSplit(label="train", column_name="col_1", column_value=0.5, inequality=">="),
            DataSplit(label="test", column_name="col_1", column_value=0.5, inequality="<"),
        ],
        dependent_vars=["target"],
    )

    # Create datacard
    datacard = DataCard(interface=data_interface)
    run.register_card(card=datacard)

    # split data
    data = datacard.split_data()

    # fit model
    reg = LinearRegression()
    reg.fit(data["train"].X.to_numpy(), data["train"].y.to_numpy())

    # create model interface
    interface = SklearnModel(model=reg, sample_data=data["train"].X.to_numpy())

    # create modelcard
    modelcard = ModelCard(interface=interface, to_onnx=True, datacard_uid=datacard.uid)

    # you can log metrics view log_metric or log_metrics
    run.log_metric("test_metric", 10)
    run.log_metrics({"test_metric2": 20})

    # log parameter
    run.log_parameter("test_parameter", 10)

    # register modelcard
    run.register_card(card=modelcard)

    # example of logging artifact to file
    with Path("artifact.txt").open("w") as f:
        f.write("This is a test")

    run.log_artifact_from_file("artifact", "artifact.txt")

You can now log into the OpsML server and see your recent run and associated metadata

Logging Hardware

Runs can also log hardware information by simply changing 1 input argument.

• Instead of with project.run() as run: use with project.run(log_hardware=True) as run: and now hardware information will be logged. For information on what hardware information is logged, see the documentation below.

• If you want to change the hardware logging time interval, simply change the hardware_interval in the run method with project.run(log_hardware=True, hardware_interval=10) as run:. Note: the hardware_interval is in seconds and the lowest value is 10 seconds. Default is 30 seconds.

opsml.RunCard

Bases: ArtifactCard

Create a RunCard from specified arguments.

Apart from required args, a RunCard must be associated with one of datacard_uid, modelcard_uids or pipelinecard_uid

Parameters:

Name	Type	Description	Default
name		Run name	required
repository		Repository that this card is associated with	required
contact		Contact to associate with card	required
info		CardInfo object containing additional metadata. If provided, it will override any values provided for name, repository, contact, and version. Name, repository, and contact are required arguments for all cards. They can be provided directly or through a CardInfo object.	required
datacard_uids		Optional DataCard uids associated with this run	required
modelcard_uids		Optional List of ModelCard uids to associate with this run	required
pipelinecard_uid		Optional PipelineCard uid to associate with this experiment	required
metrics		Optional dictionary of key (str), value (int, float) metric paris. Metrics can also be added via class methods.	required
parameters		Parameters associated with a RunCard	required
artifact_uris		Optional dictionary of artifact uris associated with artifacts.	required
uid		Unique id (assigned if card has been registered)	required
version		Current version (assigned if card has been registered)	required

Source code in opsml/cards/run.py

class RunCard(ArtifactCard):
    """
    Create a RunCard from specified arguments.

    Apart from required args, a RunCard must be associated with one of
    datacard_uid, modelcard_uids or pipelinecard_uid

    Args:
        name:
            Run name
        repository:
            Repository that this card is associated with
        contact:
            Contact to associate with card
        info:
            `CardInfo` object containing additional metadata. If provided, it will override any
            values provided for `name`, `repository`, `contact`, and `version`.

            Name, repository, and contact are required arguments for all cards. They can be provided
            directly or through a `CardInfo` object.

        datacard_uids:
            Optional DataCard uids associated with this run
        modelcard_uids:
            Optional List of ModelCard uids to associate with this run
        pipelinecard_uid:
            Optional PipelineCard uid to associate with this experiment
        metrics:
            Optional dictionary of key (str), value (int, float) metric paris.
            Metrics can also be added via class methods.
        parameters:
            Parameters associated with a RunCard
        artifact_uris:
            Optional dictionary of artifact uris associated with artifacts.
        uid:
            Unique id (assigned if card has been registered)
        version:
            Current version (assigned if card has been registered)

    """

    model_config = ConfigDict(extra="forbid")

    datacard_uids: List[str] = []
    modelcard_uids: List[str] = []
    pipelinecard_uid: Optional[str] = None
    metrics: Metrics = {}
    parameters: Params = {}
    artifact_uris: ArtifactUris = {}
    tags: Dict[str, Union[str, int]] = {}
    compute_environment: ComputeEnvironment = ComputeEnvironment()
    project: Optional[str] = None

    @model_validator(mode="before")
    @classmethod
    def validate_defaults_args(cls, card_args: Dict[str, Any]) -> Dict[str, Any]:
        # add default
        contact = card_args.get("contact")

        if contact is None:
            card_args["contact"] = CommonKwargs.UNDEFINED.value

        repository = card_args.get("repository")

        if repository is None:
            card_args["repository"] = "opsml"

        return card_args

    def add_tag(self, key: str, value: str) -> None:
        """
        Logs tags to current RunCard

        Args:
            key:
                Key for tag
            value:
                value for tag
        """
        self.tags = {**{key: value}, **self.tags}

    def add_tags(self, tags: Dict[str, str]) -> None:
        """
        Logs tags to current RunCard

        Args:
            tags:
                Dictionary of tags
        """
        self.tags = {**tags, **self.tags}

    def log_graph(
        self,
        name: str,
        x: Union[List[int], List[float], NDArray[Any]],
        y: Union[List[int], List[float], NDArray[Any], Dict[str, Union[List[int], List[float], NDArray[Any]]]],
        y_label: str,
        x_label: str,
        graph_style: str,
    ) -> None:
        """Logs a graph to the RunCard, which will be rendered in the UI as a line graph

        Args:
            name:
                Name of graph
            x:
                List or numpy array of x values

            x_label:
                Label for x axis
            y:
                Either a list or numpy array of y values or a dictionary of y values where key is the group label and
                value is a list or numpy array of y values
            y_label:
                Label for y axis
            graph_style:
                Style of graph. Options are "line" or "scatter"

        example:

            ### single line graph
            x = np.arange(1, 400, 0.5)
            y = x * x
            run.log_graph(name="graph1", x=x, y=y, x_label="x", y_label="y", graph_style="line")

            ### multi line graph
            x = np.arange(1, 1000, 0.5)
            y1 = x * x
            y2 = y1 * 1.1
            y3 = y2 * 3
            run.log_graph(
                name="multiline",
                x=x,
                y={"y1": y1, "y2": y2, "y3": y3},
                x_label="x",
                y_label="y",
                graph_style="line",
            )

        """

        if isinstance(x, np.ndarray):
            x = x.flatten().tolist()
            assert isinstance(x, list), "x must be a list or dictionary"

        x = _decimate_list(x)

        parsed_y, graph_type = _parse_y_to_list(len(x), y)

        logger.info(f"Logging graph {name} to RunCard")
        graph = RunGraph(
            name=name,
            x=x,
            x_label=x_label,
            y=parsed_y,
            y_label=y_label,
            graph_type=graph_type,
            graph_style=GraphStyle.from_str(graph_style).value,  # validate graph style
        )

        # save graph to storage so we can view in ui while run is active
        lpath, rpath = _dump_graph_artifact(graph, name, self.uri)

        self._add_artifact_uri(
            name=name,
            local_path=lpath.as_posix(),
            remote_path=rpath.as_posix(),
        )

    def log_parameters(self, parameters: Dict[str, Union[float, int, str]]) -> None:
        """
        Logs parameters to current RunCard

        Args:
            parameters:
                Dictionary of parameters
        """

        for key, value in parameters.items():
            # check key
            self.log_parameter(key, value)

    def log_parameter(self, key: str, value: Union[int, float, str]) -> None:
        """
        Logs parameter to current RunCard

        Args:
            key:
                Param name
            value:
                Param value
        """

        TypeChecker.check_param_type(param=value)
        _key = TypeChecker.replace_spaces(key)

        param = Param(name=key, value=value)

        self._registry.insert_parameter([{**param.model_dump(), **{"run_uid": self.uid}}])

        if self.parameters.get(_key) is not None:
            self.parameters[_key].append(param)

        else:
            self.parameters[_key] = [param]

    def log_metric(
        self,
        key: str,
        value: Union[int, float],
        timestamp: Optional[int] = None,
        step: Optional[int] = None,
    ) -> None:
        """
        Logs metric to the existing RunCard metric dictionary

        Args:
            key:
                Metric name
            value:
                Metric value
            timestamp:
                Optional timestamp
            step:
                Optional step associated with name and value
        """

        TypeChecker.check_metric_type(metric=value)
        _key = TypeChecker.replace_spaces(key)

        metric = Metric(name=_key, value=value, timestamp=timestamp, step=step)

        self._registry.insert_metric([{**metric.model_dump(), **{"run_uid": self.uid}}])

        if self.metrics.get(_key) is not None:
            self.metrics[_key].append(metric)
        else:
            self.metrics[_key] = [metric]

    def log_metrics(self, metrics: Dict[str, Union[float, int]], step: Optional[int] = None) -> None:
        """
        Log metrics to the existing RunCard metric dictionary

        Args:
            metrics:
                Dictionary containing key (str) and value (float or int) pairs
                to add to the current metric set
            step:
                Optional step associated with metrics
        """

        for key, value in metrics.items():
            self.log_metric(key=key, value=value, step=step)

    def log_artifact_from_file(
        self,
        name: str,
        local_path: Union[str, Path],
        artifact_path: Optional[Union[str, Path]] = None,
    ) -> None:
        """
        Log a local file or directory to the opsml server and associate with the current run.

        Args:
            name:
                Name to assign to artifact(s)
            local_path:
                Local path to file or directory. Can be string or pathlike object
            artifact_path:
                Optional path to store artifact in opsml server. If not provided, 'artifacts' will be used
        """

        lpath = Path(local_path)
        rpath = self.uri / (artifact_path or SaveName.ARTIFACTS.value)

        if lpath.is_file():
            rpath = rpath / lpath.name

        client.storage_client.put(lpath, rpath)
        self._add_artifact_uri(
            name=name,
            local_path=lpath.as_posix(),
            remote_path=rpath.as_posix(),
        )

    def create_registry_record(self) -> Dict[str, Any]:
        """Creates a registry record from the current RunCard"""

        exclude_attr = {"parameters", "metrics"}

        return self.model_dump(exclude=exclude_attr)

    def _add_artifact_uri(self, name: str, local_path: str, remote_path: str) -> None:
        """
        Adds an artifact_uri to the runcard

        Args:
            name:
                Name to associate with artifact
            uri:
                Uri where artifact is stored
        """

        self.artifact_uris[name] = Artifact(
            name=name,
            local_path=local_path,
            remote_path=remote_path,
        )

    def add_card_uid(self, card_type: str, uid: str) -> None:
        """
        Adds a card uid to the appropriate card uid list for tracking

        Args:
            card_type:
                Card class name
            uid:
                Uid of registered Card
        """

        if card_type == CardType.DATACARD:
            self.datacard_uids = [uid, *self.datacard_uids]
        elif card_type == CardType.MODELCARD:
            self.modelcard_uids = [uid, *self.modelcard_uids]

    def get_metric(self, name: str) -> List[Metric]:
        """
        Gets a metric by name

        Args:
            name:
                Name of metric

        Returns:
            List of dictionaries or dictionary containing value

        """
        _key = TypeChecker.replace_spaces(name)

        metric = self.metrics.get(_key)

        if metric is None:
            # try to get metric from registry
            assert self.uid is not None, "RunCard must be registered to get metric"
            _metric = self._registry.get_metric(run_uid=self.uid, name=[_key])

            if len(_metric) > 0:
                metric = [Metric(**i) for i in _metric]

            else:
                return cast(List[Metric], [])

        return metric

    def load_metrics(self) -> None:
        """Reloads metrics from registry"""
        assert self.uid is not None, "RunCard must be registered to load metrics"

        metrics = self._registry.get_metric(run_uid=self.uid)

        if metrics is None:
            logger.info("No metrics found for RunCard")
            return None

        # reset metrics
        self.metrics = {}
        for metric in metrics:
            _metric = Metric(**metric)
            if _metric.name not in self.metrics:
                self.metrics[_metric.name] = [_metric]
            else:
                self.metrics[_metric.name].append(_metric)
        return None

    def get_hardware_metrics(self) -> Optional[List[Dict[str, Any]]]:
        """Returns hardware metrics recorded during run.

        Returns:
            List of dictionaries containing hardware metrics
        """
        assert self.uid is not None, "RunCard must be registered to get hardware metrics"
        return self._registry.get_hw_metric(run_uid=self.uid)

    def get_parameter(self, name: str) -> List[Param]:
        """
        Gets a parameter by name

        Args:
            name:
                Name of parameter

        Returns:
            List of dictionaries or dictionary containing value

        """
        _key = TypeChecker.replace_spaces(name)
        param = self.parameters.get(_key)

        if param is None:
            # try to get metric from registry
            assert self.uid is not None, "RunCard must be registered to get metric"
            _param = self._registry.get_parameter(run_uid=self.uid, name=[_key])

            if len(_param) > 0:
                param = [Param(**i) for i in _param]

            else:
                return cast(List[Param], [])

        return param

    def load_artifacts(self, name: Optional[str] = None) -> None:
        """Loads artifacts from artifact_uris"""
        if bool(self.artifact_uris) is False:
            logger.info("No artifact uris associated with RunCard")
            return None

        if name is not None:
            artifact = self.artifact_uris.get(name)
            assert artifact is not None, f"Artifact {name} not found"
            client.storage_client.get(
                Path(artifact.remote_path),
                Path(artifact.local_path),
            )

        else:
            for _, artifact in self.artifact_uris.items():
                client.storage_client.get(
                    Path(artifact.remote_path),
                    Path(artifact.local_path),
                )
        return None

    @property
    def uri(self) -> Path:
        """The base URI to use for the card and it's artifacts."""

        # when using runcard outside of run context
        if self.version == CommonKwargs.BASE_VERSION.value:
            if self.uid is None:
                self.uid = uuid.uuid4().hex

            end_path = self.uid
        else:
            end_path = f"v{self.version}"

        return Path(
            config.storage_root,
            RegistryTableNames.from_str(self.card_type).value,
            str(self.repository),
            str(self.name),
            end_path,
        )

    @cached_property
    def _registry(self) -> RunCardRegistry:
        from opsml.registry.backend import _set_registry

        return cast(RunCardRegistry, _set_registry(RegistryType.RUN))

    @property
    def card_type(self) -> str:
        return CardType.RUNCARD.value

add_tag(key, value)

Logs tags to current RunCard

Parameters:

Name	Type	Description	Default
key	str	Key for tag	required
value	str	value for tag	required

Source code in opsml/cards/run.py

def add_tag(self, key: str, value: str) -> None:
    """
    Logs tags to current RunCard

    Args:
        key:
            Key for tag
        value:
            value for tag
    """
    self.tags = {**{key: value}, **self.tags}

add_tags(tags)

Logs tags to current RunCard

Parameters:

Name	Type	Description	Default
tags	Dict[str, str]	Dictionary of tags	required

Source code in opsml/cards/run.py

def add_tags(self, tags: Dict[str, str]) -> None:
    """
    Logs tags to current RunCard

    Args:
        tags:
            Dictionary of tags
    """
    self.tags = {**tags, **self.tags}

log_parameter(key, value)

Logs parameter to current RunCard

Parameters:

Name	Type	Description	Default
key	str	Param name	required
value	Union[int, float, str]	Param value	required

Source code in opsml/cards/run.py

def log_parameter(self, key: str, value: Union[int, float, str]) -> None:
    """
    Logs parameter to current RunCard

    Args:
        key:
            Param name
        value:
            Param value
    """

    TypeChecker.check_param_type(param=value)
    _key = TypeChecker.replace_spaces(key)

    param = Param(name=key, value=value)

    self._registry.insert_parameter([{**param.model_dump(), **{"run_uid": self.uid}}])

    if self.parameters.get(_key) is not None:
        self.parameters[_key].append(param)

    else:
        self.parameters[_key] = [param]

log_parameters(parameters)

Logs parameters to current RunCard

Parameters:

Name	Type	Description	Default
parameters	Dict[str, Union[float, int, str]]	Dictionary of parameters	required

Source code in opsml/cards/run.py

def log_parameters(self, parameters: Dict[str, Union[float, int, str]]) -> None:
    """
    Logs parameters to current RunCard

    Args:
        parameters:
            Dictionary of parameters
    """

    for key, value in parameters.items():
        # check key
        self.log_parameter(key, value)

log_metric(key, value, timestamp=None, step=None)

Logs metric to the existing RunCard metric dictionary

Parameters:

Name	Type	Description	Default
key	str	Metric name	required
value	Union[int, float]	Metric value	required
timestamp	Optional[int]	Optional timestamp	None
step	Optional[int]	Optional step associated with name and value	None

Source code in opsml/cards/run.py

def log_metric(
    self,
    key: str,
    value: Union[int, float],
    timestamp: Optional[int] = None,
    step: Optional[int] = None,
) -> None:
    """
    Logs metric to the existing RunCard metric dictionary

    Args:
        key:
            Metric name
        value:
            Metric value
        timestamp:
            Optional timestamp
        step:
            Optional step associated with name and value
    """

    TypeChecker.check_metric_type(metric=value)
    _key = TypeChecker.replace_spaces(key)

    metric = Metric(name=_key, value=value, timestamp=timestamp, step=step)

    self._registry.insert_metric([{**metric.model_dump(), **{"run_uid": self.uid}}])

    if self.metrics.get(_key) is not None:
        self.metrics[_key].append(metric)
    else:
        self.metrics[_key] = [metric]

log_metrics(metrics, step=None)

Log metrics to the existing RunCard metric dictionary

Parameters:

Name	Type	Description	Default
metrics	Dict[str, Union[float, int]]	Dictionary containing key (str) and value (float or int) pairs to add to the current metric set	required
step	Optional[int]	Optional step associated with metrics	None

Source code in opsml/cards/run.py

def log_metrics(self, metrics: Dict[str, Union[float, int]], step: Optional[int] = None) -> None:
    """
    Log metrics to the existing RunCard metric dictionary

    Args:
        metrics:
            Dictionary containing key (str) and value (float or int) pairs
            to add to the current metric set
        step:
            Optional step associated with metrics
    """

    for key, value in metrics.items():
        self.log_metric(key=key, value=value, step=step)

opsml.projects.OpsmlProject

Source code in opsml/projects/project.py

class OpsmlProject:
    def __init__(self, info: ProjectInfo):
        """
        Instantiates a project which creates cards, metrics and parameters to
        the opsml registry via a "run" object.

        If info.run_id is set, that run_id will be loaded as read only. In read
        only mode, you can retrieve cards, metrics, and parameters, however you
        cannot write new data. If you wish to record data/create a new run, you will
        need to enter the run context.

        In order to create new cards, you need to create a run using the `run`
        context manager.

        Example:

            project: OpsmlProject = OpsmlProject(
                ProjectInfo(
                    name="test-project",
                    # If run_id is omitted, a new run is created.
                    run_id="123ab123kaj8u8naskdfh813",
                )
            )
            # the project is in "read only" mode. all read operations will work
            for k, v in project.parameters:
                logger.info("{} = {}", k, v)

            # creating a project run
            with project.run() as run:
                # Now that the run context is entered, it's in read/write mode
                # You can write cards, parameters, and metrics to the project.
                run.log_parameter(key="my_param", value="12.34")

        Args:
            info:
                Run information. if a run_id is given, that run is set
                as the project's current run.
        """
        # Set the run manager and project_id (creates ProjectCard if project doesn't exist)
        registrar = _ProjectRegistrar(project_info=info)

        # get project id or register new project
        info.project_id = registrar.register_project()

        # crete run manager
        self._run_mgr = _RunManager(project_info=info, registries=registrar.registries)

    @property
    def run_id(self) -> str:
        """Current run id associated with project"""
        if self._run_mgr.run_id is not None:
            return self._run_mgr.run_id
        raise ValueError("Run id not set for current project")

    @run_id.setter
    def run_id(self, run_id: str) -> None:
        """Set the run_id to use with the active project"""
        self._run_mgr.run_id = run_id

    @property
    def project_id(self) -> int:
        return self._run_mgr.project_id

    @property
    def project_name(self) -> str:
        return self._run_mgr._project_info.name  # pylint: disable=protected-access

    @contextmanager
    def run(
        self,
        run_name: Optional[str] = None,
        log_hardware: bool = False,
        hardware_interval: int = _DEFAULT_INTERVAL,
        code_dir: Optional[Union[str, Path]] = None,
    ) -> Iterator[ActiveRun]:
        """
        Starts a new run for the project

        Args:
            run_name:
                Optional run name
            log_hardware:
                Whether to log hardware metrics
            hardware_interval:
                Interval to log hardware metrics. Default is 30 seconds.
            code_dir:
                Top-level directory containing code to be logged. If not provided,
                the directory containing the current file will be used.
        """

        try:
            # get filename
            # need to back out of project.py and contextlib.py
            filename = Path(inspect.getframeinfo(inspect.currentframe().f_back.f_back).filename)  # type: ignore
            yield self._run_mgr.start_run(
                run_name=run_name,
                log_hardware=log_hardware,
                hardware_interval=hardware_interval,
                code_dir=code_dir,
                filename=filename,
            )

        except ActiveRunException as error:
            logger.error("Run already active. Ending run.")
            raise error

        except Exception as error:
            logger.error("Error encountered. Ending run. {}", error)
            self._run_mgr.end_run()
            raise error

        self._run_mgr.end_run()

    def load_card(self, registry_name: str, info: CardInfo) -> Card:
        """
        Loads a Card.

        Args:
            registry_name:
                Name of registry to load card from
            info:
                Card information to retrieve. `uid` takes precedence if it
                exists. If the optional `version` is specified, that version
                will be loaded. If it doesn't exist, the most recent ersion will
                be loaded.

        Returns
            `Card`
        """
        card_type = CardType(registry_name.lower()).value
        return CardHandler.load_card(
            registries=self._run_mgr.registries,
            registry_name=card_type,
            info=info,
        )

    def list_runs(self, limit: int = 100) -> List[Dict[str, Any]]:
        """
        Lists all runs for the current project, sorted by timestamp

        Returns:
            List of RunCard
        """
        logger.info("Listing runs for project {}", self.project_name)

        project_runs = self._run_mgr.registries.run._registry.list_cards(  # pylint: disable=protected-access
            limit=limit,
            query_terms={"project": self.project_name},
        )

        return sorted(project_runs, key=lambda k: k["timestamp"], reverse=True)

    @property
    def runcard(self) -> RunCard:
        return cast(RunCard, self._run_mgr.registries.run.load_card(uid=self.run_id))

    @property
    def metrics(self) -> Metrics:
        runcard = self.runcard
        runcard.load_metrics()
        return runcard.metrics

    def get_metric(self, name: str) -> List[Metric]:
        """
        Get metric by name

        Args:
            name: str

        Returns:
            List of Metric or Metric

        """
        return self.runcard.get_metric(name=name)

    @property
    def parameters(self) -> Params:
        return self.runcard.parameters

    def get_parameter(self, name: str) -> List[Param]:
        """
        Get param by name

        Args:
            name: str

        Returns:
            List of Param or Param

        """
        return self.runcard.get_parameter(name=name)

    @property
    def tags(self) -> Dict[str, Union[str, int]]:
        return self.runcard.tags

    @property
    def datacard_uids(self) -> List[str]:
        """DataCards associated with the current run"""
        return self.runcard.datacard_uids

    @property
    def modelcard_uids(self) -> List[str]:
        """ModelCards associated with the current run"""
        return self.runcard.modelcard_uids

datacard_uids: List[str] property

DataCards associated with the current run

modelcard_uids: List[str] property

ModelCards associated with the current run

run_id: str property writable

Current run id associated with project

__init__(info)

Instantiates a project which creates cards, metrics and parameters to the opsml registry via a "run" object.

If info.run_id is set, that run_id will be loaded as read only. In read only mode, you can retrieve cards, metrics, and parameters, however you cannot write new data. If you wish to record data/create a new run, you will need to enter the run context.

In order to create new cards, you need to create a run using the run context manager.

Example:

project: OpsmlProject = OpsmlProject(
    ProjectInfo(
        name="test-project",
        # If run_id is omitted, a new run is created.
        run_id="123ab123kaj8u8naskdfh813",
    )
)
# the project is in "read only" mode. all read operations will work
for k, v in project.parameters:
    logger.info("{} = {}", k, v)

# creating a project run
with project.run() as run:
    # Now that the run context is entered, it's in read/write mode
    # You can write cards, parameters, and metrics to the project.
    run.log_parameter(key="my_param", value="12.34")

Parameters:

Name	Type	Description	Default
info	ProjectInfo	Run information. if a run_id is given, that run is set as the project's current run.	required

Source code in opsml/projects/project.py

def __init__(self, info: ProjectInfo):
    """
    Instantiates a project which creates cards, metrics and parameters to
    the opsml registry via a "run" object.

    If info.run_id is set, that run_id will be loaded as read only. In read
    only mode, you can retrieve cards, metrics, and parameters, however you
    cannot write new data. If you wish to record data/create a new run, you will
    need to enter the run context.

    In order to create new cards, you need to create a run using the `run`
    context manager.

    Example:

        project: OpsmlProject = OpsmlProject(
            ProjectInfo(
                name="test-project",
                # If run_id is omitted, a new run is created.
                run_id="123ab123kaj8u8naskdfh813",
            )
        )
        # the project is in "read only" mode. all read operations will work
        for k, v in project.parameters:
            logger.info("{} = {}", k, v)

        # creating a project run
        with project.run() as run:
            # Now that the run context is entered, it's in read/write mode
            # You can write cards, parameters, and metrics to the project.
            run.log_parameter(key="my_param", value="12.34")

    Args:
        info:
            Run information. if a run_id is given, that run is set
            as the project's current run.
    """
    # Set the run manager and project_id (creates ProjectCard if project doesn't exist)
    registrar = _ProjectRegistrar(project_info=info)

    # get project id or register new project
    info.project_id = registrar.register_project()

    # crete run manager
    self._run_mgr = _RunManager(project_info=info, registries=registrar.registries)

get_metric(name)

Get metric by name

Parameters:

Name	Type	Description	Default
name	str	str	required

Returns:

Type	Description
List[Metric]	List of Metric or Metric

Source code in opsml/projects/project.py

def get_metric(self, name: str) -> List[Metric]:
    """
    Get metric by name

    Args:
        name: str

    Returns:
        List of Metric or Metric

    """
    return self.runcard.get_metric(name=name)

get_parameter(name)

Get param by name

Parameters:

Name	Type	Description	Default
name	str	str	required

Returns:

Type	Description
List[Param]	List of Param or Param

Source code in opsml/projects/project.py

def get_parameter(self, name: str) -> List[Param]:
    """
    Get param by name

    Args:
        name: str

    Returns:
        List of Param or Param

    """
    return self.runcard.get_parameter(name=name)

list_runs(limit=100)

Lists all runs for the current project, sorted by timestamp

Returns:

Type	Description
List[Dict[str, Any]]	List of RunCard

Source code in opsml/projects/project.py

def list_runs(self, limit: int = 100) -> List[Dict[str, Any]]:
    """
    Lists all runs for the current project, sorted by timestamp

    Returns:
        List of RunCard
    """
    logger.info("Listing runs for project {}", self.project_name)

    project_runs = self._run_mgr.registries.run._registry.list_cards(  # pylint: disable=protected-access
        limit=limit,
        query_terms={"project": self.project_name},
    )

    return sorted(project_runs, key=lambda k: k["timestamp"], reverse=True)

load_card(registry_name, info)

Loads a Card.

Parameters:

Name	Type	Description	Default
registry_name	str	Name of registry to load card from	required
info	CardInfo	Card information to retrieve. uid takes precedence if it exists. If the optional version is specified, that version will be loaded. If it doesn't exist, the most recent ersion will be loaded.	required

Returns Card

Source code in opsml/projects/project.py

def load_card(self, registry_name: str, info: CardInfo) -> Card:
    """
    Loads a Card.

    Args:
        registry_name:
            Name of registry to load card from
        info:
            Card information to retrieve. `uid` takes precedence if it
            exists. If the optional `version` is specified, that version
            will be loaded. If it doesn't exist, the most recent ersion will
            be loaded.

    Returns
        `Card`
    """
    card_type = CardType(registry_name.lower()).value
    return CardHandler.load_card(
        registries=self._run_mgr.registries,
        registry_name=card_type,
        info=info,
    )

run(run_name=None, log_hardware=False, hardware_interval=_DEFAULT_INTERVAL, code_dir=None)

Starts a new run for the project

Parameters:

Name	Type	Description	Default
run_name	Optional[str]	Optional run name	None
log_hardware	bool	Whether to log hardware metrics	False
hardware_interval	int	Interval to log hardware metrics. Default is 30 seconds.	_DEFAULT_INTERVAL
code_dir	Optional[Union[str, Path]]	Top-level directory containing code to be logged. If not provided, the directory containing the current file will be used.	None

Source code in opsml/projects/project.py

@contextmanager
def run(
    self,
    run_name: Optional[str] = None,
    log_hardware: bool = False,
    hardware_interval: int = _DEFAULT_INTERVAL,
    code_dir: Optional[Union[str, Path]] = None,
) -> Iterator[ActiveRun]:
    """
    Starts a new run for the project

    Args:
        run_name:
            Optional run name
        log_hardware:
            Whether to log hardware metrics
        hardware_interval:
            Interval to log hardware metrics. Default is 30 seconds.
        code_dir:
            Top-level directory containing code to be logged. If not provided,
            the directory containing the current file will be used.
    """

    try:
        # get filename
        # need to back out of project.py and contextlib.py
        filename = Path(inspect.getframeinfo(inspect.currentframe().f_back.f_back).filename)  # type: ignore
        yield self._run_mgr.start_run(
            run_name=run_name,
            log_hardware=log_hardware,
            hardware_interval=hardware_interval,
            code_dir=code_dir,
            filename=filename,
        )

    except ActiveRunException as error:
        logger.error("Run already active. Ending run.")
        raise error

    except Exception as error:
        logger.error("Error encountered. Ending run. {}", error)
        self._run_mgr.end_run()
        raise error

    self._run_mgr.end_run()

opsml.types.hardware.HardwareMetrics

Bases: BaseModel

Source code in opsml/types/hardware.py

class HardwareMetrics(BaseModel):
    cpu: CPUMetrics
    memory: MemoryMetrics
    network: NetworkRates
    gpu: Optional[GPUMetrics] = None

opsml.types.hardware.CPUMetrics

Bases: BaseModel

CPU metrics data model.

Source code in opsml/types/hardware.py

class CPUMetrics(BaseModel):
    """CPU metrics data model."""

    cpu_percent_utilization: float = 0.0
    cpu_percent_per_core: Optional[List[float]] = None
    compute_overall: Optional[float] = None
    compute_utilized: Optional[float] = None
    load_avg: float

opsml.types.hardware.MemoryMetrics

Bases: BaseModel

Memory metrics data model.

Source code in opsml/types/hardware.py

class MemoryMetrics(BaseModel):
    """Memory metrics data model."""

    sys_ram_total: int = 0
    sys_ram_used: int = 0
    sys_ram_available: int = 0
    sys_ram_percent_used: float = 0.0
    sys_swap_total: Optional[int] = None
    sys_swap_used: Optional[int] = None
    sys_swap_free: Optional[int] = None
    sys_swap_percent: Optional[float] = None

opsml.types.hardware.NetworkRates

Bases: BaseModel

Network rates data model.

Source code in opsml/types/hardware.py

class NetworkRates(BaseModel):
    """Network rates data model."""

    bytes_recv: float = 0.0
    bytes_sent: float = 0.0

opsml.types.hardware.GPUMetrics

Bases: BaseModel

GPU metrics data model.

Source code in opsml/types/hardware.py

class GPUMetrics(BaseModel):
    """GPU metrics data model."""

    gpu_percent_utilization: float = 0.0
    gpu_percent_per_core: Optional[List[float]] = None