Refactoring to individual classes

Author: Ramy-Badr-Ahmed

examples/run_example.py

@@ -1 +1,7 @@
-from .signal_processor import SignalProcessor
+from .signal_generator import SignalGenerator
+from .signal_filter import SignalFilter
+from .signal_fitter import SignalFitter
+from .signal_visualizer import SignalVisualizer
+from .statistical_analyzer import StatisticalAnalyzer
+
+__version__ = '1.6.0'

src/__init__.py

@@ -1,19 +1,31 @@
-from signal_processor import SignalProcessor
+from src.signal_generator import SignalGenerator
+from src.signal_visualizer import SignalVisualizer
 import numpy as np
 
 def runProcessing():
     """
-    Run the signal processing steps using the default parameters of the SignalProcessor class.
+    Run the signal processing steps using the default parameters of the SignalProcessor classes.
     """
     timeVector = np.linspace(0, 1, 1000, endpoint = False)   # Consider importing or modifying your time vector
-    processor = SignalProcessor(timeVector)
-    processor.generateNoisySignal()
-    processor.applyFilter()
-    processor.fitDampedSineWave()
-    processor.performTTest()
-    processor.plotResults()
-    processor.plotInteractiveResults()
-    processor.printResults()
+
+    noisyInstance = SignalGenerator(timeVector)
+
+    filteredInstance = noisyInstance.generateNoisySignal().applyFilter()
+    fittedInstance = filteredInstance.fitDampedSineWave()
+    analyzedInstance = fittedInstance.analyzeFit()
+
+    # Retrieve results
+    noisySignal = noisyInstance.getNoisySignal()
+    filteredSignal = filteredInstance.getFilteredSignal()
+    fittedSignal = fittedInstance.getFittedSignal()
+    tTestResults = analyzedInstance.getTTestResults()
+
+    print(f"T-test result: statistic={tTestResults[0]}, p-value={tTestResults[1]}")
+
+    visualizer = SignalVisualizer(timeVector, noisySignal, filteredSignal, fittedSignal)
+
+    visualizer.plotResults()
+    visualizer.plotInteractiveResults()
 
 if __name__ == "__main__":
     try:

src/main.py

@@ -1,46 +1,78 @@
+from src.signal_fitter import SignalFitter
 from scipy import signal
 from typing import Optional
 import numpy as np
 
 class SignalFilter:
-    def __init__(self, noisySignal: np.ndarray, sampleRate: float = 1.0):
-        self.sampleRate = sampleRate
+    def __init__(self, timeVector: np.ndarray, noisySignal: np.ndarray):
+        """
+        Initialize with a noisy signal to filter.
+        Defaults:
+        filterOrder: Order of the Butterworth filter (default is 4).
+        cutoffFrequency: Cutoff frequency for the low-pass filter (default is 0.2).
+        filterType: Type of filter ('butter', 'bessel', 'highpass'). Default is 'butter'.
+        :param timeVector: The time vector associated with the signal.
+        :param noisySignal: The noisy signal to be filtered.
+        """
+        self.timeVector = timeVector
         self.noisySignal = noisySignal
         self.filteredSignal: Optional[np.ndarray] = None
 
+        # Default parameters
+        self.filterType: str = 'butter'
+        self.filterOrder: int = 4
+        self.cutOffFrequency: float = 0.2
+        self.bType: str = 'low'
+
         self.filterTypes = {
             "butter": lambda order, cutoff, btype: signal.butter(order, cutoff, btype, analog = False),
-            "chebyshev1": lambda order, cutoff, btype: signal.cheby1(order, 0.5, cutoff, btype, analog = False),
-            "chebyshev2": lambda order, cutoff, btype: signal.cheby2(order, 20, cutoff, btype, analog = False),
-            "elliptic": lambda order, cutoff , btype: signal.ellip(order, 0.5, 20, cutoff,  analog = False),
             "bessel": lambda order, cutoff, btype: signal.bessel(order, cutoff, btype, analog = False),
-            "notch": lambda notchFreq, Q: signal.iirnotch(notchFreq, Q, fs = self.sampleRate),
-            "highpass": lambda order, cutoff, btype: signal.butter(order, cutoff, btype, analog = False),
-            "bandpass": lambda low_cutoff, high_cutoff, btype: signal.butter(4, [low_cutoff, high_cutoff], btype, analog = False),
-            "bandstop": lambda low_cutoff, high_cutoff, btype: signal.butter(4, [low_cutoff, high_cutoff], btype, analog = False)
+            "highpass": lambda order, cutoff, btype: signal.butter(order, cutoff, btype, analog = False)
         }
 
-    def apply_filter(self, filterType: str = 'butter', **kwargs) -> np.ndarray:
+        self.applyFilter(order = self.filterOrder, cutoff = self.cutOffFrequency, btype = self.bType)
+
+    def setFilterParameters(self, filterType: str, filterOrder: int, cutOffFrequency: float, bType: str) -> None:
         """
-        Apply the specified filter type.
-        :param filterType: Type of filter ('butter', 'chebyshev1', 'chebyshev2', 'elliptic', 'bessel', 'notch', 'highpass', 'bandpass', 'bandstop').
-        :param kwargs: Additional parameters for specific filters.
-        :return: Filtered signal as a numpy array.
+        Set or change filter parameters.
+        :param filterType: Type of filter ('butter', 'bessel', 'highpass').
+        :param filterOrder: Order of the filter.
+        :param cutOffFrequency: Cutoff frequency for the filter.
+        :param bType: Filter band type ('lowpass', 'highpass', etc.).
         """
-        if filterType not in self.filterTypes:
-            raise ValueError(f"Filter type '{filterType}' is not recognized.")
+        self.filterType = filterType
+        self.throwIfNotSupported()
+        self.filterOrder = filterOrder
+        self.cutOffFrequency = cutOffFrequency
+        self.bType = bType
 
-        if self.noisySignal is None:
-            raise ValueError("Noisy signal is not generated. Please call 'generateNoisySignal' first.")
-
-        # Design Butterworth low-pass filter
-        [filterCoefficientsB, filterCoefficientsA] = self.filterTypes[filterType](**kwargs)
+    def throwIfNotSupported(self):
+        """
+        Raise an error if the filter type is not supported.
+        """
+        if self.filterType not in self.filterTypes:
+            raise ValueError(f"Filter type '{self.filterType}' is not recognized.")
 
+    def applyFilter(self, **kwargs) -> 'SignalFilter':
+        """
+        Apply the configured filter to the noisy signal.
+        :param kwargs: Additional parameters for specific filters. Given as Dict with keys: Order, cutoff, btype
+        """
+        [filterCoefficientsB, filterCoefficientsA] = self.filterTypes[self.filterType](**kwargs)
         self.filteredSignal = signal.filtfilt(filterCoefficientsB, filterCoefficientsA, self.noisySignal)
-
         print("Filter applied.")
+        return self
 
-        return self.filteredSignal
+    def fitDampedSineWave(self) -> SignalFitter:
+        """
+        Create a SignalFitter instance to fit a damped sine wave to the filtered signal.
+        :return: A SignalFitter instance.
+        """
+        return SignalFitter(self.timeVector, self.filteredSignal)
 
     def getFilteredSignal(self) -> Optional[np.ndarray]:
+        """
+        Retrieve the filtered signal.
+        :return: The filtered signal or None if filtering hasn't been performed.
+        """
         return self.filteredSignal

src/signal_filter.py

@@ -1,11 +1,18 @@
+from src.statistical_analyzer import StatisticalAnalyzer
 from scipy import optimize
 import numpy as np
 from typing import Optional
 
 class SignalFitter:
     def __init__(self, timeVector: np.ndarray, filteredSignal: np.ndarray):
+        """
+        Initialize the SignalFitter with a time vector and a filtered signal.
+        :param timeVector: The time vector associated with the signal.
+        :param filteredSignal: The filtered signal to fit.
+        """
         self.timeVector = timeVector
         self.filteredSignal = filteredSignal
+
         self.optimalParamsDamped: Optional[np.ndarray] = None
         self.fittedSignalDamped: Optional[np.ndarray] = None
 
@@ -18,6 +25,8 @@ def __init__(self, timeVector: np.ndarray, filteredSignal: np.ndarray):
         # Default bounds
         self.bounds: tuple = ([0, 0, -np.pi, 0], [np.inf, np.inf, np.pi, np.inf])
 
+        self.fitDampedSineWave()
+
     def setDampedSineWaveParameters(self, amplitudeParam: float, frequencyParam: float, phaseParam: float, decayRateParam: float) -> None:
         """
         Set the parameters for the damped sine wave fitting.
@@ -42,34 +51,56 @@ def setDampedSineWaveBounds(self, lower: list[float], upper: list[float]) -> Non
             raise ValueError("Bounds should be lists of length 4.")
         self.bounds = (lower, upper)
 
-    def fitDampedSineWave(self) -> np.ndarray:
+    def fitDampedSineWave(self) -> 'SignalFitter':
         """
         Fit a damped sine wave to the filtered signal using nonlinear least squares.
         default sine wave parameters: amplitudeParam = 1.0, frequencyParam = 10.0, phaseParam = 0.0, decayRateParam = 0.1
         change with setDampedSineWaveParameters().
         """
+        if self.filteredSignal is None or len(self.filteredSignal) == 0:
+            raise ValueError("Filtered signal is None or empty. Cannot fit a damped sine wave.")
+
         def dampedSineFunc(time: np.ndarray, amplitude: float, frequency: float, phase: float, decayRate: float) -> np.ndarray:
+            """Define the damped sine wave function."""
             return amplitude * np.exp(-decayRate * time) * np.sin(2 * np.pi * frequency * time + phase)
 
         def residualsDamped(params: np.ndarray, time: np.ndarray, data: np.ndarray) -> np.ndarray:
+            """Calculate residuals for the damped sine wave fitting."""
             return dampedSineFunc(time, *params) - data
 
         initialParams = np.array([self.amplitudeParam, self.frequencyParam, self.phaseParam, self.decayRateParam])
 
         try:
             result = optimize.least_squares(residualsDamped, initialParams, bounds = self.bounds, args = (self.timeVector, self.filteredSignal))
             self.optimalParamsDamped = result.x
+
         except Exception as e:
             raise RuntimeError(f"Optimization failed: {e}")
 
         self.fittedSignalDamped = dampedSineFunc(self.timeVector, *self.optimalParamsDamped)
-
         print("Damped sine wave fitted.")
-
-        return self.fittedSignalDamped
+        return self
 
     def getFittedSignal(self) -> Optional[np.ndarray]:
+        """
+        Retrieve the fitted damped sine wave signal.
+        :return: The fitted signal or None if fitting hasn't been performed.
+        """
         return self.fittedSignalDamped
 
     def getOptimalParams(self) -> Optional[np.ndarray]:
+        """
+        Retrieve the optimal parameters obtained from fitting a damped sine wave to the signal.
+        returns Parameters: Amplitude (A), Frequency (f), Phase (φ), Decay rate (λ)
+        :return: A numpy array containing the optimal parameters or None if fitting hasn't been performed.
+        """
         return self.optimalParamsDamped
+
+    def analyzeFit(self):
+        """
+        Create a StatisticalAnalyzer instance to analyze the fitted signal against the filtered signal.
+        :return: A StatisticalAnalyzer instance.
+        """
+        if self.fittedSignalDamped is None:
+            raise ValueError("Fitted signal is not available. Perform fitting first.")
+        return StatisticalAnalyzer(self.filteredSignal, self.fittedSignalDamped)

src/signal_fitter.py

@@ -1,22 +1,48 @@
+from src.signal_filter import SignalFilter
 from typing import Optional
 import numpy as np
 
 class SignalGenerator:
     def __init__(self, timeVector: np.ndarray):
+        """
+        Initialize the SignalGenerator with a time vector.
+        """
         self.timeVector = timeVector
         self.noisySignal: Optional[np.ndarray] = None
 
-    def generateNoisySignal(self, frequency: float = 10, noiseStdDev: float = 0.5) -> np.ndarray:
+    def generateNoisySignal(self, frequency: float = 10, noiseStdDev: float = 0.5) -> 'SignalGenerator':
         """
         Generate a noisy signal using the defined time vector.
         :param frequency: Frequency of the sine wave (default is 10 Hz).
         :param noiseStdDev: Standard deviation of the noise (default is 0.5).
         """
+        if self.timeVector is None or len(self.timeVector) == 0:
+            raise ValueError("Time array 'timeVector' is not properly initialised.")
+
         self.noisySignal = np.sin(2 * np.pi * frequency * self.timeVector) + noiseStdDev * np.random.randn(len(self.timeVector))
-        return self.noisySignal
+        return self
 
-    def importNoisySignal(self, signal: np.ndarray) -> None:
+    def importNoisySignal(self, signal: np.ndarray) -> 'SignalGenerator':
+        """
+        Import an externally generated noisy signal.
+        :param signal: The noisy signal to import.
+        """
         self.noisySignal = signal
+        return self
 
-    def getSignal(self) -> Optional[np.ndarray]:
+    def getNoisySignal(self) -> Optional[np.ndarray]:
+        """
+        Retrieve the current noisy signal.
+        :return: The noisy signal or None if it hasn't been set.
+        """
         return self.noisySignal
+
+    def applyFilter(self) -> SignalFilter:
+        """
+        Create a SignalFilter instance to filter the noisy signal.
+        :return: A SignalFilter instance.
+        """
+        if self.noisySignal is None:
+            raise ValueError("Noisy signal has not been generated or imported.")
+        return SignalFilter(self.timeVector, self.noisySignal)
+