flexmeasures-main/flexmeasures/data/models/planning/tests/utils.py

from __future__ import annotations

import pandas as pd

from flexmeasures.data.models.time_series import Sensor
from flexmeasures.tests.utils import get_test_sensor
from flexmeasures.utils.calculations import integrate_time_series
from flexmeasures.utils.unit_utils import ur


def series_to_ts_specs(s: pd.Series, unit: str) -> list[dict]:
    """Assumes the series frequency should be used as the event resolution."""
    return [
        {
            "start": i.isoformat(),
            "duration": pd.to_timedelta(s.index.freq).isoformat(),
            "value": f"{s[i]} {unit}",
        }
        for i in s.index
    ]


def check_constraints(
    sensor: Sensor,
    schedule: pd.Series,
    soc_at_start: float,
    roundtrip_efficiency: float = 1,
    storage_efficiency: float = 1,
    tolerance: float = 0.00001,
) -> pd.Series:
    soc_schedule = integrate_time_series(
        schedule,
        soc_at_start,
        up_efficiency=roundtrip_efficiency**0.5,
        down_efficiency=roundtrip_efficiency**0.5,
        storage_efficiency=storage_efficiency,
        decimal_precision=6,
    )
    with pd.option_context("display.max_rows", None, "display.max_columns", 3):
        print(soc_schedule)
    capacity = sensor.get_attribute(
        "capacity_in_mw",
        ur.Quantity(sensor.get_attribute("site-power-capacity")).to("MW").magnitude,
    )
    assert min(schedule.values) >= capacity * -1 - tolerance
    assert max(schedule.values) <= capacity + tolerance
    for soc in soc_schedule.values:
        assert soc >= sensor.get_attribute("min_soc_in_mwh")
        assert soc <= sensor.get_attribute("max_soc_in_mwh")
    return soc_schedule


def get_sensors_from_db(
    db, battery_assets, battery_name="Test battery", power_sensor_name="power"
):
    # get the sensors from the database
    epex_da = get_test_sensor(db)
    battery = [
        s for s in battery_assets[battery_name].sensors if s.name == power_sensor_name
    ][0]
    assert battery.get_attribute("consumption-price") == {"sensor": epex_da.id}

    return epex_da, battery