Skip to content

Utilities

The npfl138 package also provides a few utilities for startup initialization, parameter initialization override, and for version management.

npfl138.first_time 

first_time(tag: str) -> bool

Returns True when first called with the given tag.

Source code in npfl138/first_time.py 
10
11
12
13
14
15
16
def first_time(tag: str) -> bool:
    """Returns `True` when first called with the given `tag`."""
    if tag in seen_tags:
        return False

    seen_tags.add(tag)
    return True

npfl138.global_keras_initializers 

global_keras_initializers(
    parameter_initialization: bool = True,
    batchnorm_momentum_override: bool = True,
) -> None

Change default PyTorch initializers to Keras defaults.

The following initializers are used:

  • Linear, Conv1d, Conv2d, Conv3d, ConvTranspose1d, ConvTranspose2d, ConvTranspose3d, Bilinear: Xavier uniform for weights, zeros for biases.
  • Embedding, EmbeddingBag: Uniform [-0.05, 0.05] for weights.
  • RNN, RNNCell, LSTM, LSTMCell, GRU, GRUCell: Xavier uniform for input weights, orthogonal for recurrent weights, zeros for biases (with LSTM forget gate bias set to 1).

Furthermore, for batch normalization layers, the default momentum value is changed from 0.1 to 0.01.

Parameters:

  • parameter_initialization (bool, default: True ) –

    If True, override the default PyTorch initializers with Keras defaults.

  • batchnorm_momentum_override (bool, default: True ) –

    If True, override the default momentum value of 0.1 in batch normalization layers to 0.01.

Source code in npfl138/initializers_override.py 
 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
def global_keras_initializers(
    parameter_initialization: bool = True,
    batchnorm_momentum_override: bool = True,
) -> None:
    """Change default PyTorch initializers to Keras defaults.

    The following initializers are used:

    - `Linear`, `Conv1d`, `Conv2d`, `Conv3d`, `ConvTranspose1d`, `ConvTranspose2d`, `ConvTranspose3d`, `Bilinear`:
      Xavier uniform for weights, zeros for biases.
    - `Embedding`, `EmbeddingBag`: Uniform [-0.05, 0.05] for weights.
    - `RNN`, `RNNCell`, `LSTM`, `LSTMCell`, `GRU`, `GRUCell`: Xavier uniform for input weights,
      orthogonal for recurrent weights, zeros for biases (with LSTM forget gate bias set to 1).

    Furthermore, for batch normalization layers, the default momentum value is changed from 0.1 to 0.01.

    Parameters:
     parameter_initialization: If True, override the default PyTorch initializers with Keras defaults.
     batchnorm_momentum_override: If True, override the default momentum value of 0.1 in
       batch normalization layers to 0.01.
    """
    if parameter_initialization:
        for class_, reset_parameters_method in KerasParameterInitialization.overrides.items():
            class_.reset_parameters = reset_parameters_method

    if batchnorm_momentum_override:
        for batch_norm_super in KerasBatchNormMomentum.batch_norms:
            for batch_norm in [batch_norm_super] + batch_norm_super.__subclasses__():
                KerasBatchNormMomentum.override_default_argument_value(batch_norm.__init__, "momentum", 0.01)

npfl138.require_version 

require_version(required_version: str) -> None

Make sure the installed version is at least required_version.

If not, show a nice error message with the instructions how to upgrade.

Parameters:

  • required_version (str) –

    The required version of the npfl138 package, as a string in the format "semester.week" or "semester.week.patch".

Source code in npfl138/version.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def require_version(required_version: str) -> None:
    """Make sure the installed version is at least `required_version`.

    If not, show a nice error message with the instructions how to upgrade.

    Parameters:
      required_version: The required version of the npfl138 package, as
        a string in the format "semester.week" or "semester.week.patch".
    """

    required = required_version.split(".")
    assert len(required) <= 3, "Expected at most 3 version components"

    required = list(map(int, required))
    current = list(map(int, __version__.split(".")))

    assert current[:len(required)] >= required, (
        f"The npfl138>={required_version} is required, but found only {__version__}.\n"
        f"Please update the npfl138 package by running either:\n"
        f"- `VENV_DIR/bin/pip install --upgrade npfl138` when using a venv, or\n"
        f"- `python3 -m pip install --user --upgrade npfl138` otherwise.")

npfl138.rl_utils

npfl138.rl_utils.EvaluationEnv

Bases: Wrapper

A wrapper over gym environments capable of performing evaluation on demand.

In addition to a standard gym envrironment, it provides the following features:

  • The reset method can be called as reset(start_evaluation=False) to start the evaluation.
  • The env.episode property returns the number of completed episodes.
Source code in npfl138/rl_utils.py 
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
class EvaluationEnv(gym.Wrapper):
    """A wrapper over gym environments capable of performing evaluation on demand.

    In addition to a standard gym envrironment, it provides the following features:

    - The `reset` method can be called as `reset(start_evaluation=False)` to start the evaluation.
    - The `env.episode` property returns the number of completed episodes.
    """
    def __init__(self, env, seed=None, render_each=0, evaluate_for=100, report_each=10):
        super().__init__(env)
        self._render_each = render_each
        self._evaluate_for = evaluate_for
        self._report_each = report_each
        self._report_verbose = os.getenv("VERBOSE", "1") not in ["", "0"]

        gym.Env.reset(self.unwrapped, seed=seed)
        self.action_space.seed(seed)
        self.observation_space.seed(seed)

        self._episode_running = False
        self._episode_returns = []
        self._evaluating_from = None
        self._original_render_mode = env.render_mode
        self._pygame = __import__("pygame") if self._render_each else None

    @property
    def episode(self):
        return len(self._episode_returns)

    def reset(self, *, start_evaluation=False, logging=True, seed=None, options=None):
        start_evaluation = start_evaluation or (options or {}).get("start_evaluation", False)
        logging = logging and (options or {}).get("logging", True)

        if seed is not None:
            raise RuntimeError("The EvaluationEnv cannot be reseeded")
        if self._evaluating_from is not None and self._episode_running:
            raise RuntimeError("Cannot reset a running episode after `start_evaluation=True`")
        if start_evaluation and self._evaluating_from is None:
            self._evaluating_from = self.episode

        if logging and self._render_each and (self.episode + 1) % self._render_each == 0:
            self.unwrapped.render_mode = "human"
        elif self._render_each:
            self.unwrapped.render_mode = self._original_render_mode
        self._episode_running = True
        self._episode_return = 0 if logging or self._evaluating_from is not None else None
        return super().reset(options=options)

    def step(self, action):
        if not self._episode_running:
            raise RuntimeError("Cannot run `step` on environments without an active episode, run `reset` first")

        observation, reward, terminated, truncated, info = super().step(action)
        done = terminated or truncated

        self._episode_running = not done
        if self._episode_return is not None:
            self._episode_return += reward
        if self._episode_return is not None and done:
            self._episode_returns.append(self._episode_return)

            if self._report_each and self.episode % self._report_each == 0:
                print("Episode {}, mean {}-episode return {:.2f} +-{:.2f}{}".format(
                    self.episode, self._evaluate_for, np.mean(self._episode_returns[-self._evaluate_for:]),
                    np.std(self._episode_returns[-self._evaluate_for:]), "" if not self._report_verbose else
                    ", returns " + " ".join(map("{:g}".format, self._episode_returns[-self._report_each:]))),
                    file=sys.stderr, flush=True)
            if self._evaluating_from is not None and self.episode >= self._evaluating_from + self._evaluate_for:
                print("The mean {}-episode return after evaluation {:.2f} +-{:.2f}".format(
                    self._evaluate_for, np.mean(self._episode_returns[-self._evaluate_for:]),
                    np.std(self._episode_returns[-self._evaluate_for:])), flush=True)
                self.close()
                sys.exit(0)

        if self._pygame and self.unwrapped.render_mode == "human" and self._pygame.get_init():
            if self._pygame.event.get(self._pygame.QUIT):
                self.unwrapped.render_mode = self._original_render_mode

        return observation, reward, terminated, truncated, info

npfl138.rl_utils.typed_torch_function 

typed_torch_function(device, *types, via_np=True)

Typed Torch function decorator.

The positional input arguments are converted to torch Tensors of the given types and on the given device; for NumPy arrays on the same device, the conversion should not copy the data.

The torch Tensors generated by the wrapped function are converted back to Numpy arrays before returning (while keeping original tuples, lists, and dictionaries).

Source code in npfl138/rl_utils.py 
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
def typed_torch_function(device, *types, via_np=True):
    """Typed Torch function decorator.

    The positional input arguments are converted to torch Tensors of the given
    types and on the given device; for NumPy arrays on the same device,
    the conversion should not copy the data.

    The torch Tensors generated by the wrapped function are converted back
    to Numpy arrays before returning (while keeping original tuples, lists,
    and dictionaries).
    """
    def check_typed_torch_function(wrapped, args):
        if len(types) != len(args):
            while hasattr(wrapped, "__wrapped__"):
                wrapped = wrapped.__wrapped__
            raise AssertionError("The typed_torch_function decorator for {} expected {} arguments, but got {}".format(
                wrapped, len(types), len(args)))

    def structural_map(value):
        if isinstance(value, torch.Tensor):
            return value.numpy(force=True)
        if isinstance(value, tuple):
            return tuple(structural_map(element) for element in value)
        if isinstance(value, list):
            return [structural_map(element) for element in value]
        if isinstance(value, dict):
            return {key: structural_map(element) for key, element in value.items()}
        return value

    class TypedTorchFunctionWrapper:
        def __init__(self, func):
            self.__wrapped__ = func

        def __call__(self, *args, **kwargs):
            check_typed_torch_function(self.__wrapped__, args)
            return structural_map(self.__wrapped__(
                *[torch.as_tensor(np.asarray(arg) if via_np else arg, dtype=typ, device=device)
                  for arg, typ in zip(args, types)], **kwargs))

        def __get__(self, instance, cls):
            return TypedTorchFunctionWrapper(self.__wrapped__.__get__(instance, cls))

    return TypedTorchFunctionWrapper

npfl138.startup 

startup(
    seed: int | None = None,
    threads: int | None = None,
    forkserver_instead_of_fork: bool = False,
    recodex: bool = False,
) -> None

Initialize the environment.

  • Allow using TF32 for matrix multiplication.
  • Set the random seed if given.
  • Set the number of threads if given.
  • Use forkserver instead of fork if requested.
  • Optionally run in ReCodEx mode for better replicability. In this mode,
    • Layer initialization does not depend on the global random seed generator (it is deterministic and depends only on the parameter shape).
    • Every dataloader uses its own random generator.
    • However, the deterministic layer initialization decreases performance of trained models, so it should be used only for running tests.

Parameters:

  • seed (int | None, default: None ) –

    If not None, set the Python, Numpy, and PyTorch random seeds to this value.

  • threads (int | None, default: None ) –

    If not None of 0, set the number of threads to this value. Otherwise, use as many threads as cores.

  • forkserver_instead_of_fork (bool, default: False ) –

    If True, use forkserver instead of fork as the default multiprocessing method. This will be the default in Python 3.14.

  • recodex (bool, default: False ) –

    If True, run in ReCodEx mode for better replicability of tests.

Source code in npfl138/startup.py 
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
def startup(seed: int | None = None,
            threads: int | None = None,
            forkserver_instead_of_fork: bool = False,
            recodex: bool = False) -> None:
    """Initialize the environment.

    - Allow using TF32 for matrix multiplication.
    - Set the random seed if given.
    - Set the number of threads if given.
    - Use `forkserver` instead of `fork` if requested.
    - Optionally run in ReCodEx mode for better replicability. In this mode,
        - Layer initialization does not depend on the global random seed generator
          (it is deterministic and depends only on the parameter shape).
        - Every dataloader uses its own random generator.
        - However, the deterministic layer initialization decreases performance
          of trained models, so it should be used only for running tests.

    Parameters:
      seed: If not `None`, set the Python, Numpy, and PyTorch random seeds to this value.
      threads: If not `None` of 0, set the number of threads to this value.
        Otherwise, use as many threads as cores.
      forkserver_instead_of_fork: If `True`, use `forkserver` instead of `fork` as the
        default multiprocessing method. This will be the default in Python 3.14.
      recodex: If `True`, run in ReCodEx mode for better replicability of tests.
    """

    # Allow TF32 when available.
    torch.backends.cuda.matmul.allow_tf32 = True

    # Set random seed if not None.
    if seed is not None:
        random.seed(seed)
        np.random.seed(seed)
        torch.manual_seed(seed)

    # Set number of threads if > 0; otherwise, use as many threads as cores.
    if threads is not None and threads > 0:
        if torch.get_num_threads() != threads:
            torch.set_num_threads(threads)
        if torch.get_num_interop_threads() != threads:
            torch.set_num_interop_threads(threads)

    # If instructed, use `forkserver` instead of `fork` (which will be the default in Python 3.14).
    if "fork" in torch.multiprocessing.get_all_start_methods():
        if os.environ.get("FORCE_FORK_METHOD") == "1":
            if torch.multiprocessing.get_start_method(allow_none=True) != "fork":
                torch.multiprocessing.set_start_method("fork")
        elif forkserver_instead_of_fork or os.environ.get("FORCE_FORKSERVER_METHOD") == "1":
            if torch.multiprocessing.get_start_method(allow_none=True) != "forkserver":
                torch.multiprocessing.set_start_method("forkserver")

    # If ReCodEx mode is requested, apply various overrides for better replicability.
    if recodex:
        # Do not use accelerators.
        torch.cuda.is_available = lambda: False
        torch.backends.mps.is_available = lambda: False
        torch.xpu.is_available = lambda: False

        # Make initializers deterministic.
        def bind_generator(init):
            return lambda *args, **kwargs: init(*args, **kwargs | {"generator": torch.Generator().manual_seed(seed)})
        for name in ["uniform_", "normal_", "trunc_normal_", "xavier_uniform_", "xavier_normal_",
                     "kaiming_uniform_", "kaiming_normal_", "orthogonal_", "sparse_"]:
            setattr(torch.nn.init, name, bind_generator(getattr(torch.nn.init, name)))

        # Override the generator of every DataLoader to a fresh default generator.
        original_dataloader_init = torch.utils.data.DataLoader.__init__
        torch.utils.data.DataLoader.__init__ = lambda self, dataset, *args, **kwargs: original_dataloader_init(
            self, dataset, *args, **kwargs | {"generator": torch.Generator().manual_seed(seed)})

npfl138.__version__ module-attribute 

__version__ = '2425.14.2'