johanleroy/ENI-projet-piscine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #83 from ineszang/feat/endpoint-sensors-status

Browse Source 
feat(backend): expose GET /api/v1/sensors/status
This commit is contained in:
Johan LEROY
2026-09-17 09:16:24 +02:00
committed by GitHub
parent 9f465538bf 24bf8bf4b9
commit 3692d486c6
11 changed files with 710 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/app/api/deps.py
+8
View File
@@ -32,6 +32,7 @@ from app.repositories.user import UserRepository
from app.services.alert import AlertService
from app.services.auth import AuthService, LoginPolicy
from app.services.recommendation import RecommendationService
from app.services.sensor import SensorService
from app.services.site import SiteService
from app.services.stats import StatsService
from app.services.user import UserService
@@ -167,6 +168,13 @@ def get_stats_service(session: SessionDep) -> StatsService:
StatsServiceDep = Annotated[StatsService, Depends(get_stats_service)]
def get_sensor_service(session: SessionDep) -> SensorService:
return SensorService(sites=SiteRepository(session), readings=ReadingRepository(session))
SensorServiceDep = Annotated[SensorService, Depends(get_sensor_service)]
async def get_current_principal(
credentials: CredentialsDep,
session: SessionDep,
apps/backend/app/api/openapi.py
+4
View File
@@ -71,6 +71,10 @@ TAGS: Final[list[dict[str, Any]]] = [
"description": "Statistiques agrégées de consommation. Accessible à partir du rôle "
"`lecteur`.",
},
{
"name": "sensors",
"description": "État de santé des capteurs par site. Réservé au rôle `admin`.",
},
]
cookie_de_rafraichissement = APIKeyCookie(
apps/backend/app/api/v1/endpoints/sensors.py
+16
View File
@@ -0,0 +1,16 @@
from fastapi import APIRouter
from app.api.deps import AdminDep, SensorServiceDep
from app.schemas.sensor import SensorStatusResponse
router = APIRouter()
@router.get(
"/status",
response_model=SensorStatusResponse,
summary="État de santé des capteurs par site",
)
async def get_status(_: AdminDep, service: SensorServiceDep) -> SensorStatusResponse:
etat = await service.status()
return SensorStatusResponse.model_validate(etat)
apps/backend/app/api/v1/router.py
+4 -1
View File
@@ -1,7 +1,7 @@
from fastapi import APIRouter
from app.api.openapi import REPONSE_SERVEUR, REPONSES_ADMIN, REPONSES_LECTEUR
from app.api.v1.endpoints import alerts, auth, health, recommendations, sites, stats, users
from app.api.v1.endpoints import alerts, auth, health, recommendations, sensors, sites, stats, users
api_router = APIRouter(responses=REPONSE_SERVEUR)
api_router.include_router(health.router, prefix="/health", tags=["health"])
@@ -18,3 +18,6 @@ api_router.include_router(
responses=REPONSES_LECTEUR,
)
api_router.include_router(stats.router, prefix="/stats", tags=["stats"], responses=REPONSES_LECTEUR)
api_router.include_router(
sensors.router, prefix="/sensors", tags=["sensors"], responses=REPONSES_ADMIN
)
apps/backend/app/schemas/sensor.py
+42
View File
@@ -0,0 +1,42 @@
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class SensorDiagnosticResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
status: Literal["ok", "failing"]
since: datetime | None = Field(
description=(
"Horodatage de la dernière lecture reçue pour ce site. Ce n'est pas le début de la "
"panne : l'historique ne permet pas de le dater sans requête supplémentaire."
)
)
class SiteSensorsResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
consumption: SensorDiagnosticResponse
electrical: SensorDiagnosticResponse
temperature: SensorDiagnosticResponse
humidity: SensorDiagnosticResponse
network: SensorDiagnosticResponse
class SiteSensorStatusResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
site_id: str
site_name: str
sensors: SiteSensorsResponse
overall: Literal["ok", "degraded", "critical"]
class SensorStatusResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
timestamp: datetime
sites: list[SiteSensorStatusResponse]
apps/backend/app/services/sensor.py
+137
View File
@@ -0,0 +1,137 @@
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Literal
from app.models.energy import Reading, Site
from app.repositories.reading import ReadingRepository
from app.repositories.site import SiteRepository
CapteurStatus = Literal["ok", "failing"]
OverallStatus = Literal["ok", "degraded", "critical"]
QUALITES_CONNUES: frozenset[str] = frozenset({"good", "partial", "degraded", "critical"})
RAISON_VERS_CAPTEUR: dict[str, str] = {
"consumption_sensor_failure": "consumption",
"electrical_sensor_failure": "electrical",
"temperature_sensor_failure": "temperature",
"humidity_sensor_failure": "humidity",
"network_loss": "network",
}
CHAMPS_PAR_CAPTEUR: dict[str, tuple[str, ...]] = {
"consumption": ("consumption_kw",),
"electrical": ("voltage_v", "current_a", "power_factor"),
"temperature": ("temperature_celsius",),
"humidity": ("humidity_percent",),
}
@dataclass(frozen=True, slots=True)
class DiagnosticCapteur:
status: CapteurStatus
since: datetime | None
@dataclass(frozen=True, slots=True)
class SanteCapteurs:
consumption: DiagnosticCapteur
electrical: DiagnosticCapteur
temperature: DiagnosticCapteur
humidity: DiagnosticCapteur
network: DiagnosticCapteur
@dataclass(frozen=True, slots=True)
class SanteSite:
site_id: str
site_name: str
sensors: SanteCapteurs
overall: OverallStatus
@dataclass(frozen=True, slots=True)
class EtatCapteurs:
timestamp: datetime
sites: list[SanteSite]
class SensorService:
def __init__(self, sites: SiteRepository, readings: ReadingRepository) -> None:
self._sites = sites
self._readings = readings
async def status(self) -> EtatCapteurs:
sites = await self._sites.list_all()
dernieres = {lecture.site_id: lecture for lecture in await self._readings.latest_by_site()}
return EtatCapteurs(
timestamp=datetime.now(UTC),
sites=[_sante_site(site, dernieres.get(site.site_id)) for site in sites],
)
def _sante_site(site: Site, derniere: Reading | None) -> SanteSite:
if derniere is None:
return SanteSite(
site_id=site.site_id,
site_name=site.site_name,
sensors=_tout_en_echec(since=None),
overall="critical",
)
qualite = derniere.data_quality if derniere.data_quality in QUALITES_CONNUES else "critical"
overall = _overall_depuis_qualite(qualite)
if overall == "critical":
return SanteSite(
site_id=site.site_id,
site_name=site.site_name,
sensors=_tout_en_echec(since=derniere.timestamp),
overall="critical",
)
raisons_signalees = {
RAISON_VERS_CAPTEUR[raison]
for raison in (derniere.null_reasons or [])
if raison in RAISON_VERS_CAPTEUR
}
return SanteSite(
site_id=site.site_id,
site_name=site.site_name,
sensors=SanteCapteurs(
consumption=_diagnostic("consumption", derniere, raisons_signalees),
electrical=_diagnostic("electrical", derniere, raisons_signalees),
temperature=_diagnostic("temperature", derniere, raisons_signalees),
humidity=_diagnostic("humidity", derniere, raisons_signalees),
network=_diagnostic("network", derniere, raisons_signalees),
),
overall=overall,
)
def _overall_depuis_qualite(qualite: str) -> OverallStatus:
if qualite == "good":
return "ok"
if qualite in ("partial", "degraded"):
return "degraded"
return "critical"
def _diagnostic(capteur: str, derniere: Reading, raisons_signalees: set[str]) -> DiagnosticCapteur:
champs = CHAMPS_PAR_CAPTEUR.get(capteur, ())
en_echec = capteur in raisons_signalees or any(
getattr(derniere, champ) is None for champ in champs
)
return DiagnosticCapteur(
status="failing" if en_echec else "ok",
since=derniere.timestamp if en_echec else None,
)
def _tout_en_echec(since: datetime | None) -> SanteCapteurs:
echec = DiagnosticCapteur(status="failing", since=since)
return SanteCapteurs(
consumption=echec, electrical=echec, temperature=echec, humidity=echec, network=echec
)
apps/backend/openapi.json
+173
View File
@@ -1227,6 +1227,62 @@
}
]
}
},
"/api/v1/sensors/status": {
"get": {
"tags": [
"sensors"
],
"summary": "État de santé des capteurs par site",
"operationId": "get_status_api_v1_sensors_status_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SensorStatusResponse"
}
}
}
},
"500": {
"description": "Erreur interne. `correlation` identifie la trace côté serveur, qui n'est pas renvoyée au client.",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/InternalErrorResponse"
}
}
}
},
"401": {
"description": "Jeton absent, illisible, périmé, ou rendu caduc par un changement de rôle ou une désactivation. L'en-tête `WWW-Authenticate` porte la cause dans `error=`.",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
},
"403": {
"description": "Droits insuffisants, ou mot de passe provisoire à changer quand `detail` vaut `password_change_required`.",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
}
},
"security": [
{
"Jeton d'accès": []
}
]
}
}
},
"components": {
@@ -1572,6 +1628,59 @@
],
"title": "Role"
},
"SensorDiagnosticResponse": {
"properties": {
"status": {
"type": "string",
"enum": [
"ok",
"failing"
],
"title": "Status"
},
"since": {
"anyOf": [
{
"type": "string",
"format": "date-time"
},
{
"type": "null"
}
],
"title": "Since",
"description": "Horodatage de la dernière lecture reçue pour ce site. Ce n'est pas le début de la panne : l'historique ne permet pas de le dater sans requête supplémentaire."
}
},
"type": "object",
"required": [
"status",
"since"
],
"title": "SensorDiagnosticResponse"
},
"SensorStatusResponse": {
"properties": {
"timestamp": {
"type": "string",
"format": "date-time",
"title": "Timestamp"
},
"sites": {
"items": {
"$ref": "#/components/schemas/SiteSensorStatusResponse"
},
"type": "array",
"title": "Sites"
}
},
"type": "object",
"required": [
"timestamp",
"sites"
],
"title": "SensorStatusResponse"
},
"SiteResponse": {
"properties": {
"site_id": {
@@ -1631,6 +1740,66 @@
],
"title": "SiteResponse"
},
"SiteSensorStatusResponse": {
"properties": {
"site_id": {
"type": "string",
"title": "Site Id"
},
"site_name": {
"type": "string",
"title": "Site Name"
},
"sensors": {
"$ref": "#/components/schemas/SiteSensorsResponse"
},
"overall": {
"type": "string",
"enum": [
"ok",
"degraded",
"critical"
],
"title": "Overall"
}
},
"type": "object",
"required": [
"site_id",
"site_name",
"sensors",
"overall"
],
"title": "SiteSensorStatusResponse"
},
"SiteSensorsResponse": {
"properties": {
"consumption": {
"$ref": "#/components/schemas/SensorDiagnosticResponse"
},
"electrical": {
"$ref": "#/components/schemas/SensorDiagnosticResponse"
},
"temperature": {
"$ref": "#/components/schemas/SensorDiagnosticResponse"
},
"humidity": {
"$ref": "#/components/schemas/SensorDiagnosticResponse"
},
"network": {
"$ref": "#/components/schemas/SensorDiagnosticResponse"
}
},
"type": "object",
"required": [
"consumption",
"electrical",
"temperature",
"humidity",
"network"
],
"title": "SiteSensorsResponse"
},
"SiteSummaryResponse": {
"properties": {
"site_id": {
@@ -1959,6 +2128,10 @@
{
"name": "stats",
"description": "Statistiques agrégées de consommation. Accessible à partir du rôle `lecteur`."
},
{
"name": "sensors",
"description": "État de santé des capteurs par site. Réservé au rôle `admin`."
}
]
}
apps/backend/tests/api/test_openapi.py
+1
View File
@@ -35,6 +35,7 @@ ROUTES_A_ROLE = {
("GET", "/api/v1/recommendations"),
("GET", "/api/v1/recommendations/{recommendation_id}"),
("GET", "/api/v1/stats/summary"),
("GET", "/api/v1/sensors/status"),
}
apps/backend/tests/api/test_sensors.py
+91
View File
@@ -0,0 +1,91 @@
from collections.abc import Callable, Iterator
from datetime import UTC, datetime
from uuid import uuid4
import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from app.api.deps import get_current_principal, get_sensor_service
from app.core.principal import Principal
from app.core.roles import AccountKind, Role
from app.services.sensor import DiagnosticCapteur, EtatCapteurs, SanteCapteurs, SanteSite
TIMESTAMP = datetime(2026, 9, 16, 12, 0, tzinfo=UTC)
def principal(role: Role = Role.ADMIN) -> Principal:
return Principal(
id=uuid4(),
email=f"{role.value}@enervision.fr",
role=role,
kind=AccountKind.HUMAIN,
must_change_password=False,
)
class FauxService:
def __init__(self) -> None:
ok = DiagnosticCapteur(status="ok", since=None)
en_echec = DiagnosticCapteur(status="failing", since=TIMESTAMP)
self.etat = EtatCapteurs(
timestamp=TIMESTAMP,
sites=[
SanteSite(
site_id="SITE001",
site_name="Bureau Paris La Défense",
sensors=SanteCapteurs(
consumption=ok,
electrical=ok,
temperature=en_echec,
humidity=ok,
network=ok,
),
overall="degraded",
)
],
)
async def status(self) -> EtatCapteurs:
return self.etat
@pytest.fixture
def admin_connecte(app: FastAPI) -> Iterator[None]:
app.dependency_overrides[get_current_principal] = lambda: principal()
yield
app.dependency_overrides.pop(get_current_principal, None)
@pytest.fixture
def servi(app: FastAPI, admin_connecte: None) -> Iterator[Callable[[], FauxService]]:
def installe() -> FauxService:
service = FauxService()
app.dependency_overrides[get_sensor_service] = lambda: service
return service
yield installe
app.dependency_overrides.pop(get_sensor_service, None)
async def test_get_status_returns_the_service_result(
servi: Callable[[], FauxService], client: AsyncClient
) -> None:
servi()
response = await client.get("/api/v1/sensors/status")
assert response.status_code == 200
corps = response.json()
assert corps["sites"][0]["site_id"] == "SITE001"
assert corps["sites"][0]["overall"] == "degraded"
assert corps["sites"][0]["sensors"]["temperature"]["status"] == "failing"
assert corps["sites"][0]["sensors"]["consumption"]["status"] == "ok"
async def test_get_status_refuses_a_reader(app: FastAPI, client: AsyncClient) -> None:
app.dependency_overrides[get_current_principal] = lambda: principal(Role.LECTEUR)
response = await client.get("/api/v1/sensors/status")
assert response.status_code == 403
apps/backend/tests/services/test_sensor.py
+224
View File
@@ -0,0 +1,224 @@
from dataclasses import dataclass, field
from datetime import UTC, datetime
from app.services.sensor import SensorService
TIMESTAMP = datetime(2026, 9, 16, 12, 0, tzinfo=UTC)
@dataclass
class FauxSite:
site_id: str
site_name: str
@dataclass
class FauxLecture:
site_id: str
timestamp: datetime
data_quality: str | None
null_reasons: list[str] | None = field(default_factory=list)
consumption_kw: float | None = 10.0
voltage_v: float | None = 230.0
current_a: float | None = 5.0
power_factor: float | None = 0.95
temperature_celsius: float | None = 21.0
humidity_percent: float | None = 40.0
class FauxDepotSites:
def __init__(self, sites: list[FauxSite]) -> None:
self._sites = sites
async def list_all(self) -> list[FauxSite]:
return self._sites
class FauxDepotLectures:
def __init__(self, lectures: list[FauxLecture]) -> None:
self._lectures = lectures
async def latest_by_site(self) -> list[FauxLecture]:
return self._lectures
async def test_status_marks_a_site_without_any_reading_as_critical_with_every_sensor_failing() -> (
None
):
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures([]), # type: ignore[arg-type]
)
etat = await service.status()
site = etat.sites[0]
assert site.overall == "critical"
for capteur in (
site.sensors.consumption,
site.sensors.electrical,
site.sensors.temperature,
site.sensors.humidity,
site.sensors.network,
):
assert capteur.status == "failing"
assert capteur.since is None
async def test_status_marks_every_sensor_ok_on_a_good_quality_reading_with_no_null_field() -> None:
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures([FauxLecture("A", TIMESTAMP, "good")]), # type: ignore[arg-type]
)
etat = await service.status()
site = etat.sites[0]
assert site.overall == "ok"
for capteur in (
site.sensors.consumption,
site.sensors.electrical,
site.sensors.temperature,
site.sensors.humidity,
site.sensors.network,
):
assert capteur.status == "ok"
assert capteur.since is None
async def test_status_flags_the_sensor_named_in_null_reasons() -> None:
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures( # type: ignore[arg-type]
[
FauxLecture(
"A",
TIMESTAMP,
"partial",
null_reasons=["temperature_sensor_failure"],
temperature_celsius=None,
)
]
),
)
etat = await service.status()
site = etat.sites[0]
assert site.overall == "degraded"
assert site.sensors.temperature.status == "failing"
assert site.sensors.temperature.since == TIMESTAMP
assert site.sensors.consumption.status == "ok"
assert site.sensors.electrical.status == "ok"
assert site.sensors.humidity.status == "ok"
assert site.sensors.network.status == "ok"
async def test_status_flags_a_sensor_from_a_null_field_even_without_a_null_reason() -> None:
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures( # type: ignore[arg-type]
[FauxLecture("A", TIMESTAMP, "partial", null_reasons=[], humidity_percent=None)]
),
)
etat = await service.status()
site = etat.sites[0]
assert site.sensors.humidity.status == "failing"
assert site.sensors.humidity.since == TIMESTAMP
async def test_status_flags_electrical_as_failing_when_any_of_its_three_fields_is_null() -> None:
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures( # type: ignore[arg-type]
[FauxLecture("A", TIMESTAMP, "partial", null_reasons=[], power_factor=None)]
),
)
etat = await service.status()
site = etat.sites[0]
assert site.sensors.electrical.status == "failing"
async def test_status_forces_every_sensor_to_failing_when_overall_is_critical() -> None:
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures([FauxLecture("A", TIMESTAMP, "critical", null_reasons=[])]), # type: ignore[arg-type]
)
etat = await service.status()
site = etat.sites[0]
assert site.overall == "critical"
for capteur in (
site.sensors.consumption,
site.sensors.electrical,
site.sensors.temperature,
site.sensors.humidity,
site.sensors.network,
):
assert capteur.status == "failing"
assert capteur.since == TIMESTAMP
async def test_status_treats_an_unknown_data_quality_as_critical() -> None:
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures([FauxLecture("A", TIMESTAMP, None, null_reasons=[])]), # type: ignore[arg-type]
)
etat = await service.status()
assert etat.sites[0].overall == "critical"
async def test_status_ignores_an_unknown_null_reason() -> None:
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures( # type: ignore[arg-type]
[FauxLecture("A", TIMESTAMP, "good", null_reasons=["something_else"])]
),
)
etat = await service.status()
site = etat.sites[0]
assert site.overall == "ok"
for capteur in (
site.sensors.consumption,
site.sensors.electrical,
site.sensors.temperature,
site.sensors.humidity,
site.sensors.network,
):
assert capteur.status == "ok"
async def test_status_flags_network_from_null_reasons_only() -> None:
service = SensorService(
sites=FauxDepotSites([FauxSite("A", "Site A")]), # type: ignore[arg-type]
readings=FauxDepotLectures( # type: ignore[arg-type]
[
FauxLecture(
"A",
TIMESTAMP,
"partial",
null_reasons=["network_loss"],
)
]
),
)
etat = await service.status()
site = etat.sites[0]
assert site.overall == "degraded"
assert site.sensors.network.status == "failing"
assert site.sensors.network.since == TIMESTAMP
assert site.sensors.consumption.status == "ok"
assert site.sensors.electrical.status == "ok"
assert site.sensors.temperature.status == "ok"
assert site.sensors.humidity.status == "ok"