Comet

zenml.integrations.comet special

Initialization for the Comet integration.

The CometML integrations currently enables you to use Comet tracking as a convenient way to visualize your experiment runs within the Comet ui.

CometIntegration (Integration)

Definition of Comet integration for ZenML.

Source code in zenml/integrations/comet/__init__.py

class CometIntegration(Integration):
    """Definition of Comet integration for ZenML."""

    NAME = COMET
    REQUIREMENTS = ["comet-ml>=3.0.0"]

    @classmethod
    def flavors(cls) -> List[Type[Flavor]]:
        """Declare the stack component flavors for the Comet integration.

        Returns:
            List of stack component flavors for this integration.
        """
        from zenml.integrations.comet.flavors import (
            CometExperimentTrackerFlavor,
        )

        return [CometExperimentTrackerFlavor]

flavors() classmethod

Declare the stack component flavors for the Comet integration.

Returns:

Type	Description
List[Type[zenml.stack.flavor.Flavor]]	List of stack component flavors for this integration.

Source code in zenml/integrations/comet/__init__.py

@classmethod
def flavors(cls) -> List[Type[Flavor]]:
    """Declare the stack component flavors for the Comet integration.

    Returns:
        List of stack component flavors for this integration.
    """
    from zenml.integrations.comet.flavors import (
        CometExperimentTrackerFlavor,
    )

    return [CometExperimentTrackerFlavor]

experiment_trackers special

Initialization for the comet experiment tracker.

comet_experiment_tracker

Implementation for the Comet experiment tracker.

CometExperimentTracker (BaseExperimentTracker)

Track experiment using Comet.

Source code in zenml/integrations/comet/experiment_trackers/comet_experiment_tracker.py

class CometExperimentTracker(BaseExperimentTracker):
    """Track experiment using Comet."""

    @property
    def config(self) -> CometExperimentTrackerConfig:
        """Returns the `CometExperimentTrackerConfig` config.

        Returns:
            The configuration.
        """
        return cast(CometExperimentTrackerConfig, self._config)

    @property
    def settings_class(self) -> Type[CometExperimentTrackerSettings]:
        """Settings class for the Comet experiment tracker.

        Returns:
            The settings class.
        """
        return CometExperimentTrackerSettings

    def prepare_step_run(self, info: "StepRunInfo") -> None:
        """Configures a Comet experiment.

        Args:
            info: Info about the step that will be executed.
        """
        os.environ[COMET_API_KEY] = self.config.api_key
        settings = cast(
            CometExperimentTrackerSettings, self.get_settings(info)
        )
        tags = settings.tags + [info.run_name, info.pipeline.name]
        comet_exp_name = (
            settings.run_name or f"{info.run_name}_{info.pipeline_step_name}"
        )
        self._initialize_comet(
            run_name=comet_exp_name, tags=tags, settings=settings.settings
        )

    def get_step_run_metadata(
        self, info: "StepRunInfo"
    ) -> Dict[str, "MetadataType"]:
        """Get component- and step-specific metadata after a step ran.

        Args:
            info: Info about the step that was executed.

        Returns:
            A dictionary of metadata.
        """
        exp_url: Optional[str] = None
        exp_name: Optional[str] = None

        if self.experiment:
            exp_url = self.experiment.url
            exp_name = self.experiment.name

        # If the URL cannot be retrieved, use the default experiment URL
        default_exp_url = (
            f"https://www.comet.com/{self.config.workspace}/"
            f"{self.config.project_name}/experiments/"
        )
        exp_url = exp_url or default_exp_url

        # If the experiment name cannot be retrieved, use the default name
        default_exp_name = f"{info.run_name}_{info.pipeline_step_name}"
        settings = cast(
            CometExperimentTrackerSettings, self.get_settings(info)
        )
        exp_name = exp_name or settings.run_name or default_exp_name

        return {
            METADATA_EXPERIMENT_TRACKER_URL: Uri(exp_url),
            "comet_experiment_name": exp_name,
        }

    def cleanup_step_run(self, info: "StepRunInfo", step_failed: bool) -> None:
        """Stops the Comet experiment.

        Args:
            info: Info about the step that was executed.
            step_failed: Whether the step failed or not.
        """
        if self.experiment:
            self.experiment.end()
        os.environ.pop(COMET_API_KEY, None)

    def log_metrics(
        self,
        metrics: Dict[str, Any],
        step: Optional[int] = None,
    ) -> None:
        """Logs metrics to the Comet experiment.

        Args:
            metrics: Dictionary of metrics to log.
            step: Optional step number associated with the metrics.
        """
        if self.experiment:
            self.experiment.log_metrics(metrics, step=step)

    def log_params(self, params: Dict[str, Any]) -> None:
        """Logs parameters to the Comet experiment.

        Args:
            params: Dictionary of parameters to log.
        """
        if self.experiment:
            self.experiment.log_parameters(params)

    def _initialize_comet(
        self,
        run_name: str,
        tags: List[str],
        settings: Union[Dict[str, Any], None] = None,
    ) -> None:
        """Initializes a Comet experiment.

        Args:
            run_name: Name of the Comet experiment.
            tags: Tags to attach to the Comet experiment.
            settings: Additional settings for the Comet experiment.
        """
        logger.info(
            f"Initializing Comet with workspace {self.config.workspace}, project "
            f"name: {self.config.project_name}, run_name: {run_name}."
        )
        self.experiment = Experiment(
            workspace=self.config.workspace,
            project_name=self.config.project_name,
            **settings or {},
        )
        self.experiment.set_name(run_name)
        self.experiment.add_tags(tags)

config: CometExperimentTrackerConfig property readonly

Returns the CometExperimentTrackerConfig config.

Returns:

Type	Description
CometExperimentTrackerConfig	The configuration.

settings_class: Type[zenml.integrations.comet.flavors.comet_experiment_tracker_flavor.CometExperimentTrackerSettings] property readonly

Settings class for the Comet experiment tracker.

Returns:

Type	Description
Type[zenml.integrations.comet.flavors.comet_experiment_tracker_flavor.CometExperimentTrackerSettings]	The settings class.

cleanup_step_run(self, info, step_failed)

Stops the Comet experiment.

Parameters:

Name	Type	Description	Default
info	StepRunInfo	Info about the step that was executed.	required
step_failed	bool	Whether the step failed or not.	required

Source code in zenml/integrations/comet/experiment_trackers/comet_experiment_tracker.py

def cleanup_step_run(self, info: "StepRunInfo", step_failed: bool) -> None:
    """Stops the Comet experiment.

    Args:
        info: Info about the step that was executed.
        step_failed: Whether the step failed or not.
    """
    if self.experiment:
        self.experiment.end()
    os.environ.pop(COMET_API_KEY, None)

get_step_run_metadata(self, info)

Get component- and step-specific metadata after a step ran.

Parameters:

Name	Type	Description	Default
info	StepRunInfo	Info about the step that was executed.	required

Returns:

Type	Description
Dict[str, MetadataType]	A dictionary of metadata.

Source code in zenml/integrations/comet/experiment_trackers/comet_experiment_tracker.py

def get_step_run_metadata(
    self, info: "StepRunInfo"
) -> Dict[str, "MetadataType"]:
    """Get component- and step-specific metadata after a step ran.

    Args:
        info: Info about the step that was executed.

    Returns:
        A dictionary of metadata.
    """
    exp_url: Optional[str] = None
    exp_name: Optional[str] = None

    if self.experiment:
        exp_url = self.experiment.url
        exp_name = self.experiment.name

    # If the URL cannot be retrieved, use the default experiment URL
    default_exp_url = (
        f"https://www.comet.com/{self.config.workspace}/"
        f"{self.config.project_name}/experiments/"
    )
    exp_url = exp_url or default_exp_url

    # If the experiment name cannot be retrieved, use the default name
    default_exp_name = f"{info.run_name}_{info.pipeline_step_name}"
    settings = cast(
        CometExperimentTrackerSettings, self.get_settings(info)
    )
    exp_name = exp_name or settings.run_name or default_exp_name

    return {
        METADATA_EXPERIMENT_TRACKER_URL: Uri(exp_url),
        "comet_experiment_name": exp_name,
    }

log_metrics(self, metrics, step=None)

Logs metrics to the Comet experiment.

Parameters:

Name	Type	Description	Default
metrics	Dict[str, Any]	Dictionary of metrics to log.	required
step	Optional[int]	Optional step number associated with the metrics.	None

Source code in zenml/integrations/comet/experiment_trackers/comet_experiment_tracker.py

def log_metrics(
    self,
    metrics: Dict[str, Any],
    step: Optional[int] = None,
) -> None:
    """Logs metrics to the Comet experiment.

    Args:
        metrics: Dictionary of metrics to log.
        step: Optional step number associated with the metrics.
    """
    if self.experiment:
        self.experiment.log_metrics(metrics, step=step)

log_params(self, params)

Logs parameters to the Comet experiment.

Parameters:

Name	Type	Description	Default
params	Dict[str, Any]	Dictionary of parameters to log.	required

Source code in zenml/integrations/comet/experiment_trackers/comet_experiment_tracker.py

def log_params(self, params: Dict[str, Any]) -> None:
    """Logs parameters to the Comet experiment.

    Args:
        params: Dictionary of parameters to log.
    """
    if self.experiment:
        self.experiment.log_parameters(params)

prepare_step_run(self, info)

Configures a Comet experiment.

Parameters:

Name	Type	Description	Default
info	StepRunInfo	Info about the step that will be executed.	required

Source code in zenml/integrations/comet/experiment_trackers/comet_experiment_tracker.py

def prepare_step_run(self, info: "StepRunInfo") -> None:
    """Configures a Comet experiment.

    Args:
        info: Info about the step that will be executed.
    """
    os.environ[COMET_API_KEY] = self.config.api_key
    settings = cast(
        CometExperimentTrackerSettings, self.get_settings(info)
    )
    tags = settings.tags + [info.run_name, info.pipeline.name]
    comet_exp_name = (
        settings.run_name or f"{info.run_name}_{info.pipeline_step_name}"
    )
    self._initialize_comet(
        run_name=comet_exp_name, tags=tags, settings=settings.settings
    )

flavors special

Comet integration flavors.

comet_experiment_tracker_flavor

Comet experiment tracker flavor.

CometExperimentTrackerConfig (BaseExperimentTrackerConfig, CometExperimentTrackerSettings)

Config for the Comet experiment tracker.

Attributes:

Name	Type	Description
workspace	Optional[str]	Name of an existing Comet workspace.
project_name	Optional[str]	Name of an existing Comet project to log to.
api_key	str	API key that should be authorized to log to the configured Comet workspace and project.

Source code in zenml/integrations/comet/flavors/comet_experiment_tracker_flavor.py

class CometExperimentTrackerConfig(
    BaseExperimentTrackerConfig, CometExperimentTrackerSettings
):
    """Config for the Comet experiment tracker.

    Attributes:
        workspace: Name of an existing Comet workspace.
        project_name: Name of an existing Comet project to log to.
        api_key: API key that should be authorized to log to the configured Comet
            workspace and project.
    """

    api_key: str = SecretField()
    workspace: Optional[str] = None
    project_name: Optional[str] = None

CometExperimentTrackerFlavor (BaseExperimentTrackerFlavor)

Flavor for the Comet experiment tracker.

Source code in zenml/integrations/comet/flavors/comet_experiment_tracker_flavor.py

class CometExperimentTrackerFlavor(BaseExperimentTrackerFlavor):
    """Flavor for the Comet experiment tracker."""

    @property
    def name(self) -> str:
        """Name of the flavor.

        Returns:
            The name of the flavor.
        """
        return COMET_EXPERIMENT_TRACKER_FLAVOR

    @property
    def docs_url(self) -> Optional[str]:
        """A url to point at docs explaining this flavor.

        Returns:
            A flavor docs url.
        """
        return self.generate_default_docs_url()

    @property
    def sdk_docs_url(self) -> Optional[str]:
        """A url to point at SDK docs explaining this flavor.

        Returns:
            A flavor SDK docs url.
        """
        return self.generate_default_sdk_docs_url()

    @property
    def logo_url(self) -> str:
        """A url to represent the flavor in the dashboard.

        Returns:
            The flavor logo.
        """
        return "https://public-flavor-logos.s3.eu-central-1.amazonaws.com/experiment_tracker/comet.png"

    @property
    def config_class(self) -> Type[CometExperimentTrackerConfig]:
        """Returns `CometExperimentTrackerConfig` config class.

        Returns:
                The config class.
        """
        return CometExperimentTrackerConfig

    @property
    def implementation_class(self) -> Type["CometExperimentTracker"]:
        """Implementation class for this flavor.

        Returns:
            The implementation class.
        """
        from zenml.integrations.comet.experiment_trackers import (
            CometExperimentTracker,
        )

        return CometExperimentTracker

config_class: Type[zenml.integrations.comet.flavors.comet_experiment_tracker_flavor.CometExperimentTrackerConfig] property readonly

Returns CometExperimentTrackerConfig config class.

Returns:

Type	Description
Type[zenml.integrations.comet.flavors.comet_experiment_tracker_flavor.CometExperimentTrackerConfig]	The config class.

docs_url: Optional[str] property readonly

A url to point at docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor docs url.

implementation_class: Type[CometExperimentTracker] property readonly

Implementation class for this flavor.

Returns:

Type	Description
Type[CometExperimentTracker]	The implementation class.

logo_url: str property readonly

A url to represent the flavor in the dashboard.

Returns:

Type	Description
str	The flavor logo.

name: str property readonly

Name of the flavor.

Returns:

Type	Description
str	The name of the flavor.

sdk_docs_url: Optional[str] property readonly

A url to point at SDK docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor SDK docs url.

CometExperimentTrackerSettings (BaseSettings)

Settings for the Comet experiment tracker.

Attributes:

Name	Type	Description
run_name	Optional[str]	The Comet experiment name.
tags	List[str]	Tags for the Comet experiment.
settings	Dict[str, Any]	Settings for the Comet experiment.

Source code in zenml/integrations/comet/flavors/comet_experiment_tracker_flavor.py

class CometExperimentTrackerSettings(BaseSettings):
    """Settings for the Comet experiment tracker.

    Attributes:
        run_name: The Comet experiment name.
        tags: Tags for the Comet experiment.
        settings: Settings for the Comet experiment.
    """

    run_name: Optional[str] = None
    tags: List[str] = []
    settings: Dict[str, Any] = {}