def test_returns_numpy_array_using_alias(self):
T_array = np.zeros([2, 3, 4], dtype=np.float64) + 280.
property_dictionary = {
'air_temperature': {
'units': 'degK',
'dims': ['x', 'y', 'z'],
'alias': 'T',
},
}
state = {
'air_temperature': DataArray(
T_array,
dims=['x', 'y', 'z'],
attrs={'units': 'degK'},
),
}
return_value = get_numpy_arrays_with_properties(state, property_dictionary)
assert isinstance(return_value, dict)
assert len(return_value.keys()) == 1
assert isinstance(return_value['T'], np.ndarray)
assert np.byte_bounds(return_value['T']) == np.byte_bounds(
T_array)
assert return_value['T'].base is T_array
python类byte_bounds()的实例源码
def test_returns_scalar_array(self):
T_array = np.array(0.)
property_dictionary = {
'air_temperature': {
'units': 'degK',
'dims': [],
},
}
state = {
'air_temperature': DataArray(
T_array,
dims=[],
attrs={'units': 'degK'},
),
}
return_value = get_numpy_arrays_with_properties(state, property_dictionary)
assert isinstance(return_value, dict)
assert len(return_value.keys()) == 1
assert isinstance(return_value['air_temperature'], np.ndarray)
assert np.byte_bounds(return_value['air_temperature']) == np.byte_bounds(
T_array)
def test_scalar_becomes_multidimensional_array(self):
T_array = np.array(0.)
property_dictionary = {
'air_temperature': {
'units': 'degK',
'dims': ['z'],
},
}
state = {
'air_temperature': DataArray(
T_array,
dims=[],
attrs={'units': 'degK'},
),
}
return_value = get_numpy_arrays_with_properties(state, property_dictionary)
assert isinstance(return_value, dict)
assert len(return_value.keys()) == 1
assert isinstance(return_value['air_temperature'], np.ndarray)
assert len(return_value['air_temperature'].shape) == 1
assert np.byte_bounds(return_value['air_temperature']) == np.byte_bounds(
T_array)
assert return_value['air_temperature'].base is T_array
def test_collects_wildcard_dimension(self):
set_direction_names(z=['mid_levels'])
T_array = np.zeros([2, 3, 4], dtype=np.float64) + 280.
property_dictionary = {
'air_temperature': {
'units': 'degK',
'dims': ['x', 'y', 'z'],
},
}
state = {
'air_temperature': DataArray(
T_array,
dims=['x', 'y', 'mid_levels'],
attrs={'units': 'degK'},
),
}
return_value = get_numpy_arrays_with_properties(state, property_dictionary)
assert isinstance(return_value, dict)
assert len(return_value.keys()) == 1
assert isinstance(return_value['air_temperature'], np.ndarray)
assert np.byte_bounds(return_value['air_temperature']) == np.byte_bounds(
T_array)
assert return_value['air_temperature'].base is T_array
assert return_value['air_temperature'].shape == (2, 3, 4)
def test_creates_length_1_dimensions(self):
T_array = np.zeros([4], dtype=np.float64) + 280.
property_dictionary = {
'air_temperature': {
'dims': ['x', 'y', 'z'],
'units': 'degK',
},
}
state = {
'air_temperature': DataArray(
T_array,
dims=['z'],
attrs={'units': 'degK'},
),
}
return_value = get_numpy_arrays_with_properties(state, property_dictionary)
assert isinstance(return_value, dict)
assert len(return_value.keys()) == 1
assert isinstance(return_value['air_temperature'], np.ndarray)
assert np.byte_bounds(
return_value['air_temperature']) == np.byte_bounds(
T_array)
assert return_value['air_temperature'].base is T_array
assert return_value['air_temperature'].shape == (1, 1, 4)
def test_get_numpy_array_3d_no_change():
array = DataArray(
np.random.randn(2, 3, 4),
dims=['x', 'y', 'z'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['x', 'y', 'z'])
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert np.all(numpy_array == array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_3d_reverse():
array = DataArray(
np.random.randn(2, 3, 4),
dims=['x', 'y', 'z'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['z', 'y', 'x'])
assert numpy_array.shape == (4, 3, 2)
assert np.all(np.transpose(numpy_array, (2, 1, 0)) == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_2d_reverse():
array = DataArray(
np.random.randn(2, 3),
dims=['y', 'z'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['z', 'y'])
assert numpy_array.shape == (3, 2)
assert np.all(np.transpose(numpy_array, (1, 0)) == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_1d():
array = DataArray(
np.random.randn(2),
dims=['y'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['y'])
assert numpy_array.shape == (2,)
assert np.all(numpy_array == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_creates_new_dim():
array = DataArray(
np.random.randn(2),
dims=['x'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['x', 'y'])
assert numpy_array.shape == (2, 1)
assert np.all(numpy_array[:, 0] == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_retrieves_explicit_dimensions():
array = DataArray(
np.random.randn(2, 3),
dims=['alpha', 'zeta'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['zeta', 'alpha'])
assert numpy_array.shape == (3, 2)
assert np.all(np.transpose(numpy_array, (1, 0)) == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_invalid_dimension_collected_by_asterisk():
array = DataArray(
np.random.randn(2),
dims=['sheep'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['*'])
assert numpy_array.shape == (2,)
assert np.all(numpy_array == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_asterisk_creates_new_dim():
array = DataArray(
np.random.randn(2),
dims=['x'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['x', '*'])
assert numpy_array.shape == (2, 1)
assert np.all(numpy_array[:, 0] == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_asterisk_creates_new_dim_reversed():
array = DataArray(
np.random.randn(2),
dims=['x'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['*', 'x'])
assert numpy_array.shape == (1, 2)
assert np.all(numpy_array[0, :] == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_get_numpy_array_asterisk_flattens():
array = DataArray(
np.random.randn(2, 3),
dims=['y', 'z'],
attrs={'units': ''},
)
numpy_array = get_numpy_array(array, ['*'])
assert numpy_array.shape == (6,)
assert np.all(numpy_array.reshape((2, 3)) == array.values)
assert np.byte_bounds(numpy_array) == np.byte_bounds(array.values)
assert numpy_array.base is array.values
def test_restore_dimensions_starz_to_zyx_doesnt_copy():
array = DataArray(
np.random.randn(2, 3, 4),
dims=['z', 'y', 'x'],
attrs={'units': ''}
)
numpy_array = get_numpy_array(array, ['*', 'z'])
restored_array = restore_dimensions(
numpy_array, from_dims=['*', 'z'], result_like=array)
assert np.byte_bounds(restored_array.values) == np.byte_bounds(
array.values)
assert restored_array.values.base is array.values
def test_restore_dimensions_3d_reverse():
array = DataArray(
np.random.randn(2, 3, 4),
dims=['z', 'y', 'x'],
attrs={'units': ''}
)
numpy_array = get_numpy_array(array, ['x', 'y', 'z'])
restored_array = restore_dimensions(
numpy_array, from_dims=['x', 'y', 'z'], result_like=array)
assert np.all(restored_array.values == array.values)
assert len(restored_array.attrs) == 0
assert np.byte_bounds(restored_array.values) == np.byte_bounds(
array.values)
assert restored_array.values.base is array.values
def test_restore_dimensions_1d_flatten():
array = DataArray(
np.random.randn(2),
dims=['z'],
attrs={'units': ''}
)
numpy_array = get_numpy_array(array, ['*'])
restored_array = restore_dimensions(
numpy_array, from_dims=['*'], result_like=array)
assert np.all(restored_array.values == array.values)
assert len(restored_array.attrs) == 0
assert np.byte_bounds(restored_array.values) == np.byte_bounds(
array.values)
assert restored_array.values.base is array.values
def test_restore_dimensions_2d_flatten():
array = DataArray(
np.random.randn(2, 3),
dims=['z', 'y'],
attrs={'units': ''}
)
numpy_array = get_numpy_array(array, ['*'])
restored_array = restore_dimensions(
numpy_array, from_dims=['*'], result_like=array)
assert np.all(restored_array.values == array.values)
assert len(restored_array.attrs) == 0
assert np.byte_bounds(restored_array.values) == np.byte_bounds(
array.values)
assert restored_array.values.base is array.values
def test_restore_dimensions_removes_dummy_axes():
array = DataArray(
np.random.randn(2),
dims=['z'],
attrs={'units': ''}
)
numpy_array = get_numpy_array(array, ['x', 'y', 'z'])
restored_array = restore_dimensions(
numpy_array, from_dims=['x', 'y', 'z'], result_like=array)
assert np.all(restored_array.values == array.values)
assert len(restored_array.attrs) == 0
assert np.byte_bounds(restored_array.values) == np.byte_bounds(
array.values)
assert restored_array.values.base is array.values
def test_returns_simple_value(self):
input_state = {
'air_temperature': DataArray(
np.zeros([2, 2, 4]),
dims=['x', 'y', 'z'],
attrs={'units': 'degK'},
)
}
input_properties = {
'air_temperature': {
'dims': ['x', 'y', 'z'],
'units': 'degK',
}
}
raw_arrays = get_numpy_arrays_with_properties(input_state, input_properties)
raw_arrays = {key + '_tendency': value for key, value in raw_arrays.items()}
output_properties = {
'air_temperature_tendency': {
'dims_like': 'air_temperature',
'units': 'degK/s',
}
}
return_value = restore_data_arrays_with_properties(
raw_arrays, output_properties, input_state, input_properties
)
assert isinstance(return_value, dict)
assert len(return_value.keys()) == 1
assert isinstance(return_value['air_temperature_tendency'], DataArray)
assert return_value['air_temperature_tendency'].attrs['units'] is 'degK/s'
assert np.byte_bounds(
return_value['air_temperature_tendency'].values) == np.byte_bounds(
input_state['air_temperature'].values)
assert (return_value['air_temperature_tendency'].values.base is
input_state['air_temperature'].values)
assert return_value['air_temperature_tendency'].shape == (2, 2, 4)
def test_assumes_dims_like_own_name(self):
input_state = {
'air_temperature': DataArray(
np.zeros([2, 2, 4]),
dims=['x', 'y', 'z'],
attrs={'units': 'degK'},
)
}
input_properties = {
'air_temperature': {
'dims': ['x', 'y', 'z'],
'units': 'degK',
}
}
raw_arrays = get_numpy_arrays_with_properties(input_state, input_properties)
output_properties = {
'air_temperature': {
'units': 'degK/s',
}
}
return_value = restore_data_arrays_with_properties(
raw_arrays, output_properties, input_state, input_properties
)
assert isinstance(return_value, dict)
assert len(return_value.keys()) == 1
assert isinstance(return_value['air_temperature'], DataArray)
assert return_value['air_temperature'].attrs['units'] is 'degK/s'
assert np.byte_bounds(
return_value['air_temperature'].values) == np.byte_bounds(
input_state['air_temperature'].values)
assert (return_value['air_temperature'].values.base is
input_state['air_temperature'].values)
assert return_value['air_temperature'].shape == (2, 2, 4)
def test_restores_collected_horizontal_dimensions(self):
random = np.random.RandomState(0)
T_array = random.randn(3, 2, 4)
input_state = {
'air_temperature': DataArray(
T_array,
dims=['x', 'y', 'z'],
attrs={'units': 'degK'},
)
}
input_properties = {
'air_temperature': {
'dims': ['z', '*'],
'units': 'degK',
}
}
raw_arrays = get_numpy_arrays_with_properties(input_state, input_properties)
raw_arrays = {key + '_tendency': value for key, value in raw_arrays.items()}
output_properties = {
'air_temperature_tendency': {
'dims_like': 'air_temperature',
'units': 'degK/s',
}
}
return_value = restore_data_arrays_with_properties(
raw_arrays, output_properties, input_state, input_properties
)
assert isinstance(return_value, dict)
assert len(return_value.keys()) == 1
assert isinstance(return_value['air_temperature_tendency'], DataArray)
assert return_value['air_temperature_tendency'].attrs['units'] is 'degK/s'
assert np.byte_bounds(
return_value['air_temperature_tendency'].values) == np.byte_bounds(
input_state['air_temperature'].values)
assert (return_value['air_temperature_tendency'].values.base is
input_state['air_temperature'].values)
assert return_value['air_temperature_tendency'].shape == (3, 2, 4)
assert np.all(return_value['air_temperature_tendency'] == T_array)
assert return_value['air_temperature_tendency'].dims == input_state['air_temperature'].dims
def test_restores_aliased_name(self):
input_state = {
'air_temperature': DataArray(
np.zeros([2, 2, 4]),
dims=['x', 'y', 'z'],
attrs={'units': 'degK'},
)
}
input_properties = {
'air_temperature': {
'dims': ['x', 'y', 'z'],
'units': 'degK',
}
}
raw_arrays = {
'p': np.zeros([2, 2, 4])
}
output_properties = {
'air_pressure': {
'dims_like': 'air_temperature',
'units': 'm',
'alias': 'p',
},
}
data_arrays = restore_data_arrays_with_properties(
raw_arrays, output_properties, input_state, input_properties
)
assert len(data_arrays.keys()) == 1
assert 'air_pressure' in data_arrays.keys()
assert np.all(data_arrays['air_pressure'].values == raw_arrays['p'])
assert np.byte_bounds(data_arrays['air_pressure'].values) == np.byte_bounds(raw_arrays['p'])
def test_restores_when_name_has_alias(self):
input_state = {
'air_temperature': DataArray(
np.zeros([2, 2, 4]),
dims=['x', 'y', 'z'],
attrs={'units': 'degK'},
)
}
input_properties = {
'air_temperature': {
'dims': ['x', 'y', 'z'],
'units': 'degK',
}
}
raw_arrays = {
'air_pressure': np.zeros([2, 2, 4])
}
output_properties = {
'air_pressure': {
'dims_like': 'air_temperature',
'units': 'm',
'alias': 'p',
},
}
data_arrays = restore_data_arrays_with_properties(
raw_arrays, output_properties, input_state, input_properties
)
assert len(data_arrays.keys()) == 1
assert 'air_pressure' in data_arrays.keys()
assert np.all(data_arrays['air_pressure'].values == raw_arrays['air_pressure'])
assert np.byte_bounds(
data_arrays['air_pressure'].values) == np.byte_bounds(
raw_arrays['air_pressure'])
def test_match_dims_like_partly_hardcoded_dimensions_matching_lengths():
input_state = {
'air_temperature': DataArray(
np.zeros([2, 3, 4]),
dims=['lat', 'lon', 'mid_levels'],
attrs={'units': 'degK'},
),
'air_pressure': DataArray(
np.zeros([2, 3, 4]),
dims=['lat', 'lon', 'interface_levels'],
attrs={'units': 'Pa'},
),
}
input_properties = {
'air_temperature': {
'dims': ['*', 'mid_levels'],
'units': 'degK',
'match_dims_like': 'air_pressure',
},
'air_pressure': {
'dims': ['*', 'interface_levels'],
'units': 'Pa',
},
}
raw_arrays = get_numpy_arrays_with_properties(input_state, input_properties)
assert np.byte_bounds(input_state['air_temperature'].values) == np.byte_bounds(raw_arrays['air_temperature'])
assert np.byte_bounds(input_state['air_pressure'].values) == np.byte_bounds(raw_arrays['air_pressure'])
def test_match_dims_like_x_y_z_matching_lengths(self):
set_direction_names(x=['lat'], y=['lon'], z=['mid_levels', 'interface_levels'])
input_state = {
'air_temperature': DataArray(
np.zeros([2, 3, 4]),
dims=['lat', 'lon', 'mid_levels'],
attrs={'units': 'degK'},
),
'air_pressure': DataArray(
np.zeros([2, 3, 4]),
dims=['lat', 'lon', 'mid_levels'],
attrs={'units': 'Pa'},
),
}
input_properties = {
'air_temperature': {
'dims': ['x', 'y', 'z'],
'units': 'degK',
'match_dims_like': 'air_pressure',
},
'air_pressure': {
'dims': ['x', 'y', 'z'],
'units': 'Pa',
},
}
raw_arrays = get_numpy_arrays_with_properties(
input_state, input_properties)
assert np.byte_bounds(raw_arrays['air_temperature']) == np.byte_bounds(input_state['air_temperature'].values)
assert np.byte_bounds(raw_arrays['air_pressure']) == np.byte_bounds(input_state['air_pressure'].values)
def test_match_dims_like_star_z_matching_lengths(self):
set_direction_names(x=['lat'], y=['lon'], z=['mid_levels', 'interface_levels'])
input_state = {
'air_temperature': DataArray(
np.zeros([2, 3, 4]),
dims=['lat', 'lon', 'interface_levels'],
attrs={'units': 'degK'},
),
'air_pressure': DataArray(
np.zeros([2, 3, 4]),
dims=['lat', 'lon', 'interface_levels'],
attrs={'units': 'Pa'},
),
}
input_properties = {
'air_temperature': {
'dims': ['*', 'z'],
'units': 'degK',
'match_dims_like': 'air_pressure',
},
'air_pressure': {
'dims': ['*', 'z'],
'units': 'Pa',
},
}
raw_arrays = get_numpy_arrays_with_properties(
input_state, input_properties)
assert np.byte_bounds(raw_arrays['air_temperature']) == np.byte_bounds(input_state['air_temperature'].values)
assert np.byte_bounds(raw_arrays['air_pressure']) == np.byte_bounds(input_state['air_pressure'].values)
def _reduce_memmap_backed(a, m):
"""Pickling reduction for memmap backed arrays.
a is expected to be an instance of np.ndarray (or np.memmap)
m is expected to be an instance of np.memmap on the top of the ``base``
attribute ancestry of a. ``m.base`` should be the real python mmap object.
"""
# offset that comes from the striding differences between a and m
a_start, a_end = np.byte_bounds(a)
m_start = np.byte_bounds(m)[0]
offset = a_start - m_start
# offset from the backing memmap
offset += m.offset
if m.flags['F_CONTIGUOUS']:
order = 'F'
else:
# The backing memmap buffer is necessarily contiguous hence C if not
# Fortran
order = 'C'
if a.flags['F_CONTIGUOUS'] or a.flags['C_CONTIGUOUS']:
# If the array is a contiguous view, no need to pass the strides
strides = None
total_buffer_len = None
else:
# Compute the total number of items to map from which the strided
# view will be extracted.
strides = a.strides
total_buffer_len = (a_end - a_start) // a.itemsize
return (_strided_from_memmap,
(m.filename, a.dtype, m.mode, offset, order, a.shape, strides,
total_buffer_len))
def _reduce_memmap_backed(a, m):
"""Pickling reduction for memmap backed arrays
a is expected to be an instance of np.ndarray (or np.memmap)
m is expected to be an instance of np.memmap on the top of the ``base``
attribute ancestry of a. ``m.base`` should be the real python mmap object.
"""
# offset that comes from the striding differences between a and m
a_start, a_end = np.byte_bounds(a)
m_start = np.byte_bounds(m)[0]
offset = a_start - m_start
# offset from the backing memmap
offset += m.offset
if m.flags['F_CONTIGUOUS']:
order = 'F'
else:
# The backing memmap buffer is necessarily contiguous hence C if not
# Fortran
order = 'C'
if a.flags['F_CONTIGUOUS'] or a.flags['C_CONTIGUOUS']:
# If the array is a contiguous view, no need to pass the strides
strides = None
total_buffer_len = None
else:
# Compute the total number of items to map from which the strided
# view will be extracted.
strides = a.strides
total_buffer_len = (a_end - a_start) // a.itemsize
return (_strided_from_memmap,
(m.filename, a.dtype, m.mode, offset, order, a.shape, strides,
total_buffer_len))
