The kadi.io module is the entry point for all data reading and writing operations in KadiPy. It covers local files (CSV, Excel, JSON, NetCDF) and REST APIs, with automatic format detection based on file extension or URL.
The source classes (CSVSource, ExcelSource, etc.) are defined in kadi.kidas.sources and re-exported here for direct access. The read_* and write_* functions are shortcuts that instantiate these classes and immediately call their read() or write() method.
Quick Start
import kadi as kd
# Reading
df = kd.read_csv("recoltes_2024.csv")
df = kd.read_excel("prix_marche.xlsx")
df = kd.read_json("campagne_2023.json")
df = kd.read_api("https://api.data.bj/agriculture/prices")
# Writing
kd.write_csv(df, "export.csv")
kd.write(df, "export.xlsx") # Automatic format detection
# Inspection
kd.info("recoltes_2024.csv") # Metadata without loading the full file
kd.ping("recoltes_2024.csv") # Check if source is accessible
Read Functions
These functions read data from a source and return a pandas.DataFrame. Each function accepts optional arguments passed down to the underlying source class.
read_csv(filepath, **kwargs)
Reads a CSV file. Automatically detects encoding (via chardet), delimiter (,, ;, tab, |), and decimal separator.
import kadi as kd
df = kd.read_csv("recoltes_2024.csv")
# Override parameters if automatic detection is not sufficient
from kadi.io import CSVSource
df = CSVSource("export.csv", sep=";", decimal=",", encoding="latin-1").read()
| Parameter |
Type |
Default |
Description |
filepath |
str or Path |
required |
Path to the CSV file |
encoding |
str |
'auto' |
File encoding ('utf-8', 'latin-1', etc.) |
sep |
str |
'auto' |
Column delimiter |
decimal |
str |
'auto' |
Decimal separator |
n |
int |
None |
Maximum number of rows to read |
skip |
int |
None |
Number of rows to skip at file start |
Returns: pandas.DataFrame
Exceptions: ConnectError if file is not found, ReadError if reading fails despite trying multiple encodings.
read_excel(filepath, **kwargs)
Reads an Excel file (.xlsx or .xls).
df = kd.read_excel("prix_marche.xlsx")
# Read a specific sheet
from kadi.io import ExcelSource
df = ExcelSource("rapport.xlsx", sheet="Parakou").read()
Returns: pandas.DataFrame
read_json(filepath, **kwargs)
Reads a JSON file.
df = kd.read_json("campagne_2023.json")
Returns: pandas.DataFrame
read_netcdf(filepath, **kwargs)
Reads a NetCDF file (.nc, .netcdf). Requires optional dependencies xarray and netCDF4.
df = kd.read_netcdf("chirps_2024.nc")
Returns: xarray.Dataset or pandas.DataFrame depending on options passed.
Exception: ImportError if xarray or netCDF4 is not installed.
# Installing dependencies
pip install "kadipy[geospatial]"
read_api(url, params=None, **kwargs)
Reads data from a REST API via a GET request.
df = kd.read_api("https://api.data.bj/agriculture/prices")
# With query parameters
df = kd.read_api(
"https://api.data.bj/agriculture/prices",
params={"crop": "maize", "region": "parakou"},
)
| Parameter |
Type |
Default |
Description |
url |
str |
required |
REST API URL |
params |
dict |
None |
HTTP query parameters |
Returns: pandas.DataFrame
Targeted Write Functions
These functions write a pandas.DataFrame to a file or API. They all return True if writing succeeds.
write_csv(data, filepath, **kwargs)
kd.write_csv(df, "export.csv")
Uses delimiter and encoding detected during reading if the source was read previously, or , / utf-8 by default.
write_excel(data, filepath, **kwargs)
kd.write_excel(df, "export.xlsx")
write_json(data, filepath, **kwargs)
kd.write_json(df, "export.json")
write_netcdf(data, filepath, **kwargs)
kd.write_netcdf(df, "export.nc")
Same optional dependencies as read_netcdf.
write_api(data, url, **kwargs)
Sends a DataFrame to a REST API endpoint.
kd.write_api(df, "https://api.monservice.bj/upload")
Generic Functions
These three functions accept any format and automatically detect the source to use based on file extension or URL prefix.
Automatic Detection Table
| Extension or Prefix |
Class Used |
.csv |
CSVSource |
.xlsx, .xls |
ExcelSource |
.json |
JSONSource |
.nc, .netcdf |
NetCDFSource |
http://, https:// |
APISource |
| Other |
ValueError |
write(data, filepath_or_url, **kwargs)
Writes a DataFrame with automatic format detection.
kd.write(df, "export.csv")
kd.write(df, "export.xlsx")
kd.write(df, "https://api.monservice.bj/upload")
info(filepath_or_url, **kwargs)
Returns metadata of a source without loading all data into memory.
meta = kd.info("recoltes_2024.csv")
print(meta)
# {
# 'path': 'recoltes_2024.csv',
# 'kind': 'csv',
# 'encoding': 'utf-8',
# 'sep': ',',
# 'decimal': '.',
# 'rows': 1240,
# 'cols': 8,
# 'size_kb': 42.5,
# 'last_read': None
# }
Returns: dict containing source metadata.
Returned keys vary depending on source type:
| Key |
Description |
path |
Source path or URL |
kind |
Source type: 'csv', 'excel', 'json', 'netcdf', 'api' |
encoding |
Detected or configured encoding |
rows |
Number of data rows |
cols |
Number of columns |
size_kb |
File size in kilobytes (local files only) |
last_read |
ISO timestamp of last read, or None |
ping(filepath_or_url, **kwargs)
Checks if a source is accessible without reading it.
if kd.ping("recoltes_2024.csv"):
df = kd.read_csv("recoltes_2024.csv")
# For an API
if kd.ping("https://api.data.bj/agriculture/prices"):
df = kd.read_api("https://api.data.bj/agriculture/prices")
Returns: True if source is accessible, False otherwise.
For local files, checks that the file exists and is readable.
For APIs, checks that the endpoint responds without HTTP error.
Accessing Source Classes
Classes Source, CSVSource, ExcelSource, JSONSource, NetCDFSource, and APISource are available directly from kadi.io for advanced usage.
from kadi.io import CSVSource
source = CSVSource("recoltes.csv", sep=";", decimal=",")
df = source.read(n=100) # Read first 100 rows
meta = source.info() # Metadata
ok = source.ping() # Accessibility test
source.write(df, "copie.csv") # Rewriting
Source classes implement the abstract base class Source, which defines the shared contract: read(), write(), info(), ping().
For complete documentation on each source class, see kadi.kidas: Data Sources.
Root Level Access
All functions and classes from kadi.io are re-exported from kadi for direct access:
import kadi as kd
# Equivalent to from kadi.io import ...
kd.read_csv(...)
kd.write(...)
kd.info(...)
kd.ping(...)
kd.CSVSource(...)
io
Module kadi.io - Ingestion et exportation de données agricoles.
Ce module est le point d'entrée pour toutes les sources de données
supportées par KadiPy : fichiers locaux (CSV, Excel, JSON, NetCDF)
et APIs REST.
Il expose :
- Les classes de sources concrètes (CSVSource, ExcelSource, etc.)
- Les fonctions de lecture raccourcies (read_csv, read_excel, etc.)
- Les fonctions d'écriture raccourcies (write_csv, write_excel, etc.)
- Les fonctions génériques avec détection automatique du format (write, info, ping)
Exemples d'utilisation :
>>> from kadi.io import read_csv, write_csv
>>> df = read_csv("recoltes_2024.csv")
>>> write_csv(df, "recoltes_copie.csv")
>>> import kadi.io as io
>>> io.write(df, "donnees.json")
>>> meta = io.info("recoltes_2024.csv")
>>> ok = io.ping("recoltes_2024.csv")
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
| 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
| 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
| @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
|
|
Source code in kadi/kidas/sources/base.py
| @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
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
| @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
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
| @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__
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
| 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
| 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
| 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
| 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
| 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
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
| 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
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
| 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
| 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
| 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
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
| 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
| 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
|
|
Source code in kadi/kidas/sources/excel_source.py
| 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
| 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
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
| 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
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
| 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
| 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
| 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
| 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
| 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
| 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
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
| 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
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
| 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
|
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
| 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
| 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
| 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
| 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
| 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
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
| 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
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
| 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
|
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
| 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
| 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
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
| 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
| 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
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
|
|
Source code in kadi/kidas/sources/netcdf_source.py
| 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
|
|
Source code in kadi/kidas/sources/netcdf_source.py
| 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
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
| 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
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
| 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
|
read_csv
read_csv(filepath, **kwargs)
Lit un fichier CSV via CSVSource.
Parameters:
| Name |
Type |
Description |
Default |
filepath
|
str | Path
|
Chemin vers le fichier CSV.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à CSVSource.
|
{}
|
Returns:
| Type |
Description |
|
|
pandas.DataFrame: Contenu du fichier CSV sous forme de DataFrame.
|
Source code in kadi/io/__init__.py
| def read_csv(filepath, **kwargs):
"""
Lit un fichier CSV via CSVSource.
Args:
filepath (str | Path): Chemin vers le fichier CSV.
**kwargs: Paramètres additionnels transmis à CSVSource.
Returns:
pandas.DataFrame: Contenu du fichier CSV sous forme de DataFrame.
"""
# Instanciation de la source CSV et lecture des données
return CSVSource(filepath, **kwargs).read()
|
read_excel
read_excel(filepath, **kwargs)
Lit un fichier Excel via ExcelSource.
Parameters:
| Name |
Type |
Description |
Default |
filepath
|
str | Path
|
Chemin vers le fichier Excel.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à ExcelSource.
|
{}
|
Returns:
| Type |
Description |
|
|
pandas.DataFrame: Contenu du fichier Excel sous forme de DataFrame.
|
Source code in kadi/io/__init__.py
| def read_excel(filepath, **kwargs):
"""
Lit un fichier Excel via ExcelSource.
Args:
filepath (str | Path): Chemin vers le fichier Excel.
**kwargs: Paramètres additionnels transmis à ExcelSource.
Returns:
pandas.DataFrame: Contenu du fichier Excel sous forme de DataFrame.
"""
# Instanciation de la source Excel et lecture des données
return ExcelSource(filepath, **kwargs).read()
|
read_json
read_json(filepath, **kwargs)
Lit un fichier JSON via JSONSource.
Parameters:
| Name |
Type |
Description |
Default |
filepath
|
str | Path
|
Chemin vers le fichier JSON.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à JSONSource.
|
{}
|
Returns:
| Type |
Description |
|
|
pandas.DataFrame: Contenu du fichier JSON sous forme de DataFrame.
|
Source code in kadi/io/__init__.py
| def read_json(filepath, **kwargs):
"""
Lit un fichier JSON via JSONSource.
Args:
filepath (str | Path): Chemin vers le fichier JSON.
**kwargs: Paramètres additionnels transmis à JSONSource.
Returns:
pandas.DataFrame: Contenu du fichier JSON sous forme de DataFrame.
"""
# Instanciation de la source JSON et lecture des données
return JSONSource(filepath, **kwargs).read()
|
read_netcdf
read_netcdf(filepath, **kwargs)
Lit un fichier NetCDF via NetCDFSource.
Parameters:
| Name |
Type |
Description |
Default |
filepath
|
str | Path
|
Chemin vers le fichier NetCDF.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à NetCDFSource.
|
{}
|
Returns:
| Type |
Description |
|
|
xarray.Dataset | pandas.DataFrame: Contenu du fichier NetCDF.
|
Raises:
| Type |
Description |
ImportError
|
Si xarray ou netCDF4 n'est pas disponible.
|
Source code in kadi/io/__init__.py
| def read_netcdf(filepath, **kwargs):
"""
Lit un fichier NetCDF via NetCDFSource.
Args:
filepath (str | Path): Chemin vers le fichier NetCDF.
**kwargs: Paramètres additionnels transmis à NetCDFSource.
Returns:
xarray.Dataset | pandas.DataFrame: Contenu du fichier NetCDF.
Raises:
ImportError: Si xarray ou netCDF4 n'est pas disponible.
"""
# Vérification de la présence du module NetCDFSource
if NetCDFSource is None:
raise ImportError(
"Le support NetCDF nécessite l'installation des dépendances xarray et netCDF4."
)
# Instanciation de la source NetCDF et lecture des données
return NetCDFSource(filepath, **kwargs).read()
|
read_api
read_api(url: str, params: dict = None, **kwargs)
Lit des données depuis une API REST via APISource.
Raccourci de haut niveau pour exécuter une requête GET vers une API.
Parameters:
| Name |
Type |
Description |
Default |
url
|
str
|
|
required
|
params
|
dict
|
Paramètres de requête HTTP.
|
None
|
**kwargs
|
|
Paramètres additionnels transmis à APISource.
|
{}
|
Returns:
| Type |
Description |
|
|
pandas.DataFrame: Données renvoyées par l'API sous forme de DataFrame.
|
Source code in kadi/io/__init__.py
| def read_api(url: str, params: dict = None, **kwargs):
"""
Lit des données depuis une API REST via APISource.
Raccourci de haut niveau pour exécuter une requête GET vers une API.
Args:
url (str): URL de l'API.
params (dict, optional): Paramètres de requête HTTP.
**kwargs: Paramètres additionnels transmis à APISource.
Returns:
pandas.DataFrame: Données renvoyées par l'API sous forme de DataFrame.
"""
# Instanciation de la source API et lecture des données
source = APISource(url, **kwargs)
return source.read(params or {})
|
write_csv
write_csv(data, filepath: str, **kwargs) -> bool
Écrit un DataFrame dans un fichier CSV via CSVSource.
Parameters:
| Name |
Type |
Description |
Default |
data
|
DataFrame
|
|
required
|
filepath
|
str
|
Chemin du fichier CSV de destination.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à CSVSource.write().
|
{}
|
Returns:
| Name | Type |
Description |
bool |
bool
|
True si l'écriture est réussie.
|
Source code in kadi/io/__init__.py
| def write_csv(data, filepath: str, **kwargs) -> bool:
"""
Écrit un DataFrame dans un fichier CSV via CSVSource.
Args:
data (pandas.DataFrame): Les données à exporter.
filepath (str): Chemin du fichier CSV de destination.
**kwargs: Paramètres additionnels transmis à CSVSource.write().
Returns:
bool: True si l'écriture est réussie.
"""
# Instanciation de la source CSV et écriture des données
return CSVSource(filepath, **kwargs).write(data)
|
write_excel
write_excel(data, filepath: str, **kwargs) -> bool
Écrit un DataFrame dans un fichier Excel via ExcelSource.
Parameters:
| Name |
Type |
Description |
Default |
data
|
DataFrame
|
|
required
|
filepath
|
str
|
Chemin du fichier Excel de destination.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à ExcelSource.write().
|
{}
|
Returns:
| Name | Type |
Description |
bool |
bool
|
True si l'écriture est réussie.
|
Source code in kadi/io/__init__.py
| def write_excel(data, filepath: str, **kwargs) -> bool:
"""
Écrit un DataFrame dans un fichier Excel via ExcelSource.
Args:
data (pandas.DataFrame): Les données à exporter.
filepath (str): Chemin du fichier Excel de destination.
**kwargs: Paramètres additionnels transmis à ExcelSource.write().
Returns:
bool: True si l'écriture est réussie.
"""
# Instanciation de la source Excel et écriture des données
return ExcelSource(filepath, **kwargs).write(data)
|
write_json
write_json(data, filepath: str, **kwargs) -> bool
Écrit un DataFrame dans un fichier JSON via JSONSource.
Parameters:
| Name |
Type |
Description |
Default |
data
|
DataFrame
|
|
required
|
filepath
|
str
|
Chemin du fichier JSON de destination.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à JSONSource.write().
|
{}
|
Returns:
| Name | Type |
Description |
bool |
bool
|
True si l'écriture est réussie.
|
Source code in kadi/io/__init__.py
| def write_json(data, filepath: str, **kwargs) -> bool:
"""
Écrit un DataFrame dans un fichier JSON via JSONSource.
Args:
data (pandas.DataFrame): Les données à exporter.
filepath (str): Chemin du fichier JSON de destination.
**kwargs: Paramètres additionnels transmis à JSONSource.write().
Returns:
bool: True si l'écriture est réussie.
"""
# Instanciation de la source JSON et écriture des données
return JSONSource(filepath, **kwargs).write(data)
|
write_netcdf
write_netcdf(data, filepath: str, **kwargs) -> bool
Écrit des données dans un fichier NetCDF via NetCDFSource.
Parameters:
| Name |
Type |
Description |
Default |
data
|
DataFrame | Dataset
|
|
required
|
filepath
|
str
|
Chemin du fichier NetCDF de destination.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à NetCDFSource.write().
|
{}
|
Returns:
| Name | Type |
Description |
bool |
bool
|
True si l'écriture est réussie.
|
Raises:
| Type |
Description |
ImportError
|
Si xarray ou netCDF4 n'est pas disponible.
|
Source code in kadi/io/__init__.py
| def write_netcdf(data, filepath: str, **kwargs) -> bool:
"""
Écrit des données dans un fichier NetCDF via NetCDFSource.
Args:
data (pandas.DataFrame | xarray.Dataset): Les données à exporter.
filepath (str): Chemin du fichier NetCDF de destination.
**kwargs: Paramètres additionnels transmis à NetCDFSource.write().
Returns:
bool: True si l'écriture est réussie.
Raises:
ImportError: Si xarray ou netCDF4 n'est pas disponible.
"""
# Vérification du module NetCDFSource
if NetCDFSource is None:
raise ImportError(
"Le support NetCDF nécessite l'installation des dépendances xarray et netCDF4."
)
# Instanciation de la source NetCDF et écriture des données
return NetCDFSource(filepath, **kwargs).write(data)
|
write_api
write_api(data, url: str, **kwargs) -> bool
Écrit des données vers une API REST via APISource.
Parameters:
| Name |
Type |
Description |
Default |
data
|
DataFrame
|
|
required
|
url
|
str
|
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à APISource.write().
|
{}
|
Returns:
| Name | Type |
Description |
bool |
bool
|
True si l'envoi s'est déroulé avec succès.
|
Source code in kadi/io/__init__.py
| def write_api(data, url: str, **kwargs) -> bool:
"""
Écrit des données vers une API REST via APISource.
Args:
data (pandas.DataFrame): Les données à envoyer.
url (str): URL de l'endpoint API.
**kwargs: Paramètres additionnels transmis à APISource.write().
Returns:
bool: True si l'envoi s'est déroulé avec succès.
"""
# Instanciation de la source API et envoi des données
return APISource(url, **kwargs).write(data)
|
write
write(data, filepath_or_url: str, **kwargs) -> bool
Écrit des données dans une source en détectant automatiquement son format.
Parameters:
| Name |
Type |
Description |
Default |
data
|
DataFrame
|
|
required
|
filepath_or_url
|
str
|
Chemin du fichier ou URL de l'API REST.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à la source.
|
{}
|
Returns:
| Name | Type |
Description |
bool |
bool
|
True si l'écriture est réussie.
|
Source code in kadi/io/__init__.py
| def write(data, filepath_or_url: str, **kwargs) -> bool:
"""
Écrit des données dans une source en détectant automatiquement son format.
Args:
data (pandas.DataFrame): Les données à exporter.
filepath_or_url (str): Chemin du fichier ou URL de l'API REST.
**kwargs: Paramètres additionnels transmis à la source.
Returns:
bool: True si l'écriture est réussie.
"""
# Détection automatique de la source et exécution de l'écriture
source = _detect_source(filepath_or_url, **kwargs)
return source.write(data)
|
info
info(filepath_or_url: str, **kwargs) -> dict
Obtient les métadonnées d'une source en détectant automatiquement son format.
Parameters:
| Name |
Type |
Description |
Default |
filepath_or_url
|
str
|
Chemin du fichier ou URL de l'API REST.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à la source.
|
{}
|
Returns:
| Name | Type |
Description |
dict |
dict
|
Dictionnaire de métadonnées décrivant la source.
|
Source code in kadi/io/__init__.py
| def info(filepath_or_url: str, **kwargs) -> dict:
"""
Obtient les métadonnées d'une source en détectant automatiquement son format.
Args:
filepath_or_url (str): Chemin du fichier ou URL de l'API REST.
**kwargs: Paramètres additionnels transmis à la source.
Returns:
dict: Dictionnaire de métadonnées décrivant la source.
"""
# Détection automatique de la source et récupération des métadonnées
source = _detect_source(filepath_or_url, **kwargs)
return source.info()
|
ping
ping(filepath_or_url: str, **kwargs) -> bool
Vérifie l'accessibilité d'une source de données en détectant son format.
Parameters:
| Name |
Type |
Description |
Default |
filepath_or_url
|
str
|
Chemin du fichier local ou URL de l'API REST.
|
required
|
**kwargs
|
|
Paramètres additionnels transmis à la source.
|
{}
|
Returns:
| Name | Type |
Description |
bool |
bool
|
True si la source est accessible et fonctionnelle.
|
Source code in kadi/io/__init__.py
| def ping(filepath_or_url: str, **kwargs) -> bool:
"""
Vérifie l'accessibilité d'une source de données en détectant son format.
Args:
filepath_or_url (str): Chemin du fichier local ou URL de l'API REST.
**kwargs: Paramètres additionnels transmis à la source.
Returns:
bool: True si la source est accessible et fonctionnelle.
"""
# Détection automatique de la source et contrôle d'accès
source = _detect_source(filepath_or_url, **kwargs)
return source.ping()
|