Skip to content

Commit

Permalink
psd: backwards compatibility
Browse files Browse the repository at this point in the history
This is a temporary option which returns a "simple" spectrum without the unblocked source data

It will not be possible to perform `identify_peaks` on a spectrum generated in this way (nor should it be).
  • Loading branch information
JoepVanlier committed Sep 27, 2024
1 parent 45992c9 commit be55f2a
Show file tree
Hide file tree
Showing 2 changed files with 58 additions and 21 deletions.
29 changes: 28 additions & 1 deletion lumicks/pylake/force_calibration/power_spectrum.py
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,34 @@ def _variance(self):
return self._apply_transforms(self._raw_variance)

def downsampled_by(self, factor, reduce=np.mean) -> "PowerSpectrum":
"""Returns a spectrum downsampled by a given factor."""
"""Returns a spectrum downsampled by a given factor.
Parameters
----------
factor : int
Factor to down-sample the spectrum by.
reduce : callable
(Deprecated) Function to use for down-sampling the data. Only `np.mean` will be
supported going forward.
"""
if reduce != np.mean:
warnings.warn(
DeprecationWarning(
"Providing other reduction functions than `np.mean` is deprecated and will be "
"removed in a future version of Pylake"
)
)
return PowerSpectrum(
downsample(self.frequency, factor, reduce),
downsample(self.power, factor, reduce),
self.sample_rate,
self.total_duration,
self.unit,
num_points_per_block=self.num_points_per_block * factor,
total_samples_used=self.total_sampled_used,
variance=None,
)

ba = copy(self)
ba._downsampling_factor = ba._downsampling_factor * factor

Expand Down
50 changes: 30 additions & 20 deletions lumicks/pylake/force_calibration/tests/test_power_spectrum.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,9 @@ def test_power_spectrum_attrs(frequency, num_data, sample_rate):
np.testing.assert_allclose(power_spectrum.frequency_bin_width, sample_rate / num_data)


@pytest.mark.filterwarnings(
"ignore:Providing other reduction functions than `np.mean` is deprecated*"
)
@pytest.mark.parametrize(
"frequency, num_data, sample_rate, num_blocks, f_blocked, p_blocked",
[
Expand All @@ -100,28 +103,35 @@ def test_power_spectrum_blocking(
frequency, num_data, sample_rate, num_blocks, f_blocked, p_blocked
):
"""Functional test whether the results of blocking the power spectrum are correct"""
data = np.sin(2.0 * np.pi * frequency / sample_rate * np.arange(num_data)) / np.sqrt(2)
power_spectrum = PowerSpectrum.from_data(data, sample_rate)

downsampling_factor = len(power_spectrum.frequency) // num_blocks
blocked = power_spectrum.downsampled_by(downsampling_factor)
np.testing.assert_allclose(blocked.frequency, f_blocked)
np.testing.assert_allclose(blocked.power, p_blocked, atol=1e-16)
np.testing.assert_allclose(blocked.num_samples(), len(f_blocked))
np.testing.assert_allclose(len(power_spectrum.power), num_data // 2 + 1)
np.testing.assert_allclose(
blocked.num_points_per_block, np.floor(len(power_spectrum.power) / num_blocks)
)
np.testing.assert_allclose(
blocked.frequency_bin_width, blocked.frequency[1] - blocked.frequency[0]
)
def custom_reduce(x, *args, **kwargs):
return np.mean(x, *args, **kwargs)

for reduce in (np.mean, custom_reduce):
data = np.sin(2.0 * np.pi * frequency / sample_rate * np.arange(num_data)) / np.sqrt(2)
power_spectrum = PowerSpectrum.from_data(data, sample_rate)

downsampling_factor = len(power_spectrum.frequency) // num_blocks
blocked = power_spectrum.downsampled_by(downsampling_factor, reduce=reduce)
np.testing.assert_allclose(blocked.frequency, f_blocked)
np.testing.assert_allclose(blocked.power, p_blocked, atol=1e-16)
np.testing.assert_allclose(blocked.num_samples(), len(f_blocked))
np.testing.assert_allclose(len(power_spectrum.power), num_data // 2 + 1)
np.testing.assert_allclose(
blocked.num_points_per_block, np.floor(len(power_spectrum.power) / num_blocks)
)
np.testing.assert_allclose(
blocked.frequency_bin_width, blocked.frequency[1] - blocked.frequency[0]
)

# Downsample again and make sure the num_points_per_block is correct
dual_blocked = blocked.downsampled_by(2)
np.testing.assert_allclose(dual_blocked.num_points_per_block, blocked.num_points_per_block * 2)
np.testing.assert_allclose(
dual_blocked.frequency_bin_width, dual_blocked.frequency[1] - dual_blocked.frequency[0]
)
# Downsample again and make sure the num_points_per_block is correct
dual_blocked = blocked.downsampled_by(2, reduce=reduce)
np.testing.assert_allclose(
dual_blocked.num_points_per_block, blocked.num_points_per_block * 2
)
np.testing.assert_allclose(
dual_blocked.frequency_bin_width, dual_blocked.frequency[1] - dual_blocked.frequency[0]
)


@pytest.mark.parametrize(
Expand Down

0 comments on commit be55f2a

Please sign in to comment.