Validation (`kadi.kidas.validator`)¶

DataValidator vérifie la cohérence et la qualité des données agricoles selon un schéma défini. Il produit un rapport d'anomalies sans modifier les données.

Principe¶

La validation est non-destructive : elle ne supprime ni ne modifie les données. Elle signale les problèmes dans un rapport structuré et retourne un score de qualité global.

from kadi.kidas import DataValidator

validator = DataValidator(df)
rapport = validator.validate_schema(schema={
    "culture": "str",
    "rendement_kg_ha": "float",
    "latitude": "float",
})

Méthodes¶

`validate_schema(schema)`¶

Valide les colonnes du DataFrame selon le schéma fourni.

rapport = validator.validate_schema(schema={
    "culture": "str",           # Type de la colonne
    "rendement_kg": "float",    # Valeur numérique obligatoire
    "date_recolte": "date",     # Format date
    "superficie_ha": "float",
    "is_organic": "bool",
})

Types supportés :

Type	Vérification effectuée
`'str'`	Colonne non numérique, valeurs non nulles
`'int'`	Entiers, pas de valeurs non entières
`'float'`	Nombres réels, détection de `-inf`/`inf`
`'bool'`	Seulement `True` / `False`
`'date'`	Format ISO 8601 ou datetime pandas

`check_ranges(rules)`¶

Vérifie que les valeurs numériques se trouvent dans des plages acceptables.

rapport = validator.check_ranges({
    "rendement_kg_ha": (0, 20_000),   # Entre 0 et 20 t/ha
    "superficie_ha": (0.01, 5_000),   # Entre 10 m² et 5000 ha
    "prix_xof_kg": (10, 10_000),      # Entre 10 et 10 000 XOF/kg
    "latitude": (6.0, 12.5),          # Bornes Bénin
    "longitude": (0.5, 3.9),          # Bornes Bénin
})

`check_referential(column, allowed_values)`¶

Vérifie que les valeurs d'une colonne appartiennent à un référentiel défini.

cultures_valides = [
    "maize", "rice", "sorghum", "millet", "cowpea",
    "soybean", "yam", "cassava", "tomato", "onion",
]
rapport = validator.check_referential("culture", allowed_values=cultures_valides)

# Valider les communes béninoises
communes_benin = ["Cotonou", "Parakou", "Abomey", "Natitingou", ...]
rapport = validator.check_referential("commune", allowed_values=communes_benin)

Rapport de validation¶

rapport = validator.validate(schema={...})

# Score de qualité (0 à 1)
print(f"Score global : {rapport['quality_score']['overall']:.2f}")

# Résumé par colonne
for col, details in rapport["columns"].items():
    if details["errors"] > 0:
        print(f"  {col} : {details['errors']} erreurs — {details['error_type']}")

# Liste complète des avertissements
for avertissement in rapport["warnings"]:
    print(f"  Attention : {avertissement}")

# Lignes problématiques (index dans le DataFrame)
print(f"Lignes avec erreurs : {rapport['invalid_row_indices']}")

Structure du rapport :

Clé	Type	Description
`quality_score`	`dict`	Scores par dimension et score global
`columns`	`dict`	Rapport par colonne (nb erreurs, type d'erreur)
`warnings`	`list[str]`	Liste des avertissements
`invalid_row_indices`	`list[int]`	Index des lignes problématiques
`nb_rows_validated`	`int`	Nombre de lignes validées

Exemple complet¶

import pandas as pd
from kadi.kidas import DataValidator

df = pd.read_csv("recoltes_enquete.csv")

validator = DataValidator(df)

# Étape 1 : validation des types
rapport_types = validator.validate(schema={
    "culture": "str",
    "rendement_kg": "float",
    "date_recolte": "date",
    "commune": "str",
})

# Étape 2 : validation des plages de valeurs
rapport_ranges = validator.check_ranges({
    "rendement_kg": (0, 50_000),
    "superficie_m2": (100, 50_000_000),
})

# Étape 3 : validation référentielle
rapport_ref = validator.check_referential("culture", [
    "maize", "rice", "sorghum", "millet", "cowpea", "yam",
])

print(f"Types OK    : {rapport_types['quality_score']['overall']:.2f}")
print(f"Plages OK   : {rapport_ranges['quality_score']['overall']:.2f}")
print(f"Référentiel : {rapport_ref['quality_score']['overall']:.2f}")

DataValidator ¶

Classe de validation qualité des données agricoles tabulaires.

Fournit une suite de vérifications permettant de s'assurer que les données respectent un schéma, des types, des intervalles de valeurs et des contraintes géographiques propres au contexte béninois.

Un score de qualité global est calculé à partir des dimensions de complétude, cohérence et précision.

Attributs

df (pd.DataFrame): Le DataFrame à valider. _rapport (dict): Journal des résultats de validation.

Exemple

validator = DataValidator(df) valide, erreurs = validator.validate_schema({ ... 'culture': 'str', ... 'rendement_kg': 'float', ... }) score = validator.compute_quality_score() print(score['overall']) 0.87

Source code in kadi/kidas/validator.py

class DataValidator:
    """Classe de validation qualité des données agricoles tabulaires.

    Fournit une suite de vérifications permettant de s'assurer que les
    données respectent un schéma, des types, des intervalles de valeurs
    et des contraintes géographiques propres au contexte béninois.

    Un score de qualité global est calculé à partir des dimensions de
    complétude, cohérence et précision.

    Attributs:
        df (pd.DataFrame): Le DataFrame à valider.
        _rapport (dict): Journal des résultats de validation.

    Exemple:
        >>> validator = DataValidator(df)
        >>> valide, erreurs = validator.validate_schema({
        ...     'culture': 'str',
        ...     'rendement_kg': 'float',
        ... })
        >>> score = validator.compute_quality_score()
        >>> print(score['overall'])
        0.87
    """

    def __init__(self, df: pd.DataFrame) -> None:
        """Initialise le validateur avec le DataFrame à contrôler.

        Args:
            df (pd.DataFrame): Le DataFrame à valider.

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

        # DataFrame de référence (sans copie : lecture seule)
        self.df: pd.DataFrame = df

        # Rapport de validation initialisé
        self._rapport: Dict = {
            "lignes": len(df),
            "colonnes": len(df.columns),
            "validations": [],
        }

    def validate_schema(
        self,
        schema: Dict[str, str],
    ) -> Tuple[bool, List[str]]:
        """Vérifie que le DataFrame possède les colonnes définies dans le schéma.

        Args:
            schema (dict[str, str]): Dictionnaire nom_colonne → type attendu.
                Les types acceptés sont : 'str', 'int', 'float', 'datetime', 'bool'.
                Exemple : {'culture': 'str', 'rendement_kg': 'float'}.

        Returns:
            tuple[bool, list[str]]: Tuple contenant :
                - True si le schéma est valide, False sinon.
                - Liste des messages d'erreur (vide si valide).
        """
        erreurs: List[str] = []

        for nom_col, type_attendu in schema.items():
            # Vérification de la présence de la colonne
            if nom_col not in self.df.columns:
                erreurs.append(
                    f"Colonne manquante : '{nom_col}' (type attendu : '{type_attendu}')."
                )
                continue

            # Vérification du type si spécifié et connu
            if type_attendu in _TYPE_MAP:
                type_pandas = _TYPE_MAP[type_attendu]
                est_correct = pd.api.types.is_dtype_equal(
                    self.df[nom_col].dtype,
                    type_pandas,
                ) or isinstance(self.df[nom_col].dtype.type, type_pandas) if hasattr(type_pandas, '__mro__') else False

                # Vérification simplifiée selon la catégorie de type
                if type_attendu in ("str", "string") and not pd.api.types.is_string_dtype(self.df[nom_col]):
                    if not pd.api.types.is_object_dtype(self.df[nom_col]):
                        erreurs.append(
                            f"Type incorrect pour '{nom_col}' : "
                            f"attendu '{type_attendu}', "
                            f"reçu '{self.df[nom_col].dtype}'."
                        )
                elif type_attendu in ("int", "integer") and not pd.api.types.is_integer_dtype(self.df[nom_col]):
                    erreurs.append(
                        f"Type incorrect pour '{nom_col}' : "
                        f"attendu '{type_attendu}', "
                        f"reçu '{self.df[nom_col].dtype}'."
                    )
                elif type_attendu == "float" and not pd.api.types.is_float_dtype(self.df[nom_col]):
                    erreurs.append(
                        f"Type incorrect pour '{nom_col}' : "
                        f"attendu '{type_attendu}', "
                        f"reçu '{self.df[nom_col].dtype}'."
                    )
                elif type_attendu == "datetime" and not pd.api.types.is_datetime64_any_dtype(self.df[nom_col]):
                    erreurs.append(
                        f"Type incorrect pour '{nom_col}' : "
                        f"attendu '{type_attendu}', "
                        f"reçu '{self.df[nom_col].dtype}'."
                    )

        est_valide = len(erreurs) == 0
        logger.info(
            "Validation schéma : %s (%d erreur(s)).",
            "OK" if est_valide else "ECHEC",
            len(erreurs),
        )

        # Enregistrement dans le rapport
        self._rapport["validations"].append(
            {
                "type": "schema",
                "valide": est_valide,
                "nb_erreurs": len(erreurs),
                "erreurs": erreurs,
            }
        )

        return est_valide, erreurs

    def validate_types(
        self,
        column_dtypes: Dict[str, str],
    ) -> Tuple[bool, pd.DataFrame]:
        """Vérifie la conformité des types pandas pour chaque colonne.

        Args:
            column_dtypes (dict[str, str]): Dictionnaire nom_colonne → type
                pandas attendu (ex: 'int64', 'float64', 'object', 'datetime64[ns]').

        Returns:
            tuple[bool, pd.DataFrame]: Tuple contenant :
                - True si tous les types correspondent.
                - DataFrame des colonnes avec des types incorrects (vide si OK).
        """
        lignes_erreurs = []

        for colonne, dtype_attendu in column_dtypes.items():
            if colonne not in self.df.columns:
                lignes_erreurs.append({
                    "colonne": colonne,
                    "dtype_attendu": dtype_attendu,
                    "dtype_reel": "ABSENT",
                })
                continue

            dtype_reel = str(self.df[colonne].dtype)

            # Comparaison des types (flexible pour les variantes d'int/float)
            types_compatibles = False
            if dtype_attendu in dtype_reel or dtype_reel in dtype_attendu:
                types_compatibles = True
            elif "int" in dtype_attendu and pd.api.types.is_integer_dtype(self.df[colonne]):
                types_compatibles = True
            elif "float" in dtype_attendu and pd.api.types.is_float_dtype(self.df[colonne]):
                types_compatibles = True

            if not types_compatibles:
                lignes_erreurs.append({
                    "colonne": colonne,
                    "dtype_attendu": dtype_attendu,
                    "dtype_reel": dtype_reel,
                })

        df_erreurs = pd.DataFrame(lignes_erreurs)
        est_valide = len(df_erreurs) == 0

        logger.info(
            "Validation types : %s (%d colonne(s) incorrecte(s)).",
            "OK" if est_valide else "ECHEC",
            len(df_erreurs),
        )

        self._rapport["validations"].append(
            {"type": "types", "valide": est_valide, "nb_erreurs": len(df_erreurs)}
        )

        return est_valide, df_erreurs

    def validate_ranges(
        self,
        column_bounds: Dict[str, Tuple[Any, Any]],
    ) -> Tuple[bool, pd.DataFrame]:
        """Vérifie que les valeurs numériques respectent des intervalles.

        Args:
            column_bounds (dict[str, tuple]): Dictionnaire nom_colonne →
                (valeur_min, valeur_max). Exemple :
                {'temperature': (-10, 50), 'rendement_kg': (0, 50000)}.

        Returns:
            tuple[bool, pd.DataFrame]: Tuple contenant :
                - True si toutes les valeurs respectent les bornes.
                - DataFrame des lignes hors-intervalle (vide si OK).
        """
        masque_erreurs = pd.Series(False, index=self.df.index)

        for colonne, (borne_min, borne_max) in column_bounds.items():
            if colonne not in self.df.columns:
                logger.warning(
                    "Colonne '%s' introuvable pour la validation d'intervalle.",
                    colonne,
                )
                continue

            # Détection des valeurs hors-intervalle
            hors_borne = (
                (self.df[colonne] < borne_min) | (self.df[colonne] > borne_max)
            ) & self.df[colonne].notna()

            masque_erreurs = masque_erreurs | hors_borne

            nb_hors_borne = hors_borne.sum()
            if nb_hors_borne > 0:
                logger.warning(
                    "%d valeur(s) hors intervalle [%.2f, %.2f] dans '%s'.",
                    nb_hors_borne,
                    borne_min,
                    borne_max,
                    colonne,
                )

        df_hors_borne = self.df[masque_erreurs].copy()
        est_valide = len(df_hors_borne) == 0

        self._rapport["validations"].append(
            {"type": "ranges", "valide": est_valide, "hors_borne": len(df_hors_borne)}
        )

        return est_valide, df_hors_borne

    def validate_coordinates(
        self,
        lat_col: str,
        lon_col: str,
        region: str = "benin",
    ) -> Tuple[bool, pd.DataFrame]:
        """Vérifie la cohérence géographique des coordonnées GPS.

        Vérifie que les valeurs de latitude et longitude sont dans la
        bounding box de la région spécifiée, et détecte les inversions
        lat/lon accidentelles.

        Args:
            lat_col (str): Nom de la colonne de latitude.
            lon_col (str): Nom de la colonne de longitude.
            region (str): Région de référence pour la bbox. 'benin' utilise
                lat∈[2.5, 12.5], lon∈[-1.5, 4.0]. Par défaut 'benin'.

        Returns:
            tuple[bool, pd.DataFrame]: Tuple contenant :
                - True si toutes les coordonnées sont valides.
                - DataFrame des lignes avec coordonnées invalides.

        Raises:
            KidasValidationError: Si les colonnes lat/lon sont absentes.
        """
        # Vérification de la présence des colonnes
        for colonne in (lat_col, lon_col):
            if colonne not in self.df.columns:
                raise KidasValidationError(
                    f"Colonne de coordonnées '{colonne}' introuvable dans le DataFrame."
                )

        # Définition des bornes selon la région
        if region == "benin":
            lat_min, lat_max = _BENIN_LAT_MIN, _BENIN_LAT_MAX
            lon_min, lon_max = _BENIN_LON_MIN, _BENIN_LON_MAX
        else:
            # Valeurs mondiales par défaut
            lat_min, lat_max = -90.0, 90.0
            lon_min, lon_max = -180.0, 180.0

        # Détection des coordonnées hors bbox
        hors_bbox = (
            (self.df[lat_col] < lat_min) |
            (self.df[lat_col] > lat_max) |
            (self.df[lon_col] < lon_min) |
            (self.df[lon_col] > lon_max)
        ) & self.df[lat_col].notna() & self.df[lon_col].notna()

        df_invalides = self.df[hors_bbox].copy()
        nb_invalides = len(df_invalides)

        if nb_invalides > 0:
            logger.warning(
                "%d coordonnée(s) hors bbox %s détectée(s). "
                "Vérifiez les inversions lat/lon éventuelles.",
                nb_invalides,
                region,
            )

        est_valide = nb_invalides == 0

        self._rapport["validations"].append(
            {
                "type": "coordinates",
                "region": region,
                "valide": est_valide,
                "coords_invalides": nb_invalides,
            }
        )

        return est_valide, df_invalides

    def validate_uniqueness(
        self,
        columns: List[str],
    ) -> Tuple[bool, pd.DataFrame]:
        """Vérifie l'unicité des valeurs sur les colonnes spécifiées.

        Args:
            columns (list[str]): Colonnes dont la combinaison doit être unique
                (équivalent d'une clé primaire composite).

        Returns:
            tuple[bool, pd.DataFrame]: Tuple contenant :
                - True si la combinaison est unique sur toutes les lignes.
                - DataFrame des lignes dupliquées (vide si OK).
        """
        # Détection des lignes dupliquées sur les colonnes spécifiées
        masque_doublons = self.df.duplicated(subset=columns, keep=False)
        df_doublons = self.df[masque_doublons].copy()

        est_valide = len(df_doublons) == 0

        logger.info(
            "Validation unicité sur %s : %s (%d doublon(s)).",
            columns,
            "OK" if est_valide else "ECHEC",
            len(df_doublons),
        )

        self._rapport["validations"].append(
            {"type": "uniqueness", "columns": columns, "valide": est_valide}
        )

        return est_valide, df_doublons

    def validate_referential_integrity(
        self,
        fk_col: str,
        reference_set: Set[Any],
    ) -> Tuple[bool, pd.DataFrame]:
        """Vérifie l'intégrité référentielle d'une clé étrangère.

        Args:
            fk_col (str): Nom de la colonne contenant la clé étrangère.
            reference_set (set): Ensemble des valeurs valides de référence
                (ex: ensemble des market_id existants).

        Returns:
            tuple[bool, pd.DataFrame]: Tuple contenant :
                - True si toutes les valeurs de clé existent dans la référence.
                - DataFrame des lignes avec des références manquantes.

        Raises:
            KidasValidationError: Si la colonne de clé est absente.
        """
        if fk_col not in self.df.columns:
            raise KidasValidationError(
                f"Colonne de clé étrangère '{fk_col}' introuvable."
            )

        # Détection des valeurs absentes de l'ensemble de référence
        masque_manquants = ~self.df[fk_col].isin(reference_set) & self.df[fk_col].notna()
        df_manquants = self.df[masque_manquants].copy()

        est_valide = len(df_manquants) == 0

        logger.info(
            "Validation intégrité référentielle '%s' : %s (%d réf. manquante(s)).",
            fk_col,
            "OK" if est_valide else "ECHEC",
            len(df_manquants),
        )

        self._rapport["validations"].append(
            {
                "type": "referential_integrity",
                "fk_col": fk_col,
                "valide": est_valide,
                "refs_manquantes": len(df_manquants),
            }
        )

        return est_valide, df_manquants

    def compute_quality_score(self) -> dict:
        """Calcule un score de qualité global et par dimension.

        Le score global est la moyenne pondérée de trois dimensions :
        - Complétude (40%) : proportion de valeurs non-null.
        - Cohérence (35%) : absence de doublons.
        - Précision (25%) : score personnalisé (1.0 si non calculé).

        Returns:
            dict: Dictionnaire contenant :
                - 'overall' (float) : score global ∈ [0.0, 1.0].
                - 'completeness' (float) : proportion de cellules non-null.
                - 'consistency' (float) : 1 - taux de doublons.
                - 'accuracy' (float) : 1.0 par défaut (extensible).
                - 'columns' (dict) : score de complétude par colonne.
        """
        nb_lignes, nb_colonnes = self.df.shape

        # Calcul de la complétude : proportion de cellules non-null
        if nb_lignes * nb_colonnes > 0:
            completude = float(
                1 - (self.df.isna().sum().sum() / (nb_lignes * nb_colonnes))
            )
        else:
            completude = 1.0

        # Calcul de la cohérence : absence de doublons
        nb_doublons = self.df.duplicated().sum()
        coherence = float(1 - (nb_doublons / nb_lignes)) if nb_lignes > 0 else 1.0

        # Précision : extensible, 1.0 par défaut
        precision = 1.0

        # Score global pondéré
        score_global = round(
            0.40 * completude + 0.35 * coherence + 0.25 * precision, 4
        )

        # Score de complétude par colonne
        scores_colonnes = {
            col: round(float(1 - self.df[col].isna().mean()), 4)
            for col in self.df.columns
        }

        score = {
            "overall": score_global,
            "completeness": round(completude, 4),
            "consistency": round(coherence, 4),
            "accuracy": round(precision, 4),
            "columns": scores_colonnes,
        }

        logger.info("Score qualité calculé : overall=%.2f", score_global)

        self._rapport["quality_score"] = score
        return score

    def get_validation_report(self) -> dict:
        """Retourne le rapport complet des validations effectuées.

        Returns:
            dict: Rapport structuré contenant l'ensemble des résultats
                de validation et le score de qualité (si calculé).
        """
        return self._rapport.copy()

init ¶

__init__(df: DataFrame) -> None

Initialise le validateur avec le DataFrame à contrôler.

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	Le DataFrame à valider.	required

Raises:

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

Source code in kadi/kidas/validator.py

def __init__(self, df: pd.DataFrame) -> None:
    """Initialise le validateur avec le DataFrame à contrôler.

    Args:
        df (pd.DataFrame): Le DataFrame à valider.

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

    # DataFrame de référence (sans copie : lecture seule)
    self.df: pd.DataFrame = df

    # Rapport de validation initialisé
    self._rapport: Dict = {
        "lignes": len(df),
        "colonnes": len(df.columns),
        "validations": [],
    }

validate_schema ¶

validate_schema(schema: Dict[str, str]) -> Tuple[bool, List[str]]

Vérifie que le DataFrame possède les colonnes définies dans le schéma.

Parameters:

Name	Type	Description	Default
`schema`	`dict[str, str]`	Dictionnaire nom_colonne → type attendu. Les types acceptés sont : 'str', 'int', 'float', 'datetime', 'bool'. Exemple : {'culture': 'str', 'rendement_kg': 'float'}.	required

Returns:

Type	Description
`Tuple[bool, List[str]]`	tuple[bool, list[str]]: Tuple contenant : - True si le schéma est valide, False sinon. - Liste des messages d'erreur (vide si valide).

Source code in kadi/kidas/validator.py

def validate_schema(
    self,
    schema: Dict[str, str],
) -> Tuple[bool, List[str]]:
    """Vérifie que le DataFrame possède les colonnes définies dans le schéma.

    Args:
        schema (dict[str, str]): Dictionnaire nom_colonne → type attendu.
            Les types acceptés sont : 'str', 'int', 'float', 'datetime', 'bool'.
            Exemple : {'culture': 'str', 'rendement_kg': 'float'}.

    Returns:
        tuple[bool, list[str]]: Tuple contenant :
            - True si le schéma est valide, False sinon.
            - Liste des messages d'erreur (vide si valide).
    """
    erreurs: List[str] = []

    for nom_col, type_attendu in schema.items():
        # Vérification de la présence de la colonne
        if nom_col not in self.df.columns:
            erreurs.append(
                f"Colonne manquante : '{nom_col}' (type attendu : '{type_attendu}')."
            )
            continue

        # Vérification du type si spécifié et connu
        if type_attendu in _TYPE_MAP:
            type_pandas = _TYPE_MAP[type_attendu]
            est_correct = pd.api.types.is_dtype_equal(
                self.df[nom_col].dtype,
                type_pandas,
            ) or isinstance(self.df[nom_col].dtype.type, type_pandas) if hasattr(type_pandas, '__mro__') else False

            # Vérification simplifiée selon la catégorie de type
            if type_attendu in ("str", "string") and not pd.api.types.is_string_dtype(self.df[nom_col]):
                if not pd.api.types.is_object_dtype(self.df[nom_col]):
                    erreurs.append(
                        f"Type incorrect pour '{nom_col}' : "
                        f"attendu '{type_attendu}', "
                        f"reçu '{self.df[nom_col].dtype}'."
                    )
            elif type_attendu in ("int", "integer") and not pd.api.types.is_integer_dtype(self.df[nom_col]):
                erreurs.append(
                    f"Type incorrect pour '{nom_col}' : "
                    f"attendu '{type_attendu}', "
                    f"reçu '{self.df[nom_col].dtype}'."
                )
            elif type_attendu == "float" and not pd.api.types.is_float_dtype(self.df[nom_col]):
                erreurs.append(
                    f"Type incorrect pour '{nom_col}' : "
                    f"attendu '{type_attendu}', "
                    f"reçu '{self.df[nom_col].dtype}'."
                )
            elif type_attendu == "datetime" and not pd.api.types.is_datetime64_any_dtype(self.df[nom_col]):
                erreurs.append(
                    f"Type incorrect pour '{nom_col}' : "
                    f"attendu '{type_attendu}', "
                    f"reçu '{self.df[nom_col].dtype}'."
                )

    est_valide = len(erreurs) == 0
    logger.info(
        "Validation schéma : %s (%d erreur(s)).",
        "OK" if est_valide else "ECHEC",
        len(erreurs),
    )

    # Enregistrement dans le rapport
    self._rapport["validations"].append(
        {
            "type": "schema",
            "valide": est_valide,
            "nb_erreurs": len(erreurs),
            "erreurs": erreurs,
        }
    )

    return est_valide, erreurs

validate_types ¶

validate_types(column_dtypes: Dict[str, str]) -> Tuple[bool, pd.DataFrame]

Vérifie la conformité des types pandas pour chaque colonne.

Parameters:

Name	Type	Description	Default
`column_dtypes`	`dict[str, str]`	Dictionnaire nom_colonne → type pandas attendu (ex: 'int64', 'float64', 'object', 'datetime64[ns]').	required

Returns:

Type	Description
`Tuple[bool, DataFrame]`	tuple[bool, pd.DataFrame]: Tuple contenant : - True si tous les types correspondent. - DataFrame des colonnes avec des types incorrects (vide si OK).

Source code in kadi/kidas/validator.py

def validate_types(
    self,
    column_dtypes: Dict[str, str],
) -> Tuple[bool, pd.DataFrame]:
    """Vérifie la conformité des types pandas pour chaque colonne.

    Args:
        column_dtypes (dict[str, str]): Dictionnaire nom_colonne → type
            pandas attendu (ex: 'int64', 'float64', 'object', 'datetime64[ns]').

    Returns:
        tuple[bool, pd.DataFrame]: Tuple contenant :
            - True si tous les types correspondent.
            - DataFrame des colonnes avec des types incorrects (vide si OK).
    """
    lignes_erreurs = []

    for colonne, dtype_attendu in column_dtypes.items():
        if colonne not in self.df.columns:
            lignes_erreurs.append({
                "colonne": colonne,
                "dtype_attendu": dtype_attendu,
                "dtype_reel": "ABSENT",
            })
            continue

        dtype_reel = str(self.df[colonne].dtype)

        # Comparaison des types (flexible pour les variantes d'int/float)
        types_compatibles = False
        if dtype_attendu in dtype_reel or dtype_reel in dtype_attendu:
            types_compatibles = True
        elif "int" in dtype_attendu and pd.api.types.is_integer_dtype(self.df[colonne]):
            types_compatibles = True
        elif "float" in dtype_attendu and pd.api.types.is_float_dtype(self.df[colonne]):
            types_compatibles = True

        if not types_compatibles:
            lignes_erreurs.append({
                "colonne": colonne,
                "dtype_attendu": dtype_attendu,
                "dtype_reel": dtype_reel,
            })

    df_erreurs = pd.DataFrame(lignes_erreurs)
    est_valide = len(df_erreurs) == 0

    logger.info(
        "Validation types : %s (%d colonne(s) incorrecte(s)).",
        "OK" if est_valide else "ECHEC",
        len(df_erreurs),
    )

    self._rapport["validations"].append(
        {"type": "types", "valide": est_valide, "nb_erreurs": len(df_erreurs)}
    )

    return est_valide, df_erreurs

validate_ranges ¶

validate_ranges(column_bounds: Dict[str, Tuple[Any, Any]]) -> Tuple[bool, pd.DataFrame]

Vérifie que les valeurs numériques respectent des intervalles.

Parameters:

Name	Type	Description	Default
`column_bounds`	`dict[str, tuple]`	Dictionnaire nom_colonne → (valeur_min, valeur_max). Exemple : {'temperature': (-10, 50), 'rendement_kg': (0, 50000)}.	required

Returns:

Type	Description
`Tuple[bool, DataFrame]`	tuple[bool, pd.DataFrame]: Tuple contenant : - True si toutes les valeurs respectent les bornes. - DataFrame des lignes hors-intervalle (vide si OK).

Source code in kadi/kidas/validator.py

def validate_ranges(
    self,
    column_bounds: Dict[str, Tuple[Any, Any]],
) -> Tuple[bool, pd.DataFrame]:
    """Vérifie que les valeurs numériques respectent des intervalles.

    Args:
        column_bounds (dict[str, tuple]): Dictionnaire nom_colonne →
            (valeur_min, valeur_max). Exemple :
            {'temperature': (-10, 50), 'rendement_kg': (0, 50000)}.

    Returns:
        tuple[bool, pd.DataFrame]: Tuple contenant :
            - True si toutes les valeurs respectent les bornes.
            - DataFrame des lignes hors-intervalle (vide si OK).
    """
    masque_erreurs = pd.Series(False, index=self.df.index)

    for colonne, (borne_min, borne_max) in column_bounds.items():
        if colonne not in self.df.columns:
            logger.warning(
                "Colonne '%s' introuvable pour la validation d'intervalle.",
                colonne,
            )
            continue

        # Détection des valeurs hors-intervalle
        hors_borne = (
            (self.df[colonne] < borne_min) | (self.df[colonne] > borne_max)
        ) & self.df[colonne].notna()

        masque_erreurs = masque_erreurs | hors_borne

        nb_hors_borne = hors_borne.sum()
        if nb_hors_borne > 0:
            logger.warning(
                "%d valeur(s) hors intervalle [%.2f, %.2f] dans '%s'.",
                nb_hors_borne,
                borne_min,
                borne_max,
                colonne,
            )

    df_hors_borne = self.df[masque_erreurs].copy()
    est_valide = len(df_hors_borne) == 0

    self._rapport["validations"].append(
        {"type": "ranges", "valide": est_valide, "hors_borne": len(df_hors_borne)}
    )

    return est_valide, df_hors_borne

validate_coordinates ¶

validate_coordinates(lat_col: str, lon_col: str, region: str = 'benin') -> Tuple[bool, pd.DataFrame]

Vérifie la cohérence géographique des coordonnées GPS.

Vérifie que les valeurs de latitude et longitude sont dans la bounding box de la région spécifiée, et détecte les inversions lat/lon accidentelles.

Parameters:

Name	Type	Description	Default
`lat_col`	`str`	Nom de la colonne de latitude.	required
`lon_col`	`str`	Nom de la colonne de longitude.	required
`region`	`str`	Région de référence pour la bbox. 'benin' utilise lat∈[2.5, 12.5], lon∈[-1.5, 4.0]. Par défaut 'benin'.	`'benin'`

Returns:

Type	Description
`Tuple[bool, DataFrame]`	tuple[bool, pd.DataFrame]: Tuple contenant : - True si toutes les coordonnées sont valides. - DataFrame des lignes avec coordonnées invalides.

Raises:

Type	Description
`KidasValidationError`	Si les colonnes lat/lon sont absentes.

Source code in kadi/kidas/validator.py

def validate_coordinates(
    self,
    lat_col: str,
    lon_col: str,
    region: str = "benin",
) -> Tuple[bool, pd.DataFrame]:
    """Vérifie la cohérence géographique des coordonnées GPS.

    Vérifie que les valeurs de latitude et longitude sont dans la
    bounding box de la région spécifiée, et détecte les inversions
    lat/lon accidentelles.

    Args:
        lat_col (str): Nom de la colonne de latitude.
        lon_col (str): Nom de la colonne de longitude.
        region (str): Région de référence pour la bbox. 'benin' utilise
            lat∈[2.5, 12.5], lon∈[-1.5, 4.0]. Par défaut 'benin'.

    Returns:
        tuple[bool, pd.DataFrame]: Tuple contenant :
            - True si toutes les coordonnées sont valides.
            - DataFrame des lignes avec coordonnées invalides.

    Raises:
        KidasValidationError: Si les colonnes lat/lon sont absentes.
    """
    # Vérification de la présence des colonnes
    for colonne in (lat_col, lon_col):
        if colonne not in self.df.columns:
            raise KidasValidationError(
                f"Colonne de coordonnées '{colonne}' introuvable dans le DataFrame."
            )

    # Définition des bornes selon la région
    if region == "benin":
        lat_min, lat_max = _BENIN_LAT_MIN, _BENIN_LAT_MAX
        lon_min, lon_max = _BENIN_LON_MIN, _BENIN_LON_MAX
    else:
        # Valeurs mondiales par défaut
        lat_min, lat_max = -90.0, 90.0
        lon_min, lon_max = -180.0, 180.0

    # Détection des coordonnées hors bbox
    hors_bbox = (
        (self.df[lat_col] < lat_min) |
        (self.df[lat_col] > lat_max) |
        (self.df[lon_col] < lon_min) |
        (self.df[lon_col] > lon_max)
    ) & self.df[lat_col].notna() & self.df[lon_col].notna()

    df_invalides = self.df[hors_bbox].copy()
    nb_invalides = len(df_invalides)

    if nb_invalides > 0:
        logger.warning(
            "%d coordonnée(s) hors bbox %s détectée(s). "
            "Vérifiez les inversions lat/lon éventuelles.",
            nb_invalides,
            region,
        )

    est_valide = nb_invalides == 0

    self._rapport["validations"].append(
        {
            "type": "coordinates",
            "region": region,
            "valide": est_valide,
            "coords_invalides": nb_invalides,
        }
    )

    return est_valide, df_invalides

validate_uniqueness ¶

validate_uniqueness(columns: List[str]) -> Tuple[bool, pd.DataFrame]

Vérifie l'unicité des valeurs sur les colonnes spécifiées.

Parameters:

Name	Type	Description	Default
`columns`	`list[str]`	Colonnes dont la combinaison doit être unique (équivalent d'une clé primaire composite).	required

Returns:

Type	Description
`Tuple[bool, DataFrame]`	tuple[bool, pd.DataFrame]: Tuple contenant : - True si la combinaison est unique sur toutes les lignes. - DataFrame des lignes dupliquées (vide si OK).

Source code in kadi/kidas/validator.py

def validate_uniqueness(
    self,
    columns: List[str],
) -> Tuple[bool, pd.DataFrame]:
    """Vérifie l'unicité des valeurs sur les colonnes spécifiées.

    Args:
        columns (list[str]): Colonnes dont la combinaison doit être unique
            (équivalent d'une clé primaire composite).

    Returns:
        tuple[bool, pd.DataFrame]: Tuple contenant :
            - True si la combinaison est unique sur toutes les lignes.
            - DataFrame des lignes dupliquées (vide si OK).
    """
    # Détection des lignes dupliquées sur les colonnes spécifiées
    masque_doublons = self.df.duplicated(subset=columns, keep=False)
    df_doublons = self.df[masque_doublons].copy()

    est_valide = len(df_doublons) == 0

    logger.info(
        "Validation unicité sur %s : %s (%d doublon(s)).",
        columns,
        "OK" if est_valide else "ECHEC",
        len(df_doublons),
    )

    self._rapport["validations"].append(
        {"type": "uniqueness", "columns": columns, "valide": est_valide}
    )

    return est_valide, df_doublons

validate_referential_integrity ¶

validate_referential_integrity(fk_col: str, reference_set: Set[Any]) -> Tuple[bool, pd.DataFrame]

Vérifie l'intégrité référentielle d'une clé étrangère.

Parameters:

Name	Type	Description	Default
`fk_col`	`str`	Nom de la colonne contenant la clé étrangère.	required
`reference_set`	`set`	Ensemble des valeurs valides de référence (ex: ensemble des market_id existants).	required

Returns:

Type	Description
`Tuple[bool, DataFrame]`	tuple[bool, pd.DataFrame]: Tuple contenant : - True si toutes les valeurs de clé existent dans la référence. - DataFrame des lignes avec des références manquantes.

Raises:

Type	Description
`KidasValidationError`	Si la colonne de clé est absente.

Source code in kadi/kidas/validator.py

def validate_referential_integrity(
    self,
    fk_col: str,
    reference_set: Set[Any],
) -> Tuple[bool, pd.DataFrame]:
    """Vérifie l'intégrité référentielle d'une clé étrangère.

    Args:
        fk_col (str): Nom de la colonne contenant la clé étrangère.
        reference_set (set): Ensemble des valeurs valides de référence
            (ex: ensemble des market_id existants).

    Returns:
        tuple[bool, pd.DataFrame]: Tuple contenant :
            - True si toutes les valeurs de clé existent dans la référence.
            - DataFrame des lignes avec des références manquantes.

    Raises:
        KidasValidationError: Si la colonne de clé est absente.
    """
    if fk_col not in self.df.columns:
        raise KidasValidationError(
            f"Colonne de clé étrangère '{fk_col}' introuvable."
        )

    # Détection des valeurs absentes de l'ensemble de référence
    masque_manquants = ~self.df[fk_col].isin(reference_set) & self.df[fk_col].notna()
    df_manquants = self.df[masque_manquants].copy()

    est_valide = len(df_manquants) == 0

    logger.info(
        "Validation intégrité référentielle '%s' : %s (%d réf. manquante(s)).",
        fk_col,
        "OK" if est_valide else "ECHEC",
        len(df_manquants),
    )

    self._rapport["validations"].append(
        {
            "type": "referential_integrity",
            "fk_col": fk_col,
            "valide": est_valide,
            "refs_manquantes": len(df_manquants),
        }
    )

    return est_valide, df_manquants

compute_quality_score ¶

compute_quality_score() -> dict

Calcule un score de qualité global et par dimension.

Le score global est la moyenne pondérée de trois dimensions : - Complétude (40%) : proportion de valeurs non-null. - Cohérence (35%) : absence de doublons. - Précision (25%) : score personnalisé (1.0 si non calculé).

Returns:

Name	Type	Description
`dict`	`dict`	Dictionnaire contenant : - 'overall' (float) : score global ∈ [0.0, 1.0]. - 'completeness' (float) : proportion de cellules non-null. - 'consistency' (float) : 1 - taux de doublons. - 'accuracy' (float) : 1.0 par défaut (extensible). - 'columns' (dict) : score de complétude par colonne.

Source code in kadi/kidas/validator.py

def compute_quality_score(self) -> dict:
    """Calcule un score de qualité global et par dimension.

    Le score global est la moyenne pondérée de trois dimensions :
    - Complétude (40%) : proportion de valeurs non-null.
    - Cohérence (35%) : absence de doublons.
    - Précision (25%) : score personnalisé (1.0 si non calculé).

    Returns:
        dict: Dictionnaire contenant :
            - 'overall' (float) : score global ∈ [0.0, 1.0].
            - 'completeness' (float) : proportion de cellules non-null.
            - 'consistency' (float) : 1 - taux de doublons.
            - 'accuracy' (float) : 1.0 par défaut (extensible).
            - 'columns' (dict) : score de complétude par colonne.
    """
    nb_lignes, nb_colonnes = self.df.shape

    # Calcul de la complétude : proportion de cellules non-null
    if nb_lignes * nb_colonnes > 0:
        completude = float(
            1 - (self.df.isna().sum().sum() / (nb_lignes * nb_colonnes))
        )
    else:
        completude = 1.0

    # Calcul de la cohérence : absence de doublons
    nb_doublons = self.df.duplicated().sum()
    coherence = float(1 - (nb_doublons / nb_lignes)) if nb_lignes > 0 else 1.0

    # Précision : extensible, 1.0 par défaut
    precision = 1.0

    # Score global pondéré
    score_global = round(
        0.40 * completude + 0.35 * coherence + 0.25 * precision, 4
    )

    # Score de complétude par colonne
    scores_colonnes = {
        col: round(float(1 - self.df[col].isna().mean()), 4)
        for col in self.df.columns
    }

    score = {
        "overall": score_global,
        "completeness": round(completude, 4),
        "consistency": round(coherence, 4),
        "accuracy": round(precision, 4),
        "columns": scores_colonnes,
    }

    logger.info("Score qualité calculé : overall=%.2f", score_global)

    self._rapport["quality_score"] = score
    return score

get_validation_report ¶

get_validation_report() -> dict

Retourne le rapport complet des validations effectuées.

Returns:

Name	Type	Description
`dict`	`dict`	Rapport structuré contenant l'ensemble des résultats de validation et le score de qualité (si calculé).

Source code in kadi/kidas/validator.py

def get_validation_report(self) -> dict:
    """Retourne le rapport complet des validations effectuées.

    Returns:
        dict: Rapport structuré contenant l'ensemble des résultats
            de validation et le score de qualité (si calculé).
    """
    return self._rapport.copy()

Validation (kadi.kidas.validator)¶

Principe¶

Méthodes¶

validate_schema(schema)¶

check_ranges(rules)¶

check_referential(column, allowed_values)¶

Rapport de validation¶

Exemple complet¶

DataValidator ¶

__init__ ¶

validate_schema ¶

validate_types ¶

validate_ranges ¶

validate_coordinates ¶

validate_uniqueness ¶

validate_referential_integrity ¶

compute_quality_score ¶

get_validation_report ¶

Validation (`kadi.kidas.validator`)¶

`validate_schema(schema)`¶

`check_ranges(rules)`¶

`check_referential(column, allowed_values)`¶

init ¶