(self, batch_size: int)
| 147 | self.idx = 0 |
| 148 | |
| 149 | def read_batch(self, batch_size: int) -> Optional[np.ndarray]: |
| 150 | if self.idx >= self.shape[0]: |
| 151 | return None |
| 152 | |
| 153 | bs = min(batch_size, self.shape[0] - self.idx) |
| 154 | self.idx += bs |
| 155 | |
| 156 | if self.dtype.itemsize == 0: |
| 157 | return np.ndarray([bs, *self.shape[1:]], dtype=self.dtype) |
| 158 | |
| 159 | read_count = bs * np.prod(self.shape[1:]) |
| 160 | read_size = int(read_count * self.dtype.itemsize) |
| 161 | data = _read_bytes(self.arr_f, read_size, "array data") |
| 162 | return np.frombuffer(data, dtype=self.dtype).reshape([bs, *self.shape[1:]]) |
| 163 | |
| 164 | |
| 165 | class MemoryNpzArrayReader(NpzArrayReader): |
nothing calls this directly
no test coverage detected