Normalizer (`kadi.kidas.normalizer`)¶

Normalizer est la classe de normalisation des données agricoles de KadiPy, orientée vers le contexte béninois. Elle regroupe six transformations : normalisation des noms de colonnes en snake_case, conversion d'unités de mesure vers le kilogramme, conversion de devises, standardisation des noms de cultures vers les codes FAO, géocodage des noms de marchés, et création de géométries GPS shapely.

Chaque méthode met à jour le DataFrame interne et le retourne, ce qui permet d'enchaîner les appels en une seule expression. L'historique des transformations appliquées est accessible via mappings().

Initialisation¶

from kadi.kidas import Normalizer

normalizer = Normalizer(df)

Normalizer reçoit un pandas.DataFrame et en crée une copie interne. Le DataFrame original n'est jamais modifié.

Exception : CleanError si l'argument fourni n'est pas un DataFrame.

Méthodes de normalisation¶

`norm_cols(style='snake_case')`¶

Normalise les noms de colonnes du DataFrame en snake_case : suppression des accents, remplacement des espaces, tirets et parenthèses par des underscores, conversion en minuscules.

# Avant : ['Culture', 'Rendement (kg)', 'Température Min (°C)', 'Date Récolte']
df_norm = normalizer.norm_cols()
# Après : ['culture', 'rendement_kg', 'temperature_min_c', 'date_recolte']

Paramètre	Type	Défaut	Description
`style`	`str`	`'snake_case'`	Style cible. Seul `'snake_case'` est supporté actuellement.

Un style différent de 'snake_case' génère un avertissement dans les logs et l'opération est quand même appliquée en snake_case.

Retour : pd.DataFrame

`convert_units(unit_map)`¶

Convertit les valeurs numériques des colonnes spécifiées vers le kilogramme, en appliquant les facteurs de conversion du référentiel interne.

df_norm = normalizer.convert_units({
    "production":  "tonne",
    "recolte":     "sac_100kg",
    "stock_local": "tiya",
})

Paramètre	Type	Description
`unit_map`	`dict[str, str]`	Dictionnaire `nom_colonne` → `unité source`

Unités supportées :

Unité	Facteur vers kg
`'kg'`, `'kilogramme'`	1.0
`'tonne'`, `'tonnes'`, `'t'`	1000.0
`'sac_100kg'`, `'sac'`	100.0
`'sac_80kg'`	80.0
`'sac_50kg'`	50.0
`'tiya'`	1.5
`'quintal'`	100.0
`'boisseau'`	27.2
`'livre'`	0.4536
`'g'`, `'gramme'`	0.001

Le 'sac' standard correspond au sac de 100 kg utilisé au Bénin. Le 'tiya' est une mesure locale d'environ 1.5 kg selon la culture. Une colonne absente du DataFrame génère un avertissement dans les logs et est ignorée sans erreur.

Retour : pd.DataFrame

Exception : CleanError si une unité source est inconnue.

`convert_currency(col, from_='XOF', to='XOF', date=None)`¶

Convertit les valeurs monétaires d'une colonne selon un taux de change fixe.

# Conversion de USD vers XOF
df_norm = normalizer.convert_currency(
    col="prix_usd",
    from_="USD",
    to="XOF",
)

# Conversion d'EUR vers XOF
df_norm = normalizer.convert_currency(
    col="valeur_export",
    from_="EUR",
    to="XOF",
)

Paramètre	Type	Défaut	Description
`col`	`str`	-	Nom de la colonne à convertir
`from_`	`str`	`'XOF'`	Devise source
`to`	`str`	`'XOF'`	Devise cible
`date`	`str` ou `None`	`None`	Date de référence au format `'YYYY-MM-DD'` (non utilisée dans la version actuelle)

Taux de change disponibles (fixes, vers XOF) :

Devise	Taux vers XOF
`'XOF'`	1.0
`'EUR'`	655.957 (taux fixe UMOA)
`'USD'`	600.0
`'GBP'`	750.0

Si l'une des devises n'est pas dans ce tableau, un avertissement est émis et aucune conversion n'est appliquée.

La version actuelle utilise des taux fixes. L'intégration avec une API de change en temps réel est prévue en phase 2 du module.

Retour : pd.DataFrame

`std_crops(col, std='fao')`¶

Normalise les noms de cultures d'une colonne vers les codes FAO officiels, en gérant les variantes locales béninoises avec ou sans accents.

# Avant : ['maïs', 'Niébé', 'MANIOC', 'yam', 'mais']
df_norm = normalizer.std_crops(col="culture")
# Après : ['maize', 'cowpea', 'cassava', 'yam', 'maize']

Paramètre	Type	Défaut	Description
`col`	`str`	-	Nom de la colonne contenant les noms de cultures
`std`	`str`	`'fao'`	Standard cible (`'fao'` utilise les codes officiels FAO)

Correspondances disponibles :

Noms locaux acceptés	Code FAO
`'maïs'`, `'mais'`, `'maiz'`, `'corn'`, `'maize'`	`'maize'`
`'niébé'`, `'niebe'`, `'cowpea'`, `'haricot'`	`'cowpea'`
`'igname'`, `'yam'`	`'yam'`
`'sorgho'`, `'sorghum'`	`'sorghum'`
`'riz'`, `'rice'`	`'rice'`
`'manioc'`, `'cassava'`	`'cassava'`
`'arachide'`, `'groundnut'`, `'peanut'`	`'groundnut'`
`'mil'`, `'millet'`	`'millet'`
`'fonio'`	`'fonio'`
`'soja'`, `'soybean'`	`'soybean'`
`'tomate'`, `'tomato'`	`'tomato'`
`'oignon'`, `'onion'`	`'onion'`
`'piment'`, `'pepper'`	`'pepper'`

Les noms non reconnus sont conservés en minuscules sans accents. Une colonne absente génère un avertissement et est ignorée.

Retour : pd.DataFrame

`std_markets(col, region='benin')`¶

Normalise les noms de marchés d'une colonne et ajoute deux colonnes market_lat et market_lon avec les coordonnées GPS officielles.

# Avant : ['Dantokpa', 'PARAKOU', 'Bohicon']
df_norm = normalizer.std_markets(col="marche")
# Après : colonne 'marche' inchangée, + colonnes 'market_lat' et 'market_lon'

Paramètre	Type	Défaut	Description
`col`	`str`	-	Nom de la colonne contenant les noms de marchés
`region`	`str`	`'benin'`	Région de référence pour le dictionnaire de marchés

Marchés référencés (Bénin) :

Marché	Latitude	Longitude	Ville
`dantokpa`, `cotonou`	6.366	2.437	Cotonou
`parakou`	9.337	2.629	Parakou
`bohicon`	7.181	2.067	Bohicon
`kandi`	11.133	2.940	Kandi
`natitingou`	10.303	1.381	Natitingou
`malanville`	11.867	3.383	Malanville
`abomey`	7.183	1.983	Abomey
`porto-novo`	6.497	2.627	Porto-Novo
`lokossa`	6.617	1.717	Lokossa

La correspondance est partielle : 'Marché de Parakou' sera associé à 'parakou'. Les marchés non reconnus reçoivent None pour lat et lon. Une colonne absente génère un avertissement et est ignorée.

Retour : pd.DataFrame avec les colonnes market_lat et market_lon ajoutées.

`std_coords(lat=None, lon=None)`¶

Crée une colonne geometry contenant des objets shapely.geometry.Point à partir des colonnes de latitude et longitude.

df_norm = normalizer.std_coords(lat="latitude", lon="longitude")
# Colonne 'geometry' ajoutée : Point(lon, lat) pour chaque ligne

Paramètre	Type	Défaut	Description
`lat`	`str` ou `None`	`None`	Nom de la colonne de latitude
`lon`	`str` ou `None`	`None`	Nom de la colonne de longitude

Si lat ou lon est None ou désigne une colonne absente du DataFrame, un avertissement est émis dans les logs et le DataFrame est retourné sans modification.

Les lignes dont la latitude ou la longitude est NaN reçoivent None dans la colonne geometry.

Si le package shapely n'est pas installé, un avertissement est émis et le DataFrame est retourné sans modification.

Dépendance optionnelle : shapely >= 2.0. Installez-la avec pip install shapely>=2.0.

Retour : pd.DataFrame avec la colonne geometry ajoutée.

`mappings()`¶

Retourne l'historique complet des normalisations appliquées depuis l'initialisation du Normalizer.

normalizer.norm_cols()
normalizer.std_crops(col="culture")
normalizer.convert_units({"production": "tonne"})

hist = normalizer.mappings()
# {
#   "colonnes": {"Rendement (kg)": "rendement_kg", ...},
#   "unites":   {"production": {"unite_source": "tonne", "facteur_kg": 1000.0, "unite_cible": "kg"}},
#   "cultures": {"maïs": "maize", "Niébé": "cowpea"},
#   "marches":  {},
#   "devises":  {},
# }

Clé	Type	Contenu
`colonnes`	`dict`	`ancien_nom` → `nouveau_nom` pour chaque colonne renommée
`unites`	`dict`	`colonne` → `{unite_source, facteur_kg, unite_cible}`
`cultures`	`dict`	`nom_local` → `code_fao` pour chaque correspondance trouvée
`marches`	`dict`	Vide dans la version actuelle
`devises`	`dict`	`colonne` → `{from, to, taux}`

Retour : dict (copie de l'historique interne).

Exemple complet enchaîné¶

import kadi as kd
from kadi.kidas import Normalizer

df = kd.read_csv("recoltes_2024.csv")
normalizer = Normalizer(df)

df_norm = (
    normalizer
    .norm_cols()
    .std_crops(col="culture")
    .convert_units({"rendement_kg": "sac_100kg", "production": "tonne"})
    .std_markets(col="marche")
    .std_coords(lat="latitude", lon="longitude")
)

hist = normalizer.mappings()
print(f"Colonnes renommées : {len(hist['colonnes'])}")
print(f"Cultures normalisées : {len(hist['cultures'])}")
print(f"Unités converties : {len(hist['unites'])}")

Utilisation dans un Pipeline¶

Normalizer est utilisé en interne par Pipeline lorsqu'on ajoute une étape de normalisation via add_step ou normalize.

from kadi.kidas import Pipeline

df, rapport = (
    Pipeline()
    .add_source("recoltes_2024.csv")
    .add_step("norm_cols")
    .add_step("std_crops", col="culture")
    .add_step("convert_units", unit_map={"rendement_kg": "sac_100kg"})
    .run()
)

Pour la documentation du pipeline, voir Pipeline.

Rétrocompatibilité¶

Les anciens noms de méthodes émettent un DeprecationWarning et continuent de fonctionner jusqu'à KadiPy v2.0 :

Ancien nom (avant v1.2.0)	Nouveau nom
`normalize_column_names()`	`norm_cols()`
`normalize_units()`	`convert_units()`
`normalize_currencies()`	`convert_currency()`
`normalize_crop_names()`	`std_crops()`
`normalize_market_names()`	`std_markets()`
`normalize_geometry()`	`std_coords()`
`get_normalization_mapping()`	`mappings()`

De même, le nom de classe DataNormalizer est un alias de Normalizer.

Normalizer ¶

Classe de normalisation des données agricoles pour le contexte béninois.

Fournit des méthodes de transformation pour uniformiser les conventions de dénomination, les unités de mesure, les devises, les noms de cultures et de marchés, et pour créer des géométries GPS.

Toutes les opérations sont tracées dans un mapping interne récupérable via mappings().

Attributs

df (pd.DataFrame): Le DataFrame en cours de normalisation. _mappings (dict): Historique des transformations appliquées.

Exemple

normalizer = Normalizer(df) df_norm = ( ... normalizer ... .norm_cols() ... .std_crops(col='culture') ... .convert_units(unit_map={'production': 'tonne'}) ... )

Source code in kadi/kidas/normalizer.py

class Normalizer:
    """Classe de normalisation des données agricoles pour le contexte béninois.

    Fournit des méthodes de transformation pour uniformiser les conventions
    de dénomination, les unités de mesure, les devises, les noms de cultures
    et de marchés, et pour créer des géométries GPS.

    Toutes les opérations sont tracées dans un mapping interne récupérable
    via mappings().

    Attributs:
        df (pd.DataFrame): Le DataFrame en cours de normalisation.
        _mappings (dict): Historique des transformations appliquées.

    Exemple:
        >>> normalizer = Normalizer(df)
        >>> df_norm = (
        ...     normalizer
        ...     .norm_cols()
        ...     .std_crops(col='culture')
        ...     .convert_units(unit_map={'production': 'tonne'})
        ... )
    """

    # Table de rétrocompatibilité des méthodes d'instance.
    _METHODES_DEPRECATED = {
        "normalize_column_names": "norm_cols",
        "normalize_units":        "convert_units",
        "normalize_currencies":   "convert_currency",
        "normalize_crop_names":   "std_crops",
        "normalize_market_names": "std_markets",
        "normalize_geometry":     "std_coords",
        "get_normalization_mapping": "mappings",
    }

    def __init__(self, df: pd.DataFrame) -> None:
        """Initialise le normaliseur avec le DataFrame à transformer.

        Args:
            df (pd.DataFrame): Le DataFrame source à normaliser. Une copie
                interne est créée pour préserver l'original.

        Raises:
            CleanError: Si l'argument fourni n'est pas un DataFrame.
        """
        # Vérification du type d'entrée
        if not isinstance(df, pd.DataFrame):
            raise CleanError(
                f"Normalizer attend un pandas DataFrame, "
                f"reçu : {type(df).__name__}."
            )

        # Copie de travail du DataFrame
        self.df: pd.DataFrame = df.copy()

        # Historique des mappings de normalisation appliqués
        self._mappings: Dict = {
            "colonnes": {},
            "unites": {},
            "cultures": {},
            "marches": {},
            "devises": {},
        }

    @staticmethod
    def _to_snake(text: str) -> str:
        """Convertit une chaîne de caractères en snake_case.

        Supprime les accents, remplace les espaces et tirets par des
        underscores, et convertit en minuscules.

        Args:
            text (str): La chaîne à convertir.

        Returns:
            str: La chaîne en snake_case, sans accents ni caractères spéciaux.
        """
        # Suppression des accents via NFD + encodage ASCII
        sans_accent = (
            unicodedata.normalize("NFD", text)
            .encode("ascii", "ignore")
            .decode("utf-8")
        )

        # Remplacement des espaces, tirets et parenthèses par des underscores
        snake = re.sub(r"[\s\-\(\)/]+", "_", sans_accent)

        # Suppression des caractères non alphanumériques restants
        snake = re.sub(r"[^a-zA-Z0-9_]", "", snake)

        # Conversion en minuscules et nettoyage des underscores multiples
        snake = re.sub(r"_+", "_", snake).lower().strip("_")

        return snake

    # Alias interne maintenu pour la rétrocompatibilité du code interne existant
    _vers_snake_case = _to_snake

    def norm_cols(
        self,
        style: str = "snake_case",
    ) -> pd.DataFrame:
        """Normalise les noms de colonnes du DataFrame.

        Convertit les noms de colonnes en snake_case en supprimant les
        accents, les espaces et les caractères spéciaux.
        Exemple : 'Température Min (°C)' → 'temperature_min_c'.

        Args:
            style (str): Style de nommage cible. Seul 'snake_case' est
                supporté actuellement. Par défaut 'snake_case'.

        Returns:
            pd.DataFrame: DataFrame avec les noms de colonnes normalisés.
        """
        if style != "snake_case":
            logger.warning(
                "Style '%s' non supporté. Utilisation de 'snake_case'.", style
            )

        # Construction du mapping ancien_nom → nouveau_nom
        mapping_colonnes = {
            col: self._to_snake(col)
            for col in self.df.columns
        }

        # Application du renommage
        self.df = self.df.rename(columns=mapping_colonnes)

        # Journalisation des transformations
        changements = {
            ancien: nouveau
            for ancien, nouveau in mapping_colonnes.items()
            if ancien != nouveau
        }

        logger.info(
            "Normalisation noms de colonnes : %d colonne(s) renommée(s).",
            len(changements),
        )

        # Mise à jour du mapping de traçabilité
        self._mappings["colonnes"].update(mapping_colonnes)

        return self.df

    def convert_units(
        self,
        unit_map: Dict[str, str],
    ) -> pd.DataFrame:
        """Convertit les valeurs de colonnes numériques vers le kilogramme.

        Applique les facteurs de conversion définis dans le dictionnaire
        de référence _FACTEURS_UNITE_KG pour chaque colonne spécifiée.

        Args:
            unit_map (dict[str, str]): Dictionnaire nom_colonne → unité_source.
                Exemple : {'production': 'tonne', 'recolte': 'sac_100kg'}.

        Returns:
            pd.DataFrame: DataFrame avec les valeurs converties en kg.

        Raises:
            CleanError: Si une unité source est inconnue.
        """
        for colonne, unite_source in unit_map.items():
            if colonne not in self.df.columns:
                logger.warning(
                    "Colonne '%s' introuvable pour la normalisation d'unités.",
                    colonne,
                )
                continue

            # Normalisation du nom d'unité (minuscules, sans espaces)
            unite_normalisee = unite_source.lower().strip()

            if unite_normalisee not in _FACTEURS_UNITE_KG:
                raise CleanError(
                    f"Unité '{unite_source}' inconnue. Unités supportées : "
                    f"{list(_FACTEURS_UNITE_KG.keys())}."
                )

            # Application du facteur de conversion
            facteur = _FACTEURS_UNITE_KG[unite_normalisee]
            self.df[colonne] = self.df[colonne] * facteur

            logger.debug(
                "Colonne '%s' convertie : %s × %.4f → kg.",
                colonne,
                unite_source,
                facteur,
            )

            # Traçabilité de la conversion
            self._mappings["unites"][colonne] = {
                "unite_source": unite_source,
                "facteur_kg": facteur,
                "unite_cible": "kg",
            }

        return self.df

    def convert_currency(
        self,
        col: str,
        from_: str = "XOF",
        to: str = "XOF",
        date: Optional[str] = None,
        **kwargs,
    ) -> pd.DataFrame:
        """Convertit les valeurs monétaires d'une colonne.

        Note : Dans sa version actuelle, cette méthode applique un taux
        de change fixe. Une intégration avec une API de change peut être
        ajoutée en Phase 2 du module.

        Args:
            col (str): Nom de la colonne à convertir.
            from_ (str): Devise source (ex: 'USD', 'XOF', 'EUR').
                Par défaut 'XOF'. Alias accepté : from_currency= (rétrocompat.).
            to (str): Devise cible. Par défaut 'XOF'.
                Alias accepté : to_currency= (rétrocompat.).
            date (str | None): Date de référence pour le taux de change
                (format 'YYYY-MM-DD'). None pour le taux actuel.
                Alias accepté : exchange_date= (rétrocompat.).

        Returns:
            pd.DataFrame: DataFrame avec la colonne convertie.
        """
        # Rétrocompatibilité : anciens paramètres acceptés
        if "from_currency" in kwargs:
            from_ = kwargs.pop("from_currency")
        if "to_currency" in kwargs:
            to = kwargs.pop("to_currency")
        if "exchange_date" in kwargs:
            date = kwargs.pop("exchange_date")
        if col not in self.df.columns:
            logger.warning(
                "Colonne '%s' introuvable pour la conversion de devises.", col
            )
            return self.df

        # Taux de change fixes de référence (vers XOF)
        taux_vers_xof: Dict[str, float] = {
            "XOF": 1.0,
            "EUR": 655.957,   # Taux fixe XOF/EUR (UMOA)
            "USD": 600.0,     # Taux approximatif
            "GBP": 750.0,     # Taux approximatif
        }

        if from_ not in taux_vers_xof or to not in taux_vers_xof:
            logger.warning(
                "Paire de devises %s/%s non disponible. Aucune conversion.",
                from_,
                to,
            )
            return self.df

        # Calcul du taux de conversion croisé
        taux_conversion = taux_vers_xof[from_] / taux_vers_xof[to]

        # Application de la conversion
        self.df[col] = self.df[col] * taux_conversion

        logger.info(
            "Colonne '%s' convertie de %s vers %s (taux: %.4f).",
            col,
            from_,
            to,
            taux_conversion,
        )

        # Traçabilité de la conversion
        self._mappings["devises"][col] = {
            "from": from_,
            "to": to,
            "taux": taux_conversion,
        }

        return self.df

    def std_crops(
        self,
        col: str,
        std: str = "fao",
        **kwargs,
    ) -> pd.DataFrame:
        """Normalise les noms de cultures vers un standard international.

        Utilise le dictionnaire de référence _CULTURE_ALIASES pour mapper
        les noms locaux béninois (avec ou sans accents) vers les codes FAO.

        Args:
            col (str): Nom de la colonne contenant les noms de cultures.
            std (str): Standard cible. 'fao' utilise les codes
                officiels FAO (ex: 'maize', 'cowpea'). Par défaut 'fao'.
                Alias accepté : target_standard= (rétrocompatibilité).

        Returns:
            pd.DataFrame: DataFrame avec la colonne de cultures normalisée.
        """
        # Rétrocompatibilité : ancien paramètre accepté
        if "target_standard" in kwargs:
            std = kwargs.pop("target_standard")
        if col not in self.df.columns:
            logger.warning(
                "Colonne '%s' introuvable pour la normalisation des cultures.", col
            )
            return self.df

        def _normaliser_culture(valeur: str) -> str:
            """Normalise un nom de culture individuel.

            Args:
                valeur (str): Le nom de culture brut.

            Returns:
                str: Le code FAO correspondant ou la valeur originale en minuscules.
            """
            if not isinstance(valeur, str):
                return valeur

            # Normalisation : minuscules et suppression des accents
            valeur_norm = (
                unicodedata.normalize("NFD", valeur.lower().strip())
                .encode("ascii", "ignore")
                .decode("utf-8")
            )

            # Recherche dans le dictionnaire de référence
            if valeur_norm in _CULTURE_ALIASES:
                return _CULTURE_ALIASES[valeur_norm]

            # Clé épurée de symboles et chiffres (ex: 'mais#1' -> 'mais', 'soya*' -> 'soya')
            cle_epuree = re.sub(r"[^a-z\s]", "", valeur_norm).strip()
            if cle_epuree in _CULTURE_ALIASES:
                return _CULTURE_ALIASES[cle_epuree]

            return valeur_norm

        # Application du mapping sur toute la colonne
        avant = self.df[col].copy()
        self.df[col] = self.df[col].apply(_normaliser_culture)

        # Calcul des correspondances trouvées
        mapping_applique = {
            ancien: nouveau
            for ancien, nouveau in zip(avant, self.df[col])
            if ancien != nouveau
        }

        logger.info(
            "%d nom(s) de culture normalisé(s) vers standard '%s'.",
            len(mapping_applique),
            std,
        )

        self._mappings["cultures"].update(mapping_applique)

        return self.df

    def std_markets(
        self,
        col: str,
        region: str = "benin",
    ) -> pd.DataFrame:
        """Normalise les noms de marchés et assigne les coordonnées GPS.

        Mappe les noms de marchés béninois vers leur forme officielle
        et ajoute les colonnes 'market_lat' et 'market_lon' correspondantes.

        Args:
            col (str): Nom de la colonne contenant les noms de marchés.
            region (str): Région de référence pour le dictionnaire de marchés.
                Par défaut 'benin'.

        Returns:
            pd.DataFrame: DataFrame avec les noms de marchés normalisés
                et les colonnes 'market_lat' / 'market_lon' ajoutées.
        """
        if col not in self.df.columns:
            logger.warning(
                "Colonne '%s' introuvable pour la normalisation des marchés.", col
            )
            return self.df

        lats, lons = [], []

        for valeur in self.df[col]:
            if not isinstance(valeur, str):
                lats.append(None)
                lons.append(None)
                continue

            # Normalisation du nom de marché (minuscules, sans accents)
            nom_norm = (
                unicodedata.normalize("NFD", valeur.lower().strip())
                .encode("ascii", "ignore")
                .decode("utf-8")
            )

            # Recherche du marché dans la référence (correspondance partielle)
            coords_trouvees = None
            for cle_marche, coords in _MARCHE_COORDS.items():
                if cle_marche in nom_norm or nom_norm in cle_marche:
                    coords_trouvees = coords
                    break

            if coords_trouvees:
                lats.append(coords_trouvees["lat"])
                lons.append(coords_trouvees["lon"])
            else:
                lats.append(None)
                lons.append(None)

        # Ajout des colonnes de coordonnées au DataFrame
        self.df["market_lat"] = lats
        self.df["market_lon"] = lons

        nb_geolocalises = sum(1 for lat in lats if lat is not None)
        logger.info(
            "%d/%d marché(s) géolocalisé(s) dans '%s'.",
            nb_geolocalises,
            len(self.df),
            col,
        )

        return self.df

    def std_coords(
        self,
        lat: Optional[str] = None,
        lon: Optional[str] = None,
        **kwargs,
    ) -> pd.DataFrame:
        """Crée une colonne 'geometry' contenant des points GPS shapely.

        Args:
            lat (str | None): Nom de la colonne de latitude.
                Alias accepté : lat_col= (rétrocompatibilité).
            lon (str | None): Nom de la colonne de longitude.
                Alias accepté : lon_col= (rétrocompatibilité).

        Returns:
            pd.DataFrame: DataFrame avec la colonne 'geometry' ajoutée.
                Les lignes avec lat ou lon manquante ont geometry = None.
        """
        # Rétrocompatibilité : anciens paramètres acceptés
        if "lat_col" in kwargs:
            lat = kwargs.pop("lat_col")
        if "lon_col" in kwargs:
            lon = kwargs.pop("lon_col")

        try:
            # Import conditionnel de shapely (dépendance optionnelle)
            from shapely.geometry import Point
        except ImportError:
            logger.warning(
                "Le package 'shapely' n'est pas installé. "
                "Installez-le avec : pip install shapely>=2.0"
            )
            return self.df

        if lat not in self.df.columns or lon not in self.df.columns:
            logger.warning(
                "Colonnes '%s' ou '%s' introuvables pour la création de géométrie.",
                lat,
                lon,
            )
            return self.df

        # Création de la colonne 'geometry' avec des points shapely
        self.df["geometry"] = self.df.apply(
            lambda ligne: Point(ligne[lon], ligne[lat])
            if pd.notna(ligne[lat]) and pd.notna(ligne[lon])
            else None,
            axis=1,
        )

        nb_points = self.df["geometry"].notna().sum()
        logger.info(
            "%d point(s) géométrique(s) créé(s) depuis '%s'/'%s'.",
            nb_points,
            lat,
            lon,
        )

        return self.df

    def mappings(self) -> dict:
        """Retourne l'historique complet des normalisations appliquées.

        Returns:
            dict: Dictionnaire structuré par type de normalisation :
                - 'colonnes' (dict) : ancien → nouveau nom de colonne.
                - 'unites' (dict) : colonne → {unite_source, facteur_kg, unite_cible}.
                - 'cultures' (dict) : nom_local → code_fao.
                - 'marches' (dict) : nom_local → coordonnées.
                - 'devises' (dict) : colonne → {from, to, taux}.
        """
        return self._mappings.copy()

    def __getattr__(self, name: str):
        """Intercepte les anciens noms de méthodes pour la rétrocompatibilité.

        Args:
            name (str): Nom de la méthode demandée.

        Returns:
            callable: La méthode correspondant au nouveau nom.

        Raises:
            AttributeError: Si le nom n'est pas un alias connu.
        """
        if name in Normalizer._METHODES_DEPRECATED:
            nouveau_nom = Normalizer._METHODES_DEPRECATED[name]
            warnings.warn(
                f"Normalizer.{name}() est obsolète et sera supprimé dans KadiPy v2.0. "
                f"Utilisez Normalizer.{nouveau_nom}() à la place.",
                category=DeprecationWarning,
                stacklevel=2,
            )
            return getattr(self, nouveau_nom)
        raise AttributeError(
            f"'Normalizer' n'a pas de méthode '{name}'."
        )

init ¶

__init__(df: DataFrame) -> None

Initialise le normaliseur avec le DataFrame à transformer.

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	Le DataFrame source à normaliser. Une copie interne est créée pour préserver l'original.	required

Raises:

Type	Description
`CleanError`	Si l'argument fourni n'est pas un DataFrame.

Source code in kadi/kidas/normalizer.py

def __init__(self, df: pd.DataFrame) -> None:
    """Initialise le normaliseur avec le DataFrame à transformer.

    Args:
        df (pd.DataFrame): Le DataFrame source à normaliser. Une copie
            interne est créée pour préserver l'original.

    Raises:
        CleanError: Si l'argument fourni n'est pas un DataFrame.
    """
    # Vérification du type d'entrée
    if not isinstance(df, pd.DataFrame):
        raise CleanError(
            f"Normalizer attend un pandas DataFrame, "
            f"reçu : {type(df).__name__}."
        )

    # Copie de travail du DataFrame
    self.df: pd.DataFrame = df.copy()

    # Historique des mappings de normalisation appliqués
    self._mappings: Dict = {
        "colonnes": {},
        "unites": {},
        "cultures": {},
        "marches": {},
        "devises": {},
    }

norm_cols ¶

norm_cols(style: str = 'snake_case') -> pd.DataFrame

Normalise les noms de colonnes du DataFrame.

Convertit les noms de colonnes en snake_case en supprimant les accents, les espaces et les caractères spéciaux. Exemple : 'Température Min (°C)' → 'temperature_min_c'.

Parameters:

Name	Type	Description	Default
`style`	`str`	Style de nommage cible. Seul 'snake_case' est supporté actuellement. Par défaut 'snake_case'.	`'snake_case'`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame avec les noms de colonnes normalisés.

Source code in kadi/kidas/normalizer.py

def norm_cols(
    self,
    style: str = "snake_case",
) -> pd.DataFrame:
    """Normalise les noms de colonnes du DataFrame.

    Convertit les noms de colonnes en snake_case en supprimant les
    accents, les espaces et les caractères spéciaux.
    Exemple : 'Température Min (°C)' → 'temperature_min_c'.

    Args:
        style (str): Style de nommage cible. Seul 'snake_case' est
            supporté actuellement. Par défaut 'snake_case'.

    Returns:
        pd.DataFrame: DataFrame avec les noms de colonnes normalisés.
    """
    if style != "snake_case":
        logger.warning(
            "Style '%s' non supporté. Utilisation de 'snake_case'.", style
        )

    # Construction du mapping ancien_nom → nouveau_nom
    mapping_colonnes = {
        col: self._to_snake(col)
        for col in self.df.columns
    }

    # Application du renommage
    self.df = self.df.rename(columns=mapping_colonnes)

    # Journalisation des transformations
    changements = {
        ancien: nouveau
        for ancien, nouveau in mapping_colonnes.items()
        if ancien != nouveau
    }

    logger.info(
        "Normalisation noms de colonnes : %d colonne(s) renommée(s).",
        len(changements),
    )

    # Mise à jour du mapping de traçabilité
    self._mappings["colonnes"].update(mapping_colonnes)

    return self.df

convert_units ¶

convert_units(unit_map: Dict[str, str]) -> pd.DataFrame

Convertit les valeurs de colonnes numériques vers le kilogramme.

Applique les facteurs de conversion définis dans le dictionnaire de référence _FACTEURS_UNITE_KG pour chaque colonne spécifiée.

Parameters:

Name	Type	Description	Default
`unit_map`	`dict[str, str]`	Dictionnaire nom_colonne → unité_source. Exemple : {'production': 'tonne', 'recolte': 'sac_100kg'}.	required

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame avec les valeurs converties en kg.

Raises:

Type	Description
`CleanError`	Si une unité source est inconnue.

Source code in kadi/kidas/normalizer.py

def convert_units(
    self,
    unit_map: Dict[str, str],
) -> pd.DataFrame:
    """Convertit les valeurs de colonnes numériques vers le kilogramme.

    Applique les facteurs de conversion définis dans le dictionnaire
    de référence _FACTEURS_UNITE_KG pour chaque colonne spécifiée.

    Args:
        unit_map (dict[str, str]): Dictionnaire nom_colonne → unité_source.
            Exemple : {'production': 'tonne', 'recolte': 'sac_100kg'}.

    Returns:
        pd.DataFrame: DataFrame avec les valeurs converties en kg.

    Raises:
        CleanError: Si une unité source est inconnue.
    """
    for colonne, unite_source in unit_map.items():
        if colonne not in self.df.columns:
            logger.warning(
                "Colonne '%s' introuvable pour la normalisation d'unités.",
                colonne,
            )
            continue

        # Normalisation du nom d'unité (minuscules, sans espaces)
        unite_normalisee = unite_source.lower().strip()

        if unite_normalisee not in _FACTEURS_UNITE_KG:
            raise CleanError(
                f"Unité '{unite_source}' inconnue. Unités supportées : "
                f"{list(_FACTEURS_UNITE_KG.keys())}."
            )

        # Application du facteur de conversion
        facteur = _FACTEURS_UNITE_KG[unite_normalisee]
        self.df[colonne] = self.df[colonne] * facteur

        logger.debug(
            "Colonne '%s' convertie : %s × %.4f → kg.",
            colonne,
            unite_source,
            facteur,
        )

        # Traçabilité de la conversion
        self._mappings["unites"][colonne] = {
            "unite_source": unite_source,
            "facteur_kg": facteur,
            "unite_cible": "kg",
        }

    return self.df

convert_currency ¶

convert_currency(col: str, from_: str = 'XOF', to: str = 'XOF', date: Optional[str] = None, **kwargs) -> pd.DataFrame

Convertit les valeurs monétaires d'une colonne.

Note : Dans sa version actuelle, cette méthode applique un taux de change fixe. Une intégration avec une API de change peut être ajoutée en Phase 2 du module.

Parameters:

Name	Type	Description	Default
`col`	`str`	Nom de la colonne à convertir.	required
`from_`	`str`	Devise source (ex: 'USD', 'XOF', 'EUR'). Par défaut 'XOF'. Alias accepté : from_currency= (rétrocompat.).	`'XOF'`
`to`	`str`	Devise cible. Par défaut 'XOF'. Alias accepté : to_currency= (rétrocompat.).	`'XOF'`
`date`	`str \| None`	Date de référence pour le taux de change (format 'YYYY-MM-DD'). None pour le taux actuel. Alias accepté : exchange_date= (rétrocompat.).	`None`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame avec la colonne convertie.

Source code in kadi/kidas/normalizer.py

def convert_currency(
    self,
    col: str,
    from_: str = "XOF",
    to: str = "XOF",
    date: Optional[str] = None,
    **kwargs,
) -> pd.DataFrame:
    """Convertit les valeurs monétaires d'une colonne.

    Note : Dans sa version actuelle, cette méthode applique un taux
    de change fixe. Une intégration avec une API de change peut être
    ajoutée en Phase 2 du module.

    Args:
        col (str): Nom de la colonne à convertir.
        from_ (str): Devise source (ex: 'USD', 'XOF', 'EUR').
            Par défaut 'XOF'. Alias accepté : from_currency= (rétrocompat.).
        to (str): Devise cible. Par défaut 'XOF'.
            Alias accepté : to_currency= (rétrocompat.).
        date (str | None): Date de référence pour le taux de change
            (format 'YYYY-MM-DD'). None pour le taux actuel.
            Alias accepté : exchange_date= (rétrocompat.).

    Returns:
        pd.DataFrame: DataFrame avec la colonne convertie.
    """
    # Rétrocompatibilité : anciens paramètres acceptés
    if "from_currency" in kwargs:
        from_ = kwargs.pop("from_currency")
    if "to_currency" in kwargs:
        to = kwargs.pop("to_currency")
    if "exchange_date" in kwargs:
        date = kwargs.pop("exchange_date")
    if col not in self.df.columns:
        logger.warning(
            "Colonne '%s' introuvable pour la conversion de devises.", col
        )
        return self.df

    # Taux de change fixes de référence (vers XOF)
    taux_vers_xof: Dict[str, float] = {
        "XOF": 1.0,
        "EUR": 655.957,   # Taux fixe XOF/EUR (UMOA)
        "USD": 600.0,     # Taux approximatif
        "GBP": 750.0,     # Taux approximatif
    }

    if from_ not in taux_vers_xof or to not in taux_vers_xof:
        logger.warning(
            "Paire de devises %s/%s non disponible. Aucune conversion.",
            from_,
            to,
        )
        return self.df

    # Calcul du taux de conversion croisé
    taux_conversion = taux_vers_xof[from_] / taux_vers_xof[to]

    # Application de la conversion
    self.df[col] = self.df[col] * taux_conversion

    logger.info(
        "Colonne '%s' convertie de %s vers %s (taux: %.4f).",
        col,
        from_,
        to,
        taux_conversion,
    )

    # Traçabilité de la conversion
    self._mappings["devises"][col] = {
        "from": from_,
        "to": to,
        "taux": taux_conversion,
    }

    return self.df

std_crops ¶

std_crops(col: str, std: str = 'fao', **kwargs) -> pd.DataFrame

Normalise les noms de cultures vers un standard international.

Utilise le dictionnaire de référence _CULTURE_ALIASES pour mapper les noms locaux béninois (avec ou sans accents) vers les codes FAO.

Parameters:

Name	Type	Description	Default
`col`	`str`	Nom de la colonne contenant les noms de cultures.	required
`std`	`str`	Standard cible. 'fao' utilise les codes officiels FAO (ex: 'maize', 'cowpea'). Par défaut 'fao'. Alias accepté : target_standard= (rétrocompatibilité).	`'fao'`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame avec la colonne de cultures normalisée.

Source code in kadi/kidas/normalizer.py

def std_crops(
    self,
    col: str,
    std: str = "fao",
    **kwargs,
) -> pd.DataFrame:
    """Normalise les noms de cultures vers un standard international.

    Utilise le dictionnaire de référence _CULTURE_ALIASES pour mapper
    les noms locaux béninois (avec ou sans accents) vers les codes FAO.

    Args:
        col (str): Nom de la colonne contenant les noms de cultures.
        std (str): Standard cible. 'fao' utilise les codes
            officiels FAO (ex: 'maize', 'cowpea'). Par défaut 'fao'.
            Alias accepté : target_standard= (rétrocompatibilité).

    Returns:
        pd.DataFrame: DataFrame avec la colonne de cultures normalisée.
    """
    # Rétrocompatibilité : ancien paramètre accepté
    if "target_standard" in kwargs:
        std = kwargs.pop("target_standard")
    if col not in self.df.columns:
        logger.warning(
            "Colonne '%s' introuvable pour la normalisation des cultures.", col
        )
        return self.df

    def _normaliser_culture(valeur: str) -> str:
        """Normalise un nom de culture individuel.

        Args:
            valeur (str): Le nom de culture brut.

        Returns:
            str: Le code FAO correspondant ou la valeur originale en minuscules.
        """
        if not isinstance(valeur, str):
            return valeur

        # Normalisation : minuscules et suppression des accents
        valeur_norm = (
            unicodedata.normalize("NFD", valeur.lower().strip())
            .encode("ascii", "ignore")
            .decode("utf-8")
        )

        # Recherche dans le dictionnaire de référence
        if valeur_norm in _CULTURE_ALIASES:
            return _CULTURE_ALIASES[valeur_norm]

        # Clé épurée de symboles et chiffres (ex: 'mais#1' -> 'mais', 'soya*' -> 'soya')
        cle_epuree = re.sub(r"[^a-z\s]", "", valeur_norm).strip()
        if cle_epuree in _CULTURE_ALIASES:
            return _CULTURE_ALIASES[cle_epuree]

        return valeur_norm

    # Application du mapping sur toute la colonne
    avant = self.df[col].copy()
    self.df[col] = self.df[col].apply(_normaliser_culture)

    # Calcul des correspondances trouvées
    mapping_applique = {
        ancien: nouveau
        for ancien, nouveau in zip(avant, self.df[col])
        if ancien != nouveau
    }

    logger.info(
        "%d nom(s) de culture normalisé(s) vers standard '%s'.",
        len(mapping_applique),
        std,
    )

    self._mappings["cultures"].update(mapping_applique)

    return self.df

std_markets ¶

std_markets(col: str, region: str = 'benin') -> pd.DataFrame

Normalise les noms de marchés et assigne les coordonnées GPS.

Mappe les noms de marchés béninois vers leur forme officielle et ajoute les colonnes 'market_lat' et 'market_lon' correspondantes.

Parameters:

Name	Type	Description	Default
`col`	`str`	Nom de la colonne contenant les noms de marchés.	required
`region`	`str`	Région de référence pour le dictionnaire de marchés. Par défaut 'benin'.	`'benin'`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame avec les noms de marchés normalisés et les colonnes 'market_lat' / 'market_lon' ajoutées.

Source code in kadi/kidas/normalizer.py

def std_markets(
    self,
    col: str,
    region: str = "benin",
) -> pd.DataFrame:
    """Normalise les noms de marchés et assigne les coordonnées GPS.

    Mappe les noms de marchés béninois vers leur forme officielle
    et ajoute les colonnes 'market_lat' et 'market_lon' correspondantes.

    Args:
        col (str): Nom de la colonne contenant les noms de marchés.
        region (str): Région de référence pour le dictionnaire de marchés.
            Par défaut 'benin'.

    Returns:
        pd.DataFrame: DataFrame avec les noms de marchés normalisés
            et les colonnes 'market_lat' / 'market_lon' ajoutées.
    """
    if col not in self.df.columns:
        logger.warning(
            "Colonne '%s' introuvable pour la normalisation des marchés.", col
        )
        return self.df

    lats, lons = [], []

    for valeur in self.df[col]:
        if not isinstance(valeur, str):
            lats.append(None)
            lons.append(None)
            continue

        # Normalisation du nom de marché (minuscules, sans accents)
        nom_norm = (
            unicodedata.normalize("NFD", valeur.lower().strip())
            .encode("ascii", "ignore")
            .decode("utf-8")
        )

        # Recherche du marché dans la référence (correspondance partielle)
        coords_trouvees = None
        for cle_marche, coords in _MARCHE_COORDS.items():
            if cle_marche in nom_norm or nom_norm in cle_marche:
                coords_trouvees = coords
                break

        if coords_trouvees:
            lats.append(coords_trouvees["lat"])
            lons.append(coords_trouvees["lon"])
        else:
            lats.append(None)
            lons.append(None)

    # Ajout des colonnes de coordonnées au DataFrame
    self.df["market_lat"] = lats
    self.df["market_lon"] = lons

    nb_geolocalises = sum(1 for lat in lats if lat is not None)
    logger.info(
        "%d/%d marché(s) géolocalisé(s) dans '%s'.",
        nb_geolocalises,
        len(self.df),
        col,
    )

    return self.df

std_coords ¶

std_coords(lat: Optional[str] = None, lon: Optional[str] = None, **kwargs) -> pd.DataFrame

Crée une colonne 'geometry' contenant des points GPS shapely.

Parameters:

Name	Type	Description	Default
`lat`	`str \| None`	Nom de la colonne de latitude. Alias accepté : lat_col= (rétrocompatibilité).	`None`
`lon`	`str \| None`	Nom de la colonne de longitude. Alias accepté : lon_col= (rétrocompatibilité).	`None`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: DataFrame avec la colonne 'geometry' ajoutée. Les lignes avec lat ou lon manquante ont geometry = None.

Source code in kadi/kidas/normalizer.py

def std_coords(
    self,
    lat: Optional[str] = None,
    lon: Optional[str] = None,
    **kwargs,
) -> pd.DataFrame:
    """Crée une colonne 'geometry' contenant des points GPS shapely.

    Args:
        lat (str | None): Nom de la colonne de latitude.
            Alias accepté : lat_col= (rétrocompatibilité).
        lon (str | None): Nom de la colonne de longitude.
            Alias accepté : lon_col= (rétrocompatibilité).

    Returns:
        pd.DataFrame: DataFrame avec la colonne 'geometry' ajoutée.
            Les lignes avec lat ou lon manquante ont geometry = None.
    """
    # Rétrocompatibilité : anciens paramètres acceptés
    if "lat_col" in kwargs:
        lat = kwargs.pop("lat_col")
    if "lon_col" in kwargs:
        lon = kwargs.pop("lon_col")

    try:
        # Import conditionnel de shapely (dépendance optionnelle)
        from shapely.geometry import Point
    except ImportError:
        logger.warning(
            "Le package 'shapely' n'est pas installé. "
            "Installez-le avec : pip install shapely>=2.0"
        )
        return self.df

    if lat not in self.df.columns or lon not in self.df.columns:
        logger.warning(
            "Colonnes '%s' ou '%s' introuvables pour la création de géométrie.",
            lat,
            lon,
        )
        return self.df

    # Création de la colonne 'geometry' avec des points shapely
    self.df["geometry"] = self.df.apply(
        lambda ligne: Point(ligne[lon], ligne[lat])
        if pd.notna(ligne[lat]) and pd.notna(ligne[lon])
        else None,
        axis=1,
    )

    nb_points = self.df["geometry"].notna().sum()
    logger.info(
        "%d point(s) géométrique(s) créé(s) depuis '%s'/'%s'.",
        nb_points,
        lat,
        lon,
    )

    return self.df

mappings ¶

mappings() -> dict

Retourne l'historique complet des normalisations appliquées.

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire structuré par type de normalisation : - 'colonnes' (dict) : ancien → nouveau nom de colonne. - 'unites' (dict) : colonne → {unite_source, facteur_kg, unite_cible}. - 'cultures' (dict) : nom_local → code_fao. - 'marches' (dict) : nom_local → coordonnées. - 'devises' (dict) : colonne → {from, to, taux}.

Source code in kadi/kidas/normalizer.py

def mappings(self) -> dict:
    """Retourne l'historique complet des normalisations appliquées.

    Returns:
        dict: Dictionnaire structuré par type de normalisation :
            - 'colonnes' (dict) : ancien → nouveau nom de colonne.
            - 'unites' (dict) : colonne → {unite_source, facteur_kg, unite_cible}.
            - 'cultures' (dict) : nom_local → code_fao.
            - 'marches' (dict) : nom_local → coordonnées.
            - 'devises' (dict) : colonne → {from, to, taux}.
    """
    return self._mappings.copy()

getattr ¶

__getattr__(name: str)

Intercepte les anciens noms de méthodes pour la rétrocompatibilité.

Parameters:

Name	Type	Description	Default
`name`	`str`	Nom de la méthode demandée.	required

Returns:

Name	Type	Description
`callable`		La méthode correspondant au nouveau nom.

Raises:

Type	Description
`AttributeError`	Si le nom n'est pas un alias connu.

Source code in kadi/kidas/normalizer.py

def __getattr__(self, name: str):
    """Intercepte les anciens noms de méthodes pour la rétrocompatibilité.

    Args:
        name (str): Nom de la méthode demandée.

    Returns:
        callable: La méthode correspondant au nouveau nom.

    Raises:
        AttributeError: Si le nom n'est pas un alias connu.
    """
    if name in Normalizer._METHODES_DEPRECATED:
        nouveau_nom = Normalizer._METHODES_DEPRECATED[name]
        warnings.warn(
            f"Normalizer.{name}() est obsolète et sera supprimé dans KadiPy v2.0. "
            f"Utilisez Normalizer.{nouveau_nom}() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return getattr(self, nouveau_nom)
    raise AttributeError(
        f"'Normalizer' n'a pas de méthode '{name}'."
    )

Normalizer (kadi.kidas.normalizer)¶

Initialisation¶

Méthodes de normalisation¶

norm_cols(style='snake_case')¶

convert_units(unit_map)¶

convert_currency(col, from_='XOF', to='XOF', date=None)¶

std_crops(col, std='fao')¶

std_markets(col, region='benin')¶

std_coords(lat=None, lon=None)¶

mappings()¶

Exemple complet enchaîné¶

Utilisation dans un Pipeline¶

Rétrocompatibilité¶

Normalizer ¶

__init__ ¶

norm_cols ¶

convert_units ¶

convert_currency ¶

std_crops ¶

std_markets ¶

std_coords ¶

mappings ¶

__getattr__ ¶

Normalizer (`kadi.kidas.normalizer`)¶

`norm_cols(style='snake_case')`¶

`convert_units(unit_map)`¶

`convert_currency(col, from_='XOF', to='XOF', date=None)`¶

`std_crops(col, std='fao')`¶

`std_markets(col, region='benin')`¶

`std_coords(lat=None, lon=None)`¶

`mappings()`¶

init ¶

getattr ¶