Skip to content

Trainer

pamiq_core.trainer.Trainer 

Trainer(
    training_condition_data_user: str | None = None,
    min_buffer_size: int = 0,
    min_new_data_count: int = 0,
)

Bases: ABC, PersistentStateMixin, ThreadEventMixin

Abstract base trainer class.

The run method is called repeatedly in the training thread.

Override the following methods
  • on_training_models_attached: Callback method for when training models are attached to the trainer.
  • on_data_users_attached: Callback method when data_users are attached to the trainer.
  • is_trainable: Return whether the training can be executed.
  • setup: To setup before training starts.
  • train: The training process.
  • teardown: To teardown after training.

Models and data buffers become available after the thread has started.

Initialize a trainer.

PARAMETER DESCRIPTION
training_condition_data_user

Name of the data user to check for trainability. If None, trainer is always trainable.

TYPE: str | None DEFAULT: None

min_buffer_size

Minimum total data points required in the buffer.

TYPE: int DEFAULT: 0

min_new_data_count

Minimum number of new data points required since last training.

TYPE: int DEFAULT: 0

Source code in src/pamiq_core/trainer/base.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    training_condition_data_user: str | None = None,
    min_buffer_size: int = 0,
    min_new_data_count: int = 0,
) -> None:
    """Initialize a trainer.

    Args:
        training_condition_data_user: Name of the data user to check for trainability.
            If None, trainer is always trainable.
        min_buffer_size: Minimum total data points required in the buffer.
        min_new_data_count: Minimum number of new data points required since last training.
    """
    super().__init__()
    self._retrieved_model_names: set[str] = set()
    self._training_condition_data_user = training_condition_data_user
    self._min_buffer_size = min_buffer_size
    self._min_new_data_count = min_new_data_count
    self._previous_training_time = float("-inf")

attach_training_models 

attach_training_models(training_models: TrainingModelsDict) -> None

Attaches TrainingModelsDict to this trainer.

Source code in src/pamiq_core/trainer/base.py 
52
53
54
55
def attach_training_models(self, training_models: TrainingModelsDict) -> None:
    """Attaches TrainingModelsDict to this trainer."""
    self._training_models = training_models
    self.on_training_models_attached()

on_training_models_attached 

on_training_models_attached() -> None

Callback method for when training models are attached to the trainer.

Use :meth:get_training_model to retrieve the model that will be trained.

Source code in src/pamiq_core/trainer/base.py 
57
58
59
60
61
62
63
def on_training_models_attached(self) -> None:
    """Callback method for when training models are attached to the
    trainer.

    Use :meth:`get_training_model` to retrieve the model that will be trained.
    """
    pass

attach_data_users 

attach_data_users(data_users: DataUsersDict) -> None

Attaches DataUsersDict to this trainer.

Source code in src/pamiq_core/trainer/base.py 
65
66
67
68
def attach_data_users(self, data_users: DataUsersDict) -> None:
    """Attaches DataUsersDict to this trainer."""
    self._data_users = data_users
    self.on_data_users_attached()

on_data_users_attached 

on_data_users_attached() -> None

Callback method when data users are attached to the trainer.

Use :meth:get_data_user to obtain the data user class for dataset.

Source code in src/pamiq_core/trainer/base.py 
70
71
72
73
74
75
def on_data_users_attached(self) -> None:
    """Callback method when data users are attached to the trainer.

    Use :meth:`get_data_user` to obtain the data user class for dataset.
    """
    pass

get_training_model 

get_training_model(name: str) -> TrainingModel[Any]

Retrieves the training model.

If the specified model includes an inference model, it is automatically synchronized after training.

Source code in src/pamiq_core/trainer/base.py 
77
78
79
80
81
82
83
84
85
def get_training_model(self, name: str) -> TrainingModel[Any]:
    """Retrieves the training model.

    If the specified model includes an inference model, it is
    automatically synchronized after training.
    """
    model = self._training_models[name]
    self._retrieved_model_names.add(name)
    return model

get_data_user 

get_data_user(name: str) -> DataUser[Any]

Retrieves the data user.

Source code in src/pamiq_core/trainer/base.py 
87
88
89
def get_data_user(self, name: str) -> DataUser[Any]:
    """Retrieves the data user."""
    return self._data_users[name]

is_trainable 

is_trainable() -> bool

Determines if the training can be executed.

Checks if training can proceed based on data availability when a training condition data user is specified.

RETURNS DESCRIPTION
bool

True if training can be executed, False otherwise.
Source code in src/pamiq_core/trainer/base.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
def is_trainable(self) -> bool:
    """Determines if the training can be executed.

    Checks if training can proceed based on data availability when
    a training condition data user is specified.

    Returns:
        True if training can be executed, False otherwise.
    """
    # If no data user is specified for condition checking, always trainable
    if self._training_condition_data_user is None:
        return True

    data_user = self.get_data_user(self._training_condition_data_user)
    data_user.update()

    trainable = (
        len(data_user) >= self._min_buffer_size
        and data_user.count_data_added_since(self._previous_training_time)
        >= self._min_new_data_count
    )

    if trainable:
        self._previous_training_time = time.time()

    return trainable

setup 

setup() -> None

Setup procedure before training starts.

Source code in src/pamiq_core/trainer/base.py 
118
119
120
def setup(self) -> None:
    """Setup procedure before training starts."""
    pass

train abstractmethod 

train() -> None

Train models.

Please build the models, optimizers, dataset, and other components in this method. This method is called repeatedly.

After this method, :meth:sync_models to be called.

Source code in src/pamiq_core/trainer/base.py 
122
123
124
125
126
127
128
129
130
@abstractmethod
def train(self) -> None:
    """Train models.

    Please build the models, optimizers, dataset, and other components in this method.
    This method is called repeatedly.

    After this method, :meth:`sync_models` to be called.
    """

sync_models 

sync_models() -> None

Synchronizes params of trained models to inference models.

Source code in src/pamiq_core/trainer/base.py 
132
133
134
135
def sync_models(self) -> None:
    """Synchronizes params of trained models to inference models."""
    for name in self._retrieved_model_names:
        self._training_models[name].sync()

teardown 

teardown() -> None

Teardown procedure after training.

Source code in src/pamiq_core/trainer/base.py 
137
138
139
def teardown(self) -> None:
    """Teardown procedure after training."""
    pass

run 

run() -> bool

Runs the training process if the trainer is trainable.

RETURNS DESCRIPTION
bool

True if training was executed, False if skipped due to conditions not met.

TYPE: bool

Source code in src/pamiq_core/trainer/base.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def run(self) -> bool:
    """Runs the training process if the trainer is trainable.

    Returns:
        bool: True if training was executed, False if skipped due to conditions not met.
    """
    if not self.is_trainable():
        return False

    self.setup()
    self.train()
    self.sync_models()
    self.teardown()
    return True

save_state 

save_state(path: Path) -> None

Save the trainer state to the specified path.

PARAMETER DESCRIPTION
path

Directory path where to save the trainer state.

TYPE: Path

Source code in src/pamiq_core/trainer/base.py 
156
157
158
159
160
161
162
163
164
165
166
@override
def save_state(self, path: Path) -> None:
    """Save the trainer state to the specified path.

    Args:
        path: Directory path where to save the trainer state.
    """
    path.mkdir()
    (path / "previous_training_time").write_text(
        str(self._previous_training_time), encoding="utf-8"
    )

load_state 

load_state(path: Path) -> None

Load the trainer state from the specified path.

PARAMETER DESCRIPTION
path

Directory path from where to load the trainer state.

TYPE: Path

Source code in src/pamiq_core/trainer/base.py 
168
169
170
171
172
173
174
175
176
177
@override
def load_state(self, path: Path) -> None:
    """Load the trainer state from the specified path.

    Args:
        path: Directory path from where to load the trainer state.
    """
    self._previous_training_time = float(
        (path / "previous_training_time").read_text("utf-8")
    )