Aller au contenu

Sources de données (kadi.kidas.sources)

Le sous-package kadi.kidas.sources regroupe toutes les classes permettant de lire et d'écrire des données agricoles depuis différents formats. Chaque classe concrète hérite de Source, la classe abstraite de base, et implémente le même contrat : read(), write(), info(), ping().

Classe Format Extensions / Préfixe
CSVSource Fichiers texte délimités .csv
ExcelSource Classeurs Excel .xlsx, .xls
JSONSource Fichiers JSON ou dicts Python .json
NetCDFSource Données grillées agrométéo .nc, .netcdf
APISource APIs REST HTTP http://, https://

Import

from kadi.kidas import CSVSource, ExcelSource, JSONSource, NetCDFSource, APISource

# ou depuis le sous-package directement
from kadi.kidas.sources import CSVSource

Les classes sont également réexportées depuis kadi.io. Voir Entrées et sorties pour la documentation des fonctions read_*, write_* et write.

Source (classe abstraite)

Source est l'interface commune à toutes les sources de données. Elle ne s'instancie pas directement.

Attributs communs :

Attribut Type Description
path str Chemin local du fichier ou URI de la ressource
kind str Type de source : 'csv', 'excel', 'json', 'netcdf', 'api'
encoding str Encodage des données (peut être 'auto')
last_read datetime ou None Horodatage de la dernière lecture réussie

Méthodes abstraites que chaque sous-classe implémente :

  • read(**kwargs) : lit les données et retourne un pd.DataFrame
  • write(data, **kwargs) : écrit un DataFrame vers la source, retourne bool
  • info() : retourne un dict de métadonnées
  • ping() : retourne True si la source est accessible

CSVSource

Source pour les fichiers CSV agricoles avec détection automatique de l'encodage (via chardet), du délimiteur et du séparateur décimal. Conçue pour les exports de capteurs IoT, de coopératives et des bases de données nationales (INSAE, MAEP).

Initialisation

from kadi.kidas import CSVSource

# Détection automatique de tout
source = CSVSource("recoltes_2024.csv")

# Paramètres explicites
source = CSVSource(
    "export_insae.csv",
    encoding="latin-1",
    sep=";",
    decimal=",",
)
Paramètre Type Défaut Description
path str - Chemin vers le fichier CSV
encoding str 'auto' Encodage. 'auto' : détecté via chardet
sep str 'auto' Délimiteur de colonnes. 'auto' : auto-détecté
decimal str 'auto' Séparateur décimal. 'auto' : inféré selon le délimiteur

En mode 'auto', l'encodage est détecté sur un échantillon de 10 000 octets. Le délimiteur est détecté par csv.Sniffer avec fallback par comptage d'occurrences parmi ,, ;, \t, |. Si le délimiteur est ;, le séparateur décimal inféré est , (convention française).

read(n=None, skip=None)

df = source.read()

# 100 premières lignes
df = source.read(n=100)

# Ignorer 2 lignes de métadonnées en début de fichier
df = source.read(skip=2)
Paramètre Type Défaut Description
n int ou None None Nombre maximum de lignes à lire
skip int ou None None Lignes à ignorer en début de fichier (hors en-tête)

Tente plusieurs encodages en fallback (utf-8, latin-1, cp1252) si la lecture principale échoue.

Retour : pd.DataFrame Exceptions : ConnectError si le fichier est inaccessible, ReadError si tous les encodages échouent.

write(data, index=False)

source.write(df)
source.write(df, index=True)

Écrit en UTF-8 si l'encodage est 'auto'. Utilise le délimiteur détecté ou , par défaut.

Retour : bool Exception : WriteError

info()

print(source.info())
# {
#   'path': 'recoltes_2024.csv',
#   'kind': 'csv',
#   'encoding': 'utf-8',
#   'sep': ',',
#   'decimal': 'inféré',
#   'rows': 1500,
#   'cols': 8,
#   'size_kb': 42.31,
#   'last_read': '2024-09-01T10:23:45.123456'
# }

ping()

Vérifie que le fichier existe et est lisible en lecture. Retourne bool.

ExcelSource

Source pour les fichiers Excel (.xlsx, .xls) avec détection automatique de la ligne d'en-tête et résolution des cellules fusionnées par forward fill. Adaptée aux rapports agricoles béninois avec communes et marchés fusionnés sur plusieurs lignes.

Initialisation

from kadi.kidas import ExcelSource

source = ExcelSource("prix_marches_2024.xlsx")

# Feuille et en-tête explicites
source = ExcelSource("rapport.xlsx", sheet="Janvier", header=2)
Paramètre Type Défaut Description
path str - Chemin vers le fichier Excel
sheet str ou int 0 Feuille par défaut (nom ou index)
header str ou int 'auto' Ligne d'en-tête. 'auto' : détectée automatiquement

La détection d'en-tête inspecte les 15 premières lignes et retient la ligne la plus dense (plus grand nombre de valeurs non-null).

sheets()

Retourne la liste des noms de feuilles du classeur.

feuilles = source.sheets()
# ['Janvier', 'Fevrier', 'Mars']

Retour : list[str] Exceptions : ConnectError, ReadError

sheet_info(sheet)

Retourne les métadonnées d'une feuille spécifique.

meta = source.sheet_info("Janvier")
# {
#   'sheet': 'Janvier',
#   'rows': 245,
#   'cols': 6,
#   'columns': ['culture', 'marche', 'prix_xof_kg', ...]
# }
Paramètre Type Description
sheet str ou int Nom ou index de la feuille à inspecter

Retour : dict Exception : ReadError

read(sheet=None)

df = source.read()

# Lire une feuille spécifique
df = source.read(sheet="Fevrier")

Applique un forward fill vertical sur toutes les colonnes pour résoudre les cellules fusionnées, puis supprime les lignes entièrement vides.

Paramètre Type Défaut Description
sheet str, int ou None None Feuille à lire. None : utilise celle définie à l'initialisation

Retour : pd.DataFrame Exceptions : ConnectError, ReadError

write(data, sheet='Sheet1')

source.write(df, sheet="Export")

Retour : bool Exception : WriteError

info()

print(source.info())
# {
#   'path': 'prix_marches_2024.xlsx',
#   'kind': 'excel',
#   'sheets': ['Janvier', 'Fevrier'],
#   'active_sheet': 0,
#   'detected_header': 1,
#   'size_kb': 118.7,
#   'last_read': None
# }

ping()

Vérifie que le fichier existe et est lisible en lecture. Retourne bool.

JSONSource

Source pour les fichiers JSON plats ou imbriqués, tels que retournés par les APIs agricoles (FAO, WFP VAM, MAEP). Supporte également un dictionnaire Python fourni directement en mémoire.

Initialisation

from kadi.kidas import JSONSource

# Depuis un fichier
source = JSONSource("donnees_fao.json")

# Depuis un dictionnaire Python en mémoire
data = {"location": {"lat": 9.3, "lon": 2.4}, "crop": "maize"}
source = JSONSource(data)
Paramètre Type Description
file_path_or_dict str ou dict Chemin vers le fichier JSON ou dictionnaire Python

Quand la source est un dictionnaire, path vaut '<dict_en_memoire>' et ping() retourne toujours True.

flatten(obj, sep='.')

Aplatit récursivement un objet JSON imbriqué en utilisant la notation pointée.

obj = {"location": {"lat": 9.3, "lon": 2.4}, "crop": "maize"}
plat = source.flatten(obj)
# {"location.lat": 9.3, "location.lon": 2.4, "crop": "maize"}
Paramètre Type Défaut Description
obj dict - L'objet JSON à aplatir
sep str '.' Séparateur entre les niveaux de clés

Les listes sont indexées par leur position (location.items.0, location.items.1...).

Retour : dict

read(flatten=True)

df = source.read()

# Sans aplatissement (pd.json_normalize direct)
df = source.read(flatten=False)
Paramètre Type Défaut Description
flatten bool True Si True, aplatit chaque enregistrement avant conversion

Accepte les réponses JSON de type dict (un seul enregistrement) ou list (liste d'enregistrements).

Retour : pd.DataFrame Exceptions : ConnectError, ReadError

write(data, orient='records')

source.write(df)
source.write(df, orient="index")
Paramètre Type Défaut Description
data pd.DataFrame - Données à écrire
orient str 'records' Orientation JSON : 'records', 'index', 'columns', 'values'

Écrit avec force_ascii=False et indent=2.

Exception : WriteError si la source est un dictionnaire en mémoire (aucun fichier de destination).

Retour : bool

info()

print(source.info())
# {
#   'path': 'donnees_fao.json',
#   'kind': 'json',
#   'is_file': True,
#   'size_kb': 8.45,
#   'last_read': None
# }

ping()

Pour une source fichier : vérifie l'existence et la lisibilité du fichier. Pour une source dict : retourne toujours True. Retourne bool.

NetCDFSource

Source pour les fichiers NetCDF agrométéorologiques (.nc), utilisés pour les données CHIRPS (précipitations), TAMSAT (Afrique) et GFS (prévisions globales). Utilise xarray comme moteur de lecture et supporte le chunking Dask pour les grands fichiers.

Dépendance : xarray et netCDF4 doivent être installés.

Initialisation

from kadi.kidas import NetCDFSource

source = NetCDFSource("chirps_benin_2024.nc")

# Avec Dask pour les fichiers volumineux (> 500 Mo)
source = NetCDFSource("gfs_global_2024.nc", use_dask=True)
Paramètre Type Défaut Description
path str - Chemin vers le fichier NetCDF
use_dask bool False Active le chargement paresseux via Dask

dims()

Retourne les dimensions du dataset.

dimensions = source.dims()
# {'lat': 240, 'lon': 360, 'time': 365}

Retour : dict[str, int]

read(lat_bounds=None, lon_bounds=None, time_bounds=None)

Extrait un sous-ensemble spatial et temporel du fichier. Si aucune bounding box n'est spécifiée, utilise par défaut la bbox du Bénin (lat∈[2.5, 12.5], lon∈[-1.5, 4.0]).

# Extraction avec la bbox Bénin par défaut
da = source.read()

# Extraction sur une région personnalisée
da = source.read(
    lat_bounds=(6.0, 12.5),
    lon_bounds=(1.0, 4.0),
    time_bounds=("2024-01-01", "2024-06-30"),
)
Paramètre Type Défaut Description
lat_bounds tuple[float, float] ou None bbox Bénin Intervalle de latitude (lat_min, lat_max)
lon_bounds tuple[float, float] ou None bbox Bénin Intervalle de longitude (lon_min, lon_max)
time_bounds tuple[str, str] ou None None Intervalle temporel au format ISO ('YYYY-MM-DD', 'YYYY-MM-DD')

Détecte automatiquement les noms de dimensions lat/lon dans le dataset (compatibles avec 'lat', 'latitude', 'y', etc.). Lit la première variable de données si le dataset en contient plusieurs.

Retour : xr.DataArray Exceptions : ConnectError, ReadError

to_df()

Convertit le dernier DataArray extrait par read() en pd.DataFrame. Si read() n'a pas encore été appelé, effectue une lecture avec la bbox Bénin.

da = source.read()
df = source.to_df()
# Colonnes : lat, lon, time, <nom_variable>

Retour : pd.DataFrame Exception : ReadError

write(data)

Convertit le DataFrame en xr.Dataset via from_dataframe() et le sauvegarde au format NetCDF.

Retour : bool Exception : WriteError

info()

print(source.info())
# {
#   'path': 'chirps_benin_2024.nc',
#   'kind': 'netcdf',
#   'dimensions': {'lat': 240, 'lon': 360, 'time': 365},
#   'variables': ['precip'],
#   'use_dask': False,
#   'size_kb': 45230.0,
#   'last_read': None
# }

ping()

Vérifie que le fichier existe et est lisible en lecture. Retourne bool.

APISource

Source pour les APIs REST agricoles et climatiques (Open-Meteo, WFP VAM, FAO, SoilGrids). Intègre le rate limiting et un mécanisme de réessai avec backoff exponentiel.

Initialisation

from kadi.kidas import APISource

# API publique
source = APISource("https://api.open-meteo.com/v1/forecast")

# API authentifiée avec limite de débit personnalisée
source = APISource(
    "https://api.wfp.org/vam-data-bridges/1.0",
    token="mon_token_bearer",
    rate_limit=2.0,
)
Paramètre Type Défaut Description
url str - URL de base de l'endpoint API
token str ou None None Jeton Bearer pour les APIs authentifiées
rate_limit float 5.0 Nombre maximum de requêtes par seconde

fetch(params, retries=3, backoff=5.0)

Effectue une requête GET avec réessais et backoff exponentiel.

donnees = source.fetch(
    params={"latitude": 6.36, "longitude": 2.42},
    retries=3,
    backoff=5.0,
)
Paramètre Type Défaut Description
params dict - Paramètres de la requête GET (query string)
retries int 3 Nombre maximum de tentatives en cas d'échec
backoff float 5.0 Délai de base en secondes entre les réessais (doublé à chaque tentative, plafonné à 60 s)

Les codes HTTP réessayables sont : 429, 500, 502, 503, 504.

Retour : dict (réponse JSON parsée) Exceptions : ConnectError si l'API est inaccessible, ReadError si la réponse n'est pas un JSON valide.

read(params=None)

Effectue une requête GET et normalise la réponse en DataFrame.

df = source.read({
    "latitude": 6.36,
    "longitude": 2.42,
    "daily": "temperature_2m_max",
})

Cherche automatiquement une clé standard dans la réponse dict ('data', 'results', 'items', 'records', 'features'). Si aucune n'est trouvée, utilise le dict entier comme unique enregistrement.

Retour : pd.DataFrame Exceptions : ConnectError, ReadError

write(data)

Envoie le DataFrame vers l'API via une requête POST au format JSON (orient='records').

Retour : bool Exception : WriteError

info()

print(source.info())
# {
#   'url': 'https://api.open-meteo.com/v1/forecast',
#   'kind': 'api',
#   'requires_auth': False,
#   'rate_limit': 5.0,
#   'last_read': None
# }

ping()

Effectue une requête HEAD légère. Retourne True si le code de réponse est inférieur à 500 (l'API répond, même avec une erreur 4xx). Retourne False si la connexion échoue complètement.

Exemple complet

import kadi as kd
from kadi.kidas import CSVSource, ExcelSource, JSONSource, NetCDFSource, APISource

# CSV avec détection automatique
source_csv = CSVSource("recoltes_2024.csv")
if source_csv.ping():
    df = source_csv.read()
    print(source_csv.info())

# Excel multi-feuilles
source_excel = ExcelSource("prix_marches.xlsx")
for feuille in source_excel.sheets():
    df = source_excel.read(sheet=feuille)
    print(f"{feuille} : {len(df)} lignes")

# JSON imbriqué
source_json = JSONSource("donnees_fao.json")
df = source_json.read(flatten=True)

# NetCDF agrométéo
source_nc = NetCDFSource("chirps_benin_2024.nc")
da = source_nc.read(lat_bounds=(2.5, 12.5), lon_bounds=(-1.5, 4.0))
df = source_nc.to_df()

# API REST
source_api = APISource("https://api.open-meteo.com/v1/forecast")
df = source_api.read({
    "latitude": 6.36,
    "longitude": 2.42,
    "daily": "temperature_2m_max",
})

Rétrocompatibilité

Les anciens noms de classes émettent un DeprecationWarning et continuent de fonctionner jusqu'à KadiPy v2.0 :

Ancien nom (avant v1.2.0) Nouveau nom
DataSource Source
CSVDataSource CSVSource
ExcelDataSource ExcelSource
JSONDataSource JSONSource
NetCDFDataSource NetCDFSource
APIDataSource APISource

Source

Bases: ABC

Classe abstraite définissant l'interface commune à toutes les sources de données.

Cette classe représente le contrat que chaque source concrète (CSV, Excel, JSON, NetCDF, API) doit respecter. Elle garantit que toutes les sources exposent les mêmes méthodes fondamentales : read(), write(), info() et ping().

Attributs

path (str): Chemin local du fichier ou URI de la ressource. kind (str): Type de source parmi 'csv', 'excel', 'json', 'netcdf' ou 'api'. encoding (str): Encodage des données (ex: 'utf-8', 'latin1'). Peut être 'auto' pour une détection automatique. last_read (datetime | None): Horodatage de la dernière lecture réussie. None si la source n'a jamais été lue.

Source code in kadi/kidas/sources/base.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class Source(ABC):
    """Classe abstraite définissant l'interface commune à toutes les sources de données.

    Cette classe représente le contrat que chaque source concrète (CSV, Excel,
    JSON, NetCDF, API) doit respecter. Elle garantit que toutes les sources
    exposent les mêmes méthodes fondamentales : read(), write(), info()
    et ping().

    Attributs:
        path (str): Chemin local du fichier ou URI de la ressource.
        kind (str): Type de source parmi 'csv', 'excel', 'json',
            'netcdf' ou 'api'.
        encoding (str): Encodage des données (ex: 'utf-8', 'latin1').
            Peut être 'auto' pour une détection automatique.
        last_read (datetime | None): Horodatage de la dernière lecture
            réussie. None si la source n'a jamais été lue.
    """

    def __init__(
        self,
        path: str,
        kind: str,
        encoding: str = "utf-8",
    ) -> None:
        """Initialise les attributs communs à toutes les sources de données.

        Args:
            path (str): Chemin local du fichier ou URI de la ressource.
            kind (str): Type de la source (ex: 'csv', 'excel').
            encoding (str): Encodage des données. Par défaut 'utf-8'.
        """
        # Chemin ou URI de la source de données
        self.path: str = path

        # Type de source : 'csv', 'excel', 'json', 'netcdf' ou 'api'
        self.kind: str = kind

        # Encodage utilisé pour la lecture de la source
        self.encoding: str = encoding

        # Horodatage de la dernière lecture réussie (None si jamais lue)
        self.last_read: Optional[datetime] = None

    @abstractmethod
    def read(self, **kwargs) -> pd.DataFrame:
        """Lit les données depuis la source et les retourne sous forme de DataFrame.

        Cette méthode doit être implémentée par chaque sous-classe concrète
        selon le format de fichier concerné.

        Args:
            **kwargs: Arguments optionnels spécifiques à chaque format
                (ex: n, sheet, lat_bounds...).

        Returns:
            pd.DataFrame: Les données lues depuis la source.

        Raises:
            ReadError: Si la lecture échoue (fichier corrompu,
                format invalide, etc.).
            ConnectError: Si la source n'est pas accessible.
        """
        pass

    @abstractmethod
    def write(self, data: pd.DataFrame, **kwargs) -> bool:
        """Écrit un DataFrame vers la source de données.

        Cette méthode doit être implémentée par chaque sous-classe concrète.

        Args:
            data (pd.DataFrame): Les données à écrire vers la source.
            **kwargs: Arguments optionnels spécifiques à chaque format
                (ex: index, sheet, orient...).

        Returns:
            bool: True si l'écriture s'est déroulée avec succès.

        Raises:
            WriteError: Si l'écriture échoue.
        """
        pass

    @abstractmethod
    def info(self) -> dict:
        """Retourne un dictionnaire de métadonnées décrivant la source.

        Les métadonnées varient selon le type de source, mais incluent
        généralement : le chemin, le type, le nombre de lignes et colonnes,
        la taille du fichier et l'encodage détecté.

        Returns:
            dict: Dictionnaire contenant les métadonnées de la source.
        """
        pass

    @abstractmethod
    def ping(self) -> bool:
        """Vérifie que la source de données est accessible et lisible.

        Pour les fichiers locaux, vérifie que le fichier existe et est
        lisible. Pour les APIs, vérifie que l'endpoint répond correctement.

        Returns:
            bool: True si la source est accessible, False sinon.
        """
        pass

    def _touch(self) -> None:
        """Met à jour l'horodatage de la dernière lecture réussie.

        Cette méthode utilitaire est appelée par les sous-classes après
        chaque appel réussi à read(). Le nom s'inspire de la commande
        UNIX touch, qui met à jour l'horodatage d'un fichier.
        """
        # Enregistrement de l'heure de la lecture
        self.last_read = datetime.now()
        logger.debug(
            "Source '%s' lue avec succès à %s.",
            self.path,
            self.last_read.isoformat(),
        )

    def __repr__(self) -> str:
        """Retourne une représentation lisible de la source de données.

        Returns:
            str: Représentation de l'objet sous forme de chaîne.
        """
        return (
            f"{self.__class__.__name__}("
            f"path='{self.path}', "
            f"kind='{self.kind}', "
            f"encoding='{self.encoding}')"
        )

__init__ 

__init__(path: str, kind: str, encoding: str = 'utf-8') -> None

Initialise les attributs communs à toutes les sources de données.

Parameters:

Name Type Description Default
path str

Chemin local du fichier ou URI de la ressource.

 required
kind str

Type de la source (ex: 'csv', 'excel').

 required
encoding str

Encodage des données. Par défaut 'utf-8'.

 'utf-8'
Source code in kadi/kidas/sources/base.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    path: str,
    kind: str,
    encoding: str = "utf-8",
) -> None:
    """Initialise les attributs communs à toutes les sources de données.

    Args:
        path (str): Chemin local du fichier ou URI de la ressource.
        kind (str): Type de la source (ex: 'csv', 'excel').
        encoding (str): Encodage des données. Par défaut 'utf-8'.
    """
    # Chemin ou URI de la source de données
    self.path: str = path

    # Type de source : 'csv', 'excel', 'json', 'netcdf' ou 'api'
    self.kind: str = kind

    # Encodage utilisé pour la lecture de la source
    self.encoding: str = encoding

    # Horodatage de la dernière lecture réussie (None si jamais lue)
    self.last_read: Optional[datetime] = None

read abstractmethod 

read(**kwargs) -> pd.DataFrame

Lit les données depuis la source et les retourne sous forme de DataFrame.

Cette méthode doit être implémentée par chaque sous-classe concrète selon le format de fichier concerné.

Parameters:

Name Type Description Default
**kwargs

Arguments optionnels spécifiques à chaque format (ex: n, sheet, lat_bounds...).

 {}

Returns:

Type Description
DataFrame

pd.DataFrame: Les données lues depuis la source.

Raises:

Type Description
ReadError

Si la lecture échoue (fichier corrompu, format invalide, etc.).
ConnectError

Si la source n'est pas accessible.
Source code in kadi/kidas/sources/base.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@abstractmethod
def read(self, **kwargs) -> pd.DataFrame:
    """Lit les données depuis la source et les retourne sous forme de DataFrame.

    Cette méthode doit être implémentée par chaque sous-classe concrète
    selon le format de fichier concerné.

    Args:
        **kwargs: Arguments optionnels spécifiques à chaque format
            (ex: n, sheet, lat_bounds...).

    Returns:
        pd.DataFrame: Les données lues depuis la source.

    Raises:
        ReadError: Si la lecture échoue (fichier corrompu,
            format invalide, etc.).
        ConnectError: Si la source n'est pas accessible.
    """
    pass

write abstractmethod 

write(data: DataFrame, **kwargs) -> bool

Écrit un DataFrame vers la source de données.

Cette méthode doit être implémentée par chaque sous-classe concrète.

Parameters:

Name Type Description Default
data DataFrame

Les données à écrire vers la source.

 required
**kwargs

Arguments optionnels spécifiques à chaque format (ex: index, sheet, orient...).

 {}

Returns:

Name Type Description
bool bool

True si l'écriture s'est déroulée avec succès.

Raises:

Type Description
WriteError

Si l'écriture échoue.
Source code in kadi/kidas/sources/base.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@abstractmethod
def write(self, data: pd.DataFrame, **kwargs) -> bool:
    """Écrit un DataFrame vers la source de données.

    Cette méthode doit être implémentée par chaque sous-classe concrète.

    Args:
        data (pd.DataFrame): Les données à écrire vers la source.
        **kwargs: Arguments optionnels spécifiques à chaque format
            (ex: index, sheet, orient...).

    Returns:
        bool: True si l'écriture s'est déroulée avec succès.

    Raises:
        WriteError: Si l'écriture échoue.
    """
    pass

info abstractmethod 

info() -> dict

Retourne un dictionnaire de métadonnées décrivant la source.

Les métadonnées varient selon le type de source, mais incluent généralement : le chemin, le type, le nombre de lignes et colonnes, la taille du fichier et l'encodage détecté.

Returns:

Name Type Description
dict dict

Dictionnaire contenant les métadonnées de la source.
Source code in kadi/kidas/sources/base.py 
109
110
111
112
113
114
115
116
117
118
119
120
@abstractmethod
def info(self) -> dict:
    """Retourne un dictionnaire de métadonnées décrivant la source.

    Les métadonnées varient selon le type de source, mais incluent
    généralement : le chemin, le type, le nombre de lignes et colonnes,
    la taille du fichier et l'encodage détecté.

    Returns:
        dict: Dictionnaire contenant les métadonnées de la source.
    """
    pass

ping abstractmethod 

ping() -> bool

Vérifie que la source de données est accessible et lisible.

Pour les fichiers locaux, vérifie que le fichier existe et est lisible. Pour les APIs, vérifie que l'endpoint répond correctement.

Returns:

Name Type Description
bool bool

True si la source est accessible, False sinon.
Source code in kadi/kidas/sources/base.py 
122
123
124
125
126
127
128
129
130
131
132
@abstractmethod
def ping(self) -> bool:
    """Vérifie que la source de données est accessible et lisible.

    Pour les fichiers locaux, vérifie que le fichier existe et est
    lisible. Pour les APIs, vérifie que l'endpoint répond correctement.

    Returns:
        bool: True si la source est accessible, False sinon.
    """
    pass

__repr__ 

__repr__() -> str

Retourne une représentation lisible de la source de données.

Returns:

Name Type Description
str str

Représentation de l'objet sous forme de chaîne.
Source code in kadi/kidas/sources/base.py 
149
150
151
152
153
154
155
156
157
158
159
160
def __repr__(self) -> str:
    """Retourne une représentation lisible de la source de données.

    Returns:
        str: Représentation de l'objet sous forme de chaîne.
    """
    return (
        f"{self.__class__.__name__}("
        f"path='{self.path}', "
        f"kind='{self.kind}', "
        f"encoding='{self.encoding}')"
    )

CSVSource

Bases: Source

Source de données pour les fichiers CSV agricoles.

Gère la lecture robuste de fichiers CSV avec détection automatique de l'encodage, du délimiteur et du séparateur décimal. Conçue pour traiter les exports de capteurs IoT, de coopératives et des bases de données nationales (INSAE, MAEP).

Attributs

path (str): Chemin absolu ou relatif vers le fichier CSV. sep (str): Délimiteur de colonnes (auto-détecté si 'auto'). decimal (str): Séparateur décimal (auto-détecté si 'auto'). _sep (str | None): Délimiteur détecté en cache. _enc (str | None): Encodage détecté en cache.

Exemple

source = CSVSource('recoltes_2024.csv') df = source.read() print(source.info())

Source code in kadi/kidas/sources/csv_source.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
class CSVSource(Source):
    """Source de données pour les fichiers CSV agricoles.

    Gère la lecture robuste de fichiers CSV avec détection automatique
    de l'encodage, du délimiteur et du séparateur décimal. Conçue pour
    traiter les exports de capteurs IoT, de coopératives et des bases
    de données nationales (INSAE, MAEP).

    Attributs:
        path (str): Chemin absolu ou relatif vers le fichier CSV.
        sep (str): Délimiteur de colonnes (auto-détecté si 'auto').
        decimal (str): Séparateur décimal (auto-détecté si 'auto').
        _sep (str | None): Délimiteur détecté en cache.
        _enc (str | None): Encodage détecté en cache.

    Exemple:
        >>> source = CSVSource('recoltes_2024.csv')
        >>> df = source.read()
        >>> print(source.info())
    """

    def __init__(
        self,
        path: str,
        encoding: str = "auto",
        sep: str = "auto",
        decimal: str = "auto",
    ) -> None:
        """Initialise la source CSV avec détection automatique des paramètres.

        Args:
            path (str): Chemin vers le fichier CSV à lire.
            encoding (str): Encodage du fichier. 'auto' pour la détection
                automatique via chardet. Par défaut 'auto'.
            sep (str): Délimiteur de colonnes. 'auto' pour l'auto-
                détection. Par défaut 'auto'.
            decimal (str): Séparateur décimal. 'auto' pour l'inférence
                (détecte les fichiers français avec ','). Par défaut 'auto'.
        """
        # Initialisation de la classe parente avec le type 'csv'
        super().__init__(
            path=path,
            kind="csv",
            encoding=encoding,
        )

        # Paramètres de lecture (peuvent être 'auto' avant détection)
        self.sep: str = sep
        self.decimal: str = decimal

        # Cache interne pour éviter de re-détecter à chaque lecture
        self._sep: Optional[str] = None
        self._enc: Optional[str] = None

    def _sniff_encoding(self) -> str:
        """Détecte automatiquement l'encodage du fichier CSV via chardet.

        Lit un échantillon du fichier (10 000 octets) pour minimiser
        le temps de détection tout en gardant une bonne précision.
        Retourne 'utf-8' comme encodage par défaut si la détection échoue.

        Returns:
            str: L'encodage détecté (ex: 'utf-8', 'ISO-8859-1').

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
        """
        # Vérification de l'existence du fichier avant de lire
        if not os.path.isfile(self.path):
            raise ConnectError(
                f"Fichier CSV introuvable : '{self.path}'"
            )

        try:
            # Lecture d'un échantillon d'octets bruts pour chardet
            with open(self.path, "rb") as fichier:
                echantillon = fichier.read(_CHARDET_SAMPLE_SIZE)

            # Analyse de l'encodage via chardet
            resultat = chardet.detect(echantillon)
            detected_encoding = resultat.get("encoding")

            if not detected_encoding:
                logger.warning(
                    "Échec de la détection automatique de l'encodage pour '%s'. "
                    "Encodage de repli 'utf-8' utilisé.",
                    self.path,
                )
                encodage = "utf-8"
            else:
                encodage = detected_encoding
                logger.debug(
                    "Encodage détecté pour '%s' : %s (confiance : %.0f%%).",
                    self.path,
                    encodage,
                    (resultat.get("confidence") or 0) * 100,
                )

            # Mise en cache du résultat
            self._enc = encodage
            return encodage

        except OSError as erreur:
            raise ConnectError(
                f"Impossible de lire le fichier CSV '{self.path}' : {erreur}"
            ) from erreur

    def _sniff_sep(self) -> str:
        """Détecte automatiquement le délimiteur de colonnes du fichier CSV.

        Utilise csv.Sniffer sur les premières lignes du fichier. En cas
        d'échec, teste successivement les délimiteurs candidats (, ; \\t |)
        et choisit celui produisant le plus de colonnes.

        Returns:
            str: Le délimiteur détecté (ex: ',', ';', '\\t', '|').

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
        """
        # Détermination de l'encodage pour ouvrir le fichier correctement
        encodage = self._enc or self._sniff_encoding()

        try:
            # Tentative 1 : utilisation du Sniffer de la bibliothèque csv
            with open(self.path, encoding=encodage, errors="replace") as f:
                echantillon = f.read(2048)

            try:
                dialect = csv.Sniffer().sniff(echantillon, _CANDIDATE_DELIMITERS)
                delimiteur = dialect.delimiter
                logger.debug(
                    "Délimiteur détecté via Sniffer pour '%s' : '%s'.",
                    self.path,
                    delimiteur,
                )
            except csv.Error:
                # Tentative 2 : compter les occurrences de chaque délimiteur
                premiere_ligne = echantillon.splitlines()[0] if echantillon else ""
                delimiteur = max(
                    _CANDIDATE_DELIMITERS,
                    key=lambda d: premiere_ligne.count(d),
                )
                logger.debug(
                    "Délimiteur détecté par comptage pour '%s' : '%s'.",
                    self.path,
                    delimiteur,
                )

            # Mise en cache du résultat
            self._sep = delimiteur
            return delimiteur

        except OSError as erreur:
            raise ConnectError(
                f"Impossible d'accéder au fichier '{self.path}' : {erreur}"
            ) from erreur

    def read(
        self,
        n: Optional[int] = None,
        skip: Optional[int] = None,
    ) -> pd.DataFrame:
        """Lit le fichier CSV et retourne son contenu sous forme de DataFrame.

        Effectue la détection automatique de l'encodage et du délimiteur
        si nécessaire. Tente plusieurs encodages en fallback si la lecture
        principale échoue.

        Args:
            n (int | None): Nombre maximum de lignes à lire. None pour
                lire toutes les lignes. Par défaut None.
            skip (int | None): Nombre de lignes à ignorer en début
                de fichier (hors en-tête). Par défaut None.

        Returns:
            pd.DataFrame: Les données du fichier CSV.

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
            ReadError: Si la lecture échoue malgré les fallbacks.
        """
        # Vérification préalable de l'accessibilité
        if not self.ping():
            raise ConnectError(
                f"Le fichier CSV '{self.path}' n'est pas accessible."
            )

        # Détermination de l'encodage (détection si 'auto')
        encodage = (
            self._enc or self._sniff_encoding()
            if self.encoding == "auto"
            else self.encoding
        )

        # Détermination du délimiteur (détection si 'auto')
        delimiteur = (
            self._sep or self._sniff_sep()
            if self.sep == "auto"
            else self.sep
        )

        # Détermination du séparateur décimal
        # Heuristique : si le délimiteur est ';', le décimal est souvent ','
        if self.decimal == "auto":
            separateur_decimal = "," if delimiteur == ";" else "."
        else:
            separateur_decimal = self.decimal

        # Séquence d'encodages à essayer en cas d'échec
        encodages_fallback = [encodage, "utf-8", "latin-1", "cp1252"]
        # Déduplication tout en conservant l'ordre
        encodages_a_tester = list(dict.fromkeys(encodages_fallback))

        dernier_erreur = None

        # Tentative de lecture avec chaque encodage dans la liste
        for enc in encodages_a_tester:
            try:
                df = pd.read_csv(
                    self.path,
                    sep=delimiteur,
                    encoding=enc,
                    decimal=separateur_decimal,
                    nrows=n,
                    skiprows=skip,
                    on_bad_lines="warn",
                )
                logger.info(
                    "Fichier CSV '%s' lu avec succès : %d lignes, %d colonnes "
                    "(encodage: %s, délimiteur: '%s').",
                    self.path,
                    len(df),
                    len(df.columns),
                    enc,
                    delimiteur,
                )
                # Mise à jour de l'horodatage de lecture
                self._touch()
                return df

            except UnicodeDecodeError as erreur:
                # Échec d'encodage : on essaie le suivant
                logger.debug(
                    "Échec de lecture avec encodage '%s' pour '%s' : %s",
                    enc,
                    self.path,
                    erreur,
                )
                dernier_erreur = erreur
                continue

            except Exception as erreur:
                raise ReadError(
                    f"Erreur inattendue lors de la lecture de "
                    f"'{self.path}' : {erreur}"
                ) from erreur

        # Si tous les encodages ont échoué
        raise ReadError(
            f"Impossible de lire '{self.path}' avec les encodages "
            f"testés : {encodages_a_tester}. Dernière erreur : {dernier_erreur}"
        )

    def write(self, data: pd.DataFrame, index: bool = False) -> bool:
        """Écrit un DataFrame vers le fichier CSV de la source.

        Utilise le délimiteur détecté ou configuré. Si l'encodage est
        'auto', écrit en UTF-8 par défaut.

        Args:
            data (pd.DataFrame): Le DataFrame à écrire dans le fichier CSV.
            index (bool): Si True, inclut l'index du DataFrame dans le
                fichier CSV. Par défaut False.

        Returns:
            bool: True si l'écriture s'est déroulée avec succès.

        Raises:
            WriteError: Si l'écriture vers le fichier échoue.
        """
        # Détermination de l'encodage de sortie
        encodage_sortie = (
            "utf-8" if self.encoding == "auto" else self.encoding
        )

        # Détermination du délimiteur de sortie
        delimiteur_sortie = (
            self._sep or ","
            if self.sep == "auto"
            else self.sep
        )

        try:
            # Écriture du DataFrame au format CSV
            data.to_csv(
                self.path,
                sep=delimiteur_sortie,
                encoding=encodage_sortie,
                index=index,
            )
            logger.info(
                "DataFrame écrit avec succès vers '%s' (%d lignes).",
                self.path,
                len(data),
            )
            return True

        except OSError as erreur:
            raise WriteError(
                f"Impossible d'écrire vers '{self.path}' : {erreur}"
            ) from erreur

    def info(self) -> dict:
        """Retourne les métadonnées descriptives du fichier CSV.

        Effectue une lecture légère du fichier pour en extraire les
        statistiques sans charger toutes les données en mémoire.

        Returns:
            dict: Dictionnaire contenant les clés suivantes :
                - 'path' (str) : chemin du fichier.
                - 'kind' (str) : 'csv'.
                - 'encoding' (str) : encodage détecté ou configuré.
                - 'sep' (str) : délimiteur détecté ou configuré.
                - 'decimal' (str) : séparateur décimal utilisé.
                - 'rows' (int) : nombre de lignes de données.
                - 'cols' (int) : nombre de colonnes.
                - 'size_kb' (float) : taille du fichier en kilo-octets.
                - 'last_read' (str | None) : horodatage de la dernière lecture.
        """
        # Détection de l'encodage et du délimiteur si nécessaire
        encodage = (
            self._enc or self._sniff_encoding()
            if self.encoding == "auto"
            else self.encoding
        )
        delimiteur = (
            self._sep or self._sniff_sep()
            if self.sep == "auto"
            else self.sep
        )

        # Lecture rapide pour obtenir le nombre de lignes et colonnes
        try:
            df_apercu = pd.read_csv(
                self.path,
                sep=delimiteur,
                encoding=encodage,
                nrows=0,
            )
            nb_colonnes = len(df_apercu.columns)

            # Comptage des lignes sans charger tout le fichier en mémoire
            with open(self.path, encoding=encodage, errors="replace") as f:
                nb_lignes = sum(1 for _ in f) - 1  # -1 pour l'en-tête

        except Exception:
            nb_lignes = -1
            nb_colonnes = -1

        # Calcul de la taille du fichier
        taille_kb = os.path.getsize(self.path) / 1024 if os.path.isfile(
            self.path
        ) else 0.0

        return {
            "path": self.path,
            "kind": "csv",
            "encoding": encodage,
            "sep": delimiteur,
            "decimal": self.decimal if self.decimal != "auto" else "inféré",
            "rows": nb_lignes,
            "cols": nb_colonnes,
            "size_kb": round(taille_kb, 2),
            "last_read": (
                self.last_read.isoformat() if self.last_read else None
            ),
        }

    def ping(self) -> bool:
        """Vérifie que le fichier CSV existe et est lisible.

        Returns:
            bool: True si le fichier est accessible en lecture, False sinon.
        """
        # Vérification de l'existence et de la lisibilité du fichier
        est_accessible = os.path.isfile(self.path) and os.access(
            self.path, os.R_OK
        )

        if not est_accessible:
            logger.warning(
                "Le fichier CSV '%s' n'existe pas ou n'est pas lisible.",
                self.path,
            )

        return est_accessible

__init__ 

__init__(path: str, encoding: str = 'auto', sep: str = 'auto', decimal: str = 'auto') -> None

Initialise la source CSV avec détection automatique des paramètres.

Parameters:

Name Type Description Default
path str

Chemin vers le fichier CSV à lire.

 required
encoding str

Encodage du fichier. 'auto' pour la détection automatique via chardet. Par défaut 'auto'.

 'auto'
sep str

Délimiteur de colonnes. 'auto' pour l'auto- détection. Par défaut 'auto'.

 'auto'
decimal str

Séparateur décimal. 'auto' pour l'inférence (détecte les fichiers français avec ','). Par défaut 'auto'.

 'auto'
Source code in kadi/kidas/sources/csv_source.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def __init__(
    self,
    path: str,
    encoding: str = "auto",
    sep: str = "auto",
    decimal: str = "auto",
) -> None:
    """Initialise la source CSV avec détection automatique des paramètres.

    Args:
        path (str): Chemin vers le fichier CSV à lire.
        encoding (str): Encodage du fichier. 'auto' pour la détection
            automatique via chardet. Par défaut 'auto'.
        sep (str): Délimiteur de colonnes. 'auto' pour l'auto-
            détection. Par défaut 'auto'.
        decimal (str): Séparateur décimal. 'auto' pour l'inférence
            (détecte les fichiers français avec ','). Par défaut 'auto'.
    """
    # Initialisation de la classe parente avec le type 'csv'
    super().__init__(
        path=path,
        kind="csv",
        encoding=encoding,
    )

    # Paramètres de lecture (peuvent être 'auto' avant détection)
    self.sep: str = sep
    self.decimal: str = decimal

    # Cache interne pour éviter de re-détecter à chaque lecture
    self._sep: Optional[str] = None
    self._enc: Optional[str] = None

read 

read(n: Optional[int] = None, skip: Optional[int] = None) -> pd.DataFrame

Lit le fichier CSV et retourne son contenu sous forme de DataFrame.

Effectue la détection automatique de l'encodage et du délimiteur si nécessaire. Tente plusieurs encodages en fallback si la lecture principale échoue.

Parameters:

Name Type Description Default
n int | None

Nombre maximum de lignes à lire. None pour lire toutes les lignes. Par défaut None.

 None
skip int | None

Nombre de lignes à ignorer en début de fichier (hors en-tête). Par défaut None.

 None

Returns:

Type Description
DataFrame

pd.DataFrame: Les données du fichier CSV.

Raises:

Type Description
ConnectError

Si le fichier n'est pas accessible.
ReadError

Si la lecture échoue malgré les fallbacks.
Source code in kadi/kidas/sources/csv_source.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def read(
    self,
    n: Optional[int] = None,
    skip: Optional[int] = None,
) -> pd.DataFrame:
    """Lit le fichier CSV et retourne son contenu sous forme de DataFrame.

    Effectue la détection automatique de l'encodage et du délimiteur
    si nécessaire. Tente plusieurs encodages en fallback si la lecture
    principale échoue.

    Args:
        n (int | None): Nombre maximum de lignes à lire. None pour
            lire toutes les lignes. Par défaut None.
        skip (int | None): Nombre de lignes à ignorer en début
            de fichier (hors en-tête). Par défaut None.

    Returns:
        pd.DataFrame: Les données du fichier CSV.

    Raises:
        ConnectError: Si le fichier n'est pas accessible.
        ReadError: Si la lecture échoue malgré les fallbacks.
    """
    # Vérification préalable de l'accessibilité
    if not self.ping():
        raise ConnectError(
            f"Le fichier CSV '{self.path}' n'est pas accessible."
        )

    # Détermination de l'encodage (détection si 'auto')
    encodage = (
        self._enc or self._sniff_encoding()
        if self.encoding == "auto"
        else self.encoding
    )

    # Détermination du délimiteur (détection si 'auto')
    delimiteur = (
        self._sep or self._sniff_sep()
        if self.sep == "auto"
        else self.sep
    )

    # Détermination du séparateur décimal
    # Heuristique : si le délimiteur est ';', le décimal est souvent ','
    if self.decimal == "auto":
        separateur_decimal = "," if delimiteur == ";" else "."
    else:
        separateur_decimal = self.decimal

    # Séquence d'encodages à essayer en cas d'échec
    encodages_fallback = [encodage, "utf-8", "latin-1", "cp1252"]
    # Déduplication tout en conservant l'ordre
    encodages_a_tester = list(dict.fromkeys(encodages_fallback))

    dernier_erreur = None

    # Tentative de lecture avec chaque encodage dans la liste
    for enc in encodages_a_tester:
        try:
            df = pd.read_csv(
                self.path,
                sep=delimiteur,
                encoding=enc,
                decimal=separateur_decimal,
                nrows=n,
                skiprows=skip,
                on_bad_lines="warn",
            )
            logger.info(
                "Fichier CSV '%s' lu avec succès : %d lignes, %d colonnes "
                "(encodage: %s, délimiteur: '%s').",
                self.path,
                len(df),
                len(df.columns),
                enc,
                delimiteur,
            )
            # Mise à jour de l'horodatage de lecture
            self._touch()
            return df

        except UnicodeDecodeError as erreur:
            # Échec d'encodage : on essaie le suivant
            logger.debug(
                "Échec de lecture avec encodage '%s' pour '%s' : %s",
                enc,
                self.path,
                erreur,
            )
            dernier_erreur = erreur
            continue

        except Exception as erreur:
            raise ReadError(
                f"Erreur inattendue lors de la lecture de "
                f"'{self.path}' : {erreur}"
            ) from erreur

    # Si tous les encodages ont échoué
    raise ReadError(
        f"Impossible de lire '{self.path}' avec les encodages "
        f"testés : {encodages_a_tester}. Dernière erreur : {dernier_erreur}"
    )

write 

write(data: DataFrame, index: bool = False) -> bool

Écrit un DataFrame vers le fichier CSV de la source.

Utilise le délimiteur détecté ou configuré. Si l'encodage est 'auto', écrit en UTF-8 par défaut.

Parameters:

Name Type Description Default
data DataFrame

Le DataFrame à écrire dans le fichier CSV.

 required
index bool

Si True, inclut l'index du DataFrame dans le fichier CSV. Par défaut False.

 False

Returns:

Name Type Description
bool bool

True si l'écriture s'est déroulée avec succès.

Raises:

Type Description
WriteError

Si l'écriture vers le fichier échoue.
Source code in kadi/kidas/sources/csv_source.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
def write(self, data: pd.DataFrame, index: bool = False) -> bool:
    """Écrit un DataFrame vers le fichier CSV de la source.

    Utilise le délimiteur détecté ou configuré. Si l'encodage est
    'auto', écrit en UTF-8 par défaut.

    Args:
        data (pd.DataFrame): Le DataFrame à écrire dans le fichier CSV.
        index (bool): Si True, inclut l'index du DataFrame dans le
            fichier CSV. Par défaut False.

    Returns:
        bool: True si l'écriture s'est déroulée avec succès.

    Raises:
        WriteError: Si l'écriture vers le fichier échoue.
    """
    # Détermination de l'encodage de sortie
    encodage_sortie = (
        "utf-8" if self.encoding == "auto" else self.encoding
    )

    # Détermination du délimiteur de sortie
    delimiteur_sortie = (
        self._sep or ","
        if self.sep == "auto"
        else self.sep
    )

    try:
        # Écriture du DataFrame au format CSV
        data.to_csv(
            self.path,
            sep=delimiteur_sortie,
            encoding=encodage_sortie,
            index=index,
        )
        logger.info(
            "DataFrame écrit avec succès vers '%s' (%d lignes).",
            self.path,
            len(data),
        )
        return True

    except OSError as erreur:
        raise WriteError(
            f"Impossible d'écrire vers '{self.path}' : {erreur}"
        ) from erreur

info 

info() -> dict

Retourne les métadonnées descriptives du fichier CSV.

Effectue une lecture légère du fichier pour en extraire les statistiques sans charger toutes les données en mémoire.

Returns:

Name Type Description
dict dict

Dictionnaire contenant les clés suivantes : - 'path' (str) : chemin du fichier. - 'kind' (str) : 'csv'. - 'encoding' (str) : encodage détecté ou configuré. - 'sep' (str) : délimiteur détecté ou configuré. - 'decimal' (str) : séparateur décimal utilisé. - 'rows' (int) : nombre de lignes de données. - 'cols' (int) : nombre de colonnes. - 'size_kb' (float) : taille du fichier en kilo-octets. - 'last_read' (str | None) : horodatage de la dernière lecture.
Source code in kadi/kidas/sources/csv_source.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
def info(self) -> dict:
    """Retourne les métadonnées descriptives du fichier CSV.

    Effectue une lecture légère du fichier pour en extraire les
    statistiques sans charger toutes les données en mémoire.

    Returns:
        dict: Dictionnaire contenant les clés suivantes :
            - 'path' (str) : chemin du fichier.
            - 'kind' (str) : 'csv'.
            - 'encoding' (str) : encodage détecté ou configuré.
            - 'sep' (str) : délimiteur détecté ou configuré.
            - 'decimal' (str) : séparateur décimal utilisé.
            - 'rows' (int) : nombre de lignes de données.
            - 'cols' (int) : nombre de colonnes.
            - 'size_kb' (float) : taille du fichier en kilo-octets.
            - 'last_read' (str | None) : horodatage de la dernière lecture.
    """
    # Détection de l'encodage et du délimiteur si nécessaire
    encodage = (
        self._enc or self._sniff_encoding()
        if self.encoding == "auto"
        else self.encoding
    )
    delimiteur = (
        self._sep or self._sniff_sep()
        if self.sep == "auto"
        else self.sep
    )

    # Lecture rapide pour obtenir le nombre de lignes et colonnes
    try:
        df_apercu = pd.read_csv(
            self.path,
            sep=delimiteur,
            encoding=encodage,
            nrows=0,
        )
        nb_colonnes = len(df_apercu.columns)

        # Comptage des lignes sans charger tout le fichier en mémoire
        with open(self.path, encoding=encodage, errors="replace") as f:
            nb_lignes = sum(1 for _ in f) - 1  # -1 pour l'en-tête

    except Exception:
        nb_lignes = -1
        nb_colonnes = -1

    # Calcul de la taille du fichier
    taille_kb = os.path.getsize(self.path) / 1024 if os.path.isfile(
        self.path
    ) else 0.0

    return {
        "path": self.path,
        "kind": "csv",
        "encoding": encodage,
        "sep": delimiteur,
        "decimal": self.decimal if self.decimal != "auto" else "inféré",
        "rows": nb_lignes,
        "cols": nb_colonnes,
        "size_kb": round(taille_kb, 2),
        "last_read": (
            self.last_read.isoformat() if self.last_read else None
        ),
    }

ping 

ping() -> bool

Vérifie que le fichier CSV existe et est lisible.

Returns:

Name Type Description
bool bool

True si le fichier est accessible en lecture, False sinon.
Source code in kadi/kidas/sources/csv_source.py 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
def ping(self) -> bool:
    """Vérifie que le fichier CSV existe et est lisible.

    Returns:
        bool: True si le fichier est accessible en lecture, False sinon.
    """
    # Vérification de l'existence et de la lisibilité du fichier
    est_accessible = os.path.isfile(self.path) and os.access(
        self.path, os.R_OK
    )

    if not est_accessible:
        logger.warning(
            "Le fichier CSV '%s' n'existe pas ou n'est pas lisible.",
            self.path,
        )

    return est_accessible

ExcelSource

Bases: Source

Source de données pour les fichiers Excel agricoles (.xls, .xlsx).

Gère la lecture robuste de fichiers Excel avec des structures complexes : cellules fusionnées (forward fill), en-têtes sur des lignes variables, lignes vides intercalées et feuilles multiples.

Attributs

path (str): Chemin absolu ou relatif vers le fichier Excel. sheet (str | int): Nom ou index de la feuille à lire. header (str | int): Ligne d'en-tête. 'auto' pour la détection automatique. _header (int | None): Ligne d'en-tête détectée en cache.

Exemple

source = ExcelSource('prix_marches_2024.xlsx') print(source.sheets()) ['Janvier', 'Fevrier'] df = source.read(sheet='Janvier')

Source code in kadi/kidas/sources/excel_source.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
class ExcelSource(Source):
    """Source de données pour les fichiers Excel agricoles (.xls, .xlsx).

    Gère la lecture robuste de fichiers Excel avec des structures complexes :
    cellules fusionnées (forward fill), en-têtes sur des lignes variables,
    lignes vides intercalées et feuilles multiples.

    Attributs:
        path (str): Chemin absolu ou relatif vers le fichier Excel.
        sheet (str | int): Nom ou index de la feuille à lire.
        header (str | int): Ligne d'en-tête. 'auto' pour la détection
            automatique.
        _header (int | None): Ligne d'en-tête détectée en cache.

    Exemple:
        >>> source = ExcelSource('prix_marches_2024.xlsx')
        >>> print(source.sheets())
        ['Janvier', 'Fevrier']
        >>> df = source.read(sheet='Janvier')
    """

    def __init__(
        self,
        path: str,
        sheet: Union[str, int] = 0,
        header: Union[str, int] = "auto",
    ) -> None:
        """Initialise la source Excel avec détection automatique de l'en-tête.

        Args:
            path (str): Chemin vers le fichier Excel à lire.
            sheet (str | int): Nom ou index de la feuille par défaut.
                Par défaut 0 (première feuille).
            header (str | int): Ligne contenant les en-têtes de colonnes.
                'auto' pour la détection automatique. Par défaut 'auto'.
        """
        # Initialisation de la classe parente avec le type 'excel'
        super().__init__(
            path=path,
            kind="excel",
            encoding="utf-8",
        )

        # Feuille par défaut (nom ou index)
        self.sheet: Union[str, int] = sheet

        # Ligne d'en-tête (peut être 'auto' avant détection)
        self.header: Union[str, int] = header

        # Cache interne pour la ligne d'en-tête détectée
        self._header: Optional[int] = None

    def _sniff_header(self) -> int:
        """Détecte automatiquement la ligne contenant les en-têtes de colonnes.

        Inspecte les premières lignes du fichier et cherche la première ligne
        contenant des valeurs non-null et hétérogènes (mélange de types),
        ce qui caractérise une ligne d'en-tête typique.

        Returns:
            int: Index (0-based) de la ligne d'en-tête détectée.

        Raises:
            ConnectError: Si le fichier Excel n'est pas accessible.
            ReadError: Si la détection échoue.
        """
        if not self.ping():
            raise ConnectError(
                f"Fichier Excel introuvable : '{self.path}'"
            )

        try:
            # Lecture des premières lignes sans en-tête pour l'inspection
            df_brut = pd.read_excel(
                self.path,
                sheet_name=self.sheet,
                header=None,
                nrows=_MAX_HEADER_SCAN_ROWS,
            )

            # Recherche de la ligne la plus hétérogène (types mixtes = en-tête)
            ligne_entete = 0
            meilleur_score = -1

            for i, ligne in df_brut.iterrows():
                # Filtrage des valeurs non nulles
                valeurs_non_null = ligne.dropna()

                if len(valeurs_non_null) == 0:
                    # Ligne vide : on passe
                    continue

                # Score = nb de valeurs non-null (ligne dense = probable en-tête)
                score = len(valeurs_non_null)

                if score > meilleur_score:
                    meilleur_score = score
                    ligne_entete = int(i)

            logger.debug(
                "En-tête détecté à la ligne %d pour '%s'.",
                ligne_entete,
                self.path,
            )

            # Mise en cache du résultat
            self._header = ligne_entete
            return ligne_entete

        except Exception as erreur:
            raise ReadError(
                f"Impossible de détecter l'en-tête dans '{self.path}' : {erreur}"
            ) from erreur

    def sheets(self) -> List[str]:
        """Retourne la liste des noms de feuilles du fichier Excel.

        Returns:
            List[str]: Liste ordonnée des noms de feuilles dans le fichier.

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
            ReadError: Si la lecture de la structure échoue.
        """
        if not self.ping():
            raise ConnectError(
                f"Fichier Excel introuvable : '{self.path}'"
            )

        try:
            # Lecture de la liste des feuilles via ExcelFile
            with pd.ExcelFile(self.path) as classeur:
                feuilles = classeur.sheet_names

            logger.debug(
                "Feuilles trouvées dans '%s' : %s.", self.path, feuilles
            )
            return feuilles

        except Exception as erreur:
            raise ReadError(
                f"Impossible de lister les feuilles de '{self.path}' : {erreur}"
            ) from erreur

    def sheet_info(self, sheet: Union[str, int]) -> dict:
        """Retourne les métadonnées d'une feuille spécifique.

        Args:
            sheet (str | int): Nom ou index de la feuille à inspecter.

        Returns:
            dict: Dictionnaire contenant les clés :
                - 'sheet' (str) : nom de la feuille.
                - 'rows' (int) : nombre de lignes de données.
                - 'cols' (int) : nombre de colonnes.
                - 'columns' (list) : liste des noms de colonnes.

        Raises:
            ReadError: Si la lecture de la feuille échoue.
        """
        try:
            # Lecture de la feuille entière pour les métadonnées
            df = pd.read_excel(self.path, sheet_name=sheet)
            return {
                "sheet": sheet,
                "rows": len(df),
                "cols": len(df.columns),
                "columns": list(df.columns),
            }
        except Exception as erreur:
            raise ReadError(
                f"Impossible de lire la feuille '{sheet}' "
                f"dans '{self.path}' : {erreur}"
            ) from erreur

    def _unmerge(self, df: pd.DataFrame) -> pd.DataFrame:
        """Résout les cellules fusionnées par un forward fill vertical.

        Dans les fichiers Excel africains, les colonnes 'Commune' ou 'Marché'
        contiennent souvent des cellules fusionnées sur plusieurs lignes.
        Cette méthode propage la dernière valeur non-null vers le bas.

        Args:
            df (pd.DataFrame): Le DataFrame lu depuis Excel (avant forward fill).

        Returns:
            pd.DataFrame: DataFrame avec les cellules fusionnées résolues.
        """
        # Forward fill sur toutes les colonnes pour résoudre les cellules fusionnées
        df_resolu = df.ffill(axis=0)
        logger.debug(
            "Forward fill appliqué sur le DataFrame (%d lignes).", len(df_resolu)
        )
        return df_resolu

    def read(
        self,
        sheet: Optional[Union[str, int]] = None,
    ) -> pd.DataFrame:
        """Lit une feuille Excel et retourne son contenu sous forme de DataFrame.

        Détecte automatiquement la ligne d'en-tête et applique un forward fill
        pour résoudre les cellules fusionnées.

        Args:
            sheet (str | int | None): Feuille à lire. Si None, utilise
                la valeur définie à l'initialisation. Par défaut None.

        Returns:
            pd.DataFrame: Les données de la feuille Excel.

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
            ReadError: Si la lecture échoue.
        """
        if not self.ping():
            raise ConnectError(
                f"Fichier Excel inaccessible : '{self.path}'"
            )

        # Détermination de la feuille à lire
        feuille = sheet if sheet is not None else self.sheet

        # Détermination de la ligne d'en-tête
        if self.header == "auto":
            ligne_entete = (
                self._header
                if self._header is not None
                else self._sniff_header()
            )
        else:
            ligne_entete = int(self.header)

        try:
            # Lecture du fichier Excel avec la ligne d'en-tête correcte
            df = pd.read_excel(
                self.path,
                sheet_name=feuille,
                header=ligne_entete,
            )

            # Suppression des lignes entièrement vides
            df = df.dropna(how="all")

            # Résolution des cellules fusionnées par forward fill
            df = self._unmerge(df)

            logger.info(
                "Fichier Excel '%s' (feuille '%s') lu : %d lignes, %d colonnes.",
                self.path,
                feuille,
                len(df),
                len(df.columns),
            )

            # Mise à jour de l'horodatage de lecture
            self._touch()
            return df

        except Exception as erreur:
            raise ReadError(
                f"Erreur lors de la lecture de '{self.path}' "
                f"(feuille '{feuille}') : {erreur}"
            ) from erreur

    def write(
        self,
        data: pd.DataFrame,
        sheet: str = "Sheet1",
    ) -> bool:
        """Écrit un DataFrame vers le fichier Excel de la source.

        Args:
            data (pd.DataFrame): Les données à écrire dans le fichier Excel.
            sheet (str): Nom de la feuille de destination.
                Par défaut 'Sheet1'.

        Returns:
            bool: True si l'écriture s'est déroulée avec succès.

        Raises:
            WriteError: Si l'écriture vers le fichier échoue.
        """
        try:
            # Écriture du DataFrame en format Excel
            data.to_excel(self.path, sheet_name=sheet, index=False)
            logger.info(
                "DataFrame écrit avec succès vers '%s' (feuille '%s', %d lignes).",
                self.path,
                sheet,
                len(data),
            )
            return True

        except Exception as erreur:
            raise WriteError(
                f"Impossible d'écrire vers '{self.path}' : {erreur}"
            ) from erreur

    def info(self) -> dict:
        """Retourne les métadonnées descriptives du fichier Excel.

        Returns:
            dict: Dictionnaire contenant les clés suivantes :
                - 'path' (str) : chemin du fichier.
                - 'kind' (str) : 'excel'.
                - 'sheets' (List[str]) : liste des feuilles disponibles.
                - 'active_sheet' (str | int) : feuille active courante.
                - 'detected_header' (int | None) : ligne d'en-tête détectée.
                - 'size_kb' (float) : taille du fichier en kilo-octets.
                - 'last_read' (str | None) : horodatage de la dernière lecture.
        """
        # Récupération de la liste des feuilles
        try:
            feuilles = self.sheets()
        except ReadError:
            feuilles = []

        # Calcul de la taille du fichier
        taille_kb = os.path.getsize(self.path) / 1024 if os.path.isfile(
            self.path
        ) else 0.0

        return {
            "path": self.path,
            "kind": "excel",
            "sheets": feuilles,
            "active_sheet": self.sheet,
            "detected_header": self._header,
            "size_kb": round(taille_kb, 2),
            "last_read": (
                self.last_read.isoformat() if self.last_read else None
            ),
        }

    def ping(self) -> bool:
        """Vérifie que le fichier Excel existe et est lisible.

        Returns:
            bool: True si le fichier est accessible en lecture, False sinon.
        """
        # Vérification de l'existence et de la lisibilité du fichier
        est_accessible = os.path.isfile(self.path) and os.access(
            self.path, os.R_OK
        )

        if not est_accessible:
            logger.warning(
                "Le fichier Excel '%s' n'existe pas ou n'est pas lisible.",
                self.path,
            )

        return est_accessible

__init__ 

__init__(path: str, sheet: Union[str, int] = 0, header: Union[str, int] = 'auto') -> None

Initialise la source Excel avec détection automatique de l'en-tête.

Parameters:

Name Type Description Default
path str

Chemin vers le fichier Excel à lire.

 required
sheet str | int

Nom ou index de la feuille par défaut. Par défaut 0 (première feuille).

 0
header str | int

Ligne contenant les en-têtes de colonnes. 'auto' pour la détection automatique. Par défaut 'auto'.

 'auto'
Source code in kadi/kidas/sources/excel_source.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(
    self,
    path: str,
    sheet: Union[str, int] = 0,
    header: Union[str, int] = "auto",
) -> None:
    """Initialise la source Excel avec détection automatique de l'en-tête.

    Args:
        path (str): Chemin vers le fichier Excel à lire.
        sheet (str | int): Nom ou index de la feuille par défaut.
            Par défaut 0 (première feuille).
        header (str | int): Ligne contenant les en-têtes de colonnes.
            'auto' pour la détection automatique. Par défaut 'auto'.
    """
    # Initialisation de la classe parente avec le type 'excel'
    super().__init__(
        path=path,
        kind="excel",
        encoding="utf-8",
    )

    # Feuille par défaut (nom ou index)
    self.sheet: Union[str, int] = sheet

    # Ligne d'en-tête (peut être 'auto' avant détection)
    self.header: Union[str, int] = header

    # Cache interne pour la ligne d'en-tête détectée
    self._header: Optional[int] = None

sheets 

sheets() -> List[str]

Retourne la liste des noms de feuilles du fichier Excel.

Returns:

Type Description
List[str]

List[str]: Liste ordonnée des noms de feuilles dans le fichier.

Raises:

Type Description
ConnectError

Si le fichier n'est pas accessible.
ReadError

Si la lecture de la structure échoue.
Source code in kadi/kidas/sources/excel_source.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def sheets(self) -> List[str]:
    """Retourne la liste des noms de feuilles du fichier Excel.

    Returns:
        List[str]: Liste ordonnée des noms de feuilles dans le fichier.

    Raises:
        ConnectError: Si le fichier n'est pas accessible.
        ReadError: Si la lecture de la structure échoue.
    """
    if not self.ping():
        raise ConnectError(
            f"Fichier Excel introuvable : '{self.path}'"
        )

    try:
        # Lecture de la liste des feuilles via ExcelFile
        with pd.ExcelFile(self.path) as classeur:
            feuilles = classeur.sheet_names

        logger.debug(
            "Feuilles trouvées dans '%s' : %s.", self.path, feuilles
        )
        return feuilles

    except Exception as erreur:
        raise ReadError(
            f"Impossible de lister les feuilles de '{self.path}' : {erreur}"
        ) from erreur

sheet_info 

sheet_info(sheet: Union[str, int]) -> dict

Retourne les métadonnées d'une feuille spécifique.

Parameters:

Name Type Description Default
sheet str | int

Nom ou index de la feuille à inspecter.

 required

Returns:

Name Type Description
dict dict

Dictionnaire contenant les clés : - 'sheet' (str) : nom de la feuille. - 'rows' (int) : nombre de lignes de données. - 'cols' (int) : nombre de colonnes. - 'columns' (list) : liste des noms de colonnes.

Raises:

Type Description
ReadError

Si la lecture de la feuille échoue.
Source code in kadi/kidas/sources/excel_source.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def sheet_info(self, sheet: Union[str, int]) -> dict:
    """Retourne les métadonnées d'une feuille spécifique.

    Args:
        sheet (str | int): Nom ou index de la feuille à inspecter.

    Returns:
        dict: Dictionnaire contenant les clés :
            - 'sheet' (str) : nom de la feuille.
            - 'rows' (int) : nombre de lignes de données.
            - 'cols' (int) : nombre de colonnes.
            - 'columns' (list) : liste des noms de colonnes.

    Raises:
        ReadError: Si la lecture de la feuille échoue.
    """
    try:
        # Lecture de la feuille entière pour les métadonnées
        df = pd.read_excel(self.path, sheet_name=sheet)
        return {
            "sheet": sheet,
            "rows": len(df),
            "cols": len(df.columns),
            "columns": list(df.columns),
        }
    except Exception as erreur:
        raise ReadError(
            f"Impossible de lire la feuille '{sheet}' "
            f"dans '{self.path}' : {erreur}"
        ) from erreur

read 

read(sheet: Optional[Union[str, int]] = None) -> pd.DataFrame

Lit une feuille Excel et retourne son contenu sous forme de DataFrame.

Détecte automatiquement la ligne d'en-tête et applique un forward fill pour résoudre les cellules fusionnées.

Parameters:

Name Type Description Default
sheet str | int | None

Feuille à lire. Si None, utilise la valeur définie à l'initialisation. Par défaut None.

 None

Returns:

Type Description
DataFrame

pd.DataFrame: Les données de la feuille Excel.

Raises:

Type Description
ConnectError

Si le fichier n'est pas accessible.
ReadError

Si la lecture échoue.
Source code in kadi/kidas/sources/excel_source.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def read(
    self,
    sheet: Optional[Union[str, int]] = None,
) -> pd.DataFrame:
    """Lit une feuille Excel et retourne son contenu sous forme de DataFrame.

    Détecte automatiquement la ligne d'en-tête et applique un forward fill
    pour résoudre les cellules fusionnées.

    Args:
        sheet (str | int | None): Feuille à lire. Si None, utilise
            la valeur définie à l'initialisation. Par défaut None.

    Returns:
        pd.DataFrame: Les données de la feuille Excel.

    Raises:
        ConnectError: Si le fichier n'est pas accessible.
        ReadError: Si la lecture échoue.
    """
    if not self.ping():
        raise ConnectError(
            f"Fichier Excel inaccessible : '{self.path}'"
        )

    # Détermination de la feuille à lire
    feuille = sheet if sheet is not None else self.sheet

    # Détermination de la ligne d'en-tête
    if self.header == "auto":
        ligne_entete = (
            self._header
            if self._header is not None
            else self._sniff_header()
        )
    else:
        ligne_entete = int(self.header)

    try:
        # Lecture du fichier Excel avec la ligne d'en-tête correcte
        df = pd.read_excel(
            self.path,
            sheet_name=feuille,
            header=ligne_entete,
        )

        # Suppression des lignes entièrement vides
        df = df.dropna(how="all")

        # Résolution des cellules fusionnées par forward fill
        df = self._unmerge(df)

        logger.info(
            "Fichier Excel '%s' (feuille '%s') lu : %d lignes, %d colonnes.",
            self.path,
            feuille,
            len(df),
            len(df.columns),
        )

        # Mise à jour de l'horodatage de lecture
        self._touch()
        return df

    except Exception as erreur:
        raise ReadError(
            f"Erreur lors de la lecture de '{self.path}' "
            f"(feuille '{feuille}') : {erreur}"
        ) from erreur

write 

write(data: DataFrame, sheet: str = 'Sheet1') -> bool

Écrit un DataFrame vers le fichier Excel de la source.

Parameters:

Name Type Description Default
data DataFrame

Les données à écrire dans le fichier Excel.

 required
sheet str

Nom de la feuille de destination. Par défaut 'Sheet1'.

 'Sheet1'

Returns:

Name Type Description
bool bool

True si l'écriture s'est déroulée avec succès.

Raises:

Type Description
WriteError

Si l'écriture vers le fichier échoue.
Source code in kadi/kidas/sources/excel_source.py 
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def write(
    self,
    data: pd.DataFrame,
    sheet: str = "Sheet1",
) -> bool:
    """Écrit un DataFrame vers le fichier Excel de la source.

    Args:
        data (pd.DataFrame): Les données à écrire dans le fichier Excel.
        sheet (str): Nom de la feuille de destination.
            Par défaut 'Sheet1'.

    Returns:
        bool: True si l'écriture s'est déroulée avec succès.

    Raises:
        WriteError: Si l'écriture vers le fichier échoue.
    """
    try:
        # Écriture du DataFrame en format Excel
        data.to_excel(self.path, sheet_name=sheet, index=False)
        logger.info(
            "DataFrame écrit avec succès vers '%s' (feuille '%s', %d lignes).",
            self.path,
            sheet,
            len(data),
        )
        return True

    except Exception as erreur:
        raise WriteError(
            f"Impossible d'écrire vers '{self.path}' : {erreur}"
        ) from erreur

info 

info() -> dict

Retourne les métadonnées descriptives du fichier Excel.

Returns:

Name Type Description
dict dict

Dictionnaire contenant les clés suivantes : - 'path' (str) : chemin du fichier. - 'kind' (str) : 'excel'. - 'sheets' (List[str]) : liste des feuilles disponibles. - 'active_sheet' (str | int) : feuille active courante. - 'detected_header' (int | None) : ligne d'en-tête détectée. - 'size_kb' (float) : taille du fichier en kilo-octets. - 'last_read' (str | None) : horodatage de la dernière lecture.
Source code in kadi/kidas/sources/excel_source.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def info(self) -> dict:
    """Retourne les métadonnées descriptives du fichier Excel.

    Returns:
        dict: Dictionnaire contenant les clés suivantes :
            - 'path' (str) : chemin du fichier.
            - 'kind' (str) : 'excel'.
            - 'sheets' (List[str]) : liste des feuilles disponibles.
            - 'active_sheet' (str | int) : feuille active courante.
            - 'detected_header' (int | None) : ligne d'en-tête détectée.
            - 'size_kb' (float) : taille du fichier en kilo-octets.
            - 'last_read' (str | None) : horodatage de la dernière lecture.
    """
    # Récupération de la liste des feuilles
    try:
        feuilles = self.sheets()
    except ReadError:
        feuilles = []

    # Calcul de la taille du fichier
    taille_kb = os.path.getsize(self.path) / 1024 if os.path.isfile(
        self.path
    ) else 0.0

    return {
        "path": self.path,
        "kind": "excel",
        "sheets": feuilles,
        "active_sheet": self.sheet,
        "detected_header": self._header,
        "size_kb": round(taille_kb, 2),
        "last_read": (
            self.last_read.isoformat() if self.last_read else None
        ),
    }

ping 

ping() -> bool

Vérifie que le fichier Excel existe et est lisible.

Returns:

Name Type Description
bool bool

True si le fichier est accessible en lecture, False sinon.
Source code in kadi/kidas/sources/excel_source.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
def ping(self) -> bool:
    """Vérifie que le fichier Excel existe et est lisible.

    Returns:
        bool: True si le fichier est accessible en lecture, False sinon.
    """
    # Vérification de l'existence et de la lisibilité du fichier
    est_accessible = os.path.isfile(self.path) and os.access(
        self.path, os.R_OK
    )

    if not est_accessible:
        logger.warning(
            "Le fichier Excel '%s' n'existe pas ou n'est pas lisible.",
            self.path,
        )

    return est_accessible

JSONSource

Bases: Source

Source de données pour les fichiers JSON agricoles.

Gère la lecture de fichiers JSON plats ou fortement imbriqués. L'aplatissement automatique des clés imbriquées utilise la notation pointée (ex: {'location': {'lat': 9.3}} -> {'location.lat': 9.3}).

Attributs

path (str | None): Chemin absolu ou relatif vers le fichier JSON. '' si la source est un dictionnaire Python. _raw (dict | None): Dictionnaire Python fourni directement comme source au lieu d'un fichier.

Exemple

Depuis un fichier

source = JSONSource('donnees_fao.json') df = source.read(flatten=True)

Depuis un dictionnaire Python

data = {'location': {'lat': 9.3, 'lon': 2.4}, 'crop': 'maize'} source = JSONSource(data) df = source.read()

Source code in kadi/kidas/sources/json_source.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
class JSONSource(Source):
    """Source de données pour les fichiers JSON agricoles.

    Gère la lecture de fichiers JSON plats ou fortement imbriqués.
    L'aplatissement automatique des clés imbriquées utilise la notation
    pointée (ex: {'location': {'lat': 9.3}} -> {'location.lat': 9.3}).

    Attributs:
        path (str | None): Chemin absolu ou relatif vers le fichier JSON.
            '<dict_en_memoire>' si la source est un dictionnaire Python.
        _raw (dict | None): Dictionnaire Python fourni directement
            comme source au lieu d'un fichier.

    Exemple:
        >>> # Depuis un fichier
        >>> source = JSONSource('donnees_fao.json')
        >>> df = source.read(flatten=True)
        >>>
        >>> # Depuis un dictionnaire Python
        >>> data = {'location': {'lat': 9.3, 'lon': 2.4}, 'crop': 'maize'}
        >>> source = JSONSource(data)
        >>> df = source.read()
    """

    def __init__(
        self,
        file_path_or_dict: Union[str, dict],
    ) -> None:
        """Initialise la source JSON depuis un fichier ou un dictionnaire.

        Args:
            file_path_or_dict (str | dict): Chemin vers le fichier JSON
                ou un dictionnaire Python à utiliser directement comme source.
        """
        # Détection du type de source : fichier ou dictionnaire en mémoire
        if isinstance(file_path_or_dict, dict):
            # Source : dictionnaire Python en mémoire
            self._raw: Optional[dict] = file_path_or_dict
            chemin = "<dict_en_memoire>"
        else:
            # Source : fichier JSON sur le disque
            self._raw = None
            chemin = file_path_or_dict

        # Initialisation de la classe parente avec le type 'json'
        super().__init__(
            path=chemin,
            kind="json",
            encoding="utf-8",
        )

        # Chemin du fichier (None si source est un dict)
        self._file: Optional[str] = (
            None if isinstance(file_path_or_dict, dict) else file_path_or_dict
        )

    def flatten(self, obj: Dict[str, Any], sep: str = ".") -> dict:
        """Aplatit récursivement un objet JSON imbriqué.

        Transforme les clés imbriquées en clés à notation pointée.
        Exemple : {'location': {'lat': 9.3}} -> {'location.lat': 9.3}.

        Args:
            obj (dict): L'objet JSON à aplatir (peut contenir des dicts
                et des listes imbriqués).
            sep (str): Caractère séparateur entre les niveaux de clés.
                Par défaut '.'.

        Returns:
            dict: Dictionnaire aplati avec des clés à notation pointée.
        """
        # Dictionnaire de sortie aplati
        resultat: dict = {}

        def _aplatir(o: Any, prefixe: str = "") -> None:
            """Fonction récursive interne réalisant l'aplatissement.

            Args:
                o (Any): L'objet courant à aplatir.
                prefixe (str): Le préfixe des clés parentes accumulées.
            """
            if isinstance(o, dict):
                # Récursion sur chaque clé du dictionnaire
                for cle, valeur in o.items():
                    # Construction de la clé complète avec préfixe
                    cle_complete = f"{prefixe}{sep}{cle}" if prefixe else cle
                    _aplatir(valeur, cle_complete)
            elif isinstance(o, list):
                # Indexation des éléments de liste avec leur position
                for index, valeur in enumerate(o):
                    cle_complete = f"{prefixe}{sep}{index}" if prefixe else str(index)
                    _aplatir(valeur, cle_complete)
            else:
                # Valeur scalaire : on l'enregistre dans le résultat
                resultat[prefixe] = o

        # Lancement de la récursion à la racine
        _aplatir(obj)
        return resultat

    def _load_raw(self) -> Any:
        """Charge les données JSON brutes depuis le fichier ou le dict source.

        Returns:
            Any: Les données JSON brutes (dict, list, etc.).

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
            ReadError: Si le contenu JSON est invalide.
        """
        if self._raw is not None:
            # Source est un dictionnaire Python : retour direct
            return self._raw

        # Vérification de l'existence du fichier
        if not self.ping():
            raise ConnectError(
                f"Fichier JSON introuvable : '{self._file}'"
            )

        try:
            # Lecture et parse du fichier JSON
            with open(self._file, encoding="utf-8", errors="replace") as f:
                donnees = json.load(f)
            return donnees

        except json.JSONDecodeError as erreur:
            raise ReadError(
                f"Contenu JSON invalide dans '{self._file}' : {erreur}"
            ) from erreur
        except OSError as erreur:
            raise ConnectError(
                f"Impossible de lire '{self._file}' : {erreur}"
            ) from erreur

    def read(self, flatten: bool = True) -> pd.DataFrame:
        """Lit les données JSON et les retourne sous forme de DataFrame.

        Si flatten=True, les structures imbriquées sont automatiquement
        aplaties avant la conversion en DataFrame.

        Args:
            flatten (bool): Si True, applique self.flatten() sur chaque
                enregistrement avant normalisation. Par défaut True.

        Returns:
            pd.DataFrame: Les données JSON sous forme tabulaire.

        Raises:
            ConnectError: Si la source n'est pas accessible.
            ReadError: Si la conversion en DataFrame échoue.
        """
        # Chargement des données JSON brutes
        donnees_brutes = self._load_raw()

        # Normalisation en liste d'enregistrements si nécessaire
        if isinstance(donnees_brutes, dict):
            # Un seul enregistrement dict -> liste de un
            liste_enregistrements = [donnees_brutes]
        elif isinstance(donnees_brutes, list):
            # Déjà une liste d'enregistrements
            liste_enregistrements = donnees_brutes
        else:
            raise ReadError(
                f"Format JSON non supporté : attendu dict ou list, "
                f"reçu {type(donnees_brutes).__name__}."
            )

        try:
            if flatten:
                # Aplatissement de chaque enregistrement avant normalisation
                enregistrements_aplatis = [
                    self.flatten(enr) if isinstance(enr, dict) else enr
                    for enr in liste_enregistrements
                ]
                df = pd.DataFrame(enregistrements_aplatis)
            else:
                # Normalisation JSON directe sans aplatissement
                df = pd.json_normalize(liste_enregistrements)

            logger.info(
                "Source JSON '%s' lue : %d lignes, %d colonnes (flatten=%s).",
                self.path,
                len(df),
                len(df.columns),
                flatten,
            )

            # Mise à jour de l'horodatage de lecture
            self._touch()
            return df

        except Exception as erreur:
            raise ReadError(
                f"Impossible de convertir le JSON en DataFrame : {erreur}"
            ) from erreur

    def write(
        self,
        data: pd.DataFrame,
        orient: str = "records",
    ) -> bool:
        """Écrit un DataFrame vers le fichier JSON de la source.

        Args:
            data (pd.DataFrame): Les données à écrire au format JSON.
            orient (str): Orientation de la sérialisation JSON. Valeurs
                possibles : 'records', 'index', 'columns', 'values'.
                Par défaut 'records'.

        Returns:
            bool: True si l'écriture s'est déroulée avec succès.

        Raises:
            WriteError: Si la source est un dict en mémoire (pas de fichier)
                ou si l'écriture échoue.
        """
        # Vérification qu'on a bien un fichier de destination
        if self._file is None:
            raise WriteError(
                "Impossible d'écrire : la source JSON est un dictionnaire "
                "en mémoire, aucun fichier de destination n'est défini."
            )

        try:
            # Sérialisation du DataFrame en JSON
            data.to_json(
                self._file,
                orient=orient,
                force_ascii=False,
                indent=2,
            )
            logger.info(
                "DataFrame écrit avec succès vers '%s' (%d lignes, orient='%s').",
                self._file,
                len(data),
                orient,
            )
            return True

        except Exception as erreur:
            raise WriteError(
                f"Impossible d'écrire vers '{self._file}' : {erreur}"
            ) from erreur

    def info(self) -> dict:
        """Retourne les métadonnées descriptives de la source JSON.

        Returns:
            dict: Dictionnaire contenant les clés suivantes :
                - 'path' (str) : chemin du fichier ou '<dict_en_memoire>'.
                - 'kind' (str) : 'json'.
                - 'is_file' (bool) : True si la source est un fichier.
                - 'size_kb' (float) : taille du fichier (0 si dict en mémoire).
                - 'last_read' (str | None) : horodatage de la dernière lecture.
        """
        # Calcul de la taille du fichier si applicable
        taille_kb = 0.0
        if self._file and os.path.isfile(self._file):
            taille_kb = os.path.getsize(self._file) / 1024

        return {
            "path": self.path,
            "kind": "json",
            "is_file": self._file is not None,
            "size_kb": round(taille_kb, 2),
            "last_read": (
                self.last_read.isoformat() if self.last_read else None
            ),
        }

    def ping(self) -> bool:
        """Vérifie que la source JSON est accessible.

        Pour une source fichier, vérifie l'existence du fichier.
        Pour une source dict, retourne toujours True.

        Returns:
            bool: True si la source est accessible.
        """
        # Un dictionnaire en mémoire est toujours accessible
        if self._raw is not None:
            return True

        # Vérification de l'existence et de la lisibilité du fichier
        est_accessible = self._file is not None and os.path.isfile(
            self._file
        ) and os.access(self._file, os.R_OK)

        if not est_accessible:
            logger.warning(
                "Le fichier JSON '%s' n'existe pas ou n'est pas lisible.",
                self._file,
            )

        return est_accessible

__init__ 

__init__(file_path_or_dict: Union[str, dict]) -> None

Initialise la source JSON depuis un fichier ou un dictionnaire.

Parameters:

Name Type Description Default
file_path_or_dict str | dict

Chemin vers le fichier JSON ou un dictionnaire Python à utiliser directement comme source.

 required
Source code in kadi/kidas/sources/json_source.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    file_path_or_dict: Union[str, dict],
) -> None:
    """Initialise la source JSON depuis un fichier ou un dictionnaire.

    Args:
        file_path_or_dict (str | dict): Chemin vers le fichier JSON
            ou un dictionnaire Python à utiliser directement comme source.
    """
    # Détection du type de source : fichier ou dictionnaire en mémoire
    if isinstance(file_path_or_dict, dict):
        # Source : dictionnaire Python en mémoire
        self._raw: Optional[dict] = file_path_or_dict
        chemin = "<dict_en_memoire>"
    else:
        # Source : fichier JSON sur le disque
        self._raw = None
        chemin = file_path_or_dict

    # Initialisation de la classe parente avec le type 'json'
    super().__init__(
        path=chemin,
        kind="json",
        encoding="utf-8",
    )

    # Chemin du fichier (None si source est un dict)
    self._file: Optional[str] = (
        None if isinstance(file_path_or_dict, dict) else file_path_or_dict
    )

flatten 

flatten(obj: Dict[str, Any], sep: str = '.') -> dict

Aplatit récursivement un objet JSON imbriqué.

Transforme les clés imbriquées en clés à notation pointée. Exemple : {'location': {'lat': 9.3}} -> {'location.lat': 9.3}.

Parameters:

Name Type Description Default
obj dict

L'objet JSON à aplatir (peut contenir des dicts et des listes imbriqués).

 required
sep str

Caractère séparateur entre les niveaux de clés. Par défaut '.'.

 '.'

Returns:

Name Type Description
dict dict

Dictionnaire aplati avec des clés à notation pointée.
Source code in kadi/kidas/sources/json_source.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def flatten(self, obj: Dict[str, Any], sep: str = ".") -> dict:
    """Aplatit récursivement un objet JSON imbriqué.

    Transforme les clés imbriquées en clés à notation pointée.
    Exemple : {'location': {'lat': 9.3}} -> {'location.lat': 9.3}.

    Args:
        obj (dict): L'objet JSON à aplatir (peut contenir des dicts
            et des listes imbriqués).
        sep (str): Caractère séparateur entre les niveaux de clés.
            Par défaut '.'.

    Returns:
        dict: Dictionnaire aplati avec des clés à notation pointée.
    """
    # Dictionnaire de sortie aplati
    resultat: dict = {}

    def _aplatir(o: Any, prefixe: str = "") -> None:
        """Fonction récursive interne réalisant l'aplatissement.

        Args:
            o (Any): L'objet courant à aplatir.
            prefixe (str): Le préfixe des clés parentes accumulées.
        """
        if isinstance(o, dict):
            # Récursion sur chaque clé du dictionnaire
            for cle, valeur in o.items():
                # Construction de la clé complète avec préfixe
                cle_complete = f"{prefixe}{sep}{cle}" if prefixe else cle
                _aplatir(valeur, cle_complete)
        elif isinstance(o, list):
            # Indexation des éléments de liste avec leur position
            for index, valeur in enumerate(o):
                cle_complete = f"{prefixe}{sep}{index}" if prefixe else str(index)
                _aplatir(valeur, cle_complete)
        else:
            # Valeur scalaire : on l'enregistre dans le résultat
            resultat[prefixe] = o

    # Lancement de la récursion à la racine
    _aplatir(obj)
    return resultat

read 

read(flatten: bool = True) -> pd.DataFrame

Lit les données JSON et les retourne sous forme de DataFrame.

Si flatten=True, les structures imbriquées sont automatiquement aplaties avant la conversion en DataFrame.

Parameters:

Name Type Description Default
flatten bool

Si True, applique self.flatten() sur chaque enregistrement avant normalisation. Par défaut True.

 True

Returns:

Type Description
DataFrame

pd.DataFrame: Les données JSON sous forme tabulaire.

Raises:

Type Description
ConnectError

Si la source n'est pas accessible.
ReadError

Si la conversion en DataFrame échoue.
Source code in kadi/kidas/sources/json_source.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def read(self, flatten: bool = True) -> pd.DataFrame:
    """Lit les données JSON et les retourne sous forme de DataFrame.

    Si flatten=True, les structures imbriquées sont automatiquement
    aplaties avant la conversion en DataFrame.

    Args:
        flatten (bool): Si True, applique self.flatten() sur chaque
            enregistrement avant normalisation. Par défaut True.

    Returns:
        pd.DataFrame: Les données JSON sous forme tabulaire.

    Raises:
        ConnectError: Si la source n'est pas accessible.
        ReadError: Si la conversion en DataFrame échoue.
    """
    # Chargement des données JSON brutes
    donnees_brutes = self._load_raw()

    # Normalisation en liste d'enregistrements si nécessaire
    if isinstance(donnees_brutes, dict):
        # Un seul enregistrement dict -> liste de un
        liste_enregistrements = [donnees_brutes]
    elif isinstance(donnees_brutes, list):
        # Déjà une liste d'enregistrements
        liste_enregistrements = donnees_brutes
    else:
        raise ReadError(
            f"Format JSON non supporté : attendu dict ou list, "
            f"reçu {type(donnees_brutes).__name__}."
        )

    try:
        if flatten:
            # Aplatissement de chaque enregistrement avant normalisation
            enregistrements_aplatis = [
                self.flatten(enr) if isinstance(enr, dict) else enr
                for enr in liste_enregistrements
            ]
            df = pd.DataFrame(enregistrements_aplatis)
        else:
            # Normalisation JSON directe sans aplatissement
            df = pd.json_normalize(liste_enregistrements)

        logger.info(
            "Source JSON '%s' lue : %d lignes, %d colonnes (flatten=%s).",
            self.path,
            len(df),
            len(df.columns),
            flatten,
        )

        # Mise à jour de l'horodatage de lecture
        self._touch()
        return df

    except Exception as erreur:
        raise ReadError(
            f"Impossible de convertir le JSON en DataFrame : {erreur}"
        ) from erreur

write 

write(data: DataFrame, orient: str = 'records') -> bool

Écrit un DataFrame vers le fichier JSON de la source.

Parameters:

Name Type Description Default
data DataFrame

Les données à écrire au format JSON.

 required
orient str

Orientation de la sérialisation JSON. Valeurs possibles : 'records', 'index', 'columns', 'values'. Par défaut 'records'.

 'records'

Returns:

Name Type Description
bool bool

True si l'écriture s'est déroulée avec succès.

Raises:

Type Description
WriteError

Si la source est un dict en mémoire (pas de fichier) ou si l'écriture échoue.
Source code in kadi/kidas/sources/json_source.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def write(
    self,
    data: pd.DataFrame,
    orient: str = "records",
) -> bool:
    """Écrit un DataFrame vers le fichier JSON de la source.

    Args:
        data (pd.DataFrame): Les données à écrire au format JSON.
        orient (str): Orientation de la sérialisation JSON. Valeurs
            possibles : 'records', 'index', 'columns', 'values'.
            Par défaut 'records'.

    Returns:
        bool: True si l'écriture s'est déroulée avec succès.

    Raises:
        WriteError: Si la source est un dict en mémoire (pas de fichier)
            ou si l'écriture échoue.
    """
    # Vérification qu'on a bien un fichier de destination
    if self._file is None:
        raise WriteError(
            "Impossible d'écrire : la source JSON est un dictionnaire "
            "en mémoire, aucun fichier de destination n'est défini."
        )

    try:
        # Sérialisation du DataFrame en JSON
        data.to_json(
            self._file,
            orient=orient,
            force_ascii=False,
            indent=2,
        )
        logger.info(
            "DataFrame écrit avec succès vers '%s' (%d lignes, orient='%s').",
            self._file,
            len(data),
            orient,
        )
        return True

    except Exception as erreur:
        raise WriteError(
            f"Impossible d'écrire vers '{self._file}' : {erreur}"
        ) from erreur

info 

info() -> dict

Retourne les métadonnées descriptives de la source JSON.

Returns:

Name Type Description
dict dict

Dictionnaire contenant les clés suivantes : - 'path' (str) : chemin du fichier ou ''. - 'kind' (str) : 'json'. - 'is_file' (bool) : True si la source est un fichier. - 'size_kb' (float) : taille du fichier (0 si dict en mémoire). - 'last_read' (str | None) : horodatage de la dernière lecture.
Source code in kadi/kidas/sources/json_source.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
def info(self) -> dict:
    """Retourne les métadonnées descriptives de la source JSON.

    Returns:
        dict: Dictionnaire contenant les clés suivantes :
            - 'path' (str) : chemin du fichier ou '<dict_en_memoire>'.
            - 'kind' (str) : 'json'.
            - 'is_file' (bool) : True si la source est un fichier.
            - 'size_kb' (float) : taille du fichier (0 si dict en mémoire).
            - 'last_read' (str | None) : horodatage de la dernière lecture.
    """
    # Calcul de la taille du fichier si applicable
    taille_kb = 0.0
    if self._file and os.path.isfile(self._file):
        taille_kb = os.path.getsize(self._file) / 1024

    return {
        "path": self.path,
        "kind": "json",
        "is_file": self._file is not None,
        "size_kb": round(taille_kb, 2),
        "last_read": (
            self.last_read.isoformat() if self.last_read else None
        ),
    }

ping 

ping() -> bool

Vérifie que la source JSON est accessible.

Pour une source fichier, vérifie l'existence du fichier. Pour une source dict, retourne toujours True.

Returns:

Name Type Description
bool bool

True si la source est accessible.
Source code in kadi/kidas/sources/json_source.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def ping(self) -> bool:
    """Vérifie que la source JSON est accessible.

    Pour une source fichier, vérifie l'existence du fichier.
    Pour une source dict, retourne toujours True.

    Returns:
        bool: True si la source est accessible.
    """
    # Un dictionnaire en mémoire est toujours accessible
    if self._raw is not None:
        return True

    # Vérification de l'existence et de la lisibilité du fichier
    est_accessible = self._file is not None and os.path.isfile(
        self._file
    ) and os.access(self._file, os.R_OK)

    if not est_accessible:
        logger.warning(
            "Le fichier JSON '%s' n'existe pas ou n'est pas lisible.",
            self._file,
        )

    return est_accessible

NetCDFSource

Bases: Source

Source de données pour les fichiers NetCDF agrométéorologiques.

Gère la lecture de fichiers NetCDF avec extraction de sous-ensembles spatiaux et temporels, en utilisant xarray comme moteur de lecture. Supporte les grands fichiers via le chunking Dask.

Attributs

path (str): Chemin absolu ou relatif vers le fichier NetCDF. use_dask (bool): Si True, utilise Dask pour le chargement paresseux (lazy loading) des grands fichiers. _ds (xr.Dataset | None): Dataset xarray chargé en cache. _da (xr.DataArray | None): DataArray extrait lors du dernier read().

Exemple

source = NetCDFSource('chirps_benin_2024.nc') da = source.read(lat_bounds=(2.5, 12.5), lon_bounds=(-1.5, 4.0)) df = source.to_df() print(source.dims())

Source code in kadi/kidas/sources/netcdf_source.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
class NetCDFSource(Source):
    """Source de données pour les fichiers NetCDF agrométéorologiques.

    Gère la lecture de fichiers NetCDF avec extraction de sous-ensembles
    spatiaux et temporels, en utilisant xarray comme moteur de lecture.
    Supporte les grands fichiers via le chunking Dask.

    Attributs:
        path (str): Chemin absolu ou relatif vers le fichier NetCDF.
        use_dask (bool): Si True, utilise Dask pour le chargement paresseux
            (lazy loading) des grands fichiers.
        _ds (xr.Dataset | None): Dataset xarray chargé en cache.
        _da (xr.DataArray | None): DataArray extrait lors du dernier read().

    Exemple:
        >>> source = NetCDFSource('chirps_benin_2024.nc')
        >>> da = source.read(lat_bounds=(2.5, 12.5), lon_bounds=(-1.5, 4.0))
        >>> df = source.to_df()
        >>> print(source.dims())
        {'lat': 240, 'lon': 360, 'time': 365}
    """

    def __init__(
        self,
        path: str,
        use_dask: bool = False,
    ) -> None:
        """Initialise la source NetCDF avec support optionnel de Dask.

        Args:
            path (str): Chemin vers le fichier NetCDF (.nc) à lire.
            use_dask (bool): Si True, utilise le chunking automatique de Dask
                pour les fichiers volumineux (> 500 Mo). Par défaut False.
        """
        # Initialisation de la classe parente avec le type 'netcdf'
        super().__init__(
            path=path,
            kind="netcdf",
            encoding="utf-8",
        )

        # Activation du chargement paresseux via Dask
        self.use_dask: bool = use_dask

        # Cache interne du dataset xarray (chargé à la première lecture)
        self._ds: Optional[xr.Dataset] = None

        # Cache du DataArray extrait lors du dernier appel à read()
        self._da: Optional[xr.DataArray] = None

    def _open(self) -> xr.Dataset:
        """Charge le dataset xarray depuis le fichier NetCDF.

        Utilise le cache interne pour éviter de recharger le fichier
        à chaque appel. En mode Dask, le chargement est paresseux.

        Returns:
            xr.Dataset: Le dataset xarray chargé.

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
            ReadError: Si le fichier NetCDF est corrompu.
        """
        # Retour du cache si déjà chargé
        if self._ds is not None:
            return self._ds

        if not self.ping():
            raise ConnectError(
                f"Fichier NetCDF introuvable : '{self.path}'"
            )

        try:
            # Arguments de chargement selon l'activation de Dask
            kwargs_chargement = {"chunks": "auto"} if self.use_dask else {}

            # Chargement du dataset xarray
            self._ds = xr.open_dataset(
                self.path, **kwargs_chargement
            )
            logger.debug(
                "Dataset NetCDF '%s' chargé (Dask=%s).",
                self.path,
                self.use_dask,
            )
            return self._ds

        except Exception as erreur:
            raise ReadError(
                f"Impossible de lire le fichier NetCDF '{self.path}' : {erreur}"
            ) from erreur

    def dims(self) -> Dict[str, int]:
        """Retourne les dimensions du dataset NetCDF.

        Returns:
            dict: Dictionnaire nom -> taille pour chaque dimension.
                Exemple : {'lat': 240, 'lon': 360, 'time': 1461}.

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
            ReadError: Si la lecture du dataset échoue.
        """
        # Chargement du dataset (depuis le cache ou le disque)
        ds = self._open()

        # Extraction des dimensions du dataset via ds.sizes (compatible xarray futur)
        dimensions = {dim: int(taille) for dim, taille in ds.sizes.items()}
        logger.debug(
            "Dimensions du fichier '%s' : %s.", self.path, dimensions
        )
        return dimensions

    def read(
        self,
        lat_bounds: Optional[Tuple[float, float]] = None,
        lon_bounds: Optional[Tuple[float, float]] = None,
        time_bounds: Optional[Tuple[str, str]] = None,
    ) -> xr.DataArray:
        """Extrait un sous-ensemble spatial et temporel du fichier NetCDF.

        Par défaut, extrait les données couvrant le Bénin si aucune
        bounding box n'est spécifiée.

        Args:
            lat_bounds (tuple[float, float] | None): Intervalle de latitude
                (lat_min, lat_max). Par défaut la bbox du Bénin (2.5, 12.5).
            lon_bounds (tuple[float, float] | None): Intervalle de longitude
                (lon_min, lon_max). Par défaut la bbox du Bénin (-1.5, 4.0).
            time_bounds (tuple[str, str] | None): Intervalle temporel sous
                forme de chaînes ISO (ex: ('2024-01-01', '2024-12-31')).
                None pour toute la période disponible.

        Returns:
            xr.DataArray: Le sous-ensemble de données extrait.

        Raises:
            ConnectError: Si le fichier n'est pas accessible.
            ReadError: Si l'extraction du sous-ensemble échoue.
        """
        # Chargement du dataset
        ds = self._open()

        # Application des bounding boxes par défaut (Bénin) si non spécifiées
        lat_min, lat_max = lat_bounds if lat_bounds else _BENIN_LAT_BOUNDS
        lon_min, lon_max = lon_bounds if lon_bounds else _BENIN_LON_BOUNDS

        try:
            # Détection des noms de variables lat/lon dans le dataset
            # (peuvent s'appeler 'lat', 'latitude', 'y', etc.)
            noms_lat = [d for d in ds.sizes if "lat" in d.lower()]
            noms_lon = [d for d in ds.sizes if "lon" in d.lower() or d.lower() == "x"]

            nom_lat = noms_lat[0] if noms_lat else "lat"
            nom_lon = noms_lon[0] if noms_lon else "lon"

            # Sélection de la première variable de données si plusieurs existent
            variable_principale = list(ds.data_vars)[0]
            da = ds[variable_principale]

            # Extraction du sous-ensemble spatial
            da = da.sel(
                {
                    nom_lat: slice(lat_min, lat_max),
                    nom_lon: slice(lon_min, lon_max),
                }
            )

            # Extraction du sous-ensemble temporel si spécifié
            if time_bounds is not None:
                noms_time = [d for d in ds.sizes if "time" in d.lower()]
                nom_time = noms_time[0] if noms_time else "time"
                da = da.sel(
                    {nom_time: slice(time_bounds[0], time_bounds[1])}
                )

            logger.info(
                "Extraction NetCDF '%s' : lat[%.1f, %.1f], lon[%.1f, %.1f].",
                self.path,
                lat_min,
                lat_max,
                lon_min,
                lon_max,
            )

            # Mise en cache du DataArray et mise à jour de l'horodatage
            self._da = da
            self._touch()
            return da

        except Exception as erreur:
            raise ReadError(
                f"Erreur lors de l'extraction du sous-ensemble NetCDF : {erreur}"
            ) from erreur

    def to_df(self) -> pd.DataFrame:
        """Convertit le dernier DataArray extrait en pandas DataFrame.

        Utilise le DataArray issu du dernier appel à read(). Si read() n'a
        pas encore été appelé, lit l'ensemble du dataset avec la bbox Bénin.

        Returns:
            pd.DataFrame: Les données NetCDF sous forme tabulaire avec
                les dimensions comme colonnes (lat, lon, time, valeur).

        Raises:
            ReadError: Si la conversion échoue.
        """
        # Si read() n'a pas été appelé, effectuer une lecture par défaut
        if self._da is None:
            self.read()

        try:
            # Conversion du DataArray en DataFrame pandas
            df = self._da.to_dataframe().reset_index()
            logger.debug(
                "DataArray NetCDF converti en DataFrame : %d lignes, %d colonnes.",
                len(df),
                len(df.columns),
            )
            return df

        except Exception as erreur:
            raise ReadError(
                f"Impossible de convertir le DataArray en DataFrame : {erreur}"
            ) from erreur

    def write(self, data: pd.DataFrame) -> bool:
        """Écrit les données vers le fichier NetCDF de la source.

        Note: L'écriture de fichiers NetCDF complexes nécessite une structure
        précise (dimensions, coordonnées, attributs). Cette implémentation
        effectue une conversion basique du DataFrame en Dataset xarray.

        Args:
            data (pd.DataFrame): Les données à sauvegarder au format NetCDF.

        Returns:
            bool: True si l'écriture s'est déroulée avec succès.

        Raises:
            WriteError: Si l'écriture échoue.
        """
        try:
            # Conversion du DataFrame en Dataset xarray puis sauvegarde
            ds_sortie = xr.Dataset.from_dataframe(data)
            ds_sortie.to_netcdf(self.path)
            logger.info(
                "Données écrites vers le fichier NetCDF '%s'.", self.path
            )
            return True

        except Exception as erreur:
            raise WriteError(
                f"Impossible d'écrire vers '{self.path}' : {erreur}"
            ) from erreur

    def info(self) -> dict:
        """Retourne les métadonnées descriptives du fichier NetCDF.

        Returns:
            dict: Dictionnaire contenant les clés suivantes :
                - 'path' (str) : chemin du fichier.
                - 'kind' (str) : 'netcdf'.
                - 'dimensions' (dict) : nom -> taille de chaque dimension.
                - 'variables' (list) : liste des variables de données.
                - 'use_dask' (bool) : mode de chargement.
                - 'size_kb' (float) : taille du fichier en kilo-octets.
                - 'last_read' (str | None) : horodatage de la dernière lecture.
        """
        # Tentative de chargement des informations structurelles
        try:
            ds = self._open()
            dimensions = self.dims()
            variables = list(ds.data_vars)
        except (ReadError, ConnectError):
            dimensions = {}
            variables = []

        # Calcul de la taille du fichier
        taille_kb = os.path.getsize(self.path) / 1024 if os.path.isfile(
            self.path
        ) else 0.0

        return {
            "path": self.path,
            "kind": "netcdf",
            "dimensions": dimensions,
            "variables": variables,
            "use_dask": self.use_dask,
            "size_kb": round(taille_kb, 2),
            "last_read": (
                self.last_read.isoformat() if self.last_read else None
            ),
        }

    def ping(self) -> bool:
        """Vérifie que le fichier NetCDF existe et est lisible.

        Returns:
            bool: True si le fichier est accessible en lecture, False sinon.
        """
        # Vérification de l'existence et de la lisibilité du fichier
        est_accessible = os.path.isfile(self.path) and os.access(
            self.path, os.R_OK
        )

        if not est_accessible:
            logger.warning(
                "Le fichier NetCDF '%s' n'existe pas ou n'est pas lisible.",
                self.path,
            )

        return est_accessible

__init__ 

__init__(path: str, use_dask: bool = False) -> None

Initialise la source NetCDF avec support optionnel de Dask.

Parameters:

Name Type Description Default
path str

Chemin vers le fichier NetCDF (.nc) à lire.

 required
use_dask bool

Si True, utilise le chunking automatique de Dask pour les fichiers volumineux (> 500 Mo). Par défaut False.

 False
Source code in kadi/kidas/sources/netcdf_source.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    path: str,
    use_dask: bool = False,
) -> None:
    """Initialise la source NetCDF avec support optionnel de Dask.

    Args:
        path (str): Chemin vers le fichier NetCDF (.nc) à lire.
        use_dask (bool): Si True, utilise le chunking automatique de Dask
            pour les fichiers volumineux (> 500 Mo). Par défaut False.
    """
    # Initialisation de la classe parente avec le type 'netcdf'
    super().__init__(
        path=path,
        kind="netcdf",
        encoding="utf-8",
    )

    # Activation du chargement paresseux via Dask
    self.use_dask: bool = use_dask

    # Cache interne du dataset xarray (chargé à la première lecture)
    self._ds: Optional[xr.Dataset] = None

    # Cache du DataArray extrait lors du dernier appel à read()
    self._da: Optional[xr.DataArray] = None

dims 

dims() -> Dict[str, int]

Retourne les dimensions du dataset NetCDF.

Returns:

Name Type Description
dict Dict[str, int]

Dictionnaire nom -> taille pour chaque dimension. Exemple : {'lat': 240, 'lon': 360, 'time': 1461}.

Raises:

Type Description
ConnectError

Si le fichier n'est pas accessible.
ReadError

Si la lecture du dataset échoue.
Source code in kadi/kidas/sources/netcdf_source.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def dims(self) -> Dict[str, int]:
    """Retourne les dimensions du dataset NetCDF.

    Returns:
        dict: Dictionnaire nom -> taille pour chaque dimension.
            Exemple : {'lat': 240, 'lon': 360, 'time': 1461}.

    Raises:
        ConnectError: Si le fichier n'est pas accessible.
        ReadError: Si la lecture du dataset échoue.
    """
    # Chargement du dataset (depuis le cache ou le disque)
    ds = self._open()

    # Extraction des dimensions du dataset via ds.sizes (compatible xarray futur)
    dimensions = {dim: int(taille) for dim, taille in ds.sizes.items()}
    logger.debug(
        "Dimensions du fichier '%s' : %s.", self.path, dimensions
    )
    return dimensions

read 

read(lat_bounds: Optional[Tuple[float, float]] = None, lon_bounds: Optional[Tuple[float, float]] = None, time_bounds: Optional[Tuple[str, str]] = None) -> xr.DataArray

Extrait un sous-ensemble spatial et temporel du fichier NetCDF.

Par défaut, extrait les données couvrant le Bénin si aucune bounding box n'est spécifiée.

Parameters:

Name Type Description Default
lat_bounds tuple[float, float] | None

Intervalle de latitude (lat_min, lat_max). Par défaut la bbox du Bénin (2.5, 12.5).

 None
lon_bounds tuple[float, float] | None

Intervalle de longitude (lon_min, lon_max). Par défaut la bbox du Bénin (-1.5, 4.0).

 None
time_bounds tuple[str, str] | None

Intervalle temporel sous forme de chaînes ISO (ex: ('2024-01-01', '2024-12-31')). None pour toute la période disponible.

 None

Returns:

Type Description
DataArray

xr.DataArray: Le sous-ensemble de données extrait.

Raises:

Type Description
ConnectError

Si le fichier n'est pas accessible.
ReadError

Si l'extraction du sous-ensemble échoue.
Source code in kadi/kidas/sources/netcdf_source.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def read(
    self,
    lat_bounds: Optional[Tuple[float, float]] = None,
    lon_bounds: Optional[Tuple[float, float]] = None,
    time_bounds: Optional[Tuple[str, str]] = None,
) -> xr.DataArray:
    """Extrait un sous-ensemble spatial et temporel du fichier NetCDF.

    Par défaut, extrait les données couvrant le Bénin si aucune
    bounding box n'est spécifiée.

    Args:
        lat_bounds (tuple[float, float] | None): Intervalle de latitude
            (lat_min, lat_max). Par défaut la bbox du Bénin (2.5, 12.5).
        lon_bounds (tuple[float, float] | None): Intervalle de longitude
            (lon_min, lon_max). Par défaut la bbox du Bénin (-1.5, 4.0).
        time_bounds (tuple[str, str] | None): Intervalle temporel sous
            forme de chaînes ISO (ex: ('2024-01-01', '2024-12-31')).
            None pour toute la période disponible.

    Returns:
        xr.DataArray: Le sous-ensemble de données extrait.

    Raises:
        ConnectError: Si le fichier n'est pas accessible.
        ReadError: Si l'extraction du sous-ensemble échoue.
    """
    # Chargement du dataset
    ds = self._open()

    # Application des bounding boxes par défaut (Bénin) si non spécifiées
    lat_min, lat_max = lat_bounds if lat_bounds else _BENIN_LAT_BOUNDS
    lon_min, lon_max = lon_bounds if lon_bounds else _BENIN_LON_BOUNDS

    try:
        # Détection des noms de variables lat/lon dans le dataset
        # (peuvent s'appeler 'lat', 'latitude', 'y', etc.)
        noms_lat = [d for d in ds.sizes if "lat" in d.lower()]
        noms_lon = [d for d in ds.sizes if "lon" in d.lower() or d.lower() == "x"]

        nom_lat = noms_lat[0] if noms_lat else "lat"
        nom_lon = noms_lon[0] if noms_lon else "lon"

        # Sélection de la première variable de données si plusieurs existent
        variable_principale = list(ds.data_vars)[0]
        da = ds[variable_principale]

        # Extraction du sous-ensemble spatial
        da = da.sel(
            {
                nom_lat: slice(lat_min, lat_max),
                nom_lon: slice(lon_min, lon_max),
            }
        )

        # Extraction du sous-ensemble temporel si spécifié
        if time_bounds is not None:
            noms_time = [d for d in ds.sizes if "time" in d.lower()]
            nom_time = noms_time[0] if noms_time else "time"
            da = da.sel(
                {nom_time: slice(time_bounds[0], time_bounds[1])}
            )

        logger.info(
            "Extraction NetCDF '%s' : lat[%.1f, %.1f], lon[%.1f, %.1f].",
            self.path,
            lat_min,
            lat_max,
            lon_min,
            lon_max,
        )

        # Mise en cache du DataArray et mise à jour de l'horodatage
        self._da = da
        self._touch()
        return da

    except Exception as erreur:
        raise ReadError(
            f"Erreur lors de l'extraction du sous-ensemble NetCDF : {erreur}"
        ) from erreur

to_df 

to_df() -> pd.DataFrame

Convertit le dernier DataArray extrait en pandas DataFrame.

Utilise le DataArray issu du dernier appel à read(). Si read() n'a pas encore été appelé, lit l'ensemble du dataset avec la bbox Bénin.

Returns:

Type Description
DataFrame

pd.DataFrame: Les données NetCDF sous forme tabulaire avec les dimensions comme colonnes (lat, lon, time, valeur).

Raises:

Type Description
ReadError

Si la conversion échoue.
Source code in kadi/kidas/sources/netcdf_source.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def to_df(self) -> pd.DataFrame:
    """Convertit le dernier DataArray extrait en pandas DataFrame.

    Utilise le DataArray issu du dernier appel à read(). Si read() n'a
    pas encore été appelé, lit l'ensemble du dataset avec la bbox Bénin.

    Returns:
        pd.DataFrame: Les données NetCDF sous forme tabulaire avec
            les dimensions comme colonnes (lat, lon, time, valeur).

    Raises:
        ReadError: Si la conversion échoue.
    """
    # Si read() n'a pas été appelé, effectuer une lecture par défaut
    if self._da is None:
        self.read()

    try:
        # Conversion du DataArray en DataFrame pandas
        df = self._da.to_dataframe().reset_index()
        logger.debug(
            "DataArray NetCDF converti en DataFrame : %d lignes, %d colonnes.",
            len(df),
            len(df.columns),
        )
        return df

    except Exception as erreur:
        raise ReadError(
            f"Impossible de convertir le DataArray en DataFrame : {erreur}"
        ) from erreur

write 

write(data: DataFrame) -> bool

Écrit les données vers le fichier NetCDF de la source.

Note: L'écriture de fichiers NetCDF complexes nécessite une structure précise (dimensions, coordonnées, attributs). Cette implémentation effectue une conversion basique du DataFrame en Dataset xarray.

Parameters:

Name Type Description Default
data DataFrame

Les données à sauvegarder au format NetCDF.

 required

Returns:

Name Type Description
bool bool

True si l'écriture s'est déroulée avec succès.

Raises:

Type Description
WriteError

Si l'écriture échoue.
Source code in kadi/kidas/sources/netcdf_source.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def write(self, data: pd.DataFrame) -> bool:
    """Écrit les données vers le fichier NetCDF de la source.

    Note: L'écriture de fichiers NetCDF complexes nécessite une structure
    précise (dimensions, coordonnées, attributs). Cette implémentation
    effectue une conversion basique du DataFrame en Dataset xarray.

    Args:
        data (pd.DataFrame): Les données à sauvegarder au format NetCDF.

    Returns:
        bool: True si l'écriture s'est déroulée avec succès.

    Raises:
        WriteError: Si l'écriture échoue.
    """
    try:
        # Conversion du DataFrame en Dataset xarray puis sauvegarde
        ds_sortie = xr.Dataset.from_dataframe(data)
        ds_sortie.to_netcdf(self.path)
        logger.info(
            "Données écrites vers le fichier NetCDF '%s'.", self.path
        )
        return True

    except Exception as erreur:
        raise WriteError(
            f"Impossible d'écrire vers '{self.path}' : {erreur}"
        ) from erreur

info 

info() -> dict

Retourne les métadonnées descriptives du fichier NetCDF.

Returns:

Name Type Description
dict dict

Dictionnaire contenant les clés suivantes : - 'path' (str) : chemin du fichier. - 'kind' (str) : 'netcdf'. - 'dimensions' (dict) : nom -> taille de chaque dimension. - 'variables' (list) : liste des variables de données. - 'use_dask' (bool) : mode de chargement. - 'size_kb' (float) : taille du fichier en kilo-octets. - 'last_read' (str | None) : horodatage de la dernière lecture.
Source code in kadi/kidas/sources/netcdf_source.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def info(self) -> dict:
    """Retourne les métadonnées descriptives du fichier NetCDF.

    Returns:
        dict: Dictionnaire contenant les clés suivantes :
            - 'path' (str) : chemin du fichier.
            - 'kind' (str) : 'netcdf'.
            - 'dimensions' (dict) : nom -> taille de chaque dimension.
            - 'variables' (list) : liste des variables de données.
            - 'use_dask' (bool) : mode de chargement.
            - 'size_kb' (float) : taille du fichier en kilo-octets.
            - 'last_read' (str | None) : horodatage de la dernière lecture.
    """
    # Tentative de chargement des informations structurelles
    try:
        ds = self._open()
        dimensions = self.dims()
        variables = list(ds.data_vars)
    except (ReadError, ConnectError):
        dimensions = {}
        variables = []

    # Calcul de la taille du fichier
    taille_kb = os.path.getsize(self.path) / 1024 if os.path.isfile(
        self.path
    ) else 0.0

    return {
        "path": self.path,
        "kind": "netcdf",
        "dimensions": dimensions,
        "variables": variables,
        "use_dask": self.use_dask,
        "size_kb": round(taille_kb, 2),
        "last_read": (
            self.last_read.isoformat() if self.last_read else None
        ),
    }

ping 

ping() -> bool

Vérifie que le fichier NetCDF existe et est lisible.

Returns:

Name Type Description
bool bool

True si le fichier est accessible en lecture, False sinon.
Source code in kadi/kidas/sources/netcdf_source.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
def ping(self) -> bool:
    """Vérifie que le fichier NetCDF existe et est lisible.

    Returns:
        bool: True si le fichier est accessible en lecture, False sinon.
    """
    # Vérification de l'existence et de la lisibilité du fichier
    est_accessible = os.path.isfile(self.path) and os.access(
        self.path, os.R_OK
    )

    if not est_accessible:
        logger.warning(
            "Le fichier NetCDF '%s' n'existe pas ou n'est pas lisible.",
            self.path,
        )

    return est_accessible

APISource

Bases: Source

Source de données pour les APIs REST agricoles et climatiques.

Gère les appels HTTP GET/POST avec gestion du rate limiting, réessais automatiques avec backoff exponentiel, et parse de la réponse en DataFrame pandas.

Attributs

url (str): URL de base de l'endpoint API. token (str | None): Jeton d'authentification Bearer. None pour les APIs publiques. rate_limit (float): Nombre maximum de requêtes par seconde. _last_req (float): Timestamp de la dernière requête HTTP. _schema (dict | None): Schéma API mis en cache.

Exemple

source = APISource( ... 'https://api.open-meteo.com/v1/forecast', ... rate_limit=10, ... ) df = source.read({ ... 'latitude': 6.36, ... 'longitude': 2.42, ... 'daily': 'temperature_2m_max', ... })

Source code in kadi/kidas/sources/api_source.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
class APISource(Source):
    """Source de données pour les APIs REST agricoles et climatiques.

    Gère les appels HTTP GET/POST avec gestion du rate limiting,
    réessais automatiques avec backoff exponentiel, et parse de la
    réponse en DataFrame pandas.

    Attributs:
        url (str): URL de base de l'endpoint API.
        token (str | None): Jeton d'authentification Bearer.
            None pour les APIs publiques.
        rate_limit (float): Nombre maximum de requêtes par seconde.
        _last_req (float): Timestamp de la dernière requête HTTP.
        _schema (dict | None): Schéma API mis en cache.

    Exemple:
        >>> source = APISource(
        ...     'https://api.open-meteo.com/v1/forecast',
        ...     rate_limit=10,
        ... )
        >>> df = source.read({
        ...     'latitude': 6.36,
        ...     'longitude': 2.42,
        ...     'daily': 'temperature_2m_max',
        ... })
    """

    def __init__(
        self,
        url: str,
        token: Optional[str] = None,
        rate_limit: float = 5.0,
    ) -> None:
        """Initialise la source API avec gestion du rate limiting.

        Args:
            url (str): URL de base de l'endpoint API (sans paramètres).
            token (str | None): Jeton Bearer pour les APIs authentifiées.
                None pour les APIs publiques (Open-Meteo, etc.).
            rate_limit (float): Nombre maximal de requêtes autorisées
                par seconde. Par défaut 5.0.
        """
        # Initialisation de la classe parente avec le type 'api'
        super().__init__(
            path=url,
            kind="api",
            encoding="utf-8",
        )

        # URL de base de l'endpoint API
        self.url: str = url

        # Jeton d'authentification (None pour APIs publiques)
        self.token: Optional[str] = token

        # Limite de débit en requêtes/seconde
        self.rate_limit: float = rate_limit

        # Timestamp de la dernière requête (pour le throttling)
        self._last_req: float = 0.0

        # Cache du schéma API
        self._schema: Optional[dict] = None

    def _build_headers(self) -> Dict[str, str]:
        """Construit les en-têtes HTTP pour la requête API.

        Returns:
            dict: Dictionnaire des en-têtes HTTP avec authentification
                si un jeton est configuré.
        """
        # En-têtes de base communs à toutes les requêtes
        entetes: Dict[str, str] = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            "User-Agent": "KadiPy/1.1 (AgriTech Benin)",
        }

        # Ajout du jeton d'authentification si disponible
        if self.token:
            entetes["Authorization"] = f"Bearer {self.token}"

        return entetes

    def _throttle(self) -> None:
        """Applique le délai nécessaire pour respecter le rate limit.

        Calcule le temps écoulé depuis la dernière requête et attend
        si nécessaire pour ne pas dépasser la limite de débit configurée.
        """
        # Délai minimum entre deux requêtes (en secondes)
        delai_minimum = 1.0 / self.rate_limit

        # Temps écoulé depuis la dernière requête
        temps_ecoule = time.monotonic() - self._last_req

        if temps_ecoule < delai_minimum:
            # Attente du temps restant pour respecter la limite
            temps_attente = delai_minimum - temps_ecoule
            logger.debug(
                "Rate limiting : attente de %.3f secondes.", temps_attente
            )
            time.sleep(temps_attente)

    def fetch(
        self,
        params: Dict[str, Any],
        retries: int = 3,
        backoff: float = 5.0,
    ) -> dict:
        """Effectue une requête GET avec réessais et backoff exponentiel.

        En cas d'erreur temporaire (codes 429, 500, 502, 503, 504),
        réessaie automatiquement avec un délai croissant.

        Args:
            params (dict): Paramètres de la requête GET (query string).
            retries (int): Nombre maximum de tentatives en cas d'échec.
                Par défaut 3.
            backoff (float): Délai de base (en secondes) entre les
                réessais. Le délai est doublé à chaque tentative.
                Par défaut 5.0.

        Returns:
            dict: La réponse JSON parsée de l'API.

        Raises:
            ConnectError: Si l'API est inaccessible après toutes
                les tentatives.
            ReadError: Si la réponse n'est pas un JSON valide.
        """
        derniere_erreur: Optional[Exception] = None

        for tentative in range(retries + 1):
            try:
                # Respect du rate limit avant chaque requête
                self._throttle()

                # Enregistrement du timestamp de la requête
                self._last_req = time.monotonic()

                # Exécution de la requête HTTP GET
                reponse = requests.get(
                    self.url,
                    params=params,
                    headers=self._build_headers(),
                    timeout=30,
                )

                # Journalisation de la requête
                logger.debug(
                    "Requête API [tentative %d/%d] : %s (status: %d).",
                    tentative + 1,
                    retries + 1,
                    reponse.url,
                    reponse.status_code,
                )

                # Vérification si le code est réessayable
                if reponse.status_code in _HTTP_RETRY_CODES and tentative < retries:
                    # Calcul du backoff exponentiel
                    delai_attente = min(backoff * (2 ** tentative), _MAX_BACKOFF_SEC)
                    logger.warning(
                        "Erreur HTTP %d. Réessai dans %.1f secondes (tentative %d/%d).",
                        reponse.status_code,
                        delai_attente,
                        tentative + 1,
                        retries,
                    )
                    time.sleep(delai_attente)
                    continue

                # Levée d'exception pour les codes d'erreur non-réessayables
                reponse.raise_for_status()

                # Parse et retour de la réponse JSON
                return reponse.json()

            except requests.exceptions.ConnectionError as erreur:
                derniere_erreur = erreur
                if tentative < retries:
                    delai_attente = min(backoff * (2 ** tentative), _MAX_BACKOFF_SEC)
                    logger.warning(
                        "Erreur de connexion. Réessai dans %.1f sec.", delai_attente
                    )
                    time.sleep(delai_attente)

            except requests.exceptions.HTTPError as erreur:
                # Erreur HTTP non-réessayable
                raise ConnectError(
                    f"Erreur HTTP lors de l'appel à '{self.url}' : {erreur}"
                ) from erreur

            except ValueError as erreur:
                # Réponse non-JSON
                raise ReadError(
                    f"La réponse de '{self.url}' n'est pas du JSON valide : {erreur}"
                ) from erreur

        # Toutes les tentatives ont échoué
        raise ConnectError(
            f"API '{self.url}' inaccessible après {retries + 1} tentatives. "
            f"Dernière erreur : {derniere_erreur}"
        )

    def read(self, params: Optional[Dict[str, Any]] = None) -> pd.DataFrame:
        """Effectue une requête GET et retourne la réponse sous forme de DataFrame.

        Gère automatiquement les réponses JSON de type dict (avec clé 'data'
        ou 'results') et les réponses de type liste.

        Args:
            params (dict | None): Paramètres de la requête GET.
                Par défaut None (requête sans paramètres).

        Returns:
            pd.DataFrame: Les données de la réponse API sous forme tabulaire.

        Raises:
            ConnectError: Si l'API est inaccessible.
            ReadError: Si la conversion en DataFrame échoue.
        """
        # Paramètres de requête par défaut
        parametres = params or {}

        # Exécution de la requête avec réessais
        donnees_brutes = self.fetch(parametres)

        try:
            # Normalisation de la structure de réponse en liste d'enregistrements
            if isinstance(donnees_brutes, list):
                # Réponse déjà sous forme de liste
                enregistrements = donnees_brutes
            elif isinstance(donnees_brutes, dict):
                # Recherche d'une clé standard contenant les données
                for cle_donnees in ("data", "results", "items", "records", "features"):
                    if cle_donnees in donnees_brutes:
                        enregistrements = donnees_brutes[cle_donnees]
                        break
                else:
                    # Pas de clé standard : utilise le dict entier
                    enregistrements = [donnees_brutes]
            else:
                raise ReadError(
                    f"Format de réponse non supporté : {type(donnees_brutes).__name__}"
                )

            # Conversion en DataFrame pandas
            df = pd.json_normalize(enregistrements)

            logger.info(
                "Données API '%s' reçues : %d lignes, %d colonnes.",
                self.url,
                len(df),
                len(df.columns),
            )

            # Mise à jour de l'horodatage de lecture
            self._touch()
            return df

        except ReadError:
            raise
        except Exception as erreur:
            raise ReadError(
                f"Impossible de convertir la réponse de '{self.url}' "
                f"en DataFrame : {erreur}"
            ) from erreur

    def write(self, data: pd.DataFrame) -> bool:
        """Envoie des données vers l'API via une requête POST.

        Args:
            data (pd.DataFrame): Les données à envoyer à l'API en JSON.

        Returns:
            bool: True si l'envoi a réussi (code 200 ou 201).

        Raises:
            WriteError: Si la requête POST échoue.
        """
        try:
            # Conversion du DataFrame en liste de dictionnaires JSON
            payload = data.to_dict(orient="records")

            # Respect du rate limit avant la requête
            self._throttle()
            self._last_req = time.monotonic()

            # Envoi de la requête POST
            reponse = requests.post(
                self.url,
                json=payload,
                headers=self._build_headers(),
                timeout=30,
            )
            reponse.raise_for_status()

            logger.info(
                "Données envoyées avec succès à '%s' (%d enregistrements, "
                "status: %d).",
                self.url,
                len(data),
                reponse.status_code,
            )
            return True

        except requests.exceptions.HTTPError as erreur:
            raise WriteError(
                f"Erreur HTTP lors du POST vers '{self.url}' : {erreur}"
            ) from erreur
        except Exception as erreur:
            raise WriteError(
                f"Impossible d'envoyer les données vers '{self.url}' : {erreur}"
            ) from erreur

    def info(self) -> dict:
        """Retourne les métadonnées de la connexion API.

        Returns:
            dict: Dictionnaire contenant les clés suivantes :
                - 'url' (str) : URL de l'API.
                - 'kind' (str) : 'api'.
                - 'requires_auth' (bool) : True si un jeton est requis.
                - 'rate_limit' (float) : limite de débit en req/sec.
                - 'last_read' (str | None) : horodatage de la dernière lecture.
        """
        return {
            "url": self.url,
            "kind": "api",
            "requires_auth": self.token is not None,
            "rate_limit": self.rate_limit,
            "last_read": (
                self.last_read.isoformat() if self.last_read else None
            ),
        }

    def ping(self) -> bool:
        """Vérifie que l'endpoint API est accessible via une requête HEAD.

        Returns:
            bool: True si l'endpoint répond (même avec une erreur 4xx),
                False si la connexion échoue complètement.
        """
        try:
            # Requête HEAD légère pour tester la connectivité
            reponse = requests.head(
                self.url,
                headers=self._build_headers(),
                timeout=10,
            )
            # Tout code de réponse inférieur à 500 indique que l'API est accessible
            est_accessible = reponse.status_code < 500
            logger.debug(
                "Validation connexion API '%s' : status %d -> %s.",
                self.url,
                reponse.status_code,
                "OK" if est_accessible else "KO",
            )
            return est_accessible

        except requests.exceptions.RequestException as erreur:
            logger.warning(
                "API '%s' inaccessible : %s", self.url, erreur
            )
            return False

__init__ 

__init__(url: str, token: Optional[str] = None, rate_limit: float = 5.0) -> None

Initialise la source API avec gestion du rate limiting.

Parameters:

Name Type Description Default
url str

URL de base de l'endpoint API (sans paramètres).

 required
token str | None

Jeton Bearer pour les APIs authentifiées. None pour les APIs publiques (Open-Meteo, etc.).

 None
rate_limit float

Nombre maximal de requêtes autorisées par seconde. Par défaut 5.0.

 5.0
Source code in kadi/kidas/sources/api_source.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    url: str,
    token: Optional[str] = None,
    rate_limit: float = 5.0,
) -> None:
    """Initialise la source API avec gestion du rate limiting.

    Args:
        url (str): URL de base de l'endpoint API (sans paramètres).
        token (str | None): Jeton Bearer pour les APIs authentifiées.
            None pour les APIs publiques (Open-Meteo, etc.).
        rate_limit (float): Nombre maximal de requêtes autorisées
            par seconde. Par défaut 5.0.
    """
    # Initialisation de la classe parente avec le type 'api'
    super().__init__(
        path=url,
        kind="api",
        encoding="utf-8",
    )

    # URL de base de l'endpoint API
    self.url: str = url

    # Jeton d'authentification (None pour APIs publiques)
    self.token: Optional[str] = token

    # Limite de débit en requêtes/seconde
    self.rate_limit: float = rate_limit

    # Timestamp de la dernière requête (pour le throttling)
    self._last_req: float = 0.0

    # Cache du schéma API
    self._schema: Optional[dict] = None

fetch 

fetch(params: Dict[str, Any], retries: int = 3, backoff: float = 5.0) -> dict

Effectue une requête GET avec réessais et backoff exponentiel.

En cas d'erreur temporaire (codes 429, 500, 502, 503, 504), réessaie automatiquement avec un délai croissant.

Parameters:

Name Type Description Default
params dict

Paramètres de la requête GET (query string).

 required
retries int

Nombre maximum de tentatives en cas d'échec. Par défaut 3.

 3
backoff float

Délai de base (en secondes) entre les réessais. Le délai est doublé à chaque tentative. Par défaut 5.0.

 5.0

Returns:

Name Type Description
dict dict

La réponse JSON parsée de l'API.

Raises:

Type Description
ConnectError

Si l'API est inaccessible après toutes les tentatives.
ReadError

Si la réponse n'est pas un JSON valide.
Source code in kadi/kidas/sources/api_source.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def fetch(
    self,
    params: Dict[str, Any],
    retries: int = 3,
    backoff: float = 5.0,
) -> dict:
    """Effectue une requête GET avec réessais et backoff exponentiel.

    En cas d'erreur temporaire (codes 429, 500, 502, 503, 504),
    réessaie automatiquement avec un délai croissant.

    Args:
        params (dict): Paramètres de la requête GET (query string).
        retries (int): Nombre maximum de tentatives en cas d'échec.
            Par défaut 3.
        backoff (float): Délai de base (en secondes) entre les
            réessais. Le délai est doublé à chaque tentative.
            Par défaut 5.0.

    Returns:
        dict: La réponse JSON parsée de l'API.

    Raises:
        ConnectError: Si l'API est inaccessible après toutes
            les tentatives.
        ReadError: Si la réponse n'est pas un JSON valide.
    """
    derniere_erreur: Optional[Exception] = None

    for tentative in range(retries + 1):
        try:
            # Respect du rate limit avant chaque requête
            self._throttle()

            # Enregistrement du timestamp de la requête
            self._last_req = time.monotonic()

            # Exécution de la requête HTTP GET
            reponse = requests.get(
                self.url,
                params=params,
                headers=self._build_headers(),
                timeout=30,
            )

            # Journalisation de la requête
            logger.debug(
                "Requête API [tentative %d/%d] : %s (status: %d).",
                tentative + 1,
                retries + 1,
                reponse.url,
                reponse.status_code,
            )

            # Vérification si le code est réessayable
            if reponse.status_code in _HTTP_RETRY_CODES and tentative < retries:
                # Calcul du backoff exponentiel
                delai_attente = min(backoff * (2 ** tentative), _MAX_BACKOFF_SEC)
                logger.warning(
                    "Erreur HTTP %d. Réessai dans %.1f secondes (tentative %d/%d).",
                    reponse.status_code,
                    delai_attente,
                    tentative + 1,
                    retries,
                )
                time.sleep(delai_attente)
                continue

            # Levée d'exception pour les codes d'erreur non-réessayables
            reponse.raise_for_status()

            # Parse et retour de la réponse JSON
            return reponse.json()

        except requests.exceptions.ConnectionError as erreur:
            derniere_erreur = erreur
            if tentative < retries:
                delai_attente = min(backoff * (2 ** tentative), _MAX_BACKOFF_SEC)
                logger.warning(
                    "Erreur de connexion. Réessai dans %.1f sec.", delai_attente
                )
                time.sleep(delai_attente)

        except requests.exceptions.HTTPError as erreur:
            # Erreur HTTP non-réessayable
            raise ConnectError(
                f"Erreur HTTP lors de l'appel à '{self.url}' : {erreur}"
            ) from erreur

        except ValueError as erreur:
            # Réponse non-JSON
            raise ReadError(
                f"La réponse de '{self.url}' n'est pas du JSON valide : {erreur}"
            ) from erreur

    # Toutes les tentatives ont échoué
    raise ConnectError(
        f"API '{self.url}' inaccessible après {retries + 1} tentatives. "
        f"Dernière erreur : {derniere_erreur}"
    )

read 

read(params: Optional[Dict[str, Any]] = None) -> pd.DataFrame

Effectue une requête GET et retourne la réponse sous forme de DataFrame.

Gère automatiquement les réponses JSON de type dict (avec clé 'data' ou 'results') et les réponses de type liste.

Parameters:

Name Type Description Default
params dict | None

Paramètres de la requête GET. Par défaut None (requête sans paramètres).

 None

Returns:

Type Description
DataFrame

pd.DataFrame: Les données de la réponse API sous forme tabulaire.

Raises:

Type Description
ConnectError

Si l'API est inaccessible.
ReadError

Si la conversion en DataFrame échoue.
Source code in kadi/kidas/sources/api_source.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def read(self, params: Optional[Dict[str, Any]] = None) -> pd.DataFrame:
    """Effectue une requête GET et retourne la réponse sous forme de DataFrame.

    Gère automatiquement les réponses JSON de type dict (avec clé 'data'
    ou 'results') et les réponses de type liste.

    Args:
        params (dict | None): Paramètres de la requête GET.
            Par défaut None (requête sans paramètres).

    Returns:
        pd.DataFrame: Les données de la réponse API sous forme tabulaire.

    Raises:
        ConnectError: Si l'API est inaccessible.
        ReadError: Si la conversion en DataFrame échoue.
    """
    # Paramètres de requête par défaut
    parametres = params or {}

    # Exécution de la requête avec réessais
    donnees_brutes = self.fetch(parametres)

    try:
        # Normalisation de la structure de réponse en liste d'enregistrements
        if isinstance(donnees_brutes, list):
            # Réponse déjà sous forme de liste
            enregistrements = donnees_brutes
        elif isinstance(donnees_brutes, dict):
            # Recherche d'une clé standard contenant les données
            for cle_donnees in ("data", "results", "items", "records", "features"):
                if cle_donnees in donnees_brutes:
                    enregistrements = donnees_brutes[cle_donnees]
                    break
            else:
                # Pas de clé standard : utilise le dict entier
                enregistrements = [donnees_brutes]
        else:
            raise ReadError(
                f"Format de réponse non supporté : {type(donnees_brutes).__name__}"
            )

        # Conversion en DataFrame pandas
        df = pd.json_normalize(enregistrements)

        logger.info(
            "Données API '%s' reçues : %d lignes, %d colonnes.",
            self.url,
            len(df),
            len(df.columns),
        )

        # Mise à jour de l'horodatage de lecture
        self._touch()
        return df

    except ReadError:
        raise
    except Exception as erreur:
        raise ReadError(
            f"Impossible de convertir la réponse de '{self.url}' "
            f"en DataFrame : {erreur}"
        ) from erreur

write 

write(data: DataFrame) -> bool

Envoie des données vers l'API via une requête POST.

Parameters:

Name Type Description Default
data DataFrame

Les données à envoyer à l'API en JSON.

 required

Returns:

Name Type Description
bool bool

True si l'envoi a réussi (code 200 ou 201).

Raises:

Type Description
WriteError

Si la requête POST échoue.
Source code in kadi/kidas/sources/api_source.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def write(self, data: pd.DataFrame) -> bool:
    """Envoie des données vers l'API via une requête POST.

    Args:
        data (pd.DataFrame): Les données à envoyer à l'API en JSON.

    Returns:
        bool: True si l'envoi a réussi (code 200 ou 201).

    Raises:
        WriteError: Si la requête POST échoue.
    """
    try:
        # Conversion du DataFrame en liste de dictionnaires JSON
        payload = data.to_dict(orient="records")

        # Respect du rate limit avant la requête
        self._throttle()
        self._last_req = time.monotonic()

        # Envoi de la requête POST
        reponse = requests.post(
            self.url,
            json=payload,
            headers=self._build_headers(),
            timeout=30,
        )
        reponse.raise_for_status()

        logger.info(
            "Données envoyées avec succès à '%s' (%d enregistrements, "
            "status: %d).",
            self.url,
            len(data),
            reponse.status_code,
        )
        return True

    except requests.exceptions.HTTPError as erreur:
        raise WriteError(
            f"Erreur HTTP lors du POST vers '{self.url}' : {erreur}"
        ) from erreur
    except Exception as erreur:
        raise WriteError(
            f"Impossible d'envoyer les données vers '{self.url}' : {erreur}"
        ) from erreur

info 

info() -> dict

Retourne les métadonnées de la connexion API.

Returns:

Name Type Description
dict dict

Dictionnaire contenant les clés suivantes : - 'url' (str) : URL de l'API. - 'kind' (str) : 'api'. - 'requires_auth' (bool) : True si un jeton est requis. - 'rate_limit' (float) : limite de débit en req/sec. - 'last_read' (str | None) : horodatage de la dernière lecture.
Source code in kadi/kidas/sources/api_source.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def info(self) -> dict:
    """Retourne les métadonnées de la connexion API.

    Returns:
        dict: Dictionnaire contenant les clés suivantes :
            - 'url' (str) : URL de l'API.
            - 'kind' (str) : 'api'.
            - 'requires_auth' (bool) : True si un jeton est requis.
            - 'rate_limit' (float) : limite de débit en req/sec.
            - 'last_read' (str | None) : horodatage de la dernière lecture.
    """
    return {
        "url": self.url,
        "kind": "api",
        "requires_auth": self.token is not None,
        "rate_limit": self.rate_limit,
        "last_read": (
            self.last_read.isoformat() if self.last_read else None
        ),
    }

ping 

ping() -> bool

Vérifie que l'endpoint API est accessible via une requête HEAD.

Returns:

Name Type Description
bool bool

True si l'endpoint répond (même avec une erreur 4xx), False si la connexion échoue complètement.
Source code in kadi/kidas/sources/api_source.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
def ping(self) -> bool:
    """Vérifie que l'endpoint API est accessible via une requête HEAD.

    Returns:
        bool: True si l'endpoint répond (même avec une erreur 4xx),
            False si la connexion échoue complètement.
    """
    try:
        # Requête HEAD légère pour tester la connectivité
        reponse = requests.head(
            self.url,
            headers=self._build_headers(),
            timeout=10,
        )
        # Tout code de réponse inférieur à 500 indique que l'API est accessible
        est_accessible = reponse.status_code < 500
        logger.debug(
            "Validation connexion API '%s' : status %d -> %s.",
            self.url,
            reponse.status_code,
            "OK" if est_accessible else "KO",
        )
        return est_accessible

    except requests.exceptions.RequestException as erreur:
        logger.warning(
            "API '%s' inaccessible : %s", self.url, erreur
        )
        return False