Skip to content

FileSystemLogger

A file system logger interface.

minnt.loggers.FileSystemLogger

Bases: Logger

Source code in minnt/loggers/filesystem_logger.py 
 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
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
class FileSystemLogger(Logger):
    def __init__(self, logdir: str) -> None:
        """Initialize the file system logger.

        Parameters:
          logdir: The root directory where the log files will be stored.
        """
        self._logdir: str = logdir
        self._log_file: TextIO | None = None

    def __del__(self) -> None:
        # Close the log file if it was opened.
        if self._log_file:
            self._log_file.close()

    def get_file(self) -> TextIO:
        """Possibly open and return log file object.

        Returns:
          file: The opened log file.
        """
        if not self._log_file:
            os.makedirs(self._logdir, exist_ok=True)
            self._log_file = open(os.path.join(self._logdir, "logs.txt"), "a", encoding="utf-8")

        return self._log_file

    def _sanitize_path(self, input_string: str) -> str:
        """Sanitize the given string to be safe for file paths.

        Parameters:
          input_string: The input string to sanitize.
        Returns:
          The sanitized string.
        """
        return re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", input_string)

    def _split_label(self, label: str) -> tuple[str, str]:
        """Split the given label into directory and base label, and sanitize them.

        Returns:
          directory: The directory part of the label.
          label: The base label.
        """
        directory, label = label.split(":", maxsplit=1) if ":" in label else ("train", label)
        directory, label = map(self._sanitize_path, (directory, label))
        return os.path.join(self._logdir, directory), label

    def _maybe_epoch(self, epoch: int) -> str:
        """Return epoch suffix if epoch is non-zero.

        Returns:
          suffix: The epoch suffix.
        """
        return f".{epoch}" if epoch else ""

    def log_audio(self, label: str, audio: AnyArray, sample_rate: int, epoch: int) -> Self:
        audio = self._process_audio(audio)

        directory, label = self._split_label(label)
        os.makedirs(directory, exist_ok=True)
        with wave.open(os.path.join(directory, f"{label}{self._maybe_epoch(epoch)}.wav"), "wb") as wav_file:
            wav_file.setsampwidth(2)  # 16 bits
            wav_file.setnchannels(audio.shape[-1])
            wav_file.setframerate(sample_rate)
            wav_file.writeframes(audio.numpy().tobytes())

        return self

    def log_config(self, config: dict[str, Any], epoch: int) -> Self:
        config = dict(sorted(config.items()))

        print("Config", f"epoch={epoch}", *[f"{k}={v}" for k, v in config.items()],
              file=self.get_file(), flush=True)

        os.makedirs(self._logdir, exist_ok=True)
        with open(os.path.join(self._logdir, f"config{self._maybe_epoch(epoch)}.json"), "w", encoding="utf-8") as file:
            json.dump(config, file, ensure_ascii=False, indent=2)

        return self

    def log_epoch(
        self, logs: dict[str, float], epoch: int, epochs: int | None = None, elapsed: float | None = None,
    ) -> Self:
        print(f"Epoch {epoch}" + (f"/{epochs}" if epochs is not None else ""),
              *[f"{elapsed:.1f}s"] if elapsed is not None else [],
              *[f"{k}={v:#.{0 < abs(v) < 2e-4 and '2e' or '4f'}}" for k, v in logs.items()],
              file=self.get_file(), flush=True)
        return self

    def log_figure(self, label: str, figure: Any, epoch: int, tight_layout: bool = True, close: bool = True) -> Self:
        return super().log_figure(label, figure, epoch, tight_layout, close)

    def log_graph(self, graph: torch.nn.Module, data: TensorOrTensors, epoch: int) -> Self:
        os.makedirs(self._logdir, exist_ok=True)
        with open(os.path.join(self._logdir, f"graph{self._maybe_epoch(epoch)}.txt"), "w", encoding="utf-8") as file:
            if isinstance(graph, torch.nn.Sequential):
                print("# Sequential Module", graph, file=file, sep="\n", end="\n\n")
            traced = torch.jit.trace(graph, data, strict=False)
            print("# Traced Code", traced.code, file=file, sep="\n", end="\n\n")
            print("# Traced Graph", traced.graph, file=file, sep="\n")
            print("# Traced Inlined Graph", traced.inlined_graph, file=file, sep="\n")
        return self

    def log_image(self, label: str, image: AnyArray, epoch: int) -> Self:
        image = self._process_image(image).numpy()

        directory, label = self._split_label(label)
        os.makedirs(directory, exist_ok=True)
        with open(os.path.join(directory, f"{label}{self._maybe_epoch(epoch)}.png"), "wb") as file:
            def add_block(chunk_type: bytes, data: bytes) -> None:
                file.write(struct.pack("!I", len(data)))
                data = chunk_type + data
                file.write(data)
                file.write(struct.pack("!I", zlib.crc32(data)))

            file.write(b"\x89PNG\r\n\x1a\n")
            add_block(b"IHDR", struct.pack(
                "!2I5B", image.shape[1], image.shape[0], 8, [0, 4, 2, 6][image.shape[2] - 1], 0, 0, 0))
            pixels = [b"\2" + (image[y] - (image[y - 1] if y else 0)).tobytes() for y in range(len(image))]
            add_block(b"IDAT", zlib.compress(b"".join(pixels), level=9))
            add_block(b"IEND", b"")

        return self

    def log_text(self, label: str, text: str, epoch: int) -> Self:
        directory, label = self._split_label(label)
        os.makedirs(directory, exist_ok=True)
        with open(os.path.join(directory, f"{label}{self._maybe_epoch(epoch)}.txt"), "w", encoding="utf-8") as file:
            file.write(text)
        return self

__init__ 

__init__(logdir: str) -> None

Initialize the file system logger.

Parameters:

  • logdir (str) –

    The root directory where the log files will be stored.

Source code in minnt/loggers/filesystem_logger.py 
22
23
24
25
26
27
28
29
def __init__(self, logdir: str) -> None:
    """Initialize the file system logger.

    Parameters:
      logdir: The root directory where the log files will be stored.
    """
    self._logdir: str = logdir
    self._log_file: TextIO | None = None

get_file 

get_file() -> TextIO

Possibly open and return log file object.

Returns:

  • file ( TextIO ) –

    The opened log file.

Source code in minnt/loggers/filesystem_logger.py 
36
37
38
39
40
41
42
43
44
45
46
def get_file(self) -> TextIO:
    """Possibly open and return log file object.

    Returns:
      file: The opened log file.
    """
    if not self._log_file:
        os.makedirs(self._logdir, exist_ok=True)
        self._log_file = open(os.path.join(self._logdir, "logs.txt"), "a", encoding="utf-8")

    return self._log_file

log_audio 

log_audio(label: str, audio: AnyArray, sample_rate: int, epoch: int) -> Self

Log the given audio with the given label at the given epoch.

Parameters:

  • label (str) –

    The label of the logged audio.

  • audio (AnyArray) –

    The audio to log, represented as an array with any of the following shapes:

    • (L,) of (L, 1) for mono audio,
    • (L, 2) for stereo audio.

    If the sample values are floating-point numbers, they are expected to be in the [-1, 1] range; otherwise, they are assumed to be in the [-32_768, 32_767] range.

  • sample_rate (int) –

    The sample rate of the audio.

  • epoch (int) –

    The epoch number at which the audio is logged.

Source code in minnt/loggers/filesystem_logger.py 
77
78
79
80
81
82
83
84
85
86
87
88
def log_audio(self, label: str, audio: AnyArray, sample_rate: int, epoch: int) -> Self:
    audio = self._process_audio(audio)

    directory, label = self._split_label(label)
    os.makedirs(directory, exist_ok=True)
    with wave.open(os.path.join(directory, f"{label}{self._maybe_epoch(epoch)}.wav"), "wb") as wav_file:
        wav_file.setsampwidth(2)  # 16 bits
        wav_file.setnchannels(audio.shape[-1])
        wav_file.setframerate(sample_rate)
        wav_file.writeframes(audio.numpy().tobytes())

    return self

log_config 

log_config(config: dict[str, Any], epoch: int) -> Self

Log the given configuration dictionary at the given epoch.

Parameters:

  • config (dict[str, Any]) –

    A JSON-serializable dictionary representing the configuration to log.

  • epoch (int) –

    The epoch number at which the configuration is logged.

Source code in minnt/loggers/filesystem_logger.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def log_config(self, config: dict[str, Any], epoch: int) -> Self:
    config = dict(sorted(config.items()))

    print("Config", f"epoch={epoch}", *[f"{k}={v}" for k, v in config.items()],
          file=self.get_file(), flush=True)

    os.makedirs(self._logdir, exist_ok=True)
    with open(os.path.join(self._logdir, f"config{self._maybe_epoch(epoch)}.json"), "w", encoding="utf-8") as file:
        json.dump(config, file, ensure_ascii=False, indent=2)

    return self

log_epoch 

log_epoch(
    logs: dict[str, float],
    epoch: int,
    epochs: int | None = None,
    elapsed: float | None = None,
) -> Self

Log metrics collected during a given epoch.

Parameters:

  • logs (dict[str, float]) –

    A dictionary of logged metrics for the epoch.

  • epoch (int) –

    The epoch number at which the logs were collected.

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

    The total number of epochs, if known.

  • elapsed (float | None, default: None ) –

    The time elapsed during the epoch, in seconds, if known.

Source code in minnt/loggers/filesystem_logger.py 
102
103
104
105
106
107
108
109
def log_epoch(
    self, logs: dict[str, float], epoch: int, epochs: int | None = None, elapsed: float | None = None,
) -> Self:
    print(f"Epoch {epoch}" + (f"/{epochs}" if epochs is not None else ""),
          *[f"{elapsed:.1f}s"] if elapsed is not None else [],
          *[f"{k}={v:#.{0 < abs(v) < 2e-4 and '2e' or '4f'}}" for k, v in logs.items()],
          file=self.get_file(), flush=True)
    return self

log_figure 

log_figure(
    label: str,
    figure: Any,
    epoch: int,
    tight_layout: bool = True,
    close: bool = True,
) -> Self

Log the given matplotlib Figure with the given label at the given epoch.

Parameters:

  • label (str) –

    The label of the logged image.

  • figure (Any) –

    A matplotlib Figure.

  • epoch (int) –

    The epoch number at which the image is logged.

  • tight_layout (bool, default: True ) –

    Whether to apply tight layout to the figure before logging it.

  • close (bool, default: True ) –

    Whether to close the figure after logging it.

Source code in minnt/loggers/filesystem_logger.py 
111
112
def log_figure(self, label: str, figure: Any, epoch: int, tight_layout: bool = True, close: bool = True) -> Self:
    return super().log_figure(label, figure, epoch, tight_layout, close)

log_graph 

log_graph(graph: Module, data: TensorOrTensors, epoch: int) -> Self

Log the given computation graph by tracing it with the given data.

Alternatively, loggers may choose to log the graph using TorchScript or other mechanisms.

Parameters:

  • graph (Module) –

    The computation graph to log, represented as a PyTorch module.

  • data (TensorOrTensors) –

    The input data to use for tracing the computation graph.

  • epoch (int) –

    The epoch number at which the computation graph is logged.

Source code in minnt/loggers/filesystem_logger.py 
114
115
116
117
118
119
120
121
122
123
def log_graph(self, graph: torch.nn.Module, data: TensorOrTensors, epoch: int) -> Self:
    os.makedirs(self._logdir, exist_ok=True)
    with open(os.path.join(self._logdir, f"graph{self._maybe_epoch(epoch)}.txt"), "w", encoding="utf-8") as file:
        if isinstance(graph, torch.nn.Sequential):
            print("# Sequential Module", graph, file=file, sep="\n", end="\n\n")
        traced = torch.jit.trace(graph, data, strict=False)
        print("# Traced Code", traced.code, file=file, sep="\n", end="\n\n")
        print("# Traced Graph", traced.graph, file=file, sep="\n")
        print("# Traced Inlined Graph", traced.inlined_graph, file=file, sep="\n")
    return self

log_image 

log_image(label: str, image: AnyArray, epoch: int) -> Self

Log the given image with the given label at the given epoch.

Parameters:

  • label (str) –

    The label of the logged image.

  • image (AnyArray) –

    The image to log, represented as an array, which can have any of the following shapes:

    • (H, W) or (H, W, 1) for grayscale images,
    • (H, W, 2) for grayscale images with alpha channel,
    • (H, W, 3) for RGB images,
    • (H, W, 4) for RGBA images.

    If the pixel values are floating-point numbers, they are expected to be in the [0, 1] range; otherwise, they are assumed to be in the [0, 255] range.

  • epoch (int) –

    The epoch number at which the image is logged.

Source code in minnt/loggers/filesystem_logger.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def log_image(self, label: str, image: AnyArray, epoch: int) -> Self:
    image = self._process_image(image).numpy()

    directory, label = self._split_label(label)
    os.makedirs(directory, exist_ok=True)
    with open(os.path.join(directory, f"{label}{self._maybe_epoch(epoch)}.png"), "wb") as file:
        def add_block(chunk_type: bytes, data: bytes) -> None:
            file.write(struct.pack("!I", len(data)))
            data = chunk_type + data
            file.write(data)
            file.write(struct.pack("!I", zlib.crc32(data)))

        file.write(b"\x89PNG\r\n\x1a\n")
        add_block(b"IHDR", struct.pack(
            "!2I5B", image.shape[1], image.shape[0], 8, [0, 4, 2, 6][image.shape[2] - 1], 0, 0, 0))
        pixels = [b"\2" + (image[y] - (image[y - 1] if y else 0)).tobytes() for y in range(len(image))]
        add_block(b"IDAT", zlib.compress(b"".join(pixels), level=9))
        add_block(b"IEND", b"")

    return self

log_text 

log_text(label: str, text: str, epoch: int) -> Self

Log the given text with the given label at the given epoch.

Parameters:

  • label (str) –

    The label of the logged text.

  • text (str) –

    The text to log.

  • epoch (int) –

    The epoch number at which the text is logged.

Source code in minnt/loggers/filesystem_logger.py 
146
147
148
149
150
151
def log_text(self, label: str, text: str, epoch: int) -> Self:
    directory, label = self._split_label(label)
    os.makedirs(directory, exist_ok=True)
    with open(os.path.join(directory, f"{label}{self._maybe_epoch(epoch)}.txt"), "w", encoding="utf-8") as file:
        file.write(text)
    return self