agreg-server/server/backend/adapters/http/_readings_schemas.py

28 lines
754 B
Python
Raw Normal View History

2026-05-10 12:36:38 +02:00
from datetime import datetime
from pydantic import BaseModel
from domain.entities import ConsumptionResponse
class ConsumptionPointSchema(BaseModel):
period: datetime
pulse_count_start: int
pulse_count_end: int
delta_pulses: int
delta_m3: float
class ConsumptionResponseSchema(BaseModel):
dev_eui: str
start: datetime
end: datetime
granularity: str
points: list[ConsumptionPointSchema]
@classmethod
def from_domain(cls, r: ConsumptionResponse) -> "ConsumptionResponseSchema":
return cls(
dev_eui=r.dev_eui,
start=r.start,
end=r.end,
granularity=r.granularity,
points=[ConsumptionPointSchema(**p.__dict__) for p in r.points],
)