Skip to content

API Reference

Class extension

deeptool.core

Notebook-friendly primitives: class extension and hyperparameter capture.

HyperParameters

Mixin that saves __init__ arguments as attributes and in self.hparams.

Source code in deeptool/core.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class HyperParameters:
    """Mixin that saves `__init__` arguments as attributes and in `self.hparams`."""

    def save_hyperparameters(self, ignore: Iterable[str] = ()) -> None:
        """Store the calling `__init__`'s arguments on the instance.

        Reads the declared arguments of the calling frame — positional and
        keyword-only — so local variables are never captured.

        Call `super().__init__()` first and `save_hyperparameters()` second.
        Each class stores its own arguments, but `self.hparams` holds the last
        call's, so the subclass has to run last to win.

        Args:
            ignore: Argument names to leave out of both the attributes and
                `self.hparams`.
        """
        frame = inspect.currentframe().f_back
        code = frame.f_code
        n_args = code.co_argcount + code.co_kwonlyargcount
        names = code.co_varnames[:n_args]
        local_vars = frame.f_locals

        self.hparams = {
            name: local_vars[name]
            for name in names
            if name != "self" and name not in ignore and name in local_vars
        }
        for name, value in self.hparams.items():
            setattr(self, name, value)
save_hyperparameters 
save_hyperparameters(ignore: Iterable[str] = ()) -> None

Store the calling __init__'s arguments on the instance.

Reads the declared arguments of the calling frame — positional and keyword-only — so local variables are never captured.

Call super().__init__() first and save_hyperparameters() second. Each class stores its own arguments, but self.hparams holds the last call's, so the subclass has to run last to win.

Parameters:

Name Type Description Default
ignore Iterable[str]

Argument names to leave out of both the attributes and self.hparams.

 ()
Source code in deeptool/core.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def save_hyperparameters(self, ignore: Iterable[str] = ()) -> None:
    """Store the calling `__init__`'s arguments on the instance.

    Reads the declared arguments of the calling frame — positional and
    keyword-only — so local variables are never captured.

    Call `super().__init__()` first and `save_hyperparameters()` second.
    Each class stores its own arguments, but `self.hparams` holds the last
    call's, so the subclass has to run last to win.

    Args:
        ignore: Argument names to leave out of both the attributes and
            `self.hparams`.
    """
    frame = inspect.currentframe().f_back
    code = frame.f_code
    n_args = code.co_argcount + code.co_kwonlyargcount
    names = code.co_varnames[:n_args]
    local_vars = frame.f_locals

    self.hparams = {
        name: local_vars[name]
        for name in names
        if name != "self" and name not in ignore and name in local_vars
    }
    for name, value in self.hparams.items():
        setattr(self, name, value)

add_to_class 

add_to_class(Class: type) -> Callable[[Callable], Callable]

Register the decorated function as a method on Class.

Lets you define a class in one notebook cell and attach methods from later cells, without re-running the class definition.

Parameters:

Name Type Description Default
Class type

The class to attach the method to.

 required

Returns:

Type Description
Callable[[Callable], Callable]

A decorator that registers the function and returns it unchanged, so the
Callable[[Callable], Callable]

name stays usable in the cell that defined it.
Example 
@add_to_class(MyNet)
def loss(self, y_hat, y):
    return F.cross_entropy(y_hat, y)
Source code in deeptool/core.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def add_to_class(Class: type) -> Callable[[Callable], Callable]:
    """Register the decorated function as a method on `Class`.

    Lets you define a class in one notebook cell and attach methods from later
    cells, without re-running the class definition.

    Args:
        Class: The class to attach the method to.

    Returns:
        A decorator that registers the function and returns it unchanged, so the
        name stays usable in the cell that defined it.

    Example:
        ```python
        @add_to_class(MyNet)
        def loss(self, y_hat, y):
            return F.cross_entropy(y_hat, y)
        ```
    """

    def wrapper(func: Callable) -> Callable:
        setattr(Class, func.__name__, func)
        return func

    return wrapper

Data

deeptool.data

Data loading contract: training and validation loaders on one object.

DataModule

Bases: HyperParameters

Base class supplying training and validation dataloaders.

Subclasses implement get_dataloader(train) and nothing else.

Call super().__init__() before save_hyperparameters(), or the parent's defaults overwrite the subclass arguments:

class ToyData(DataModule):
    def __init__(self, batch_size=32):
        super().__init__()
        self.save_hyperparameters()
Source code in deeptool/data.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class DataModule(HyperParameters):
    """Base class supplying training and validation dataloaders.

    Subclasses implement `get_dataloader(train)` and nothing else.

    Call `super().__init__()` before `save_hyperparameters()`, or the parent's
    defaults overwrite the subclass arguments:

    ```python
    class ToyData(DataModule):
        def __init__(self, batch_size=32):
            super().__init__()
            self.save_hyperparameters()
    ```
    """

    def __init__(self, root: str = "../data", num_workers: int = 0,
                 batch_size: int = 32) -> None:
        self.save_hyperparameters()

    def get_dataloader(self, train: bool) -> torch_data.DataLoader:
        raise NotImplementedError

    def train_dataloader(self) -> torch_data.DataLoader:
        return self.get_dataloader(train=True)

    def val_dataloader(self) -> torch_data.DataLoader:
        return self.get_dataloader(train=False)

    def get_tensorloader(self, tensors: Sequence[torch.Tensor], train: bool,
                         indices: slice = slice(0, None)) -> torch_data.DataLoader:
        """Wrap a tuple of tensors in a `DataLoader`.

        Args:
            tensors: Tensors sliced together; the last one holds the targets.
            train: Shuffles when true.
            indices: Slice applied to every tensor, for splitting training from
                validation.

        Returns:
            A `torch.utils.data.DataLoader` over a `TensorDataset`.
        """
        tensors = tuple(a[indices] for a in tensors)
        dataset = torch_data.TensorDataset(*tensors)
        return torch_data.DataLoader(
            dataset, self.batch_size, shuffle=train,
            num_workers=self.num_workers,
        )
get_tensorloader 
get_tensorloader(tensors: Sequence[Tensor], train: bool, indices: slice = slice(0, None)) -> torch_data.DataLoader

Wrap a tuple of tensors in a DataLoader.

Parameters:

Name Type Description Default
tensors Sequence[Tensor]

Tensors sliced together; the last one holds the targets.

 required
train bool

Shuffles when true.

 required
indices slice

Slice applied to every tensor, for splitting training from validation.

 slice(0, None)

Returns:

Type Description
DataLoader

A torch.utils.data.DataLoader over a TensorDataset.
Source code in deeptool/data.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def get_tensorloader(self, tensors: Sequence[torch.Tensor], train: bool,
                     indices: slice = slice(0, None)) -> torch_data.DataLoader:
    """Wrap a tuple of tensors in a `DataLoader`.

    Args:
        tensors: Tensors sliced together; the last one holds the targets.
        train: Shuffles when true.
        indices: Slice applied to every tensor, for splitting training from
            validation.

    Returns:
        A `torch.utils.data.DataLoader` over a `TensorDataset`.
    """
    tensors = tuple(a[indices] for a in tensors)
    dataset = torch_data.TensorDataset(*tensors)
    return torch_data.DataLoader(
        dataset, self.batch_size, shuffle=train,
        num_workers=self.num_workers,
    )

Model

deeptool.module

Model contract: nn.Module plus hyperparameter capture and plotting hooks.

Module

Bases: Module, HyperParameters

Base class for your models.

You fill in three things: forward (or just assign self.net), loss, and configure_optimizers. In a notebook you can attach them from later cells with @add_to_class.

board and trainer are injected by Trainer.fit.

Source code in deeptool/module.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class Module(nn.Module, HyperParameters):
    """Base class for your models.

    You fill in three things: `forward` (or just assign `self.net`), `loss`, and
    `configure_optimizers`. In a notebook you can attach them from later cells
    with `@add_to_class`.

    `board` and `trainer` are injected by `Trainer.fit`.
    """

    def __init__(self, plot_train_per_epoch: int = 2,
                 plot_valid_per_epoch: int = 1) -> None:
        super().__init__()
        self.save_hyperparameters()
        self.board = None
        self.trainer = None

    def forward(self, X: torch.Tensor) -> torch.Tensor:
        assert hasattr(self, "net"), "implement forward() or assign self.net"
        return self.net(X)

    def loss(self, y_hat: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
        raise NotImplementedError

    def configure_optimizers(
        self,
    ) -> (
        torch.optim.Optimizer
        | tuple[
            torch.optim.Optimizer,
            torch.optim.lr_scheduler.LRScheduler
            | torch.optim.lr_scheduler.ReduceLROnPlateau,
        ]
    ):
        """Build an optimizer and, optionally, a learning-rate scheduler.

        Returns:
            An optimizer, or `(optimizer, scheduler)`. Normal scheduler timing
            follows `Trainer.scheduler_interval`; `ReduceLROnPlateau` receives
            validation loss whenever validation runs.
        """
        raise NotImplementedError

    def log(self, key: str, value: torch.Tensor | float) -> None:
        """Aggregate one custom scalar under its unchanged name for this interval.

        When a live board is attached, the same value is also plotted using the
        model's current training or validation phase. Calling this method before
        a Trainer is attached is a no-op.

        Args:
            key: Free-form metric name stored in progress history.
            value: A scalar tensor or plain number.

        Raises:
            ValueError: If `value` is not scalar or `key` is reserved by Trainer.
        """
        if self.trainer is None:
            return
        if torch.is_tensor(value):
            if value.numel() != 1:
                raise ValueError("logged tensors must contain one scalar value")
            value = value.detach().cpu().item()
        scalar = float(value)
        self.trainer._log_scalar(key, scalar)
        self.plot(key, scalar, train=self.training)

    def plot(self, key: str, value: torch.Tensor | float, train: bool) -> None:
        """Draw one scalar on the live board. A no-op when there is no board.

        Args:
            key: Curve name. Rendered as `train_<key>` or `val_<key>`.
            value: A scalar tensor or plain float.
            train: Selects the training or validation curve. Training points use
                the active Trainer progress unit on the x-axis.
        """
        if self.board is None or self.trainer is None:
            return
        if torch.is_tensor(value):
            value = value.detach().cpu().item()
        x = self.trainer.plot_x(train)
        if train:
            every_n = self.trainer.num_train_batches / self.plot_train_per_epoch
        else:
            every_n = self.trainer.num_val_batches / self.plot_valid_per_epoch
        prefix = "train_" if train else "val_"
        self.board.draw(x, float(value), prefix + key,
                        every_n=max(1, int(every_n)))

    def training_step(self, batch: Sequence[torch.Tensor]) -> torch.Tensor:
        loss = self.loss(self(*batch[:-1]), batch[-1])
        self.plot("loss", loss, train=True)
        return loss

    def validation_step(self, batch: Sequence[torch.Tensor]) -> torch.Tensor:
        loss = self.loss(self(*batch[:-1]), batch[-1])
        self.plot("loss", loss, train=False)
        return loss
configure_optimizers 
configure_optimizers() -> torch.optim.Optimizer | tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LRScheduler | torch.optim.lr_scheduler.ReduceLROnPlateau]

Build an optimizer and, optionally, a learning-rate scheduler.

Returns:

Type Description
Optimizer | tuple[Optimizer, LRScheduler | ReduceLROnPlateau]

An optimizer, or (optimizer, scheduler). Normal scheduler timing
Optimizer | tuple[Optimizer, LRScheduler | ReduceLROnPlateau]

follows Trainer.scheduler_interval; ReduceLROnPlateau receives
Optimizer | tuple[Optimizer, LRScheduler | ReduceLROnPlateau]

validation loss whenever validation runs.
Source code in deeptool/module.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def configure_optimizers(
    self,
) -> (
    torch.optim.Optimizer
    | tuple[
        torch.optim.Optimizer,
        torch.optim.lr_scheduler.LRScheduler
        | torch.optim.lr_scheduler.ReduceLROnPlateau,
    ]
):
    """Build an optimizer and, optionally, a learning-rate scheduler.

    Returns:
        An optimizer, or `(optimizer, scheduler)`. Normal scheduler timing
        follows `Trainer.scheduler_interval`; `ReduceLROnPlateau` receives
        validation loss whenever validation runs.
    """
    raise NotImplementedError
log 
log(key: str, value: Tensor | float) -> None

Aggregate one custom scalar under its unchanged name for this interval.

When a live board is attached, the same value is also plotted using the model's current training or validation phase. Calling this method before a Trainer is attached is a no-op.

Parameters:

Name Type Description Default
key str

Free-form metric name stored in progress history.

 required
value Tensor | float

A scalar tensor or plain number.

 required

Raises:

Type Description
ValueError

If value is not scalar or key is reserved by Trainer.
Source code in deeptool/module.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def log(self, key: str, value: torch.Tensor | float) -> None:
    """Aggregate one custom scalar under its unchanged name for this interval.

    When a live board is attached, the same value is also plotted using the
    model's current training or validation phase. Calling this method before
    a Trainer is attached is a no-op.

    Args:
        key: Free-form metric name stored in progress history.
        value: A scalar tensor or plain number.

    Raises:
        ValueError: If `value` is not scalar or `key` is reserved by Trainer.
    """
    if self.trainer is None:
        return
    if torch.is_tensor(value):
        if value.numel() != 1:
            raise ValueError("logged tensors must contain one scalar value")
        value = value.detach().cpu().item()
    scalar = float(value)
    self.trainer._log_scalar(key, scalar)
    self.plot(key, scalar, train=self.training)
plot 
plot(key: str, value: Tensor | float, train: bool) -> None

Draw one scalar on the live board. A no-op when there is no board.

Parameters:

Name Type Description Default
key str

Curve name. Rendered as train_<key> or val_<key>.

 required
value Tensor | float

A scalar tensor or plain float.

 required
train bool

Selects the training or validation curve. Training points use the active Trainer progress unit on the x-axis.

 required
Source code in deeptool/module.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def plot(self, key: str, value: torch.Tensor | float, train: bool) -> None:
    """Draw one scalar on the live board. A no-op when there is no board.

    Args:
        key: Curve name. Rendered as `train_<key>` or `val_<key>`.
        value: A scalar tensor or plain float.
        train: Selects the training or validation curve. Training points use
            the active Trainer progress unit on the x-axis.
    """
    if self.board is None or self.trainer is None:
        return
    if torch.is_tensor(value):
        value = value.detach().cpu().item()
    x = self.trainer.plot_x(train)
    if train:
        every_n = self.trainer.num_train_batches / self.plot_train_per_epoch
    else:
        every_n = self.trainer.num_val_batches / self.plot_valid_per_epoch
    prefix = "train_" if train else "val_"
    self.board.draw(x, float(value), prefix + key,
                    every_n=max(1, int(every_n)))

Trainer

deeptool.trainer

Training loop: device placement, epochs, loss aggregation, early stopping.

Trainer

Bases: HyperParameters

Runs the training loop over a Module and a DataModule.

All configuration lives on the constructor; fit takes only the model and the data. One Trainer therefore represents one training setup, and trainer.hparams records it in full.

Parameters:

Name Type Description Default
max_epochs int | None

Upper bound on epochs. Supply this or max_steps.

 None
device device | str | None

Where to train. Defaults to default_device().

 None
gradient_clip_val float

Clips gradient norm after backward when above zero.

 0
plot bool

Draws a live loss curve in the notebook.

 True
snapshot_best bool

Keeps weights from the best monitored progress point.

 True
best_path str | Path | None

Writes that snapshot to this file instead of memory.

 None
best_with_optim bool

Also stores optimizer state in the snapshot file, so it can resume training.

 False
patience int | None

Stops after this many monitor checks without improvement.

 None
log_dir str | Path | None

Writes run metadata and flushed JSONL progress rows here.

 None
monitor str

Exact scalar key used for best snapshots and early stopping.

 'val_loss'
mode Literal['min', 'max']

Whether lower (min) or higher (max) monitor values improve.

 'min'
max_steps int | None

Upper bound on completed optimizer updates. Supply this or max_epochs.

 None
log_every_n_steps int

Step interval between training records.

 1
val_every_n_steps int | None

Optional step interval between validation runs. Without it, step mode validates only at the final step.

 None
scheduler_interval Literal['auto', 'epoch', 'step']

Normal scheduler cadence. auto follows the active training unit. ReduceLROnPlateau always follows validation.

 'auto'

Raises:

Type Description
ValueError

If training limits, intervals, monitor, mode, or patience are invalid.
Source code in deeptool/trainer.py 
 38
 39
 40
 41
 42
 43
 44
 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
 89
 90
 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
166
167
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
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
class Trainer(HyperParameters):
    """Runs the training loop over a `Module` and a `DataModule`.

    All configuration lives on the constructor; `fit` takes only the model and
    the data. One `Trainer` therefore represents one training setup, and
    `trainer.hparams` records it in full.

    Args:
        max_epochs: Upper bound on epochs. Supply this or `max_steps`.
        device: Where to train. Defaults to `default_device()`.
        gradient_clip_val: Clips gradient norm after backward when above zero.
        plot: Draws a live loss curve in the notebook.
        snapshot_best: Keeps weights from the best monitored progress point.
        best_path: Writes that snapshot to this file instead of memory.
        best_with_optim: Also stores optimizer state in the snapshot file, so it
            can resume training.
        patience: Stops after this many monitor checks without improvement.
        log_dir: Writes run metadata and flushed JSONL progress rows here.
        monitor: Exact scalar key used for best snapshots and early stopping.
        mode: Whether lower (`min`) or higher (`max`) monitor values improve.
        max_steps: Upper bound on completed optimizer updates. Supply this or
            `max_epochs`.
        log_every_n_steps: Step interval between training records.
        val_every_n_steps: Optional step interval between validation runs.
            Without it, step mode validates only at the final step.
        scheduler_interval: Normal scheduler cadence. `auto` follows the active
            training unit. `ReduceLROnPlateau` always follows validation.

    Raises:
        ValueError: If training limits, intervals, monitor, mode, or patience
            are invalid.
    """

    _ROW_FIELDS = frozenset({
        "epoch", "step", "train_loss", "val_loss", "lr", "sec"
    })

    def __init__(self, max_epochs: int | None = None,
                 device: torch.device | str | None = None,
                 gradient_clip_val: float = 0, plot: bool = True,
                 snapshot_best: bool = True,
                 best_path: str | Path | None = None,
                 best_with_optim: bool = False,
                 patience: int | None = None,
                 log_dir: str | Path | None = None,
                 monitor: str = "val_loss",
                 mode: Literal["min", "max"] = "min",
                 max_steps: int | None = None,
                 log_every_n_steps: int = 1,
                 val_every_n_steps: int | None = None,
                 scheduler_interval: Literal[
                     "auto", "epoch", "step"
                 ] = "auto") -> None:
        self.save_hyperparameters()
        if (max_epochs is None) == (max_steps is None):
            raise ValueError("exactly one of max_epochs and max_steps is required")
        if max_epochs is not None:
            _require_positive_int("max_epochs", max_epochs)
        if max_steps is not None:
            _require_positive_int("max_steps", max_steps)
        _require_positive_int("log_every_n_steps", log_every_n_steps)
        if val_every_n_steps is not None:
            _require_positive_int("val_every_n_steps", val_every_n_steps)
        if scheduler_interval not in {"auto", "epoch", "step"}:
            raise ValueError(
                "scheduler_interval must be 'auto', 'epoch', or 'step'"
            )
        # patience=0 이면 최저점 epoch 에서도 epoch - best_epoch >= 0 이 참이 되어
        # 첫 epoch 직후 멈춘다. 의미가 없으므로 막는다.
        if patience is not None and patience < 1:
            raise ValueError(f"patience must be at least 1 (got {patience})")
        if not isinstance(monitor, str) or not monitor:
            raise ValueError("monitor must be a non-empty string")
        if mode not in {"min", "max"}:
            raise ValueError("mode must be 'min' or 'max'")
        self.device = torch.device(device) if device is not None else default_device()
        self.training_unit = "epoch" if max_epochs is not None else "step"
        self.board = (
            ProgressBoard(
                xlabel=self.training_unit,
                ylabel="loss",
                display=_in_notebook(),
            )
            if plot else None
        )
        self.recorder = RunRecorder(log_dir) if log_dir is not None else None
        self.history = (
            {"train_loss": [], "val_loss": []}
            if self.training_unit == "epoch"
            else {"step": [], "train_loss": [], "val_loss": []}
        )
        self.global_step = 0
        self.epoch = 0
        self.train_batch_idx = 0
        self.val_batch_idx = 0
        self._epoch_scalars: dict[str, list[float]] = {}
        self._logged_metric_names: list[str] = []
        self._bad_monitor_checks = 0
        self._best_val_loss: float | None = None
        self._best = BestSnapshot(
            snapshot_best, best_path, best_with_optim,
            monitor=monitor, mode=mode,
        )

    @property
    def best_val_loss(self) -> float | None:
        """Lowest validation loss seen, or `None` before the first check."""
        return self._best_val_loss

    @property
    def best_score(self) -> float | None:
        """Best value seen for the configured monitor."""
        return self._best.score

    @property
    def best_epoch(self) -> int | None:
        """Epoch that produced the best score, or `None` in step mode."""
        if self._best.progress_name == "epoch":
            return self._best.progress
        return None

    @property
    def best_step(self) -> int | None:
        """Step that produced the best score, or `None` in epoch mode."""
        if self._best.progress_name == "step":
            return self._best.progress
        return None

    def prepare_data(self, data: DataModule) -> None:
        self.train_dataloader = data.train_dataloader()
        self.val_dataloader = data.val_dataloader()
        try:
            self.num_train_batches = len(self.train_dataloader)
        except TypeError as error:
            if self.training_unit == "step":
                raise ValueError(
                    "step training needs a finite dataloader with a length"
                ) from error
            raise
        if self.training_unit == "step" and self.num_train_batches == 0:
            raise ValueError("train dataloader must not be empty")
        self.num_val_batches = (
            len(self.val_dataloader) if self.val_dataloader is not None else 0
        )

    def prepare_model(self, model: Module) -> None:
        model.trainer = self
        model.board = self.board
        self.model = model.to(self.device)

    def prepare_batch(self, batch: Sequence[torch.Tensor]) -> list[torch.Tensor]:
        return [a.to(self.device) for a in batch]

    def plot_x(self, train: bool) -> float:
        """Return the live-plot coordinate for the active training unit."""
        if self.training_unit == "step":
            return float(self.global_step + 1 if train else self.global_step)
        if train:
            return self.train_batch_idx / self.num_train_batches
        return float(self.epoch + 1)

    def materialize_lazy_parameters(self) -> None:
        """Materialize lazy layers with a dummy forward pass.

        `nn.LazyLinear` and friends have no parameters until the first forward,
        so building an optimizer before one would fail.
        """
        batch = self.prepare_batch(next(iter(self.train_dataloader)))
        with torch.no_grad():
            self.model(*batch[:-1])

    def fit(self, model: Module,
            data: DataModule) -> dict[str, list[float | None]]:
        self.prepare_data(data)
        # 검증 데이터가 없으면 best_epoch 가 계속 None 이라 조기 종료가 영원히
        # 발동하지 않는다. 조용히 무시하면 왜 안 멈추는지 알 수 없으므로 막는다.
        if (
            self.patience is not None
            and self.monitor == "val_loss"
            and self.num_val_batches == 0
        ):
            raise ValueError("patience needs validation data.")
        self.prepare_model(model)
        self.materialize_lazy_parameters()
        self._configure_optimizers()
        plateau = torch.optim.lr_scheduler.ReduceLROnPlateau
        if isinstance(self.scheduler, plateau) and self.num_val_batches == 0:
            raise ValueError("ReduceLROnPlateau needs validation data.")
        self._record_meta()
        if self.training_unit == "step":
            self._fit_steps()
            return self.history
        for self.epoch in range(self.max_epochs):
            started = perf_counter()
            self._epoch_scalars = {}
            self.fit_epoch()
            metrics = self._finish_logged_scalars()
            row = self._epoch_row(metrics, perf_counter() - started)
            improved = self._update_best(row, "epoch", self.epoch)
            self._record_row(row)
            self._step_scheduler()
            if self._should_stop_early(improved):
                break
        return self.history

    def _fit_steps(self) -> None:
        self.model.train()
        train_iterator = iter(self.train_dataloader)
        losses: list[float] = []
        last_lr = float(self.optim.param_groups[0]["lr"])
        window_started = perf_counter()
        self._epoch_scalars = {}
        while self.global_step < self.max_steps:
            try:
                batch = next(train_iterator)
            except StopIteration:
                self.epoch += 1
                if self._resolved_scheduler_interval() == "epoch":
                    self._step_normal_scheduler()
                train_iterator = iter(self.train_dataloader)
                batch = next(train_iterator)
            loss, last_lr = self._train_batch(batch)
            losses.append(loss)
            validate = (
                self.global_step == self.max_steps
                or self.val_every_n_steps is not None
                and self.global_step % self.val_every_n_steps == 0
            )
            val_loss = self._run_validation() if validate else None
            if val_loss is not None:
                self._track_val_loss(val_loss)
            emit = (
                validate
                or self.global_step % self.log_every_n_steps == 0
                or self.global_step == self.max_steps
            )
            if not emit:
                continue
            metrics = self._finish_logged_scalars(append_history=False)
            row = self._step_row(
                losses,
                val_loss,
                metrics,
                last_lr,
                perf_counter() - window_started,
            )
            improved = (
                self._update_best(row, "step", self.global_step)
                if validate else None
            )
            self._append_step_history(row)
            self._record_row(row)
            if validate:
                self._step_plateau_scheduler(row)
            if validate and self._should_stop_early(improved):
                return
            losses = []
            self._epoch_scalars = {}
            window_started = perf_counter()
            self.model.train()

    def _configure_optimizers(self) -> None:
        configured = self.model.configure_optimizers()
        if isinstance(configured, tuple):
            if len(configured) != 2:
                raise ValueError(
                    "configure_optimizers must return an optimizer or an "
                    "(optimizer, scheduler) pair"
                )
            self.optim, self.scheduler = configured
        else:
            self.optim = configured
            self.scheduler = None

    def _step_scheduler(self) -> None:
        if isinstance(
            self.scheduler, torch.optim.lr_scheduler.ReduceLROnPlateau
        ):
            self.scheduler.step(self.history["val_loss"][-1])
        elif self._resolved_scheduler_interval() == "epoch":
            self._step_normal_scheduler()

    def _resolved_scheduler_interval(self) -> Literal["epoch", "step"]:
        if self.scheduler_interval == "auto":
            return self.training_unit
        return self.scheduler_interval

    def _step_normal_scheduler(self) -> None:
        plateau = torch.optim.lr_scheduler.ReduceLROnPlateau
        if self.scheduler is not None and not isinstance(self.scheduler, plateau):
            self.scheduler.step()

    def _step_plateau_scheduler(self, row: dict[str, object]) -> None:
        if isinstance(
            self.scheduler, torch.optim.lr_scheduler.ReduceLROnPlateau
        ):
            self.scheduler.step(float(row["val_loss"]))

    def _record_meta(self) -> None:
        if self.recorder is None:
            return
        model_class = type(self.model)
        self.recorder.meta(
            started_at=datetime.now(UTC).isoformat(),
            device=str(self.device),
            model_class=f"{model_class.__module__}.{model_class.__qualname__}",
            trainer_hparams=self.hparams,
            model_hparams=getattr(self.model, "hparams", {}),
        )

    def _epoch_row(self, metrics: dict[str, float],
                   seconds: float) -> dict[str, object]:
        row: dict[str, object] = {
            "epoch": self.epoch,
            "train_loss": self.history["train_loss"][-1],
        }
        if self.num_val_batches > 0:
            row["val_loss"] = self.history["val_loss"][-1]
        row.update(metrics)
        row["lr"] = float(self.optim.param_groups[0]["lr"])
        row["sec"] = float(seconds)
        return row

    def _record_row(self, row: dict[str, object]) -> None:
        if self.recorder is None:
            return
        self.recorder.epoch(**row)
        if not _in_notebook():
            print(" ".join(
                f"{key}={_format_scalar(value)}" for key, value in row.items()
            ))

    def _update_best(
        self, row: dict[str, object],
        progress_name: Literal["epoch", "step"], progress: int,
    ) -> bool | None:
        if self.monitor == "val_loss" and "val_loss" not in row:
            return None
        if self.monitor not in row:
            raise ValueError(
                f"monitor {self.monitor!r} was not logged at "
                f"{progress_name} {progress}"
            )
        score = float(row[self.monitor])
        if not math.isfinite(score):
            raise ValueError(
                f"monitor {self.monitor!r} must be finite at "
                f"{progress_name} {progress}"
            )
        return self._best.update(
            score, progress_name, progress, self.model, self.optim
        )

    def _log_scalar(self, key: str, value: float) -> None:
        if key in self._ROW_FIELDS:
            raise ValueError(f"{key!r} is reserved for Trainer progress rows")
        self._epoch_scalars.setdefault(key, []).append(value)

    def _finish_logged_scalars(
        self, *, append_history: bool = True
    ) -> dict[str, float]:
        metrics = {
            key: sum(values) / len(values)
            for key, values in self._epoch_scalars.items()
        }
        if append_history:
            self._append_epoch_metrics(metrics)
        return metrics

    def _append_epoch_metrics(self, metrics: dict[str, float]) -> None:
        for key in self._logged_metric_names:
            if key not in metrics:
                self.history[key].append(None)
        for key, value in metrics.items():
            if key not in self._logged_metric_names:
                self._logged_metric_names.append(key)
                self.history[key] = [None] * self.epoch
            self.history[key].append(value)

    def _step_row(
        self,
        train_losses: list[float],
        val_loss: float | None,
        metrics: dict[str, float],
        lr: float,
        seconds: float,
    ) -> dict[str, object]:
        row: dict[str, object] = {
            "step": self.global_step,
            "train_loss": sum(train_losses) / len(train_losses),
        }
        if val_loss is not None:
            row["val_loss"] = val_loss
        row.update(metrics)
        row["lr"] = lr
        row["sec"] = seconds
        return row

    def _append_step_history(self, row: dict[str, object]) -> None:
        history_keys = {
            key for key in row if key not in {"lr", "sec"}
        }
        previous = len(self.history["step"])
        for key in history_keys - self.history.keys():
            self.history[key] = [None] * previous
        for key in self.history:
            self.history[key].append(row.get(key))

    def fit_epoch(self) -> None:
        self.model.train()
        losses = []
        for batch in self.train_dataloader:
            loss, _ = self._train_batch(batch)
            losses.append(loss)
        self.history["train_loss"].append(sum(losses) / len(losses))

        val_loss = self._run_validation()
        if val_loss is None:
            return
        self.history["val_loss"].append(val_loss)
        self._track_val_loss(val_loss)

    def _run_validation(self) -> float | None:
        if self.num_val_batches == 0:
            return None
        self.model.eval()
        losses = []
        for batch in self.val_dataloader:
            with torch.no_grad():
                loss = self.model.validation_step(self.prepare_batch(batch))
            self.val_batch_idx += 1
            losses.append(loss.detach().cpu().item())
        return sum(losses) / len(losses)

    def _track_val_loss(self, val_loss: float) -> None:
        if self._best_val_loss is None or val_loss < self._best_val_loss:
            self._best_val_loss = val_loss

    def _train_batch(
        self, batch: Sequence[torch.Tensor]
    ) -> tuple[float, float]:
        loss = self.model.training_step(self.prepare_batch(batch))
        self.optim.zero_grad()
        loss.backward()
        if self.gradient_clip_val > 0:
            self.clip_gradients(self.gradient_clip_val)
        lr = float(self.optim.param_groups[0]["lr"])
        self.optim.step()
        self.train_batch_idx += 1
        self.global_step += 1
        if self._resolved_scheduler_interval() == "step":
            self._step_normal_scheduler()
        return float(loss.detach().cpu().item()), lr

    def clip_gradients(self, grad_clip_val: float) -> None:
        params = [p for p in self.model.parameters() if p.requires_grad]
        torch.nn.utils.clip_grad_norm_(params, grad_clip_val)

    def _should_stop_early(self, improved: bool | None) -> bool:
        """True once `patience` monitor checks pass without improvement."""
        if self.patience is None or improved is None:
            return False
        self._bad_monitor_checks = (
            0 if improved else self._bad_monitor_checks + 1
        )
        return self._bad_monitor_checks >= self.patience

    def restore_best(self) -> int:
        """Load weights from the best value of the configured monitor.

        `fit` never does this on its own. Until you call it the model holds the
        last epoch's weights, so you can compare the two.

        Only model weights are restored; optimizer state is left alone.

        Returns:
            The epoch or step value that was restored.

        Raises:
            RuntimeError: If `fit` has not run, if there was no validation data,
                or if `snapshot_best` was off.
        """
        if not hasattr(self, "model"):
            raise RuntimeError("fit() has not run yet.")
        return self._best.restore(self.model)

    def save_checkpoint(self, path: str | Path) -> None:
        """Save model, optimizer, progress, and hyperparameters to a file.

        Args:
            path: Destination file.
        """
        step = self.global_step if self.training_unit == "step" else None
        _save_checkpoint(
            self.model, self.optim, self.epoch, path, step=step
        )

    @staticmethod
    def load_checkpoint(path: str | Path, model: torch.nn.Module,
                        optim: torch.optim.Optimizer | None = None) -> dict[str, Any]:
        """Restore a checkpoint into `model` in place.

        Args:
            path: Checkpoint file.
            model: Model to restore into.
            optim: Optimizer to restore as well, for resuming training. Leave it
                out to restore weights only, for inference.

        Returns:
            Stored progress (`epoch` and optional `step`) plus `hparams`.
        """
        return _load_checkpoint(path, model, optim)

    def predict(self, data: DataModule, train: bool = False,
                keep_inputs: bool = False) -> Predictions:
        """Run the trained model over `data` and collect per-sample results.

        Args:
            data: A `DataModule`.
            train: Uses the training split instead of validation.
            keep_inputs: Also collect the input tensors, for visualizing
                individual samples.

        Returns:
            A `Predictions` holding CPU tensors.
        """
        loader = data.train_dataloader() if train else data.val_dataloader()
        return _predict(self.model, loader, self.device, keep_inputs)
best_val_loss property 
best_val_loss: float | None

Lowest validation loss seen, or None before the first check.

best_score property 
best_score: float | None

Best value seen for the configured monitor.

best_epoch property 
best_epoch: int | None

Epoch that produced the best score, or None in step mode.

best_step property 
best_step: int | None

Step that produced the best score, or None in epoch mode.

plot_x 
plot_x(train: bool) -> float

Return the live-plot coordinate for the active training unit.

Source code in deeptool/trainer.py 
191
192
193
194
195
196
197
def plot_x(self, train: bool) -> float:
    """Return the live-plot coordinate for the active training unit."""
    if self.training_unit == "step":
        return float(self.global_step + 1 if train else self.global_step)
    if train:
        return self.train_batch_idx / self.num_train_batches
    return float(self.epoch + 1)
materialize_lazy_parameters 
materialize_lazy_parameters() -> None

Materialize lazy layers with a dummy forward pass.

nn.LazyLinear and friends have no parameters until the first forward, so building an optimizer before one would fail.

Source code in deeptool/trainer.py 
199
200
201
202
203
204
205
206
207
def materialize_lazy_parameters(self) -> None:
    """Materialize lazy layers with a dummy forward pass.

    `nn.LazyLinear` and friends have no parameters until the first forward,
    so building an optimizer before one would fail.
    """
    batch = self.prepare_batch(next(iter(self.train_dataloader)))
    with torch.no_grad():
        self.model(*batch[:-1])
restore_best 
restore_best() -> int

Load weights from the best value of the configured monitor.

fit never does this on its own. Until you call it the model holds the last epoch's weights, so you can compare the two.

Only model weights are restored; optimizer state is left alone.

Returns:

Type Description
int

The epoch or step value that was restored.

Raises:

Type Description
RuntimeError

If fit has not run, if there was no validation data, or if snapshot_best was off.
Source code in deeptool/trainer.py 
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
def restore_best(self) -> int:
    """Load weights from the best value of the configured monitor.

    `fit` never does this on its own. Until you call it the model holds the
    last epoch's weights, so you can compare the two.

    Only model weights are restored; optimizer state is left alone.

    Returns:
        The epoch or step value that was restored.

    Raises:
        RuntimeError: If `fit` has not run, if there was no validation data,
            or if `snapshot_best` was off.
    """
    if not hasattr(self, "model"):
        raise RuntimeError("fit() has not run yet.")
    return self._best.restore(self.model)
save_checkpoint 
save_checkpoint(path: str | Path) -> None

Save model, optimizer, progress, and hyperparameters to a file.

Parameters:

Name Type Description Default
path str | Path

Destination file.

 required
Source code in deeptool/trainer.py 
524
525
526
527
528
529
530
531
532
533
def save_checkpoint(self, path: str | Path) -> None:
    """Save model, optimizer, progress, and hyperparameters to a file.

    Args:
        path: Destination file.
    """
    step = self.global_step if self.training_unit == "step" else None
    _save_checkpoint(
        self.model, self.optim, self.epoch, path, step=step
    )
load_checkpoint staticmethod 
load_checkpoint(path: str | Path, model: Module, optim: Optimizer | None = None) -> dict[str, Any]

Restore a checkpoint into model in place.

Parameters:

Name Type Description Default
path str | Path

Checkpoint file.

 required
model Module

Model to restore into.

 required
optim Optimizer | None

Optimizer to restore as well, for resuming training. Leave it out to restore weights only, for inference.

 None

Returns:

Type Description
dict[str, Any]

Stored progress (epoch and optional step) plus hparams.
Source code in deeptool/trainer.py 
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
@staticmethod
def load_checkpoint(path: str | Path, model: torch.nn.Module,
                    optim: torch.optim.Optimizer | None = None) -> dict[str, Any]:
    """Restore a checkpoint into `model` in place.

    Args:
        path: Checkpoint file.
        model: Model to restore into.
        optim: Optimizer to restore as well, for resuming training. Leave it
            out to restore weights only, for inference.

    Returns:
        Stored progress (`epoch` and optional `step`) plus `hparams`.
    """
    return _load_checkpoint(path, model, optim)
predict 
predict(data: DataModule, train: bool = False, keep_inputs: bool = False) -> Predictions

Run the trained model over data and collect per-sample results.

Parameters:

Name Type Description Default
data DataModule

A DataModule.

 required
train bool

Uses the training split instead of validation.

 False
keep_inputs bool

Also collect the input tensors, for visualizing individual samples.

 False

Returns:

Type Description
Predictions

A Predictions holding CPU tensors.
Source code in deeptool/trainer.py 
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
def predict(self, data: DataModule, train: bool = False,
            keep_inputs: bool = False) -> Predictions:
    """Run the trained model over `data` and collect per-sample results.

    Args:
        data: A `DataModule`.
        train: Uses the training split instead of validation.
        keep_inputs: Also collect the input tensors, for visualizing
            individual samples.

    Returns:
        A `Predictions` holding CPU tensors.
    """
    loader = data.train_dataloader() if train else data.val_dataloader()
    return _predict(self.model, loader, self.device, keep_inputs)

default_device 

default_device() -> torch.device

Pick the first available accelerator, in the order cuda, mps, cpu.

Returns:

Type Description
device

A torch.device.
Source code in deeptool/trainer.py 
25
26
27
28
29
30
31
32
33
34
35
def default_device() -> torch.device:
    """Pick the first available accelerator, in the order cuda, mps, cpu.

    Returns:
        A `torch.device`.
    """
    if torch.cuda.is_available():
        return torch.device("cuda")
    if torch.backends.mps.is_available():
        return torch.device("mps")
    return torch.device("cpu")

Evaluation

deeptool.evaluate

Post-hoc evaluation: collect predictions over a whole dataset at once.

Predictions

The result of predict.

Stores only the raw outputs and targets; everything else is derived as a property and recomputed each time rather than cached.

preds, probs, confidence, correct and accuracy are classification only. For a regression model, use outputs directly.

Attributes:

Name Type Description
outputs

Raw model output, shape (N, ...). Always on CPU.
targets

Ground truth, shape (N, ...). Always on CPU.
inputs

Input tensors, present only when predict was called with keep_inputs=True. Otherwise None.
Source code in deeptool/evaluate.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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
class Predictions:
    """The result of `predict`.

    Stores only the raw `outputs` and `targets`; everything else is derived as a
    property and recomputed each time rather than cached.

    `preds`, `probs`, `confidence`, `correct` and `accuracy` are
    **classification only**. For a regression model, use `outputs` directly.

    Attributes:
        outputs: Raw model output, shape `(N, ...)`. Always on CPU.
        targets: Ground truth, shape `(N, ...)`. Always on CPU.
        inputs: Input tensors, present only when `predict` was called with
            `keep_inputs=True`. Otherwise `None`.
    """

    def __init__(self, outputs: torch.Tensor, targets: torch.Tensor,
                 inputs: torch.Tensor | None = None) -> None:
        self.outputs = outputs
        self.targets = targets
        self.inputs = inputs

    def __len__(self) -> int:
        return len(self.targets)

    def __repr__(self) -> str:
        body = f"n={len(self)} outputs={tuple(self.outputs.shape)}"
        # 회귀 결과에 accuracy 를 계산하면 브로드캐스트로 (N, N) 텐서가 만들어진다.
        # shape 이 맞을 때만 붙인다.
        if self.preds.shape == self.targets.shape:
            body += f" accuracy={self.accuracy:.4f}"
        return f"<Predictions {body}>"

    @property
    def preds(self) -> torch.Tensor:
        """Predicted class per sample, `outputs.argmax(dim=-1)`."""
        return self.outputs.argmax(dim=-1)

    @property
    def probs(self) -> torch.Tensor:
        """Class probabilities, `outputs.softmax(dim=-1)`."""
        return self.outputs.softmax(dim=-1)

    @property
    def confidence(self) -> torch.Tensor:
        """Probability assigned to the predicted class."""
        return self.probs.max(dim=-1).values

    @property
    def correct(self) -> torch.Tensor:
        """Boolean tensor of whether each prediction matches its target.

        Raises:
            ValueError: If `preds` and `targets` have different shapes, which
                means this is not a classification result.
        """
        preds = self.preds
        if preds.shape != self.targets.shape:
            raise ValueError(
                f"preds shape {tuple(preds.shape)} does not match targets shape "
                f"{tuple(self.targets.shape)}. This is a classification-only "
                "property; use outputs directly for regression."
            )
        return preds == self.targets

    @property
    def accuracy(self) -> float:
        """Fraction of correct predictions, as a plain float."""
        return self.correct.float().mean().item()
preds property 
preds: Tensor

Predicted class per sample, outputs.argmax(dim=-1).

probs property 
probs: Tensor

Class probabilities, outputs.softmax(dim=-1).

confidence property 
confidence: Tensor

Probability assigned to the predicted class.

correct property 
correct: Tensor

Boolean tensor of whether each prediction matches its target.

Raises:

Type Description
ValueError

If preds and targets have different shapes, which means this is not a classification result.
accuracy property 
accuracy: float

Fraction of correct predictions, as a plain float.

predict 

predict(model: Module, dataloader: Iterable[Sequence[Tensor]], device: device | str | None = None, keep_inputs: bool = False) -> Predictions

Run the model over an entire dataloader and collect per-sample results.

Results are moved to CPU as they arrive, so the dataset never accumulates in accelerator memory and downstream code gets the CPU tensors it expects.

Puts the model in eval mode and leaves it there. Trainer.fit_epoch calls model.train() at the start of every epoch, so resuming training is safe.

Parameters:

Name Type Description Default
model Module

A model following the Module batch convention — batch[:-1] are inputs and batch[-1] are targets.

 required
dataloader Iterable[Sequence[Tensor]]

Batches to run through the model.

 required
device device | str | None

Where to run inference. Defaults to the model's own device.

 None
keep_inputs bool

Also collect batch[0]. Off by default because inputs are much larger than outputs — 10,000 28x28 images is 31MB against 400KB of logits.

 False

Returns:

Type Description
Predictions

A Predictions holding CPU tensors.
Source code in deeptool/evaluate.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 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
@torch.no_grad()
def predict(model: torch.nn.Module,
            dataloader: Iterable[Sequence[torch.Tensor]],
            device: torch.device | str | None = None,
            keep_inputs: bool = False) -> Predictions:
    """Run the model over an entire dataloader and collect per-sample results.

    Results are moved to CPU as they arrive, so the dataset never accumulates in
    accelerator memory and downstream code gets the CPU tensors it expects.

    Puts the model in eval mode and leaves it there. `Trainer.fit_epoch` calls
    `model.train()` at the start of every epoch, so resuming training is safe.

    Args:
        model: A model following the `Module` batch convention — `batch[:-1]`
            are inputs and `batch[-1]` are targets.
        dataloader: Batches to run through the model.
        device: Where to run inference. Defaults to the model's own device.
        keep_inputs: Also collect `batch[0]`. Off by default because inputs are
            much larger than outputs — 10,000 28x28 images is 31MB against 400KB
            of logits.

    Returns:
        A `Predictions` holding CPU tensors.
    """
    model.eval()
    if device is None:
        device = next(model.parameters()).device

    outputs, targets, inputs = [], [], []
    for batch in dataloader:
        batch = [a.to(device) for a in batch]
        outputs.append(model(*batch[:-1]).cpu())
        targets.append(batch[-1].cpu())
        if keep_inputs:
            inputs.append(batch[0].cpu())

    return Predictions(
        torch.cat(outputs),
        torch.cat(targets),
        torch.cat(inputs) if keep_inputs else None,
    )

Live plotting

deeptool.board

Live training curves that update in place inside a notebook cell.

ProgressBoard

Bases: HyperParameters

A live plot that accumulates one curve per label.

Each draw call buffers a point. Once every_n points have arrived they are averaged into a single point on the curve and the figure is redrawn.

In a notebook the previous output is replaced with the new figure. In a plain script the data still accumulates but nothing is rendered.

Source code in deeptool/board.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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
89
90
91
92
93
94
95
class ProgressBoard(HyperParameters):
    """A live plot that accumulates one curve per label.

    Each `draw` call buffers a point. Once `every_n` points have arrived they
    are averaged into a single point on the curve and the figure is redrawn.

    In a notebook the previous output is replaced with the new figure. In a
    plain script the data still accumulates but nothing is rendered.
    """

    def __init__(self, xlabel: str | None = None, ylabel: str | None = None,
                 xlim: tuple[float, float] | None = None,
                 ylim: tuple[float, float] | None = None,
                 xscale: str = "linear", yscale: str = "linear",
                 ls: Sequence[str] = ("-", "--", "-.", ":"),
                 colors: Sequence[str] = ("C0", "C1", "C2", "C3"),
                 figsize: tuple[float, float] = (3.5, 2.5),
                 display: bool = True) -> None:
        self.save_hyperparameters()
        self.raw_points = {}
        self.data = {}
        self.fig = None
        self.axes = None

    def draw(self, x: float, y: float, label: str, every_n: int = 1) -> None:
        if label not in self.raw_points:
            self.raw_points[label] = []
            self.data[label] = []
        points = self.raw_points[label]
        points.append((float(x), float(y)))
        if len(points) < every_n:
            return
        n = len(points)
        sum_x = 0.0
        sum_y = 0.0
        for px, py in points:
            sum_x += px
            sum_y += py
        self.data[label].append((sum_x / n, sum_y / n))
        points.clear()
        if self.display:
            self._render()

    def _render(self) -> None:
        if self.fig is None:
            self.fig, self.axes = plt.subplots(figsize=self.figsize)
            # pyplot 의 figure 매니저에서 떼어내 inline 백엔드가 셀 끝에서
            # 같은 그림을 한 번 더 출력하는 것을 막는다. 참조는 우리가 들고 있다.
            plt.close(self.fig)
        self.axes.cla()
        for i, (label, line) in enumerate(self.data.items()):
            self.axes.plot(
                [p[0] for p in line], [p[1] for p in line],
                linestyle=self.ls[i % len(self.ls)],
                color=self.colors[i % len(self.colors)],
                label=label,
            )
        self.axes.set_xlabel(self.xlabel)
        self.axes.set_ylabel(self.ylabel)
        self.axes.set_xscale(self.xscale)
        self.axes.set_yscale(self.yscale)
        if self.xlim is not None:
            self.axes.set_xlim(self.xlim)
        if self.ylim is not None:
            self.axes.set_ylim(self.ylim)
        self.axes.grid(True)
        self.axes.legend()
        self._show()

    def _show(self) -> None:
        if not _in_notebook():
            return
        from IPython import display as ipy_display

        ipy_display.clear_output(wait=True)
        ipy_display.display(self.fig)

Run recording

deeptool.record

Persistent JSONL training records and run comparison plots.

RunRecorder

Write one run's metadata and completed epoch rows to disk.

Parameters:

Name Type Description Default
log_dir str | Path

Directory containing meta.json and history.jsonl.

 required
Source code in deeptool/record.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class RunRecorder:
    """Write one run's metadata and completed epoch rows to disk.

    Args:
        log_dir: Directory containing `meta.json` and `history.jsonl`.
    """

    def __init__(self, log_dir: str | Path) -> None:
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)
        self.meta_path = self.log_dir / "meta.json"
        self.history_path = self.log_dir / "history.jsonl"

    def meta(self, **info: object) -> None:
        """Atomically replace this run's metadata document."""
        temporary = self.log_dir / "meta.json.tmp"
        with temporary.open("w", encoding="utf-8") as file:
            json.dump(info, file, ensure_ascii=False, indent=2, default=str)
            file.write("\n")
        temporary.replace(self.meta_path)

    def epoch(self, **scalars: object) -> None:
        """Append and immediately flush one completed epoch row."""
        with self.history_path.open("a", encoding="utf-8") as file:
            file.write(json.dumps(scalars, ensure_ascii=False) + "\n")
            file.flush()
meta 
meta(**info: object) -> None

Atomically replace this run's metadata document.

Source code in deeptool/record.py 
28
29
30
31
32
33
34
def meta(self, **info: object) -> None:
    """Atomically replace this run's metadata document."""
    temporary = self.log_dir / "meta.json.tmp"
    with temporary.open("w", encoding="utf-8") as file:
        json.dump(info, file, ensure_ascii=False, indent=2, default=str)
        file.write("\n")
    temporary.replace(self.meta_path)
epoch 
epoch(**scalars: object) -> None

Append and immediately flush one completed epoch row.

Source code in deeptool/record.py 
36
37
38
39
40
def epoch(self, **scalars: object) -> None:
    """Append and immediately flush one completed epoch row."""
    with self.history_path.open("a", encoding="utf-8") as file:
        file.write(json.dumps(scalars, ensure_ascii=False) + "\n")
        file.flush()

load_runs 

load_runs(root: str | Path) -> RunData

Load immediate child runs below root from their JSONL histories.

Missing metrics are represented by None so every metric stays aligned with its run's epoch list.

Parameters:

Name Type Description Default
root str | Path

Directory whose child directories represent runs.

 required

Returns:

Type Description
RunData

Run names mapped to metric names and aligned value lists.

Raises:

Type Description
FileNotFoundError

If root does not exist.
ValueError

If a non-empty history line is not a JSON object.
Source code in deeptool/record.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def load_runs(root: str | Path) -> RunData:
    """Load immediate child runs below `root` from their JSONL histories.

    Missing metrics are represented by `None` so every metric stays aligned
    with its run's epoch list.

    Args:
        root: Directory whose child directories represent runs.

    Returns:
        Run names mapped to metric names and aligned value lists.

    Raises:
        FileNotFoundError: If `root` does not exist.
        ValueError: If a non-empty history line is not a JSON object.
    """
    runs: RunData = {}
    for run_dir in sorted(Path(root).iterdir()):
        history_path = run_dir / "history.jsonl"
        if not run_dir.is_dir() or not history_path.is_file():
            continue
        rows = _load_history(history_path)
        runs[run_dir.name] = _transpose_rows(rows)
    return runs

plot_runs 

plot_runs(source: str | Path | Mapping[str, Mapping[str, list[Any]]]) -> list[Figure]

Plot every recorded metric, overlaying all runs that contain it.

Parameters:

Name Type Description Default
source str | Path | Mapping[str, Mapping[str, list[Any]]]

A run root accepted by load_runs, or its loaded result.

 required

Returns:

Type Description
list[Figure]

One Matplotlib figure per metric, excluding epoch and step.
Source code in deeptool/record.py 
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
def plot_runs(source: str | Path | Mapping[str, Mapping[str, list[Any]]]
              ) -> list[Figure]:
    """Plot every recorded metric, overlaying all runs that contain it.

    Args:
        source: A run root accepted by `load_runs`, or its loaded result.

    Returns:
        One Matplotlib figure per metric, excluding `epoch` and `step`.
    """
    runs = source if isinstance(source, Mapping) else load_runs(source)
    metrics = dict.fromkeys(
        metric
        for run in runs.values()
        for metric in run
        if metric not in {"epoch", "step"}
    )
    figures = []
    for metric in metrics:
        figure, axes = plt.subplots()
        xlabels = set()
        for name, run in runs.items():
            if metric not in run:
                continue
            if "epoch" in run:
                xlabel = "epoch"
            elif "step" in run:
                xlabel = "step"
            else:
                xlabel = "index"
            xlabels.add(xlabel)
            x = run.get(xlabel, list(range(len(run[metric]))))
            axes.plot(x, run[metric], label=name)
        axes.set_xlabel(xlabels.pop() if len(xlabels) == 1 else "progress")
        axes.set_ylabel(metric)
        axes.grid(True)
        axes.legend()
        figures.append(figure)
    return figures

Checkpointing

deeptool.checkpoint

Checkpoint saving and restoring, and best-validation-loss snapshots.

BestSnapshot

Keeps model weights from the best value of one monitored scalar.

With path unset the snapshot lives in memory as a deepcopy; with a path it is written to that file.

Score and progress are tracked even when enabled is False. Disabling only skips the copy or write.

Attributes:

Name Type Description
score

Best monitored score, or None before the first update.
progress_name

Whether progress represents an epoch or step.
progress

Epoch or step that produced the best score.
Source code in deeptool/checkpoint.py 
 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
166
167
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
class BestSnapshot:
    """Keeps model weights from the best value of one monitored scalar.

    With `path` unset the snapshot lives in memory as a `deepcopy`; with a path
    it is written to that file.

    Score and progress are tracked even when `enabled` is False. Disabling only
    skips the copy or write.

    Attributes:
        score: Best monitored score, or `None` before the first update.
        progress_name: Whether `progress` represents an epoch or step.
        progress: Epoch or step that produced the best score.
    """

    def __init__(self, enabled: bool = True, path: str | Path | None = None,
                 with_optim: bool = False, monitor: str = "val_loss",
                 mode: Literal["min", "max"] = "min") -> None:
        self.enabled = enabled
        self.path = path
        self.with_optim = with_optim
        self.monitor = monitor
        self.mode = mode
        self.score = None
        self.progress_name = None
        self.progress = None
        self._state = None

    def update(self, score: float,
               progress_name: Literal["epoch", "step"], progress: int,
               model: torch.nn.Module,
               optim: torch.optim.Optimizer) -> bool:
        """Record an improved score and snapshot the weights.

        Does nothing when `score` is not strictly better in the configured
        direction.

        Args:
            score: Finite monitored value for this progress point.
            progress_name: Whether `progress` is an epoch or step.
            progress: Epoch or step stored when this is a new best.
            model: Model whose `state_dict` is snapshotted.
            optim: Optimizer, used only when `with_optim` is set.

        Returns:
            `True` when the score improved, otherwise `False`.
        """
        improved = (
            self.score is None
            or (self.mode == "min" and score < self.score)
            or (self.mode == "max" and score > self.score)
        )
        if not improved:
            return False
        self.score = score
        self.progress_name = progress_name
        self.progress = progress
        if not self.enabled:
            return True
        if self.path is None:
            self._state = copy.deepcopy(model.state_dict())
            return True
        # optimizer 상태는 restore() 가 읽지 않는다. Adam 기준 모델의 2배라
        # 매 개선마다 쓰면 낭비이므로 기본값은 가중치 전용이다.
        if self.with_optim:
            payload = checkpoint_payload(model, optim, progress)
            if progress_name == "step":
                payload["step"] = payload.pop("epoch")
            if self.monitor != "val_loss" or progress_name != "epoch":
                payload.update({
                    "monitor": self.monitor,
                    "mode": self.mode,
                    "score": score,
                })
        elif self.monitor != "val_loss" or progress_name != "epoch":
            payload = {
                "model": model.state_dict(),
                progress_name: progress,
                "monitor": self.monitor,
                "mode": self.mode,
                "score": score,
            }
        else:
            payload = {
                "model": model.state_dict(),
                "epoch": progress,
                "val_loss": score,
            }
        atomic_save(payload, self.path)
        return True

    def restore(self, model: torch.nn.Module) -> int:
        """Restore `model` to the best weights and return its progress value.

        Only model weights are restored; optimizer state is left alone. The
        point is to evaluate with the best model, not to resume training.

        Args:
            model: Model to restore into.

        Returns:
            The epoch or step value that was restored.

        Raises:
            RuntimeError: If no snapshot exists — either there was no validation
                data, or snapshotting was disabled.
        """
        if self.progress is None:
            raise RuntimeError("No snapshot: there was no validation data.")
        if not self.enabled:
            raise RuntimeError(
                f"No snapshot: trained with snapshot_best=False. "
                f"(best was {self.progress_name} {self.progress}, "
                f"{self.monitor} {self.score:.4f})"
            )
        if self.path is None:
            model.load_state_dict(self._state)
        else:
            ckpt = torch.load(self.path, map_location="cpu", weights_only=False)
            model.load_state_dict(ckpt["model"])
        return self.progress
update 
update(score: float, progress_name: Literal['epoch', 'step'], progress: int, model: Module, optim: Optimizer) -> bool

Record an improved score and snapshot the weights.

Does nothing when score is not strictly better in the configured direction.

Parameters:

Name Type Description Default
score float

Finite monitored value for this progress point.

 required
progress_name Literal['epoch', 'step']

Whether progress is an epoch or step.

 required
progress int

Epoch or step stored when this is a new best.

 required
model Module

Model whose state_dict is snapshotted.

 required
optim Optimizer

Optimizer, used only when with_optim is set.

 required

Returns:

Type Description
bool

True when the score improved, otherwise False.
Source code in deeptool/checkpoint.py 
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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def update(self, score: float,
           progress_name: Literal["epoch", "step"], progress: int,
           model: torch.nn.Module,
           optim: torch.optim.Optimizer) -> bool:
    """Record an improved score and snapshot the weights.

    Does nothing when `score` is not strictly better in the configured
    direction.

    Args:
        score: Finite monitored value for this progress point.
        progress_name: Whether `progress` is an epoch or step.
        progress: Epoch or step stored when this is a new best.
        model: Model whose `state_dict` is snapshotted.
        optim: Optimizer, used only when `with_optim` is set.

    Returns:
        `True` when the score improved, otherwise `False`.
    """
    improved = (
        self.score is None
        or (self.mode == "min" and score < self.score)
        or (self.mode == "max" and score > self.score)
    )
    if not improved:
        return False
    self.score = score
    self.progress_name = progress_name
    self.progress = progress
    if not self.enabled:
        return True
    if self.path is None:
        self._state = copy.deepcopy(model.state_dict())
        return True
    # optimizer 상태는 restore() 가 읽지 않는다. Adam 기준 모델의 2배라
    # 매 개선마다 쓰면 낭비이므로 기본값은 가중치 전용이다.
    if self.with_optim:
        payload = checkpoint_payload(model, optim, progress)
        if progress_name == "step":
            payload["step"] = payload.pop("epoch")
        if self.monitor != "val_loss" or progress_name != "epoch":
            payload.update({
                "monitor": self.monitor,
                "mode": self.mode,
                "score": score,
            })
    elif self.monitor != "val_loss" or progress_name != "epoch":
        payload = {
            "model": model.state_dict(),
            progress_name: progress,
            "monitor": self.monitor,
            "mode": self.mode,
            "score": score,
        }
    else:
        payload = {
            "model": model.state_dict(),
            "epoch": progress,
            "val_loss": score,
        }
    atomic_save(payload, self.path)
    return True
restore 
restore(model: Module) -> int

Restore model to the best weights and return its progress value.

Only model weights are restored; optimizer state is left alone. The point is to evaluate with the best model, not to resume training.

Parameters:

Name Type Description Default
model Module

Model to restore into.

 required

Returns:

Type Description
int

The epoch or step value that was restored.

Raises:

Type Description
RuntimeError

If no snapshot exists — either there was no validation data, or snapshotting was disabled.
Source code in deeptool/checkpoint.py 
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
def restore(self, model: torch.nn.Module) -> int:
    """Restore `model` to the best weights and return its progress value.

    Only model weights are restored; optimizer state is left alone. The
    point is to evaluate with the best model, not to resume training.

    Args:
        model: Model to restore into.

    Returns:
        The epoch or step value that was restored.

    Raises:
        RuntimeError: If no snapshot exists — either there was no validation
            data, or snapshotting was disabled.
    """
    if self.progress is None:
        raise RuntimeError("No snapshot: there was no validation data.")
    if not self.enabled:
        raise RuntimeError(
            f"No snapshot: trained with snapshot_best=False. "
            f"(best was {self.progress_name} {self.progress}, "
            f"{self.monitor} {self.score:.4f})"
        )
    if self.path is None:
        model.load_state_dict(self._state)
    else:
        ckpt = torch.load(self.path, map_location="cpu", weights_only=False)
        model.load_state_dict(ckpt["model"])
    return self.progress

save_checkpoint 

save_checkpoint(model: Module, optim: Optimizer, epoch: int, path: str | Path, *, step: int | None = None) -> None

Save model and optimizer state, epoch and hyperparameters to one file.

Parameters:

Name Type Description Default
model Module

Model to save.

 required
optim Optimizer

Optimizer to save.

 required
epoch int

Epoch index to record.

 required
path str | Path

Destination file.

 required
step int | None

Optional completed optimizer-step count.

 None
Source code in deeptool/checkpoint.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def save_checkpoint(model: torch.nn.Module, optim: torch.optim.Optimizer,
                    epoch: int, path: str | Path, *,
                    step: int | None = None) -> None:
    """Save model and optimizer state, epoch and hyperparameters to one file.

    Args:
        model: Model to save.
        optim: Optimizer to save.
        epoch: Epoch index to record.
        path: Destination file.
        step: Optional completed optimizer-step count.
    """
    torch.save(checkpoint_payload(model, optim, epoch, step=step), path)

load_checkpoint 

load_checkpoint(path: str | Path, model: Module, optim: Optimizer | None = None) -> dict[str, Any]

Restore a checkpoint into model in place.

Warning

hparams can hold arbitrary Python objects, so this reads with weights_only=False. Only load checkpoints you trust.

Parameters:

Name Type Description Default
path str | Path

Checkpoint file.

 required
model Module

Model to restore into.

 required
optim Optimizer | None

Optimizer to restore as well, for resuming training. Leave it out to restore weights only, for inference.

 None

Returns:

Type Description
dict[str, Any]

Stored progress (epoch and/or step) plus hparams.
Source code in deeptool/checkpoint.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def load_checkpoint(path: str | Path, model: torch.nn.Module,
                    optim: torch.optim.Optimizer | None = None) -> dict[str, Any]:
    """Restore a checkpoint into `model` in place.

    Warning:
        `hparams` can hold arbitrary Python objects, so this reads with
        `weights_only=False`. Only load checkpoints you trust.

    Args:
        path: Checkpoint file.
        model: Model to restore into.
        optim: Optimizer to restore as well, for resuming training. Leave it out
            to restore weights only, for inference.

    Returns:
        Stored progress (`epoch` and/or `step`) plus `hparams`.
    """
    ckpt = torch.load(path, map_location="cpu", weights_only=False)
    model.load_state_dict(ckpt["model"])
    if optim is not None:
        optim.load_state_dict(ckpt["optim"])
    metadata = {"hparams": ckpt["hparams"]}
    for progress_name in ("epoch", "step"):
        if progress_name in ckpt:
            metadata[progress_name] = ckpt[progress_name]
    return metadata

Dashboard launcher

deeptool.dashboard

Command-line launcher for the optional Streamlit run dashboard.

main 

main(argv: Sequence[str] | None = None) -> int

Validate options and launch the packaged Streamlit application.

Source code in deeptool/dashboard.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def main(argv: Sequence[str] | None = None) -> int:
    """Validate options and launch the packaged Streamlit application."""
    parser = _parser()
    args = parser.parse_args(argv)
    root = args.runs_dir.expanduser().resolve()
    if not root.is_dir():
        parser.error(f"runs directory does not exist: {root}")
    if not args.host:
        parser.error("host must be non-empty")
    if not 1 <= args.port <= 65535:
        parser.error("port must be between 1 and 65535")
    if not math.isfinite(args.refresh) or args.refresh < 0:
        parser.error("refresh must be a finite non-negative number")
    if importlib.util.find_spec("streamlit") is None:
        parser.error(
            'dashboard dependencies are missing; run: '
            'uv add "deeptool[dashboard]"'
        )
    app = Path(__file__).with_name("_dashboard_app.py")
    command = [
        sys.executable,
        "-m",
        "streamlit",
        "run",
        str(app),
        f"--server.address={args.host}",
        f"--server.port={args.port}",
        f"--server.headless={str(args.no_browser).lower()}",
        "--",
        str(root),
        "--refresh",
        str(args.refresh),
    ]
    subprocess.run(command, check=True)
    return 0