def test_triangles(self):
"""Test the creation of triangles"""
ds = psyd.open_dataset(os.path.join(bt.test_dir, 'icon_test.nc'))
decoder = psyd.CFDecoder(ds)
var = ds.t2m[0, 0]
var.attrs.pop('grid_type', None)
self.assertTrue(decoder.is_triangular(var))
self.assertTrue(decoder.is_unstructured(var))
triangles = decoder.get_triangles(var)
self.assertEqual(len(triangles.triangles), var.size)
# Test for correct falsification
ds = psyd.open_dataset(os.path.join(bt.test_dir, 'test-t2m-u-v.nc'))
decoder = psyd.CFDecoder(ds)
self.assertFalse(decoder.is_triangular(ds.t2m[0, 0]))
self.assertFalse(decoder.is_unstructured(ds.t2m[0, 0]))
python类open_dataset()的实例源码
def test_update_01_isel(self):
"""test the update of a single array through the isel method"""
ds = psyd.open_dataset(bt.get_file('test-t2m-u-v.nc'))
arr = ds.psy.t2m.psy[0, 0, 0]
arr.attrs['test'] = 4
self.assertNotIn('test', ds.t2m.attrs)
self.assertIs(arr.psy.base, ds)
self.assertEqual(dict(arr.psy.idims), {'time': 0, 'lev': 0, 'lat': 0,
'lon': slice(None)})
# update to next time step
arr.psy.update(time=1)
self.assertEqual(arr.time, ds.time[1])
self.assertEqual(arr.values.tolist(),
ds.t2m[1, 0, 0, :].values.tolist())
self.assertEqual(dict(arr.psy.idims), {'time': 1, 'lev': 0, 'lat': 0,
'lon': slice(None)})
self.assertNotIn('test', ds.t2m.attrs)
self.assertIn('test', arr.attrs)
self.assertEqual(arr.test, 4)
def test_update_02_sel(self):
"""test the update of a single array through the sel method"""
ds = psyd.open_dataset(bt.get_file('test-t2m-u-v.nc'))
arr = ds.psy.t2m.psy[0, 0, 0]
arr.attrs['test'] = 4
self.assertNotIn('test', ds.t2m.attrs)
self.assertIs(arr.psy.base, ds)
self.assertEqual(dict(arr.psy.idims), {'time': 0, 'lev': 0, 'lat': 0,
'lon': slice(None)})
# update to next time step
arr.psy.update(time='1979-02-28T18:00', method='nearest')
self.assertEqual(arr.time, ds.time[1])
self.assertEqual(arr.values.tolist(),
ds.t2m[1, 0, 0, :].values.tolist())
self.assertEqual(dict(arr.psy.idims), {'time': 1, 'lev': 0, 'lat': 0,
'lon': slice(None)})
self.assertNotIn('test', ds.t2m.attrs)
self.assertIn('test', arr.attrs)
self.assertEqual(arr.test, 4)
def test_update_03_isel_concat(self):
"""test the update of a concatenated array through the isel method"""
ds = psyd.open_dataset(bt.get_file('test-t2m-u-v.nc'))[['t2m', 'u']]
arr = ds.psy.to_array().psy.isel(time=0, lev=0, lat=0)
arr.attrs['test'] = 4
self.assertNotIn('test', ds.t2m.attrs)
arr.name = 'something'
self.assertIs(arr.psy.base, ds)
self.assertEqual(dict(arr.psy.idims), {'time': 0, 'lev': 0, 'lat': 0,
'lon': slice(None)})
self.assertEqual(arr.coords['variable'].values.tolist(), ['t2m', 'u'])
# update to next time step
arr.psy.update(time=1)
self.assertEqual(arr.time, ds.time[1])
self.assertEqual(arr.coords['variable'].values.tolist(), ['t2m', 'u'])
self.assertEqual(arr.values.tolist(),
ds[['t2m', 'u']].to_array()[
:, 1, 0, 0, :].values.tolist())
self.assertEqual(dict(arr.psy.idims), {'time': 1, 'lev': 0, 'lat': 0,
'lon': slice(None)})
self.assertNotIn('test', ds.t2m.attrs)
self.assertIn('test', arr.attrs)
self.assertEqual(arr.test, 4)
self.assertEqual(arr.name, 'something')
def test_update_05_1variable(self):
"""Test to change the variable"""
ds = psyd.open_dataset(bt.get_file('test-t2m-u-v.nc'))
arr = ds.psy.t2m.psy[0, 0, 0]
arr.attrs['test'] = 4
self.assertNotIn('test', ds.t2m.attrs)
self.assertIs(arr.psy.base, ds)
self.assertEqual(dict(arr.psy.idims), {'time': 0, 'lev': 0, 'lat': 0,
'lon': slice(None)})
# update to next time step
arr.psy.update(name='u', time=1)
self.assertEqual(arr.time, ds.time[1])
self.assertEqual(arr.name, 'u')
self.assertEqual(arr.values.tolist(),
ds.u[1, 0, 0, :].values.tolist())
self.assertEqual(dict(arr.psy.idims), {'time': 1, 'lev': 0, 'lat': 0,
'lon': slice(None)})
self.assertNotIn('test', ds.t2m.attrs)
self.assertIn('test', arr.attrs)
self.assertEqual(arr.test, 4)
def test_update_06_2variables(self):
"""test the change of the variable of a concatenated array"""
ds = psyd.open_dataset(bt.get_file('test-t2m-u-v.nc'))
arr = ds[['t2m', 'u']].to_array().isel(time=0, lev=0, lat=0)
arr.attrs['test'] = 4
self.assertNotIn('test', ds.t2m.attrs)
arr.name = 'something'
arr.psy.base = ds
self.assertEqual(dict(arr.psy.idims), {'time': 0, 'lev': 0, 'lat': 0,
'lon': slice(None)})
self.assertEqual(arr.coords['variable'].values.tolist(), ['t2m', 'u'])
# update to next time step
arr.psy.update(time=1, name=['u', 'v'])
self.assertEqual(arr.time, ds.time[1])
self.assertEqual(arr.coords['variable'].values.tolist(), ['u', 'v'])
self.assertEqual(arr.values.tolist(),
ds[['u', 'v']].to_array()[
:, 1, 0, 0, :].values.tolist())
self.assertEqual(dict(arr.psy.idims), {'time': 1, 'lev': 0, 'lat': 0,
'lon': slice(None)})
self.assertNotIn('test', ds.t2m.attrs)
self.assertIn('test', arr.attrs)
self.assertEqual(arr.test, 4)
self.assertEqual(arr.name, 'something')
def test_netcdf_monitor_single_time_all_vars():
try:
assert not os.path.isfile('out.nc')
monitor = NetCDFMonitor('out.nc')
monitor.store(state)
assert not os.path.isfile('out.nc') # not set to write on store
monitor.write()
assert os.path.isfile('out.nc')
with xr.open_dataset('out.nc') as ds:
assert len(ds.data_vars.keys()) == 2
assert 'air_temperature' in ds.data_vars.keys()
assert ds.data_vars['air_temperature'].attrs['units'] == 'degK'
assert tuple(ds.data_vars['air_temperature'].shape) == (1, nx, ny, nz)
assert 'air_pressure' in ds.data_vars.keys()
assert ds.data_vars['air_pressure'].attrs['units'] == 'Pa'
assert tuple(ds.data_vars['air_pressure'].shape) == (1, nx, ny, nz)
assert len(ds['time']) == 1
assert ds['time'][0] == np.datetime64(state['time'])
finally: # make sure we remove the output file
if os.path.isfile('out.nc'):
os.remove('out.nc')
def test_netcdf_monitor_single_write_on_store():
try:
assert not os.path.isfile('out.nc')
monitor = NetCDFMonitor('out.nc', write_on_store=True)
monitor.store(state)
assert os.path.isfile('out.nc')
with xr.open_dataset('out.nc') as ds:
assert len(ds.data_vars.keys()) == 2
assert 'air_temperature' in ds.data_vars.keys()
assert ds.data_vars['air_temperature'].attrs['units'] == 'degK'
assert tuple(ds.data_vars['air_temperature'].shape) == (1, nx, ny, nz)
assert 'air_pressure' in ds.data_vars.keys()
assert ds.data_vars['air_pressure'].attrs['units'] == 'Pa'
assert tuple(ds.data_vars['air_pressure'].shape) == (1, nx, ny, nz)
assert len(ds['time']) == 1
assert ds['time'][0] == np.datetime64(state['time'])
finally: # make sure we remove the output file
if os.path.isfile('out.nc'):
os.remove('out.nc')
def test_fractional_cover(sr_filepath, fc_filepath):
print(sr_filepath)
print(fc_filepath)
sr_dataset = open_dataset(sr_filepath)
measurements = [
{'name': 'PV', 'dtype': 'int8', 'nodata': -1, 'units': 'percent'},
{'name': 'NPV', 'dtype': 'int8', 'nodata': -1, 'units': 'percent'},
{'name': 'BS', 'dtype': 'int8', 'nodata': -1, 'units': 'percent'},
{'name': 'UE', 'dtype': 'int8', 'nodata': -1, 'units': '1'}
]
fc_dataset = fractional_cover(sr_dataset, measurements)
assert set(fc_dataset.data_vars.keys()) == {m['name'] for m in measurements}
validation_ds = open_dataset(fc_filepath)
assert validation_ds == fc_dataset
assert validation_ds.equals(fc_dataset)
def read_netcdf(data_handle, domain=None, iter_dims=['lat', 'lon'],
start=None, stop=None, calendar='standard',
var_dict=None) -> xr.Dataset:
"""Read in a NetCDF file"""
ds = xr.open_dataset(data_handle)
if var_dict is not None:
ds.rename(var_dict, inplace=True)
if start is not None and stop is not None:
ds = ds.sel(time=slice(start, stop))
dates = ds.indexes['time']
ds['day_of_year'] = xr.Variable(('time', ), dates.dayofyear)
if domain is not None:
ds = ds.sel(**{d: domain[d] for d in iter_dims})
out = ds.load()
ds.close()
return out
def convert_hdf5_to_netcdf4(input_file, output_file):
ds = xr.open_dataset(input_file)
chl = ds['chlor_a'].to_dataset()
chl.to_netcdf(output_file, format='NETCDF4_CLASSIC')
def test_version_metadata_with_streaming(self, api, opener):
np.random.seed(123)
times = pd.date_range('2000-01-01', '2001-12-31', name='time')
annual_cycle = np.sin(2 * np.pi * (times.dayofyear / 365.25 - 0.28))
base = 10 + 15 * np.array(annual_cycle).reshape(-1, 1)
tmin_values = base + 3 * np.random.randn(annual_cycle.size, 3)
tmax_values = base + 3 * np.random.randn(annual_cycle.size, 3)
ds = xr.Dataset({'tmin': (('time', 'location'), tmin_values),
'tmax': (('time', 'location'), tmax_values)},
{'time': times, 'location': ['IA', 'IN', 'IL']})
var = api.create('streaming_test')
with var.get_local_path(
bumpversion='patch',
dependencies={'arch1': '0.1.0', 'arch2': '0.2.0'}) as f:
ds.to_netcdf(f)
ds.close()
assert var.get_history()[-1]['dependencies']['arch2'] == '0.2.0'
tmin_values = base + 10 * np.random.randn(annual_cycle.size, 3)
ds.update({'tmin': (('time', 'location'), tmin_values)})
with var.get_local_path(
bumpversion='patch',
dependencies={'arch1': '0.1.0', 'arch2': '1.2.0'}) as f:
with xr.open_dataset(f) as ds:
mem = ds.load()
ds.close()
mem.to_netcdf(f)
assert var.get_history()[-1]['dependencies']['arch2'] == '1.2.0'
assert var.get_history()[-1][
'checksum'] != var.get_history()[-2]['checksum']
def decode_coords(ds, gridfile=None, inplace=True):
"""
Sets the coordinates and bounds in a dataset
This static method sets those coordinates and bounds that are marked
marked in the netCDF attributes as coordinates in :attr:`ds` (without
deleting them from the variable attributes because this information is
necessary for visualizing the data correctly)
Parameters
----------
ds: xarray.Dataset
The dataset to decode
gridfile: str
The path to a separate grid file or a xarray.Dataset instance which
may store the coordinates used in `ds`
inplace: bool, optional
If True, `ds` is modified in place
Returns
-------
xarray.Dataset
`ds` with additional coordinates"""
def add_attrs(obj):
if 'coordinates' in obj.attrs:
extra_coords.update(obj.attrs['coordinates'].split())
if 'bounds' in obj.attrs:
extra_coords.add(obj.attrs['bounds'])
if gridfile is not None and not isinstance(gridfile, xr.Dataset):
gridfile = open_dataset(gridfile)
extra_coords = set(ds.coords)
for k, v in six.iteritems(ds.variables):
add_attrs(v)
add_attrs(ds)
if gridfile is not None:
ds = ds.update({k: v for k, v in six.iteritems(gridfile.variables)
if k in extra_coords}, inplace=inplace)
ds = ds.set_coords(extra_coords.intersection(ds.variables),
inplace=inplace)
return ds
def decode_coords(ds, gridfile=None, inplace=True):
"""
Reimplemented to set the mesh variables as coordinates
Parameters
----------
%(CFDecoder.decode_coords.parameters)s
Returns
-------
%(CFDecoder.decode_coords.returns)s"""
extra_coords = set(ds.coords)
for var in six.itervalues(ds.variables):
if 'mesh' in var.attrs:
mesh = var.attrs['mesh']
if mesh not in extra_coords:
extra_coords.add(mesh)
try:
mesh_var = ds.variables[mesh]
except KeyError:
warn('Could not find mesh variable %s' % mesh)
continue
if 'node_coordinates' in mesh_var.attrs:
extra_coords.update(
mesh_var.attrs['node_coordinates'].split())
if 'face_node_connectivity' in mesh_var.attrs:
extra_coords.add(
mesh_var.attrs['face_node_connectivity'])
if gridfile is not None and not isinstance(gridfile, xr.Dataset):
gridfile = open_dataset(gridfile)
ds = ds.update({k: v for k, v in six.iteritems(gridfile.variables)
if k in extra_coords}, inplace=inplace)
ds = ds.set_coords(extra_coords.intersection(ds.variables),
inplace=inplace)
return ds
def open_dataset(filename_or_obj, decode_cf=True, decode_times=True,
decode_coords=True, engine=None, gridfile=None, **kwargs):
"""
Open an instance of :class:`xarray.Dataset`.
This method has the same functionality as the :func:`xarray.open_dataset`
method except that is supports an additional 'gdal' engine to open
gdal Rasters (e.g. GeoTiffs) and that is supports absolute time units like
``'day as %Y%m%d.%f'`` (if `decode_cf` and `decode_times` are True).
Parameters
----------
%(xarray.open_dataset.parameters.no_engine)s
engine: {'netcdf4', 'scipy', 'pydap', 'h5netcdf', 'gdal'}, optional
Engine to use when reading netCDF files. If not provided, the default
engine is chosen based on available dependencies, with a preference for
'netcdf4'.
%(CFDecoder.decode_coords.parameters.gridfile)s
Returns
-------
xarray.Dataset
The dataset that contains the variables from `filename_or_obj`"""
# use the absolute path name (is saver when saving the project)
if isstring(filename_or_obj) and os.path.exists(filename_or_obj):
filename_or_obj = os.path.abspath(filename_or_obj)
if engine == 'gdal':
from psyplot.gdal_store import GdalStore
filename_or_obj = GdalStore(filename_or_obj)
engine = None
ds = xr.open_dataset(filename_or_obj, decode_cf=decode_cf,
decode_coords=False, engine=engine,
decode_times=decode_times, **kwargs)
if decode_cf:
ds = CFDecoder.decode_ds(
ds, decode_coords=decode_coords, decode_times=decode_times,
gridfile=gridfile, inplace=True)
return ds
def _open_ds_from_store(fname, store_mod=None, store_cls=None, **kwargs):
"""Open a dataset and return it"""
if isinstance(fname, xr.Dataset):
return fname
if store_mod is not None and store_cls is not None:
fname = getattr(import_module(store_mod), store_cls)(fname)
return open_dataset(fname, **kwargs)
def _test_coord(self, func_name, name, uname=None, name2d=False,
circ_name=None):
def check_ds(name):
self.assertEqual(getattr(d, func_name)(ds.t2m).name, name)
if name2d:
self.assertEqual(getattr(d, func_name)(ds.t2m_2d).name, name)
else:
self.assertIsNone(getattr(d, func_name)(ds.t2m_2d))
if six.PY3:
# Test whether the warning is raised if the decoder finds
# multiple dimensions
with self.assertWarnsRegex(RuntimeWarning,
'multiple matches'):
coords = 'time lat lon lev x y latitude longitude'.split()
ds.t2m.attrs.pop('coordinates', None)
for dim in 'xytz':
getattr(d, dim).update(coords)
for coord in set(coords).intersection(ds.coords):
ds.coords[coord].attrs.pop('axis', None)
getattr(d, func_name)(ds.t2m)
uname = uname or name
circ_name = circ_name or name
ds = psyd.open_dataset(os.path.join(bt.test_dir, 'test-t2m-u-v.nc'))
d = psyd.CFDecoder(ds)
check_ds(name)
ds.close()
ds = psyd.open_dataset(os.path.join(bt.test_dir, 'icon_test.nc'))
d = psyd.CFDecoder(ds)
check_ds(uname)
ds.close()
ds = psyd.open_dataset(
os.path.join(bt.test_dir, 'circumpolar_test.nc'))
d = psyd.CFDecoder(ds)
check_ds(circ_name)
ds.close()
def test_standardization(self):
"""Test the :meth:`psyplot.data.CFDecoder.standardize_dims` method"""
ds = psyd.open_dataset(os.path.join(bt.test_dir, 'test-t2m-u-v.nc'))
decoder = psyd.CFDecoder(ds)
dims = {'time': 1, 'lat': 2, 'lon': 3, 'lev': 4}
replaced = decoder.standardize_dims(ds.t2m, dims)
for dim, rep in [('time', 't'), ('lat', 'y'), ('lon', 'x'),
('lev', 'z')]:
self.assertIn(rep, replaced)
self.assertEqual(replaced[rep], dims[dim],
msg="Wrong value for %s (%s-) dimension" % (
dim, rep))
def test_idims(self):
"""Test the extraction of the slicers of the dimensions"""
ds = psyd.open_dataset(bt.get_file('test-t2m-u-v.nc'))
arr = ds.t2m[1:, 1]
arr.psy.init_accessor(base=ds)
dims = arr.psy.idims
for dim in ['time', 'lev', 'lat', 'lon']:
self.assertEqual(
psyd.safe_list(ds[dim][dims[dim]]),
psyd.safe_list(arr.coords[dim]),
msg="Slice %s for dimension %s is wrong!" % (dims[dim], dim))
# test with unknown dimensions
if xr.__version__ >= '0.9':
ds = ds.drop('time')
arr = ds.t2m[1:, 1]
arr.psy.init_accessor(base=ds)
if not six.PY2:
with self.assertWarnsRegex(UserWarning, 'time'):
dims = arr.psy.idims
l = psyd.ArrayList.from_dataset(
ds, name='t2m', time=slice(1, None), lev=85000., method='sel')
arr = l[0]
dims = arr.psy.idims
for dim in ['time', 'lev', 'lat', 'lon']:
if dim == 'time':
self.assertEqual(dims[dim], slice(1, 5, 1))
else:
self.assertEqual(
psyd.safe_list(ds[dim][dims[dim]]),
psyd.safe_list(arr.coords[dim]),
msg="Slice %s for dimension %s is wrong!" % (dims[dim],
dim))
def test_is_circumpolar(self):
"""Test whether the is_circumpolar method works"""
ds = psyd.open_dataset(os.path.join(bt.test_dir,
'circumpolar_test.nc'))
decoder = psyd.CFDecoder(ds)
self.assertTrue(decoder.is_circumpolar(ds.t2m))
# test for correct falsification
ds = psyd.open_dataset(os.path.join(bt.test_dir, 'icon_test.nc'))
decoder = psyd.CFDecoder(ds)
self.assertFalse(decoder.is_circumpolar(ds.t2m))
def test_get_decoder(self):
"""Test to get the right decoder"""
ds = psyd.open_dataset(bt.get_file('simple_triangular_grid_si0.nc'))
d = psyd.CFDecoder.get_decoder(ds, ds.Mesh2_fcvar)
self.assertIsInstance(d, psyd.UGridDecoder)
return ds, d
def test_auto_update(self):
"""Test the :attr:`psyplot.plotter.Plotter.no_auto_update` attribute"""
ds = psyd.open_dataset(bt.get_file('test-t2m-u-v.nc'))
arr = ds.psy.t2m.psy[0, 0, 0]
arr.psy.init_accessor(auto_update=False)
arr.psy.update(time=1)
self.assertEqual(arr.time, ds.time[0])
arr.psy.start_update()
self.assertEqual(arr.time, ds.time[1])
arr.psy.no_auto_update = False
arr.psy.update(time=2)
self.assertEqual(arr.time, ds.time[2])
def test_array_info(self):
variables, coords = self._from_dataset_test_variables
variables['v4'] = variables['v3'].copy()
ds = xr.Dataset(variables, coords)
fname = osp.relpath(bt.get_file('test-t2m-u-v.nc'), '.')
ds2 = xr.open_dataset(fname)
l = ds.psy.create_list(
name=[['v1', ['v3', 'v4']], ['v1', 'v2']], prefer_list=True)
l.extend(ds2.psy.create_list(name=['t2m'], x=0, t=1),
new_name=True)
self.assertEqual(l.array_info(engine='netCDF4'), OrderedDict([
# first list contating an array with two variables
('arr0', OrderedDict([
('arr0', {'dims': {'t': slice(None), 'x': slice(None)},
'attrs': OrderedDict(), 'store': (None, None),
'name': 'v1', 'fname': None}),
('arr1', {'dims': {'y': slice(None)},
'attrs': OrderedDict(), 'store': (None, None),
'name': [['v3', 'v4']], 'fname': None}),
('attrs', OrderedDict())])),
# second list with two arrays containing each one variable
('arr1', OrderedDict([
('arr0', {'dims': {'t': slice(None), 'x': slice(None)},
'attrs': OrderedDict(), 'store': (None, None),
'name': 'v1', 'fname': None}),
('arr1', {'dims': {'y': slice(None), 'x': slice(None)},
'attrs': OrderedDict(), 'store': (None, None),
'name': 'v2', 'fname': None}),
('attrs', OrderedDict())])),
# last array from real dataset
('arr2', {'dims': {'z': slice(None), 'y': slice(None),
't': 1, 'x': 0},
'attrs': ds2.t2m.attrs,
'store': ('xarray.backends.netCDF4_',
'NetCDF4DataStore'),
'name': 't2m', 'fname': fname}),
('attrs', OrderedDict())]))
return l
def test_open_dataset(self):
fname = self.test_to_netcdf()
ref_ds = self._test_ds
ds = psyd.open_dataset(fname)
self.assertEqual(
pd.to_datetime(ds.time.values).tolist(),
pd.to_datetime(ref_ds.time.values).tolist())
def _test_engine(self, engine):
from importlib import import_module
fname = self.fname
ds = psyd.open_dataset(fname, engine=engine).load()
self.assertEqual(ds.psy.filename, fname)
store_mod, store = ds.psy.data_store
# try to load the dataset
mod = import_module(store_mod)
ds2 = psyd.open_dataset(getattr(mod, store)(fname))
ds.close()
ds2.close()
ds.psy.filename = None
dumped_fname, dumped_store_mod, dumped_store = psyd.get_filename_ds(
ds, dump=True, engine=engine, paths=True)
self.assertTrue(dumped_fname)
self.assertTrue(osp.exists(dumped_fname),
msg='Missing %s' % fname)
self.assertEqual(dumped_store_mod, store_mod)
self.assertEqual(dumped_store, store)
ds.close()
ds.psy.filename = None
os.remove(dumped_fname)
dumped_fname, dumped_store_mod, dumped_store = psyd.get_filename_ds(
ds, dump=True, engine=engine, paths=dumped_fname)
self.assertTrue(dumped_fname)
self.assertTrue(osp.exists(dumped_fname),
msg='Missing %s' % fname)
self.assertEqual(dumped_store_mod, store_mod)
self.assertEqual(dumped_store, store)
ds.close()
os.remove(dumped_fname)
def test_netcdf_monitor_multiple_times_batched_all_vars():
time_list = [
datetime(2013, 7, 20, 0),
datetime(2013, 7, 20, 6),
datetime(2013, 7, 20, 12),
]
current_state = state.copy()
try:
assert not os.path.isfile('out.nc')
monitor = NetCDFMonitor('out.nc')
for time in time_list:
current_state['time'] = time
monitor.store(current_state)
assert not os.path.isfile('out.nc') # not set to write on store
monitor.write()
assert os.path.isfile('out.nc')
with xr.open_dataset('out.nc') as ds:
assert len(ds.data_vars.keys()) == 2
assert 'air_temperature' in ds.data_vars.keys()
assert ds.data_vars['air_temperature'].attrs['units'] == 'degK'
assert tuple(ds.data_vars['air_temperature'].shape) == (
len(time_list), nx, ny, nz)
assert 'air_pressure' in ds.data_vars.keys()
assert ds.data_vars['air_pressure'].attrs['units'] == 'Pa'
assert tuple(ds.data_vars['air_pressure'].shape) == (
len(time_list), nx, ny, nz)
assert len(ds['time']) == len(time_list)
assert np.all(
ds['time'].values == [np.datetime64(time) for time in time_list])
finally: # make sure we remove the output file
if os.path.isfile('out.nc'):
os.remove('out.nc')
def test_netcdf_monitor_multiple_times_sequential_all_vars():
time_list = [
datetime(2013, 7, 20, 0),
datetime(2013, 7, 20, 6),
datetime(2013, 7, 20, 12),
]
current_state = state.copy()
try:
assert not os.path.isfile('out.nc')
monitor = NetCDFMonitor('out.nc')
for time in time_list:
current_state['time'] = time
monitor.store(current_state)
monitor.write()
assert os.path.isfile('out.nc')
with xr.open_dataset('out.nc') as ds:
assert len(ds.data_vars.keys()) == 2
assert 'air_temperature' in ds.data_vars.keys()
assert ds.data_vars['air_temperature'].attrs['units'] == 'degK'
assert tuple(ds.data_vars['air_temperature'].shape) == (
len(time_list), nx, ny, nz)
assert 'air_pressure' in ds.data_vars.keys()
assert ds.data_vars['air_pressure'].attrs['units'] == 'Pa'
assert tuple(ds.data_vars['air_pressure'].shape) == (
len(time_list), nx, ny, nz)
assert len(ds['time']) == len(time_list)
assert np.all(
ds['time'].values == [np.datetime64(time) for time in time_list])
finally: # make sure we remove the output file
if os.path.isfile('out.nc'):
os.remove('out.nc')
def test_netcdf_monitor_multiple_times_sequential_all_vars_timedelta():
time_list = [
timedelta(hours=0),
timedelta(hours=6),
timedelta(hours=12),
]
current_state = state.copy()
try:
assert not os.path.isfile('out.nc')
monitor = NetCDFMonitor('out.nc')
for time in time_list:
current_state['time'] = time
monitor.store(current_state)
monitor.write()
assert os.path.isfile('out.nc')
with xr.open_dataset('out.nc') as ds:
assert len(ds.data_vars.keys()) == 2
assert 'air_temperature' in ds.data_vars.keys()
assert ds.data_vars['air_temperature'].attrs['units'] == 'degK'
assert tuple(ds.data_vars['air_temperature'].shape) == (
len(time_list), nx, ny, nz)
assert 'air_pressure' in ds.data_vars.keys()
assert ds.data_vars['air_pressure'].attrs['units'] == 'Pa'
assert tuple(ds.data_vars['air_pressure'].shape) == (
len(time_list), nx, ny, nz)
assert len(ds['time']) == len(time_list)
assert np.all(
ds['time'].values == [np.timedelta64(time) for time in time_list])
finally: # make sure we remove the output file
if os.path.isfile('out.nc'):
os.remove('out.nc')
def test_netcdf_monitor_multiple_times_batched_single_var():
time_list = [
datetime(2013, 7, 20, 0),
datetime(2013, 7, 20, 6),
datetime(2013, 7, 20, 12),
]
current_state = state.copy()
try:
assert not os.path.isfile('out.nc')
monitor = NetCDFMonitor('out.nc', store_names=['air_temperature'])
for time in time_list:
current_state['time'] = time
monitor.store(current_state)
assert not os.path.isfile('out.nc') # not set to write on store
monitor.write()
assert os.path.isfile('out.nc')
with xr.open_dataset('out.nc') as ds:
assert len(ds.data_vars.keys()) == 1
assert 'air_temperature' in ds.data_vars.keys()
assert ds.data_vars['air_temperature'].attrs['units'] == 'degK'
assert tuple(ds.data_vars['air_temperature'].shape) == (
len(time_list), nx, ny, nz)
assert len(ds['time']) == len(time_list)
assert np.all(
ds['time'].values == [np.datetime64(time) for time in time_list])
finally: # make sure we remove the output file
if os.path.isfile('out.nc'):
os.remove('out.nc')
def test_netcdf_monitor_multiple_write_on_store():
time_list = [
datetime(2013, 7, 20, 0),
datetime(2013, 7, 20, 6),
datetime(2013, 7, 20, 12),
]
current_state = state.copy()
try:
assert not os.path.isfile('out.nc')
monitor = NetCDFMonitor('out.nc', write_on_store=True)
for time in time_list:
current_state['time'] = time
monitor.store(current_state)
assert os.path.isfile('out.nc')
with xr.open_dataset('out.nc') as ds:
assert len(ds.data_vars.keys()) == 2
assert 'air_temperature' in ds.data_vars.keys()
assert ds.data_vars['air_temperature'].attrs['units'] == 'degK'
assert tuple(ds.data_vars['air_temperature'].shape) == (
len(time_list), nx, ny, nz)
assert 'air_pressure' in ds.data_vars.keys()
assert ds.data_vars['air_pressure'].attrs['units'] == 'Pa'
assert tuple(ds.data_vars['air_pressure'].shape) == (
len(time_list), nx, ny, nz)
assert len(ds['time']) == len(time_list)
assert np.all(
ds['time'].values == [np.datetime64(time) for time in time_list])
finally: # make sure we remove the output file
if os.path.isfile('out.nc'):
os.remove('out.nc')
