Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_edit_zmetadata(self):
optimize_dataset(TEST_CUBE_ZARR, unchunk_coords=True, output_path=TEST_CUBE_ZARR_OPTIMIZED)
edit_metadata(TEST_CUBE_ZARR_OPTIMIZED, metadata_path=TEST_NEW_META_YML, in_place=False,
output_path=TEST_CUBE_ZARR_OPTIMIZED_EDIT, monitor=print)
ds1 = zarr.open(TEST_CUBE_ZARR)
ds2 = zarr.convenience.open_consolidated(TEST_CUBE_ZARR_OPTIMIZED_EDIT)
self.assertEqual(ds1.__len__(), ds2.__len__())
self.assertEqual(ds1.attrs.__getitem__('start_date'), ds2.attrs.__getitem__('start_date'))
self.assertEqual('happiness', ds2['conc_chl'].attrs.__getitem__('units'))
self.assertNotIn('creator_name', ds1.attrs.keys())
self.assertIn('creator_name', ds2.attrs.keys())
def _unchunk_vars(dataset_path: str, var_names: List[str]):
for var_name in var_names:
var_path = os.path.join(dataset_path, var_name)
# Optimization: if "shape" and "chunks" are equal in ${var}/.zarray, we are done
var_array_info_path = os.path.join(var_path, '.zarray')
with open(var_array_info_path, 'r') as fp:
var_array_info = json.load(fp)
if var_array_info.get('shape') == var_array_info.get('chunks'):
continue
# Open array and remove chunks from the data
var_array = zarr.convenience.open_array(var_path, 'r+')
if var_array.shape != var_array.chunks:
# TODO (forman): Fully loading data is inefficient and dangerous for large arrays.
# Instead save unchunked to temp and replace existing chunked array dir with temp.
# Fully load data and attrs so we no longer depend on files
data = np.array(var_array)
attributes = var_array.attrs.asdict()
# Save array data
zarr.convenience.save_array(var_path, data, chunks=False, fill_value=var_array.fill_value)
# zarr.convenience.save_array() does not seem save user attributes (file ".zattrs" not written),
# therefore we must modify attrs explicitly:
var_array = zarr.convenience.open_array(var_path, 'r+')
var_array.attrs.update(attributes)
if os.path.exists(output_path):
raise exception_type(f'Output path already exists.')
if not in_place:
shutil.copytree(input_path, output_path)
if unchunk_coords:
if isinstance(unchunk_coords, str):
var_names = (unchunk_coords,)
elif isinstance(unchunk_coords, bool):
var_names = None
else:
var_names = tuple(unchunk_coords)
unchunk_dataset(output_path, var_names=var_names, coords_only=True)
zarr.convenience.consolidate_metadata(output_path)