DSP

Digital Signal Processing module for Bob.

DSP

Main module for the DSP algorithm.

Warning: the DSP _dsp_bob_shared_clock_shared_lo, _dsp_bob_shared_clock_unshared_lo and _dsp_bob_unshared_clock_shared_lo are adapated versions of old DSP and might not work. There are untested.

class qosst_bob.dsp.dsp.DSPDebug(begin_zadoff_chu: int = 0, end_zadoff_chu: int = 0, begin_data: int = 0, end_data: int = 0, tones: ~typing.List[~numpy.ndarray] = <factory>, uncorrected_data: ~typing.List[~numpy.ndarray] = <factory>, real_pilot_frequencies: ~typing.List[float] = <factory>, beat_frequency: float = 0, delta_frequency_pilots: float = 0, equi_adc_rate: float = 0)

Dataclass for debug information for the DSP.

begin_zadoff_chu: int = 0: Beginning of the Zadoff-Chu sequence.

end_zadoff_chu: int = 0: End of the Zadoff-Chu sequence.

begin_data: int = 0: Beginning of useful data.

end_data: int = 0: End of useful data.

tones: List[ndarray]: List of arrays containing filtered tone used for phase recovery.

uncorrected_data: List[ndarray]: List of arrays containing data before phase correction.

real_pilot_frequencies: List[float]: List of recovered pilot frequencies.

delta_frequency_pilots: float = 0: Difference of frequency between the two tones.

equi_adc_rate: float = 0: Equivalent ADC rate in case the clock is not shared.

class qosst_bob.dsp.dsp.SpecialDSPParams(symbol_rate: float, adc_rate: float, roll_off: float, frequency_shift: float, schema: DetectionSchema)

Dataclass for the parameters to give to the special DSP function for the elec and shot noise.

symbol_rate: float: Symbol rate in Symbols/s,.

adc_rate: float: Symbol rate in Samples/s, recovered in case clock is not shared.

roll_off: float: Roll off of the RRC filter.

frequency_shift: float: Frequency shift of the data, recovered in case clock is not shared and/or LLO setup.

schema: DetectionSchema: Detection schema to know how to interpret the data.

qosst_bob.dsp.dsp.dsp_bob(data: ndarray, config: Configuration) → Tuple[List[ndarray] | None, SpecialDSPParams | None, DSPDebug | None]

DSP function for Bob, given the data and the configuration.

Parameters:

data (np.ndarray) – the data on which to apply the DSP.
config (Configuration) – the configuration object.

Returns:

array of symbols, SpecialDSPParams containing data to apply the exact same DSP to other data and DSPDebug containing debug information,.

Return type:

Tuple[Optional[List[np.ndarray]], Optional[SpecialDSPParams], Optional[DSPDebug]]

qosst_bob.dsp.dsp.dsp_bob_params(data: ~numpy.ndarray, symbol_rate: float, dac_rate: float, adc_rate: float, num_symbols: int, roll_off: float, frequency_shift: float, num_pilots: int, pilots_frequencies: ~numpy.ndarray, zc_length: int, zc_root: int, zc_rate: float, shared_clock: bool = False, shared_lo: bool = False, process_subframes: bool = False, subframe_length: int = 0, fir_size: int = 500, tone_filtering_cutoff: float = 10000000.0, abort_clock_recovery: float = 0, excl: ~typing.List[~typing.Tuple[float, float]] | None = None, pilot_phase_filtering_size: int = 0, num_samples_fbeat_estimation: int = 100000, schema: ~qosst_core.schema.detection.DetectionSchema = <qosst_core.schema.detection.DetectionSchema object>, debug: bool = False) → Tuple[List[ndarray] | None, SpecialDSPParams | None, DSPDebug | None]

Apply the DSP to the data given the DSP parameters.

Parameters:

data (np.ndarray) – data on which to apply the DSP.
symbol_rate (float) – symbol rate in Symbols per second.
dac_rate (float) – DAC rate in Hz.
adc_rate (float) – ADC rate in Hz.
num_symbols (int) – number of sent symbols.
roll_off (float) – roll off value for the RRC filter
frequency_shift (float) – frequency shift of the quantum data in Hz.
num_pilots (int) – number of pilots.
pilots_frequencies (np.ndarray) – list of pilot frequencies, in Hz.
zc_length (int) – length of the Zadoff-Chu sequence.
zc_root (int) – root of the Zadoff-Chu sequence.
zc_rate (float) – rate of the Zadoff-Chu sequence.
shared_clock (bool, optional) – if the clock is shared between Alice and Bob. Defaults to False.
shared_lo (bool, optional) – if the local oscillator is shared between Alice and Bob. Defaults to False.
process_subframes (bool, optional) – if the data should be processed at subframes. Defaults to False.
subframe_length (int, optional) – if the previous parameter is True, the length, in samples, of the subframe. Defaults to 0.
fir_size (int, optional) – FIR size. Defaults to 500.
tone_filtering_cutoff (float, optional) – cutoff for the FIR filter for the pilot filtering, in Hz.
abort_clock_recovery (float, optional) – Maximal mismatch allowed by the clock recovery algorithm before aborting. If 0, the algorithm never aborts. Defaults to 0.
excl (Optional[List[Tuple[float, float]]], optional) – exclusion zones for the research of pilots (i.e. frequencies where we are sure the pilots are not), given as a list of tuples of float, each elements defining excluded segment (start frequency, stop frequency).
pilot_phase_filtering_size (int, optional) – Size of the uniform1d filter to apply to the phase of the recovered pilots for correction. Defaults to 0.
num_samples_fbeat_estimation (int, optional) – number of samples to estimate the beat frequency between the two lasers. Defaults to 100000.
schema (DetectionSchema, optional) – detection schema to use for the DSP. Defaults to qosst_core.schema.emission.SINGLE_POLARISATION_RF_HETERODYNE.
debug (bool, optional) – wether to return a debug dict. Defaults to False.

Returns:

array of symbols, SpecialDSPParams containing data to apply the exact same DPS to other data and DSPDebug containing debug information,.

Return type:

Tuple[Optional[np.ndarray], Optional[SpecialDSPParams], Optional[DSPDebug]]

qosst_bob.dsp.dsp._dsp_bob_shared_clock_shared_lo(data: ~numpy.ndarray, symbol_rate: float, dac_rate: float, adc_rate: float, num_symbols: int, roll_off: float, frequency_shift: float, num_pilots: int, pilots_frequencies: ~numpy.ndarray, zc_length: int, zc_root: int, zc_rate: float, process_subframes: bool = False, subframe_length: int = 0, fir_size: int = 500, tone_filtering_cutoff: float = 10000000.0, pilot_phase_filtering_size: int = 0, schema: ~qosst_core.schema.detection.DetectionSchema = <qosst_core.schema.detection.DetectionSchema object>, debug: bool = False) → Tuple[List[ndarray] | None, SpecialDSPParams | None, DSPDebug | None]

DSP in the case of a shared clock and a shared local oscillator.

This simplifies a lot the DSP, since there is no clock difference or beat frequency.

The procedure is the following:

Recovery of the Zadoff-Chu sequence
Recovery of the pilot (per subframe)
Unshift signal (per subframe)
Apply match filter (per subframe)
Downsample (per subframe)
Correct relative phase noise (per subframe)

The output has still a global phase noise.

Parameters:

data (np.ndarray) – data received by Bob.
symbol_rate (float) – symbol rate for the quantum data, in Symbols per second.
dac_rate (float) – DAC rate, in Hz.
adc_rate (float) – ADC rate in Hz.
num_symbols (int) – number of symbols.
roll_off (float) – roll-off for the RRC filter.
frequency_shift (float) – frequnecy shift in Hz for the quantum data.
num_pilots (int) – number of pilots.
pilots_frequencies (np.ndarray) – list of frequencies of the pilots.
zc_length (int) – length of the Zadoff-Chu sequence.
zc_root (int) – root of the Zadoff-Chu sequence.
zc_rate (float) – rate of the Zadoff-Chu sequence.
process_subframes (bool, optional) – if True, data is processed as subframes. Defaults to False.
subframe_length (int, optional) – number of symbols to recover in each subframe. Defaults to 0.
fir_size (int, optional) – size for the FIR filters. Defaults to 500.
tone_filtering_cutoff (float, optional) – cutoff, in Hz, for the filter of the tone. Defaults to 10e6.
pilot_phase_filtering_size (int, optional) – size of the uniform1d filter to filter the phase correction. Defaults to 0.
schema (DetectionSchema, optional) – detection schema to use for the DSP. Defaults to qosst_core.schema.emission.SINGLE_POLARISATION_RF_HETERODYNE.
debug (bool, optional) – if True, a debug dict is returned. Defaults to False.

Returns:

list of np.ndarray, each one corresponding to the recovered symbols for a subframe, SpecialDSPParams object to give to the special dsp, and DSPDebug object if debug was true.

Return type:

Tuple[List[np.ndarray], SpecialDSPParams, Optional[DSPDebug]]

qosst_bob.dsp.dsp._dsp_bob_shared_clock_unshared_lo(data: ~numpy.ndarray, symbol_rate: float, dac_rate: float, adc_rate: float, num_symbols: int, roll_off: float, frequency_shift: float, num_pilots: int, pilots_frequencies: ~numpy.ndarray, zc_length: int, zc_root: int, zc_rate: float, process_subframes: bool = False, subframe_length: int = 0, fir_size: int = 500, tone_filtering_cutoff: float = 10000000.0, excl: ~typing.List[~typing.Tuple[float, float]] | None = None, pilot_phase_filtering_size: int = 0, schema: ~qosst_core.schema.detection.DetectionSchema = <qosst_core.schema.detection.DetectionSchema object>, debug: bool = False) → Tuple[List[ndarray] | None, SpecialDSPParams | None, DSPDebug | None]

DSP in the case of a shared clock and an unshared local oscillator.

This simplifies the DSP, since there is no clock difference.

The procedure is the following:

Estimation of f_beat (to find the Zadoff-Chu sequence)
Recovery of the Zadoff-Chu sequence
Recovery of the pilot (per subframe)
Estimation of f_beat (per subframe)
Unshift signal (per subframe)
Apply match filter (per subframe)
Downsample (per subframe)
Correct relative phase noise (per subframe)

The output has still a global phase noise.

Parameters:

data (np.ndarray) – data measured by Bob.
symbol_rate (float) – symbol rate in Symbols per second.
dac_rate (float) – DAC rate, in Hz.
adc_rate (float) – ADC rate, in Hz.
num_symbols (int) – number of symbols.
roll_off (float) – roll off factor for the RRC filter.
frequency_shift (float) – frequency shift of the quantum symbols in Hz.
num_pilots (int) – number pilots.
pilots_frequencies (np.ndarray) – list of the frequencies of the pilots.
zc_length (int) – length of the Zadoff-Chu sequence.
zc_root (int) – root of the Zadoff-Chu sequence.
zc_rate (float) – shift, in Hz, of the Zadoff-Chu sequence.
process_subframes (bool, optional) – if True, process the data with subframes. Defaults to False.
subframe_length (int, optional) – number of symbols to recover in each subframe. Defaults to 0.
fir_size (int, optional) – size of the FIR filters.. Defaults to 500.
tone_filtering_cutoff (float, optional) – cutoff, in Hz, for the filtering of the pilots.. Defaults to 10e6.
excl (Optional[List[Tuple[float, float]]], optional) – exclusion zones for the research of pilots (i.e. frequencies where we are sure the pilots are not), given as a list of tuples of float, each elements defining excluded segment (start frequency, stop frequency). Defaults to None.
pilot_phase_filtering_size (int, optional) – size of the uniform1d filter to filter the phase correction. Defaults to 0.
schema (DetectionSchema, optional) – detection schema to use for the DSP. Defaults to qosst_core.schema.emission.SINGLE_POLARISATION_RF_HETERODYNE.
debug (bool, optional) – if True, the DSPDebug object is returned. Defaults to False.

Returns:

list of np.ndarray, each one corresponding to the recovered symbols for a subframe, SpecialDSPParams object to give to the special dsp, and DSPDebug object if debug was true.

Return type:

Tuple[List[np.ndarray], SpecialDSPParams, Optional[DSPDebug]]

qosst_bob.dsp.dsp._dsp_bob_unshared_clock_shared_lo(data: ~numpy.ndarray, symbol_rate: float, dac_rate: float, adc_rate: float, num_symbols: int, roll_off: float, frequency_shift: float, num_pilots: int, pilots_frequencies: ~numpy.ndarray, zc_length: int, zc_root: int, zc_rate: float, process_subframes: bool = False, subframe_length: int = 0, fir_size: int = 500, tone_filtering_cutoff: float = 10000000.0, pilot_phase_filtering_size: int = 0, schema: ~qosst_core.schema.detection.DetectionSchema = <qosst_core.schema.detection.DetectionSchema object>, debug: bool = False) → Tuple[List[ndarray] | None, SpecialDSPParams | None, DSPDebug | None]

DSP in the case of an unshared clock and a shared local oscillator.

This simplifies the DSP, since there is no frequency beat.

The procedure is the following:

Recovery of the Zadoff-Chu sequence
Recovery of the pilot (per subframe)
Recovery of the clock (per subframe)
Unshift signal (per subframe)
Apply match filter (per subframe)
Downsample (per subframe)
Correct relative phase noise (per subframe)

Parameters:

data (np.ndarray) – data measured by Bob.
symbol_rate (float) – symbol rate in Symbols per second.
dac_rate (float) – DAC rate, in Hz.
adc_rate (float) – ADC rate, in Hz.
num_symbols (int) – number of symbols.
roll_off (float) – roll-off factor for the RRC filter.
frequency_shift (float) – frequency shift of the quantum symbols, in Hz.
num_pilots (int) – number of pilots.
pilots_frequencies (np.ndarray) – list of the frequencies of the pilots.
zc_length (int) – length of the Zadoff-Chu sequence.
zc_root (int) – root of the Zadoff-Chu sequence.
zc_rate (float) – rate of the Zadoff-Chu sequence.
process_subframes (bool, optional) – if True, process the data in subframes. Defaults to False.
subframe_length (int, optional) – number of symbols to recover in each subframe. Defaults to 0.
fir_size (int, optional) – size of the FIR filters. Defaults to 500.
tone_filtering_cutoff (float, optional) – cutoff, in Hz, for the filtering of the tone. Defaults to 10e6.
pilot_phase_filtering_size (int, optional) – size of the uniform1d filter to filter the phase correction. Defaults to 0.
schema (DetectionSchema, optional) – detection schema to use for the DSP. Defaults to qosst_core.schema.emission.SINGLE_POLARISATION_RF_HETERODYNE.
debug (bool, optional) – if True, the DSPDebug object is returned. Defaults to False.

Returns:

list of np.ndarray, each one corresponding to the recovered symbols for a subframe, SpecialDSPParams object to give to the special dsp, and DSPDebug object if debug was true.

Return type:

Tuple[List[np.ndarray], SpecialDSPParams, Optional[DSPDebug]]

qosst_bob.dsp.dsp._dsp_bob_general(data: ~numpy.ndarray, symbol_rate: float, dac_rate: float, adc_rate: float, num_symbols: int, roll_off: float, frequency_shift: float, num_pilots: int, pilots_frequencies: ~numpy.ndarray, zc_length: int, zc_root: int, zc_rate: float, process_subframes: bool = False, subframe_length: int = 0, fir_size: int = 500, tone_filtering_cutoff: float = 10000000.0, abort_clock_recovery: float = 0, excl: ~typing.List[~typing.Tuple[float, float]] | None = None, pilot_phase_filtering_size: int = 0, num_samples_fbeat_estimation: int = 100000, schema: ~qosst_core.schema.detection.DetectionSchema = <qosst_core.schema.detection.DetectionSchema object>, debug: bool = False) → Tuple[List[ndarray] | None, SpecialDSPParams | None, DSPDebug | None]

General DSP.

The steps are the following:

Find an approximative start of the Zadoff-Chu sequence
Find the pilots
Correct clock difference
Find the pilots again with the good clock
Estimate the beat frequency
Find the Zadoff-Chu sequence
Estimate the beat frequency (per subframe)
Find one pilot (per subframe)
Unshift the quantum signal (per subframe)
Apply matched RRC filter (per subframe)
Downsample (per subframe)
Correct relative phase noise (per subframe)

The output has stil a global phase difference.

Parameters:

data (np.ndarray) – data measured by Bob.
symbol_rate (float) – symbol rate in symbols per second.
dac_rate (float) – DAC rate, in Hz.
adc_rate (float) – ADC rate, in Hz.
num_symbols (int) – number of symbols.
roll_off (float) – roll-off factor for the RRC filter.
frequency_shift (float) – frequency shift of the quantum symbol, in Hz.
num_pilots (int) – number of pilots.
pilots_frequencies (np.ndarray) – list of the frequencies of the pilots.
zc_length (int) – length of the Zadoff-Chu sequence.
zc_root (int) – root of the Zadoff-Chu sequence.
zc_rate (float) – rate of the Zadoff-Chu sequence.
process_subframes (bool, optional) – if True, process the data with subframes. Defaults to False.
subframe_length (int, optional) – number of symbols to recover in each subframes. Defaults to 0.
fir_size (int, optional) – size of the FIR filters. Defaults to 500.
tone_filtering_cutoff (float, optional) – cutoff, in Hz, for the pilot filtering. Defaults to 10e6.
abort_clock_recovery (float, optional) – maximal mismatch allowed by the clock recovery algorithm before aborting. If 0, the algorithm never aborts.. Defaults to 0.
excl (Optional[List[Tuple[float, float]]], optional) – exclusion zones for the research of pilots (i.e. frequencies where we are sure the pilots are not), given as a list of tuples of float, each elements defining excluded segment (start frequency, stop frequency). Defaults to None.
pilot_phase_filtering_size (int, optional) – size of the uniform1d filter to filter the phase correction. Defaults to 0.
num_samples_fbeat_estimation (int, optional) – number of samples for the estimation of fbeat. Defaults to 100000.
schema (DetectionSchema, optional) – detection schema to use for the DSP. Defaults to qosst_core.schema.emission.SINGLE_POLARISATION_RF_HETERODYNE.
debug (bool, optional) – if True, the DSPDebug object is returned. Defaults to False.

Returns:

list of np.ndarray, each one corresponding to the recovered symbols for a subframe, SpecialDSPParams object to give to the special dsp, and DSPDebug object if debug was true.

Return type:

Tuple[Optional[List[np.ndarray]], Optional[SpecialDSPParams], Optional[DSPDebug]]

qosst_bob.dsp.dsp.find_global_angle(received_data: ndarray, sent_data: ndarray, precision: float = 0.001) → Tuple[float, float]

Find global angle between received and sent data by exhaustive search.

The best angle is found when the real part of the covariance is the highset between the two sets. A certain number of angles will be tested to statisfy the required precision. In fact the number of tested points will ceil(2*pi/precision) with an actual precision of 2*pi/(number of points) with a precision lower or equal to the targeted precision.

The returned value is an angle in radian, between -pi and pi.

Parameters:

received_data (np.ndarray) – the symbols received by Bob after the DSP.
sent_data (np.ndarray) – the send symbols by Alice.
precision (float, optional) – the precision wanted on the angle, in radians. Defaults to 0.001.

Returns:

the angle that maximises the covariance, in radians, and the maximal covariance.

Return type:

Tuple[float,float]

qosst_bob.dsp.dsp.special_dsp(elec_noise_data: List[ndarray], elec_shot_noise_data: List[ndarray], params: SpecialDSPParams) → Tuple[ndarray, ndarray]

Special DSP to apply on the electronic and electronic and shot noises.

Parameters:

elec_noise_data (List[np.ndarray]) – list of arrays (for each channel) of electronic noise data.
elec_shot_noise_data (List[np.ndarray]) – list of arrays (for each channel) of electronic and shot noise data.
params (SpecialDSPParams) – the dictionnary returned by the DSP containing the required parameters.

Returns:

the electronic symbols and electronic and shot symbols.

Return type:

Tuple[np.ndarray, np.ndarray]

qosst_bob.dsp.dsp._special_dsp_params(elec_noise_data: ndarray, elec_shot_noise_data: ndarray, symbol_rate: float, adc_rate: float, roll_off: float, frequency_shift: float, _schema: DetectionSchema) → Tuple[ndarray, ndarray]

Special DSP to apply the electronic and electronic and shot noises taking the parameters.

Parameters:

elec_noise_data (np.ndarray) – array of the electronic noise.
elec_shot_noise_data (np.ndarray) – array of the electronic and shot noise.
symbol_rate (float) – symbol rate of the quantum symbols, in Symbols per second.
adc_rate (float) – ADC rate, in Hz.
roll_off (float) – roll-off factor of the RRC filter.
frequency_shift (float) – frequency shift of the quantum data, in Hz.
schema (DetectionSchema) – schema to know how to interpret the data.

Returns:

the electronic symbols and electronic and shot symbols.

Return type:

Tuple[np.ndarray, np.ndarray]

Equalizers

Module for equalization.

class qosst_bob.dsp.equalizers.CMAEqualizer(length: int, step: float, p_param: int = 2, q_param: int = 2, target_radius: float = 1, error_threshold: float = 0.02)

Channel equalizer based on the Constant Modulus Algorithm (CMA).

Initialize the CMA equalizer.

Parameters:

length (int) – length of the equalizer.
step (float) – step of the equalizer.
p_param (int, optional) – p parameter of the equalizer. Defaults to 2.
q_param (int, optional) – q parametwer of the equalizer. Defaults to 2.
target_radius (float, optional) – target radius of the equalizer. Defaults to 1.
error_threshold (float, optional) – error threshold. The training will stop wen the desired error is reached. Setting 0 will make the algorithm run on all the data. Defaults to 0.02.

length: int: Length of the equalizer.

step: float: Step of the equalizer.

p_param: int: P parameter of the equalizer.

q_param: int: Q parameter of the equalizer.

target_radius: float: Target radius of the equalizer.

error_threshold: float: Error threshold (training stop when this value is reached).

weights: ndarray | None: Weights of the equalizer.

train(train_data: ndarray) → Tuple[ndarray, ndarray | None, ndarray | None]

Train the CMA on the train data and updates the weights.

Parameters:: train_data (np.ndarray) – the data that should have a constant modulus.
Returns:: a tuple containing the corrected tain data (only before the error threshold was reached), the errors vector and the weights vector.
Return type:: Tuple[np.ndarray, Optional[np.ndarray], Optional[np.ndarray]]

apply(data: ndarray) → ndarray

Apply the equalizer to the data.

Parameters:: data (np.ndarray) – the data to apply the equalizer on.
Returns:: the equalized data.
Return type:: np.ndarray

Pilots

Pilots processing and related functions.

qosst_bob.dsp.pilots.recover_tones(data: ndarray, frequencies: List[float], rate: float, fir_size: int, cutoff: float = 10000000.0) → List[ndarray]

Recover all the tones within data given the list of frequencies of the tones.

This will get the tones by applying a FIR of size fir_size and cut-off cutoff on the data.

Parameters:

data (np.ndarray) – data on which the tones should be recovered.
frequencies (List[float]) – list of the frequencies of the tones.
rate (float) – rate of the data, in Samples per second.
fir_size (int) – the size of the FIR.
cutoff (float, optional) – the cut-off frequency of the filter, in Hz. Defaults to 10e6.

Returns:

the list of received tones.

Return type:

List[np.ndarray]

qosst_bob.dsp.pilots.recover_tone(data: ndarray, frequency: float, rate: float, fir_size: int, cutoff: float = 10000000.0) → ndarray

Recover the tone within data given the frequency of the tone.

This will get the tones by applying a FIR of size fir_size and cut-off cutoff on the data.

Parameters:

data (np.ndarray) – the data on which the tones should be recovered.
frequencies (float) – the frequency of the tone.
rate (float) – the rate of the data.
fir_size (int) – the size of the FIR.
cutoff (float, optional) – the cut-off frequency of the filter. Defaults to 10e6.

Returns:

the received tone.

Return type:

np.ndarray

qosst_bob.dsp.pilots.find_one_pilot(data: ndarray, rate: float, excl: List[Tuple[float, float]] | None = None) → float

Find the frequency of one pilot by looking at maximums in the FFT.

Parameters:

data (np.ndarray) – the data to analyze.
rate (float) – the sampling rate, in Samples per second.
excl (Optional[List[Tuple[float, float]]], optional) – List of exclusion zones. Each tuple will be considered as (beginning of exclusion zone in Hz, end of exclusion zone in Hz). Defaults to None.

Returns:

frequency of the pilot, in Hz.

Return type:

float

qosst_bob.dsp.pilots.find_two_pilots(data: ndarray, rate: float, tone_excl: float = 5000000.0, excl: List[Tuple[float, float]] | None = None) → Tuple[float, float]

Find the frequency of two pilots by looking at maximums in the FFT.

Parameters:

data (np.ndarray) – the data to analyze.
rate (float) – the sampling rate, in Samples per second.
tone_excl (float, optional) – exclusion zone after first pilot is found (f_pilot1 - tone_excl, f_pilot1 + tone_excl). Defaults to 5e6.
excl (List[Tuple[float, float]], optional) – List of exclusion zones. Each tuple will be considered as (beginning of exclusion zone in Hz, end of exclusion zone in Hz). Defaults to None.

Returns:

frequency of the first pilot and frequency of the second pilot, in Hz. freq1 < freq2.

Return type:

Tuple[float, float]

qosst_bob.dsp.pilots.equivalent_adc_rate_one_pilot(data: ndarray, frequency: float, rate: float, fir_size: int, cutoff: float = 10000000.0) → float

Find the equivalent ADC rate with a linear fit on the angle difference of the received tone.

Parameters:

data (np.ndarray) – the received data.
frequencies (float) – the frequency of the pilot tone.
rate (float) – the rate of the ADC in Hz.
fir_size (int) – the fir size of the filters for the tone recovery.
cutoff (float, optional) – the cut-off frequency of the filter. Defaults to 10e6.

Returns:

the equivalent ADC rate in Hz.

Return type:

float

qosst_bob.dsp.pilots.phase_noise_correction(received_tone: ndarray, frequency: float, rate: float) → ndarray

Return the phase difference to apply to correct the relative phase noise.

The phase noise is computed as the difference between the received tone and the expected tone.

Parameters:

received_tone (np.ndarray) – the received tone.
frequency (float) – the frequency of the received tone, in Hz.
rate (float) – the rate of the data, in Samples per second.

Returns:

the array of phase difference.

Return type:

np.ndarray

qosst_bob.dsp.pilots.correct_noise(data: ndarray, sampling_point: int, sps: float, received_tone: ndarray, frequency: float, rate: float, filter_size: int = 0) → ndarray

Correct phase noise using phase reference.

Parameters:

data (np.ndarray) – data to correct.
sampling_point (int) – the best sampling point found.
sps (float) – the value of samples per symbol.
received_tone (np.ndarray) – the data for the received tone.
frequency (float) – the frequency of this received tone, in Hz.
rate (float) – the sampling rate, in Samples per second.
filter_size (int, optional) – size of the uniform1d filter to apply to the phase. Defaults to 0.

Returns:

the corrected data.

Return type:

np.ndarray

Resample

Modules to resample data (mostly downsample) and associated functions.

qosst_bob.dsp.resample.downsample(data: ndarray, start_point: int, downsampling_factor: float) → ndarray

Downsample data starting with start_point and with a downsampling factor of downsampling_factor.

If the downsampling_factor is an integer, it uses the standard slice method.

Parameters:

data (np.ndarray) – the data to downsample.
start_point (int) – the start point: i.e. the first point in the downsample array is data[start_point].
downsampling_factor (float) – the downsampling factor: i.e. points in the downsampled arry are in the form data[start_point + k*downsampling_factor].

Returns:

downsampled data.

Return type:

np.ndarray

qosst_bob.dsp.resample._downsample_int(data: ndarray, start_point: int, downsampling_factor: int) → ndarray

Downsample data starting with start_point and with a downsampling factor of downsampling_factor, in the case where downsampling_factor is an integer.

It uses the standard slice method.

Parameters:

data (np.ndarray) – the data to downsample.
start_point (int) – the start point: i.e. the first point in the downsample array is data[start_point].
downsampling_factor (float) – the downsampling factor: i.e. points in the downsampled arry are in the form data[start_point + k*downsampling_factor].

Returns:

downsampled data.

Return type:

np.ndarray

qosst_bob.dsp.resample._downsample_float(data: ndarray, start_point: int, downsampling_factor: float) → ndarray

Downsample by a floating factor.

Usually the operation is to take the point start_point + k * downsampling_factor.

However, here downsampling_factor is a float so we want to compensate by approximating to the nearest integer with rint.

Hence if L denotes the length of the list, the requirement on k is that

rint(start_point + k * downsampling_factor) < L start_point+k*downsampling_factor <= L-0.5 k <= (L-0.5-start_point)/downsampling_factor

As k is an integer

k <= np.floor((L-0.5-start_point)/downsampling_factor)

As k must take this last value to recover everything the arange is np.arange(np.floor((L-0.5-start_point)/downsampling_factor) + 1)

Note that the rint function was not used as its behavior is weird when being exactly between two integers (i.e. it rounds to the nearest integer value) which is hard to take into account for every possibility.

Instead, we use np.ceil(x-0.5) that has the same behavior as np.rint except that is always round down when between exactly two values.

Parameters:

data (np.ndarray) – the data to downsample.
start_point (int) – the start point: i.e. the first point in the downsample array is data[start_point].
downsampling_factor (float) – the downsampling factor: i.e. points in the downsampled arry are in the form data[start_point + k*downsampling_factor].

Returns:

downsampled data.

Return type:

np.ndarray

qosst_bob.dsp.resample.best_sampling_point(data: ndarray, sps: float) → int

Find the best sampling point with maximal variance, for the downsampling after the matched filter.

The best sampling point is found by testing all the possible sampling point and by taking the one with maximal variance.

Parameters:

data (np.ndarray) – the data from which to find the best sampling point.
sps (float) – the samples per symbol value.

Returns:

the best sampling point.

Return type:

int

qosst_bob.dsp.resample._best_sampling_point_int(data: ndarray, sps: int) → int

Find the best sampling point with maximal variance, with the assumption that the sps is an integer value.

The best sampling point is found by testing all the possible sampling point and by taking the one with maximal variance.

Parameters:

data (np.ndarray) – the data from which to find the best sampling point.
sps (int) – the samples per symbol value.

Returns:

the best sampling point.

Return type:

int

qosst_bob.dsp.resample._best_sampling_point_float(data: ndarray, sps: float) → int

Find the best sampling point with maximal variance, with the assumption that the sps is not an integer value.

The best sampling point is found by testing all the possible sampling point and by taking the one with maximal variance.

Parameters:

data (np.ndarray) – the data from which to find the best sampling point.
sps (float) – the samples per symbol value.

Returns:

the best sampling point.

Return type:

int

ZC

DSP functions to deal with Zadoff-Chu sequences and synchronisation.

qosst_bob.dsp.zc.synchronisation_zc(data: ndarray, zc_root: int, zc_length: int, resample: float = 1, use_abs=True, ratio_approx=50) → Tuple[int, int]

Find the beginning of a Zadoff-Chu sequence in data.

This function finds the beginning and the end of a Zadoff-Chu sequence by computing the cross-correlation of the Zadoff-Chu sequence and the data.

From version 0.4.27, the behavior of this function is a little different:

first we find a first approximate of the Zadoff-Chu location by making a rolling average of the data, and then, we make the cross-correlation around this point.

Parameters:

data (np.ndarray) – the data from where the Zadoff-Chu should be found.
zc_root (int) – the root of the Zadoff-Chu sequence.
zc_length (int) – the length of the Zadoff-Chu sequence.
resample (float, optional) – the optional resample to apply to the Zadoff-Chu sequence. Defaults to 1.
ratio_approx (int, optional) – the length of the data will be divided by this value to get the window size of the rolling average for the approximation. Defaults to 50.

Returns:

tuple including the beginning and the end of the Zadoff-Chu sequence.

Return type:

Tuple[int, int]