IDKHowToCodeFr/tinyml-backend

1import pandas as pd2import numpy as np3import os4import joblib5from sklearn.model_selection import train_test_split6from sklearn.preprocessing import StandardScaler, LabelEncoder7from sklearn.impute import SimpleImputer8 9def resolve_model_dir():10    return os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'model')11 12def preprocess_data(df, is_training=True):13    model_dir = resolve_model_dir()14    scaler_path = f'{model_dir}/scaler.pkl'15    label_encoder_path = f'{model_dir}/label_encoder.pkl'16    imputer_path = f'{model_dir}/imputer.pkl'17        18    columns_to_drop = ['Patient Number', 'Data Accuracy (%)', 'Heart Rate Alert', 'SpO2 Level Alert', 'Blood Pressure Alert', 'Temperature Alert', 'Fall Detection']19    cols_drop = [c for c in columns_to_drop if c in df.columns]20    X_raw = df.drop(columns=cols_drop)21    22    y = None23    if 'Predicted Disease' in X_raw.columns:24        y_raw = X_raw['Predicted Disease']25        X_raw = X_raw.drop(columns=['Predicted Disease'])26    27        if is_training:28            label_encoder = LabelEncoder()29            y = label_encoder.fit_transform(y_raw)30            os.makedirs(os.path.dirname(label_encoder_path), exist_ok=True)31            joblib.dump(label_encoder, label_encoder_path)32            joblib.dump(label_encoder.classes_, f'{model_dir}/classes.pkl')33        else:34            if os.path.exists(label_encoder_path):35                label_encoder = joblib.load(label_encoder_path)36                y = label_encoder.transform(y_raw)37            else:38                y = y_raw 39    40    # Fix broken encoding column names safely41    X_raw.rename(columns=lambda x: x.replace('\ufffd', '°') if isinstance(x, str) else x, inplace=True)42    43    if 'Fall Detection' in X_raw.columns:44        X_raw['Fall Detection'] = X_raw['Fall Detection'].map({'Yes': 1, 'No': 0}).fillna(0)45        46    # Engineered Feature47    X_raw['Risk_Severity'] = (X_raw['Heart Rate (bpm)'] > 105).astype(int) + (X_raw['SpO2 Level (%)'] < 94).astype(int)48    49    continuous_features = ['Heart Rate (bpm)', 'SpO2 Level (%)', 'Systolic Blood Pressure (mmHg)', 'Diastolic Blood Pressure (mmHg)', 'Body Temperature (°C)', 'Risk_Severity']50    51    if is_training:52        imputer = SimpleImputer(strategy='mean')53        X_raw[continuous_features] = imputer.fit_transform(X_raw[continuous_features])54        joblib.dump(imputer, imputer_path)55        56        scaler = StandardScaler()57        X_raw[continuous_features] = scaler.fit_transform(X_raw[continuous_features])58        joblib.dump(scaler, scaler_path)59    else:60        if os.path.exists(imputer_path):61            imputer = joblib.load(imputer_path)62            X_raw[continuous_features] = imputer.transform(X_raw[continuous_features])63            64        if os.path.exists(scaler_path):65            scaler = joblib.load(scaler_path)66            X_raw[continuous_features] = scaler.transform(X_raw[continuous_features])67            68    return X_raw, y69 70def get_train_test_split(df):71    X, y = preprocess_data(df, is_training=True)72    return train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)73