Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _download_chunk(self, cloudpath):
chunk = self._storage.get_or_none(cloudpath)
if chunk:
chunk = self._codec.decode(chunk)
else:
chunk = np.zeros(shape=self.chunk, dtype=self.dtype)
bbox = Bbox.from_filename(cloudpath)
return chunk, bbox
def _chunkify(self, cloudpaths, requested_bbox, item):
chunks = []
for path in cloudpaths:
cloudchunk = Bbox.from_filename(path)
intersection = Bbox.intersection(cloudchunk, requested_bbox)
chunk_slices = (intersection - cloudchunk.minpt).to_slices()
item_slices = (intersection - requested_bbox.minpt).to_slices()
chunk = None
if np.any(np.array(intersection.to_shape()) != np.array(self.chunk)):
chunk, _ = self._download_chunk(path)
else:
chunk = np.zeros(shape=self.chunk, dtype=self.dtype)
chunk.setflags(write=1)
chunk[chunk_slices] = item[item_slices]
chunks.append(chunk)
return zip(cloudpaths, chunks)