test_netcdf_monitor.py 文件源码

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