211 lines
8.0 KiB
Python
211 lines
8.0 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import func
|
|
from sqlmodel import Session, select
|
|
|
|
from models.measurement import IndoorMeasurement, IndoorMeasurementCreateRequest, OutdoorMeasurementCreateRequest, \
|
|
OutdoorMeasurement, MeasurementListResponse, StationMeasurementResponse, MeasurementResponse, MeasurementResolution
|
|
from models.station import Station, StationCreateRequest, StationListResponse
|
|
from services import stationService
|
|
from coolname import generate_slug
|
|
|
|
|
|
def push_indoor_measurement(session: Session, raw_measurement: IndoorMeasurementCreateRequest):
|
|
statement = select(Station).where(Station.mac == raw_measurement.mac)
|
|
station = session.exec(statement).first()
|
|
|
|
if not station:
|
|
station = stationService.create_station(session, StationCreateRequest(
|
|
mac=raw_measurement.mac,
|
|
name=generate_slug(2)
|
|
))
|
|
|
|
measurement = IndoorMeasurement(
|
|
station_id=station.id,
|
|
temperature=raw_measurement.temperature,
|
|
humidity=raw_measurement.humidity
|
|
)
|
|
|
|
session.add(IndoorMeasurement.model_validate(measurement))
|
|
session.commit()
|
|
|
|
def push_outdoor_measurement(session: Session, raw_measurement: OutdoorMeasurementCreateRequest):
|
|
statement = select(Station).where(Station.mac == raw_measurement.mac)
|
|
station = session.exec(statement).first()
|
|
|
|
if not station:
|
|
station = stationService.create_station(session, StationCreateRequest(
|
|
mac=raw_measurement.mac,
|
|
name=generate_slug(2)
|
|
))
|
|
|
|
measurement = OutdoorMeasurement(
|
|
station_id=station.id,
|
|
temperature=raw_measurement.temperature,
|
|
humidity=raw_measurement.humidity,
|
|
pressure=raw_measurement.pressure
|
|
)
|
|
|
|
session.add(OutdoorMeasurement.model_validate(measurement))
|
|
session.commit()
|
|
from typing import Type, Union
|
|
|
|
|
|
def period_format(resolution: MeasurementResolution) -> str:
|
|
return {
|
|
MeasurementResolution.hourly: "%Y-%m-%d %H",
|
|
MeasurementResolution.daily: "%Y-%m-%d",
|
|
MeasurementResolution.weekly: "%Y-%W",
|
|
MeasurementResolution.monthly: "%Y-%m",
|
|
MeasurementResolution.yearly: "%Y",
|
|
}[resolution]
|
|
from sqlalchemy import func
|
|
from sqlmodel import select
|
|
|
|
|
|
def _query_measurements(
|
|
session: Session,
|
|
model: Type[IndoorMeasurement | OutdoorMeasurement],
|
|
indoor: bool,
|
|
station_ids: list[int] | None,
|
|
from_timestamp: datetime | None = None,
|
|
to_timestamp: datetime | None = None,
|
|
limit: int | None = None,
|
|
resolution: MeasurementResolution = MeasurementResolution.hourly,
|
|
) -> list[StationMeasurementResponse]:
|
|
|
|
#
|
|
# RAW
|
|
#
|
|
if resolution == MeasurementResolution.raw:
|
|
statement = select(model)
|
|
|
|
if station_ids:
|
|
statement = statement.where(model.station_id.in_(station_ids))
|
|
|
|
if from_timestamp:
|
|
statement = statement.where(model.timestamp >= from_timestamp)
|
|
|
|
if to_timestamp:
|
|
statement = statement.where(model.timestamp <= to_timestamp)
|
|
|
|
statement = statement.order_by(model.timestamp.desc())
|
|
|
|
if limit:
|
|
statement = statement.limit(limit)
|
|
|
|
rows = session.exec(statement).all()
|
|
|
|
grouped: dict[int, list[MeasurementResponse]] = {}
|
|
|
|
for row in rows:
|
|
grouped.setdefault(row.station_id, []).append(
|
|
MeasurementResponse.model_validate(row)
|
|
)
|
|
|
|
#
|
|
# AGGREGATED
|
|
#
|
|
else:
|
|
period = func.strftime(period_format(resolution), model.timestamp)
|
|
|
|
columns = [
|
|
model.station_id.label("station_id"),
|
|
func.max(model.timestamp).label("timestamp"),
|
|
func.round(func.avg(model.temperature), 1).label("temperature"),
|
|
func.round(func.avg(model.humidity), 1).label("humidity"),
|
|
]
|
|
|
|
if model is OutdoorMeasurement:
|
|
columns.append(
|
|
func.round(func.avg(model.pressure), 1).label("pressure")
|
|
)
|
|
|
|
statement = (
|
|
select(*columns)
|
|
.group_by(model.station_id, period)
|
|
.order_by(func.max(model.timestamp).desc())
|
|
)
|
|
|
|
if station_ids:
|
|
statement = statement.where(model.station_id.in_(station_ids))
|
|
|
|
if from_timestamp:
|
|
statement = statement.where(model.timestamp >= from_timestamp)
|
|
|
|
if to_timestamp:
|
|
statement = statement.where(model.timestamp <= to_timestamp)
|
|
|
|
print(str(statement.compile(compile_kwargs={"literal_binds": True})))
|
|
rows = session.exec(statement).all()
|
|
|
|
grouped: dict[int, list[MeasurementResponse]] = {}
|
|
|
|
for row in rows:
|
|
grouped.setdefault(row.station_id, []).append(
|
|
MeasurementResponse(
|
|
timestamp=row.timestamp,
|
|
temperature=row.temperature,
|
|
humidity=row.humidity,
|
|
pressure=getattr(row, "pressure", None),
|
|
)
|
|
)
|
|
|
|
if limit:
|
|
grouped = {
|
|
station_id: measurements[:limit]
|
|
for station_id, measurements in grouped.items()
|
|
}
|
|
|
|
#
|
|
# Stations laden
|
|
#
|
|
station_ids = list(grouped.keys())
|
|
|
|
stations = {
|
|
s.id: s
|
|
for s in session.exec(
|
|
select(Station).where(Station.id.in_(station_ids))
|
|
)
|
|
}
|
|
|
|
return [
|
|
StationMeasurementResponse(
|
|
station=StationListResponse.model_validate(stations[station_id]),
|
|
measurements=measurements,
|
|
indoor=indoor,
|
|
)
|
|
for station_id, measurements in grouped.items()
|
|
]
|
|
|
|
def get_indoor_measurements(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None = None, to_timestamp: datetime | None = None, limit: int | None = None, resolution: MeasurementResolution = MeasurementResolution.hourly) -> list[StationMeasurementResponse]:
|
|
return _query_measurements(session, IndoorMeasurement, True, station_ids, from_timestamp, to_timestamp, limit, resolution)
|
|
|
|
def get_outdoor_measurements(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None = None, to_timestamp: datetime | None = None, limit: int | None = None, resolution: MeasurementResolution = MeasurementResolution.hourly) -> list[StationMeasurementResponse]:
|
|
return _query_measurements(session, OutdoorMeasurement, False, station_ids, from_timestamp, to_timestamp, limit, resolution)
|
|
|
|
def get_measurements(session: Session, station_ids: list[int] | None, indoor: bool, from_timestamp: datetime | None, to_timestamp: datetime | None,limit: int | None, resolution: MeasurementResolution ):
|
|
if indoor is None:
|
|
indoor_results = get_indoor_measurements(session, station_ids, from_timestamp, to_timestamp, limit, resolution)
|
|
outdoor_results = get_outdoor_measurements(session, station_ids, from_timestamp, to_timestamp, limit, resolution)
|
|
return MeasurementListResponse(
|
|
stations=[
|
|
*[StationMeasurementResponse(station=indoor_result.station, measurements=indoor_result.measurements, indoor=True) for indoor_result in indoor_results],
|
|
*[StationMeasurementResponse(station=outdoor_result.station, measurements=outdoor_result.measurements, indoor=False) for outdoor_result in outdoor_results],
|
|
]
|
|
)
|
|
else:
|
|
if indoor:
|
|
indoor_results = get_indoor_measurements(session, station_ids, from_timestamp, to_timestamp, limit, resolution)
|
|
return MeasurementListResponse(
|
|
stations=[
|
|
*[StationMeasurementResponse(station=indoor_result.station, measurements=indoor_result.measurements, indoor=True) for indoor_result in indoor_results],
|
|
]
|
|
)
|
|
else:
|
|
outdoor_results = get_outdoor_measurements(session, station_ids, from_timestamp, to_timestamp, limit, resolution)
|
|
return MeasurementListResponse(
|
|
stations=[
|
|
*[StationMeasurementResponse(station=outdoor_result.station, measurements=outdoor_result.measurements, indoor=False) for outdoor_result in outdoor_results],
|
|
]
|
|
) |