Spaces:
Sleeping
Sleeping
| import numpy as np | |
| import pandas as pd | |
| from sklearn.preprocessing import StandardScaler, MinMaxScaler | |
| import os | |
| def generate_mock_csi_data(samples=1000, subcarriers=30, antennas=3): | |
| """ | |
| Génère des données CSI factices pour tester le pipeline | |
| Args: | |
| samples: Nombre d'échantillons temporels | |
| subcarriers: Nombre de sous-porteuses | |
| antennas: Nombre d'antennes | |
| Returns: | |
| DataFrame avec des données CSI simulées | |
| """ | |
| # Génération de données aléatoires complexes (phase et amplitude) | |
| csi_data = np.random.randn(samples, subcarriers, antennas) + 1j * np.random.randn(samples, subcarriers, antennas) | |
| # Création des noms de colonnes | |
| columns = [] | |
| for ant in range(1, antennas + 1): | |
| for sub in range(1, subcarriers + 1): | |
| columns.append(f"antenna_{ant}_subcarrier_{sub}_real") | |
| columns.append(f"antenna_{ant}_subcarrier_{sub}_imag") | |
| # Conversion en format réel (concaténer partie réelle et imaginaire) | |
| real_data = np.concatenate([np.real(csi_data), np.imag(csi_data)], axis=2) | |
| real_data = real_data.reshape(samples, -1) | |
| # Création du DataFrame | |
| df = pd.DataFrame(real_data, columns=columns) | |
| # Ajout d'une colonne d'activité simulée | |
| activities = ['walking', 'standing', 'sitting', 'falling'] | |
| df['activity'] = np.random.choice(activities, size=samples) | |
| # Ajout d'un timestamp simulé | |
| df['timestamp'] = pd.date_range(start='2023-01-01', periods=samples, freq='100ms') | |
| return df | |
| def normalize_data(df, normalization_type='standard'): | |
| """ | |
| Normalise les données CSI (sauf les colonnes d'activité et timestamp) | |
| Args: | |
| df: DataFrame pandas avec les données CSI | |
| normalization_type: 'standard' (StandardScaler) ou 'minmax' (MinMaxScaler) | |
| Returns: | |
| DataFrame normalisé | |
| """ | |
| # Colonnes à ne pas normaliser | |
| non_feature_cols = ['activity', 'timestamp'] | |
| feature_cols = [col for col in df.columns if col not in non_feature_cols] | |
| if normalization_type == 'standard': | |
| scaler = StandardScaler() | |
| elif normalization_type == 'minmax': | |
| scaler = MinMaxScaler() | |
| else: | |
| raise ValueError(f"Type de normalisation inconnu: {normalization_type}") | |
| # Application de la normalisation | |
| df[feature_cols] = scaler.fit_transform(df[feature_cols]) | |
| return df | |
| def load_and_preprocess_dataset(base_path='mock_data', normalization_type='standard'): | |
| """ | |
| Fonction principale qui charge (ou génère) et prétraite les données | |
| Args: | |
| base_path: Chemin vers les données (ignoré pour les données factices) | |
| normalization_type: Type de normalisation | |
| Returns: | |
| DataFrame prétraité | |
| """ | |
| # Ici on génère des données factices au lieu de charger depuis un fichier | |
| print(f"Génération de données factices (base_path '{base_path}' ignoré)") | |
| df = generate_mock_csi_data(samples=500) # Réduit à 500 échantillons pour les tests | |
| # Normalisation | |
| df = normalize_data(df, normalization_type) | |
| return df | |
| # Test du module | |
| if __name__ == "__main__": | |
| print("Génération d'un exemple de données...") | |
| test_df = load_and_preprocess_dataset() | |
| print("Données générées :") | |
| print(test_df.head()) | |
| print("\nStatistiques descriptives :") | |
| print(test_df.describe()) |