ENI-projet-piscine/etl

Pipeline ETL — EnerVision

Objectif

Le pipeline ETL EnerVision permet d'intégrer les données énergétiques dans PostgreSQL/TimescaleDB à partir de deux sources :

• le dataset historique CSV/JSON fourni dans le cadre du projet ;
• l'API Mock EnerVision.

Le pipeline assure :

• l'extraction des données sources ;
• la validation de leur structure et de leur cohérence ;
• la normalisation des données nécessaires au stockage ;
• le suivi de la qualité des données ;
• la traçabilité des données importées ;
• le chargement des données dans PostgreSQL/TimescaleDB ;
• la conservation des valeurs manquantes et des informations de qualité ;
• l'idempotence du chargement afin d'éviter la création de doublons.

Données sources

Dataset historique

Le dataset est fourni par le formateur dans le cadre du projet EnerVision.

Il contient les deux fichiers suivants :

all_sites_combined.csv
dataset_metadata.json

Ces fichiers sont nécessaires une seule fois pour initialiser les données historiques de l'environnement.

Ils ne sont pas versionnés dans Git. Chaque membre de l'équipe récupère manuellement les fichiers fournis par le formateur et les place dans :

data/raw/

Structure locale attendue :

data/
└── raw/
    ├── .gitkeep
    ├── all_sites_combined.csv
    └── dataset_metadata.json

Le fichier .gitkeep est versionné afin de conserver le répertoire data/raw/ dans Git. Les fichiers CSV et JSON sont ignorés par Git.

API Mock

La deuxième source est l'API Mock EnerVision.

Elle permet de récupérer :

• les informations des sites avec GET /api/v1/sites ;
• les mesures simulées avec GET /api/v1/readings.

L'API Mock est utilisée pour compléter les données historiques avec des mesures simulées récupérées sur une période donnée.

Technologies utilisées

Technologie	Utilisation
Python	Développement du pipeline ETL
Pandas	Lecture, validation et transformation du dataset historique
JSON	Lecture des métadonnées et conservation des données sources
HTTPX	Appels HTTP asynchrones vers l'API Mock
hashlib / SHA-256	Identification, intégrité et traçabilité du dataset historique
SQLAlchemy Async	Connexion et chargement asynchrone en base
PostgreSQL	Stockage relationnel
TimescaleDB	Stockage des séries temporelles énergétiques
Docker Compose	Exécution de l'environnement local
Alembic	Gestion des migrations du schéma
uv	Gestion et exécution de l'environnement Python
Ruff	Contrôle de la qualité du code
mypy	Vérification du typage
Pytest	Tests automatisés

Import du dataset historique

Fonctionnement du pipeline historique

Le script d'import se trouve dans :

apps/backend/app/etl/historical_import.py

Le flux d'import est le suivant :

CSV + métadonnées JSON
          |
          v
      Extraction
          |
          v
      Validation
          |
          v
   Traçabilité SHA-256
          |
          v
     Transformation
          |
          v
 Chargement par batches
          |
          v
PostgreSQL / TimescaleDB

1. Extraction

Le pipeline charge :

• all_sites_combined.csv avec Pandas ;
• dataset_metadata.json avec le module JSON de Python.

2. Validation

Avant toute écriture en base, le pipeline contrôle notamment :

• la présence des colonnes obligatoires ;
• le nombre de lignes ;
• la cohérence des identifiants des sites ;
• la cohérence des informations associées aux sites ;
• les doublons sur le couple (site_id, timestamp) ;
• les timestamps ;
• les valeurs manquantes.

Une incohérence détectée pendant cette étape interrompt l'import avant le chargement.

3. Dry-run

Un mode --dry-run permet d'exécuter les contrôles sans écrire de données dans PostgreSQL.

Il permet notamment de vérifier :

• le nombre de lignes ;
• le nombre de sites ;
• la période couverte ;
• les doublons ;
• les valeurs NULL ;
• l'empreinte SHA-256.

4. Traçabilité

Une empreinte SHA-256 est calculée à partir du fichier CSV afin d'identifier le dataset utilisé.

Empreinte SHA-256 du dataset validé :

6E3777A97A5660B11855750B9028F70BE72138A11F26795F3A35D9CE74CE0C8D

Cette empreinte participe à la traçabilité du dataset chargé.

5. Transformation

Les timestamps sont normalisés avec la timezone :

UTC

Le pipeline détermine également la qualité des mesures à partir des données disponibles.

Les valeurs manquantes sont conservées pendant cette phase afin de préserver la donnée source.

Aucune imputation n'est réalisée pendant l'ingestion :

imputed_values = NULL
imputation_method = NULL

6. Chargement

Le chargement est réalisé avec SQLAlchemy Async dans PostgreSQL/TimescaleDB.

Les données sont enregistrées dans les tables :

dataset
site
reading

Les mesures sont chargées par batches de :

1000 lignes

Les mesures provenant du dataset CSV utilisent :

source = "csv"
dataset_id = identifiant du dataset

Cette représentation respecte les contraintes définies dans le schéma de la base.

Dataset validé

Le dataset traité contient :

• 122 647 mesures ;
• 7 sites ;
• une période du 01/01/2023 au 31/12/2024 ;
• 0 doublon détecté dans les données sources.

Valeurs manquantes identifiées :

Variable	Nombre de valeurs NULL
consumption_kwh	2 840
consumption_euros	2 487
temperature_celsius	3 416
humidity_percent	3 423
solar_irradiance_wm2	3 964

Exécution historique en dry-run

Depuis le dossier :

apps/backend/

exécuter :

uv run python -m app.etl.historical_import `
  --csv ..\..\data\raw\all_sites_combined.csv `
  --metadata ..\..\data\raw\dataset_metadata.json `
  --source-timezone UTC `
  --dry-run

Aucune donnée n'est écrite dans la base pendant cette exécution.

Chargement historique réel

Depuis apps/backend/ :

uv run python -m app.etl.historical_import `
  --csv ..\..\data\raw\all_sites_combined.csv `
  --metadata ..\..\data\raw\dataset_metadata.json `
  --source-timezone UTC

Le chargement est effectué progressivement par batches.

Exemple :

Chargement : 1000/122647
Chargement : 2000/122647
...
Chargement : 122647/122647

Résultats obtenus pour le dataset historique

Après le chargement initial, les contrôles en base ont confirmé :

datasets = 1
sites    = 7
readings = 122647
source   = csv

Le premier import a créé :

nouvelles lectures : 122647

Idempotence du dataset historique

Le pipeline a été exécuté une deuxième fois avec exactement le même dataset afin de vérifier son idempotence.

Résultat :

lectures avant     : 122647
lectures après     : 122647
nouvelles lectures : 0

Une nouvelle exécution du même import ne crée donc pas de mesures supplémentaires pour le dataset testé.

Vérifications SQL du dataset historique

Depuis la racine du projet, vérifier le nombre d'enregistrements avec :

docker compose exec db psql -U enervision -d enervision -c "SELECT COUNT(*) AS datasets FROM dataset; SELECT COUNT(*) AS sites FROM site; SELECT COUNT(*) AS readings FROM reading;"

Résultat attendu après l'import initial :

datasets = 1
sites    = 7
readings = 122647

Vérifier la source des mesures avec :

docker compose exec db psql -U enervision -d enervision -c "SELECT source, COUNT(*) FROM reading GROUP BY source ORDER BY source;"

Résultat attendu pour le dataset historique :

csv | 122647

Import depuis l'API Mock

Fonctionnement

Le script d'import de l'API Mock se trouve dans :

apps/backend/app/etl/mock_api_import.py

Le flux est le suivant :

        API Mock
           |
     +-----+------+
     |            |
     v            v
   /sites      /readings
     |            |
     +-----+------+
           |
           v
 mock_api_import.py
           |
           v
 Transformation
 + qualité data
           |
           v
PostgreSQL / TimescaleDB
     |          |
     v          v
   site       reading

Le pipeline commence par récupérer les sites avec :

GET /api/v1/sites

Il récupère ensuite les mesures de chaque site avec :

GET /api/v1/readings

Les paramètres envoyés à /api/v1/readings sont :

site_id
start_time
end_time
limit

Le paramètre limit doit être compris entre 1 et 1000.

Configuration de l'API Mock

La connexion à l'API Mock est configurée avec les variables d'environnement suivantes :

APP_MOCK_API_BASE_URL
APP_MOCK_API_USERNAME
APP_MOCK_API_PASSWORD
APP_MOCK_API_TIMEOUT_SECONDS

Les identifiants réels ne sont pas versionnés dans Git.

Les fichiers .env.example indiquent uniquement les variables nécessaires à l'exécution.

Transformation des mesures API

Les mesures provenant de l'API Mock sont enregistrées dans reading avec :

source = "api_history"
dataset_id = NULL

Les mesures provenant de l'API ne sont donc pas rattachées à un dataset historique.

Le timestamp reçu depuis l'API est converti en datetime avec timezone avant le chargement.

La réponse source est conservée dans :

raw_data

afin de préserver la donnée reçue et faciliter la traçabilité.

Qualité des données API

Les valeurs NULL fournies par l'API sont conservées telles quelles.

Une valeur manquante n'est pas transformée en zéro et la mesure n'est pas supprimée.

Le pipeline conserve également :

data_quality
null_reasons

Les niveaux de qualité possibles sont :

good
partial
degraded
critical

Ce sont les quatre seules valeurs que la contrainte ck_reading_quality accepte. Toute autre valeur renvoyée par l'API est remplacée par NULL plutôt que de faire échouer le lot entier.

Aucune imputation n'est réalisée pendant l'ingestion :

imputed_values = NULL
imputation_method = NULL

Cette stratégie permet de distinguer une véritable valeur nulle ou manquante d'une consommation égale à zéro et de conserver les informations liées aux défaillances de capteurs.

Bornes physiques et frontière de confiance

La réponse de l'API Mock est traitée comme une entrée hostile : l'API n'a pas d'authentification et expose un endpoint mutatif à quiconque. Voir API10 dans docs/architecture/owasp-traceabilite.md.

Les plages acceptées sont déclarées dans PHYSICAL_BOUNDS :

Grandeur	Plage acceptée
consumption_kw	0 à 100 000
consumption_kwh	0 à 100 000
voltage_v	0 à 1 000
current_a	0 à 10 000
power_factor	0 à 1
temperature_celsius	-90 à 60
humidity_percent	0 à 100
capacity_kw	0 à 100 000

Une valeur hors plage, d'un type inattendu, NaN ou infinie devient NULL :

null_reasons += "out_of_physical_bounds:<colonne>"
data_quality = "degraded"

L'import ne s'interrompt pas pour autant : le mock émet des anomalies par construction, et raw_data conserve la réponse d'origine.

La taille des réponses est plafonnée : au plus MAX_SITES sites, et au plus --limit mesures par site. Au-delà, l'import échoue au lieu de charger.

Enfin, seuls les champs attendus sont recopiés vers la base. Une clé supplémentaire renvoyée par l'API n'atteint jamais une colonne.

Dry-run de l'API Mock

Le mode --dry-run permet de tester la connexion, la récupération des sites et la récupération des mesures sans écrire dans PostgreSQL.

Depuis apps/backend/ :

uv run python -m app.etl.mock_api_import `
  --start-time "2024-06-15T12:00:00" `
  --end-time "2024-06-15T13:00:00" `
  --limit 60 `
  --dry-run

Chargement réel depuis l'API Mock

Depuis apps/backend/ :

uv run python -m app.etl.mock_api_import `
  --start-time "2024-06-15T12:00:00" `
  --end-time "2024-06-15T13:00:00" `
  --limit 60

Résultat validé pour l'API Mock

Le scénario de validation utilisé couvre la période :

15/06/2024 12:00 UTC
à
15/06/2024 13:00 UTC

avec une limite de 60 lectures par site.

Résultat obtenu :

sites récupérés       : 7
lectures par site     : 60
lectures récupérées   : 420
source                : api_history
dataset_id            : NULL

Les contrôles effectués directement dans PostgreSQL/TimescaleDB ont confirmé :

• l'enregistrement des mesures dans reading ;
• la présence des 7 sites ;
• source = "api_history" ;
• dataset_id = NULL ;
• la conservation des valeurs NULL ;
• la conservation de data_quality ;
• la conservation de null_reasons ;
• la conservation de la donnée source dans raw_data.

Idempotence de l'import API Mock

Le même import a été exécuté plusieurs fois afin de vérifier qu'une mesure déjà présente n'est pas créée une seconde fois.

L'idempotence repose sur la contrainte d'unicité de la table reading et sur la gestion des conflits lors de l'insertion.

Un test d'intégration automatisé vérifie également ce comportement.

Tests et qualité

Les tests automatisés des pipelines ETL sont situés dans :

apps/backend/tests/etl/

Les tests de l'import historique couvrent notamment :

• la validation du dataset ;
• les colonnes obligatoires ;
• la détection des doublons ;
• la cohérence des sites ;
• la normalisation des timestamps ;
• la gestion des valeurs manquantes ;
• la classification de la qualité des données ;
• la construction des mesures destinées à la BDD ;
• le respect des contraintes du modèle de données.

Les tests de l'import API Mock couvrent notamment :

• la récupération des sites ;
• l'appel à /api/v1/readings ;
• les paramètres site_id, start_time, end_time et limit ;
• la gestion des erreurs HTTP ;
• la validation du format de la réponse ;
• la transformation des mesures ;
• la conservation des valeurs NULL ;
• la conservation de data_quality et null_reasons ;
• source = "api_history" ;
• dataset_id = NULL ;
• la conservation de raw_data ;
• l'idempotence du chargement.

Exécuter les tests ETL :

uv run pytest tests\etl -v

Exécuter les tests unitaires de l'import API Mock :

uv run pytest tests\etl\test_mock_api_import.py -v

Exécuter le test d'intégration de l'import API Mock :

uv run pytest tests\etl\test_mock_api_import.py -m integration -v

Contrôler la qualité du code :

uv run ruff check app\etl tests\etl

Contrôler le typage :

uv run mypy app

Exécuter la suite complète avec le seuil de couverture :

uv run pytest --cov-fail-under=85

Lors de la validation de l'import API Mock :

8 tests unitaires passés
1 test d'intégration passé

La suite backend complète a également été validée avec une couverture supérieure au seuil de 85 %.

Suite du pipeline Data

Deux sources de données sont maintenant prises en charge :

Dataset CSV/JSON
      |
      v
historical_import.py
      |
      +-----------------+
                        |
                        v
             PostgreSQL / TimescaleDB
                        ^
                        |
      +-----------------+
      |
mock_api_import.py
      ^
      |
   API Mock

La logique d'extraction, de transformation et de chargement est donc disponible pour les deux sources de données du MVP.

Airflow tourne désormais réellement (etl/airflow/, make airflow-up) et orchestre le pipeline ML (ml_train/ml_score, issue #115) ainsi que la détection d'alertes et la génération des recommandations (alertes, issue #116). Il n'orchestre pas encore ces deux imports : historical_import.py et mock_api_import.py (normalisation et chargement micro-batch, issues #15/#16) restent à faire.

Airflow permet de planifier les traitements, gérer leur ordre d'exécution, suivre leur état et remonter les erreurs. Il ne remplace pas la logique ETL Python existante : les scripts actuels restent responsables de l'extraction, de la validation, de la transformation et du chargement. etl/airflow/dags/ml_train.py, ml_score.py et alertes.py montrent le patron retenu (des BashOperator qui invoquent le script tel quel, dans l'environnement uv que l'image embarque pour lui).

Le pipeline Data servira ensuite à préparer les données nécessaires au modèle de Machine Learning.