diff --git a/.github/workflows/ml.yml b/.github/workflows/ml.yml index 0b745eb..4145c4e 100644 --- a/.github/workflows/ml.yml +++ b/.github/workflows/ml.yml @@ -8,10 +8,28 @@ on: paths: - "ml/**" - ".github/workflows/ml.yml" + # Le job `integration` monte son schema avec les migrations du backend et joue le test de + # chaine qui vit dans ses tests : sans ces chemins, une migration modifiee ne declencherait + # rien et le schema deriverait du SQL du pipeline sans que rien ne casse. Meme raisonnement + # que le filtre d'airflow.yml, qui inclut deja des chemins de ml/ et de apps/backend/. + - "apps/backend/alembic/**" + - "apps/backend/app/models/**" + - "apps/backend/tests/test_chaine_ml_api.py" + - "apps/backend/pyproject.toml" + - "apps/backend/uv.lock" pull_request: paths: - "ml/**" - ".github/workflows/ml.yml" + # Le job `integration` monte son schema avec les migrations du backend et joue le test de + # chaine qui vit dans ses tests : sans ces chemins, une migration modifiee ne declencherait + # rien et le schema deriverait du SQL du pipeline sans que rien ne casse. Meme raisonnement + # que le filtre d'airflow.yml, qui inclut deja des chemins de ml/ et de apps/backend/. + - "apps/backend/alembic/**" + - "apps/backend/app/models/**" + - "apps/backend/tests/test_chaine_ml_api.py" + - "apps/backend/pyproject.toml" + - "apps/backend/uv.lock" permissions: contents: read @@ -53,11 +71,91 @@ jobs: - name: Typage run: uv run mypy enervision_ml tests - # Aucun test ne touche PostgreSQL ni MLflow distant : tout tourne sur donnees - # synthetiques ou un magasin SQLite local jetable (cf. ml/tests/test_train.py). + # Les tests exigeant une base portent le marqueur `integration`, ecarte par defaut et + # joue par le job `integration` ci-dessous. - name: Tests run: uv run pytest + # Le seul job du depot qui dispose a la fois des deux environnements uv et d'une base. Piege : + # le schema de la base ML est celui du backend (apps/backend/alembic, proprietaire du schema). + # Le reconstruire ici a la main rendrait ce job vert sur une base qui n'est pas la notre. + integration: + name: ML - DB et chaîne ML - DB - API + runs-on: ubuntu-latest + + services: + db: + image: timescale/timescaledb-ha:pg17 + env: + POSTGRES_USER: enervision + POSTGRES_PASSWORD: change_me + POSTGRES_DB: enervision_test + ports: + - "5433:5432" + options: >- + --health-cmd "pg_isready -U enervision -d enervision_test" + --health-interval 10s + --health-timeout 5s + --health-retries 12 + --health-start-period 40s + + env: + # Deux variables, deux dialectes : Alembic et l'API parlent asyncpg, le pipeline ML parle + # psycopg en synchrone. Cf. docs/ML-START.md, section 1. + DATABASE_URL: postgresql+asyncpg://enervision:change_me@localhost:5433/enervision_test + ML_DATABASE_URL: postgresql+psycopg://enervision:change_me@localhost:5433/enervision_test + APP_SECRET_KEY: secret-de-test-assez-long-pour-le-validateur + PGPASSWORD: change_me + + steps: + - name: Récupère le dépôt + uses: actions/checkout@v7 + + - name: Installe uv + uses: astral-sh/setup-uv@v7 + with: + enable-cache: true + cache-dependency-glob: | + ml/uv.lock + apps/backend/uv.lock + + - name: Installe l'interpréteur déclaré par .python-version + working-directory: ml + run: uv python install + + - name: Synchronise le pipeline ML sans dévier du verrou + working-directory: ml + run: uv sync --all-groups --frozen + + # Le backend est installé ici parce qu'il porte les migrations, seule source du schéma, et + # le test de chaîne, qui interroge l'API. + - name: Synchronise le backend sans dévier du verrou + working-directory: apps/backend + run: uv sync --all-groups --frozen + + # db/init/110-test-database.sql n'est pas monté ici, et sans l'extension la première + # révision Alembic refuse de s'appliquer. + - name: Active TimescaleDB sur la base de test + run: psql -h localhost -p 5433 -U enervision -d enervision_test -c "CREATE EXTENSION IF NOT EXISTS timescaledb" + + - name: Applique les migrations du backend, propriétaire du schéma + working-directory: apps/backend + run: uv run alembic upgrade head + + # `-m` en ligne de commande écrase celui d'addopts. Couverture désactivée : ce job ne joue + # qu'une partie de la suite, son taux n'aurait pas de sens (même raison que backend.yml). + - name: Tests ML exigeant une base + working-directory: ml + run: uv run pytest -m integration --no-cov + + # Lance les vrais binaires enervision_ml.train et .score en sous-processus, comme les DAGs + # ml_train et ml_score, puis relit le résultat par GET /api/v1/predictions. + - name: Chaîne complète ML vers DB vers API + working-directory: apps/backend + env: + ML_PYTHON: ${{ github.workspace }}/ml/.venv/bin/python + run: uv run pytest -m chaine --no-cov + sast: name: Analyse statique de sécurité runs-on: ubuntu-latest diff --git a/Makefile b/Makefile index c749f76..01e8afd 100644 --- a/Makefile +++ b/Makefile @@ -25,6 +25,13 @@ MAILPIT_UI_PORT := $(or $(strip $(call env-val,MAILPIT_UI_PORT)),8025) ML_DATABASE_URL ?= postgresql+psycopg://$(PG_USER):$(PG_PASSWORD)@localhost:$(PG_PORT)/$(PG_DB) export ML_DATABASE_URL +# Piege : la base des tests d'integration n'est pas la base de developpement. Ces tests ecrivent +# et suppriment des lignes, et leurs fixtures refusent de demarrer ailleurs que sur +# `enervision_test` (garde sur le nom, cf. ml/tests/conftest.py). +PG_TEST_DB ?= enervision_test +TEST_DATABASE_URL ?= postgresql+asyncpg://$(PG_USER):$(PG_PASSWORD)@localhost:$(PG_PORT)/$(PG_TEST_DB) +ML_TEST_DATABASE_URL ?= postgresql+psycopg://$(PG_USER):$(PG_PASSWORD)@localhost:$(PG_PORT)/$(PG_TEST_DB) + # Le jeu historique s'arrete au 31/12/2024 : score et detection ancres a l'horloge reelle ne # verraient qu'un parc muet depuis des mois. Cf. `--now` de enervision_ml.score. DEMO_NOW ?= 2024-12-31T00:00:00Z @@ -32,9 +39,10 @@ DEMO_NOW ?= 2024-12-31T00:00:00Z .DEFAULT_GOAL := help .PHONY: help install install-backend install-frontend install-ml install-airflow \ dev dev-backend dev-frontend \ - lint format typecheck test test-cov test-integration check \ + lint format typecheck test test-cov test-integration ml-test-integration \ + test-chaine check \ openapi docker-build db-up db-down db-reset db-logs db-psql db-wait db-ensure-airflow \ - migrate bootstrap-admin services-up demo-data demo-data-force \ + migrate migrate-test bootstrap-admin services-up demo-data demo-data-force \ ml-lint ml-typecheck ml-test ml-check ml-train ml-score detect-alerts recommendations \ airflow-lint airflow-test airflow-check airflow-up airflow-down airflow-logs \ tls-selfsigned tls-acme tls-renew stack-up stack-down stack-logs @@ -112,6 +120,16 @@ ml-test: ## Exécute les tests du pipeline ML (donnees synthetiques, sans base n ml-check: ml-lint ml-typecheck ml-test ## Chaîne de vérification complète du pipeline ML +# La cible surcharge ML_DATABASE_URL, que ce Makefile exporte vers la base de développement : la +# garde du conftest ferait échouer la cible sans cette surcharge. +ml-test-integration: ML_DATABASE_URL := $(ML_TEST_DATABASE_URL) +ml-test-integration: ## Tests ML exigeant une base migrée. Faire `make db-up migrate-test` avant + cd $(ML) && uv run pytest -m integration --no-cov + +test-chaine: ## Chaîne ML -> DB -> API, vrais binaires. Exige les deux environnements uv + cd $(BACKEND) && DATABASE_URL=$(TEST_DATABASE_URL) ML_PYTHON=$(CURDIR)/$(ML)/.venv/bin/python \ + uv run pytest -m chaine --no-cov + ml-train: ## Entraine le modele LightGBM. CSV=chemin optionnel, sinon lit ML_DATABASE_URL cd $(ML) && uv run python -m enervision_ml.train $(if $(CSV),--csv $(CSV),) @@ -209,6 +227,9 @@ db-ensure-airflow: ## Crée la base de métadonnées Airflow si le volume pgdata migrate: ## Applique les migrations Alembic cd $(BACKEND) && uv run alembic upgrade head +migrate-test: ## Applique les migrations sur enervision_test, la base des tests d'intégration + cd $(BACKEND) && DATABASE_URL=$(TEST_DATABASE_URL) uv run alembic upgrade head + bootstrap-admin: ## Crée le premier administrateur, mot de passe saisi au clavier cd $(BACKEND) && uv run python -m app.cli create-admin --email $${EMAIL:?EMAIL=... requis} diff --git a/apps/backend/pyproject.toml b/apps/backend/pyproject.toml index cfe6481..e884616 100644 --- a/apps/backend/pyproject.toml +++ b/apps/backend/pyproject.toml @@ -87,8 +87,11 @@ disallow_untyped_defs = false testpaths = ["tests"] asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "function" -addopts = "-q --strict-markers -m 'not integration' --cov=app --cov-report=term-missing" -markers = ["integration: requiert une base PostgreSQL joignable, hors `make test`"] +addopts = "-q --strict-markers -m 'not integration and not chaine' --cov=app --cov-report=term-missing" +markers = [ + "integration: requiert une base PostgreSQL joignable, hors `make test`", + "chaine: requiert en plus l'environnement uv de ml/, hors `make test` et hors `-m integration`", +] [tool.coverage.run] source = ["app"] diff --git a/apps/backend/tests/test_chaine_ml_api.py b/apps/backend/tests/test_chaine_ml_api.py new file mode 100644 index 0000000..f2718a5 --- /dev/null +++ b/apps/backend/tests/test_chaine_ml_api.py @@ -0,0 +1,223 @@ +"""Piege : ce fichier porte le marqueur `chaine`, pas `integration` - test_the_ml_binaries...() + +Il lance les vrais binaires `enervision_ml.train` et `enervision_ml.score` dans l'environnement +uv de `ml/`, que le job `integration` de `backend.yml` n'installe pas. Un marqueur distinct evite +que ce job, et `make test`, ne le selectionnent et n'echouent faute de `ml/.venv`. +""" + +import math +import os +import subprocess +from collections.abc import AsyncIterator, Iterator +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +from functools import partial +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import anyio +import pytest +from fastapi import FastAPI +from httpx import AsyncClient +from sqlalchemy import delete, insert, make_url + +from app.api.deps import get_current_principal +from app.core.config import get_settings +from app.core.principal import Principal +from app.core.roles import AccountKind, Role +from app.db.session import get_session_factory +from app.models.energy import Prediction, Reading, Site + +pytestmark = pytest.mark.chaine + +RACINE = Path(__file__).resolve().parents[3] +ML = RACINE / "ml" +PYTHON_ML = Path(os.environ.get("ML_PYTHON", ML / ".venv" / "bin" / "python")) + +HEURES_COMPLETES = 400 +HEURES_INSUFFISANTES = 100 + + +def lecteur() -> Principal: + return Principal( + id=uuid4(), + email="lecteur@enervision.fr", + role=Role.LECTEUR, + kind=AccountKind.HUMAIN, + must_change_password=False, + ) + + +def url_ml() -> str: + """Derive la chaine du pipeline de celle du backend plutot que de la recopier : les deux + cotes visent ainsi la meme base, dans leur dialecte respectif.""" + return ( + make_url(get_settings().database_url) + .set(drivername="postgresql+psycopg") + .render_as_string(hide_password=False) + ) + + +def lance_ml(module: str, *arguments: str, journal: Path) -> subprocess.CompletedProcess[str]: + if not PYTHON_ML.exists(): + pytest.fail( + f"Environnement ml/ absent ({PYTHON_ML}). Lancer `cd ml && uv sync --all-groups`." + ) + + return subprocess.run( # noqa: S603 -- argv en liste, sans shell, binaire resolu dans le depot + [str(PYTHON_ML), "-m", module, *arguments], + cwd=ML, + text=True, + capture_output=True, + timeout=600, + check=False, + env={ + **os.environ, + "ML_DATABASE_URL": url_ml(), + "MLFLOW_TRACKING_URI": f"sqlite:///{journal}/mlflow.db", + }, + ) + + +async def executer(module: str, *arguments: str, journal: Path) -> subprocess.CompletedProcess[str]: + resultat = await anyio.to_thread.run_sync( + partial(lance_ml, module, *arguments, journal=journal) + ) + assert resultat.returncode == 0, resultat.stderr + return resultat + + +@dataclass +class Parc: + sites: list[str] = field(default_factory=list) + + +def lignes_horaires(site_id: str, *, heures: int, fin: datetime) -> list[dict[str, Any]]: + return [ + { + "site_id": site_id, + "timestamp": fin - timedelta(hours=decalage), + "source": "api_history", + "consumption_kwh": 50.0 + math.sin(decalage / 12.0) * 10.0, + "temperature_celsius": 15.0, + "humidity_percent": 50.0, + "solar_irradiance_wm2": 0.0, + "is_working_hours": True, + "raw_data": {}, + } + for decalage in reversed(range(heures)) + ] + + +@pytest.fixture +async def parc() -> AsyncIterator[Parc]: + """Deux sites dotes d'un historique complet, un troisieme qui n'atteint pas le lag de 168 h. + + Les ecritures sont validees : les binaires ML ouvrent leur propre connexion et ne verraient + pas une transaction en cours. + """ + fin = datetime.now(UTC).replace(minute=0, second=0, microsecond=0) - timedelta(hours=1) + marque = uuid4().hex[:12] + complets = [f"TEST-{marque}-A", f"TEST-{marque}-B"] + partiel = f"TEST-{marque}-C" + parc = Parc(sites=[*complets, partiel]) + + async with get_session_factory()() as session: + await session.execute( + insert(Site), + [ + { + "site_id": site_id, + "site_name": f"Site {site_id}", + "site_type": "office", + "capacity_kw": 100.0, + } + for site_id in parc.sites + ], + ) + for site_id in complets: + await session.execute( + insert(Reading), lignes_horaires(site_id, heures=HEURES_COMPLETES, fin=fin) + ) + await session.execute( + insert(Reading), lignes_horaires(partiel, heures=HEURES_INSUFFISANTES, fin=fin) + ) + await session.commit() + + try: + yield parc + finally: + async with get_session_factory()() as session: + await session.execute(delete(Prediction).where(Prediction.site_id.in_(parc.sites))) + await session.execute(delete(Reading).where(Reading.site_id.in_(parc.sites))) + await session.execute(delete(Site).where(Site.site_id.in_(parc.sites))) + await session.commit() + + +@pytest.fixture +def principal_lecteur(app: FastAPI) -> Iterator[None]: + app.dependency_overrides[get_current_principal] = lecteur + yield + app.dependency_overrides.pop(get_current_principal, None) + + +async def resume_du_site(client: AsyncClient, site_id: str) -> dict[str, Any]: + reponse = await client.get("/api/v1/predictions") + + assert reponse.status_code == 200 + sites = reponse.json()["sites"] + return next(site for site in sites if site["site_id"] == site_id) + + +async def entraine_et_score(parc: Parc, tmp_path: Path, *arguments: str) -> Path: + modele = tmp_path / "lightgbm-consumption.txt" + + await executer( + "enervision_ml.train", + "--model-output", + str(modele), + "--mlflow-tracking-uri", + f"sqlite:///{tmp_path}/mlflow.db", + journal=tmp_path, + ) + await executer("enervision_ml.score", "--model", str(modele), *arguments, journal=tmp_path) + + return modele + + +async def test_the_ml_binaries_produce_a_prediction_that_the_api_serves( + parc: Parc, tmp_path: Path, client: AsyncClient, principal_lecteur: None +) -> None: + await entraine_et_score(parc, tmp_path) + + servi = await resume_du_site(client, parc.sites[0]) + + assert servi["prediction"]["status"] == "available" + assert servi["prediction"]["predicted_value"] is not None + assert servi["prediction"]["target_metric"] == "consumption_kwh" + + +async def test_the_api_exposes_the_failure_reason_of_a_site_without_enough_history( + parc: Parc, tmp_path: Path, client: AsyncClient, principal_lecteur: None +) -> None: + await entraine_et_score(parc, tmp_path) + + servi = await resume_du_site(client, parc.sites[-1]) + + assert servi["prediction"]["status"] == "insufficient_data" + assert servi["prediction"]["predicted_value"] is None + assert servi["prediction"]["failure_reason"] is not None + + +async def test_the_api_serves_the_latest_run_when_the_score_cli_runs_twice( + parc: Parc, tmp_path: Path, client: AsyncClient, principal_lecteur: None +) -> None: + modele = await entraine_et_score(parc, tmp_path) + premier = await resume_du_site(client, parc.sites[0]) + + await executer("enervision_ml.score", "--model", str(modele), journal=tmp_path) + + second = await resume_du_site(client, parc.sites[0]) + assert second["prediction"]["created_at"] >= premier["prediction"]["created_at"] + assert second["prediction"]["model_reference"] == premier["prediction"]["model_reference"] diff --git a/ml/tests/__init__.py b/ml/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ml/tests/conftest.py b/ml/tests/conftest.py new file mode 100644 index 0000000..33f65b9 --- /dev/null +++ b/ml/tests/conftest.py @@ -0,0 +1,332 @@ +"""Piege : deux fixtures d'acces a la base, jamais interchangeables - `connexion_ml` et `parc`. + +`connexion_ml` ouvre une transaction annulee a la fin du test : rien ne subsiste, et rien n'est +visible hors de cette connexion. Elle sert aux fonctions qui recoivent leur connexion en +argument (`load_from_database`, `load_recent_from_database`, `write_predictions`). + +`run_scoring` fabrique en revanche son propre engine depuis `ML_DATABASE_URL` : il ne verrait +pas des lignes semees dans une transaction non validee, et ses propres ecritures survivraient a +l'annulation. Les tests qui l'appellent passent donc par `parc`, qui valide ce qu'il ecrit et +nettoie lui-meme, dans l'ordre impose par les cles etrangeres `RESTRICT`. +""" + +import math +import os +from collections.abc import Iterator +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import lightgbm as lgb +import pandas as pd +import pytest +from sqlalchemy import Connection, Engine, Row, bindparam, create_engine, text +from sqlalchemy.engine import URL, make_url + +from enervision_ml.features import TARGET_COLUMN, build_features, feature_columns + +BASE_ATTENDUE = "enervision_test" + +# Piege : `load_from_database` lit toute la table, et `enervision_test` est partagee entre un run +# local et la CI. Les tests ancrent donc leurs lectures au-dela de tout jeu de donnees reel +# (l'historique s'arrete au 31/12/2024) pour que leur borne `since` ne ramene qu'eux. +ANCRAGE = datetime(2035, 1, 1, tzinfo=UTC) + +SITE_TYPE = "office" +CAPACITY_KW = 100.0 + +_INSERT_SITE = text( + """ + INSERT INTO site (site_id, site_name, site_type, capacity_kw) + VALUES (:site_id, :site_name, :site_type, :capacity_kw) + """ +) + +# `source = 'api_history'` impose `dataset_id IS NULL` (ck_reading_dataset_source), ce qui evite +# de creer une ligne `dataset`. `raw_data` est NOT NULL, d'ou le litteral jsonb. +_INSERT_READING = text( + """ + INSERT INTO reading ( + site_id, timestamp, source, consumption_kwh, temperature_celsius, + humidity_percent, solar_irradiance_wm2, is_working_hours, raw_data + ) VALUES ( + :site_id, :timestamp, :source, :consumption_kwh, :temperature_celsius, + :humidity_percent, :solar_irradiance_wm2, :is_working_hours, '{}'::jsonb + ) + """ +) + +_SELECT_PREDICTIONS = text( + """ + SELECT target_at, predicted_value, status, failure_reason, model_reference + FROM prediction + WHERE site_id = :site_id + ORDER BY prediction_id + """ +) + +_INSERT_PREDICTION = text( + """ + INSERT INTO prediction ( + site_id, target_at, target_metric, period_minutes, + predicted_value, model_reference, status, failure_reason + ) VALUES ( + :site_id, :target_at, 'consumption_kwh', 60, + :predicted_value, :model_reference, :status, :failure_reason + ) + """ +) + + +# Ordre impose par les cles etrangeres `RESTRICT` : une lecture avant son site, une prediction +# avant sa lecture. +_SUPPRESSIONS = tuple( + text(requete).bindparams(bindparam("sites", expanding=True)) + for requete in ( + "DELETE FROM prediction WHERE site_id IN :sites", + "DELETE FROM reading WHERE site_id IN :sites", + "DELETE FROM site WHERE site_id IN :sites", + ) +) + + +def insere_site( + connexion: Connection, + *, + site_type: str = SITE_TYPE, + capacity_kw: float | None = CAPACITY_KW, +) -> str: + site_id = f"TEST-{uuid4().hex[:12]}" + connexion.execute( + _INSERT_SITE, + { + "site_id": site_id, + "site_name": "Site de test", + "site_type": site_type, + "capacity_kw": capacity_kw, + }, + ) + return site_id + + +def insere_lectures( + connexion: Connection, + site_id: str, + *, + heures: int, + fin: datetime, + valeur: float = 50.0, + source: str = "api_history", + is_working_hours: bool | None = True, +) -> list[datetime]: + """Grille horaire contigue finissant a `fin`, incluse. + + Contigue parce que les lags de `build_features` sont des `shift()` positionnels : un trou + dans la grille decalerait le lag de 168 h sans qu'aucune erreur ne se declenche. + """ + instants = [fin - timedelta(hours=decalage) for decalage in reversed(range(heures))] + connexion.execute( + _INSERT_READING, + [ + { + "site_id": site_id, + "timestamp": instant, + "source": source, + "consumption_kwh": valeur + math.sin(rang / 12.0) * 10.0, + "temperature_celsius": 15.0, + "humidity_percent": 50.0, + "solar_irradiance_wm2": 0.0, + "is_working_hours": is_working_hours, + } + for rang, instant in enumerate(instants) + ], + ) + return instants + + +def insere_lecture( + connexion: Connection, + site_id: str, + *, + instant: datetime, + consumption_kwh: float | None = 50.0, + source: str = "api_history", + is_working_hours: bool | None = True, +) -> None: + """Une lecture isolee, quand le test pilote sa valeur plutot que sa forme.""" + connexion.execute( + _INSERT_READING, + { + "site_id": site_id, + "timestamp": instant, + "source": source, + "consumption_kwh": consumption_kwh, + "temperature_celsius": 15.0, + "humidity_percent": 50.0, + "solar_irradiance_wm2": 0.0, + "is_working_hours": is_working_hours, + }, + ) + + +def insere_prediction( + connexion: Connection, + site_id: str, + *, + target_at: datetime, + predicted_value: float | None = 42.0, + model_reference: str = "lightgbm-test000000", + status: str = "available", + failure_reason: str | None = None, +) -> None: + connexion.execute( + _INSERT_PREDICTION, + { + "site_id": site_id, + "target_at": target_at, + "predicted_value": predicted_value, + "model_reference": model_reference, + "status": status, + "failure_reason": failure_reason, + }, + ) + + +@pytest.fixture(scope="session") +def url_ml() -> URL: + valeur = os.environ.get("ML_DATABASE_URL") + if not valeur: + pytest.fail("ML_DATABASE_URL absente. Voir `make ml-test-integration`.") + + url = make_url(valeur) + if url.database != BASE_ATTENDUE: + pytest.fail( + f"Ces tests ecrivent et suppriment : ML_DATABASE_URL doit viser {BASE_ATTENDUE}, " + f"pas {url.database}." + ) + return url + + +@pytest.fixture(scope="session") +def moteur_ml(url_ml: URL) -> Iterator[Engine]: + moteur = create_engine(url_ml) + try: + yield moteur + finally: + moteur.dispose() + + +@pytest.fixture +def connexion_ml(moteur_ml: Engine) -> Iterator[Connection]: + with moteur_ml.connect() as connexion: + transaction = connexion.begin() + try: + yield connexion + finally: + transaction.rollback() + + +@dataclass +class Parc: + """Semis valide en base, et son nettoyage, pour les tests qui appellent `run_scoring`. + + Chaque `site_id` porte une marque unique : la base de test est partagee entre un run local + et la CI. + """ + + moteur: Engine + sites: list[str] = field(default_factory=list) + + def site(self, *, site_type: str = SITE_TYPE, capacity_kw: float | None = CAPACITY_KW) -> str: + with self.moteur.begin() as connexion: + site_id = insere_site(connexion, site_type=site_type, capacity_kw=capacity_kw) + self.sites.append(site_id) + return site_id + + def lectures(self, site_id: str, **arguments: Any) -> list[datetime]: + with self.moteur.begin() as connexion: + return insere_lectures(connexion, site_id, **arguments) + + def lecture(self, site_id: str, **arguments: Any) -> None: + with self.moteur.begin() as connexion: + insere_lecture(connexion, site_id, **arguments) + + def prediction(self, site_id: str, **arguments: Any) -> None: + with self.moteur.begin() as connexion: + insere_prediction(connexion, site_id, **arguments) + + def predictions_ecrites(self, site_id: str) -> list[Row[Any]]: + with self.moteur.connect() as connexion: + return list(connexion.execute(_SELECT_PREDICTIONS, {"site_id": site_id})) + + def nettoie(self) -> None: + if not self.sites: + return + + with self.moteur.begin() as connexion: + for suppression in _SUPPRESSIONS: + connexion.execute(suppression, {"sites": self.sites}) + + +@pytest.fixture +def parc(moteur_ml: Engine) -> Iterator[Parc]: + semis = Parc(moteur=moteur_ml) + try: + yield semis + finally: + semis.nettoie() + + +def trame_synthetique(*, sites: int = 2, heures: int = 400) -> pd.DataFrame: + """Lectures horaires deterministes, assez longues pour que le lag de 168 h existe.""" + depart = datetime(2024, 1, 1, tzinfo=UTC) + morceaux = [ + pd.DataFrame( + { + "site_id": f"SITE{numero:03d}", + "timestamp": [depart + timedelta(hours=rang) for rang in range(heures)], + TARGET_COLUMN: [ + 50.0 + 10.0 * math.sin(rang / 12.0) + numero * 5.0 for rang in range(heures) + ], + "temperature_celsius": 15.0, + "humidity_percent": 50.0, + "solar_irradiance_wm2": 0.0, + "is_working_hours": True, + "site_type": SITE_TYPE, + "capacity_kw": CAPACITY_KW, + } + ) + for numero in range(sites) + ] + return pd.concat(morceaux, ignore_index=True) + + +@pytest.fixture(scope="session") +def modele_jetable(tmp_path_factory: pytest.TempPathFactory) -> Path: + """Booster reel entraine sur une trame synthetique, ecrit dans un repertoire temporaire. + + Ni `ml/models/` (ignore par git, et le polluer serait un effet de bord), ni + `enervision_ml.train.train()` (qui journalise dans MLflow sans garde). Le typage `category` + de `site_type` reproduit celui de l'entrainement : c'est le `pandas_categorical` enregistre + dans le modele que `score()` devra retrouver. + """ + features = build_features(trame_synthetique()).dropna(subset=feature_columns()) + typee = features.copy() + typee["site_type"] = typee["site_type"].astype("category") + + donnees = lgb.Dataset( + typee[feature_columns()], + label=typee[TARGET_COLUMN], + categorical_feature=["site_type"], + ) + booster = lgb.train( + {"objective": "regression", "num_leaves": 7, "min_data_in_leaf": 5, "verbosity": -1}, + donnees, + num_boost_round=5, + ) + + chemin = tmp_path_factory.mktemp("modele") / "lightgbm-consumption.txt" + booster.save_model(str(chemin)) + return chemin diff --git a/ml/tests/test_data_integration.py b/ml/tests/test_data_integration.py new file mode 100644 index 0000000..2352a00 --- /dev/null +++ b/ml/tests/test_data_integration.py @@ -0,0 +1,154 @@ +from datetime import timedelta +from pathlib import Path + +import pandas as pd +import pytest +from sqlalchemy import Connection + +from enervision_ml.data import ( + OUTPUT_COLUMNS, + load_from_csv, + load_from_database, + load_recent_from_database, +) +from tests.conftest import ANCRAGE, insere_lecture, insere_lectures, insere_site + +pytestmark = pytest.mark.integration + + +def test_load_from_database_returns_the_nine_contract_columns(connexion_ml: Connection) -> None: + site_id = insere_site(connexion_ml) + insere_lectures(connexion_ml, site_id, heures=3, fin=ANCRAGE) + + frame = load_from_database(connexion_ml) + + assert list(frame.columns) == OUTPUT_COLUMNS + + +def test_load_from_database_joins_the_site_attributes_to_every_reading( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml, site_type="factory", capacity_kw=250.0) + insere_lectures(connexion_ml, site_id, heures=3, fin=ANCRAGE) + + frame = load_from_database(connexion_ml) + + mien = frame[frame["site_id"] == site_id] + assert len(mien) == 3 + assert set(mien["site_type"]) == {"factory"} + assert set(mien["capacity_kw"]) == {250.0} + + +def test_load_recent_from_database_excludes_readings_before_the_since_bound( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + insere_lectures(connexion_ml, site_id, heures=5, fin=ANCRAGE) + + frame = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=2)) + + assert list(frame["timestamp"]) == [ + ANCRAGE - timedelta(hours=2), + ANCRAGE - timedelta(hours=1), + ANCRAGE, + ] + + +def test_load_recent_from_database_includes_a_reading_exactly_at_the_since_bound( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + insere_lecture(connexion_ml, site_id, instant=ANCRAGE) + + frame = load_recent_from_database(connexion_ml, since=ANCRAGE) + + assert len(frame) == 1 + + +def test_load_recent_from_database_keeps_timestamps_timezone_aware( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + insere_lecture(connexion_ml, site_id, instant=ANCRAGE) + + frame = load_recent_from_database(connexion_ml, since=ANCRAGE) + + assert frame["timestamp"].dt.tz is not None + + +def test_load_recent_from_database_orders_readings_by_site_then_timestamp( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + for decalage in (2, 0, 1): + insere_lecture(connexion_ml, site_id, instant=ANCRAGE + timedelta(hours=decalage)) + + frame = load_recent_from_database(connexion_ml, since=ANCRAGE) + + assert list(frame["timestamp"]) == [ + ANCRAGE, + ANCRAGE + timedelta(hours=1), + ANCRAGE + timedelta(hours=2), + ] + + +def test_load_recent_from_database_returns_the_contract_columns_even_without_any_row( + connexion_ml: Connection, +) -> None: + frame = load_recent_from_database(connexion_ml, since=ANCRAGE + timedelta(days=365)) + + assert frame.empty + assert list(frame.columns) == OUTPUT_COLUMNS + + +def test_load_recent_from_database_types_a_fully_null_capacity_kw_as_float64( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml, capacity_kw=None) + insere_lectures(connexion_ml, site_id, heures=3, fin=ANCRAGE) + + frame = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=2)) + + assert frame["capacity_kw"].dtype == "float64" + assert frame["capacity_kw"].isna().all() + + +def test_load_recent_from_database_types_a_null_is_working_hours_as_float64( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + insere_lecture(connexion_ml, site_id, instant=ANCRAGE, is_working_hours=None) + insere_lecture( + connexion_ml, site_id, instant=ANCRAGE + timedelta(hours=1), is_working_hours=True + ) + + frame = load_recent_from_database(connexion_ml, since=ANCRAGE) + + assert frame["is_working_hours"].dtype == "float64" + assert list(frame["is_working_hours"].isna()) == [True, False] + + +def test_both_loaders_produce_the_same_columns_in_the_same_order( + connexion_ml: Connection, tmp_path: Path +) -> None: + site_id = insere_site(connexion_ml) + insere_lectures(connexion_ml, site_id, heures=2, fin=ANCRAGE) + csv_path = tmp_path / "lectures.csv" + pd.DataFrame( + { + "site_id": [site_id], + "timestamp": [ANCRAGE], + "consumption_kwh": [50.0], + "temperature_celsius": [15.0], + "humidity_percent": [50.0], + "solar_irradiance_wm2": [0.0], + "is_working_hours": [True], + "site_type": ["office"], + } + ).to_csv(csv_path, index=False) + + depuis_la_base = load_recent_from_database(connexion_ml, since=ANCRAGE - timedelta(hours=1)) + depuis_le_csv = load_from_csv(csv_path) + + assert list(depuis_la_base.columns) == list(depuis_le_csv.columns) + assert depuis_la_base.dtypes.to_dict() == depuis_le_csv.dtypes.to_dict() diff --git a/ml/tests/test_score_integration.py b/ml/tests/test_score_integration.py new file mode 100644 index 0000000..9859568 --- /dev/null +++ b/ml/tests/test_score_integration.py @@ -0,0 +1,225 @@ +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any + +import pytest +from sqlalchemy import Connection, Row, text +from sqlalchemy.exc import IntegrityError + +from enervision_ml.score import ( + INSUFFICIENT_DATA_REASON, + LOOKBACK, + MAX_STALENESS, + ScoredSite, + model_reference, + run_scoring, + write_predictions, +) +from tests.conftest import ANCRAGE, Parc, insere_site + +pytestmark = pytest.mark.integration + +REFERENCE = "lightgbm-000000000000" + +_SELECT = text( + """ + SELECT target_at, target_metric, period_minutes, predicted_value, + model_reference, status, failure_reason + FROM prediction + WHERE site_id = :site_id + ORDER BY prediction_id + """ +) + + +def lignes(connexion: Connection, site_id: str) -> list[Row[Any]]: + return list(connexion.execute(_SELECT, {"site_id": site_id})) + + +def disponible( + site_id: str, + *, + target_at: datetime = ANCRAGE, + predicted_value: float | None = 12.5, +) -> ScoredSite: + return ScoredSite( + site_id=site_id, + target_at=target_at, + status="available", + predicted_value=predicted_value, + failure_reason=None, + ) + + +def test_write_predictions_inserts_one_row_per_scored_site(connexion_ml: Connection) -> None: + premier = insere_site(connexion_ml) + second = insere_site(connexion_ml) + + write_predictions(connexion_ml, [disponible(premier), disponible(second)], reference=REFERENCE) + + assert len(lignes(connexion_ml, premier)) == 1 + assert len(lignes(connexion_ml, second)) == 1 + + +def test_write_predictions_stores_the_model_reference_and_the_hourly_period( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + + write_predictions(connexion_ml, [disponible(site_id)], reference=REFERENCE) + + ligne = lignes(connexion_ml, site_id)[0] + assert ligne.model_reference == REFERENCE + assert ligne.target_metric == "consumption_kwh" + assert ligne.period_minutes == 60 + + +def test_write_predictions_stacks_a_second_run_instead_of_overwriting_the_first( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + + write_predictions( + connexion_ml, [disponible(site_id, predicted_value=10.0)], reference=REFERENCE + ) + write_predictions( + connexion_ml, [disponible(site_id, predicted_value=20.0)], reference=REFERENCE + ) + + assert [ligne.predicted_value for ligne in lignes(connexion_ml, site_id)] == [10.0, 20.0] + + +def test_write_predictions_writes_nothing_when_no_site_was_scored( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + + write_predictions(connexion_ml, [], reference=REFERENCE) + + assert lignes(connexion_ml, site_id) == [] + + +def test_write_predictions_rejects_an_available_row_without_a_predicted_value( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + + with pytest.raises(IntegrityError, match="ck_prediction_status"): + write_predictions( + connexion_ml, [disponible(site_id, predicted_value=None)], reference=REFERENCE + ) + + +def test_write_predictions_rejects_an_insufficient_data_row_carrying_a_value( + connexion_ml: Connection, +) -> None: + site_id = insere_site(connexion_ml) + incoherent = ScoredSite( + site_id=site_id, + target_at=ANCRAGE, + status="insufficient_data", + predicted_value=12.5, + failure_reason=INSUFFICIENT_DATA_REASON, + ) + + with pytest.raises(IntegrityError, match="ck_prediction_status"): + write_predictions(connexion_ml, [incoherent], reference=REFERENCE) + + +def test_write_predictions_rejects_a_prediction_for_an_unknown_site( + connexion_ml: Connection, +) -> None: + with pytest.raises(IntegrityError, match="fk_prediction_site"): + write_predictions(connexion_ml, [disponible("SITE-INCONNU")], reference=REFERENCE) + + +def test_run_scoring_writes_an_available_prediction_for_a_site_with_a_full_week( + parc: Parc, modele_jetable: Path +) -> None: + site_id = parc.site() + parc.lectures(site_id, heures=200, fin=ANCRAGE) + + run_scoring(model_path=modele_jetable, now=ANCRAGE) + + ligne = parc.predictions_ecrites(site_id)[0] + assert ligne.status == "available" + assert ligne.predicted_value is not None + assert ligne.target_at == ANCRAGE + timedelta(hours=1) + + +def test_run_scoring_writes_insufficient_data_when_the_weekly_lag_is_missing( + parc: Parc, modele_jetable: Path +) -> None: + site_id = parc.site() + parc.lectures(site_id, heures=100, fin=ANCRAGE) + + run_scoring(model_path=modele_jetable, now=ANCRAGE) + + ligne = parc.predictions_ecrites(site_id)[0] + assert ligne.status == "insufficient_data" + assert ligne.predicted_value is None + assert ligne.failure_reason == INSUFFICIENT_DATA_REASON + + +def test_run_scoring_writes_a_staleness_reason_when_the_last_reading_is_too_old( + parc: Parc, modele_jetable: Path +) -> None: + site_id = parc.site() + parc.lectures(site_id, heures=200, fin=ANCRAGE) + + run_scoring(model_path=modele_jetable, now=ANCRAGE + MAX_STALENESS + timedelta(hours=1)) + + ligne = parc.predictions_ecrites(site_id)[0] + assert ligne.status == "insufficient_data" + assert ligne.failure_reason != INSUFFICIENT_DATA_REASON + + +def test_run_scoring_writes_nothing_when_every_reading_is_older_than_the_window( + parc: Parc, modele_jetable: Path +) -> None: + site_id = parc.site() + parc.lectures(site_id, heures=200, fin=ANCRAGE) + + run_scoring(model_path=modele_jetable, now=ANCRAGE + LOOKBACK + timedelta(days=1)) + + assert parc.predictions_ecrites(site_id) == [] + + +def test_run_scoring_only_writes_the_site_that_was_requested( + parc: Parc, modele_jetable: Path +) -> None: + demande = parc.site() + ignore = parc.site() + parc.lectures(demande, heures=200, fin=ANCRAGE) + parc.lectures(ignore, heures=200, fin=ANCRAGE) + + run_scoring(model_path=modele_jetable, site_id=demande, now=ANCRAGE) + + assert len(parc.predictions_ecrites(demande)) == 1 + assert parc.predictions_ecrites(ignore) == [] + + +def test_run_scoring_uses_the_model_file_hash_as_model_reference( + parc: Parc, modele_jetable: Path +) -> None: + site_id = parc.site() + parc.lectures(site_id, heures=200, fin=ANCRAGE) + + run_scoring(model_path=modele_jetable, now=ANCRAGE) + + ligne = parc.predictions_ecrites(site_id)[0] + assert ligne.model_reference == model_reference(modele_jetable) + + +def test_run_scoring_appends_a_second_row_when_it_runs_twice( + parc: Parc, modele_jetable: Path +) -> None: + site_id = parc.site() + parc.lectures(site_id, heures=200, fin=ANCRAGE) + + run_scoring(model_path=modele_jetable, now=ANCRAGE) + run_scoring(model_path=modele_jetable, now=ANCRAGE) + + ecrites = parc.predictions_ecrites(site_id) + assert len(ecrites) == 2 + assert ecrites[0].target_at == ecrites[1].target_at