(self, pid: Any)
| 784 | self.zip_file = zip_file |
| 785 | |
| 786 | def persistent_load(self, pid: Any) -> Any: |
| 787 | assert pid[0] == 'storage' |
| 788 | assert isinstance(pid[1], LazyStorageKind) |
| 789 | data_type = pid[1].data_type |
| 790 | filename_stem = pid[2] |
| 791 | filename = f'{self.data_base_path}/{filename_stem}' |
| 792 | info = self.zip_file.getinfo(filename) |
| 793 | |
| 794 | def load(offset: int, elm_count: int) -> NDArray: |
| 795 | dtype = data_type.dtype |
| 796 | fp = self.zip_file.open(info) |
| 797 | fp.seek(offset * dtype.itemsize) |
| 798 | size = elm_count * dtype.itemsize |
| 799 | data = fp.read(size) |
| 800 | assert len(data) == size |
| 801 | return np.frombuffer(data, dtype) |
| 802 | description = f'storage data_type={data_type} path-in-zip={filename} path={self.zip_file.filename}' |
| 803 | return LazyStorage(load=load, kind=pid[1], description=description) |
| 804 | |
| 805 | @staticmethod |
| 806 | def lazy_rebuild_tensor_v2(storage: Any, storage_offset: Any, size: Any, stride: Any, |
nothing calls this directly
no test coverage detected