zebris_extractor.py

import io
import numpy as np
import pandas as pd


STANDARD_COLUMNS = [
    "Nom",
    "Date",
    "Poids (kg)",
    "Vitesse (km/h)",
    "Cadence (pas/min)",
    "Contact (%)",
    "Flight (%)",
    "Force talon G (N)",
    "Force talon D (N)",
    "Force avant-pied G (N)",
    "Force avant-pied D (N)",
    "Pression talon G (N/cm²)",
    "Pression talon D (N/cm²)",
    "COP G (mm)",
    "COP D (mm)",
    "Rotation G (°)",
    "Rotation D (°)",
    "Transition G (s)",
    "Transition D (s)",
    "Longueur foulée (cm)",
    "Largeur pas (cm)",
]


def _to_numeric(series):
    return pd.to_numeric(
        series.astype(str).str.replace(",", ".", regex=False),
        errors="coerce"
    )


def _read_csv_flex(uploaded_file):
    raw = uploaded_file.read()
    uploaded_file.seek(0)

    for encoding in ["utf-8-sig", "utf-8", "latin1", "cp1252"]:
        for sep in [",", ";", "\t"]:
            try:
                txt = raw.decode(encoding)
                df = pd.read_csv(io.StringIO(txt), sep=sep)
                if df.shape[1] > 1:
                    return df
            except Exception:
                pass

    return pd.read_csv(uploaded_file)


def _pick(df, candidates):
    for c in candidates:
        if c in df.columns:
            return c
    return None


def extract_zebris_csv(uploaded_file):
    df = _read_csv_flex(uploaded_file)

    out = pd.DataFrame(index=df.index)
    mapping = {}
    manquantes = []

    # Nom
    first_name_col = _pick(df, ["Prénom", "First Name"])
    last_name_col = _pick(df, ["Nom de famille", "Last Name"])
    if first_name_col and last_name_col:
        out["Nom"] = (
            df[first_name_col].astype(str).str.strip() + " " +
            df[last_name_col].astype(str).str.strip()
        )
        mapping["Nom"] = [first_name_col, last_name_col]
    else:
        out["Nom"] = "Inconnu"
        manquantes.append("Nom")

    # Mapping direct depuis ton CSV Zebris
    direct_map = {
        "Date": ["Measurement date", "Date"],
        "Poids (kg)": ["Body weight [Kg]", "Weight (kg)", "Poids (kg)"],
        "Vitesse (km/h)": ["Vitesse [km/h]", "Speed [km/h]", "Speed (km/h)"],
        "Cadence (pas/min)": ["Cadence [pass/min]", "Cadence [pas/min]", "Cadence"],
        "Contact (%)": ["Total contact [%]", "Contact [%]"],
        "Flight (%)": ["Total flight [%]", "Flight [%]"],
        "Force talon G (N)": ["Force maximale Heel (Three zones) Gauche [N]"],
        "Force talon D (N)": ["Force maximale Heel (Three zones) Droite [N]"],
        "Force avant-pied G (N)": ["Force maximale Forefoot (Three zones) Gauche [N]"],
        "Force avant-pied D (N)": ["Force maximale Forefoot (Three zones) Droite [N]"],
        "Pression talon G (N/cm²)": ["Pression maximale Heel (Three zones) Gauche [N/cm²]", "Pression maximale Heel (Three zones) Gauche [N/cm2]"],
        "Pression talon D (N/cm²)": ["Pression maximale Heel (Three zones) Droite [N/cm²]", "Pression maximale Heel (Three zones) Droite [N/cm2]"],
        "COP G (mm)": ["Longueur lors de la phase d'appui Gauche [mm]"],
        "COP D (mm)": ["Longueur lors de la phase d'appui Droite [mm]"],
        "Rotation G (°)": ["Rotation du pied Gauche [degré]"],
        "Rotation D (°)": ["Rotation du pied Droite [degré]"],
        "Transition G (s)": ["Instant du passage du talon vers l'avant-pied Gauche [s]"],
        "Transition D (s)": ["Instant du passage du talon vers l'avant-pied Droite [s]"],
        "Longueur foulée (cm)": ["Longueur de la foulée [cm]"],
        "Largeur pas (cm)": ["Largeur du pas [cm]"],
    }

    for target, candidates in direct_map.items():
        col = _pick(df, candidates)
        if col is None:
            out[target] = np.nan
            manquantes.append(target)
        else:
            mapping[target] = col
            if target == "Date":
                out[target] = df[col]
            else:
                out[target] = _to_numeric(df[col])

    # Garde seulement les lignes avec une vitesse
    out = out[out["Vitesse (km/h)"].notna()].copy()

    # Réordonne
    out = out.reindex(columns=STANDARD_COLUMNS).reset_index(drop=True)

    debug = {
        "mapping": mapping,
        "manquantes": manquantes,
        "colonnes_csv": list(df.columns),
        "nb_lignes_csv": len(df),
        "nb_lignes_extractees": len(out),
    }

    return out, debug