Skip to content

FileSystemLogger

minnt.loggers.FileSystemLogger

Bases: BaseLogger

A file system logger interface.

Source code in minnt/loggers/filesystem_logger.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
 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
class FileSystemLogger(BaseLogger):
    """A file system logger interface."""

    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 close(self) -> None:
        if self._log_file:
            self._log_file.close()
            self._log_file = None

    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 _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(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.preprocess_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:
        print(self.format_config_as_text(config, epoch), 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")
            with self.graph_in_eval_mode(graph), torch.no_grad():
                traced = torch.jit.trace(graph, data, check_trace=False, 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")
        del traced
        gc.collect()  # Make sure the traced module is collected.
        return self

    def log_image(self, label: str, image: AnyArray, epoch: int, data_format: DataFormat = "HWC") -> Self:
        image = self.preprocess_image(image, data_format).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_metrics(self, logs: dict[str, float], epoch: int, description: str | None = None) -> Self:
        print(description or f"Epoch {epoch}", self.format_metrics(logs), file=self.get_file(), flush=True)
        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 
23
24
25
26
27
28
29
30
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

close 

close() -> None

Close the logger and release its resources.

Source code in minnt/loggers/filesystem_logger.py 
32
33
34
35
def close(self) -> None:
    if self._log_file:
        self._log_file.close()
        self._log_file = 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 
37
38
39
40
41
42
43
44
45
46
47
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 
68
69
70
71
72
73
74
75
76
77
78
79
def log_audio(self, label: str, audio: AnyArray, sample_rate: int, epoch: int) -> Self:
    audio = self.preprocess_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 
81
82
83
def log_config(self, config: dict[str, Any], epoch: int) -> Self:
    print(self.format_config_as_text(config, epoch), 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 
85
86
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, run it on the given data, or use any other mechanism they see fit.

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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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")
        with self.graph_in_eval_mode(graph), torch.no_grad():
            traced = torch.jit.trace(graph, data, check_trace=False, 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")
    del traced
    gc.collect()  # Make sure the traced module is collected.
    return self

log_image 

log_image(
    label: str, image: AnyArray, epoch: int, data_format: DataFormat = "HWC"
) -> 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 a PIL image or as an array of any of the following shapes (assuming "HWC" data format):

    • (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.

  • data_format (DataFormat, default: 'HWC' ) –

    The data format of the image specifying whether the channels are stored in the last dimension ("HWC", the default) or in the first dimension ("CHW"); ignored for a PIL image.

Source code in minnt/loggers/filesystem_logger.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def log_image(self, label: str, image: AnyArray, epoch: int, data_format: DataFormat = "HWC") -> Self:
    image = self.preprocess_image(image, data_format).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_metrics 

log_metrics(
    logs: dict[str, float], epoch: int, description: str | None = None
) -> Self

Log metrics collected during a given epoch, with an optional description.

Parameters:

  • logs (dict[str, float]) –

    A dictionary of logged metrics for the epoch.

  • epoch (int) –

    The epoch number at which the logs were collected.

  • description (str | None, default: None ) –

    An optional description of the logged metrics (used only by some loggers).

Source code in minnt/loggers/filesystem_logger.py 
123
124
125
def log_metrics(self, logs: dict[str, float], epoch: int, description: str | None = None) -> Self:
    print(description or f"Epoch {epoch}", self.format_metrics(logs), file=self.get_file(), flush=True)
    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 
127
128
129
130
131
132
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