ユーティリティ¶

ユーティリティ関数を提供します。

共通ユーティリティ (`ml_networks.utils`)¶

save_blosc2¶

save_blosc2 ¶

save_blosc2(path, x)

Save numpy array with blosc2 compression.

Args:

path : str Path to save. x : np.ndarray Numpy array to save.

Examples:

>>> save_blosc2("test.blosc2", np.random.randn(10, 10))

Source code in src/ml_networks/utils.py

def save_blosc2(path: str, x: np.ndarray) -> None:
    """Save numpy array with blosc2 compression.

    Args:
    -----
    path : str
        Path to save.
    x : np.ndarray
        Numpy array to save.

    Examples
    --------
    >>> save_blosc2("test.blosc2", np.random.randn(10, 10))

    """
    Path(path).write_bytes(blosc2.pack_array2(x))

load_blosc2¶

load_blosc2 ¶

load_blosc2(path)

Load numpy array with blosc2 compression.

Args:

path : str Path to load.

Returns:

Type	Description
`ndarray`	Numpy array.

Examples:

>>> data = load_blosc2("test.blosc2")
>>> type(data)
<class 'numpy.ndarray'>

Source code in src/ml_networks/utils.py

def load_blosc2(path: str) -> np.ndarray:
    """Load numpy array with blosc2 compression.

    Args:
    -----
    path : str
        Path to load.

    Returns
    -------
    np.ndarray
        Numpy array.

    Examples
    --------
    >>> data = load_blosc2("test.blosc2")
    >>> type(data)
    <class 'numpy.ndarray'>
    """
    return blosc2.unpack_array2(Path(path).read_bytes())

determine_loader¶

determine_loader ¶

determine_loader(data, seed, batch_size, shuffle=True, collate_fn=None)

Determine DataLoader with fixed seed.

Args:

data : Dataset Dataset to load. seed : int Random seed. batch_size : int Batch size. shuffle : bool Whether to shuffle data. Default is True. collate_fn : callable Collate function. Default is None.

Returns:

Type	Description
`DataLoader`	DataLoader with fixed seed.

Source code in src/ml_networks/utils.py

def determine_loader(
    data: Dataset,
    seed: int,
    batch_size: int,
    shuffle: bool = True,
    collate_fn: Callable | None = None,
) -> DataLoader:
    """
    Determine DataLoader with fixed seed.

    Args:
    -----
    data : Dataset
        Dataset to load.
    seed : int
        Random seed.
    batch_size : int
        Batch size.
    shuffle : bool
        Whether to shuffle data. Default is True.
    collate_fn : callable
        Collate function. Default is None.

    Returns
    -------
    DataLoader
        DataLoader with fixed seed.
    """
    g = torch.Generator()  # type: ignore[attr-defined]
    g.manual_seed(seed)
    return DataLoader(
        data,
        batch_size=batch_size,
        shuffle=shuffle,
        worker_init_fn=seed_worker,
        generator=g,
        num_workers=2,
        pin_memory=False,
        collate_fn=collate_fn,
    )

seed_worker¶

seed_worker ¶

seed_worker(_worker_id)

DataLoaderのworkerの固定.

Dataloaderの乱数固定にはgeneratorの固定も必要らしい

Source code in src/ml_networks/utils.py

def seed_worker(_worker_id: int) -> None:
    """
    DataLoaderのworkerの固定.

    Dataloaderの乱数固定にはgeneratorの固定も必要らしい
    """
    worker_seed = torch.initial_seed() % 2**32
    pl.seed_everything(worker_seed)

conv_out¶

畳み込み層の出力サイズを計算します。

conv_out ¶

conv_out(h_in, padding, kernel_size, stride, dilation=1)

Calculate the output size of convolutional layer.

Args:

h_in : int Input size. padding : int Padding size. kernel_size : int Kernel size. stride : int Stride size. dilation : int Dilation size. Default is 1.

Returns:

Type	Description
`int`	Output size.

Examples:

>>> conv_out(32, 1, 3, 1)
32
>>> conv_out(32, 1, 3, 2)
16
>>> conv_out(32, 1, 3, 1, 2)
30

Source code in src/ml_networks/utils.py

def conv_out(h_in: int, padding: int, kernel_size: int, stride: int, dilation: int = 1) -> int:
    """
    Calculate the output size of convolutional layer.

    Args:
    -----
    h_in : int
        Input size.
    padding : int
        Padding size.
    kernel_size : int
        Kernel size.
    stride : int
        Stride size.
    dilation : int
        Dilation size. Default is 1.

    Returns
    -------
    int
        Output size.

    Examples
    --------
    >>> conv_out(32, 1, 3, 1)
    32
    >>> conv_out(32, 1, 3, 2)
    16
    >>> conv_out(32, 1, 3, 1, 2)
    30
    """
    return int((h_in + 2.0 * padding - dilation * (kernel_size - 1.0) - 1.0) / stride + 1.0)

conv_transpose_out¶

転置畳み込み層の出力サイズを計算します。

conv_transpose_out ¶

conv_transpose_out(h_in, padding, kernel_size, stride, dilation=1, output_padding=0)

Calculate the output size of transposed convolutional layer.

Parameters:

Name	Type	Description	Default
`h_in`	`int`	Input size.	required
`padding`	`int`	Padding size.	required
`kernel_size`	`int`	Kernel size.	required
`stride`	`int`	Stride size.	required
`dilation`	`int`	Dilation size. Default is 1.	`1`
`output_padding`	`int`	Output padding size. Default is 0.	`0`

Returns:

Type	Description
`int`	Output size.

Examples:

>>> conv_transpose_out(32, 1, 3, 1)
32
>>> conv_transpose_out(32, 1, 3, 2)
63
>>> conv_transpose_out(32, 1, 3, 1, 2)
34
>>> conv_transpose_out(32, 1, 3, 1, 1, 1)
33

Source code in src/ml_networks/utils.py

def conv_transpose_out(
    h_in: int,
    padding: int,
    kernel_size: int,
    stride: int,
    dilation: int = 1,
    output_padding: int = 0,
) -> int:
    """
    Calculate the output size of transposed convolutional layer.

    Parameters
    ----------
    h_in : int
        Input size.
    padding : int
        Padding size.
    kernel_size : int
        Kernel size.
    stride : int
        Stride size.
    dilation : int
        Dilation size. Default is 1.
    output_padding : int
        Output padding size. Default is 0.

    Returns
    -------
    int
        Output size.

    Examples
    --------
    >>> conv_transpose_out(32, 1, 3, 1)
    32
    >>> conv_transpose_out(32, 1, 3, 2)
    63
    >>> conv_transpose_out(32, 1, 3, 1, 2)
    34
    >>> conv_transpose_out(32, 1, 3, 1, 1, 1)
    33


    """
    return (h_in - 1) * stride - 2 * padding + dilation * (kernel_size - 1) + output_padding + 1

output_padding¶

転置畳み込み層の出力パディングを計算します。

output_padding ¶

output_padding(h_in, h_out, padding, kernel_size, stride, dilation=1)

Calculate the output padding size of transposed convolutional layer.

Args:

h_in : int Input size. h_out : int Output size. padding : int Padding size. kernel_size : int Kernel size. stride : int Stride size. dilation : int Dilation size. Default is 1.

Returns:

Type	Description
`int`	Output padding size.

Examples:

>>> output_padding(32, 32, 1, 3, 1)
0
>>> output_padding(32, 16, 1, 3, 2)
1
>>> output_padding(32, 30, 1, 3, 1, 2)
0

Source code in src/ml_networks/utils.py

def output_padding(h_in: int, h_out: int, padding: int, kernel_size: int, stride: int, dilation: int = 1) -> int:
    """
    Calculate the output padding size of transposed convolutional layer.

    Args:
    -----
    h_in : int
        Input size.
    h_out : int
        Output size.
    padding : int
        Padding size.
    kernel_size : int
        Kernel size.
    stride : int
        Stride size.
    dilation : int
        Dilation size. Default is 1.

    Returns
    -------
    int
        Output padding size.

    Examples
    --------
    >>> output_padding(32, 32, 1, 3, 1)
    0
    >>> output_padding(32, 16, 1, 3, 2)
    1
    >>> output_padding(32, 30, 1, 3, 1, 2)
    0

    """
    return h_in - (h_out - 1) * stride + 2 * padding - dilation * (kernel_size - 1) - 1

PyTorch固有ユーティリティ (`ml_networks.torch.torch_utils`)¶

get_optimizer¶

get_optimizer ¶

get_optimizer(param, name, **kwargs)

Get optimizer from torch.optim or pytorch_optimizer.

Args:

param : Iterator[nn.Parameter] Parameters of models to optimize. name : str Optimizer name. kwargs : dict Optimizer arguments(settings).

Returns:

Type	Description
`Optimizer`

Examples:

>>> get_optimizer([nn.Parameter(torch.randn(1, 3))], "Adam", lr=0.01)
Adam (
Parameter Group 0
    amsgrad: False
    betas: (0.9, 0.999)
    capturable: False
    differentiable: False
    eps: 1e-08
    foreach: None
    fused: None
    lr: 0.01
    maximize: False
    weight_decay: 0
)

Source code in src/ml_networks/torch/torch_utils.py

def get_optimizer(
    param: Iterator[nn.Parameter],
    name: str,
    **kwargs: float | str | bool,
) -> torch.optim.Optimizer:
    """
    Get optimizer from torch.optim or pytorch_optimizer.

    Args:
    -----
    param : Iterator[nn.Parameter]
        Parameters of models to optimize.
    name : str
        Optimizer name.
    kwargs : dict
        Optimizer arguments(settings).

    Returns
    -------
    torch.optim.Optimizer

    Examples
    --------
    >>> get_optimizer([nn.Parameter(torch.randn(1, 3))], "Adam", lr=0.01)
    Adam (
    Parameter Group 0
        amsgrad: False
        betas: (0.9, 0.999)
        capturable: False
        differentiable: False
        eps: 1e-08
        foreach: None
        fused: None
        lr: 0.01
        maximize: False
        weight_decay: 0
    )
    """
    if hasattr(schedulefree, name):
        optimizer = getattr(schedulefree, name)
    elif hasattr(torch.optim, name):
        optimizer = getattr(torch.optim, name)
    elif hasattr(pytorch_optimizer, name):
        optimizer = getattr(pytorch_optimizer, name)
    else:
        msg = f"Optimizer {name} is not implemented in torch.optim or pytorch_optimizer, schedulefree. "
        msg += "Please check the name and capitalization."
        raise NotImplementedError(msg)
    return optimizer(param, **kwargs)

torch_fix_seed¶

torch_fix_seed ¶

torch_fix_seed(seed=42)

乱数を固定する関数.

References

https://qiita.com/north_redwing/items/1e153139125d37829d2d

Source code in src/ml_networks/torch/torch_utils.py

def torch_fix_seed(seed: int = 42) -> None:
    """
    乱数を固定する関数.

    References
    ----------
    - https://qiita.com/north_redwing/items/1e153139125d37829d2d
    """
    random.seed(seed)
    pl.seed_everything(seed, workers=True)
    torch.set_float32_matmul_precision("medium")
    torch.backends.cudnn.benchmark = False
    torch.backends.cudnn.deterministic = True

gumbel_softmax¶

gumbel_softmax ¶

gumbel_softmax(inputs, dim, temperature=1.0)

Gumbel softmax function with temperature. This prevents overflow and underflow.

Parameters:

Name	Type	Description	Default
`inputs`	`Tensor`	Input tensor.	required
`dim`	`int`	Dimension to apply softmax.	required
`temperature`	`float`	Temperature. Default is 1.0.	`1.0`

Returns:

Type	Description
`Tensor`	Gumbel softmaxed tensor.

Raises:

Type	Description
`ValueError`	If the gumbel_softmax is inf or nan.

Source code in src/ml_networks/torch/torch_utils.py

def gumbel_softmax(
    inputs: torch.Tensor,
    dim: int,
    temperature: float = 1.0,
) -> torch.Tensor:
    """
    Gumbel softmax function with temperature. This prevents overflow and underflow.

    Parameters
    ----------
    inputs : torch.Tensor
        Input tensor.
    dim : int
        Dimension to apply softmax.
    temperature : float
        Temperature. Default is 1.0.

    Returns
    -------
    torch.Tensor
        Gumbel softmaxed tensor.

    Raises
    ------
    ValueError
        If the gumbel_softmax is inf or nan.
    """
    x = inputs - torch.max(inputs.detach(), dim=-1, keepdim=True)[0]
    x = F.gumbel_softmax(x, dim=dim, tau=temperature, hard=True)
    if torch.isinf(x).any() or torch.isnan(x).any():
        msg = "gumbel_softmax is inf or nan"
        raise ValueError(msg)
    return x

softmax¶

softmax ¶

softmax(inputs, dim, temperature=1.0)

Softmax function with temperature. This prevents overflow and underflow.

Parameters:

Name	Type	Description	Default
`inputs`	`Tensor`	Input tensor.	required
`dim`	`int`	Dimension to apply softmax.	required
`temperature`	`float`	Temperature. Default is 1.0.	`1.0`

Returns:

Type	Description
`Tensor`	Softmaxed tensor.

Raises:

Type	Description
`ValueError`	If the softmax is inf or nan.

Source code in src/ml_networks/torch/torch_utils.py

def softmax(
    inputs: torch.Tensor,
    dim: int,
    temperature: float = 1.0,
) -> torch.Tensor:
    """
    Softmax function with temperature. This prevents overflow and underflow.

    Parameters
    ----------
    inputs : torch.Tensor
        Input tensor.
    dim : int
        Dimension to apply softmax.
    temperature : float
        Temperature. Default is 1.0.

    Returns
    -------
    torch.Tensor
        Softmaxed tensor.

    Raises
    ------
    ValueError
        If the softmax is inf or nan.
    """
    x = inputs / temperature
    x = torch.exp(F.log_softmax(x, dim=dim))
    if torch.isinf(x).any() or torch.isnan(x).any():
        msg = "softmax is inf or nan"
        raise ValueError(msg)
    return x

MinMaxNormalize¶

MinMaxNormalize ¶

MinMaxNormalize(min_val, max_val, old_min=0.0, old_max=1.0)

Bases: Normalize

MinMax 正規化変換.

MinMaxNormalize の初期化.

Args:

min_val : float 最小値. max_val : float 最大値. old_min : float 元の最小値. old_max : float 元の最大値.

Source code in src/ml_networks/torch/torch_utils.py

def __init__(self, min_val: float, max_val: float, old_min: float = 0.0, old_max: float = 1.0) -> None:
    """
    MinMaxNormalize の初期化.

    Args:
    -----
    min_val : float
        最小値.
    max_val : float
        最大値.
    old_min : float
        元の最小値.
    old_max : float
        元の最大値.


    """
    scale = (max_val - min_val) / (old_max - old_min)
    shift = min_val - old_min * scale  # new = scale·x + shift

    #   Normalize does (x - mean)/std  =>  x · (1/std)  - mean/std
    mean = -shift / scale  # invert the affine form
    std = 1.0 / scale

    super().__init__(mean=[mean] * 3, std=[std] * 3)
    self.min = min_val
    self.max = max_val

Attributes¶

max `instance-attribute` ¶

max = max_val

min `instance-attribute` ¶

min = min_val

Functions¶

SoftmaxTransformation¶

SoftmaxTransformation ¶

SoftmaxTransformation(cfg)

Softmax 変換クラス.

SoftmaxTransformation の初期化.

Args:

cfg : SoftmaxTransConfig SoftmaxTransformation の設定.

Source code in src/ml_networks/torch/torch_utils.py

def __init__(
    self,
    cfg: SoftmaxTransConfig,
) -> None:
    """
    SoftmaxTransformation の初期化.

    Args:
    -----
    cfg : SoftmaxTransConfig
        SoftmaxTransformation の設定.
    """
    super().__init__()
    self.vector = cfg.vector
    self.sigma = cfg.sigma
    self.n_ignore = cfg.n_ignore
    self.max = cfg.max
    self.min = cfg.min
    self.k = torch.linspace(self.min, self.max, self.vector)

Attributes¶

k `instance-attribute` ¶

k = linspace(min, max, vector)

max `instance-attribute` ¶

max = max

min `instance-attribute` ¶

min = min

n_ignore `instance-attribute` ¶

n_ignore = n_ignore

sigma `instance-attribute` ¶

sigma = sigma

vector `instance-attribute` ¶

vector = vector

Functions¶

call ¶

__call__(x)

Source code in src/ml_networks/torch/torch_utils.py

def __call__(self, x: torch.Tensor) -> torch.Tensor:
    return self.transform(x)

get_transformed_dim ¶

get_transformed_dim(dim)

Source code in src/ml_networks/torch/torch_utils.py

def get_transformed_dim(self, dim: int) -> int:
    return (dim - self.n_ignore) * self.vector + self.n_ignore

inverse ¶

inverse(x)

SoftmaxTransformation の逆変換.

Args:

x : torch.Tensor 入力テンソル.

Returns:

Type	Description
`Tensor`	出力テンソル.

Source code in src/ml_networks/torch/torch_utils.py

def inverse(self, x: torch.Tensor) -> torch.Tensor:
    """
    SoftmaxTransformation の逆変換.

    Args:
    -----
    x : torch.Tensor
        入力テンソル.

    Returns
    -------
    torch.Tensor
        出力テンソル.
    """
    *batch, dim = x.shape
    x = x.reshape(-1, dim)
    if self.n_ignore:
        data, ignored = x[:, : -self.n_ignore], x[:, -self.n_ignore :]
    else:
        data = x

    data = data.reshape([len(data), -1, self.vector])

    data = rearrange(data, "b d v -> v b d")

    data = torch.stack([data[v] * self.k[v] for v in range(self.vector)]).sum(dim=0)

    data = torch.cat([data, ignored], dim=-1) if self.n_ignore else data
    return data.reshape(*batch, -1)

transform ¶

transform(x)

SoftmaxTransformation の実行.

Args:

x : torch.Tensor 入力テンソル.

Returns:

Type	Description
`Tensor`	出力テンソル.

Examples:

>>> trans = SoftmaxTransformation(SoftmaxTransConfig(vector=16, sigma=0.01, n_ignore=1, min=-1.0, max=1.0))
>>> x = torch.randn(2, 3, 4)
>>> transformed = trans(x)
>>> transformed.shape
torch.Size([2, 3, 49])

>>> trans = SoftmaxTransformation(SoftmaxTransConfig(vector=11, sigma=0.05, n_ignore=0, min=-1.0, max=1.0))
>>> x = torch.randn(2, 3, 4)
>>> transformed = trans(x)
>>> transformed.shape
torch.Size([2, 3, 44])

Source code in src/ml_networks/torch/torch_utils.py

def transform(self, x: torch.Tensor) -> torch.Tensor:
    """
    SoftmaxTransformation の実行.

    Args:
    -----
    x : torch.Tensor
        入力テンソル.

    Returns
    -------
    torch.Tensor
        出力テンソル.

    Examples
    --------
    >>> trans = SoftmaxTransformation(SoftmaxTransConfig(vector=16, sigma=0.01, n_ignore=1, min=-1.0, max=1.0))
    >>> x = torch.randn(2, 3, 4)
    >>> transformed = trans(x)
    >>> transformed.shape
    torch.Size([2, 3, 49])

    >>> trans = SoftmaxTransformation(SoftmaxTransConfig(vector=11, sigma=0.05, n_ignore=0, min=-1.0, max=1.0))
    >>> x = torch.randn(2, 3, 4)
    >>> transformed = trans(x)
    >>> transformed.shape
    torch.Size([2, 3, 44])

    """
    *batch, dim = x.shape
    x = x.reshape(-1, dim)
    if self.n_ignore:
        data, ignored = x[:, : -self.n_ignore], x[:, -self.n_ignore :]
    else:
        data = x

    negative = torch.stack([torch.exp((-((data - self.k[v]) ** 2)) / self.sigma) for v in range(self.vector)])
    negative_sum = negative.sum(dim=0)

    transformed = negative / (negative_sum + 1e-8)
    transformed = rearrange(transformed, "v b d -> b (d v)")

    transformed = torch.cat([transformed, ignored], dim=-1) if self.n_ignore else transformed
    return transformed.reshape(*batch, self.get_transformed_dim(dim))

ユーティリティ¶

共通ユーティリティ (ml_networks.utils)¶

save_blosc2¶

save_blosc2 ¶

load_blosc2¶

load_blosc2 ¶

determine_loader¶

determine_loader ¶

seed_worker¶

seed_worker ¶

conv_out¶

conv_out ¶

conv_transpose_out¶

conv_transpose_out ¶

output_padding¶

output_padding ¶

PyTorch固有ユーティリティ (ml_networks.torch.torch_utils)¶

get_optimizer¶

get_optimizer ¶

torch_fix_seed¶

torch_fix_seed ¶

gumbel_softmax¶

gumbel_softmax ¶

softmax¶

softmax ¶

MinMaxNormalize¶

MinMaxNormalize ¶

Attributes¶

max instance-attribute ¶

min instance-attribute ¶

Functions¶

SoftmaxTransformation¶

SoftmaxTransformation ¶

Attributes¶

k instance-attribute ¶

max instance-attribute ¶

min instance-attribute ¶

n_ignore instance-attribute ¶

sigma instance-attribute ¶

vector instance-attribute ¶

Functions¶

__call__ ¶

get_transformed_dim ¶

inverse ¶

transform ¶

共通ユーティリティ (`ml_networks.utils`)¶

PyTorch固有ユーティリティ (`ml_networks.torch.torch_utils`)¶

max `instance-attribute` ¶

min `instance-attribute` ¶

k `instance-attribute` ¶

max `instance-attribute` ¶

min `instance-attribute` ¶

n_ignore `instance-attribute` ¶

sigma `instance-attribute` ¶

vector `instance-attribute` ¶

call ¶