trackers

ConsoleResultTracker

Bases: ResultTracker

A class that directly prints to console.

Source code in klinker/trackers.py

class ConsoleResultTracker(ResultTracker):
    """A class that directly prints to console."""

    def __init__(
        self,
        *,
        track_parameters: bool = True,
        parameter_filter: Union[None, str, Pattern[str]] = None,
        track_metrics: bool = True,
        metric_filter: Union[None, str, Pattern[str]] = None,
        start_end_run: bool = False,
        writer: str = "tqdm",
    ):
        """
        Initialize the tracker.

        Args:
            track_parameters: Whether to print parameters.
            parameter_filter: A regular expression to filter parameters. If None, print all parameters.
            track_metrics: Whether to print metrics.
            metric_filter: A regular expression to filter metrics. If None, print all parameters.
            start_end_run: Whether to print start/end run messages.
            writer: The writer to use - one of "tqdm", "builtin", or "logger".
        """
        self.start_end_run = start_end_run

        self.track_parameters = track_parameters
        if isinstance(parameter_filter, str):
            parameter_filter = re.compile(parameter_filter)
        self.parameter_filter = parameter_filter

        self.track_metrics = track_metrics
        if isinstance(metric_filter, str):
            metric_filter = re.compile(metric_filter)
        self.metric_filter = metric_filter

        if writer == "tqdm":
            self.write = tqdm.write
        elif writer == "builtin":
            self.write = print  # noqa:T202
        elif writer == "logging":
            self.write = logging.getLogger("klinker").info

    def start_run(self, run_name: Optional[str] = None) -> None:
        if run_name is not None and self.start_end_run:
            self.write(f"Starting run: {run_name}")

    def log_params(
        self, params: Mapping[str, Any], prefix: Optional[str] = None
    ) -> None:
        if not self.track_parameters:
            return

        for key, value in flatten_dictionary(dictionary=params).items():
            if not self.parameter_filter or self.parameter_filter.match(key):
                self.write(f"Parameter: {key} = {value}")

    def log_metrics(
        self,
        metrics: Mapping[str, float],
        step: Optional[int] = None,
        prefix: Optional[str] = None,
    ) -> None:
        if not self.track_metrics:
            return
        self.write(f"Step: {step}")
        for key, value in flatten_dictionary(dictionary=metrics, prefix=prefix).items():
            if not self.metric_filter or self.metric_filter.match(key):
                self.write(f"Metric: {key} = {value}")

    def end_run(self, success: bool = True) -> None:
        if not success:
            self.write("Run failed.")
        if self.start_end_run:
            self.write("Finished run.")

__init__(*, track_parameters=True, parameter_filter=None, track_metrics=True, metric_filter=None, start_end_run=False, writer='tqdm')

Initialize the tracker.

Parameters:

Name	Type	Description	Default
track_parameters	bool	Whether to print parameters.	True
parameter_filter	Union[None, str, Pattern[str]]	A regular expression to filter parameters. If None, print all parameters.	None
track_metrics	bool	Whether to print metrics.	True
metric_filter	Union[None, str, Pattern[str]]	A regular expression to filter metrics. If None, print all parameters.	None
start_end_run	bool	Whether to print start/end run messages.	False
writer	str	The writer to use - one of "tqdm", "builtin", or "logger".	'tqdm'

Source code in klinker/trackers.py

def __init__(
    self,
    *,
    track_parameters: bool = True,
    parameter_filter: Union[None, str, Pattern[str]] = None,
    track_metrics: bool = True,
    metric_filter: Union[None, str, Pattern[str]] = None,
    start_end_run: bool = False,
    writer: str = "tqdm",
):
    """
    Initialize the tracker.

    Args:
        track_parameters: Whether to print parameters.
        parameter_filter: A regular expression to filter parameters. If None, print all parameters.
        track_metrics: Whether to print metrics.
        metric_filter: A regular expression to filter metrics. If None, print all parameters.
        start_end_run: Whether to print start/end run messages.
        writer: The writer to use - one of "tqdm", "builtin", or "logger".
    """
    self.start_end_run = start_end_run

    self.track_parameters = track_parameters
    if isinstance(parameter_filter, str):
        parameter_filter = re.compile(parameter_filter)
    self.parameter_filter = parameter_filter

    self.track_metrics = track_metrics
    if isinstance(metric_filter, str):
        metric_filter = re.compile(metric_filter)
    self.metric_filter = metric_filter

    if writer == "tqdm":
        self.write = tqdm.write
    elif writer == "builtin":
        self.write = print  # noqa:T202
    elif writer == "logging":
        self.write = logging.getLogger("klinker").info

ResultTracker

A class that tracks the results from a pipeline run.

Source code in klinker/trackers.py

class ResultTracker:
    """A class that tracks the results from a pipeline run."""

    def start_run(self, run_name: Optional[str] = None) -> None:
        """Start a run with an optional name.

        Args:
          run_name: Optional[str]:  (Default value = None)

        Returns:

        """

    def log_params(
        self, params: Mapping[str, Any], prefix: Optional[str] = None
    ) -> None:
        """Log parameters to result store."""

    def log_metrics(
        self,
        metrics: Mapping[str, float],
        step: Optional[int] = None,
        prefix: Optional[str] = None,
    ) -> None:
        """Log metrics to result store.

        Args:
          metrics: The metrics to log.
          step: An optional step to attach the metrics to (e.g. the epoch).
          prefix: An optional prefix to prepend to every key in metrics.
        """

    def end_run(self, success: bool = True) -> None:
        """End a run.

        HAS to be called after the experiment is finished.

        Args:
          success: Can be used to signal failed runs. May be ignored.
        """

end_run(success=True)

End a run.

HAS to be called after the experiment is finished.

Parameters:

Name	Type	Description	Default
success	bool	Can be used to signal failed runs. May be ignored.	True

Source code in klinker/trackers.py

def end_run(self, success: bool = True) -> None:
    """End a run.

    HAS to be called after the experiment is finished.

    Args:
      success: Can be used to signal failed runs. May be ignored.
    """

log_metrics(metrics, step=None, prefix=None)

Log metrics to result store.

Parameters:

Name	Type	Description	Default
metrics	Mapping[str, float]	The metrics to log.	required
step	Optional[int]	An optional step to attach the metrics to (e.g. the epoch).	None
prefix	Optional[str]	An optional prefix to prepend to every key in metrics.	None

Source code in klinker/trackers.py

def log_metrics(
    self,
    metrics: Mapping[str, float],
    step: Optional[int] = None,
    prefix: Optional[str] = None,
) -> None:
    """Log metrics to result store.

    Args:
      metrics: The metrics to log.
      step: An optional step to attach the metrics to (e.g. the epoch).
      prefix: An optional prefix to prepend to every key in metrics.
    """

log_params(params, prefix=None)

Log parameters to result store.

Source code in klinker/trackers.py

def log_params(
    self, params: Mapping[str, Any], prefix: Optional[str] = None
) -> None:
    """Log parameters to result store."""

start_run(run_name=None)

Start a run with an optional name.

Parameters:

Name	Type	Description	Default
run_name	Optional[str]	Optional[str]: (Default value = None)	None

Returns:

Source code in klinker/trackers.py

def start_run(self, run_name: Optional[str] = None) -> None:
    """Start a run with an optional name.

    Args:
      run_name: Optional[str]:  (Default value = None)

    Returns:

    """

WANDBResultTracker

Bases: ResultTracker

A tracker for Weights and Biases.

Note that you have to perform wandb login beforehand.

Source code in klinker/trackers.py

class WANDBResultTracker(ResultTracker):
    """A tracker for Weights and Biases.

    Note that you have to perform wandb login beforehand.
    """

    #: The WANDB run
    run: Optional["wandb.wandb_run.Run"]

    def __init__(
        self,
        project: str,
        offline: bool = False,
        **kwargs,
    ):
        """Initialize result tracking via WANDB.

        Args:
            project:
                project name your WANDB login has access to.
            offline:
                whether to run in offline mode, i.e, without syncing with the wandb server.
            kwargs:
                additional keyword arguments passed to :func:`wandb.init`.

        Raises:
            ValueError: If the project name is given as None
        """
        import wandb as _wandb

        self.wandb = _wandb
        if project is None:
            raise ValueError("Weights & Biases requires a project name.")
        self.project = project

        if offline:
            os.environ[self.wandb.env.MODE] = "dryrun"  # type: ignore
        self.kwargs = kwargs
        self.run = None

    def start_run(self, run_name: Optional[str] = None) -> None:
        self.run = self.wandb.init(project=self.project, name=run_name, **self.kwargs)  # type: ignore

    def end_run(self, success: bool = True) -> None:
        assert self.run
        self.run.finish(exit_code=0 if success else -1)
        self.run = None

    def log_metrics(
        self,
        metrics: Mapping[str, float],
        step: Optional[int] = None,
        prefix: Optional[str] = None,
    ) -> None:
        if self.run is None:
            raise AssertionError("start_run must be called before logging any metrics")
        metrics = flatten_dictionary(dictionary=metrics, prefix=prefix)
        self.run.log(metrics, step=step)

    def log_params(
        self, params: Mapping[str, Any], prefix: Optional[str] = None
    ) -> None:
        if self.run is None:
            raise AssertionError("start_run must be called before logging any metrics")
        params = flatten_dictionary(dictionary=params, prefix=prefix)
        self.run.config.update(params)

__init__(project, offline=False, **kwargs)

Initialize result tracking via WANDB.

Parameters:

Name	Type	Description	Default
project	str	project name your WANDB login has access to.	required
offline	bool	whether to run in offline mode, i.e, without syncing with the wandb server.	False
kwargs		additional keyword arguments passed to :func:wandb.init.	{}

Raises:

Type	Description
ValueError	If the project name is given as None

Source code in klinker/trackers.py

def __init__(
    self,
    project: str,
    offline: bool = False,
    **kwargs,
):
    """Initialize result tracking via WANDB.

    Args:
        project:
            project name your WANDB login has access to.
        offline:
            whether to run in offline mode, i.e, without syncing with the wandb server.
        kwargs:
            additional keyword arguments passed to :func:`wandb.init`.

    Raises:
        ValueError: If the project name is given as None
    """
    import wandb as _wandb

    self.wandb = _wandb
    if project is None:
        raise ValueError("Weights & Biases requires a project name.")
    self.project = project

    if offline:
        os.environ[self.wandb.env.MODE] = "dryrun"  # type: ignore
    self.kwargs = kwargs
    self.run = None

flatten_dictionary(dictionary, prefix=None, sep='.')

Flatten a nested dictionary.

Source code in klinker/trackers.py

def flatten_dictionary(
    dictionary: Mapping[str, Any],
    prefix: Optional[str] = None,
    sep: str = ".",
) -> Dict[str, Any]:
    """Flatten a nested dictionary."""
    real_prefix = tuple() if prefix is None else (prefix,)
    partial_result = _flatten_dictionary(dictionary=dictionary, prefix=real_prefix)
    return {sep.join(map(str, k)): v for k, v in partial_result.items()}