Skip to content

kadi.kidas: Data Processing and Standardization

The kadi.kidas module is the agricultural data processing pipeline of KadiPy. It handles data ingestion, cleaning, validation, and normalization from any source.

kidas = KadiPy Data Ingestion, Alignment and Standardization

Architecture

    Source (CSV / Excel / JSON / NetCDF / API)
                        |
                        |
                    Pipeline             <- Orchestrator
                        |
                        |
             ┌──────────┴──────────┐
             |                     |
             |                     |
         Cleaner               Validator
             |                     |
             |                     |
             └──────────┬──────────┘
                        |
                        |
                   Normalizer            <- Final standardization
                        |
                        |
                      Cache               <- SQLite persistence
                        |
                        |
                DataFrame + Report

Each component can be used independently or chained together in a pipeline.

Quick Start

In One Line with load_clean

import kadi.kidas as kidas

df, report = kidas.load_clean("recolte_2024.csv")
print(f"{len(df)} rows loaded")

load_clean automatically creates a pipeline with duplicate removal and missing value imputation using the mean.

With kadi.io (Read Only)

import kadi as kd

df = kd.read_csv("recolte_2024.csv")
kd.info("recolte_2024.csv")

Custom Pipeline

from kadi.kidas import Pipeline

pipeline = Pipeline()

df, report = (
    pipeline
    .add_source("donnees_marche.xlsx")
    .add_step("drop_dupes")
    .add_step("fill_missing", strategy="median")
    .add_step("check_schema", {"culture": "str", "rendement_kg": "float"})
    .add_step("norm_cols", {"crops": "culture"})
    .run(cache=True)
)

print(report["steps_summary"])
print(f"Quality score: {report.get('quality_score')}")
print(f"Output rows: {report['nb_rows_out']}")

Fluid API with Short Aliases

from kadi.kidas import Pipeline

df, report = (
    Pipeline()
    .load("donnees_marche.xlsx")
    .clean("drop_dupes")
    .clean("fill_missing", strategy="median")
    .validate({"culture": "str", "rendement_kg": "float"})
    .normalize({"crops": "culture"})
    .run(cache=True)
)

Supported File Formats

Format Extension Source Class
CSV .csv, .tsv, .txt CSVSource
Excel .xlsx, .xls, .xlsm ExcelSource
JSON .json JSONSource
NetCDF .nc, .nc4, .netcdf NetCDFSource
REST API URL http:// or https:// APISource

The format is automatically detected from the extension or URL prefix.

Pipeline Report

Every call to run() returns a structured report:

df, report = pipeline.add_source("recoltes.csv").add_step("check_schema", {"culture": "str"}).run()

# Number of rows before and after processing
print(f"Before: {report['nb_rows_in']} rows")
print(f"After: {report['nb_rows_out']} rows")

# List of applied steps
print(report["steps_summary"])

# Quality score (present if a validation step was executed)
if report.get("quality_score"):
    print(f"Overall score: {report['quality_score']['overall']:.2f}")

# Validation warnings
for warning in report.get("warnings", []):
    print(warning)

# Data loaded from cache
print(f"Cache used: {report['cache_utilise']}")
Key Type Description
nb_rows_in int Number of rows loaded from source
nb_rows_out int Number of rows after processing
steps_summary list[str] Names of applied steps in execution order
quality_score dict or None Quality score (generated by a validation step)
warnings list[str] Warnings produced during validation
cache_utilise bool True if data was loaded from cache
details dict Detailed internal reports for each step

Submodules