# Path: thrifty/soa_estimator.py
# def _clip_offset(offset, max_=0.6):
# def calculate_window(block_len, history_len, template_len):
#     def __init__(self, template, thresh_coeffs, block_len, history_len):
#     def soa_estimate(self, fft):
#     def __call__(self, fft):
#     def despread(self, fft):
#     def get_peak(self, corr):
#     def estimate_noise(self, peak_mag, fft):
#     def calculate_threshold(self, corr, noise_rms):
# def calculate_threshold(corr, noise_rms, thresh_coeffs):
# def get_peak(corr, window):
# def parabolic_interpolation(corr_mag, peak_idx):
# def gaussian_interpolation(corr_mag, peak_idx):
# class SoaEstimator(object):
# 
# Path: thrifty/signal_utils.py
# class Signal(np.ndarray):
#     """Representation of signal with caching and on-demand calculations."""
#     # Based on https://docs.scipy.org/doc/numpy/user/basics.subclassing.html
# 
#     def __new__(cls, input_array, fft=None, ifft=None):
#         obj = np.asarray(input_array).view(cls)
# 
#         if fft is not None:
#             obj._fft = fft
#             fft._ifft = obj
# 
#         if ifft is not None:
#             obj._ifft = ifft
#             ifft._fft = obj
# 
#         return obj
# 
#     def __array_finalize__(self, obj):
#         if obj is None:
#             return
#         self._reset()
# 
#     def __array_wrap__(self, obj, context=None):
#         # http://stackoverflow.com/questions/19223926/
# 
#         if obj.shape == ():
#             return obj[()]  # if ufunc output is scalar, return it
#         else:
#             ret = np.ndarray.__array_wrap__(self, obj, context)
#             if isinstance(ret, Signal):
#                 ret._reset()
#             return ret
# 
#     def _reset(self):
#         self._fft = None
#         self._ifft = None
#         self._rms = None
#         self._mag = None
#         self._power = None
#         self._conj = None
# 
#     @property
#     def fft(self):
#         """FFT of signal."""
#         if self._fft is None:
#             self._fft = Signal(compute_fft(self), ifft=self)
#         return self._fft
# 
#     @property
#     def ifft(self):
#         """IFFT of signal."""
#         if self._ifft is None:
#             self._ifft = Signal(compute_ifft(self), fft=self)
#         return self._ifft
# 
#     @property
#     def mag(self):
#         """Calculate the magnitude of the samples."""
#         if self._mag is None:
#             if self._power is not None:
#                 self._mag = np.sqrt(self._power)
#             else:
#                 self._mag = np.abs(self)  # TODO: Convert to Signal?
#         return self._mag
# 
#     @property
#     def power(self):
#         """Calculate the power (magnitude^2) of the samples."""
#         if self._power is None:
#             # Alternative: self._power = (samples * np.conj(samples)).real
#             # or: self._power = np.abs(self)**2
#             self._power = self.mag**2
#         return self._power
# 
#     @property
#     def rms(self):
#         """Calculate signal's root mean square value."""
#         if self._rms is None:
#             if self._fft is not None and self._fft._rms is not None:
#                 self._rms = self._fft._rms / np.sqrt(len(self))
#             elif self._ifft is not None and self._ifft._rms is not None:
#                 self._rms = self._ifft._rms * np.sqrt(len(self))
#             else:
#                 self._rms = np.sqrt(np.mean(self.power))
#         return self._rms
# 
#     @property
#     def conj(self):
#         """Calculate and cache the complex conjugate."""
#         if self._conj is None:
#             self._conj = np.ndarray.conj(self)
#             self._conj._conj = self
#         return self._conj


import pytest
import numpy as np
import scipy.signal
from thrifty import soa_estimator
from thrifty.signal_utils import Signal
        assert corr_abs[peak_idx] >= PEAK_MAG - 0.1
        np.testing.assert_array_less(non_peak, MAX_SIDEBAND + 0.1)
    else:
        np.testing.assert_array_less(corr_abs, MAX_SIDEBAND + 0.1)


@pytest.mark.parametrize("pos", DESPREADER_TESTDATA)
def test_despreader_cross_validate(pos):
    """Cross-validate despreader against scipy's correlate function."""
    estimator = soa_estimator.SoaEstimator(TEMPLATE, None,
                                           BLOCK_LEN, len(TEMPLATE))
    block = gen_block(pos)
    fft = np.fft.fft(block)
    corr = estimator.despread(fft)
    corr2 = scipy.signal.correlate(block, TEMPLATE, mode='valid')
    np.testing.assert_allclose(corr, corr2, atol=1e-12, rtol=1e-12)


WINDOW_TESTDATA = [
    # (block_len, history_len, template_len), (expected_start, expected_stop)
    ((64, 31, 32), (0, 33)),
    ((64, 32, 32), (0, 32)),
    ((64, 33, 32), (1, 32)),
    ((64, 63, 32), (16, 17)),
]


@pytest.mark.parametrize("params,expected", WINDOW_TESTDATA)
def test_calculate_window(params, expected):
    """Test calculate_window with different parameters."""

# Path: thrifty/soa_estimator.py
# def _clip_offset(offset, max_=0.6):
# def calculate_window(block_len, history_len, template_len):
#     def __init__(self, template, thresh_coeffs, block_len, history_len):
#     def soa_estimate(self, fft):
#     def __call__(self, fft):
#     def despread(self, fft):
#     def get_peak(self, corr):
#     def estimate_noise(self, peak_mag, fft):
#     def calculate_threshold(self, corr, noise_rms):
# def calculate_threshold(corr, noise_rms, thresh_coeffs):
# def get_peak(corr, window):
# def parabolic_interpolation(corr_mag, peak_idx):
# def gaussian_interpolation(corr_mag, peak_idx):
# class SoaEstimator(object):
# 
# Path: thrifty/signal_utils.py
# class Signal(np.ndarray):
#     """Representation of signal with caching and on-demand calculations."""
#     # Based on https://docs.scipy.org/doc/numpy/user/basics.subclassing.html
# 
#     def __new__(cls, input_array, fft=None, ifft=None):
#         obj = np.asarray(input_array).view(cls)
# 
#         if fft is not None:
#             obj._fft = fft
#             fft._ifft = obj
# 
#         if ifft is not None:
#             obj._ifft = ifft
#             ifft._fft = obj
# 
#         return obj
# 
#     def __array_finalize__(self, obj):
#         if obj is None:
#             return
#         self._reset()
# 
#     def __array_wrap__(self, obj, context=None):
#         # http://stackoverflow.com/questions/19223926/
# 
#         if obj.shape == ():
#             return obj[()]  # if ufunc output is scalar, return it
#         else:
#             ret = np.ndarray.__array_wrap__(self, obj, context)
#             if isinstance(ret, Signal):
#                 ret._reset()
#             return ret
# 
#     def _reset(self):
#         self._fft = None
#         self._ifft = None
#         self._rms = None
#         self._mag = None
#         self._power = None
#         self._conj = None
# 
#     @property
#     def fft(self):
#         """FFT of signal."""
#         if self._fft is None:
#             self._fft = Signal(compute_fft(self), ifft=self)
#         return self._fft
# 
#     @property
#     def ifft(self):
#         """IFFT of signal."""
#         if self._ifft is None:
#             self._ifft = Signal(compute_ifft(self), fft=self)
#         return self._ifft
# 
#     @property
#     def mag(self):
#         """Calculate the magnitude of the samples."""
#         if self._mag is None:
#             if self._power is not None:
#                 self._mag = np.sqrt(self._power)
#             else:
#                 self._mag = np.abs(self)  # TODO: Convert to Signal?
#         return self._mag
# 
#     @property
#     def power(self):
#         """Calculate the power (magnitude^2) of the samples."""
#         if self._power is None:
#             # Alternative: self._power = (samples * np.conj(samples)).real
#             # or: self._power = np.abs(self)**2
#             self._power = self.mag**2
#         return self._power
# 
#     @property
#     def rms(self):
#         """Calculate signal's root mean square value."""
#         if self._rms is None:
#             if self._fft is not None and self._fft._rms is not None:
#                 self._rms = self._fft._rms / np.sqrt(len(self))
#             elif self._ifft is not None and self._ifft._rms is not None:
#                 self._rms = self._ifft._rms * np.sqrt(len(self))
#             else:
#                 self._rms = np.sqrt(np.mean(self.power))
#         return self._rms
# 
#     @property
#     def conj(self):
#         """Calculate and cache the complex conjugate."""
#         if self._conj is None:
#             self._conj = np.ndarray.conj(self)
#             self._conj._conj = self
#         return self._conj


# Path: tests/test_soa_estimator.py
import pytest
import numpy as np
import scipy.signal
from thrifty import soa_estimator
from thrifty.signal_utils import Signal

        assert corr_abs[peak_idx] >= PEAK_MAG - 0.1
        np.testing.assert_array_less(non_peak, MAX_SIDEBAND + 0.1)
    else:
        np.testing.assert_array_less(corr_abs, MAX_SIDEBAND + 0.1)


@pytest.mark.parametrize("pos", DESPREADER_TESTDATA)
def test_despreader_cross_validate(pos):
    """Cross-validate despreader against scipy's correlate function."""
    estimator = soa_estimator.SoaEstimator(TEMPLATE, None,
                                           BLOCK_LEN, len(TEMPLATE))
    block = gen_block(pos)
    fft = np.fft.fft(block)
    corr = estimator.despread(fft)
    corr2 = scipy.signal.correlate(block, TEMPLATE, mode='valid')
    np.testing.assert_allclose(corr, corr2, atol=1e-12, rtol=1e-12)


WINDOW_TESTDATA = [
    # (block_len, history_len, template_len), (expected_start, expected_stop)
    ((64, 31, 32), (0, 33)),
    ((64, 32, 32), (0, 32)),
    ((64, 33, 32), (1, 32)),
    ((64, 63, 32), (16, 17)),
]


@pytest.mark.parametrize("params,expected", WINDOW_TESTDATA)
def test_calculate_window(params, expected):
    """Test calculate_window with different parameters."""

    got = soa_estimator.calculate_window(*params)