Skip to content

Data Sources (kadi.kidas.sources)

The kadi.kidas.sources subpackage gathers all classes for reading and writing agricultural data from various formats. Each concrete class inherits from Source, the abstract base class, and implements the same contract: read(), write(), info(), ping().

Class Format Extensions / Prefix
CSVSource Delimited text files .csv
ExcelSource Excel workbooks .xlsx, .xls
JSONSource JSON files or Python dicts .json
NetCDFSource Gridded agrometeorological data .nc, .netcdf
APISource HTTP REST APIs http://, https://

Import

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

# or directly from the subpackage
from kadi.kidas.sources import CSVSource

The classes are also re-exported from kadi.io. See Input and Output for documentation on read_*, write_*, and write functions.

Source (Abstract Base Class)

Source is the shared interface for all data sources. It cannot be instantiated directly.

Common Attributes:

Attribute Type Description
path str Local file path or resource URI
kind str Source type: 'csv', 'excel', 'json', 'netcdf', 'api'
encoding str Data encoding (can be 'auto')
last_read datetime or None Timestamp of the last successful read

Abstract Methods Implemented by Each Subclass:

  • read(**kwargs): reads data and returns a pd.DataFrame
  • write(data, **kwargs): writes a DataFrame to the source, returns bool
  • info(): returns a metadata dict
  • ping(): returns True if the source is accessible

CSVSource

Source for agricultural CSV files with automatic detection of encoding (via chardet), delimiter, and decimal separator. Designed for exports from IoT sensors, cooperatives, and national databases (INSAE, MAEP).

Initialization

from kadi.kidas import CSVSource

# Automatic detection of everything
source = CSVSource("recoltes_2024.csv")

# Explicit parameters
source = CSVSource(
    "export_insae.csv",
    encoding="latin-1",
    sep=";",
    decimal=",",
)
Parameter Type Default Description
path str required Path to the CSV file
encoding str 'auto' Encoding. 'auto': detected via chardet
sep str 'auto' Column delimiter. 'auto': auto-detected
decimal str 'auto' Decimal separator. 'auto': inferred from delimiter

In 'auto' mode, encoding is detected from a 10,000-byte sample. Delimiter is detected by csv.Sniffer with fallback frequency counting among ,, ;, \t, |. If delimiter is ;, the inferred decimal separator is , (French convention).

read(n=None, skip=None)

df = source.read()

# First 100 rows
df = source.read(n=100)

# Skip 2 metadata rows at file start
df = source.read(skip=2)
Parameter Type Default Description
n int or None None Maximum number of rows to read
skip int or None None Rows to skip at file start (excluding header)

Tries multiple fallback encodings (utf-8, latin-1, cp1252) if primary reading fails.

Returns: pd.DataFrame Exceptions: ConnectError if file is inaccessible, ReadError if all encodings fail.

write(data, index=False)

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

Writes in UTF-8 if encoding is 'auto'. Uses detected delimiter or , by default.

Returns: bool Exception: WriteError

info()

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

ping()

Checks if file exists and is readable. Returns bool.

ExcelSource

Source for Excel files (.xlsx, .xls) with automatic header row detection and merged cell resolution via forward fill. Adapted for Beninese agricultural reports with municipalities and markets merged across multiple rows.

Initialization

from kadi.kidas import ExcelSource

source = ExcelSource("prix_marches_2024.xlsx")

# Explicit sheet and header
source = ExcelSource("rapport.xlsx", sheet="Janvier", header=2)
Parameter Type Default Description
path str required Path to the Excel file
sheet str or int 0 Default sheet (name or index)
header str or int 'auto' Header row. 'auto': automatically detected

Header detection inspects the first 15 rows and selects the densest row (highest count of non-null values).

sheets()

Returns the list of sheet names in the workbook.

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

Returns: list[str] Exceptions: ConnectError, ReadError

sheet_info(sheet)

Returns metadata for a specific sheet.

meta = source.sheet_info("Janvier")
# {
#   'sheet': 'Janvier',
#   'rows': 245,
#   'cols': 6,
#   'columns': ['culture', 'marche', 'prix_xof_kg', ...]
# }
Parameter Type Description
sheet str or int Sheet name or index to inspect

Returns: dict Exception: ReadError

read(sheet=None)

df = source.read()

# Read a specific sheet
df = source.read(sheet="Fevrier")

Applies a vertical forward fill on all columns to resolve merged cells, then drops completely empty rows.

Parameter Type Default Description
sheet str, int, or None None Sheet to read. None: uses sheet defined at initialization

Returns: pd.DataFrame Exceptions: ConnectError, ReadError

write(data, sheet='Sheet1')

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

Returns: 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()

Checks if file exists and is readable. Returns bool.

JSONSource

Source for flat or nested JSON files, such as returned by agricultural APIs (FAO, WFP VAM, MAEP). Also supports an in-memory Python dictionary directly.

Initialization

from kadi.kidas import JSONSource

# From a file
source = JSONSource("donnees_fao.json")

# From an in-memory Python dictionary
data = {"location": {"lat": 9.3, "lon": 2.4}, "crop": "maize"}
source = JSONSource(data)
Parameter Type Description
file_path_or_dict str or dict Path to JSON file or Python dictionary

When the source is a dictionary, path equals '<dict_in_memory>' and ping() always returns True.

flatten(obj, sep='.')

Recursively flattens a nested JSON object using dot notation.

obj = {"location": {"lat": 9.3, "lon": 2.4}, "crop": "maize"}
flat = source.flatten(obj)
# {"location.lat": 9.3, "location.lon": 2.4, "crop": "maize"}
Parameter Type Default Description
obj dict required JSON object to flatten
sep str '.' Separator between key levels

Lists are indexed by position (location.items.0, location.items.1...).

Returns: dict

read(flatten=True)

df = source.read()

# Without flattening (direct pd.json_normalize)
df = source.read(flatten=False)
Parameter Type Default Description
flatten bool True If True, flattens each record before conversion

Accepts JSON responses of type dict (single record) or list (list of records).

Returns: pd.DataFrame Exceptions: ConnectError, ReadError

write(data, orient='records')

source.write(df)
source.write(df, orient="index")
Parameter Type Default Description
data pd.DataFrame required Data to write
orient str 'records' JSON orientation: 'records', 'index', 'columns', 'values'

Writes with force_ascii=False and indent=2.

Exception: WriteError if source is an in-memory dictionary (no destination file).

Returns: bool

info()

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

ping()

For file source: checks existence and readability. For dict source: always returns True. Returns bool.

NetCDFSource

Source for agrometeorological NetCDF files (.nc), used for CHIRPS (precipitation), TAMSAT (Africa), and GFS (global forecasts) data. Uses xarray as reading engine and supports Dask chunking for large files.

Dependency: xarray and netCDF4 must be installed.

Initialization

from kadi.kidas import NetCDFSource

source = NetCDFSource("chirps_benin_2024.nc")

# With Dask for large files (> 500 MB)
source = NetCDFSource("gfs_global_2024.nc", use_dask=True)
Parameter Type Default Description
path str required Path to the NetCDF file
use_dask bool False Enables lazy loading via Dask

dims()

Returns dataset dimensions.

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

Returns: dict[str, int]

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

Extracts a spatial and temporal subset of the file. If no bounding box is specified, defaults to the Benin bbox (lat in [2.5, 12.5], lon in [-1.5, 4.0]).

# Extraction with default Benin bbox
da = source.read()

# Extraction on a custom region
da = source.read(
    lat_bounds=(6.0, 12.5),
    lon_bounds=(1.0, 4.0),
    time_bounds=("2024-01-01", "2024-06-30"),
)
Parameter Type Default Description
lat_bounds tuple[float, float] or None Benin bbox Latitude interval (lat_min, lat_max)
lon_bounds tuple[float, float] or None Benin bbox Longitude interval (lon_min, lon_max)
time_bounds tuple[str, str] or None None Time interval in ISO format ('YYYY-MM-DD', 'YYYY-MM-DD')

Automatically detects lat/lon dimension names in the dataset (compatible with 'lat', 'latitude', 'y', etc.). Reads the first data variable if dataset contains multiple variables.

Returns: xr.DataArray Exceptions: ConnectError, ReadError

to_df()

Converts the last DataArray extracted by read() to a pd.DataFrame. If read() has not been called yet, performs a read with the Benin bbox.

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

Returns: pd.DataFrame Exception: ReadError

write(data)

Converts the DataFrame to xr.Dataset via from_dataframe() and saves it in NetCDF format.

Returns: 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()

Checks if file exists and is readable. Returns bool.

APISource

Source for agricultural and climate REST APIs (Open-Meteo, WFP VAM, FAO, SoilGrids). Features rate limiting and retry mechanism with exponential backoff.

Initialization

from kadi.kidas import APISource

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

# Authenticated API with custom rate limit
source = APISource(
    "https://api.wfp.org/vam-data-bridges/1.0",
    token="my_bearer_token",
    rate_limit=2.0,
)
Parameter Type Default Description
url str required Base URL of the API endpoint
token str or None None Bearer token for authenticated APIs
rate_limit float 5.0 Maximum number of requests per second

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

Performs a GET request with retries and exponential backoff.

data = source.fetch(
    params={"latitude": 6.36, "longitude": 2.42},
    retries=3,
    backoff=5.0,
)
Parameter Type Default Description
params dict required GET request query parameters
retries int 3 Maximum number of attempts on failure
backoff float 5.0 Base delay in seconds between retries (doubled each attempt, capped at 60 s)

Retryable HTTP status codes are: 429, 500, 502, 503, 504.

Returns: dict (parsed JSON response) Exceptions: ConnectError if API is unreachable, ReadError if response is not valid JSON.

read(params=None)

Performs a GET request and normalizes the response to a DataFrame.

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

Automatically searches for a standard key in dict response ('data', 'results', 'items', 'records', 'features'). If none is found, uses the whole dict as a single record.

Returns: pd.DataFrame Exceptions: ConnectError, ReadError

write(data)

Sends the DataFrame to the API via a POST request in JSON format (orient='records').

Returns: 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()

Performs a lightweight HEAD request. Returns True if HTTP status code is below 500 (API responds, even with 4xx error). Returns False if connection fails completely.

Full Example

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

# CSV with automatic detection
source_csv = CSVSource("recoltes_2024.csv")
if source_csv.ping():
    df = source_csv.read()
    print(source_csv.info())

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

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

# Agrometeorological NetCDF
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()

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

Backward Compatibility

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

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