Source code for src.internal.experiment_connector.connector
"""Provides a base class for all experiment connectors to inherit from.""" ""
from abc import ABC, abstractmethod
from logging import Logger
from typing import Dict, List, Optional
from ...models.model import Artifact
[docs]class ExperimentConnector(ABC):
"""Base class for experiment connectors."""
def __init__(self):
"""Initalize an experiment connector"""
self.project_name: Optional[str] = None
self.exp_name: Optional[str] = None
self.id: Optional[str] = None
self.output_url: Optional[str] = None
self.logger: Logger = Logger(__name__)
self.user: Optional[str] = None
@property
@abstractmethod
def config(self) -> Dict:
"""Returns config associated with experiment
Raises:
NotImplementedError: If experiment connector
does not implement this method.
Returns:
Dict: Configuration of experiment
"""
raise NotImplementedError
@property
@abstractmethod
def tags(self) -> List[str]:
"""Returns tags associated with experiment
Raises:
NotImplementedError: If experiment connector
does not implement this method.
Returns:
List: List of tags associated with experiment
"""
raise NotImplementedError
@property
@abstractmethod
def artifacts(self) -> Dict[str, Artifact]:
"""Returns artifacts associated with experiment
Raises:
ValueError: If not currently connected to any experiments
NotImplementedError: If experiment connector
does not implement this method.
Returns:
Dict[str, Artifact]: Mapping of artifact names to artifacts
"""
if not self.id:
raise ValueError("Not currently connected to any experiments")
raise NotImplementedError
@property
@abstractmethod
def models(self) -> Dict[str, Artifact]:
"""Returns models associated with experiment
Raises:
ValueError: If not currently connected to any experiments
NotImplementedError: If experiment connector
does not implement this method.
Returns:
Dict[str, Artifact]: Mapping of model names to artifacts
"""
if not self.id:
raise ValueError("Not currently connected to any experiments")
raise NotImplementedError
[docs] @classmethod
@abstractmethod
def get(cls) -> "ExperimentConnector":
"""Get an existing experiment.
Raises:
NotImplementedError: If experiment connector
does not implement this method.
Returns:
ExperimentConnector: Experiment connector
"""
raise NotImplementedError
[docs] @classmethod
@abstractmethod
def clone(
cls, exp_id: str, clone_name: Optional[str] = None
) -> "ExperimentConnector":
"""Clone an existing experiment.
Args:
exp_id (str): Id of experiment to clone
clone_name (Optional[str], optional): Name of cloned experiment. Defaults to None.
Raises:
NotImplementedError: If experiment connector
does not implement this method.
Returns:
ExperimentConnector: Cloned experiment connector
"""
raise NotImplementedError
[docs] @abstractmethod
def delete(self) -> bool:
"""Delete the experiment.
Raises:
NotImplementedError: If experiment connector
does not implement this method.
Returns:
bool: True if experiment was deleted, False otherwise
"""
raise NotImplementedError
[docs] @abstractmethod
def close(self):
"""Close the experiment connector.
Raises:
NotImplementedError: If experiment connector
does not implement this method.
"""
raise NotImplementedError