Validator (`kadi.kidas.validator`)¶

Validator is the data quality validation class for tabular agricultural data in KadiPy. It gathers six targeted checks: schema, pandas data types, numeric ranges, GPS coordinates, uniqueness, and referential integrity. It also calculates an overall data quality score across three dimensions: completeness, consistency, and accuracy.

Each validation method returns a tuple (bool, result), allowing you to inspect results or halt processing immediately if an anomaly is detected. All performed checks are recorded in an internal report accessible via report().

Initialization¶

from kadi.kidas import Validator

validator = Validator(df)

Validator receives a pandas.DataFrame. The DataFrame is not copied: it is held by reference and is never mutated by validation methods.

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

Validation Methods¶

`check_schema(schema)`¶

Verifies that the DataFrame contains the columns specified in the schema and that their data types match expectations.

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

if not is_valid:
    for msg in errors:
        print(msg)

Parameter	Type	Description
`schema`	`dict[str, str]`	Dictionary mapping `column_name` -> `expected_type`

Accepted Types:

Value	Interpretation
`'str'` or `'string'`	Text column (`object` dtype)
`'int'` or `'integer'`	Integer column
`'float'`	Floating point column
`'datetime'`	`datetime64` column
`'bool'`	Boolean column

Returns: tuple[bool, list[str]]

True if the schema is valid, False otherwise.
List of error messages (empty if valid).

`check_types(dtypes)`¶

Verifies compliance of actual pandas data types for each column, using flexible type comparison (for example: int64 and int32 are both accepted for 'int').

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

if not is_valid:
    print(df_errors)

Parameter	Type	Description
`dtypes`	`dict[str, str]`	Dictionary mapping `column_name` -> expected pandas dtype

Returns: tuple[bool, pd.DataFrame]

True if all types match.
DataFrame of non-conforming columns (empty if OK), containing columns colonne, dtype_attendu, and dtype_reel.

`check_ranges(bounds)`¶

Verifies that numeric values in specified columns fall within defined intervals. NaN values are ignored.

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

if not is_valid:
    print(f"{len(df_out_of_bounds)} row(s) out of bounds.")
    print(df_out_of_bounds)

Parameter	Type	Description
`bounds`	`dict[str, tuple]`	Dictionary mapping `column_name` -> `(min, max)`

Bounds are inclusive. A column missing from the DataFrame logs a warning and is skipped without raising an error.

Returns: tuple[bool, pd.DataFrame]

True if all values satisfy the bounds.
DataFrame of out-of-bounds rows, including all columns (empty if OK).

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

Verifies that GPS coordinates lie within the bounding box of the specified region. Also detects accidental lat/lon inversions.

is_valid, df_invalid = validator.check_coords(
    lat="latitude",
    lon="longitude",
)

# With the default region ('benin'):
# latitude  in [2.5,  12.5]
# longitude in [-1.5,  4.0]

if not is_valid:
    print(f"{len(df_invalid)} invalid coordinate(s).")

When region is set to something other than 'benin', worldwide bounding box defaults [-90, 90] for latitude and [-180, 180] for longitude are applied.

Parameter	Type	Default	Description
`lat`	`str`	-	Name of the latitude column
`lon`	`str`	-	Name of the longitude column
`region`	`str`	`'benin'`	Reference region for the bounding box

Returns: tuple[bool, pd.DataFrame]

True if all coordinates fall within the bounding box.
DataFrame of rows with coordinates outside the bounding box (empty if OK).

Exception: ValidationError if either lat or lon column is missing from the DataFrame.

`check_unique(cols)`¶

Verifies that the combination of specified columns is unique across all rows (equivalent to a composite primary key).

is_valid, df_duplicates = validator.check_unique(
    cols=["culture", "marche", "date_recolte"]
)

if not is_valid:
    print(f"{len(df_duplicates)} duplicate row(s).")
    print(df_duplicates)

Parameter	Type	Description
`cols`	`list[str]`	List of columns whose combined values must be unique

Returns: tuple[bool, pd.DataFrame]

True if the column combination is unique.
DataFrame of all rows involved in duplicate combinations (empty if OK).

`check_fk(fk_col, ref)`¶

Verifies referential integrity of a column: every non-null value must exist within the provided reference set.

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

is_valid, df_missing = validator.check_fk(
    fk_col="marche",
    ref=valid_markets,
)

if not is_valid:
    print(f"{len(df_missing)} unknown reference(s).")

Parameter	Type	Description
`fk_col`	`str`	Name of the column holding foreign key values
`ref`	`set`	Set of expected valid reference values

Returns: tuple[bool, pd.DataFrame]

True if all values are present in the reference set.
DataFrame of rows with values missing from the reference set (empty if OK).

Exception: ValidationError if column fk_col is missing.

`quality_score()`¶

Calculates an overall quality score and scores per dimension for the DataFrame.

score = validator.quality_score()

print(score["overall"])       # ex: 0.8742
print(score["completeness"])  # non-null cell ratio
print(score["consistency"])   # 1 - duplicate ratio
print(score["accuracy"])      # 1.0 (extensible)
print(score["columns"])       # completeness score per column

The overall score is a weighted average:

Dimension	Weight	Calculation
Completeness	40%	Proportion of non-null cells
Consistency	35%	`1 - (nb_duplicates / total_rows)`
Accuracy	25%	`1.0` by default (extensible)

Returns: dict with keys:

Key	Type	Description
`overall`	`float`	Overall score in range [0.0, 1.0]
`completeness`	`float`	Completeness score
`consistency`	`float`	Consistency score
`accuracy`	`float`	Accuracy score
`columns`	`dict[str, float]`	Completeness score per column

`report()`¶

Returns the complete validation report accumulated since the initialization of Validator.

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

report = 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,
#     ...
#   }
# }

Key	Type	Description
`lignes`	`int`	Number of DataFrame rows at initialization
`colonnes`	`int`	Number of DataFrame columns at initialization
`validations`	`list`	Log of each performed validation
`quality_score`	`dict`	Present only if `quality_score()` was invoked

Returns: dict (copy of internal report).

Complete Example¶

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",
}

is_valid, errors = validator.check_schema(schema)
if not is_valid:
    for msg in errors:
        print(msg)

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

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

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

report = validator.report()

Usage in a Pipeline¶

Validator is used internally by Pipeline when adding a validation step via add_step or validate.

from kadi.kidas import Pipeline

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

For pipeline documentation, see Pipeline.

Backward Compatibility¶

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

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

Similarly, class name DataValidator is an alias for 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)¶

Initialization¶

Validation Methods¶

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

Complete Example¶

Usage in a Pipeline¶

Backward Compatibility¶

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 ¶