| from __future__ import annotations |
|
|
| from typing import TYPE_CHECKING, Union |
|
|
|
|
| if TYPE_CHECKING: |
| from collections.abc import Sequence |
|
|
| from . import Dim |
|
|
| import torch |
|
|
|
|
| |
| |
| class DimEntry: |
| |
| data: Union[Dim, int] |
|
|
| def __init__(self, data: Union[Dim, int, None] = None) -> None: |
| from . import Dim |
|
|
| if type(data) is int: |
| if data >= 0: |
| raise AssertionError(f"Expected negative int, got {data}") |
| elif data is None: |
| data = 0 |
| else: |
| if not isinstance(data, Dim): |
| raise AssertionError(f"Expected Dim, got {type(data)}") |
| self.data = data |
|
|
| def __eq__(self, other: object) -> bool: |
| if not isinstance(other, DimEntry): |
| return False |
| |
| |
| if self.is_positional() and other.is_positional(): |
| |
| return self.data == other.data |
| elif not self.is_positional() and not other.is_positional(): |
| |
| return self.data is other.data |
| else: |
| |
| return False |
|
|
| def is_positional(self) -> bool: |
| return type(self.data) is int and self.data < 0 |
|
|
| def is_none(self) -> bool: |
| |
| from . import Dim |
|
|
| if isinstance(self.data, Dim): |
| |
| return False |
| else: |
| |
| return self.data == 0 |
|
|
| def position(self) -> int: |
| if not isinstance(self.data, int): |
| raise AssertionError(f"Expected int, got {type(self.data)}") |
| return self.data |
|
|
| def dim(self) -> Dim: |
| if isinstance(self.data, int): |
| raise AssertionError("Expected Dim, got int") |
| return self.data |
|
|
| def __repr__(self) -> str: |
| return repr(self.data) |
|
|
|
|
| def ndim_of_levels(levels: Sequence[DimEntry]) -> int: |
| r = 0 |
| for l in levels: |
| if l.is_positional(): |
| r += 1 |
| return r |
|
|
|
|
| def _match_levels( |
| tensor: torch.Tensor, |
| from_levels: list[DimEntry], |
| to_levels: list[DimEntry], |
| drop_levels: bool = False, |
| ) -> torch.Tensor: |
| """ |
| Reshape a tensor to match target levels using as_strided. |
| |
| Args: |
| tensor: Input tensor to reshape |
| from_levels: Current levels of the tensor |
| to_levels: Target levels to match |
| drop_levels: If True, missing dimensions are assumed to have stride 0 |
| |
| Returns: |
| Reshaped tensor |
| """ |
| if from_levels == to_levels: |
| return tensor |
|
|
| sizes = tensor.size() |
| strides = tensor.stride() |
|
|
| if not drop_levels: |
| if len(from_levels) > len(to_levels): |
| raise AssertionError("Cannot expand dimensions without drop_levels") |
|
|
| new_sizes = [] |
| new_strides = [] |
|
|
| for level in to_levels: |
| |
| try: |
| idx = from_levels.index(level) |
| except ValueError: |
| |
| if level.is_positional(): |
| new_sizes.append(1) |
| else: |
| new_sizes.append(level.dim().size) |
| new_strides.append(0) |
| else: |
| new_sizes.append(sizes[idx]) |
| new_strides.append(strides[idx]) |
|
|
| return tensor.as_strided(new_sizes, new_strides, tensor.storage_offset()) |
|
|