kurzti88066/Fahrsimulator_MSY2526_AI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

aufraeumen branch

Browse Source
This commit is contained in:
weigmi87303
2026-01-27 19:10:24 +01:00
parent eee173dc0b
commit 9951d8b4f9
4 changed files with 748 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
model_training/tools/scaler.py
+96 -18
View File
@@ -1,5 +1,7 @@
from sklearn.preprocessing import MinMaxScaler, StandardScaler
import pandas as pd
import pickle
from sklearn.preprocessing import StandardScaler, MinMaxScaler
import numpy as np
import os
def fit_normalizer(train_data, au_columns, method='standard', scope='global'):
"""
@@ -19,9 +21,8 @@ def fit_normalizer(train_data, au_columns, method='standard', scope='global'):
Returns:
--------
dict
Dictionary containing fitted scalers
Dictionary containing fitted scalers and statistics for new subjects
"""
# Select scaler based on method
if method == 'standard':
Scaler = StandardScaler
elif method == 'minmax':
@@ -30,19 +31,54 @@ def fit_normalizer(train_data, au_columns, method='standard', scope='global'):
raise ValueError("method must be 'standard' or 'minmax'")
scalers = {}
if scope == 'subject':
# Fit one scaler per subject
subject_stats = []
for subject in train_data['subjectID'].unique():
subject_mask = train_data['subjectID'] == subject
scaler = Scaler()
scaler.fit(train_data.loc[subject_mask, au_columns])
scaler.fit(train_data.loc[subject_mask, au_columns].values)
scalers[subject] = scaler
# Store statistics for averaging
if method == 'standard':
subject_stats.append({
'mean': scaler.mean_,
'std': scaler.scale_
})
elif method == 'minmax':
subject_stats.append({
'min': scaler.data_min_,
'max': scaler.data_max_
})
# Calculate average statistics for new subjects
if method == 'standard':
avg_mean = np.mean([s['mean'] for s in subject_stats], axis=0)
avg_std = np.mean([s['std'] for s in subject_stats], axis=0)
fallback_scaler = StandardScaler()
fallback_scaler.mean_ = avg_mean
fallback_scaler.scale_ = avg_std
fallback_scaler.var_ = avg_std ** 2
fallback_scaler.n_features_in_ = len(au_columns)
elif method == 'minmax':
avg_min = np.mean([s['min'] for s in subject_stats], axis=0)
avg_max = np.mean([s['max'] for s in subject_stats], axis=0)
fallback_scaler = MinMaxScaler()
fallback_scaler.data_min_ = avg_min
fallback_scaler.data_max_ = avg_max
fallback_scaler.data_range_ = avg_max - avg_min
fallback_scaler.scale_ = 1.0 / fallback_scaler.data_range_
fallback_scaler.min_ = -avg_min * fallback_scaler.scale_
fallback_scaler.n_features_in_ = len(au_columns)
scalers['_fallback'] = fallback_scaler
elif scope == 'global':
# Fit one scaler for all subjects
scaler = Scaler()
scaler.fit(train_data[au_columns])
scaler.fit(train_data[au_columns].values)
scalers['global'] = scaler
else:
@@ -50,7 +86,7 @@ def fit_normalizer(train_data, au_columns, method='standard', scope='global'):
return {'scalers': scalers, 'method': method, 'scope': scope}
def apply_normalizer(data, au_columns, normalizer_dict):
def apply_normalizer(data, columns, normalizer_dict):
"""
Apply fitted normalization scalers to data.
@@ -71,28 +107,70 @@ def apply_normalizer(data, au_columns, normalizer_dict):
normalized_data = data.copy()
scalers = normalizer_dict['scalers']
scope = normalizer_dict['scope']
normalized_data[columns] = normalized_data[columns].astype(np.float64)
if scope == 'subject':
# Apply per-subject normalization
for subject in data['subjectID'].unique():
subject_mask = data['subjectID'] == subject
# Use the subject's scaler if available, otherwise use a fitted scaler from training
# Use the subject's scaler if available, otherwise use fallback
if subject in scalers:
scaler = scalers[subject]
else:
# For new subjects not seen in training, use the first available scaler
# (This is a fallback - ideally all test subjects should be in training for subject-level normalization)
print(f"Warning: Subject {subject} not found in training data. Using fallback scaler.")
scaler = list(scalers.values())[0]
# Use averaged scaler for new subjects
scaler = scalers['_fallback']
print(f"Info: Subject {subject} not in training data. Using averaged scaler from training subjects.")
normalized_data.loc[subject_mask, au_columns] = scaler.transform(
data.loc[subject_mask, au_columns]
normalized_data.loc[subject_mask, columns] = scaler.transform(
data.loc[subject_mask, columns].values
)
elif scope == 'global':
# Apply global normalization
scaler = scalers['global']
normalized_data[au_columns] = scaler.transform(data[au_columns])
normalized_data[columns] = scaler.transform(data[columns].values)
return normalized_data
return normalized_data
def save_normalizer(normalizer_dict, filepath):
"""
Save fitted normalizer to disk.
Parameters:
-----------
normalizer_dict : dict
Dictionary containing fitted scalers from fit_normalizer()
filepath : str
Path to save the normalizer (e.g., 'normalizer.pkl')
"""
# Create directory if it does not exist
dirpath = os.path.dirname(filepath)
if dirpath:
os.makedirs(dirpath, exist_ok=True)
with open(filepath, 'wb') as f:
pickle.dump(normalizer_dict, f)
print(f"Normalizer saved to {filepath}")
def load_normalizer(filepath):
"""
Load fitted normalizer from disk.
Parameters:
-----------
filepath : str
Path to the saved normalizer file
Returns:
--------
dict
Dictionary containing fitted scalers
"""
with open(filepath, 'rb') as f:
normalizer_dict = pickle.load(f)
print(f"Normalizer loaded from {filepath}")
return normalizer_dict