test_netcdf_monitor.py 文件源码

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')