Validator (`kadi.kidas.validator`)¶

Validator est la classe de validation qualité des données agricoles tabulaires de KadiPy. Elle regroupe six vérifications ciblées : schéma, types pandas, intervalles numériques, coordonnées GPS, unicité et intégrité référentielle. Elle calcule également un score de qualité global sur trois dimensions : complétude, cohérence et précision.

Chaque méthode de validation retourne un tuple (bool, résultat), ce qui permet d'inspecter le résultat ou d'interrompre le traitement immédiatement en cas d'anomalie. Toutes les vérifications effectuées sont enregistrées dans un rapport interne accessible via report().

Initialisation¶

from kadi.kidas import Validator

validator = Validator(df)

Validator reçoit un pandas.DataFrame. Le DataFrame n'est pas copié : il est conservé en référence et n'est jamais modifié par les méthodes de validation.

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

Méthodes de validation¶

`check_schema(schema)`¶

Vérifie que le DataFrame contient les colonnes définies dans le schéma et que leurs types correspondent.

valide, erreurs = validator.check_schema({
    "culture":       "str",
    "rendement_kg":  "float",
    "date_recolte":  "datetime",
})

if not valide:
    for msg in erreurs:
        print(msg)

Paramètre	Type	Description
`schema`	`dict[str, str]`	Dictionnaire `nom_colonne` → `type attendu`

Types acceptés :

Valeur	Interprétation
`'str'` ou `'string'`	Colonne de type texte (`object`)
`'int'` ou `'integer'`	Colonne entière
`'float'`	Colonne flottante
`'datetime'`	Colonne de type `datetime64`
`'bool'`	Colonne booléenne

Retour : tuple[bool, list[str]]

True si le schéma est valide, False sinon.
Liste des messages d'erreur (vide si valide).

`check_types(dtypes)`¶

Vérifie la conformité des types pandas réels pour chaque colonne, avec une comparaison flexible (ex: int64 et int32 sont tous deux acceptés pour 'int').

valide, df_erreurs = validator.check_types({
    "culture":      "object",
    "rendement_kg": "float64",
    "date_recolte": "datetime64[ns]",
})

if not valide:
    print(df_erreurs)

Paramètre	Type	Description
`dtypes`	`dict[str, str]`	Dictionnaire `nom_colonne` → dtype pandas attendu

Retour : tuple[bool, pd.DataFrame]

True si tous les types correspondent.
DataFrame des colonnes non conformes (vide si OK), avec les colonnes colonne, dtype_attendu, dtype_reel.

`check_ranges(bounds)`¶

Vérifie que les valeurs numériques des colonnes spécifiées sont comprises dans les intervalles indiqués. Les valeurs NaN sont ignorées.

valide, df_hors_borne = validator.check_ranges({
    "temperature":   (-10, 50),
    "rendement_kg":  (0, 50000),
    "prix_xof_kg":   (0, 5000),
})

if not valide:
    print(f"{len(df_hors_borne)} ligne(s) hors intervalle.")
    print(df_hors_borne)

Paramètre	Type	Description
`bounds`	`dict[str, tuple]`	Dictionnaire `nom_colonne` → `(min, max)`

Les bornes sont inclusives. Une colonne absente du DataFrame génère un avertissement dans les logs et est ignorée sans erreur.

Retour : tuple[bool, pd.DataFrame]

True si toutes les valeurs respectent les bornes.
DataFrame des lignes hors-intervalle, toutes colonnes incluses (vide si OK).

`check_coords(lat, lon, region='benin')`¶

Vérifie que les coordonnées GPS sont dans la bounding box de la région spécifiée. Détecte également les inversions lat/lon accidentelles.

valide, df_invalides = validator.check_coords(
    lat="latitude",
    lon="longitude",
)

# Avec la région par défaut ('benin') :
# latitude  ∈ [2.5,  12.5]
# longitude ∈ [-1.5,  4.0]

if not valide:
    print(f"{len(df_invalides)} coordonnée(s) invalide(s).")

Lorsque region est différente de 'benin', les bornes mondiales [-90, 90] pour la latitude et [-180, 180] pour la longitude sont utilisées.

Paramètre	Type	Défaut	Description
`lat`	`str`	-	Nom de la colonne de latitude
`lon`	`str`	-	Nom de la colonne de longitude
`region`	`str`	`'benin'`	Région de référence pour la bbox

Retour : tuple[bool, pd.DataFrame]

True si toutes les coordonnées sont dans la bbox.
DataFrame des lignes avec des coordonnées hors-bbox (vide si OK).

Exception : ValidationError si l'une des colonnes lat ou lon est absente du DataFrame.

`check_unique(cols)`¶

Vérifie que la combinaison des colonnes spécifiées est unique sur l'ensemble des lignes (équivalent d'une clé primaire composite).

valide, df_doublons = validator.check_unique(
    cols=["culture", "marche", "date_recolte"]
)

if not valide:
    print(f"{len(df_doublons)} ligne(s) en doublon.")
    print(df_doublons)

Paramètre	Type	Description
`cols`	`list[str]`	Colonnes dont la combinaison doit être unique

Retour : tuple[bool, pd.DataFrame]

True si la combinaison est unique.
DataFrame de toutes les occurrences impliquées dans un doublon (vide si OK).

`check_fk(fk_col, ref)`¶

Vérifie l'intégrité référentielle d'une colonne : chaque valeur non-null doit figurer dans l'ensemble de référence fourni.

marches_valides = {"Cotonou", "Porto-Novo", "Parakou", "Bohicon"}

valide, df_manquants = validator.check_fk(
    fk_col="marche",
    ref=marches_valides,
)

if not valide:
    print(f"{len(df_manquants)} référence(s) inconnue(s).")

Paramètre	Type	Description
`fk_col`	`str`	Nom de la colonne contenant la clé étrangère
`ref`	`set`	Ensemble des valeurs valides attendues

Retour : tuple[bool, pd.DataFrame]

True si toutes les valeurs sont dans l'ensemble de référence.
DataFrame des lignes avec des valeurs absentes de la référence (vide si OK).

Exception : ValidationError si la colonne fk_col est absente.

`quality_score()`¶

Calcule un score de qualité global et par dimension pour le DataFrame.

score = validator.quality_score()

print(score["overall"])       # ex: 0.8742
print(score["completeness"])  # proportion de cellules non-null
print(score["consistency"])   # 1 - taux de doublons
print(score["accuracy"])      # 1.0 (extensible)
print(score["columns"])       # score de complétude par colonne

Le score global est une moyenne pondérée :

Dimension	Poids	Calcul
Complétude	40 %	Proportion de cellules non-null
Cohérence	35 %	`1 - (nb_doublons / nb_lignes)`
Précision	25 %	`1.0` par défaut (extensible)

Retour : dict avec les clés :

Clé	Type	Description
`overall`	`float`	Score global ∈ [0.0, 1.0]
`completeness`	`float`	Score de complétude
`consistency`	`float`	Score de cohérence
`accuracy`	`float`	Score de précision
`columns`	`dict[str, float]`	Complétude par colonne

`report()`¶

Retourne le rapport complet des validations effectuées depuis l'initialisation du Validator.

validator.check_schema({"culture": "str", "rendement_kg": "float"})
validator.check_ranges({"rendement_kg": (0, 50000)})
validator.quality_score()

rapport = validator.report()
# {
#   "lignes": 1500,
#   "colonnes": 8,
#   "validations": [
#     {"type": "schema",  "valide": True,  "nb_erreurs": 0, "erreurs": []},
#     {"type": "ranges",  "valide": False, "hors_borne": 3},
#   ],
#   "quality_score": {
#     "overall": 0.9245,
#     ...
#   }
# }

Clé	Type	Description
`lignes`	`int`	Nombre de lignes du DataFrame à l'initialisation
`colonnes`	`int`	Nombre de colonnes à l'initialisation
`validations`	`list`	Résultats de chaque validation effectuée
`quality_score`	`dict`	Présent uniquement si `quality_score()` a été appelé

Retour : dict (copie du rapport interne).

Exemple complet¶

import kadi as kd
from kadi.kidas import Validator

df = kd.read_csv("enquete_prix_2024.csv")
validator = Validator(df)

schema = {
    "culture":      "str",
    "marche":       "str",
    "date_recolte": "datetime",
    "rendement_kg": "float",
    "prix_xof_kg":  "float",
    "latitude":     "float",
    "longitude":    "float",
}

valide, erreurs = validator.check_schema(schema)
if not valide:
    for msg in erreurs:
        print(msg)

valide, df_hors_borne = validator.check_ranges({
    "rendement_kg": (0, 50000),
    "prix_xof_kg":  (0, 5000),
})

valide, df_coords = validator.check_coords(lat="latitude", lon="longitude")

score = validator.quality_score()
print(f"Score qualité : {score['overall']:.2%}")

rapport = validator.report()

Utilisation dans un Pipeline¶

Validator est utilisé en interne par Pipeline lorsqu'on ajoute une étape de validation via add_step ou validate.

from kadi.kidas import Pipeline

df, rapport = (
    Pipeline()
    .add_source("enquete_prix_2024.csv")
    .add_step("check_schema", schema={"culture": "str", "rendement_kg": "float"})
    .add_step("check_ranges", bounds={"rendement_kg": (0, 50000)})
    .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
`validate_schema()`	`check_schema()`
`validate_types()`	`check_types()`
`validate_ranges()`	`check_ranges()`
`validate_coordinates()`	`check_coords()`
`validate_uniqueness()`	`check_unique()`
`validate_referential_integrity()`	`check_fk()`
`compute_quality_score()`	`quality_score()`
`get_validation_report()`	`report()`

De même, le nom de classe DataValidator est un alias de Validator.

Validator ¶

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. _report (dict): Journal des résultats de validation.

Exemple

validator = Validator(df) valide, erreurs = validator.check_schema({ ... 'culture': 'str', ... 'rendement_kg': 'float', ... }) score = validator.quality_score() print(score['overall']) 0.87

Source code in kadi/kidas/validator.py

class Validator:
    """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.
        _report (dict): Journal des résultats de validation.

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

    # Table de rétrocompatibilité des méthodes d'instance.
    _METHODES_DEPRECATED = {
        "validate_schema":               "check_schema",
        "validate_types":                "check_types",
        "validate_ranges":               "check_ranges",
        "validate_coordinates":          "check_coords",
        "validate_uniqueness":           "check_unique",
        "validate_referential_integrity":"check_fk",
        "compute_quality_score":         "quality_score",
        "get_validation_report":         "report",
    }

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

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

        Raises:
            ValidationError: Si l'argument fourni n'est pas un DataFrame.
        """
        # Vérification du type d'entrée
        if not isinstance(df, pd.DataFrame):
            raise ValidationError(
                f"Validator 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._report: Dict = {
            "lignes": len(df),
            "colonnes": len(df.columns),
            "validations": [],
        }

    def check_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}'."
                    )
                elif type_attendu == "bool" and not pd.api.types.is_bool_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._report["validations"].append(
            {
                "type": "schema",
                "valide": est_valide,
                "nb_erreurs": len(erreurs),
                "erreurs": erreurs,
            }
        )

        return est_valide, erreurs

    def check_types(
        self,
        dtypes: Optional[Dict[str, str]] = None,
        **kwargs,
    ) -> Tuple[bool, pd.DataFrame]:
        """Vérifie la conformité des types pandas pour chaque colonne.

        Args:
            dtypes (dict[str, str]): Dictionnaire nom_colonne → type
                pandas attendu (ex: 'int64', 'float64', 'object', 'datetime64[ns]').
                Alias accepté : column_dtypes= (rétrocompatibilité).

        Returns:
            tuple[bool, pd.DataFrame]: Tuple contenant :
                - True si tous les types correspondent.
                - DataFrame des colonnes avec des types incorrects (vide si OK).
        """
        # Rétrocompatibilité : ancien paramètre accepté
        if dtypes is None and "column_dtypes" in kwargs:
            dtypes = kwargs.pop("column_dtypes")
        if dtypes is None:
            dtypes = {}

        lignes_erreurs = []

        for colonne, dtype_attendu in 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 ("str" in dtype_attendu or "object" in dtype_attendu or "string" in dtype_attendu) and (
                pd.api.types.is_string_dtype(self.df[colonne]) or pd.api.types.is_object_dtype(self.df[colonne])
            ):
                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._report["validations"].append(
            {"type": "types", "valide": est_valide, "nb_erreurs": len(df_erreurs)}
        )

        return est_valide, df_erreurs

    def check_ranges(
        self,
        bounds: Optional[Dict[str, Tuple[Any, Any]]] = None,
        **kwargs,
    ) -> Tuple[bool, pd.DataFrame]:
        """Vérifie que les valeurs numériques respectent des intervalles.

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

        Returns:
            tuple[bool, pd.DataFrame]: Tuple contenant :
                - True si toutes les valeurs respectent les bornes.
                - DataFrame des lignes hors-intervalle (vide si OK).
        """
        # Rétrocompatibilité : ancien paramètre accepté
        if bounds is None and "column_bounds" in kwargs:
            bounds = kwargs.pop("column_bounds")
        if bounds is None:
            bounds = {}

        masque_erreurs = pd.Series(False, index=self.df.index)

        for colonne, (borne_min, borne_max) in 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._report["validations"].append(
            {"type": "ranges", "valide": est_valide, "hors_borne": len(df_hors_borne)}
        )

        return est_valide, df_hors_borne

    def check_coords(
        self,
        lat: Optional[str] = None,
        lon: Optional[str] = None,
        region: str = "benin",
        **kwargs,
    ) -> 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 (str): Nom de la colonne de latitude.
                Alias accepté : lat_col= (rétrocompatibilité).
            lon (str): Nom de la colonne de longitude.
                Alias accepté : lon_col= (rétrocompatibilité).
            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:
            ValidationError: Si les colonnes lat/lon sont absentes.
        """
        # Rétrocompatibilité : anciens paramètres acceptés
        if lat is None and "lat_col" in kwargs:
            lat = kwargs.pop("lat_col")
        if lon is None and "lon_col" in kwargs:
            lon = kwargs.pop("lon_col")

        # Vérification de la présence des colonnes
        for colonne in (lat, lon):
            if colonne not in self.df.columns:
                raise ValidationError(
                    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] < lat_min) |
            (self.df[lat] > lat_max) |
            (self.df[lon] < lon_min) |
            (self.df[lon] > lon_max)
        ) & self.df[lat].notna() & self.df[lon].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._report["validations"].append(
            {
                "type": "coordinates",
                "region": region,
                "valide": est_valide,
                "coords_invalides": nb_invalides,
            }
        )

        return est_valide, df_invalides

    def check_unique(
        self,
        cols: Optional[List[str]] = None,
        **kwargs,
    ) -> Tuple[bool, pd.DataFrame]:
        """Vérifie l'unicité des valeurs sur les colonnes spécifiées.

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

        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).
        """
        # Rétrocompatibilité : ancien paramètre accepté
        if cols is None and "columns" in kwargs:
            cols = kwargs.pop("columns")
        if cols is None:
            cols = []

        # Détection des lignes dupliquées sur les colonnes spécifiées
        masque_doublons = self.df.duplicated(subset=cols, 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)).",
            cols,
            "OK" if est_valide else "ECHEC",
            len(df_doublons),
        )

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

        return est_valide, df_doublons

    def check_fk(
        self,
        fk_col: Optional[str] = None,
        ref: Optional[Set[Any]] = None,
        **kwargs,
    ) -> 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.
            ref (set): Ensemble des valeurs valides de référence
                (ex: ensemble des market_id existants).
                Alias accepté : reference_set= (rétrocompatibilité).

        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:
            ValidationError: Si la colonne de clé est absente.
        """
        # Rétrocompatibilité : ancien paramètre accepté
        if ref is None and "reference_set" in kwargs:
            ref = kwargs.pop("reference_set")

        if fk_col not in self.df.columns:
            raise ValidationError(
                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(ref) & 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._report["validations"].append(
            {
                "type": "referential_integrity",
                "fk_col": fk_col,
                "valide": est_valide,
                "refs_manquantes": len(df_manquants),
            }
        )

        return est_valide, df_manquants

    def 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._report["quality_score"] = score
        return score

    def 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._report.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 Validator._METHODES_DEPRECATED:
            # Récupération du nouveau nom
            nouveau_nom = Validator._METHODES_DEPRECATED[name]
            warnings.warn(
                f"Validator.{name}() est obsolète et sera supprimé dans KadiPy v2.0. "
                f"Utilisez Validator.{nouveau_nom}() à la place.",
                category=DeprecationWarning,
                stacklevel=2,
            )
            return getattr(self, nouveau_nom)
        raise AttributeError(
            f"'Validator' n'a pas de méthode '{name}'."
        )

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
`ValidationError`	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:
        ValidationError: Si l'argument fourni n'est pas un DataFrame.
    """
    # Vérification du type d'entrée
    if not isinstance(df, pd.DataFrame):
        raise ValidationError(
            f"Validator 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._report: Dict = {
        "lignes": len(df),
        "colonnes": len(df.columns),
        "validations": [],
    }

check_schema ¶

check_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 check_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}'."
                )
            elif type_attendu == "bool" and not pd.api.types.is_bool_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._report["validations"].append(
        {
            "type": "schema",
            "valide": est_valide,
            "nb_erreurs": len(erreurs),
            "erreurs": erreurs,
        }
    )

    return est_valide, erreurs

check_types ¶

check_types(dtypes: Optional[Dict[str, str]] = None, **kwargs) -> Tuple[bool, pd.DataFrame]

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

Parameters:

Name	Type	Description	Default
`dtypes`	`dict[str, str]`	Dictionnaire nom_colonne → type pandas attendu (ex: 'int64', 'float64', 'object', 'datetime64[ns]'). Alias accepté : column_dtypes= (rétrocompatibilité).	`None`

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 check_types(
    self,
    dtypes: Optional[Dict[str, str]] = None,
    **kwargs,
) -> Tuple[bool, pd.DataFrame]:
    """Vérifie la conformité des types pandas pour chaque colonne.

    Args:
        dtypes (dict[str, str]): Dictionnaire nom_colonne → type
            pandas attendu (ex: 'int64', 'float64', 'object', 'datetime64[ns]').
            Alias accepté : column_dtypes= (rétrocompatibilité).

    Returns:
        tuple[bool, pd.DataFrame]: Tuple contenant :
            - True si tous les types correspondent.
            - DataFrame des colonnes avec des types incorrects (vide si OK).
    """
    # Rétrocompatibilité : ancien paramètre accepté
    if dtypes is None and "column_dtypes" in kwargs:
        dtypes = kwargs.pop("column_dtypes")
    if dtypes is None:
        dtypes = {}

    lignes_erreurs = []

    for colonne, dtype_attendu in 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 ("str" in dtype_attendu or "object" in dtype_attendu or "string" in dtype_attendu) and (
            pd.api.types.is_string_dtype(self.df[colonne]) or pd.api.types.is_object_dtype(self.df[colonne])
        ):
            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._report["validations"].append(
        {"type": "types", "valide": est_valide, "nb_erreurs": len(df_erreurs)}
    )

    return est_valide, df_erreurs

check_ranges ¶

check_ranges(bounds: Optional[Dict[str, Tuple[Any, Any]]] = None, **kwargs) -> Tuple[bool, pd.DataFrame]

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

Parameters:

Name	Type	Description	Default
`bounds`	`dict[str, tuple]`	Dictionnaire nom_colonne → (valeur_min, valeur_max). Exemple : {'temperature': (-10, 50), 'rendement_kg': (0, 50000)}. Alias accepté : column_bounds= (rétrocompatibilité).	`None`

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 check_ranges(
    self,
    bounds: Optional[Dict[str, Tuple[Any, Any]]] = None,
    **kwargs,
) -> Tuple[bool, pd.DataFrame]:
    """Vérifie que les valeurs numériques respectent des intervalles.

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

    Returns:
        tuple[bool, pd.DataFrame]: Tuple contenant :
            - True si toutes les valeurs respectent les bornes.
            - DataFrame des lignes hors-intervalle (vide si OK).
    """
    # Rétrocompatibilité : ancien paramètre accepté
    if bounds is None and "column_bounds" in kwargs:
        bounds = kwargs.pop("column_bounds")
    if bounds is None:
        bounds = {}

    masque_erreurs = pd.Series(False, index=self.df.index)

    for colonne, (borne_min, borne_max) in 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._report["validations"].append(
        {"type": "ranges", "valide": est_valide, "hors_borne": len(df_hors_borne)}
    )

    return est_valide, df_hors_borne

check_coords ¶

check_coords(lat: Optional[str] = None, lon: Optional[str] = None, region: str = 'benin', **kwargs) -> 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`	`str`	Nom de la colonne de latitude. Alias accepté : lat_col= (rétrocompatibilité).	`None`
`lon`	`str`	Nom de la colonne de longitude. Alias accepté : lon_col= (rétrocompatibilité).	`None`
`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
`ValidationError`	Si les colonnes lat/lon sont absentes.

Source code in kadi/kidas/validator.py

def check_coords(
    self,
    lat: Optional[str] = None,
    lon: Optional[str] = None,
    region: str = "benin",
    **kwargs,
) -> 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 (str): Nom de la colonne de latitude.
            Alias accepté : lat_col= (rétrocompatibilité).
        lon (str): Nom de la colonne de longitude.
            Alias accepté : lon_col= (rétrocompatibilité).
        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:
        ValidationError: Si les colonnes lat/lon sont absentes.
    """
    # Rétrocompatibilité : anciens paramètres acceptés
    if lat is None and "lat_col" in kwargs:
        lat = kwargs.pop("lat_col")
    if lon is None and "lon_col" in kwargs:
        lon = kwargs.pop("lon_col")

    # Vérification de la présence des colonnes
    for colonne in (lat, lon):
        if colonne not in self.df.columns:
            raise ValidationError(
                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] < lat_min) |
        (self.df[lat] > lat_max) |
        (self.df[lon] < lon_min) |
        (self.df[lon] > lon_max)
    ) & self.df[lat].notna() & self.df[lon].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._report["validations"].append(
        {
            "type": "coordinates",
            "region": region,
            "valide": est_valide,
            "coords_invalides": nb_invalides,
        }
    )

    return est_valide, df_invalides

check_unique ¶

check_unique(cols: Optional[List[str]] = None, **kwargs) -> Tuple[bool, pd.DataFrame]

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

Parameters:

Name	Type	Description	Default
`cols`	`list[str]`	Colonnes dont la combinaison doit être unique (équivalent d'une clé primaire composite). Alias accepté : columns= (rétrocompatibilité).	`None`

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 check_unique(
    self,
    cols: Optional[List[str]] = None,
    **kwargs,
) -> Tuple[bool, pd.DataFrame]:
    """Vérifie l'unicité des valeurs sur les colonnes spécifiées.

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

    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).
    """
    # Rétrocompatibilité : ancien paramètre accepté
    if cols is None and "columns" in kwargs:
        cols = kwargs.pop("columns")
    if cols is None:
        cols = []

    # Détection des lignes dupliquées sur les colonnes spécifiées
    masque_doublons = self.df.duplicated(subset=cols, 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)).",
        cols,
        "OK" if est_valide else "ECHEC",
        len(df_doublons),
    )

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

    return est_valide, df_doublons

check_fk ¶

check_fk(fk_col: Optional[str] = None, ref: Optional[Set[Any]] = None, **kwargs) -> 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.	`None`
`ref`	`set`	Ensemble des valeurs valides de référence (ex: ensemble des market_id existants). Alias accepté : reference_set= (rétrocompatibilité).	`None`

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
`ValidationError`	Si la colonne de clé est absente.

Source code in kadi/kidas/validator.py

def check_fk(
    self,
    fk_col: Optional[str] = None,
    ref: Optional[Set[Any]] = None,
    **kwargs,
) -> 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.
        ref (set): Ensemble des valeurs valides de référence
            (ex: ensemble des market_id existants).
            Alias accepté : reference_set= (rétrocompatibilité).

    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:
        ValidationError: Si la colonne de clé est absente.
    """
    # Rétrocompatibilité : ancien paramètre accepté
    if ref is None and "reference_set" in kwargs:
        ref = kwargs.pop("reference_set")

    if fk_col not in self.df.columns:
        raise ValidationError(
            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(ref) & 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._report["validations"].append(
        {
            "type": "referential_integrity",
            "fk_col": fk_col,
            "valide": est_valide,
            "refs_manquantes": len(df_manquants),
        }
    )

    return est_valide, df_manquants

quality_score ¶

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 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._report["quality_score"] = score
    return score

report ¶

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 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._report.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/validator.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 Validator._METHODES_DEPRECATED:
        # Récupération du nouveau nom
        nouveau_nom = Validator._METHODES_DEPRECATED[name]
        warnings.warn(
            f"Validator.{name}() est obsolète et sera supprimé dans KadiPy v2.0. "
            f"Utilisez Validator.{nouveau_nom}() à la place.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return getattr(self, nouveau_nom)
    raise AttributeError(
        f"'Validator' n'a pas de méthode '{name}'."
    )

Validator (kadi.kidas.validator)¶

Initialisation¶

Méthodes de validation¶

check_schema(schema)¶

check_types(dtypes)¶

check_ranges(bounds)¶

check_coords(lat, lon, region='benin')¶

check_unique(cols)¶

check_fk(fk_col, ref)¶

quality_score()¶

report()¶

Exemple complet¶

Utilisation dans un Pipeline¶

Rétrocompatibilité¶

Validator ¶

__init__ ¶

check_schema ¶

check_types ¶

check_ranges ¶

check_coords ¶

check_unique ¶

check_fk ¶

quality_score ¶

report ¶

__getattr__ ¶

Validator (`kadi.kidas.validator`)¶

`check_schema(schema)`¶

`check_types(dtypes)`¶

`check_ranges(bounds)`¶

`check_coords(lat, lon, region='benin')`¶

`check_unique(cols)`¶

`check_fk(fk_col, ref)`¶

`quality_score()`¶

`report()`¶

init ¶

getattr ¶