test_nwm_data_read_streamflow.py 文件源码

def file_to_read_streamflow_setup(request):
    ids = [2, 4, 6]
    flows = [1.3, -9999.0, 5.1]
    date = '2017-04-29_04:00:00'
    flows = ma.masked_array(flows, mask=[0, 1, 0])  # explicit mask
    with Dataset(_file_to_read_streamflow, 'w') as nc:
        nc.model_output_valid_time = date
        dim = nc.createDimension('feature_id', 3)
        id_var = nc.createVariable('feature_id', 'i', ('feature_id',))
        id_var[:] = ids
        flow_var = nc.createVariable('streamflow', 'f', ('feature_id',),
                                     fill_value=-9999.0)
        flow_var[:] = flows
    def file_to_read_streamflow_teardown():
        os.remove(_file_to_read_streamflow)
    request.addfinalizer(file_to_read_streamflow_teardown)