flexmeasures-main/flexmeasures/data/services/timerange.py

from __future__ import annotations

from datetime import datetime

from sqlalchemy import func, select

from flexmeasures.data import db
from flexmeasures.utils import time_utils


def get_timerange(sensor_ids: list[int]) -> tuple[datetime, datetime]:
    """Get the start and end of the least recent and most recent event, respectively.

    In case of no data, defaults to (now, now).
    """
    from flexmeasures.data.models.time_series import Sensor, TimedBelief

    least_recent_event_start_and_most_recent_event_end = db.session.execute(
        select(
            func.min(TimedBelief.event_start),
            func.max(TimedBelief.event_start + Sensor.event_resolution),
        )
        .select_from(TimedBelief)
        .join(Sensor, TimedBelief.sensor_id == Sensor.id)
        .filter(TimedBelief.sensor_id.in_(sensor_ids))
    ).one_or_none()
    if least_recent_event_start_and_most_recent_event_end == (None, None):
        # return now in case there is no data for any of the sensors
        now = time_utils.server_now()
        return now, now
    return least_recent_event_start_and_most_recent_event_end