Natuurlijk! Hier is een voorbeeld van een modulaire, efficiënte en onderhoudsvriendelijke datatransformatie-pipeline in Python met NumPy, die de beschreven preprocessingstappen uitvoert: ```python import numpy as np class DataPreprocessor: def __init__(self): self.min = None self.max = None def remove_nan_rows(self, data): """ Verwijdert rijen met NaN-waarden uit de dataset. """ return data[~np.isnan(data).any(axis=1)] def fit(self, data): """ Berekent de min en max voor normalisatie op basis van de data zonder NaN-rijen. """ self.min = np.min(data, axis=0) self.max = np.max(data, axis=0) # Om division door 0 te voorkomen, vervang gelijke min en max door 1 self.range = self.max - self.min self.range[self.range == 0] = 1 def transform(self, data): """ Normaliseert de data naar het bereik 0-1. """ return (data - self.min) / self.range def fit_transform(self, data): """ Voert zowel fit als transform uit. """ data_clean = self.remove_nan_rows(data) self.fit(data_clean) return self.transform(data_clean) # Voorbeeld gebruik: # Stel dat 'sensor_data' je NumPy array is met tijdreeks sensordata # sensor_data = np.array([...]) # Initializeer de pipeline preprocessor = DataPreprocessor() # Voer preprocessing uit processed_data = preprocessor.fit_transform(sensor_data) # 'processed_data' bevat nu de genormaliseerde data zonder NaN-rijen ``` **Toelichting:** - De klasse `DataPreprocessor` is modulair en onderhoudsvriendelijk ontworpen. - De methode `remove_nan_rows` verwijdert alle rijen met NaN-waarden. - De methoden `fit`, `transform` en `fit_transform` zorgen voor een gescheiden en herbruikbare normalisatie. - Bij gelijke min en max wordt de range op 1 gezet om deling door nul te voorkomen. - Deze aanpak is efficiënt doordat numpy-operaties vectorized zijn. Indien je dataset erg groot is, kun je overwegen om NumPy's masked arrays of andere technieken te gebruiken voor optimalisatie.

Hier is een modulaire en efficiënte NumPy-pipeline voor uw tijdreekssensordata: ```python import numpy as np from typing import Tuple class DataPreprocessor: def __init__(self): self.feature_min = None self.feature_max = None def verwijder_nan_rijen(self, data: np.ndarray) -> np.ndarray: """Verwijdert rijen met NaN-waarden""" if data.size == 0: return data # Behoud alleen rijen zonder NaN-waarden geldige_rijen = ~np.isnan(data).any(axis=1) return data[geldige_rijen] def fit_normalisatie(self, data: np.ndarray) -> None: """Bereken normalisatieparameters op trainingsdata""" if data.size == 0: raise ValueError("Data mag niet leeg zijn voor normalisatie") self.feature_min = np.nanmin(data, axis=0) self.feature_max = np.nanmax(data, axis=0) # Voorkom deling door nul nul_range_mask = self.feature_max == self.feature_min self.feature_max[nul_range_mask] = self.feature_min[nul_range_mask] + 1 def transform_normalisatie(self, data: np.ndarray) -> np.ndarray: """Pas normalisatie toe op data""" if self.feature_min is None or self.feature_max is None: raise ValueError("Normalisatie moet eerst gefit worden") if data.size == 0: return data # Normaliseer naar range [0, 1] genormaliseerd = (data - self.feature_min) / (self.feature_max - self.feature_min) return np.clip(genormaliseerd, 0, 1) # Zekere begrenzing def fit_transform(self, data: np.ndarray) -> np.ndarray: """Volledige preprocessing pipeline""" # Stap 1: Verwijder NaN rijen schone_data = self.verwijder_nan_rijen(data) # Stap 2: Fit en transform normalisatie self.fit_normalisatie(schone_data) return self.transform_normalisatie(schone_data) def transform(self, data: np.ndarray) -> np.ndarray: """Transformeer nieuwe data met bestaande parameters""" schone_data = self.verwijder_nan_rijen(data) return self.transform_normalisatie(schone_data) # Gebruiksvoorbeeld: if __name__ == "__main__": # Voorbeeld dataset (vervang met uw eigen data) sensor_data = np.array([ [1.0, 2.0, 3.0], [4.0, np.nan, 6.0], # Rij met NaN [7.0, 8.0, 9.0], [10.0, 11.0, 12.0] ]) # Initialiseer en voer pipeline uit preprocessor = DataPreprocessor() verwerkte_data = preprocessor.fit_transform(sensor_data) print("Originele data vorm:", sensor_data.shape) print("Verwerkte data vorm:", verwerkte_data.shape) print("Verwerkte data:\n", verwerkte_data) # Voor nieuwe data (bijvoorbeeld testset) nieuwe_data = np.array([[13.0, 14.0, 15.0], [16.0, np.nan, 18.0]]) nieuwe_data_verwerkt = preprocessor.transform(nieuwe_data) print("Nieuwe data verwerkt:\n", nieuwe_data_verwerkt) ``` **Belangrijke kenmerken van deze pipeline:** 1. **Modulair**: Elke preprocessingstap is separaat implementeerd 2. **Onderhoudsvriendelijk**: Duidelijke methoden en documentatie 3. **Efficiënt**: Gebruikt NumPy's vectorized operations 4. **Robuust**: Behandelt edge cases (lege data, deling door nul) 5. **Herbruikbaar**: Fit-transform pattern voor consistente verwerking **Aanbevelingen voor gebruik:** - Fit de preprocessor alleen op trainingsdata - Gebruik `transform()` voor testdata om data leakage te voorkomen - Sla de preprocessor op voor consistentie in productieomgevingen