Normalizer (`kadi.kidas.normalizer`)¶

Normalizer is the agricultural data normalization class of KadiPy, tailored for the Beninese context. It gathers six transformations: column name normalization to snake_case, unit conversion to kilograms, currency conversion, crop name standardization to FAO codes, market geocoding, and creation of Shapely GPS geometries.

Each method updates the internal DataFrame and returns it, allowing method chaining in a single expression. The history of applied transformations is accessible via mappings().

Initialization¶

from kadi.kidas import Normalizer

normalizer = Normalizer(df)

Normalizer receives a pandas.DataFrame and creates an internal copy. The original DataFrame is never modified.

Exception: CleanError if the provided argument is not a DataFrame.

Normalization Methods¶

`norm_cols(style='snake_case')`¶

Normalizes DataFrame column names to snake_case: accent removal, replacing spaces, hyphens, and parentheses with underscores, converting to lowercase.

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

Parameter	Type	Default	Description
`style`	`str`	`'snake_case'`	Target style. Only `'snake_case'` is currently supported.

A style other than 'snake_case' emits a log warning and the operation is still applied in snake_case.

Returns: pd.DataFrame

`convert_units(unit_map)`¶

Converts numeric values of specified columns to kilograms, applying conversion factors from the internal reference table.

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

Parameter	Type	Description
`unit_map`	`dict[str, str]`	Dictionary `column_name` -> `source_unit`

Supported Units:

Unit	Factor to 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

The standard 'sac' corresponds to the 100 kg bag used in Benin. The 'tiya' is a local measure of approximately 1.5 kg depending on the crop. A column missing from the DataFrame emits a log warning and is ignored without raising an error.

Returns: pd.DataFrame

Exception: CleanError if a source unit is unknown.

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

Converts monetary values of a column using a fixed exchange rate.

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

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

Parameter	Type	Default	Description
`col`	`str`	required	Name of the column to convert
`from_`	`str`	`'XOF'`	Source currency
`to`	`str`	`'XOF'`	Target currency
`date`	`str` or `None`	`None`	Reference date in `'YYYY-MM-DD'` format (unused in current version)

Available Exchange Rates (fixed, to XOF):

Currency	Rate to XOF
`'XOF'`	1.0
`'EUR'`	655.957 (WAEMU fixed rate)
`'USD'`	600.0
`'GBP'`	750.0

If one of the currencies is not in this table, a warning is emitted and no conversion is applied.

The current version uses fixed rates. Integration with a real-time exchange rate API is planned for phase 2 of the module.

Returns: pd.DataFrame

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

Normalizes crop names in a column to official FAO codes, handling local Beninese variants with or without accents.

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

Parameter	Type	Default	Description
`col`	`str`	required	Name of the column containing crop names
`std`	`str`	`'fao'`	Target standard (`'fao'` uses official FAO codes)

Available Mappings:

Accepted Local Names	FAO Code
`'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'`

Unrecognized names are kept in lowercase without accents. A missing column emits a warning and is ignored.

Returns: pd.DataFrame

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

Normalizes market names in a column and adds two columns market_lat and market_lon with official GPS coordinates.

# Before: ['Dantokpa', 'PARAKOU', 'Bohicon']
df_norm = normalizer.std_markets(col="marche")
# After: 'marche' column unchanged, + 'market_lat' and 'market_lon' columns

Parameter	Type	Default	Description
`col`	`str`	required	Name of the column containing market names
`region`	`str`	`'benin'`	Reference region for market dictionary

Referenced Markets (Benin):

Market	Latitude	Longitude	City
`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

Matching is partial: 'Marché de Parakou' will be matched to 'parakou'. Unrecognized markets receive None for lat and lon. A missing column emits a warning and is ignored.

Returns: pd.DataFrame with added market_lat and market_lon columns.

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

Creates a geometry column containing shapely.geometry.Point objects from latitude and longitude columns.

df_norm = normalizer.std_coords(lat="latitude", lon="longitude")
# 'geometry' column added: Point(lon, lat) for each row

Parameter	Type	Default	Description
`lat`	`str` or `None`	`None`	Latitude column name
`lon`	`str` or `None`	`None`	Longitude column name

If lat or lon is None or refers to a missing column in the DataFrame, a log warning is emitted and the DataFrame is returned unchanged.

Rows with NaN latitude or longitude receive None in the geometry column.

If the shapely package is not installed, a warning is emitted and the DataFrame is returned unchanged.

Optional dependency: shapely >= 2.0. Install it with pip install shapely>=2.0.

Returns: pd.DataFrame with added geometry column.

`mappings()`¶

Returns the full history of applied normalizations since Normalizer initialization.

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":  {},
# }

Key	Type	Content
`colonnes`	`dict`	`old_name` -> `new_name` for each renamed column
`unites`	`dict`	`column` -> `{unite_source, facteur_kg, unite_cible}`
`cultures`	`dict`	`local_name` -> `fao_code` for each match found
`marches`	`dict`	Empty in current version
`devises`	`dict`	`column` -> `{from, to, taux}`

Returns: dict (copy of internal history).

Full Chained Example¶

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"Renamed columns: {len(hist['colonnes'])}")
print(f"Normalized crops: {len(hist['cultures'])}")
print(f"Converted units: {len(hist['unites'])}")

Usage in a Pipeline¶

Normalizer is used internally by Pipeline when adding a normalization step via add_step or normalize.

from kadi.kidas import Pipeline

df, report = (
    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()
)

For pipeline documentation, see Pipeline.

Backward Compatibility¶

Legacy method names emit a DeprecationWarning and continue to function until KadiPy v2.0:

Legacy Name (before v1.2.0)	New Name
`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()`

Likewise, the class name DataNormalizer is an alias of 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)¶

Initialization¶

Normalization Methods¶

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()¶

Full Chained Example¶

Usage in a Pipeline¶

Backward Compatibility¶

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 ¶