Skip to content

trackers

ConsoleResultTracker

Bases: ResultTracker

A class that directly prints to console.

Source code in klinker/trackers.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
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 = "logging",
    ):
        """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
        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='logging')

Initialize the tracker.

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".
Source code in klinker/trackers.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
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 = "logging",
):
    """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
    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 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
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.

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

Source code in klinker/trackers.py 
80
81
82
83
84
85
86
87
88
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.

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.

Source code in klinker/trackers.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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 
60
61
62
63
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.

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

Returns:
Source code in klinker/trackers.py 
48
49
50
51
52
53
54
55
56
57
58
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 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
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,
            settings=self.wandb.Settings(start_method="fork"),
            **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.

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`.

ValueError: If the project name is given as None
Source code in klinker/trackers.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
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