Refactor modules to use centralized validators

Author: sanketsabharwal

pyeyesweb/data_models/sliding_window.py

@@ -2,6 +2,7 @@
 import threading
 import numpy as np
 from typing import Optional, Union
+from pyeyesweb.utils.validators import validate_integer
 
 
 class SlidingWindow:
@@ -59,30 +60,14 @@ def max_length(self, value: int):
             self._resize(old_max_length)
 
     def __init__(self, max_length: int, n_columns: int):
-        # Validate inputs
-        if not isinstance(max_length, int):
-            raise TypeError(f"max_length must be an integer, got {type(max_length).__name__}")
-        if not isinstance(n_columns, int):
-            raise TypeError(f"n_columns must be an integer, got {type(n_columns).__name__}")
-
-        if max_length <= 0:
-            raise ValueError(f"max_length must be positive, got {max_length}")
-        if n_columns <= 0:
-            raise ValueError(f"n_columns must be positive, got {n_columns}")
-
-        # Reasonable limits to prevent memory exhaustion
-        if max_length > 10_000_000:  # Have added 10 million samples
-            raise ValueError(f"max_length too large ({max_length}), maximum is 10,000,000")
-        if n_columns > 10_000:  # 10k features
-            raise ValueError(f"n_columns too large ({n_columns}), maximum is 10,000")
+        # Validate inputs using centralized validators
+        self._max_length = validate_integer(max_length, 'max_length', min_val=1, max_val=10_000_000)
+        self._n_columns = validate_integer(n_columns, 'n_columns', min_val=1, max_val=10_000)
 
         self._lock = threading.RLock()
 
-        self._max_length = max_length
-        self._n_columns = n_columns
-
-        self._buffer = np.empty((max_length, n_columns), dtype=np.float32)
-        self._timestamp = np.empty(max_length, dtype=np.float64)
+        self._buffer = np.empty((self._max_length, self._n_columns), dtype=np.float32)
+        self._timestamp = np.empty(self._max_length, dtype=np.float64)
 
         self._start = 0
         self._size = 0

pyeyesweb/mid_level/smoothness.py

@@ -16,6 +16,7 @@
 from pyeyesweb.data_models.sliding_window import SlidingWindow
 from pyeyesweb.utils.signal_processing import apply_savgol_filter
 from pyeyesweb.utils.math_utils import compute_sparc, compute_jerk_rms, normalize_signal
+from pyeyesweb.utils.validators import validate_numeric, validate_boolean
 
 
 class Smoothness:
@@ -64,20 +65,11 @@ class Smoothness:
     """
 
     def __init__(self, rate_hz=50.0, use_filter=True):
-        # Validate rate_hz
-        if not isinstance(rate_hz, (int, float)):
-            raise TypeError(f"rate_hz must be a number, got {type(rate_hz).__name__}")
-        if rate_hz <= 0:
-            raise ValueError(f"rate_hz must be positive, got {rate_hz}")
-        if rate_hz > 100000:  # 100 kHz is a reasonable upper limit
-            raise ValueError(f"rate_hz too high ({rate_hz} Hz), maximum is 100,000 Hz")
-
-        # Validate use_filter
-        if not isinstance(use_filter, bool):
-            raise TypeError(f"use_filter must be boolean, got {type(use_filter).__name__}")
-
-        self.rate_hz = float(rate_hz)  # Ensure it's a float
-        self.use_filter = use_filter
+        # Validate rate_hz using centralized validator
+        self.rate_hz = validate_numeric(rate_hz, 'rate_hz', min_val=0.01, max_val=100000)
+
+        # Validate use_filter using centralized validator
+        self.use_filter = validate_boolean(use_filter, 'use_filter')
 
     def _filter_signal(self, signal):
         """Apply Savitzky-Golay filter if enabled and enough data.

pyeyesweb/sync.py

@@ -34,8 +34,9 @@
 import numpy as np
 
 from pyeyesweb.data_models.sliding_window import SlidingWindow
-from pyeyesweb.utils.signal_processing import bandpass_filter, compute_hilbert_phases
+from pyeyesweb.utils.signal_processing import bandpass_filter, compute_hilbert_phases, validate_filter_params
 from pyeyesweb.utils.math_utils import compute_phase_locking_value, center_signals
+from pyeyesweb.utils.validators import validate_integer, validate_boolean, validate_numeric
 
 
 class Synchronization:
@@ -125,45 +126,31 @@ class Synchronization:
     """
 
     def __init__(self, sensitivity=100, output_phase=False, filter_params=None, phase_threshold=0.7):
-        # Validate sensitivity
-        if not isinstance(sensitivity, int):
-            raise TypeError(f"sensitivity must be an integer, got {type(sensitivity).__name__}")
-        if sensitivity <= 0:
-            raise ValueError(f"sensitivity must be positive, got {sensitivity}")
-        if sensitivity > 10000:  # Reasonable upper limit
-            raise ValueError(f"sensitivity too large ({sensitivity}), maximum is 10,000")
-
-        # Validate output_phase
-        if not isinstance(output_phase, bool):
-            raise TypeError(f"output_phase must be boolean, got {type(output_phase).__name__}")
-
-        # Validate phase_threshold
-        if not isinstance(phase_threshold, (int, float)):
-            raise TypeError(f"phase_threshold must be a number, got {type(phase_threshold).__name__}")
-        if not 0 <= phase_threshold <= 1:
-            raise ValueError(f"phase_threshold must be between 0 and 1, got {phase_threshold}")
+        # Validate sensitivity using centralized validator
+        sensitivity = validate_integer(sensitivity, 'sensitivity', min_val=1, max_val=10000)
+
+        # Validate output_phase using centralized validator
+        self.output_phase = validate_boolean(output_phase, 'output_phase')
+
+        # Validate phase_threshold using centralized validator with range check
+        self.phase_threshold = validate_numeric(phase_threshold, 'phase_threshold', min_val=0, max_val=1)
 
         # Validate filter_params if provided
         if filter_params is not None:
             if not isinstance(filter_params, (tuple, list)):
                 raise TypeError(f"filter_params must be a tuple or list, got {type(filter_params).__name__}")
             if len(filter_params) != 3:
                 raise ValueError(f"filter_params must have 3 elements (lowcut, highcut, fs), got {len(filter_params)}")
-            lowcut, highcut, fs = filter_params
             if not all(isinstance(x, (int, float)) for x in filter_params):
                 raise TypeError("filter_params must contain only numbers")
-            if lowcut <= 0 or highcut <= 0 or fs <= 0:
-                raise ValueError("filter_params frequencies must be positive")
-            if lowcut >= highcut:
-                raise ValueError(f"lowcut ({lowcut}) must be less than highcut ({highcut})")
-            if highcut >= fs / 2:
-                raise ValueError(f"highcut ({highcut}) must be less than Nyquist frequency ({fs/2})")
+            # Use centralized filter validation
+            lowcut, highcut, fs = validate_filter_params(*filter_params)
+            # Store the validated parameters
+            filter_params = (lowcut, highcut, fs)
 
         self.plv_history = deque(maxlen=sensitivity)
         self._history_lock = threading.Lock()
-        self.output_phase = output_phase
         self.filter_params = filter_params
-        self.phase_threshold = phase_threshold
 
 
     def compute_synchronization(self, signals: SlidingWindow):