| 16 | |
| 17 | |
| 18 | class SequenceBase(IterableDataset[T_Data], ABC, ConfigTestableSubclass): |
| 19 | @classmethod |
| 20 | def name(cls) -> str: |
| 21 | """ |
| 22 | Assign a short name for the dataset class. By default will be the class name. |
| 23 | Overwrite this function if you want to create a more readable name used in `name` field in config. |
| 24 | """ |
| 25 | return cls.__name__ |
| 26 | |
| 27 | @abstractmethod |
| 28 | def __getitem__(self, local_index: int) -> T_Data: ... |
| 29 | |
| 30 | # No need to read further ### Implementation details below ############ |
| 31 | def __init__(self, length: int) -> None: |
| 32 | super().__init__() |
| 33 | self.origin_length: Final[int] = length |
| 34 | self.indices = np.arange(0, length, 1) |
| 35 | |
| 36 | def get_index(self, local_index: int) -> int: |
| 37 | """ |
| 38 | SequenceBase class supports masking / sampling of sequences. |
| 39 | The 'actual index' refer to the index in the original sequence, |
| 40 | In contrast with the 'logical index' (index after mask is applied) used by the user. |
| 41 | """ |
| 42 | return self.indices[local_index].item() |
| 43 | |
| 44 | @final |
| 45 | def clip(self, start_idx: int | None = None, end_idx: int | None = None, step: int | None = None) -> Self: |
| 46 | self.indices = self.indices[start_idx:end_idx:step] |
| 47 | return self |
| 48 | |
| 49 | def preload(self) -> "PreloadedSequence[T_Data]": |
| 50 | return PreloadedSequence(self) |
| 51 | |
| 52 | def transform(self, actions: list[Callable[[T_Data,], T_Data]] | Callable[[T_Data,], T_Data]) -> "TransformSequence[T_Data] | Self": |
| 53 | if isinstance(actions, list) and len(actions) == 0: return self |
| 54 | return TransformSequence(self, actions) |
| 55 | |
| 56 | def __len__(self) -> int: |
| 57 | return self.indices.size |
| 58 | |
| 59 | def __iter__(self) -> Generator[T_Data, None, None]: |
| 60 | for idx in range(len(self)): yield self[idx] |
| 61 | |
| 62 | def __repr__(self) -> str: |
| 63 | return f"{self.name()}(orig_len={self.origin_length}, clip_len={len(self)})" |
| 64 | |
| 65 | @staticmethod |
| 66 | def collate_fn(batch: list[T_Data]) -> T_Data: |
| 67 | """ |
| 68 | Collate function for DataLoader. |
| 69 | """ |
| 70 | return batch[0].collate(batch) |
| 71 | |
| 72 | @staticmethod |
| 73 | def config_dict2ns(cfg: SimpleNamespace | dict[str, Any]) -> SimpleNamespace: |
| 74 | if isinstance(cfg, SimpleNamespace): return cfg |
| 75 | return build_dynamic_config(cfg)[0] |
nothing calls this directly
no outgoing calls
no test coverage detected