Stop the asset page's KPIs counting a day the chart does not show

documentation/changelog.rst

@@ -22,6 +22,8 @@ Infrastructure / Support
 Bugfixes
 -----------
 
+* KPIs on the asset page counted one day more than the selected time range [see `PR #2434 <https://www.github.com/FlexMeasures/flexmeasures/pull/2434>`_]
+* KPIs on the asset page now total the values the chart beside them draws, counting each event under the day it starts in: a sensor reported by several sources counted only one of them, and a revised value was counted on top of the value it revised [see `PR #2434 <https://www.github.com/FlexMeasures/flexmeasures/pull/2434>`_]
 * The time range sent when loading an asset's KPIs was off by the viewer's UTC offset, so KPIs could cover the wrong days [see `PR #2435 <https://www.github.com/FlexMeasures/flexmeasures/pull/2435>`_]
 
 

flexmeasures/api/v3_0/assets.py

@@ -26,7 +26,6 @@
 )
 from flexmeasures.data.services.sensors import (
     build_asset_jobs_data,
-    get_sensor_stats,
 )
 from flexmeasures.api.common.schemas.scheduling import (
     flex_context_schema_openAPI,
@@ -1887,10 +1886,23 @@ def get_kpis(self, id: int, asset: GenericAsset, start, end):
         kpis = []
         for kpi in asset_kpis:
             sensor = Sensor.query.get(kpi["sensor"])
-            sensor_stats = get_sensor_stats(sensor, start, end, sort_keys=False)
+            # The beliefs the chart draws: one value per event, the most recent one.
+            # Aggregating belief rows instead would count a revision on top of what it revised,
+            # and would count each source separately when several report the same sensor.
+            beliefs = sensor.search_beliefs(
+                event_starts_after=start,
+                event_ends_before=end,
+                most_recent_beliefs_only=True,
+            )
+            # Count each event once, under the window it starts in.
+            # The search also returns events that merely overlap the window, which the chart draws,
+            # but a total that included them would count one event under two adjacent selections.
+            event_starts = beliefs.index.get_level_values("event_start")
+            beliefs = beliefs[(event_starts >= start) & (event_starts < end)]
+            values = beliefs["event_value"].dropna()
 
             downsample_function, downsample_value = get_downsample_function_and_value(
-                kpi, sensor, sensor_stats
+                kpi, sensor, values
             )
             kpi_dict = {
                 "title": kpi["title"],

flexmeasures/api/v3_0/tests/test_assets_api.py

@@ -1,11 +1,15 @@
 import json
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 from flask import url_for
 import pytest
 from sqlalchemy import select, func
 
+from pytz import utc
+
 from flexmeasures.data.models.audit_log import AssetAuditLog
+from flexmeasures.data.models.generic_assets import GenericAssetType
+from flexmeasures.data.models.time_series import TimedBelief
 from flexmeasures.data.models.generic_assets import GenericAsset
 from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.services.users import find_user_by_email
@@ -1782,3 +1786,287 @@ def test_get_asset_chart_session_vars_with_canonical_params(
     with client.session_transaction() as sess:
         assert sess.get("event_starts_after") == "2025-05-01T00:00:00+02:00"
         assert sess.get("event_ends_before") == "2025-05-02T00:00:00+02:00"
+
+
+def _asset_with_daily_kpi(
+    db, requesting_user, setup_sources, days: int
+) -> tuple[GenericAsset, datetime]:
+    """Build an asset whose KPI sums a distinct value per day.
+
+    The values differ per day, so that a window covering the wrong days totals differently,
+    rather than merely covering the same number of days.
+    """
+    window_start = datetime(2022, 1, 1, tzinfo=utc)
+    asset_type = (
+        db.session.query(GenericAssetType).filter_by(name="battery").one_or_none()
+    )
+    if asset_type is None:
+        asset_type = GenericAssetType(name="battery")
+        db.session.add(asset_type)
+        db.session.flush()
+    asset = GenericAsset(
+        name=f"kpi window asset ({days} days)",
+        generic_asset_type=asset_type,
+        account_id=requesting_user.account_id,
+    )
+    db.session.add(asset)
+    db.session.flush()
+    sensor = Sensor(
+        name=f"kpi window sensor ({days} days)",
+        generic_asset=asset,
+        event_resolution=timedelta(days=1),
+        unit="MWh",
+    )
+    db.session.add(sensor)
+    db.session.flush()
+    source = list(setup_sources.values())[0]
+    db.session.bulk_insert_mappings(
+        TimedBelief,
+        [
+            dict(
+                event_start=window_start + timedelta(days=day),
+                belief_horizon=timedelta(0),
+                event_value=float(day + 1),
+                sensor_id=sensor.id,
+                source_id=source.id,
+                cumulative_probability=0.5,
+            )
+            for day in range(days)
+        ],
+    )
+    asset.sensors_to_show_as_kpis = [
+        {"title": "Total", "sensor": sensor.id, "function": "sum"}
+    ]
+    db.session.flush()
+    return asset, window_start
+
+
+def _kpi_total(client, asset: GenericAsset, start: str, end: str) -> float:
+    """Ask the KPI endpoint for one window and return its single value."""
+    response = client.get(
+        url_for("AssetAPI:get_kpis", id=asset.id),
+        query_string={"start": start, "end": end},
+    )
+    assert response.status_code == 200, response.json
+    kpis = response.json["data"]
+    assert len(kpis) == 1, f"expected exactly one KPI, got {kpis}"
+    return kpis[0]["downsample_value"]
+
+
+@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True)
+def test_kpi_window_end_is_exclusive(
+    db, client, setup_api_test_data, setup_sources, requesting_user
+):
+    """The KPI window ends before `end`, as the chart's own window does.
+
+    The asset page derives the KPI window from the chart's,
+    so the two have to agree on whether the end is part of the window.
+    """
+    asset, window_start = _asset_with_daily_kpi(
+        db, requesting_user, setup_sources, days=5
+    )
+
+    # Days one to five carry the values 1 to 5, so the first three total six.
+    # Treating the end as inclusive would add the fourth day and total ten.
+    total = _kpi_total(
+        client,
+        asset,
+        window_start.isoformat(),
+        (window_start + timedelta(days=3)).isoformat(),
+    )
+    assert total == pytest.approx(
+        6.0
+    ), "the window must end before its end date, not on it"
+
+
+@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True)
+def test_kpi_window_honours_the_offset_it_is_given(
+    db, client, setup_api_test_data, setup_sources, requesting_user
+):
+    """A window written with a UTC offset names the instants that offset implies.
+
+    The asset page sends its window as local clock times carrying an offset,
+    so the endpoint has to read the offset rather than the clock time alone.
+    """
+    asset, _ = _asset_with_daily_kpi(db, requesting_user, setup_sources, days=6)
+
+    # Midnight UTC on the first, written as one o'clock in +01:00.
+    # These are the first three days, worth 1 + 2 + 3.
+    total = _kpi_total(
+        client, asset, "2022-01-01T01:00:00+01:00", "2022-01-04T01:00:00+01:00"
+    )
+    assert total == pytest.approx(
+        6.0
+    ), "the offset must be read, not just the clock time"
+
+    # The same clock times read as UTC name instants an hour later,
+    # which drops the first day and picks up the fourth: 2 + 3 + 4.
+    shifted = _kpi_total(
+        client, asset, "2022-01-01T01:00:00+00:00", "2022-01-04T01:00:00+00:00"
+    )
+    assert shifted == pytest.approx(
+        9.0
+    ), "an hour later is a different set of days, so the two windows must not agree"
+    assert total != shifted, "the assertion above only means something if these differ"
+
+
+@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True)
+def test_kpi_reports_what_the_chart_draws(
+    db, client, setup_api_test_data, setup_sources, requesting_user
+):
+    """A KPI is read beside the chart, so it must describe the same beliefs.
+
+    Two ways that used to diverge: several sources reporting one sensor were counted separately,
+    and a revised belief was counted on top of the belief it revised.
+    """
+    asset_type = (
+        db.session.query(GenericAssetType).filter_by(name="battery").one_or_none()
+    )
+    asset = GenericAsset(
+        name="kpi agrees with chart",
+        generic_asset_type=asset_type,
+        account_id=requesting_user.account_id,
+    )
+    db.session.add(asset)
+    db.session.flush()
+    sensor = Sensor(
+        name="kpi agrees with chart sensor",
+        generic_asset=asset,
+        event_resolution=timedelta(days=1),
+        unit="EUR",
+    )
+    db.session.add(sensor)
+    db.session.flush()
+
+    sources = list(setup_sources.values())
+    scheduled, uploaded = sources[0], sources[-1]
+    assert scheduled.id != uploaded.id, "this test needs two distinct sources"
+
+    window_start = datetime(2030, 1, 15, tzinfo=utc)
+    db.session.bulk_insert_mappings(
+        TimedBelief,
+        [
+            # One day per source, as when a single point is uploaded beside computed data.
+            dict(
+                event_start=window_start,
+                belief_horizon=timedelta(0),
+                event_value=122.0,
+                sensor_id=sensor.id,
+                source_id=scheduled.id,
+                cumulative_probability=0.5,
+            ),
+            dict(
+                event_start=window_start + timedelta(days=1),
+                belief_horizon=timedelta(0),
+                event_value=100.0,
+                sensor_id=sensor.id,
+                source_id=uploaded.id,
+                cumulative_probability=0.5,
+            ),
+            # A third day believed twice, the later belief revising the earlier one.
+            dict(
+                event_start=window_start + timedelta(days=2),
+                belief_horizon=timedelta(days=2),
+                event_value=50.0,
+                sensor_id=sensor.id,
+                source_id=scheduled.id,
+                cumulative_probability=0.5,
+            ),
+            dict(
+                event_start=window_start + timedelta(days=2),
+                belief_horizon=timedelta(days=1),
+                event_value=7.0,
+                sensor_id=sensor.id,
+                source_id=scheduled.id,
+                cumulative_probability=0.5,
+            ),
+        ],
+    )
+    asset.sensors_to_show_as_kpis = [
+        {"title": "Daily costs", "sensor": sensor.id, "function": "sum"}
+    ]
+    db.session.flush()
+
+    window_end = window_start + timedelta(days=3)
+    total = _kpi_total(client, asset, window_start.isoformat(), window_end.isoformat())
+
+    drawn = sensor.search_beliefs(
+        event_starts_after=window_start,
+        event_ends_before=window_end,
+        most_recent_beliefs_only=True,
+    )
+    starts = drawn.index.get_level_values("event_start")
+    within = drawn[(starts >= window_start) & (starts < window_end)]
+    assert total == pytest.approx(
+        float(within["event_value"].sum())
+    ), "the KPI must total the values the chart draws, for the events this window owns"
+    assert total == pytest.approx(
+        229.0
+    ), "122 from one source, 100 from another, and 7 revising 50, is 229"
+
+
+@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True)
+def test_kpi_counts_each_event_under_one_day_only(
+    db, client, setup_api_test_data, setup_sources, requesting_user
+):
+    """A day's KPI covers the events that day owns, not those merely overlapping it.
+
+    A daily sensor on the UTC grid, read from a timezone an hour ahead, has every event
+    straddling the boundary between two local days.
+    Counting an event under both would total it twice across neighbouring selections.
+    """
+    asset_type = (
+        db.session.query(GenericAssetType).filter_by(name="battery").one_or_none()
+    )
+    asset = GenericAsset(
+        name="kpi owns its events",
+        generic_asset_type=asset_type,
+        account_id=requesting_user.account_id,
+    )
+    db.session.add(asset)
+    db.session.flush()
+    sensor = Sensor(
+        name="kpi owns its events sensor",
+        generic_asset=asset,
+        event_resolution=timedelta(days=1),
+        unit="EUR",
+        timezone="UTC",
+    )
+    db.session.add(sensor)
+    db.session.flush()
+    source = list(setup_sources.values())[0]
+
+    first = datetime(2030, 1, 15, tzinfo=utc)
+    db.session.bulk_insert_mappings(
+        TimedBelief,
+        [
+            dict(
+                event_start=first + timedelta(days=offset),
+                belief_horizon=timedelta(0),
+                event_value=value,
+                sensor_id=sensor.id,
+                source_id=source.id,
+                cumulative_probability=0.5,
+            )
+            for offset, value in ((0, 122.0), (1, 100.0))
+        ],
+    )
+    asset.sensors_to_show_as_kpis = [
+        {"title": "Daily costs", "sensor": sensor.id, "function": "sum"}
+    ]
+    db.session.flush()
+
+    # Local days in +01:00, so each runs from 23:00 UTC to 23:00 UTC and straddles both events.
+    def local_day(day: int) -> str:
+        return f"2030-01-{day:02d}T00:00:00+01:00"
+
+    totals = {
+        day: _kpi_total(client, asset, local_day(day), local_day(day + 1))
+        for day in (15, 16, 17)
+    }
+    assert totals[15] == pytest.approx(122.0), "the 15th owns only its own event"
+    assert totals[16] == pytest.approx(100.0), "the 16th must not also count the 15th"
+    assert totals[17] == pytest.approx(0.0), "the 17th owns nothing"
+    assert sum(totals.values()) == pytest.approx(
+        222.0
+    ), "each event counts once across neighbouring days, not twice"

flexmeasures/data/utils.py

@@ -273,26 +273,31 @@ def save_to_db(
     return status
 
 
-def get_downsample_function_and_value(
-    kpi: dict, sensor: Sensor, sensor_stats: dict
-) -> tuple:
+def get_downsample_function_and_value(kpi: dict, sensor: Sensor, values) -> tuple:
+    """Reduce a sensor's values over a window to the single number a KPI shows.
+
+    :param kpi:     The `sensors_to_show_as_kpis` entry, which may name a function.
+    :param sensor:  The sensor the KPI describes, whose unit decides the default function.
+    :param values:  One value per event, as the chart draws them, rather than one row per belief.
+    :returns:       The function used, and the value it produced.
+    """
     downsample_function = kpi.get("function", None)
     if downsample_function is None:
         if sensor.unit == "%":
             downsample_function = "mean"
         else:
             downsample_function = "sum"
-    try:
-        if downsample_function == "mean":
-            downsample_value = dict(next(iter(sensor_stats.values())))["Mean value"]
-        elif downsample_function == "max":
-            downsample_value = dict(next(iter(sensor_stats.values())))["Max value"]
-        elif downsample_function == "min":
-            downsample_value = dict(next(iter(sensor_stats.values())))["Min value"]
-        else:
-            downsample_value = dict(next(iter(sensor_stats.values())))[
-                "Sum over values"
-            ]
-    except StopIteration:
-        downsample_value = 0
+
+    # An empty window has nothing to reduce, and sum() over none of it is not a KPI of zero cost.
+    if len(values) == 0:
+        return downsample_function, 0
+
+    if downsample_function == "mean":
+        downsample_value = values.mean()
+    elif downsample_function == "max":
+        downsample_value = values.max()
+    elif downsample_function == "min":
+        downsample_value = values.min()
+    else:
+        downsample_value = values.sum()
     return downsample_function, downsample_value