summaryrefslogtreecommitdiffstats
path: root/dpd/src
diff options
context:
space:
mode:
Diffstat (limited to 'dpd/src')
-rw-r--r--dpd/src/Const.py2
-rw-r--r--dpd/src/Heuristics.py13
-rw-r--r--dpd/src/MER.py7
-rw-r--r--dpd/src/Measure.py21
-rw-r--r--dpd/src/Measure_Shoulders.py37
-rw-r--r--dpd/src/Model_AM.py2
-rw-r--r--dpd/src/Model_PM.py21
-rw-r--r--dpd/src/Symbol_align.py67
-rw-r--r--dpd/src/TX_Agc.py2
-rw-r--r--dpd/src/Test_data.py136
-rwxr-xr-xdpd/src/subsample_align.py19
-rw-r--r--dpd/src/test_dab_Util.py92
-rw-r--r--dpd/src/test_measure.py62
13 files changed, 105 insertions, 376 deletions
diff --git a/dpd/src/Const.py b/dpd/src/Const.py
index 390dfbb..ac48b59 100644
--- a/dpd/src/Const.py
+++ b/dpd/src/Const.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
#
-# DPD Calculation Engine, constants
+# DPD Calculation Engine, constants.
#
# Source for DAB standard: etsi_EN_300_401_v010401p p145
#
diff --git a/dpd/src/Heuristics.py b/dpd/src/Heuristics.py
index a32ccff..f98490d 100644
--- a/dpd/src/Heuristics.py
+++ b/dpd/src/Heuristics.py
@@ -1,26 +1,33 @@
# -*- coding: utf-8 -*-
#
-# DPD Calculation Engine, heuristics we use to tune the parameters
+# DPD Calculation Engine, heuristics we use to tune the parameters.
#
# http://www.opendigitalradio.org
# Licence: The MIT License, see notice at the end of this file
import numpy as np
+
def get_learning_rate(idx_run):
+ """Gradually reduce learning rate from lr_max to lr_min within
+ idx_max steps, then keep the learning rate at lr_min"""
idx_max = 10.0
lr_min = 0.05
lr_max = 0.4
lr_delta = lr_max - lr_min
idx_run = min(idx_run, idx_max)
- learning_rate = lr_max - lr_delta * idx_run/idx_max
+ learning_rate = lr_max - lr_delta * idx_run / idx_max
return learning_rate
+
def get_n_meas(idx_run):
+ """Gradually increase number of measurements used to extract
+ a statistic from n_meas_min to n_meas_max within idx_max steps,
+ then keep number of measurements at n_meas_max"""
idx_max = 10.0
n_meas_min = 10
n_meas_max = 20
n_meas_delta = n_meas_max - n_meas_min
idx_run = min(idx_run, idx_max)
- learning_rate = n_meas_delta * idx_run/idx_max + n_meas_min
+ learning_rate = n_meas_delta * idx_run / idx_max + n_meas_min
return int(np.round(learning_rate))
diff --git a/dpd/src/MER.py b/dpd/src/MER.py
index 69c94f9..0f169a7 100644
--- a/dpd/src/MER.py
+++ b/dpd/src/MER.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
#
-# Modulation Error Rate
+# DPD Calculation Engine, Modulation Error Rate.
#
# http://www.opendigitalradio.org
# Licence: The MIT License, see notice at the end of this file
@@ -13,7 +13,6 @@ try:
except:
logging_path = "/tmp/"
-import src.Const
import numpy as np
import matplotlib
matplotlib.use('agg')
@@ -72,8 +71,8 @@ class MER:
return x_mean, y_mean, U_RMS, U_ERR, MER
def calc_mer(self, tx, debug_name=""):
- assert tx.shape[0] == self.c.T_U,\
- "Wrong input length"
+ """Calculate MER for input signal from a symbol aligned signal."""
+ assert tx.shape[0] == self.c.T_U, "Wrong input length"
spectrum = self._calc_spectrum(tx)
diff --git a/dpd/src/Measure.py b/dpd/src/Measure.py
index 7a9246c..d485a86 100644
--- a/dpd/src/Measure.py
+++ b/dpd/src/Measure.py
@@ -6,14 +6,11 @@
# http://www.opendigitalradio.org
# Licence: The MIT License, see notice at the end of this file
-import sys
import socket
import struct
import numpy as np
-import logging
import src.Dab_Util as DU
import os
-import datetime
import logging
logging_path = os.path.dirname(logging.getLoggerClass().root.handlers[0].baseFilename)
@@ -78,15 +75,15 @@ class Measure:
rxframe = np.array([], dtype=np.complex64)
if logging.getLogger().getEffectiveLevel() == logging.DEBUG:
- logging.debug("txframe: min %f, max %f, median %f" %
- (np.min(np.abs(txframe)),
- np.max(np.abs(txframe)),
- np.median(np.abs(txframe))))
-
- logging.debug("rxframe: min %f, max %f, median %f" %
- (np.min(np.abs(rxframe)),
- np.max(np.abs(rxframe)),
- np.median(np.abs(rxframe))))
+ logging.debug('txframe: min {}, max {}, median {}'.format(
+ np.min(np.abs(txframe)),
+ np.max(np.abs(txframe)),
+ np.median(np.abs(txframe))))
+
+ logging.debug('rxframe: min {}, max {}, median {}'.format(
+ np.min(np.abs(rxframe)),
+ np.max(np.abs(rxframe)),
+ np.median(np.abs(rxframe))))
logging.debug("Disconnecting")
s.close()
diff --git a/dpd/src/Measure_Shoulders.py b/dpd/src/Measure_Shoulders.py
index 7d48a2b..4cf7d0d 100644
--- a/dpd/src/Measure_Shoulders.py
+++ b/dpd/src/Measure_Shoulders.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
#
-# DPD Calculation Engine, calculate peak to shoulder difference
+# DPD Calculation Engine, calculate peak to shoulder difference.
#
# http://www.opendigitalradio.org
# Licence: The MIT License, see notice at the end of this file
@@ -15,28 +15,35 @@ logging_path = os.path.dirname(logging.getLoggerClass().root.handlers[0].baseFil
import numpy as np
import matplotlib.pyplot as plt
+
def plt_next_axis(sub_rows, sub_cols, i_sub):
i_sub += 1
ax = plt.subplot(sub_rows, sub_cols, i_sub)
return i_sub, ax
-def plt_annotate(ax, x,y,title=None,legend_loc=None):
+
+def plt_annotate(ax, x, y, title=None, legend_loc=None):
ax.set_xlabel(x)
ax.set_ylabel(y)
- if title is not None: ax.set_title(title)
- if legend_loc is not None: ax.legend(loc=legend_loc)
+ if title is not None:
+ ax.set_title(title)
+ if legend_loc is not None:
+ ax.legend(loc=legend_loc)
+
def calc_fft_db(signal, offset, c):
fft = np.fft.fftshift(np.fft.fft(signal[offset:offset + c.MS_FFT_size]))
fft_db = 20 * np.log10(np.abs(fft))
return fft_db
+
def _calc_peak(fft, c):
assert fft.shape == (c.MS_FFT_size,), fft.shape
idxs = (c.MS_peak_start, c.MS_peak_end)
peak = np.mean(fft[idxs[0]:idxs[1]])
return peak, idxs
+
def _calc_shoulder_hight(fft_db, c):
assert fft_db.shape == (c.MS_FFT_size,), fft_db.shape
idxs_left = (c.MS_shoulder_left_start, c.MS_shoulder_left_end)
@@ -48,26 +55,26 @@ def _calc_shoulder_hight(fft_db, c):
shoulder = np.mean((shoulder_left, shoulder_right))
return shoulder, (idxs_left, idxs_right)
+
def calc_shoulder(fft, c):
peak = _calc_peak(fft, c)[0]
shoulder = _calc_shoulder_hight(fft, c)[0]
assert (peak >= shoulder), (peak, shoulder)
return peak, shoulder
+
def shoulder_from_sig_offset(arg):
signal, offset, c = arg
fft_db = calc_fft_db(signal, offset, c)
peak, shoulder = calc_shoulder(fft_db, c)
- return peak-shoulder, peak, shoulder
+ return peak - shoulder, peak, shoulder
class Measure_Shoulders:
"""Calculate difference between the DAB signal and the shoulder hight in the
power spectrum"""
- def __init__(self,
- c,
- plot=False):
+ def __init__(self, c):
self.c = c
self.plot = c.MS_plot
@@ -75,9 +82,9 @@ class Measure_Shoulders:
dt = datetime.datetime.now().isoformat()
fig_path = logging_path + "/" + dt + "_sync_subsample_aligned.svg"
- fft = calc_fft_db(signal, 100, 10)
- peak, idxs_peak = self._calc_peak(fft)
- shoulder, idxs_sh = self._calc_shoulder_hight(fft, self.c)
+ fft = calc_fft_db(signal, 100, self.c)
+ peak, idxs_peak = _calc_peak(fft, self.c)
+ shoulder, idxs_sh = _calc_shoulder_hight(fft, self.c)
sub_rows = 1
sub_cols = 1
@@ -93,7 +100,7 @@ class Measure_Shoulders:
ax.plot(idxs_sh[1], (shoulder, shoulder), color='blue')
plt_annotate(ax, "Frequency", "Magnitude [dB]", None, 4)
- ax.text(100, -17, str(self.calc_shoulder(fft, self.c)))
+ ax.text(100, -17, str(calc_shoulder(fft, self.c)))
ax.set_ylim(-20, 30)
fig.tight_layout()
@@ -106,14 +113,13 @@ class Measure_Shoulders:
return None
assert signal.shape[0] > 4 * self.c.MS_FFT_size
- if n_avg is None: n_avg = self.c.MS_averaging_size
+ if n_avg is None:
+ n_avg = self.c.MS_averaging_size
off_min = 0
off_max = signal.shape[0] - self.c.MS_FFT_size
offsets = np.linspace(off_min, off_max, num=n_avg, dtype=int)
- shoulders = []
-
args = zip(
[signal, ] * offsets.shape[0],
offsets,
@@ -129,7 +135,6 @@ class Measure_Shoulders:
return np.mean(shoulders_diff), np.mean(shoulders), np.mean(peaks)
-
# The MIT License (MIT)
#
# Copyright (c) 2017 Andreas Steger
diff --git a/dpd/src/Model_AM.py b/dpd/src/Model_AM.py
index cdc3de1..a40f421 100644
--- a/dpd/src/Model_AM.py
+++ b/dpd/src/Model_AM.py
@@ -90,6 +90,8 @@ class Model_AM:
plt.close(fig)
def get_next_coefs(self, tx_dpd, rx_received, coefs_am):
+ """Calculate the next AM/AM coefficients using the extracted
+ statistic of TX and RX amplitude"""
check_input_get_next_coefs(tx_dpd, rx_received)
coefs_am_new = fit_poly(tx_dpd, rx_received)
diff --git a/dpd/src/Model_PM.py b/dpd/src/Model_PM.py
index e0fcb55..b0c7ae2 100644
--- a/dpd/src/Model_PM.py
+++ b/dpd/src/Model_PM.py
@@ -15,17 +15,16 @@ import numpy as np
import matplotlib.pyplot as plt
+def is_npfloat32(array):
+ assert isinstance(array, np.ndarray), type(array)
+ assert array.dtype == np.float32, array.dtype
+ assert array.flags.contiguous
+ assert not any(np.isnan(array))
+
+
def check_input_get_next_coefs(tx_dpd, phase_diff):
- is_float32 = lambda x: (isinstance(x, np.ndarray) and
- x.dtype == np.float32 and
- x.flags.contiguous)
- assert is_float32(tx_dpd), \
- "tx_dpd is not float32 but {}".format(tx_dpd[0].dtype)
- assert is_float32(phase_diff), \
- "phase_diff is not float32 but {}".format(tx_dpd[0].dtype)
- assert tx_dpd.shape == phase_diff.shape, \
- "tx_dpd.shape {}, phase_diff.shape {}".format(
- tx_dpd.shape, phase_diff.shape)
+ is_npfloat32(tx_dpd)
+ is_npfloat32(phase_diff)
class Model_PM:
@@ -93,6 +92,8 @@ class Model_PM:
return tx_range, phase_diff
def get_next_coefs(self, tx_dpd, phase_diff, coefs_pm):
+ """Calculate the next AM/PM coefficients using the extracted
+ statistic of TX amplitude and phase difference"""
tx_dpd, phase_diff = self._discard_small_values(tx_dpd, phase_diff)
check_input_get_next_coefs(tx_dpd, phase_diff)
diff --git a/dpd/src/Symbol_align.py b/dpd/src/Symbol_align.py
index e21e793..4ea79c5 100644
--- a/dpd/src/Symbol_align.py
+++ b/dpd/src/Symbol_align.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
#
-# Modulation Error Rate
+# DPD Calculation Engine, Modulation Error Rate.
#
# http://www.opendigitalradio.org
# Licence: The MIT License, see notice at the end of this file
@@ -8,24 +8,50 @@
import datetime
import os
import logging
-import time
+
try:
logging_path = os.path.dirname(logging.getLoggerClass().root.handlers[0].baseFilename)
except:
logging_path = "/tmp/"
import numpy as np
-import src.Const
import scipy
import matplotlib
+
matplotlib.use('agg')
import matplotlib.pyplot as plt
+
+def _remove_outliers(x, stds=5):
+ deviation_from_mean = np.abs(x - np.mean(x))
+ inlier_idxs = deviation_from_mean < stds * np.std(x)
+ x = x[inlier_idxs]
+ return x
+
+
+def _calc_delta_angle(fft):
+ # Introduce invariance against carrier
+ angles = np.angle(fft) % (np.pi / 2.)
+
+ # Calculate Angle difference and compensate jumps
+ deltas_angle = np.diff(angles)
+ deltas_angle[deltas_angle > np.pi / 4.] = \
+ deltas_angle[deltas_angle > np.pi / 4.] - np.pi / 2.
+ deltas_angle[-deltas_angle > np.pi / 4.] = \
+ deltas_angle[-deltas_angle > np.pi / 4.] + np.pi / 2.
+ deltas_angle = _remove_outliers(deltas_angle)
+
+ delta_angle = np.mean(deltas_angle)
+
+ return delta_angle
+
+
class Symbol_align:
"""
Find the phase offset to the start of the DAB symbols in an
unaligned dab signal.
"""
+
def __init__(self, c, plot=False):
self.c = c
self.plot = plot
@@ -95,8 +121,8 @@ class Symbol_align:
ax.set_title("Min Filter")
ax = fig.add_subplot(4, 1, 4)
- tx_product_crop = tx_product[peaks[0]-50:peaks[0]+50]
- x = range(tx_product.shape[0])[peaks[0]-50:peaks[0]+50]
+ tx_product_crop = tx_product[peaks[0] - 50:peaks[0] + 50]
+ x = range(tx_product.shape[0])[peaks[0] - 50:peaks[0] + 50]
ax.plot(x, tx_product_crop)
ylim = ax.get_ylim()
ax.plot((peaks[0], peaks[0]), (ylim[0], ylim[1]))
@@ -113,38 +139,16 @@ class Symbol_align:
# of the shift to find the offset of the symbol start.
return (offset + self.c.T_C) % self.c.T_S
- def _remove_outliers(self, x, stds=5):
- deviation_from_mean = np.abs(x - np.mean(x))
- inlier_idxs = deviation_from_mean < stds * np.std(x)
- x = x[inlier_idxs]
- return x
-
- def _calc_delta_angle(self, fft):
- # Introduce invariance against carrier
- angles = np.angle(fft) % (np.pi / 2.)
-
- # Calculate Angle difference and compensate jumps
- deltas_angle = np.diff(angles)
- deltas_angle[deltas_angle > np.pi/4.] =\
- deltas_angle[deltas_angle > np.pi/4.] - np.pi/2.
- deltas_angle[-deltas_angle > np.pi/4.] = \
- deltas_angle[-deltas_angle > np.pi/4.] + np.pi/2.
- deltas_angle = self._remove_outliers(deltas_angle)
-
- delta_angle = np.mean(deltas_angle)
-
- return delta_angle
-
def _delta_angle_to_samples(self, angle):
return - angle / self.c.phase_offset_per_sample
- def _calc_sample_offset(self, sig, debug=False):
- assert sig.shape[0] == self.c.T_U,\
+ def _calc_sample_offset(self, sig):
+ assert sig.shape[0] == self.c.T_U, \
"Input length is not a Symbol without cyclic prefix"
fft = np.fft.fftshift(np.fft.fft(sig))
fft_crop = np.delete(fft[self.c.FFT_start:self.c.FFT_end], self.c.FFT_delete)
- delta_angle = self._calc_delta_angle(fft_crop)
+ delta_angle = _calc_delta_angle(fft_crop)
delta_sample = self._delta_angle_to_samples(delta_angle)
delta_sample_int = np.round(delta_sample).astype(int)
error = np.abs(delta_sample_int - delta_sample)
@@ -154,6 +158,7 @@ class Symbol_align:
return delta_sample_int
def calc_offset(self, tx):
+ """Calculate the offset the first symbol"""
off_sym = self._calc_offset_to_first_symbol_without_prefix(
tx)
off_sam = self._calc_sample_offset(
@@ -168,7 +173,7 @@ class Symbol_align:
off = self.calc_offset(tx)
return tx[
off:
- off+self.c.T_U
+ off + self.c.T_U
]
# The MIT License (MIT)
diff --git a/dpd/src/TX_Agc.py b/dpd/src/TX_Agc.py
index 2602ea6..857ae06 100644
--- a/dpd/src/TX_Agc.py
+++ b/dpd/src/TX_Agc.py
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
#
-# Automatic Gain Control
+# DPD Calculation Engine, Automatic Gain Control.
#
# http://www.opendigitalradio.org
# Licence: The MIT License, see notice at the end of this file
diff --git a/dpd/src/Test_data.py b/dpd/src/Test_data.py
deleted file mode 100644
index bbef282..0000000
--- a/dpd/src/Test_data.py
+++ /dev/null
@@ -1,136 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# Modulation Error Rate
-#
-# http://www.opendigitalradio.org
-# Licence: The MIT License, see notice at the end of this file
-
-import datetime
-import os
-import logging
-import time
-try:
- logging_path = os.path.dirname(logging.getLoggerClass().root.handlers[0].baseFilename)
-except:
- logging_path = "/tmp/"
-
-import src.Const
-import src.Dab_Util
-import numpy as np
-import matplotlib
-matplotlib.use('agg')
-import matplotlib.pyplot as plt
-
-
-class Test_data:
- def __init__(self, sample_rate, type):
- """
- Standardized access to complex64 test data files.
-
- :param sample_rate:
- :param type:
-
- Testing:
- TD = src.Test_data.Test_data(8192000, 'file')
- tx_orig = TD.get_symbol(0,0)
- fig = plt.figure(figsize=(9,6))
- ax = fig.add_subplot(2,1,1)
- ax.plot(tx_orig)
- ax = fig.add_subplot(2,1,2)
- plt.plot(np.angle(np.fft.fftshift(np.fft.fft(tx_orig))), 'p')
- """
-
- self.c = src.Const.Const(sample_rate,, False
- self.du = src.Dab_Util.Dab_Util(sample_rate)
-
- self.file_paths = {
- (2048000, 'file'):
- ("./test_data/odr-dabmod_to_file_2048_NoFir_noDPD.iq",
- (
- self.c.T_F + # Pipelineing
- self.c.T_NULL + # NULL Symbol
- self.c.T_S + # Synchronization Symbol
- self.c.T_C # Cyclic Prefix
- )),
- (8192000, 'file'):
- ("./test_data/odr-dabmod_to_file_8192_NoFir_noDPD.iq",
- (
- self.c.T_F + # Pipelining
- self.c.T_U + # Pipelining Resampler TODO(?)
- self.c.T_NULL + # NULL Symbol
- self.c.T_S + # Synchronization Symbol
- self.c.T_C # Cyclic Prefix
- )),
- (8192000, 'rec_noFir'):
- ("./test_data/odr-dabmod_reconding_8192_NoFir_DPD_2104.iq",
- ( 64 )),
- (8192000, 'rec_fir'):
- ("./test_data/odr-dabmod_reconding_8192_Fir_DPD_2104.iq",
- ( 232 )),
- }
-
- config = (sample_rate, type)
- if not config in self.file_paths.keys():
- raise RuntimeError("Configuration not found, possible are:\n {}".
- format(self.file_paths))
-
- self.path, self.file_offset = self.file_paths[(sample_rate, type)]
-
- def _load_from_file(self, offset, length):
- print(offset, length, self.file_offset)
- return self.du.fromfile(
- self.path,
- length=length,
- offset=offset + self.file_offset)
-
- def get_symbol_without_prefix(self,
- frame_idx=0,
- symbol_idx=0,
- off=0):
- return self._load_from_file(
- frame_idx*self.c.T_F +
- symbol_idx*self.c.T_S +
- off,
- self.c.T_U)
-
- def get_symbol_with_prefix(self,
- frame_idx=0,
- symbol_idx=0,
- n_symbols=1,
- off=0):
- offset = (
- frame_idx*self.c.T_F +
- symbol_idx*self.c.T_S -
- self.c.T_C +
- off)
- return self._load_from_file( offset, self.c.T_S * n_symbols)
-
- def get_file_length_in_symbols(self):
- symbol_size = float(
- 64/8 * # complex 64
- self.c.T_S # Symbol Size
- )
- return os.path.getsize(self.path) / symbol_size
-
-
-# The MIT License (MIT)
-#
-# Copyright (c) 2017 Andreas Steger
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
diff --git a/dpd/src/subsample_align.py b/dpd/src/subsample_align.py
index b0cbe88..0c98e50 100755
--- a/dpd/src/subsample_align.py
+++ b/dpd/src/subsample_align.py
@@ -1,23 +1,25 @@
# -*- coding: utf-8 -*-
#
-# DPD Calculation Engine, utility to do subsample alignment
+# DPD Calculation Engine, utility to do subsample alignment.
#
# http://www.opendigitalradio.org
# Licence: The MIT License, see notice at the end of this file
import datetime
-import os
import logging
+import os
+
logging_path = os.path.dirname(logging.getLoggerClass().root.handlers[0].baseFilename)
import numpy as np
-from scipy import signal, optimize
+from scipy import optimize
import matplotlib.pyplot as plt
+
def gen_omega(length):
if (length % 2) == 1:
raise ValueError("Needs an even length array.")
- halflength = int(length/2)
+ halflength = int(length / 2)
factor = 2.0 * np.pi / length
omega = np.zeros(length, dtype=np.float)
@@ -29,6 +31,7 @@ def gen_omega(length):
return omega
+
def subsample_align(sig, ref_sig, plot=False):
"""Do subsample alignment for sig relative to the reference signal
ref_sig. The delay between the two must be less than sample
@@ -38,7 +41,7 @@ def subsample_align(sig, ref_sig, plot=False):
n = len(sig)
if (n % 2) == 1:
raise ValueError("Needs an even length signal.")
- halflen = int(n/2)
+ halflen = int(n / 2)
fft_sig = np.fft.fft(sig)
@@ -63,7 +66,8 @@ def subsample_align(sig, ref_sig, plot=False):
return -np.abs(np.sum(np.conj(corr_sig) * ref_sig))
- optim_result = optimize.minimize_scalar(correlate_for_delay, bounds=(-1,1), method='bounded', options={'disp': True})
+ optim_result = optimize.minimize_scalar(correlate_for_delay, bounds=(-1, 1), method='bounded',
+ options={'disp': True})
if optim_result.success:
best_tau = optim_result.x
@@ -85,10 +89,9 @@ def subsample_align(sig, ref_sig, plot=False):
rotate_vec[halflen] = np.cos(np.pi * best_tau)
return np.fft.ifft(rotate_vec * fft_sig).astype(np.complex64)
else:
- #print("Could not optimize: " + optim_result.message)
+ # print("Could not optimize: " + optim_result.message)
return np.zeros(0, dtype=np.complex64)
-
# The MIT License (MIT)
#
# Copyright (c) 2017 Andreas Steger
diff --git a/dpd/src/test_dab_Util.py b/dpd/src/test_dab_Util.py
deleted file mode 100644
index 86ee2d8..0000000
--- a/dpd/src/test_dab_Util.py
+++ /dev/null
@@ -1,92 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# Test code for DAB util
-#
-# http://www.opendigitalradio.org
-# Licence: The MIT License, see notice at the end of this file
-
-from unittest import TestCase
-
-import numpy as np
-import pandas as pd
-import src.Dab_Util as DU
-
-class TestDab_Util(TestCase):
-
- def test_subsample_align(self, sample_orig=r'../test_data/orig_rough_aligned.dat',
- sample_rec =r'../test_data/recored_rough_aligned.dat',
- length = 10240, max_size = 1000000):
- du = DU
- res1 = []
- res2 = []
- for i in range(10):
- start = np.random.randint(50, max_size)
- r = np.random.randint(-50, 50)
-
- s1 = du.fromfile(sample_orig, offset=start+r, length=length)
- s2 = du.fromfile(sample_rec, offset=start, length=length)
-
- res1.append(du.lag_upsampling(s2, s1, 32))
-
- s1_aligned, s2_aligned = du.subsample_align(s1, s2)
-
- res2.append(du.lag_upsampling(s2_aligned, s1_aligned, 32))
-
- error_rate = np.mean(np.array(res2) != 0)
- self.assertEqual(error_rate, 0.0, "The error rate for aligning was %.2f%%"
- % error_rate * 100)
-
-#def test_using_aligned_pair(sample_orig=r'../data/orig_rough_aligned.dat', sample_rec =r'../data/recored_rough_aligned.dat', length = 10240, max_size = 1000000):
-# res = []
-# for i in tqdm(range(100)):
-# start = np.random.randint(50, max_size)
-# r = np.random.randint(-50, 50)
-#
-# s1 = du.fromfile(sample_orig, offset=start+r, length=length)
-# s2 = du.fromfile(sample_rec, offset=start, length=length)
-#
-# res.append({'offset':r,
-# '1':r - du.lag_upsampling(s2, s1, n_up=1),
-# '2':r - du.lag_upsampling(s2, s1, n_up=2),
-# '3':r - du.lag_upsampling(s2, s1, n_up=3),
-# '4':r - du.lag_upsampling(s2, s1, n_up=4),
-# '8':r - du.lag_upsampling(s2, s1, n_up=8),
-# '16':r - du.lag_upsampling(s2, s1, n_up=16),
-# '32':r - du.lag_upsampling(s2, s1, n_up=32),
-# })
-# df = pd.DataFrame(res)
-# df = df.reindex_axis(sorted(df.columns), axis=1)
-# print(df.describe())
-#
-#
-#print("Align using upsampling")
-#for n_up in [1, 2, 3, 4, 7, 8, 16]:
-# correct_ratio = test_phase_offset(lambda x,y: du.lag_upsampling(x,y,n_up), tol=1./n_up)
-# print("%.1f%% of the tested offsets were measured within tolerance %.4f for n_up = %d" % (correct_ratio * 100, 1./n_up, n_up))
-#test_using_aligned_pair()
-#
-#print("Phase alignment")
-#test_subsample_alignment()
-
-
-# The MIT License (MIT)
-#
-# Copyright (c) 2017 Andreas Steger
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
diff --git a/dpd/src/test_measure.py b/dpd/src/test_measure.py
deleted file mode 100644
index ee3cfdb..0000000
--- a/dpd/src/test_measure.py
+++ /dev/null
@@ -1,62 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# DPD Calculation Engine, test case for measure
-#
-# http://www.opendigitalradio.org
-# Licence: The MIT License, see notice at the end of this file
-from unittest import TestCase
-from Measure import Measure
-import socket
-
-
-class TestMeasure(TestCase):
-
- def _open_socks(self):
- sock_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- sock_server.bind(('localhost', 1234))
- sock_server.listen(1)
-
- sock_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock_client.connect(('localhost', 1234))
-
- conn_server, addr_server = sock_server.accept()
- return conn_server, sock_client
-
- def test__recv_exact(self):
- m = Measure(1234, 1)
- payload = b"test payload"
-
- conn_server, sock_client = self._open_socks()
- conn_server.send(payload)
- rec = m._recv_exact(sock_client, len(payload))
-
- self.assertEqual(rec, payload,
- "Did not receive the same message as sended. (%s, %s)" %
- (rec, payload))
-
- def test_get_samples(self):
- self.fail()
-
-
-# The MIT License (MIT)
-#
-# Copyright (c) 2017 Andreas Steger
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-28 17:28:05 +0000