refactoring analytics

NanoVNASaver/Analysis/PeakSearchAnalysis.py

@@ -26,9 +26,6 @@ from NanoVNASaver.Analysis.Base import QHLine
 from NanoVNASaver.Analysis.SimplePeakSearchAnalysis import (
     SimplePeakSearchAnalysis)
 
-from NanoVNASaver.Formatting import format_vswr
-from NanoVNASaver.Formatting import format_gain
-from NanoVNASaver.Formatting import format_resistance
 from NanoVNASaver.Formatting import format_frequency_short
 
 
@@ -58,50 +55,19 @@ class PeakSearchAnalysis(SimplePeakSearchAnalysis):
         self.reset()
 
         s11 = self.app.data.s11
-        s21 = self.app.data.s21
+        data, fmt_fnc = self.data_and_format()
 
-        if not s21:
-            self.button['gain'].setEnabled(False)
-            if self.button['gain'].isChecked():
-                self.button['vswr'].setChecked(True)
+        inverted = False
+        if self.button['peak_l'].isChecked():
+            inverted = True
+            peaks, _ = scipy.signal.find_peaks(
+                -np.array(data), width=3, distance=3, prominence=1)
         else:
-            self.button['gain'].setEnabled(True)
-
-        count = self.peak_cnt.value()
-        if self.button['vswr'].isChecked():
-            fn = format_vswr
-            data = [d.vswr for d in s11]
-        elif self.button['gain'].isChecked():
-            fn = format_gain
-            data = [d.gain for d in s21]
-        elif self.button['resistance'].isChecked():
-            fn = format_resistance
-            data = [d.impedance().real for d in s11]
-        elif self.button['reactance'].isChecked():
-            fn = format_resistance
-            data = [d.impedance().imag for d in s11]
-        else:
-            logger.warning("Searching for peaks on unknown data")
-            return
-
-        sign = 1
-        if self.button['peak_h'].isChecked():
+            self.button['peak_h'].setChecked(True)
             peaks, _ = scipy.signal.find_peaks(
                 data, width=3, distance=3, prominence=1)
-        elif self.button['peak_l'].isChecked():
-            sign = -1
-            data = [x * sign for x in data]
-            peaks, _ = scipy.signal.find_peaks(
-                data, width=3, distance=3, prominence=1)
-        else:
-            # Both is not yet in
-            logger.warning(
-                "Searching for peaks,"
-                " but neither looking at positive nor negative?")
-            return
 
         # Having found the peaks, get the prominence data
-
         for i, p in np.ndenumerate(peaks):
             logger.debug("Peak %i at %d", i, p)
         prominences = scipy.signal.peak_prominences(data, peaks)[0]
@@ -109,19 +75,20 @@ class PeakSearchAnalysis(SimplePeakSearchAnalysis):
 
         # Find the peaks with the most extreme values
         # Alternately, allow the user to select "most prominent"?
+        count = self.peak_cnt.value()
+        if count > len(prominences):
+            count = len(prominences)
+            self.peak_cnt.setValue(count)
+
         indices = np.argpartition(prominences, -count)[-count:]
         logger.debug("%d indices", len(indices))
         for i in indices:
-            logger.debug("Index %d", i)
-            logger.debug("Prominence %f", prominences[i])
-            logger.debug("Index in sweep %d", peaks[i])
-            logger.debug("Frequency %d", s11[peaks[i]].freq)
-            logger.debug("Value %f", sign * data[peaks[i]])
+            pos = peaks[i]
             self.layout.addRow(
-                f"Freq"
-                f" {format_frequency_short(s11[peaks[i]].freq)}",
-                QtWidgets.QLabel(f" value {fn(sign * data[peaks[i]])}"
-                                 ))
+                f"Freq: {format_frequency_short(s11[pos].freq)}",
+                QtWidgets.QLabel(
+                    f" Value: {fmt_fnc(-data[pos] if inverted else data[pos])}"
+                ))
 
         if self.button['move_marker'].isChecked():
             if count > len(self.app.markers):
@@ -130,18 +97,8 @@ class PeakSearchAnalysis(SimplePeakSearchAnalysis):
                 self.app.markers[i].setFrequency(
                     str(s11[peaks[indices[i]]].freq))
 
-        max_val = -10**10
-        max_idx = -1
-        for p in peaks:
-            if data[p] > max_val:
-                max_val = data[p]
-                max_idx = p
-
-        logger.debug("Max peak at %d, value %f", max_idx, max_val)
-
     def reset(self):
-        logger.debug("Reset analysis")
-
+        super().reset()
         logger.debug("Results start at %d, out of %d",
                      self.results_header, self.layout.rowCount())
         for _ in range(self.results_header, self.layout.rowCount()):

NanoVNASaver/Analysis/ResonanceAnalysis.py

@@ -196,7 +196,7 @@ class EFHWAnalysis(ResonanceAnalysis):
                     f" ({diff[i]['r']}) {diff[i]['lambda']} m"))
 
         if filename and extended_data:
-            with open(filename, 'w', newline='') as csvfile:
+            with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
                 fieldnames = extended_data[sorted(
                     extended_data.keys())[0]].keys()
                 writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

NanoVNASaver/Analysis/SimplePeakSearchAnalysis.py

@@ -17,13 +17,14 @@
 #  You should have received a copy of the GNU General Public License
 #  along with this program.  If not, see <https://www.gnu.org/licenses/>.
 import logging
+from typing import Callable, List, Tuple
 
 from PyQt5 import QtWidgets
 import numpy as np
 
 from NanoVNASaver.Analysis.Base import Analysis, QHLine
-from NanoVNASaver.Formatting import format_frequency
-
+from NanoVNASaver.Formatting import (
+    format_frequency, format_gain, format_resistance, format_vswr)
 
 logger = logging.getLogger(__name__)
 
@@ -82,6 +83,22 @@ class SimplePeakSearchAnalysis(Analysis):
         if not self.app.data.s11:
             return
 
+        s11 = self.app.data.s11
+        data, fmt_fnc = self.data_and_format()
+
+        if self.button['peak_l'].isChecked():
+            idx_peak = np.argmin(data)
+        else:
+            self.button['peak_h'].setChecked(True)
+            idx_peak = np.argmax(data)
+
+        self.label['peak_freq'].setText(format_frequency(s11[idx_peak].freq))
+        self.label['peak_db'].setText(fmt_fnc(data[idx_peak]))
+
+        if self.button['move_marker'].isChecked() and self.app.markers:
+            self.app.markers[0].setFrequency(f"{s11[idx_peak].freq}")
+
+    def data_and_format(self) -> Tuple[List[float], Callable]:
         s11 = self.app.data.s11
         s21 = self.app.data.s21
 
@@ -92,36 +109,11 @@ class SimplePeakSearchAnalysis(Analysis):
         else:
             self.button['gain'].setEnabled(True)
 
-        if self.button['vswr'].isChecked():
-            suffix = ""
-            data = [d.vswr for d in s11]
-        elif self.button['resistance'].isChecked():
-            suffix = " \N{OHM SIGN}"
-            data = [d.impedance().real for d in s11]
-        elif self.button['reactance'].isChecked():
-            suffix = " \N{OHM SIGN}"
-            data = [d.impedance().imag for d in s11]
-        elif self.button['gain'].isChecked():
-            suffix = " dB"
-            data = [d.gain for d in s21]
-        else:
-            logger.warning("Searching for peaks on unknown data")
-            return
-
-        if self.button['peak_h'].isChecked():
-            idx_peak = np.argmax(data)
-        elif self.button['peak_l'].isChecked():
-            idx_peak = np.argmin(data)
-        else:
-            # Both is not yet in
-            logger.warning(
-                "Searching for peaks,"
-                " but neither looking at positive nor negative?")
-            return
-
-        self.label['peak_freq'].setText(
-            format_frequency(s11[idx_peak].freq))
-        self.label['peak_db'].setText(f"{round(data[idx_peak], 3)}{suffix}")
-
-        if self.button['move_marker'].isChecked() and self.app.markers:
-            self.app.markers[0].setFrequency(f"{s11[idx_peak].freq}")
+        if self.button['gain'].isChecked():
+            return ([d.gain for d in s21], format_gain)
+        if self.button['resistance'].isChecked():
+            return ([d.impedance().real for d in s11], format_resistance)
+        if self.button['reactance'].isChecked():
+            return ([d.impedance().imag for d in s11], format_resistance)
+        # default
+        return ([d.vswr for d in s11], format_vswr)

NanoVNASaver/AnalyticTools.py

@@ -115,7 +115,7 @@ def center_from_idx(gains: List[float],
     Args:
         gains (List[float]): gain values
         idx (int): start position to search from
-        delta (float=3.0): max gain delta from start
+        delta (float, optional): max gain delta from start. Defaults to 3.0.
 
     Returns:
         int: position of highest gain from start in range (-1 if no data)
@@ -128,6 +128,19 @@ def center_from_idx(gains: List[float],
 
 def cut_off_left(gains: List[float], idx: int,
                  peak_gain: float, attn: float = 3.0) -> int:
+    """find first position in list where gain in attn lower then peak
+    left from index
+
+    Args:
+        gains (List[float]): gain values
+        idx (int): start position to search from
+        peak_gain (float): reference gain value
+        attn (float, optional): attenuation to search position for.
+                                Defaults to 3.0.
+
+    Returns:
+        int: position of attenuation point. (-1 if no data)
+    """
     return next(
         (i for i in range(idx, -1, -1) if
             (peak_gain - gains[i]) > attn),
@@ -136,6 +149,20 @@ def cut_off_left(gains: List[float], idx: int,
 
 def cut_off_right(gains: List[float], idx: int,
                   peak_gain: float, attn: float = 3.0) -> int:
+    """find first position in list where gain in attn lower then peak
+    right from index
+
+    Args:
+        gains (List[float]): gain values
+        idx (int): start position to search from
+        peak_gain (float): reference gain value
+        attn (float, optional): attenuation to search position for.
+                                Defaults to 3.0.
+
+    Returns:
+        int: position of attenuation point. (-1 if no data)
+    """
+
     return next(
         (i for i in range(idx, len(gains)) if
             (peak_gain - gains[i]) > attn),
@@ -152,12 +179,10 @@ def calculate_rolloff(s21: List[Datapoint],
                       idx_1: int, idx_2: int) -> Tuple[float, float]:
     if idx_1 == idx_2:
         return (math.nan, math.nan)
-    freq_1 = s21[idx_1].freq
-    freq_2 = s21[idx_2].freq
-    gain1 = s21[idx_1].gain
-    gain2 = s21[idx_2].gain
+    freq_1, freq_2 = s21[idx_1].freq, s21[idx_2].freq
+    gain_1, gain_2 = s21[idx_1].gain, s21[idx_2].gain
     factor = freq_1 / freq_2 if freq_1 > freq_2 else freq_2 / freq_1
-    attn = abs(gain1 - gain2)
-    octave_attn = attn / (math.log10(factor) / math.log10(2))
+    attn = abs(gain_1 - gain_2)
     decade_attn = attn / math.log10(factor)
+    octave_attn = decade_attn * math.log10(2)
     return (octave_attn, decade_attn)