From fcbfcc8eb2d5c21ece42a3b60e656d1e9f66a20d Mon Sep 17 00:00:00 2001 From: Meryemel-gham Date: Wed, 16 Sep 2026 12:49:21 +0200 Subject: [PATCH 1/6] feat(data): ajoute l'import historique des donnees --- .gitignore | 3 +- apps/backend/app/etl/__init__.py | 0 apps/backend/app/etl/historical_import.py | 783 ++++++++++++++++++++++ apps/backend/pyproject.toml | 1 + apps/backend/uv.lock | 95 +++ data/raw/.gitkeep | 0 6 files changed, 881 insertions(+), 1 deletion(-) create mode 100644 apps/backend/app/etl/__init__.py create mode 100644 apps/backend/app/etl/historical_import.py create mode 100644 data/raw/.gitkeep diff --git a/.gitignore b/.gitignore index bb3dca3..38ef5cf 100644 --- a/.gitignore +++ b/.gitignore @@ -52,7 +52,8 @@ standalone_admin_password.txt secrets/ # Donnees locales -data/ +data/raw/* +!data/raw/.gitkeep *.sqlite3 monitoring/grafana/data/ monitoring/prometheus/data/ diff --git a/apps/backend/app/etl/__init__.py b/apps/backend/app/etl/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/backend/app/etl/historical_import.py b/apps/backend/app/etl/historical_import.py new file mode 100644 index 0000000..e876c26 --- /dev/null +++ b/apps/backend/app/etl/historical_import.py @@ -0,0 +1,783 @@ +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +from pathlib import Path +from typing import Any + +import pandas as pd +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncConnection, create_async_engine + +from app.core.config import get_settings + +REQUIRED_COLUMNS = { + "timestamp", + "site_id", + "site_type", + "site_name", + "consumption_kwh", + "consumption_euros", + "temperature_celsius", + "humidity_percent", + "solar_irradiance_wm2", + "hour", + "day_of_week", + "day_name", + "month", + "is_weekend", + "is_working_hours", +} + +MEASURE_COLUMNS = [ + "consumption_kwh", + "consumption_euros", + "temperature_celsius", + "humidity_percent", + "solar_irradiance_wm2", +] + +SOURCE_NAME = "historical_csv" + + +def compute_sha256(path: Path) -> str: + """Calcule l'empreinte SHA-256 du fichier source.""" + sha256 = hashlib.sha256() + + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + sha256.update(block) + + return sha256.hexdigest() + + +def load_metadata(path: Path) -> dict[str, Any]: + """Charge les métadonnées fournies avec le dataset.""" + with path.open("r", encoding="utf-8") as source: + return json.load(source) + + +def classify_quality( + row: dict[str, Any], +) -> tuple[str, list[str]]: + """ + Déduit une qualité technique à partir des champs manquants. + + Les valeurs NULL sont conservées. On ne cherche pas ici à + déterminer la cause physique exacte de leur absence. + """ + missing = [ + column + for column in MEASURE_COLUMNS + if pd.isna(row.get(column)) + ] + + if not missing: + quality = "good" + elif len(missing) == len(MEASURE_COLUMNS): + quality = "critical" + elif "consumption_kwh" in missing: + quality = "degraded" + else: + quality = "partial" + + reasons = [ + f"missing:{column}" + for column in missing + ] + + return quality, reasons + + +def validate_source( + frame: pd.DataFrame, + metadata: dict[str, Any], +) -> None: + """Valide le dataset avant tout chargement en base.""" + missing_columns = REQUIRED_COLUMNS.difference( + frame.columns + ) + + if missing_columns: + raise ValueError( + "Colonnes obligatoires absentes : " + f"{sorted(missing_columns)}" + ) + + expected_records = int(metadata["total_records"]) + + if len(frame) != expected_records: + raise ValueError( + "Nombre de lignes inattendu : " + f"{len(frame)} au lieu de " + f"{expected_records}" + ) + + expected_sites = set(metadata["sites"].keys()) + actual_sites = set(frame["site_id"].unique()) + + if actual_sites != expected_sites: + raise ValueError( + "Sites incohérents. " + f"Attendus={sorted(expected_sites)}, " + f"trouvés={sorted(actual_sites)}" + ) + + duplicated = frame.duplicated( + subset=["site_id", "timestamp"] + ).sum() + + if duplicated: + raise ValueError( + f"{duplicated} doublons " + "(site_id, timestamp) détectés" + ) + + static_variants = ( + frame.groupby("site_id")[ + ["site_type", "site_name"] + ] + .nunique() + ) + + if (static_variants > 1).any().any(): + raise ValueError( + "Un site possède plusieurs valeurs " + "de site_type ou site_name." + ) + + # Vérifie également que tous les timestamps + # peuvent être interprétés correctement. + pd.to_datetime( + frame["timestamp"], + errors="raise", + ) + + +def normalize_timestamps( + frame: pd.DataFrame, + source_timezone: str, +) -> pd.DataFrame: + """ + Normalise les timestamps et leur associe une timezone. + + Les timestamps originaux sont conservés dans une colonne + temporaire afin de pouvoir les stocker dans raw_data. + """ + normalized = frame.copy() + + normalized["_source_timestamp"] = ( + normalized["timestamp"] + ) + + timestamps = pd.to_datetime( + normalized["timestamp"], + errors="raise", + ) + + if timestamps.dt.tz is None: + timestamps = timestamps.dt.tz_localize( + source_timezone + ) + else: + timestamps = timestamps.dt.tz_convert( + source_timezone + ) + + normalized["timestamp"] = timestamps + + return normalized + + +def to_json_value(value: Any) -> Any: + """ + Convertit une valeur Pandas/Numpy en valeur + compatible JSON. + """ + if value is None: + return None + + try: + if pd.isna(value): + return None + except (TypeError, ValueError): + pass + + if isinstance(value, pd.Timestamp): + return value.isoformat() + + if hasattr(value, "item"): + return value.item() + + return value + + +async def ensure_dataset( + connection: AsyncConnection, + metadata: dict[str, Any], + sha256: str, + source_timezone: str, + storage_uri: str, +) -> int: + """ + Crée l'entrée dataset si elle n'existe pas. + + Le SHA-256 permet de reconnaître un fichier déjà importé + et participe à l'idempotence et à la traçabilité. + """ + result = await connection.execute( + text( + """ + SELECT dataset_id + FROM dataset + WHERE archive_sha256 = :sha256 + LIMIT 1 + """ + ), + { + "sha256": sha256, + }, + ) + + existing = result.scalar_one_or_none() + + if existing is not None: + return int(existing) + + metadata_summary = { + "generator_version": metadata.get( + "generator_version" + ), + "total_sites": metadata.get( + "total_sites" + ), + "total_records": metadata.get( + "total_records" + ), + "date_range": metadata.get( + "date_range" + ), + "frequency": metadata.get( + "frequency" + ), + "null_injection_enabled": metadata.get( + "null_injection_enabled" + ), + "null_strategies": metadata.get( + "null_strategies" + ), + "importer": "historical_import_v1", + } + + result = await connection.execute( + text( + """ + INSERT INTO dataset ( + dataset_name, + archive_sha256, + storage_uri, + source_timezone, + "metadata" + ) + VALUES ( + :dataset_name, + :archive_sha256, + :storage_uri, + :source_timezone, + CAST(:metadata AS jsonb) + ) + RETURNING dataset_id + """ + ), + { + "dataset_name": ( + "EnerVision historical dataset " + "2023-2024" + ), + "archive_sha256": sha256, + "storage_uri": storage_uri, + "source_timezone": source_timezone, + "metadata": json.dumps( + metadata_summary, + ensure_ascii=False, + ), + }, + ) + + return int(result.scalar_one()) + + +async def upsert_sites( + connection: AsyncConnection, + frame: pd.DataFrame, +) -> None: + """Insère ou met à jour les sites du dataset.""" + sites = ( + frame[ + [ + "site_id", + "site_type", + "site_name", + ] + ] + .drop_duplicates( + subset=["site_id"] + ) + .to_dict( + orient="records" + ) + ) + + await connection.execute( + text( + """ + INSERT INTO site ( + site_id, + site_type, + site_name + ) + VALUES ( + :site_id, + :site_type, + :site_name + ) + ON CONFLICT (site_id) + DO UPDATE SET + site_type = EXCLUDED.site_type, + site_name = EXCLUDED.site_name + """ + ), + sites, + ) + + +def build_reading_batch( + chunk: pd.DataFrame, + dataset_id: int, +) -> list[dict[str, Any]]: + """ + Transforme un chunk Pandas en lignes prêtes + à être chargées dans la table reading. + """ + rows: list[dict[str, Any]] = [] + + for record in chunk.to_dict( + orient="records" + ): + quality, reasons = classify_quality( + record + ) + + raw_data = { + column: to_json_value(value) + for column, value in record.items() + if column != "_source_timestamp" + } + + # Dans raw_data, on conserve le timestamp + # exactement tel qu'il était dans le CSV. + raw_data["timestamp"] = to_json_value( + record["_source_timestamp"] + ) + + rows.append( + { + "site_id": record["site_id"], + "timestamp": record["timestamp"], + "source": SOURCE_NAME, + "dataset_id": dataset_id, + + # Non fourni par le dataset historique. + "consumption_kw": None, + + "consumption_kwh": to_json_value( + record["consumption_kwh"] + ), + "consumption_euros": to_json_value( + record["consumption_euros"] + ), + + # Non fournis par le CSV historique. + "voltage_v": None, + "current_a": None, + "power_factor": None, + + "temperature_celsius": ( + to_json_value( + record[ + "temperature_celsius" + ] + ) + ), + "humidity_percent": ( + to_json_value( + record[ + "humidity_percent" + ] + ) + ), + "solar_irradiance_wm2": ( + to_json_value( + record[ + "solar_irradiance_wm2" + ] + ) + ), + + "is_working_hours": bool( + record[ + "is_working_hours" + ] + ), + + "data_quality": quality, + "null_reasons": reasons, + + # Aucune imputation pendant + # l'ingestion RAW. + "imputed_values": json.dumps( + {} + ), + "imputation_method": None, + + # Conservation de la donnée source + # pour la traçabilité. + "raw_data": json.dumps( + raw_data, + ensure_ascii=False, + ), + } + ) + + return rows + + +READING_INSERT = text( + """ + INSERT INTO reading ( + site_id, + timestamp, + source, + dataset_id, + consumption_kw, + consumption_kwh, + consumption_euros, + voltage_v, + current_a, + power_factor, + temperature_celsius, + humidity_percent, + solar_irradiance_wm2, + is_working_hours, + data_quality, + null_reasons, + imputed_values, + imputation_method, + raw_data + ) + VALUES ( + :site_id, + :timestamp, + :source, + :dataset_id, + :consumption_kw, + :consumption_kwh, + :consumption_euros, + :voltage_v, + :current_a, + :power_factor, + :temperature_celsius, + :humidity_percent, + :solar_irradiance_wm2, + :is_working_hours, + :data_quality, + :null_reasons, + CAST(:imputed_values AS jsonb), + :imputation_method, + CAST(:raw_data AS jsonb) + ) + ON CONFLICT DO NOTHING + """ +) + + +async def import_historical( + csv_path: Path, + metadata_path: Path, + source_timezone: str, + batch_size: int, + dry_run: bool, + storage_uri: str, +) -> None: + """ + Exécute le pipeline ETL historique EnerVision. + + Étapes : + 1. Extract + 2. Validate + 3. Transform + 4. Load + """ + metadata = load_metadata( + metadata_path + ) + + frame = pd.read_csv( + csv_path + ) + + validate_source( + frame, + metadata, + ) + + print( + f"Lignes : {len(frame)}" + ) + print( + "Sites : " + f"{frame['site_id'].nunique()}" + ) + print( + "Période : " + f"{frame['timestamp'].min()} -> " + f"{frame['timestamp'].max()}" + ) + print( + "Doublons : " + f"{frame.duplicated(['site_id', 'timestamp']).sum()}" + ) + + print("\nValeurs NULL :") + print( + frame[ + MEASURE_COLUMNS + ].isna().sum() + ) + + sha256 = compute_sha256( + csv_path + ) + + print( + f"\nSHA-256 : {sha256}" + ) + + if dry_run: + print( + "\nDry-run terminé : " + "aucune donnée écrite." + ) + return + + normalized = normalize_timestamps( + frame, + source_timezone, + ) + + settings = get_settings() + + engine = create_async_engine( + str(settings.database_url), + pool_pre_ping=True, + ) + + try: + async with engine.begin() as connection: + dataset_id = await ensure_dataset( + connection=connection, + metadata=metadata, + sha256=sha256, + source_timezone=source_timezone, + storage_uri=storage_uri, + ) + + await upsert_sites( + connection, + normalized, + ) + + result = await connection.execute( + text( + """ + SELECT COUNT(*) + FROM reading + WHERE dataset_id = :dataset_id + AND source = :source + """ + ), + { + "dataset_id": dataset_id, + "source": SOURCE_NAME, + }, + ) + + before = int( + result.scalar_one() + ) + + for start in range( + 0, + len(normalized), + batch_size, + ): + chunk = normalized.iloc[ + start : start + batch_size + ] + + rows = build_reading_batch( + chunk, + dataset_id, + ) + + await connection.execute( + READING_INSERT, + rows, + ) + + loaded = min( + start + batch_size, + len(normalized), + ) + + print( + "Chargement : " + f"{loaded}/" + f"{len(normalized)}" + ) + + result = await connection.execute( + text( + """ + SELECT COUNT(*) + FROM reading + WHERE dataset_id = :dataset_id + AND source = :source + """ + ), + { + "dataset_id": dataset_id, + "source": SOURCE_NAME, + }, + ) + + after = int( + result.scalar_one() + ) + + print( + "\nImport terminé." + ) + print( + "dataset_id : " + f"{dataset_id}" + ) + print( + "lectures avant : " + f"{before}" + ) + print( + "lectures après : " + f"{after}" + ) + print( + "nouvelles lectures : " + f"{after - before}" + ) + + finally: + await engine.dispose() + + +def parse_args() -> argparse.Namespace: + """Définit les arguments CLI de l'import.""" + parser = argparse.ArgumentParser( + description=( + "Import historique EnerVision" + ) + ) + + parser.add_argument( + "--csv", + type=Path, + required=True, + help="Chemin vers le CSV historique.", + ) + + parser.add_argument( + "--metadata", + type=Path, + required=True, + help=( + "Chemin vers le fichier " + "dataset_metadata.json." + ), + ) + + parser.add_argument( + "--source-timezone", + default="UTC", + help=( + "Timezone associée aux timestamps " + "du dataset. Défaut : UTC." + ), + ) + + parser.add_argument( + "--batch-size", + type=int, + default=1000, + help=( + "Nombre de lignes insérées " + "par batch. Défaut : 1000." + ), + ) + + parser.add_argument( + "--dry-run", + action="store_true", + help=( + "Valide les données sans " + "écrire en base." + ), + ) + + return parser.parse_args() + + +def main() -> None: + """Point d'entrée CLI du pipeline.""" + args = parse_args() + + if args.batch_size <= 0: + raise ValueError( + "--batch-size doit être " + "strictement supérieur à 0." + ) + + # resolve() est volontairement exécuté ici, + # dans la partie synchrone du programme. + # Cela évite une opération filesystem bloquante + # à l'intérieur d'une fonction async. + storage_uri = ( + args.csv.resolve().as_uri() + ) + + asyncio.run( + import_historical( + csv_path=args.csv, + metadata_path=args.metadata, + source_timezone=( + args.source_timezone + ), + batch_size=args.batch_size, + dry_run=args.dry_run, + storage_uri=storage_uri, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/apps/backend/pyproject.toml b/apps/backend/pyproject.toml index 18bf979..6cf42a5 100644 --- a/apps/backend/pyproject.toml +++ b/apps/backend/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "pyjwt>=2.10", "argon2-cffi>=23.1", "anyio>=4.0", + "pandas>=3.0.5", ] [dependency-groups] diff --git a/apps/backend/uv.lock b/apps/backend/uv.lock index 7c2b8f4..fb43797 100644 --- a/apps/backend/uv.lock +++ b/apps/backend/uv.lock @@ -1,6 +1,11 @@ version = 1 revision = 3 requires-python = "==3.14.*" +resolution-markers = [ + "sys_platform == 'win32'", + "sys_platform == 'emscripten'", + "sys_platform != 'emscripten' and sys_platform != 'win32'", +] [[package]] name = "alembic" @@ -311,6 +316,7 @@ dependencies = [ { name = "argon2-cffi" }, { name = "asyncpg" }, { name = "fastapi" }, + { name = "pandas" }, { name = "prometheus-fastapi-instrumentator" }, { name = "pydantic", extra = ["email"] }, { name = "pydantic-settings" }, @@ -337,6 +343,7 @@ requires-dist = [ { name = "argon2-cffi", specifier = ">=23.1" }, { name = "asyncpg", specifier = ">=0.31.0" }, { name = "fastapi", specifier = ">=0.141.1" }, + { name = "pandas", specifier = ">=3.0.5" }, { name = "prometheus-fastapi-instrumentator", specifier = ">=8.1.0" }, { name = "pydantic", extras = ["email"], specifier = ">=2.13.5" }, { name = "pydantic-settings", specifier = ">=2.15.0" }, @@ -595,6 +602,35 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/79/7b/2c79738432f5c924bef5071f933bcc9efd0473bac3b4aa584a6f7c1c8df8/mypy_extensions-1.1.0-py3-none-any.whl", hash = "sha256:1be4cccdb0f2482337c4743e60421de3a356cd97508abadd57d47403e94f5505", size = 4963, upload-time = "2025-04-22T14:54:22.983Z" }, ] +[[package]] +name = "numpy" +version = "2.5.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/13/01/11703282db468b85f6f7b8c7f22d058de5970d5c7e60a3a8aaa313c3de36/numpy-2.5.3.tar.gz", hash = "sha256:df2d5874ff183595a4ba404edd04f6bd9b5505c1d7708573f6a6c17489a67563", size = 20791231, upload-time = "2026-09-06T16:27:47.073Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/70/78/cf416f15dc29375a229d9dfebf8db6e313f291580b39fa1a568b6052bb07/numpy-2.5.3-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:350ba9783ce969cf9f7ce6e6a9a58e1a6e2a19ca025b7ee448c4db727706212a", size = 16998686, upload-time = "2026-09-06T16:25:33.171Z" }, + { url = "https://files.pythonhosted.org/packages/9e/59/abcc2d8def4fd60eec7d87f92d27c13448ffd9ab14339bcc63a0d7a2fdea/numpy-2.5.3-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:012e66aca395d795496446e52aeeb5866312a5d4d3f27da270e5a0b43f70dc5c", size = 12013862, upload-time = "2026-09-06T16:25:36.748Z" }, + { url = "https://files.pythonhosted.org/packages/94/75/4640d2d6e4b64a049e48425a82728a41ef4adb61332d2cba68055774878b/numpy-2.5.3-cp314-cp314-macosx_14_0_arm64.whl", hash = "sha256:adc1ada2662f8a5f960b8a10d9986897e7499ef07e06d4cfe7197f8cce923c07", size = 5449793, upload-time = "2026-09-06T16:25:39.476Z" }, + { url = "https://files.pythonhosted.org/packages/96/cd/625b57ae33d4ca560f32cc0b47b4a5922146d9beb998ddf773900d440a73/numpy-2.5.3-cp314-cp314-macosx_14_0_x86_64.whl", hash = "sha256:54a115e5a73b8fc44f0cebef486365a1894b5c9760685d4558b72b7c3eb846e0", size = 6785176, upload-time = "2026-09-06T16:25:42.069Z" }, + { url = "https://files.pythonhosted.org/packages/9c/72/12918652e7912ef9751e8694c88820fcd1908e0618cb23f5f3caa6004b7b/numpy-2.5.3-cp314-cp314-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:be5a8381859b6da607c84f4f7d6847725f1cf1853ef8a2c9e115b7d58bef47dc", size = 15703377, upload-time = "2026-09-06T16:25:45.135Z" }, + { url = "https://files.pythonhosted.org/packages/45/8f/9beacf79ca7c650688ad0baa80931adb988fe6e6e5d5903c23cc3dbd70eb/numpy-2.5.3-cp314-cp314-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b0521d0f4aebb6e06189451025fa17a913287b13c03d5fe05c017333b654ea5b", size = 16711928, upload-time = "2026-09-06T16:25:48.461Z" }, + { url = "https://files.pythonhosted.org/packages/09/8d/41d0a56e1ac4c87495c897a211b1368691b7237aadabec8b3b8f3a74d48f/numpy-2.5.3-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:9deb49575e5b0b94ed72c8a64ec4d033381adc27e9060ae842971f697ba96104", size = 17059507, upload-time = "2026-09-06T16:25:51.873Z" }, + { url = "https://files.pythonhosted.org/packages/08/1e/0dfbc5cc251d54e2af790f254d24ec38637fa97ec7d5d11de7ffed787098/numpy-2.5.3-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:b00eefbcf0f292945c4b4dec2ae845389ef5bcdcd596e6e4328051db5b5ba694", size = 18471002, upload-time = "2026-09-06T16:25:55.233Z" }, + { url = "https://files.pythonhosted.org/packages/b5/2c/dfa40f6991f8185c8c30ffd023dfcbb11888e823cfab9557b920f3bb7bed/numpy-2.5.3-cp314-cp314-win32.whl", hash = "sha256:c2381f82999704f818e2c987a865050e285ec3621262c66d40f5a96c8f899f8e", size = 6180485, upload-time = "2026-09-06T16:25:58.157Z" }, + { url = "https://files.pythonhosted.org/packages/a4/73/d2c08231e4fde7e415501fd02c715d96e98599b2d8384445933944152984/numpy-2.5.3-cp314-cp314-win_amd64.whl", hash = "sha256:2c25dfa72943e4336ddb6b0ee4277b47a0c85bede0807530ec68103bf58e2c10", size = 12698179, upload-time = "2026-09-06T16:26:00.789Z" }, + { url = "https://files.pythonhosted.org/packages/5c/e9/dcdcc9b95cf5f49815055573aee1b11cfbf5299f38a180e437ded050810f/numpy-2.5.3-cp314-cp314-win_arm64.whl", hash = "sha256:15aa985ac73a8db02db7663381aa109510449d3819d37206caed27b33a65a8a6", size = 10769383, upload-time = "2026-09-06T16:26:04.011Z" }, + { url = "https://files.pythonhosted.org/packages/49/c4/af8bc08a7ef4e1529a7c0cf24969accce316b783999802089a581ec99272/numpy-2.5.3-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:ac7bb1c52d445bd4f8f7f97fefe6abc3a084dc4d63df50d79b17fa2b78e89297", size = 12132668, upload-time = "2026-09-06T16:26:07.138Z" }, + { url = "https://files.pythonhosted.org/packages/c5/ae/0f15eb56d4ec5e13c1f7ff04ff407f997d1acbadb45d3e1f2e2645a8f43c/numpy-2.5.3-cp314-cp314t-macosx_14_0_arm64.whl", hash = "sha256:e6ab667ba76450084eb64013762c438ea76d9d29cc676dcd6c2e9892ba37f841", size = 5568580, upload-time = "2026-09-06T16:26:09.828Z" }, + { url = "https://files.pythonhosted.org/packages/23/fb/c72a8f25d4b6e96c354e7ab45ace3b27dc11e5d6a13b6c7d0cd6b08bf112/numpy-2.5.3-cp314-cp314t-macosx_14_0_x86_64.whl", hash = "sha256:f7fabeb6cea87d65f3b926de33d03fb016cfdc29314c90974383b5582ae72891", size = 6882634, upload-time = "2026-09-06T16:26:12.524Z" }, + { url = "https://files.pythonhosted.org/packages/07/a9/968c90ed2ab15060c338e8137f1215b5a60756ae07328e0a60d1c6734df4/numpy-2.5.3-cp314-cp314t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1fb6f8fb9ff0b3a69f52c66ce397b0246583e9f28616231b0e32ca49259a5fa6", size = 15748923, upload-time = "2026-09-06T16:26:15.092Z" }, + { url = "https://files.pythonhosted.org/packages/59/08/9df04103947b95e3b6b1f2ed1a70521f325647a31b82da6a2aae3a485508/numpy-2.5.3-cp314-cp314t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:93e1f5447e2b1e479d7bd74701e84746b86450cff1fc368b132d195e2b8f8211", size = 16746748, upload-time = "2026-09-06T16:26:18.43Z" }, + { url = "https://files.pythonhosted.org/packages/41/a0/14c8d5fe5b53a334aabb653deb391c0fef49558f491880ea300ed6785224/numpy-2.5.3-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:c00abe94c1a69d75d827dcf1c025b25c8a45d230b3bcd77a9020883a1b047653", size = 17111561, upload-time = "2026-09-06T16:26:22.113Z" }, + { url = "https://files.pythonhosted.org/packages/c4/a6/d7e96e42f01522e154c32489640f16dfc4f6181d165d05fc3bec8c2c4999/numpy-2.5.3-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:536f963710a4e63934d80ac0dc4f478804a83e9a84b6828018f25d09953ada33", size = 18513945, upload-time = "2026-09-06T16:26:25.401Z" }, + { url = "https://files.pythonhosted.org/packages/25/39/3453afb7119d0449ef11c886874120ff180e2c337760e0e2d88f70f1a945/numpy-2.5.3-cp314-cp314t-win32.whl", hash = "sha256:4c8a6d2ebce6305fd82fbefca827775437147052a976ee7c94b36a0c1b52ac6c", size = 6335421, upload-time = "2026-09-06T16:26:28.175Z" }, + { url = "https://files.pythonhosted.org/packages/99/01/22815d2b19a1a746b1d45205cffebb3fe511a18acb75fba6c88491fc9894/numpy-2.5.3-cp314-cp314t-win_amd64.whl", hash = "sha256:9a37475425b431b4d060f23b4f52cd2f3aef6bc7c654bd760adf0040eec9d435", size = 12896420, upload-time = "2026-09-06T16:26:31.265Z" }, + { url = "https://files.pythonhosted.org/packages/fa/ee/a7cbba67eeaff038dc29ca8b98a88396c8b0cc9c89d4924f4a27a5c9150b/numpy-2.5.3-cp314-cp314t-win_arm64.whl", hash = "sha256:2d8240cb4c16fd831074aa2b2cf9fc54664d826341d61c372245b96a74a49a9a", size = 10857177, upload-time = "2026-09-06T16:26:34.167Z" }, +] + [[package]] name = "packaging" version = "26.3" @@ -604,6 +640,35 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956, upload-time = "2026-08-04T18:15:27.159Z" }, ] +[[package]] +name = "pandas" +version = "3.0.5" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy" }, + { name = "python-dateutil" }, + { name = "tzdata", marker = "sys_platform == 'emscripten' or sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/be/4f/5f3422a2afec5ffc46308b79e53291365a93748b498ac2e58bead0197916/pandas-3.0.5.tar.gz", hash = "sha256:dca3734d6ab7c906e6730f0788b0a1dbb9f2467731f9711f77995c8e9d62d712", size = 4658219, upload-time = "2026-07-22T22:19:28.819Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/51/2f/cf6aae281264f4463f0875bcbb15fd2bb6d291cc535187dad1732475e4a9/pandas-3.0.5-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:2f264fc46911cc8131a7322a16199bbf8e353d27c10bb211f5bd0c814324dc36", size = 10390034, upload-time = "2026-07-22T22:18:49.818Z" }, + { url = "https://files.pythonhosted.org/packages/06/ec/5189518c7a7659c4bdcc6b1eb32c46c6f3c86b0661ffd84143d1112c7732/pandas-3.0.5-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:53730687fcd161883b24e10411c06d6a4c0f2275d2faf3bb2bc25deb4ba8007c", size = 9980065, upload-time = "2026-07-22T22:18:52.249Z" }, + { url = "https://files.pythonhosted.org/packages/ea/f1/598503ce8d7e3c35601e0747ba288c7864baae66380725bc12f13f884dfe/pandas-3.0.5-cp314-cp314-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:960d3ebcf249f75206899fcd2c6de53f736b7265759ced0d3e559df0b8b709b0", size = 10545532, upload-time = "2026-07-22T22:18:54.813Z" }, + { url = "https://files.pythonhosted.org/packages/fa/de/ceae2adf7034e07e9910299fe412e1819c4f0dd520700a888bcb03625448/pandas-3.0.5-cp314-cp314-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9e94c2c5ca43bd3ca32bf64d32308887b65e5f9bfd8023ea52755107a999f93b", size = 10963120, upload-time = "2026-07-22T22:18:57.42Z" }, + { url = "https://files.pythonhosted.org/packages/66/25/86e0f4451874eb79e688deeebe3c451fec4557f8952005818d800ee8ac7e/pandas-3.0.5-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:e819dd5f62966b481a8cb649d3299ebd886a1ea91ed5a99bf7ce77c98d18ab94", size = 11563178, upload-time = "2026-07-22T22:18:59.729Z" }, + { url = "https://files.pythonhosted.org/packages/f3/45/8643daa3b4147e433adfcccefdd0380d3aad79d86b15d8999730fe1944d5/pandas-3.0.5-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:3c5ed2e7c06e91d340dfd091d7934f9bc82e4a36b95f647f090b9d1c9ac649da", size = 12028708, upload-time = "2026-07-22T22:19:02.164Z" }, + { url = "https://files.pythonhosted.org/packages/96/58/ad979ae617615576e8aafd569c9d4b62f1191d896e38f51d66ba06f3b89a/pandas-3.0.5-cp314-cp314-win_amd64.whl", hash = "sha256:cd8f7c6dc98527058ee6264219343f5392240a6f1bfa654fc5d79023020d0c92", size = 9951806, upload-time = "2026-07-22T22:19:04.596Z" }, + { url = "https://files.pythonhosted.org/packages/69/32/7ac03886b304049a9d2625ee88f59af760d8a93bd30ed9239bce7b9869a8/pandas-3.0.5-cp314-cp314-win_arm64.whl", hash = "sha256:5183427f5a8156d480f30333777bc978be93650a49a7c01db26adffe95b31e85", size = 9238297, upload-time = "2026-07-22T22:19:06.836Z" }, + { url = "https://files.pythonhosted.org/packages/be/ed/1d1f2ee5547d5167face2376d11c8b2a4c7bfff5a416ee7a9046891fab1e/pandas-3.0.5-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:303da736987d481074ca720ada325f8bd80c64ebc2d45ed79b29df3aaa4a26ca", size = 10849690, upload-time = "2026-07-22T22:19:09.391Z" }, + { url = "https://files.pythonhosted.org/packages/57/55/17e17152e98fbb0c4b1e562bc65387a2f20a80db0f4a86bf8d3a0e4248d4/pandas-3.0.5-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:3b2801bbb049d0136f6c213eae02b5fca969384fc2064dd728d8620552aa49da", size = 10509945, upload-time = "2026-07-22T22:19:11.773Z" }, + { url = "https://files.pythonhosted.org/packages/88/90/817d44dbf83facf9556f33576d9af0a241981e7bb5c00606c0bcb5df8dda/pandas-3.0.5-cp314-cp314t-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:cce3a9d11d2b1f82c69a27ec1f4948a170e2c403c4bbfa8cca62e3fdebe2ef3a", size = 10392197, upload-time = "2026-07-22T22:19:14.024Z" }, + { url = "https://files.pythonhosted.org/packages/f1/da/889f00c0a6f5aa1545add70abbf01502dff87ab577adb855bd631c54d2f2/pandas-3.0.5-cp314-cp314t-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ef01af4d8dc6cd2c8d6c7736f149574ef93fe043811eeb5e445f2647154b5040", size = 10862726, upload-time = "2026-07-22T22:19:16.351Z" }, + { url = "https://files.pythonhosted.org/packages/bc/98/f1e934fb3c98fce859c6147c6785816c7b5b9ab7821115c5d8c4de9842b9/pandas-3.0.5-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:e2759e890db96dfcffdbd9b86c3c2cb6afaf58def482820317e06163ec1066cd", size = 11414864, upload-time = "2026-07-22T22:19:18.981Z" }, + { url = "https://files.pythonhosted.org/packages/fe/be/d448af7d657d82e1888dd8551f79c6d6fb161080b5b9752d84d910ec2319/pandas-3.0.5-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:b58b1b39d46a5862e3fb18f50d1a201398619d16a0f9f73f57eea5583cf0e63c", size = 11925105, upload-time = "2026-07-22T22:19:21.515Z" }, + { url = "https://files.pythonhosted.org/packages/29/c1/ccb4238212c8c4f496c584f3044d94e0c030ed8e1d68999db46c91c2242f/pandas-3.0.5-cp314-cp314t-win_amd64.whl", hash = "sha256:1c10461f6eeb35d8f05b6184c65c8b9991663b66c46b1d559b682cb34ae7c6ea", size = 10387612, upload-time = "2026-07-22T22:19:24.257Z" }, + { url = "https://files.pythonhosted.org/packages/d2/cf/6a51b2c38980e04c279fd2fa908a1b0982064e860444acfca4ec2e2c8359/pandas-3.0.5-cp314-cp314t-win_arm64.whl", hash = "sha256:3c5015fd1730fbf883647e88068176c839c102cea883ba1769a6f4593bfc1f8c", size = 9509776, upload-time = "2026-07-22T22:19:26.694Z" }, +] + [[package]] name = "pathspec" version = "1.1.1" @@ -788,6 +853,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9d/7a/d968e294073affff457b041c2be9868a40c1c71f4a35fcc1e45e5493067b/pytest_cov-7.1.0-py3-none-any.whl", hash = "sha256:a0461110b7865f9a271aa1b51e516c9a95de9d696734a2f71e3e78f46e1d4678", size = 22876, upload-time = "2026-03-21T20:11:14.438Z" }, ] +[[package]] +name = "python-dateutil" +version = "2.9.0.post0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "six" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, +] + [[package]] name = "python-dotenv" version = "1.2.3" @@ -857,6 +934,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/8b/4b/51327018d056f0dad2c2238f26d1fb0f53707a9d91b75dea6d1b3039f136/ruff-0.16.7-py3-none-win_arm64.whl", hash = "sha256:aab7f39e2c9df6c596216070f98eef1207b94f8516cca20c808826974971855b", size = 10412401, upload-time = "2026-09-10T18:04:04.098Z" }, ] +[[package]] +name = "six" +version = "1.17.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" }, +] + [[package]] name = "sqlalchemy" version = "2.0.52" @@ -916,6 +1002,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/67/81/4add07e5172b7ac40d8ed5ff580409a7801a4fe26d529bdd915401dabfbe/typing_inspection-0.4.4-py3-none-any.whl", hash = "sha256:65b8397ba37ccbce054456aaccddfc91e6e3083c92824df348d96ca832f3f147", size = 14750, upload-time = "2026-08-12T12:37:24.648Z" }, ] +[[package]] +name = "tzdata" +version = "2026.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e4/31/3d74fa778a63b98b7374323befcc0be5ab3bd94afd4096a0124e7379152c/tzdata-2026.4.tar.gz", hash = "sha256:f1b8bd365d8d210c55353f4d7f8d6d8561c0ba50d704b700d195a9424bba0d79", size = 199350, upload-time = "2026-09-12T12:56:03.251Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f9/bc/8737e8d54cf51106118039b83f485a4783112fab49ea9d044b234978a46e/tzdata-2026.4-py2.py3-none-any.whl", hash = "sha256:c2169a8b0a7a5e9674da5a135ccdfb2b3e671b333ed9fed17b41f73c34476e81", size = 347494, upload-time = "2026-09-12T12:56:01.67Z" }, +] + [[package]] name = "uvicorn" version = "0.53.0" diff --git a/data/raw/.gitkeep b/data/raw/.gitkeep new file mode 100644 index 0000000..e69de29 From b2d52823bae33df93b2808f33556a36b4baad663 Mon Sep 17 00:00:00 2001 From: Meryemel-gham Date: Wed, 16 Sep 2026 14:16:22 +0200 Subject: [PATCH 2/6] fix(data): aligne l'import historique avec les contraintes BDD --- apps/backend/app/etl/historical_import.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/apps/backend/app/etl/historical_import.py b/apps/backend/app/etl/historical_import.py index e876c26..c3b9a70 100644 --- a/apps/backend/app/etl/historical_import.py +++ b/apps/backend/app/etl/historical_import.py @@ -39,7 +39,7 @@ MEASURE_COLUMNS = [ "solar_irradiance_wm2", ] -SOURCE_NAME = "historical_csv" +SOURCE_NAME = "csv" def compute_sha256(path: Path) -> str: @@ -435,11 +435,10 @@ def build_reading_batch( "data_quality": quality, "null_reasons": reasons, - # Aucune imputation pendant - # l'ingestion RAW. - "imputed_values": json.dumps( - {} - ), + # Aucune imputation pendant l'ingestion RAW. + # Les valeurs manquantes sont conservées telles quelles + # afin de préserver la donnée source. + "imputed_values": None, "imputation_method": None, # Conservation de la donnée source From ebb72fb39996cebaafa53e4c727e6ec25257d054 Mon Sep 17 00:00:00 2001 From: Meryemel-gham Date: Wed, 16 Sep 2026 14:29:22 +0200 Subject: [PATCH 3/6] test(data): couvre l'import historique --- .../tests/etl/test_historical_import.py | 244 ++++++++++++++++++ 1 file changed, 244 insertions(+) create mode 100644 apps/backend/tests/etl/test_historical_import.py diff --git a/apps/backend/tests/etl/test_historical_import.py b/apps/backend/tests/etl/test_historical_import.py new file mode 100644 index 0000000..f311b2d --- /dev/null +++ b/apps/backend/tests/etl/test_historical_import.py @@ -0,0 +1,244 @@ +import hashlib +import json + +import pandas as pd +import pytest + +from app.etl.historical_import import ( + SOURCE_NAME, + build_reading_batch, + classify_quality, + compute_sha256, + load_metadata, + normalize_timestamps, + validate_source, +) + + +def make_metadata() -> dict: + return { + "total_records": 2, + "sites": { + "SITE001": {}, + }, + } + + +def make_dataframe() -> pd.DataFrame: + return pd.DataFrame( + [ + { + "timestamp": "2023-01-01 00:00:00", + "site_id": "SITE001", + "site_type": "office", + "site_name": "Site 1", + "consumption_kwh": 10.5, + "consumption_euros": 2.5, + "temperature_celsius": 20.0, + "humidity_percent": 50.0, + "solar_irradiance_wm2": 0.0, + "hour": 0, + "day_of_week": 6, + "day_name": "Sunday", + "month": 1, + "is_weekend": True, + "is_working_hours": False, + }, + { + "timestamp": "2023-01-01 01:00:00", + "site_id": "SITE001", + "site_type": "office", + "site_name": "Site 1", + "consumption_kwh": 11.0, + "consumption_euros": 2.7, + "temperature_celsius": 19.5, + "humidity_percent": 52.0, + "solar_irradiance_wm2": 0.0, + "hour": 1, + "day_of_week": 6, + "day_name": "Sunday", + "month": 1, + "is_weekend": True, + "is_working_hours": False, + }, + ] + ) + + +def test_compute_sha256(tmp_path): + file_path = tmp_path / "dataset.csv" + content = b"hello-enervision" + + file_path.write_bytes(content) + + expected = hashlib.sha256(content).hexdigest() + + assert compute_sha256(file_path) == expected + + +def test_load_metadata(tmp_path): + metadata_path = tmp_path / "metadata.json" + + metadata = { + "total_records": 2, + "sites": { + "SITE001": {}, + }, + } + + metadata_path.write_text( + json.dumps(metadata), + encoding="utf-8", + ) + + assert load_metadata(metadata_path) == metadata + + +def test_validate_source_accepts_valid_dataset(): + frame = make_dataframe() + + validate_source( + frame, + make_metadata(), + ) + + +def test_validate_source_rejects_missing_column(): + frame = make_dataframe().drop( + columns=["consumption_kwh"] + ) + + with pytest.raises( + ValueError, + match="Colonnes obligatoires absentes", + ): + validate_source( + frame, + make_metadata(), + ) + + +def test_validate_source_rejects_duplicates(): + frame = make_dataframe() + + frame.loc[1, "timestamp"] = frame.loc[ + 0, + "timestamp", + ] + + with pytest.raises( + ValueError, + match="doublons", + ): + validate_source( + frame, + make_metadata(), + ) + + +def test_validate_source_rejects_unknown_site(): + frame = make_dataframe() + + frame.loc[1, "site_id"] = "SITE999" + + with pytest.raises( + ValueError, + match="Sites incohérents", + ): + validate_source( + frame, + make_metadata(), + ) + + +def test_normalize_timestamps_adds_timezone(): + frame = make_dataframe() + + normalized = normalize_timestamps( + frame, + "UTC", + ) + + assert normalized["timestamp"].dt.tz is not None + + assert "_source_timestamp" in normalized.columns + + +def test_classify_quality_good(): + row = make_dataframe().iloc[0].to_dict() + + quality, reasons = classify_quality(row) + + assert quality == "good" + assert reasons == [] + + +def test_classify_quality_degraded_when_consumption_missing(): + row = make_dataframe().iloc[0].to_dict() + row["consumption_kwh"] = None + + quality, reasons = classify_quality(row) + + assert quality == "degraded" + + assert "missing:consumption_kwh" in reasons + + +def test_build_reading_batch_respects_database_contract(): + frame = normalize_timestamps( + make_dataframe(), + "UTC", + ) + + rows = build_reading_batch( + frame.iloc[:1], + dataset_id=3, + ) + + assert len(rows) == 1 + + row = rows[0] + + assert row["dataset_id"] == 3 + + # Important : + # contrainte ck_reading_dataset_source. + assert row["source"] == "csv" + assert SOURCE_NAME == "csv" + + # Important : + # contrainte ck_reading_imputation. + assert row["imputed_values"] is None + assert row["imputation_method"] is None + + assert row["data_quality"] == "good" + assert row["null_reasons"] == [] + + +def test_build_reading_batch_keeps_missing_values(): + frame = make_dataframe() + + frame.loc[0, "temperature_celsius"] = None + + frame = normalize_timestamps( + frame, + "UTC", + ) + + rows = build_reading_batch( + frame.iloc[:1], + dataset_id=3, + ) + + row = rows[0] + + assert row["temperature_celsius"] is None + + assert ( + "missing:temperature_celsius" + in row["null_reasons"] + ) + + # RAW ingestion : aucune imputation. + assert row["imputed_values"] is None + assert row["imputation_method"] is None From 74ac1b45778c27935d936a51d9460d15932874f3 Mon Sep 17 00:00:00 2001 From: Meryemel-gham Date: Thu, 17 Sep 2026 09:55:27 +0200 Subject: [PATCH 4/6] docs(data): documente le pipeline d'import historique --- docs/architecture/40-data.md | 70 +++++++ etl/README.md | 354 ++++++++++++++++++++++++++++++++++- 2 files changed, 417 insertions(+), 7 deletions(-) diff --git a/docs/architecture/40-data.md b/docs/architecture/40-data.md index 6566753..ffd5e6d 100644 --- a/docs/architecture/40-data.md +++ b/docs/architecture/40-data.md @@ -231,3 +231,73 @@ et ne sont pas considérées comme des alertes actuelles. - Les mesures API ne sont pas rattachées à un dataset historique. - Une alerte peut être associée à une prévision du même site. - Une alerte peut donner lieu à plusieurs recommandations. + +## Ingestion des données historiques + +Le MVP EnerVision initialise les données énergétiques à partir du dataset fourni dans le cadre du projet. + +Le dataset de référence contient 122 647 mesures issues de 7 sites et couvre la période du 1er janvier 2023 au 31 décembre 2024. + +Les fichiers sources CSV et JSON sont nécessaires uniquement pour l'initialisation des données. Ils ne sont pas versionnés dans Git et sont placés localement dans `data/raw/`. + +### Architecture du flux + +```text +Dataset CSV + métadonnées JSON + | + v + historical_import.py + | + +------+------+ + | | + v v + Validation SHA-256 + | Traçabilité + +------+------+ + | + v + Normalisation + + qualité data + | + v + Chargement par batches + | + v + PostgreSQL / TimescaleDB + | | | + v v v + dataset site reading +``` + +Le pipeline est développé en Python. + +Pandas est utilisé pour l'extraction, la validation et la préparation des données. SQLAlchemy Async assure le chargement transactionnel dans PostgreSQL/TimescaleDB. + +Une empreinte SHA-256 permet d'identifier le dataset utilisé et d'assurer sa traçabilité. + +Les valeurs manquantes sont conservées pendant l'ingestion afin de préserver les données sources. Aucune imputation n'est réalisée à cette étape. + +Le chargement des mesures est effectué par batches de 1 000 lignes. + +Les données provenant du dataset CSV sont identifiées par `source = "csv"` et associées à leur `dataset_id`. + +### Résultats validés + +Le chargement de référence a permis d'obtenir : + +- 1 dataset ; +- 7 sites ; +- 122 647 mesures ; +- 0 doublon détecté dans le dataset source. + +L'idempotence a également été vérifiée par une deuxième exécution du pipeline : aucune nouvelle mesure n'a été créée et le nombre de `reading` est resté à 122 647. + +La procédure détaillée d'installation, d'exécution, de validation et de contrôle du pipeline est disponible dans `etl/README.md`. + +### Évolution prévue + +L'étape suivante consiste à orchestrer les traitements Data avec Apache Airflow. + +L'orchestration réutilisera la logique ETL existante afin de séparer la logique de traitement de la planification, du suivi des exécutions et de la gestion des erreurs. + +Le pipeline servira ensuite de base à la préparation des données nécessaires au modèle de Machine Learning. diff --git a/etl/README.md b/etl/README.md index cac0f3d..b835311 100644 --- a/etl/README.md +++ b/etl/README.md @@ -1,9 +1,349 @@ -# ETL +# Pipeline ETL — EnerVision -Orchestration Apache Airflow : ingestion des mesures, agregations continues, -controles de qualite. Non initialise, voir le ticket dedie. +## Objectif -- `airflow/dags` : DAGs. -- `airflow/plugins` : operateurs et hooks maison. -- `airflow/include` : requetes SQL et ressources referencees par les DAGs. -- `airflow/tests` : tests d'integrite des DAGs. +Le pipeline ETL EnerVision permet d'intégrer les données énergétiques historiques dans PostgreSQL/TimescaleDB. + +Cette première étape du pipeline Data permet de charger le dataset fourni dans le cadre du projet, contenant les mesures énergétiques de 7 sites sur la période du 1er janvier 2023 au 31 décembre 2024. + +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é du dataset importé ; +- le chargement des données dans PostgreSQL/TimescaleDB ; +- l'idempotence du chargement afin d'éviter la création de doublons. + +## Données sources + +Le dataset est fourni par le formateur dans le cadre du projet EnerVision. + +Il contient les deux fichiers suivants : + +```text +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 une fois les fichiers fournis par le formateur et les place dans : + +```text +data/raw/ +``` + +Structure locale attendue : + +```text +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. + +## Technologies utilisées + +| Technologie | Utilisation | +|---|---| +| Python | Développement du pipeline ETL | +| Pandas | Lecture, validation et transformation des données | +| JSON | Lecture des métadonnées du dataset | +| hashlib / SHA-256 | Identification, intégrité et traçabilité du dataset | +| 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 | +| Pytest | Tests automatisés | + +## Fonctionnement du pipeline + +Le script principal d'import se trouve dans : + +```text +apps/backend/app/etl/historical_import.py +``` + +Le flux d'import est le suivant : + +```text +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é : + +```text +6E3777A97A5660B11855750B9028F70BE72138A11F26795F3A35D9CE74CE0C8D +``` + +Cette empreinte participe à la traçabilité du dataset chargé. + +### 5. Transformation + +Les timestamps sont normalisés avec la timezone : + +```text +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 : + +```text +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 : + +```text +dataset +site +reading +``` + +Les mesures sont chargées par batches de : + +```text +1000 lignes +``` + +Les mesures provenant du dataset CSV utilisent : + +```text +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 en dry-run + +Depuis le dossier : + +```text +apps/backend/ +``` + +exécuter : + +```powershell +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 réel + +Depuis `apps/backend/` : + +```powershell +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 : + +```text +Chargement : 1000/122647 +Chargement : 2000/122647 +... +Chargement : 122647/122647 +``` + +## Résultats obtenus + +Après le chargement initial, les contrôles en base ont confirmé : + +```text +datasets = 1 +sites = 7 +readings = 122647 +source = csv +``` + +Le premier import a créé : + +```text +nouvelles lectures : 122647 +``` + +## Idempotence + +Le pipeline a été exécuté une deuxième fois avec exactement le même dataset afin de vérifier son idempotence. + +Résultat : + +```text +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 + +Depuis la racine du projet, vérifier le nombre d'enregistrements avec : + +```powershell +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 : + +```text +datasets = 1 +sites = 7 +readings = 122647 +``` + +Vérifier la source des mesures avec : + +```powershell +docker compose exec db psql -U enervision -d enervision -c "SELECT source, COUNT(*) FROM reading GROUP BY source ORDER BY source;" +``` + +Résultat attendu : + +```text +csv | 122647 +``` + +## Tests et qualité + +Les tests automatisés du pipeline sont situés dans : + +```text +apps/backend/tests/etl/ +``` + +Ils 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. + +Exécuter les tests ETL : + +```powershell +uv run pytest tests\etl -v +``` + +Contrôler la qualité du code : + +```powershell +uv run ruff check app\etl tests\etl +``` + +## Suite du pipeline Data + +L'import historique constitue la première brique du pipeline Data EnerVision. + +La prochaine étape consiste à orchestrer les traitements ETL avec Apache Airflow, puis à préparer les données nécessaires à l'entraînement du modèle de Machine Learning. + +Airflow sera utilisé comme orchestrateur des traitements existants et ne remplacera pas la logique métier déjà implémentée dans le pipeline ETL. \ No newline at end of file From f03dce5fe37225a022bd6e3f9077e0e33cdf43d4 Mon Sep 17 00:00:00 2001 From: Meryemel-gham Date: Thu, 17 Sep 2026 10:06:54 +0200 Subject: [PATCH 5/6] style(data): applique le formatage Ruff --- apps/backend/app/etl/historical_import.py | 292 ++++-------------- .../tests/etl/test_historical_import.py | 9 +- 2 files changed, 62 insertions(+), 239 deletions(-) diff --git a/apps/backend/app/etl/historical_import.py b/apps/backend/app/etl/historical_import.py index c3b9a70..b4f2089 100644 --- a/apps/backend/app/etl/historical_import.py +++ b/apps/backend/app/etl/historical_import.py @@ -68,11 +68,7 @@ def classify_quality( Les valeurs NULL sont conservées. On ne cherche pas ici à déterminer la cause physique exacte de leur absence. """ - missing = [ - column - for column in MEASURE_COLUMNS - if pd.isna(row.get(column)) - ] + missing = [column for column in MEASURE_COLUMNS if pd.isna(row.get(column))] if not missing: quality = "good" @@ -83,10 +79,7 @@ def classify_quality( else: quality = "partial" - reasons = [ - f"missing:{column}" - for column in missing - ] + reasons = [f"missing:{column}" for column in missing] return quality, reasons @@ -96,57 +89,33 @@ def validate_source( metadata: dict[str, Any], ) -> None: """Valide le dataset avant tout chargement en base.""" - missing_columns = REQUIRED_COLUMNS.difference( - frame.columns - ) + missing_columns = REQUIRED_COLUMNS.difference(frame.columns) if missing_columns: - raise ValueError( - "Colonnes obligatoires absentes : " - f"{sorted(missing_columns)}" - ) + raise ValueError(f"Colonnes obligatoires absentes : {sorted(missing_columns)}") expected_records = int(metadata["total_records"]) if len(frame) != expected_records: - raise ValueError( - "Nombre de lignes inattendu : " - f"{len(frame)} au lieu de " - f"{expected_records}" - ) + raise ValueError(f"Nombre de lignes inattendu : {len(frame)} au lieu de {expected_records}") expected_sites = set(metadata["sites"].keys()) actual_sites = set(frame["site_id"].unique()) if actual_sites != expected_sites: raise ValueError( - "Sites incohérents. " - f"Attendus={sorted(expected_sites)}, " - f"trouvés={sorted(actual_sites)}" + f"Sites incohérents. Attendus={sorted(expected_sites)}, trouvés={sorted(actual_sites)}" ) - duplicated = frame.duplicated( - subset=["site_id", "timestamp"] - ).sum() + duplicated = frame.duplicated(subset=["site_id", "timestamp"]).sum() if duplicated: - raise ValueError( - f"{duplicated} doublons " - "(site_id, timestamp) détectés" - ) + raise ValueError(f"{duplicated} doublons (site_id, timestamp) détectés") - static_variants = ( - frame.groupby("site_id")[ - ["site_type", "site_name"] - ] - .nunique() - ) + static_variants = frame.groupby("site_id")[["site_type", "site_name"]].nunique() if (static_variants > 1).any().any(): - raise ValueError( - "Un site possède plusieurs valeurs " - "de site_type ou site_name." - ) + raise ValueError("Un site possède plusieurs valeurs de site_type ou site_name.") # Vérifie également que tous les timestamps # peuvent être interprétés correctement. @@ -168,9 +137,7 @@ def normalize_timestamps( """ normalized = frame.copy() - normalized["_source_timestamp"] = ( - normalized["timestamp"] - ) + normalized["_source_timestamp"] = normalized["timestamp"] timestamps = pd.to_datetime( normalized["timestamp"], @@ -178,13 +145,9 @@ def normalize_timestamps( ) if timestamps.dt.tz is None: - timestamps = timestamps.dt.tz_localize( - source_timezone - ) + timestamps = timestamps.dt.tz_localize(source_timezone) else: - timestamps = timestamps.dt.tz_convert( - source_timezone - ) + timestamps = timestamps.dt.tz_convert(source_timezone) normalized["timestamp"] = timestamps @@ -202,7 +165,7 @@ def to_json_value(value: Any) -> Any: try: if pd.isna(value): return None - except (TypeError, ValueError): + except TypeError, ValueError: pass if isinstance(value, pd.Timestamp): @@ -247,27 +210,13 @@ async def ensure_dataset( return int(existing) metadata_summary = { - "generator_version": metadata.get( - "generator_version" - ), - "total_sites": metadata.get( - "total_sites" - ), - "total_records": metadata.get( - "total_records" - ), - "date_range": metadata.get( - "date_range" - ), - "frequency": metadata.get( - "frequency" - ), - "null_injection_enabled": metadata.get( - "null_injection_enabled" - ), - "null_strategies": metadata.get( - "null_strategies" - ), + "generator_version": metadata.get("generator_version"), + "total_sites": metadata.get("total_sites"), + "total_records": metadata.get("total_records"), + "date_range": metadata.get("date_range"), + "frequency": metadata.get("frequency"), + "null_injection_enabled": metadata.get("null_injection_enabled"), + "null_strategies": metadata.get("null_strategies"), "importer": "historical_import_v1", } @@ -292,10 +241,7 @@ async def ensure_dataset( """ ), { - "dataset_name": ( - "EnerVision historical dataset " - "2023-2024" - ), + "dataset_name": ("EnerVision historical dataset 2023-2024"), "archive_sha256": sha256, "storage_uri": storage_uri, "source_timezone": source_timezone, @@ -322,12 +268,8 @@ async def upsert_sites( "site_name", ] ] - .drop_duplicates( - subset=["site_id"] - ) - .to_dict( - orient="records" - ) + .drop_duplicates(subset=["site_id"]) + .to_dict(orient="records") ) await connection.execute( @@ -363,12 +305,8 @@ def build_reading_batch( """ rows: list[dict[str, Any]] = [] - for record in chunk.to_dict( - orient="records" - ): - quality, reasons = classify_quality( - record - ) + for record in chunk.to_dict(orient="records"): + quality, reasons = classify_quality(record) raw_data = { column: to_json_value(value) @@ -378,9 +316,7 @@ def build_reading_batch( # Dans raw_data, on conserve le timestamp # exactement tel qu'il était dans le CSV. - raw_data["timestamp"] = to_json_value( - record["_source_timestamp"] - ) + raw_data["timestamp"] = to_json_value(record["_source_timestamp"]) rows.append( { @@ -388,59 +324,25 @@ def build_reading_batch( "timestamp": record["timestamp"], "source": SOURCE_NAME, "dataset_id": dataset_id, - # Non fourni par le dataset historique. "consumption_kw": None, - - "consumption_kwh": to_json_value( - record["consumption_kwh"] - ), - "consumption_euros": to_json_value( - record["consumption_euros"] - ), - + "consumption_kwh": to_json_value(record["consumption_kwh"]), + "consumption_euros": to_json_value(record["consumption_euros"]), # Non fournis par le CSV historique. "voltage_v": None, "current_a": None, "power_factor": None, - - "temperature_celsius": ( - to_json_value( - record[ - "temperature_celsius" - ] - ) - ), - "humidity_percent": ( - to_json_value( - record[ - "humidity_percent" - ] - ) - ), - "solar_irradiance_wm2": ( - to_json_value( - record[ - "solar_irradiance_wm2" - ] - ) - ), - - "is_working_hours": bool( - record[ - "is_working_hours" - ] - ), - + "temperature_celsius": (to_json_value(record["temperature_celsius"])), + "humidity_percent": (to_json_value(record["humidity_percent"])), + "solar_irradiance_wm2": (to_json_value(record["solar_irradiance_wm2"])), + "is_working_hours": bool(record["is_working_hours"]), "data_quality": quality, "null_reasons": reasons, - # Aucune imputation pendant l'ingestion RAW. # Les valeurs manquantes sont conservées telles quelles # afin de préserver la donnée source. "imputed_values": None, "imputation_method": None, - # Conservation de la donnée source # pour la traçabilité. "raw_data": json.dumps( @@ -519,56 +421,29 @@ async def import_historical( 3. Transform 4. Load """ - metadata = load_metadata( - metadata_path - ) + metadata = load_metadata(metadata_path) - frame = pd.read_csv( - csv_path - ) + frame = pd.read_csv(csv_path) validate_source( frame, metadata, ) - print( - f"Lignes : {len(frame)}" - ) - print( - "Sites : " - f"{frame['site_id'].nunique()}" - ) - print( - "Période : " - f"{frame['timestamp'].min()} -> " - f"{frame['timestamp'].max()}" - ) - print( - "Doublons : " - f"{frame.duplicated(['site_id', 'timestamp']).sum()}" - ) + print(f"Lignes : {len(frame)}") + print(f"Sites : {frame['site_id'].nunique()}") + print(f"Période : {frame['timestamp'].min()} -> {frame['timestamp'].max()}") + print(f"Doublons : {frame.duplicated(['site_id', 'timestamp']).sum()}") print("\nValeurs NULL :") - print( - frame[ - MEASURE_COLUMNS - ].isna().sum() - ) + print(frame[MEASURE_COLUMNS].isna().sum()) - sha256 = compute_sha256( - csv_path - ) + sha256 = compute_sha256(csv_path) - print( - f"\nSHA-256 : {sha256}" - ) + print(f"\nSHA-256 : {sha256}") if dry_run: - print( - "\nDry-run terminé : " - "aucune donnée écrite." - ) + print("\nDry-run terminé : aucune donnée écrite.") return normalized = normalize_timestamps( @@ -613,18 +488,14 @@ async def import_historical( }, ) - before = int( - result.scalar_one() - ) + before = int(result.scalar_one()) for start in range( 0, len(normalized), batch_size, ): - chunk = normalized.iloc[ - start : start + batch_size - ] + chunk = normalized.iloc[start : start + batch_size] rows = build_reading_batch( chunk, @@ -641,11 +512,7 @@ async def import_historical( len(normalized), ) - print( - "Chargement : " - f"{loaded}/" - f"{len(normalized)}" - ) + print(f"Chargement : {loaded}/{len(normalized)}") result = await connection.execute( text( @@ -662,29 +529,13 @@ async def import_historical( }, ) - after = int( - result.scalar_one() - ) + after = int(result.scalar_one()) - print( - "\nImport terminé." - ) - print( - "dataset_id : " - f"{dataset_id}" - ) - print( - "lectures avant : " - f"{before}" - ) - print( - "lectures après : " - f"{after}" - ) - print( - "nouvelles lectures : " - f"{after - before}" - ) + print("\nImport terminé.") + print(f"dataset_id : {dataset_id}") + print(f"lectures avant : {before}") + print(f"lectures après : {after}") + print(f"nouvelles lectures : {after - before}") finally: await engine.dispose() @@ -692,11 +543,7 @@ async def import_historical( def parse_args() -> argparse.Namespace: """Définit les arguments CLI de l'import.""" - parser = argparse.ArgumentParser( - description=( - "Import historique EnerVision" - ) - ) + parser = argparse.ArgumentParser(description=("Import historique EnerVision")) parser.add_argument( "--csv", @@ -709,38 +556,26 @@ def parse_args() -> argparse.Namespace: "--metadata", type=Path, required=True, - help=( - "Chemin vers le fichier " - "dataset_metadata.json." - ), + help=("Chemin vers le fichier dataset_metadata.json."), ) parser.add_argument( "--source-timezone", default="UTC", - help=( - "Timezone associée aux timestamps " - "du dataset. Défaut : UTC." - ), + help=("Timezone associée aux timestamps du dataset. Défaut : UTC."), ) parser.add_argument( "--batch-size", type=int, default=1000, - help=( - "Nombre de lignes insérées " - "par batch. Défaut : 1000." - ), + help=("Nombre de lignes insérées par batch. Défaut : 1000."), ) parser.add_argument( "--dry-run", action="store_true", - help=( - "Valide les données sans " - "écrire en base." - ), + help=("Valide les données sans écrire en base."), ) return parser.parse_args() @@ -751,26 +586,19 @@ def main() -> None: args = parse_args() if args.batch_size <= 0: - raise ValueError( - "--batch-size doit être " - "strictement supérieur à 0." - ) + raise ValueError("--batch-size doit être strictement supérieur à 0.") # resolve() est volontairement exécuté ici, # dans la partie synchrone du programme. # Cela évite une opération filesystem bloquante # à l'intérieur d'une fonction async. - storage_uri = ( - args.csv.resolve().as_uri() - ) + storage_uri = args.csv.resolve().as_uri() asyncio.run( import_historical( csv_path=args.csv, metadata_path=args.metadata, - source_timezone=( - args.source_timezone - ), + source_timezone=(args.source_timezone), batch_size=args.batch_size, dry_run=args.dry_run, storage_uri=storage_uri, diff --git a/apps/backend/tests/etl/test_historical_import.py b/apps/backend/tests/etl/test_historical_import.py index f311b2d..31f6e2d 100644 --- a/apps/backend/tests/etl/test_historical_import.py +++ b/apps/backend/tests/etl/test_historical_import.py @@ -104,9 +104,7 @@ def test_validate_source_accepts_valid_dataset(): def test_validate_source_rejects_missing_column(): - frame = make_dataframe().drop( - columns=["consumption_kwh"] - ) + frame = make_dataframe().drop(columns=["consumption_kwh"]) with pytest.raises( ValueError, @@ -234,10 +232,7 @@ def test_build_reading_batch_keeps_missing_values(): assert row["temperature_celsius"] is None - assert ( - "missing:temperature_celsius" - in row["null_reasons"] - ) + assert "missing:temperature_celsius" in row["null_reasons"] # RAW ingestion : aucune imputation. assert row["imputed_values"] is None From 6798d355722cf0520888ee08aa648c2bc3941e35 Mon Sep 17 00:00:00 2001 From: Meryemel-gham Date: Thu, 17 Sep 2026 10:41:18 +0200 Subject: [PATCH 6/6] fix(data): corrige le typage de l'import historique --- apps/backend/app/etl/historical_import.py | 21 ++++++++++++++++----- apps/backend/pyproject.toml | 1 + apps/backend/uv.lock | 14 ++++++++++++++ 3 files changed, 31 insertions(+), 5 deletions(-) diff --git a/apps/backend/app/etl/historical_import.py b/apps/backend/app/etl/historical_import.py index b4f2089..22d9b03 100644 --- a/apps/backend/app/etl/historical_import.py +++ b/apps/backend/app/etl/historical_import.py @@ -5,7 +5,7 @@ import asyncio import hashlib import json from pathlib import Path -from typing import Any +from typing import Any, cast import pandas as pd from sqlalchemy import text @@ -56,7 +56,12 @@ def compute_sha256(path: Path) -> str: def load_metadata(path: Path) -> dict[str, Any]: """Charge les métadonnées fournies avec le dataset.""" with path.open("r", encoding="utf-8") as source: - return json.load(source) + metadata = json.load(source) + + if not isinstance(metadata, dict): + raise ValueError("Le fichier de métadonnées doit contenir un objet JSON.") + + return cast(dict[str, Any], metadata) def classify_quality( @@ -260,7 +265,8 @@ async def upsert_sites( frame: pd.DataFrame, ) -> None: """Insère ou met à jour les sites du dataset.""" - sites = ( + sites = cast( + list[dict[str, Any]], frame[ [ "site_id", @@ -269,7 +275,7 @@ async def upsert_sites( ] ] .drop_duplicates(subset=["site_id"]) - .to_dict(orient="records") + .to_dict(orient="records"), ) await connection.execute( @@ -305,7 +311,12 @@ def build_reading_batch( """ rows: list[dict[str, Any]] = [] - for record in chunk.to_dict(orient="records"): + records = cast( + list[dict[str, Any]], + chunk.to_dict(orient="records"), + ) + + for record in records: quality, reasons = classify_quality(record) raw_data = { diff --git a/apps/backend/pyproject.toml b/apps/backend/pyproject.toml index 6cf42a5..c330c8d 100644 --- a/apps/backend/pyproject.toml +++ b/apps/backend/pyproject.toml @@ -27,6 +27,7 @@ dev = [ "pytest-asyncio>=1.4.0", "pytest-cov>=7.1.0", "httpx>=0.28.1", + "pandas-stubs>=3.0.5.260914", ] [build-system] diff --git a/apps/backend/uv.lock b/apps/backend/uv.lock index fb43797..6836990 100644 --- a/apps/backend/uv.lock +++ b/apps/backend/uv.lock @@ -330,6 +330,7 @@ dependencies = [ dev = [ { name = "httpx" }, { name = "mypy" }, + { name = "pandas-stubs" }, { name = "pytest" }, { name = "pytest-asyncio" }, { name = "pytest-cov" }, @@ -357,6 +358,7 @@ requires-dist = [ dev = [ { name = "httpx", specifier = ">=0.28.1" }, { name = "mypy", specifier = ">=2.3.1" }, + { name = "pandas-stubs", specifier = ">=3.0.5.260914" }, { name = "pytest", specifier = ">=9.1.1" }, { name = "pytest-asyncio", specifier = ">=1.4.0" }, { name = "pytest-cov", specifier = ">=7.1.0" }, @@ -669,6 +671,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d2/cf/6a51b2c38980e04c279fd2fa908a1b0982064e860444acfca4ec2e2c8359/pandas-3.0.5-cp314-cp314t-win_arm64.whl", hash = "sha256:3c5015fd1730fbf883647e88068176c839c102cea883ba1769a6f4593bfc1f8c", size = 9509776, upload-time = "2026-07-22T22:19:26.694Z" }, ] +[[package]] +name = "pandas-stubs" +version = "3.0.5.260914" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c1/93/8948ae6c1e1e3d6833596fd266f7be2d27c1451b8be094975ad42c5e842e/pandas_stubs-3.0.5.260914.tar.gz", hash = "sha256:3f6fc1f147f68fd89c007105e7c94a948acb4ecd7eb20dc1c02e153c4ed5c250", size = 117622, upload-time = "2026-09-14T16:42:35.065Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9a/cb/5ad79e02a556cc23fed5816de0109fa8af660c66cfa5f4af74c3e8d4cd26/pandas_stubs-3.0.5.260914-py3-none-any.whl", hash = "sha256:39a1300c5c5c55fdf609e3476805decce5d5015539a4dcb683449f8feaeee2fb", size = 177344, upload-time = "2026-09-14T16:42:33.771Z" }, +] + [[package]] name = "pathspec" version = "1.1.1"