300 lines
13 KiB
Python
300 lines
13 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import func, Select, Integer
|
|
from sqlmodel import Session, select
|
|
|
|
from models.measurement import IndoorMeasurement, IndoorMeasurementCreateRequest, OutdoorMeasurementCreateRequest, \
|
|
OutdoorMeasurement, MeasurementListResponse, StationMeasurementResponse, MeasurementResponse, MeasurementResolution, \
|
|
MeasurementRecord, RecordResponse, MeasurementRecordItem
|
|
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, List
|
|
|
|
|
|
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],
|
|
]
|
|
)
|
|
|
|
def get_measurement_record(
|
|
session: Session,
|
|
model: Type[IndoorMeasurement | OutdoorMeasurement],
|
|
field: str,
|
|
station_ids: list[int] | None,
|
|
from_timestamp: datetime | None,
|
|
order: str,
|
|
night_only: bool = False,
|
|
) -> MeasurementRecordItem | None:
|
|
column = getattr(model, field)
|
|
statement = select(model)
|
|
|
|
if station_ids is not None:
|
|
statement = statement.where(model.station_id.in_(station_ids))
|
|
|
|
if from_timestamp is not None:
|
|
to_timestamp = datetime.now()
|
|
statement = statement.where(model.timestamp >= from_timestamp)
|
|
statement = statement.where(model.timestamp <= to_timestamp)
|
|
|
|
if night_only:
|
|
hour = func.cast(func.strftime("%H", model.timestamp), Integer)
|
|
statement = statement.where((hour >= 22) | (hour < 6))
|
|
|
|
statement = statement.where(column.is_not(None))
|
|
statement = statement.order_by(column.desc() if order == 'max' else column.asc())
|
|
statement = statement.limit(1)
|
|
|
|
result = session.exec(statement).first()
|
|
if result is None:
|
|
return None
|
|
|
|
station = session.get(Station, result.station_id)
|
|
return MeasurementRecordItem(
|
|
recordValue=getattr(result, field),
|
|
timestamp=result.timestamp,
|
|
station=StationListResponse.model_validate(station),
|
|
)
|
|
|
|
def get_inner_temperature_record(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None, order: str) -> MeasurementRecordItem | None:
|
|
return get_measurement_record(session, IndoorMeasurement, "temperature", station_ids, from_timestamp, order)
|
|
|
|
def get_outer_temperature_record(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None, order: str) -> MeasurementRecordItem | None:
|
|
return get_measurement_record(session, OutdoorMeasurement, "temperature", station_ids, from_timestamp, order)
|
|
|
|
def get_inner_humidity_record(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None, order: str) -> MeasurementRecordItem | None:
|
|
return get_measurement_record(session, IndoorMeasurement, "humidity", station_ids, from_timestamp, order)
|
|
|
|
def get_outer_humidity_record(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None, order: str) -> MeasurementRecordItem | None:
|
|
return get_measurement_record(session, OutdoorMeasurement, "humidity", station_ids, from_timestamp, order)
|
|
|
|
def get_pressure_record(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None, order: str) -> MeasurementRecordItem | None:
|
|
return get_measurement_record(session, OutdoorMeasurement, "pressure", station_ids, from_timestamp, order)
|
|
|
|
def get_warmest_night_record(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None) -> MeasurementRecordItem | None:
|
|
return get_measurement_record(session, OutdoorMeasurement, "temperature", station_ids, from_timestamp, "max", night_only=True)
|
|
|
|
def get_coldest_night_record(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None) -> MeasurementRecordItem | None:
|
|
return get_measurement_record(session, OutdoorMeasurement, "temperature", station_ids, from_timestamp, "min", night_only=True)
|
|
|
|
def get_measurement_records(session: Session, station_ids: list[int] | None, from_timestamp: datetime | None):
|
|
return RecordResponse(
|
|
inner_temperature_record = MeasurementRecord(
|
|
max_value=get_inner_temperature_record(session, station_ids, from_timestamp, order='max'),
|
|
min_value=get_inner_temperature_record(session, station_ids, from_timestamp, order='min')
|
|
),
|
|
outer_temperature_record = MeasurementRecord(
|
|
max_value=get_outer_temperature_record(session, station_ids, from_timestamp, order='max'),
|
|
min_value=get_outer_temperature_record(session, station_ids, from_timestamp, order='min')
|
|
),
|
|
inner_humidity_record = MeasurementRecord(
|
|
max_value=get_inner_humidity_record(session, station_ids, from_timestamp, order='max'),
|
|
min_value=get_inner_humidity_record(session, station_ids, from_timestamp, order='min')
|
|
),
|
|
outer_humidity_record = MeasurementRecord(
|
|
max_value=get_outer_humidity_record(session, station_ids, from_timestamp, order='max'),
|
|
min_value=get_outer_humidity_record(session, station_ids, from_timestamp, order='min')
|
|
),
|
|
pressure_record = MeasurementRecord(
|
|
max_value=get_pressure_record(session, station_ids, from_timestamp, order='max'),
|
|
min_value=get_pressure_record(session, station_ids, from_timestamp, order='min')
|
|
),
|
|
night_temperature_record = MeasurementRecord(
|
|
max_value=get_warmest_night_record(session, None, from_timestamp),
|
|
min_value=get_coldest_night_record(session, None, from_timestamp),
|
|
)
|
|
) |