From b4d0b306dbbd6713694051fda5a79e289a865a5a Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 15:03:23 +0200 Subject: [PATCH 01/14] Count an event once in a KPI, when several sources report it Two sources reporting one event are two claims about it, not two contributions to it, so adding them up produced a number no source ever reported, and that no point on the chart showed. A sensor with a forecast later corrected by an upload, or with two forecasters configured differently, totalled both. The KPI query now asks for one deterministic belief per event, which prefers the latest source version, and the most recent belief within it. `most_recent_beliefs_only` alone did not do this: it is per source, and `use_latest_version_per_event` only collapses sources sharing a name, type and model, and only within one belief time. The chart beside the KPI still draws every source, so it can show more points than the KPI counted, which the KPI documentation now says. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Lp1bUhWjEQtyDbnvRZQgQs Signed-off-by: F.N. Claessen --- documentation/views/asset-data.rst | 4 ++ flexmeasures/api/v3_0/assets.py | 10 ++- .../api/v3_0/tests/test_assets_api.py | 71 +++++++++++++++++++ 3 files changed, 82 insertions(+), 3 deletions(-) diff --git a/documentation/views/asset-data.rst b/documentation/views/asset-data.rst index 6ccf0f55ae..7dbba1596a 100644 --- a/documentation/views/asset-data.rst +++ b/documentation/views/asset-data.rst @@ -131,6 +131,10 @@ Currently, this supports only a daily resolution (which fits the date picker on So you will need a sensor with daily resolution (probably generated with FlexMeasures' reporting tooling). From this data, you can display summed totals, means, max or min values (the image above shows two KPIs with totals). +The function is applied to one value per event. +Where several data sources reported the same event, the value is the one from the latest source version, and from the most recent belief within that, rather than each source's value in turn. +The chart beside the KPI still draws every source, so it can show more points than the KPI counted. + We aim to support a graphical tool to edit these KPIs in the future. For now, you can set them by editing the asset's `kpi_sensors_to_show` field in the properties page, which will validate that the format is correct and tell you what to change. Read more about the format below. diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index 33fe92af26..9a9abf78de 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -2098,13 +2098,17 @@ def get_kpis(self, id: int, asset: GenericAsset, start, end): kpis = [] for kpi in asset_kpis: sensor = Sensor.query.get(kpi["sensor"]) - # 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. + # One value per event, which is what a KPI reduces. + # Aggregating belief rows instead would count a revision on top of the belief it revised, + # and would count each source separately when several report the same event, + # so that a total came out higher than anything anyone reported. + # Where several do report an event, the value is the one from the latest source version, + # and from the most recent belief within that. beliefs = sensor.search_beliefs( event_starts_after=start, event_ends_before=end, most_recent_beliefs_only=True, + one_deterministic_belief_per_event=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, diff --git a/flexmeasures/api/v3_0/tests/test_assets_api.py b/flexmeasures/api/v3_0/tests/test_assets_api.py index 4bd78a2cc3..92489d7d8d 100644 --- a/flexmeasures/api/v3_0/tests/test_assets_api.py +++ b/flexmeasures/api/v3_0/tests/test_assets_api.py @@ -1910,6 +1910,77 @@ def test_kpi_window_honours_the_offset_it_is_given( 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_counts_an_event_once_when_two_sources_report_it( + db, client, setup_api_test_data, setup_sources, requesting_user +): + """Two sources reporting one event are two claims about it, not two contributions to it. + + Summing them produced a number no source ever reported, and that no point on the chart showed. + The KPI now reduces one value per event, preferring the latest source version and the most recent belief in it. + """ + asset_type = ( + db.session.query(GenericAssetType).filter_by(name="battery").one_or_none() + ) + asset = GenericAsset( + name="kpi with two sources on one event", + generic_asset_type=asset_type, + account_id=requesting_user.account_id, + ) + db.session.add(asset) + db.session.flush() + sensor = Sensor( + name="kpi with two sources sensor", + generic_asset=asset, + event_resolution=timedelta(days=1), + unit="EUR", + ) + db.session.add(sensor) + db.session.flush() + + sources = list(setup_sources.values()) + reported, corrected = sources[0], sources[-1] + assert reported.id != corrected.id, "this test needs two distinct sources" + + window_start = datetime(2030, 3, 15, tzinfo=utc) + db.session.bulk_insert_mappings( + TimedBelief, + [ + # One event, claimed by two sources, the second more recently than the first. + dict( + event_start=window_start, + belief_horizon=timedelta(days=2), + event_value=100.0, + sensor_id=sensor.id, + source_id=reported.id, + cumulative_probability=0.5, + ), + dict( + event_start=window_start, + belief_horizon=timedelta(days=1), + event_value=80.0, + sensor_id=sensor.id, + source_id=corrected.id, + cumulative_probability=0.5, + ), + ], + ) + asset.sensors_to_show_as_kpis = [ + {"title": "Daily costs", "sensor": sensor.id, "function": "sum"} + ] + db.session.flush() + + total = _kpi_total( + client, + asset, + window_start.isoformat(), + (window_start + timedelta(days=1)).isoformat(), + ) + assert total == pytest.approx( + 80.0 + ), "the more recent belief about the event, rather than 180.0, which neither source reported" + + @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 From d9423d1156d50f1289b5c134e16f71d6d3740b1f Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 15:04:21 +0200 Subject: [PATCH 02/14] docs/changelog: record the KPI fix Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Lp1bUhWjEQtyDbnvRZQgQs Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 1 + 1 file changed, 1 insertion(+) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 3940b4edac..501d43c73b 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -30,6 +30,7 @@ Infrastructure / Support Bugfixes ----------- +* A KPI on the asset page counted an event once per data source that reported it, so a total could come out higher than any source reported; it now reduces one value per event, from the latest source version and the most recent belief within it [see `PR #2472 `_] * Sensor data ingestion now preserves ``null`` gaps when converting posted values to the sensor's unit, instead of failing the request [see `PR #2461 `_] * KPIs on the asset page counted one day more than the selected time range [see `PR #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 `_] From 097070f0b52c6172126fe5d3fb7bf24ad06cbc2c Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 20:47:39 +0200 Subject: [PATCH 03/14] tests: cover the version priority a KPI applies, and say where ties stop The regression test only varied the belief time, while the docstring and the documentation also claimed a preference for the latest source version, which the fixture's unversioned sources could not show. A second test now gives one reporter two versions and has the newer one speak first, so the KPI answers with the newer version's value despite the older belief time. Ties beyond that are left alone, and `_select_latest_version_and_belief_per_event` now says why: beliefs which tie on version and belief time keep the order they came in, which is how a caller expresses its own precedence by the order it passes its sources. `test_source_transition` documents and relies on that. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LtMZ49GH6LaiY5MdgtfNe2 Signed-off-by: F.N. Claessen --- .../api/v3_0/tests/test_assets_api.py | 80 ++++++++++++++++++- flexmeasures/data/models/time_series.py | 4 + 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/flexmeasures/api/v3_0/tests/test_assets_api.py b/flexmeasures/api/v3_0/tests/test_assets_api.py index 92489d7d8d..07362d5eee 100644 --- a/flexmeasures/api/v3_0/tests/test_assets_api.py +++ b/flexmeasures/api/v3_0/tests/test_assets_api.py @@ -1917,7 +1917,8 @@ def test_kpi_counts_an_event_once_when_two_sources_report_it( """Two sources reporting one event are two claims about it, not two contributions to it. Summing them produced a number no source ever reported, and that no point on the chart showed. - The KPI now reduces one value per event, preferring the latest source version and the most recent belief in it. + The KPI now reduces one value per event, and these two sources are of the same version, + so the one that believed the event more recently is the one it counts. """ asset_type = ( db.session.query(GenericAssetType).filter_by(name="battery").one_or_none() @@ -1981,6 +1982,83 @@ def test_kpi_counts_an_event_once_when_two_sources_report_it( ), "the more recent belief about the event, rather than 180.0, which neither source reported" +@pytest.mark.parametrize("requesting_user", ["test_admin_user@seita.nl"], indirect=True) +def test_kpi_prefers_the_latest_source_version_over_the_most_recent_belief( + db, client, setup_api_test_data, requesting_user +): + """A newer version of a source wins the event, even when an older version believed it more recently. + + Version comes first because it says which code produced the value, + where the belief time only says when it was said. + """ + from flexmeasures.data.models.data_sources import DataSource + + asset_type = ( + db.session.query(GenericAssetType).filter_by(name="battery").one_or_none() + ) + asset = GenericAsset( + name="kpi with two source versions", + generic_asset_type=asset_type, + account_id=requesting_user.account_id, + ) + db.session.add(asset) + db.session.flush() + sensor = Sensor( + name="kpi with two source versions sensor", + generic_asset=asset, + event_resolution=timedelta(days=1), + unit="EUR", + ) + db.session.add(sensor) + # Two versions of one reporter, which is what a release upgrade leaves behind. + older_version = DataSource( + name="Reporter", type="reporter", model="Rep", version="1" + ) + newer_version = DataSource( + name="Reporter", type="reporter", model="Rep", version="2" + ) + db.session.add_all([older_version, newer_version]) + db.session.flush() + + window_start = datetime(2030, 4, 15, tzinfo=utc) + db.session.bulk_insert_mappings( + TimedBelief, + [ + # The newer version spoke first, and the older version spoke later. + dict( + event_start=window_start, + belief_horizon=timedelta(days=2), + event_value=42.0, + sensor_id=sensor.id, + source_id=newer_version.id, + cumulative_probability=0.5, + ), + dict( + event_start=window_start, + belief_horizon=timedelta(days=1), + event_value=99.0, + sensor_id=sensor.id, + source_id=older_version.id, + cumulative_probability=0.5, + ), + ], + ) + asset.sensors_to_show_as_kpis = [ + {"title": "Daily costs", "sensor": sensor.id, "function": "sum"} + ] + db.session.flush() + + total = _kpi_total( + client, + asset, + window_start.isoformat(), + (window_start + timedelta(days=1)).isoformat(), + ) + assert total == pytest.approx( + 42.0 + ), "the newer version's value, despite the older belief time" + + @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 diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index 196fc1cd6f..51e1d87829 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -902,6 +902,10 @@ def _select_latest_version_and_belief_per_event( """Keep, per event, the single belief with the latest source version, breaking version ties by most recent belief time. + Beliefs that tie on both keep the order they came in, which is what lets a caller + express its own precedence by the order in which it passes its sources. + See `test_source_transition`, where the first source in the list wins the events both sources report. + Assumes deterministic beliefs (probabilistic depth 1) and a belief_time index level. """ source_codes, unique_sources = pd.factorize(bdf.index.get_level_values("source")) From da9ebb9b1f5f8ecf945bc977d784be99d1c149d7 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Thu, 3 Sep 2026 20:53:22 +0200 Subject: [PATCH 04/14] docs: reflow the tie-order docstring to break after punctuation Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LtMZ49GH6LaiY5MdgtfNe2 Signed-off-by: F.N. Claessen --- flexmeasures/data/models/time_series.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index 51e1d87829..a22a3d9c62 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -902,8 +902,8 @@ def _select_latest_version_and_belief_per_event( """Keep, per event, the single belief with the latest source version, breaking version ties by most recent belief time. - Beliefs that tie on both keep the order they came in, which is what lets a caller - express its own precedence by the order in which it passes its sources. + Beliefs that tie on both keep the order they came in, + which is what lets a caller express its own precedence by the order in which it passes its sources. See `test_source_transition`, where the first source in the list wins the events both sources report. Assumes deterministic beliefs (probabilistic depth 1) and a belief_time index level. From 7cbbe70c89a62d1b85c71ae08a51468c8c669209 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 11:25:38 +0200 Subject: [PATCH 05/14] Settle which source wins an event with one rule, in two steps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A version number orders the releases of one source and says nothing about a different one, but the choice between sources ranked every version in the frame together, so a forecaster at v9 outranked a scheduler at v1 on the strength of the number alone, against a fresher belief and against a caller that had asked for the scheduler. Sources are now grouped into families sharing a name, type and model, and versions are only compared inside one. Between families, the order in which a caller passed its sources decides. That is what the `AggregatorReporter` has documented since #819 — "the first source defined in the sources array is prioritized" — and what it never did: the winner was whichever source name sorted first, so `test_source_transition` passed because "source1" sorts before "source2", and reversing the list changed nothing. It does now, which that test also checks. What neither settles falls to the most recent belief, and then to the highest source id, which does not move when a source is renamed. The probabilistic path used to repeat the same ranking in pandas, with the same cross-family bug, so it now makes its beliefs deterministic and hands them to the one implementation. `keep_latest_version` is skipped when one belief per event is asked for, since that path settles versions itself, family by family, and it would otherwise drop a fresher belief before the choice was made. Closes #2476. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 1 + .../models/reporting/tests/test_aggregator.py | 12 ++ flexmeasures/data/models/time_series.py | 163 +++++++++++------ .../data/tests/test_search_postprocessing.py | 171 ++++++++++++++++-- 4 files changed, 270 insertions(+), 77 deletions(-) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 955f975631..8807ca5aa2 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -39,6 +39,7 @@ Infrastructure / Support Bugfixes ----------- +* Which data source wins an event, when several report it and one value per event is asked for, now follows one rule: a source's version is only compared within a family of sources sharing a name, type and model, where it used to be compared between unrelated sources, so a forecaster at v9 no longer outranks a scheduler at v1; between families, the order in which a caller passed its sources decides, which is what the ``AggregatorReporter`` has documented all along; and a tie neither settles falls to the most recent belief and then the highest source id, rather than to the source whose name sorts first [see `PR #2483 `_] * A KPI on the asset page counted an event once per data source that reported it, so a total could come out higher than any source reported; it now reduces one value per event, from the latest source version and the most recent belief within it [see `PR #2472 `_] * ``flexmeasures add schedule --dry-run`` no longer saves a schedule when it is combined with ``--as-job``, where the flag used to be dropped without a word and the queued job stored its schedule anyway; that combination is now rejected, and a dry run says how many beliefs it would have saved and which events they cover [see `PR #2483 `_] * Upgrading a database old enough to still carry the pre-``GenericAsset``/``Sensor`` tables now works, where the v0.18.0 migration that removes them crashed twice over: once while checking whether those tables hold data, as soon as one of them held more than a single row, and once while dropping them, because it dropped each table before the ones referencing it [see `PR #2475 `_] diff --git a/flexmeasures/data/models/reporting/tests/test_aggregator.py b/flexmeasures/data/models/reporting/tests/test_aggregator.py index 044c9fa08b..e32f980fa2 100644 --- a/flexmeasures/data/models/reporting/tests/test_aggregator.py +++ b/flexmeasures/data/models/reporting/tests/test_aggregator.py @@ -189,6 +189,18 @@ def test_source_transition(setup_dummy_data, db): ) # the data from the first source is used assert (result[13:] == -1).all().event_value + # Naming the sources the other way round hands the overlapping event to the other source, + # which is what "the first source defined in the sources array" means. + reversed_result = agg_reporter.compute( + start=tz.localize(datetime(2023, 4, 24)), + end=tz.localize(datetime(2023, 4, 25)), + input=[dict(sensor=s3, sources=[ds2, ds1])], + output=[dict(sensor=report_sensor)], + belief_time=tz.localize(datetime(2023, 12, 1)), + )[0]["data"] + assert (reversed_result[:12] == 1).all().event_value + assert (reversed_result[12:] == -1).all().event_value + # only considering DataSource 1 result = agg_reporter.compute( start=tz.localize(datetime(2023, 4, 24)), diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index a22a3d9c62..e84eda95aa 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -431,7 +431,7 @@ def search_beliefs( # noqa: C901 :param beliefs_before: only return beliefs formed before this datetime (inclusive) :param horizons_at_least: only return beliefs with a belief horizon equal or greater than this timedelta (for example, use timedelta(0) to get ante knowledge time beliefs) :param horizons_at_most: only return beliefs with a belief horizon equal or less than this timedelta (for example, use timedelta(0) to get post knowledge time beliefs) - :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources. Without this set and a most recent parameter used (see below), the results can be of any source. + :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources. Without this set and a most recent parameter used (see below), the results can be of any source. Where a list is given, its order says which source to prefer for an event that several of them report, which matters when asking for one deterministic belief per event. :param user_source_ids: Optional list of user source ids to query only specific user sources :param source_account_ids: Optional account ID (or list thereof) to query only sources linked to specific accounts :param source_types: Optional list of source type names to query only specific source types * @@ -440,7 +440,7 @@ def search_beliefs( # noqa: C901 :param most_recent_beliefs_only: only return the most recent beliefs for each event from each source (minimum belief horizon). Defaults to True. :param most_recent_events_only: only return (post knowledge time) beliefs for the most recent event (maximum event start). Defaults to False. :param most_recent_only: only return a single belief, the most recent from the most recent event. Fastest method if you only need one. Defaults to False. Setting this to True will turn off usage of most_recent_beliefs_only and most_recent_events_only. Use with care when data uses cumulative probability (more than one belief per event_start and horizon). - :param one_deterministic_belief_per_event: only return a single value per event (no probabilistic distribution and only 1 source) + :param one_deterministic_belief_per_event: only return a single value per event (no probabilistic distribution and only 1 source). Where several sources report an event, the one to keep is chosen by version within a family of sources sharing a name, type and model, and between families by the order the sources were passed in, then by the most recent belief, then by the highest source id. :param one_deterministic_belief_per_event_per_source: only return a single value per event per source (no probabilistic distribution) :param as_json: return beliefs in JSON format (e.g. for use in charts) rather than as BeliefsDataFrame :param compress_json: return beliefs, sensors and sources as separate datasets to be used for lookups @@ -896,39 +896,108 @@ def to_epoch_list(column: str, np_dtype: str | None, divisor: int) -> list: return all_records, sources_metadata +def _first_of_each_group(order: np.ndarray, keys: tuple[np.ndarray, ...]) -> np.ndarray: + """The positions, within `order`, of the first row of each group. + + The rows are already sorted by `order`, so a group ends wherever one of its keys changes. + """ + is_first = np.empty(len(order), dtype=bool) + is_first[:1] = True + if len(order) > 1: + changed = np.zeros(len(order) - 1, dtype=bool) + for key in keys: + sorted_key = key[order] + changed |= sorted_key[1:] != sorted_key[:-1] + is_first[1:] = changed + return order[is_first] + + +def _belief_recency(bdf: tb.BeliefsDataFrame) -> np.ndarray: + """How recently each belief was formed, as a number that grows with recency.""" + if "belief_time" in bdf.index.names: + return bdf.index.get_level_values("belief_time").asi8 + # A shorter horizon means the belief was formed closer to the event, so later. + return -bdf.index.get_level_values("belief_horizon").asi8 + + def _select_latest_version_and_belief_per_event( bdf: tb.BeliefsDataFrame, + preferred_sources: list["DataSource"] | None = None, ) -> tb.BeliefsDataFrame: - """Keep, per event, the single belief with the latest source version, - breaking version ties by most recent belief time. + """Keep one belief per event, choosing between the sources that reported it. + + The choice is made in two steps, because a version number orders the releases of one source, + and says nothing about a different source. + Sources are therefore grouped into families sharing a name, type and model, + and versions are only ever compared inside a family. + + Within a family, the latest version wins, then the most recent belief, then the highest source id. - Beliefs that tie on both keep the order they came in, - which is what lets a caller express its own precedence by the order in which it passes its sources. - See `test_source_transition`, where the first source in the list wins the events both sources report. + Between families, the order the caller named its sources in wins, for a caller that named any, + then the most recent belief, then the highest source id. + That order is how a caller says which source it prefers where two of them report one event, + which is what `AggregatorReporter` offers through its `sources` field. - Assumes deterministic beliefs (probabilistic depth 1) and a belief_time index level. + The highest source id is a last resort, so that a tie no one else broke is at least answered the + same way every time, and does not move when a source is renamed. + + Assumes deterministic beliefs (probabilistic depth 1). """ + if len(bdf) < 2: + return bdf + source_codes, unique_sources = pd.factorize(bdf.index.get_level_values("source")) - versions = [ - Version(source.version if source.version else "0.0.0") - for source in unique_sources - ] - version_ranks = { + + families: dict = {} + family_per_source = np.array( + [ + families.setdefault((source.name, source.type, source.model), len(families)) + for source in unique_sources + ] + ) + versions = [Version(source.version or "0.0.0") for source in unique_sources] + version_order = { version: rank for rank, version in enumerate(sorted(set(versions))) } - rank_per_row = np.array([version_ranks[version] for version in versions])[ - source_codes - ] - event_values = bdf.index.get_level_values("event_start").asi8 - belief_values = bdf.index.get_level_values("belief_time").asi8 - # Sort by event (ascending), then version rank and belief time (both descending) - order = np.lexsort((-belief_values, -rank_per_row, event_values)) - sorted_events = event_values[order] - is_first_of_event = np.empty(len(order), dtype=bool) - is_first_of_event[:1] = True - is_first_of_event[1:] = sorted_events[1:] != sorted_events[:-1] - mask = np.zeros(len(order), dtype=bool) - mask[order[is_first_of_event]] = True + version_per_source = np.array([version_order[version] for version in versions]) + id_per_source = np.array( + [source.id if source.id is not None else -1 for source in unique_sources] + ) + # Sources the caller did not name rank behind the ones it did, in the order it gave them. + positions: dict = {} + for position, source in enumerate(preferred_sources or []): + positions.setdefault(source.id, position) + position_per_source = np.array( + [positions.get(source.id, len(positions)) for source in unique_sources] + ) + + events = bdf.index.get_level_values("event_start").asi8 + recency = _belief_recency(bdf) + family = family_per_source[source_codes] + version = version_per_source[source_codes] + source_id = id_per_source[source_codes] + position = position_per_source[source_codes] + + # np.lexsort takes its keys least significant first. + within_family = _first_of_each_group( + np.lexsort((-source_id, -recency, -version, family, events)), + (events, family), + ) + between_families = _first_of_each_group( + within_family[ + np.lexsort( + ( + -source_id[within_family], + -recency[within_family], + position[within_family], + events[within_family], + ) + ) + ], + (events,), + ) + mask = np.zeros(len(bdf), dtype=bool) + mask[between_families] = True return bdf[mask] @@ -1061,7 +1130,7 @@ def search( :param most_recent_beliefs_only: only return the most recent beliefs for each event from each source (minimum belief horizon). Defaults to True. :param most_recent_events_only: only return (post knowledge time) beliefs for the most recent event (maximum event start) :param most_recent_only: only return a single belief, the most recent from the most recent event. Fastest method if you only need one. - :param one_deterministic_belief_per_event: only return a single value per event (no probabilistic distribution and only 1 source) + :param one_deterministic_belief_per_event: only return a single value per event (no probabilistic distribution and only 1 source). Where several sources report an event, the one to keep is chosen by version within a family of sources sharing a name, type and model, and between families by the order the sources were passed in, then by the most recent belief, then by the highest source id. :param one_deterministic_belief_per_event_per_source: only return a single value per event per source (no probabilistic distribution) :param resolution: Optional timedelta or pandas freqstr used to resample the results ** :param sum_multiple: if True, sum over multiple sensors; otherwise, return a dictionary with sensors as key, each holding a BeliefsDataFrame as its value @@ -1129,11 +1198,10 @@ def search( custom_filter_criteria=source_criteria, custom_join_targets=custom_join_targets, ) - if use_latest_version_per_event: - bdf = keep_latest_version( - bdf=bdf, - one_deterministic_belief_per_event=one_deterministic_belief_per_event, - ) + if use_latest_version_per_event and not one_deterministic_belief_per_event: + # Asking for one belief per event settles the versions itself, family by family, + # so it does not need this pass first. + bdf = keep_latest_version(bdf=bdf) if one_deterministic_belief_per_event: if ( bdf.lineage.number_of_sources <= 1 @@ -1141,34 +1209,13 @@ def search( ): # Fast track, no need to loop over beliefs pass - elif ( - bdf.lineage.probabilistic_depth == 1 - and "belief_time" in bdf.index.names - ): - # Deterministic beliefs: no need to take the median, - # just pick the winning belief per event directly - bdf = _select_latest_version_and_belief_per_event(bdf) else: - # First make deterministic - bdf = bdf.for_each_belief(get_median_belief) - # Then sort each event by latest source version and most recent belief_time - version_per_source = { - source: Version(source.version if source.version else "0.0.0") - for source in bdf.lineage.sources - } - bdf = bdf.sort_values( - by=["event_start", "source", "belief_time"], - ascending=[True, False, False], - key=lambda col: ( - col.map(version_per_source) if col.name == "source" else col - ), + if bdf.lineage.probabilistic_depth != 1: + # Make the beliefs deterministic, so that one of them can be chosen. + bdf = bdf.for_each_belief(get_median_belief) + bdf = _select_latest_version_and_belief_per_event( + bdf, preferred_sources=parsed_sources ) - # Finally, take the first belief for each event, thus preference latest version first, most recent belief_time second - bdf = bdf[ - ~bdf.index.get_level_values("event_start").duplicated( - keep="first" - ) - ] elif one_deterministic_belief_per_event_per_source: if len(bdf) == 0 or bdf.lineage.probabilistic_depth == 1: # Fast track, no need to loop over beliefs diff --git a/flexmeasures/data/tests/test_search_postprocessing.py b/flexmeasures/data/tests/test_search_postprocessing.py index 8e794c6d5c..3fa53637f8 100644 --- a/flexmeasures/data/tests/test_search_postprocessing.py +++ b/flexmeasures/data/tests/test_search_postprocessing.py @@ -36,12 +36,37 @@ def make_random_deterministic_bdf( def naive_select_latest_version_and_belief_per_event( bdf: tb.BeliefsDataFrame, + preferred_sources: list[DataSource] | None = None, ) -> tb.BeliefsDataFrame: - """Reference implementation: per event, pick the belief with the latest - source version, breaking version ties by most recent belief time.""" - winners: dict = {} + """Reference implementation, written per row rather than vectorised. + + Per event, keep one belief per family of sources sharing a name, type and model, + choosing the latest version, then the most recent belief, then the highest source id. + Then, among those, choose the source the caller named first, + again falling back on the most recent belief and then the highest source id. + """ + positions: dict = {} + for position, source in enumerate(preferred_sources or []): + positions.setdefault(source.id, position) + unlisted = len(positions) + + per_family: dict = {} for i, (event_start, belief_time, source, _cp) in enumerate(bdf.index): - candidate = (Version(source.version or "0.0.0"), belief_time) + family = (event_start, source.name, source.type, source.model) + candidate = (Version(source.version or "0.0.0"), belief_time, source.id or -1) + incumbent = per_family.get(family) + if incumbent is None or candidate > incumbent[0]: + per_family[family] = (candidate, i) + + winners: dict = {} + for (event_start, _name, _type, _model), (_key, i) in per_family.items(): + _, belief_time, source, _cp = bdf.index[i] + # A lower position is preferred, so it is negated to keep "greater is better". + candidate = ( + -positions.get(source.id, unlisted), + belief_time, + source.id or -1, + ) incumbent = winners.get(event_start) if incumbent is None or candidate > incumbent[0]: winners[event_start] = (candidate, i) @@ -50,26 +75,134 @@ def naive_select_latest_version_and_belief_per_event( def test_select_latest_version_and_belief_per_event_equivalence(): + """The vectorised choice agrees with the plainly written one, over random frames. + + The sources span two families, so that the two steps of the choice are both exercised, + and each trial is run with and without a caller's preference. + """ rng = np.random.default_rng(7) sources = [ + DataSource(id=1, name="s1", model="model 1", type="forecaster", version=None), DataSource( - id=i + 1, - name="s1", - model="model 1", - type="forecaster", - version=version, - ) - for i, version in enumerate([None, "0.1.0", "0.2.0", "0.2.0", "1.0.0"]) + id=2, name="s1", model="model 1", type="forecaster", version="0.1.0" + ), + DataSource( + id=3, name="s1", model="model 1", type="forecaster", version="0.2.0" + ), + DataSource( + id=4, name="s1", model="model 1", type="forecaster", version="0.2.0" + ), + DataSource(id=5, name="s2", model="model 2", type="scheduler", version="1.0.0"), + DataSource(id=6, name="s2", model="model 2", type="scheduler", version="9.0.0"), ] - for trial in range(10): - bdf = make_random_deterministic_bdf( - rng, sources, n_beliefs=int(rng.integers(2, 30)) + for preference in (None, [sources[4], sources[0]], [sources[0], sources[5]]): + for _ in range(10): + bdf = make_random_deterministic_bdf( + rng, sources, n_beliefs=int(rng.integers(2, 30)) + ) + result = _select_latest_version_and_belief_per_event( + bdf, preferred_sources=preference + ) + expected = naive_select_latest_version_and_belief_per_event( + bdf, preferred_sources=preference + ) + pd.testing.assert_frame_equal(pd.DataFrame(result), pd.DataFrame(expected)) + # Exactly one belief per event + assert not result.index.get_level_values("event_start").duplicated().any() + + +def _one_event_frame( + beliefs: list[tuple[DataSource, str, float]], +) -> tb.BeliefsDataFrame: + """A frame of beliefs about one event, each given as its source, belief time and value.""" + sensor = tb.Sensor("precedence sensor", event_resolution=timedelta(hours=1)) + event_start = pd.Timestamp("2025-01-01T00:00:00+00:00") + return tb.BeliefsDataFrame( + [ + tb.TimedBelief( + sensor=sensor, + source=source, + event_start=event_start, + belief_time=pd.Timestamp(belief_time), + event_value=value, + ) + for source, belief_time, value in beliefs + ] + ) + + +def _chosen_value(beliefs, preferred_sources=None) -> float: + frame = _select_latest_version_and_belief_per_event( + _one_event_frame(beliefs), preferred_sources=preferred_sources + ) + assert len(frame) == 1 + return frame["event_value"].iloc[0] + + +def test_a_later_version_of_one_source_wins_its_family(): + """Within a family, the version says which code produced the value, so it comes first.""" + old = DataSource(id=1, name="rep", model="Rep", type="reporter", version="1.0.0") + new = DataSource(id=2, name="rep", model="Rep", type="reporter", version="2.0.0") + # The older version spoke more recently, and still loses. + assert ( + _chosen_value( + [(new, "2024-12-31T00:00+00:00", 2.0), (old, "2024-12-31T06:00+00:00", 1.0)] ) - result = _select_latest_version_and_belief_per_event(bdf) - expected = naive_select_latest_version_and_belief_per_event(bdf) - pd.testing.assert_frame_equal(pd.DataFrame(result), pd.DataFrame(expected)) - # Exactly one belief per event - assert not result.index.get_level_values("event_start").duplicated().any() + == 2.0 + ) + + +def test_versions_are_not_compared_between_families(): + """A version number orders one source's releases, and says nothing about another source. + + A scheduler at v1 and a forecaster at v9 are unrelated numbering, + so the choice falls to the more recent belief instead. + """ + scheduler = DataSource( + id=1, name="Seita", model="StorageScheduler", type="scheduler", version="1" + ) + forecaster = DataSource( + id=2, name="Seita", model="Prophet", type="forecaster", version="9" + ) + assert ( + _chosen_value( + [ + (scheduler, "2024-12-31T06:00+00:00", 1.0), + (forecaster, "2024-12-31T00:00+00:00", 9.0), + ] + ) + == 1.0 + ) + + +def test_a_caller_that_names_its_sources_says_which_it_prefers(): + """Between families, the order the caller named its sources in decides, whatever the belief times say.""" + meter = DataSource(id=1, name="meter", model="M", type="other") + scheduler = DataSource(id=2, name="Seita", model="S", type="scheduler") + beliefs = [ + (meter, "2024-12-31T06:00+00:00", 1.0), + (scheduler, "2024-12-31T00:00+00:00", 2.0), + ] + assert _chosen_value(beliefs, preferred_sources=[scheduler, meter]) == 2.0 + assert _chosen_value(beliefs, preferred_sources=[meter, scheduler]) == 1.0 + # Naming neither leaves the more recent belief to decide. + assert _chosen_value(beliefs) == 1.0 + + +def test_a_tie_no_one_broke_is_answered_the_same_way_every_time(): + """Two sources that nothing else tells apart are settled by the highest id. + + The id does not move when a source is renamed, where sorting on the name would. + """ + first = DataSource(id=1, name="zzz", model="M", type="reporter") + second = DataSource(id=2, name="aaa", model="M", type="reporter") + beliefs = [ + (first, "2024-12-31T00:00+00:00", 1.0), + (second, "2024-12-31T00:00+00:00", 2.0), + ] + assert _chosen_value(beliefs) == 2.0 + # And the same answer whichever order the rows arrive in. + assert _chosen_value(list(reversed(beliefs))) == 2.0 def naive_compress_belief_records(df: pd.DataFrame, sensor_id: int): From f84cf1b2d34dbd5ff89a17a50dcf5400f48e5bb7 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 11:26:23 +0200 Subject: [PATCH 06/14] docs/changelog: point the entry at the PR Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 8807ca5aa2..81ba8d5dfd 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -17,7 +17,7 @@ v1.1.0 | September XX, 2026 New features ------------- -* Try out a forecast without recording it, using ``flexmeasures add forecasts --dry-run``, which computes the forecast in full and reports the sensor, data source, number of beliefs and event range it would have saved [see `PR #2483 `_] +* Try out a forecast without recording it, using ``flexmeasures add forecasts --dry-run``, which computes the forecast in full and reports the sensor, data source, number of beliefs and event range it would have saved [see `PR #2494 `_] * Run one-off reports as background jobs from the CLI or the asset API, with sensor-level authorization and a dedicated reporting worker queue [see `PR #2298 `_] * A single automation can now be run on demand, from the CLI (``flexmeasures jobs run-automation``), the API (``POST /assets//automations//trigger``) and the asset's *Automations* page (a *Run now* button), which is useful to try out a new automation, to re-run one after fixing what made it fail, or to refresh its results after late input data arrived [see `PR #2460 `_] * Changing the selected time range on an asset or sensor chart now only loads the data that is actually new, instead of reloading the whole range, which makes stepping through or extending a long period much faster; reloading the page, or leaving it open for five minutes, still fetches everything afresh [see `PR #2433 `_] @@ -39,9 +39,9 @@ Infrastructure / Support Bugfixes ----------- -* Which data source wins an event, when several report it and one value per event is asked for, now follows one rule: a source's version is only compared within a family of sources sharing a name, type and model, where it used to be compared between unrelated sources, so a forecaster at v9 no longer outranks a scheduler at v1; between families, the order in which a caller passed its sources decides, which is what the ``AggregatorReporter`` has documented all along; and a tie neither settles falls to the most recent belief and then the highest source id, rather than to the source whose name sorts first [see `PR #2483 `_] +* Which data source wins an event, when several report it and one value per event is asked for, now follows one rule: a source's version is only compared within a family of sources sharing a name, type and model, where it used to be compared between unrelated sources, so a forecaster at v9 no longer outranks a scheduler at v1; between families, the order in which a caller passed its sources decides, which is what the ``AggregatorReporter`` has documented all along; and a tie neither settles falls to the most recent belief and then the highest source id, rather than to the source whose name sorts first [see `PR #2494 `_] * A KPI on the asset page counted an event once per data source that reported it, so a total could come out higher than any source reported; it now reduces one value per event, from the latest source version and the most recent belief within it [see `PR #2472 `_] -* ``flexmeasures add schedule --dry-run`` no longer saves a schedule when it is combined with ``--as-job``, where the flag used to be dropped without a word and the queued job stored its schedule anyway; that combination is now rejected, and a dry run says how many beliefs it would have saved and which events they cover [see `PR #2483 `_] +* ``flexmeasures add schedule --dry-run`` no longer saves a schedule when it is combined with ``--as-job``, where the flag used to be dropped without a word and the queued job stored its schedule anyway; that combination is now rejected, and a dry run says how many beliefs it would have saved and which events they cover [see `PR #2494 `_] * Upgrading a database old enough to still carry the pre-``GenericAsset``/``Sensor`` tables now works, where the v0.18.0 migration that removes them crashed twice over: once while checking whether those tables hold data, as soon as one of them held more than a single row, and once while dropping them, because it dropped each table before the ones referencing it [see `PR #2475 `_] * A forecaster that is told both where to start training and how much history to train on now trains on whichever of the two asks for less data, rather than training back to the start date: ``train-start`` says where training may begin, and ``train-period`` says how much history to use [see `PR #2482 `_] * ``max-training-period`` said the same thing as ``train-period``, so the two are now one setting, and the former is a deprecated alias that hosts should stop using [see `PR #2482 `_] From a17ccf67718d5f14e5c06805b93d2ddeadaf2e29 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 11:27:15 +0200 Subject: [PATCH 07/14] docs/changelog: restore two entries' own PR references A blanket replacement of the placeholder PR number also rewrote the links of two unrelated entries that legitimately cite PR #2483. Only this branch's own entry points at #2494. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 81ba8d5dfd..eb4846160a 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -17,7 +17,7 @@ v1.1.0 | September XX, 2026 New features ------------- -* Try out a forecast without recording it, using ``flexmeasures add forecasts --dry-run``, which computes the forecast in full and reports the sensor, data source, number of beliefs and event range it would have saved [see `PR #2494 `_] +* Try out a forecast without recording it, using ``flexmeasures add forecasts --dry-run``, which computes the forecast in full and reports the sensor, data source, number of beliefs and event range it would have saved [see `PR #2483 `_] * Run one-off reports as background jobs from the CLI or the asset API, with sensor-level authorization and a dedicated reporting worker queue [see `PR #2298 `_] * A single automation can now be run on demand, from the CLI (``flexmeasures jobs run-automation``), the API (``POST /assets//automations//trigger``) and the asset's *Automations* page (a *Run now* button), which is useful to try out a new automation, to re-run one after fixing what made it fail, or to refresh its results after late input data arrived [see `PR #2460 `_] * Changing the selected time range on an asset or sensor chart now only loads the data that is actually new, instead of reloading the whole range, which makes stepping through or extending a long period much faster; reloading the page, or leaving it open for five minutes, still fetches everything afresh [see `PR #2433 `_] @@ -41,7 +41,7 @@ Bugfixes * Which data source wins an event, when several report it and one value per event is asked for, now follows one rule: a source's version is only compared within a family of sources sharing a name, type and model, where it used to be compared between unrelated sources, so a forecaster at v9 no longer outranks a scheduler at v1; between families, the order in which a caller passed its sources decides, which is what the ``AggregatorReporter`` has documented all along; and a tie neither settles falls to the most recent belief and then the highest source id, rather than to the source whose name sorts first [see `PR #2494 `_] * A KPI on the asset page counted an event once per data source that reported it, so a total could come out higher than any source reported; it now reduces one value per event, from the latest source version and the most recent belief within it [see `PR #2472 `_] -* ``flexmeasures add schedule --dry-run`` no longer saves a schedule when it is combined with ``--as-job``, where the flag used to be dropped without a word and the queued job stored its schedule anyway; that combination is now rejected, and a dry run says how many beliefs it would have saved and which events they cover [see `PR #2494 `_] +* ``flexmeasures add schedule --dry-run`` no longer saves a schedule when it is combined with ``--as-job``, where the flag used to be dropped without a word and the queued job stored its schedule anyway; that combination is now rejected, and a dry run says how many beliefs it would have saved and which events they cover [see `PR #2483 `_] * Upgrading a database old enough to still carry the pre-``GenericAsset``/``Sensor`` tables now works, where the v0.18.0 migration that removes them crashed twice over: once while checking whether those tables hold data, as soon as one of them held more than a single row, and once while dropping them, because it dropped each table before the ones referencing it [see `PR #2475 `_] * A forecaster that is told both where to start training and how much history to train on now trains on whichever of the two asks for less data, rather than training back to the start date: ``train-start`` says where training may begin, and ``train-period`` says how much history to use [see `PR #2482 `_] * ``max-training-period`` said the same thing as ``train-period``, so the two are now one setting, and the former is a deprecated alias that hosts should stop using [see `PR #2482 `_] From c189aa93be0d746723aa34013f3d16617ac42c4d Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 11:32:02 +0200 Subject: [PATCH 08/14] docs: reflow the tie-break sentence to break after punctuation Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- flexmeasures/data/models/time_series.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index e84eda95aa..6ce67ae723 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -938,8 +938,8 @@ def _select_latest_version_and_belief_per_event( That order is how a caller says which source it prefers where two of them report one event, which is what `AggregatorReporter` offers through its `sources` field. - The highest source id is a last resort, so that a tie no one else broke is at least answered the - same way every time, and does not move when a source is renamed. + The highest source id is a last resort, so that a tie no one else broke is answered the same way every time, + and does not move when a source is renamed. Assumes deterministic beliefs (probabilistic depth 1). """ From 9c54db4855d057124973bd26105344e727279995 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 11:47:03 +0200 Subject: [PATCH 09/14] Treat one named source as one preference, however many it matches A source name is not unique, and the query that resolves one has no ordering, so a caller naming a single source could hand the choice several of them in whatever order the database returned. Read as a precedence, that would have let the database decide, which is the opposite of the point. The sources of one entry now share a rank, and are told apart by belief time and then id, like any other tie. `parse_source_arg_per_entry` keeps that grouping, and `parse_source_arg` flattens it, so the older function behaves exactly as before. Also shorten the changelog entry to what a reader of the changelog needs, and leave the rule itself to the docstring and the pull request. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- documentation/changelog.rst | 2 +- flexmeasures/data/models/parsing_utils.py | 68 +++++++++++++------ flexmeasures/data/models/time_series.py | 28 ++++++-- .../data/tests/test_search_postprocessing.py | 23 +++++++ 4 files changed, 92 insertions(+), 29 deletions(-) diff --git a/documentation/changelog.rst b/documentation/changelog.rst index eb4846160a..7c8e573648 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -39,7 +39,7 @@ Infrastructure / Support Bugfixes ----------- -* Which data source wins an event, when several report it and one value per event is asked for, now follows one rule: a source's version is only compared within a family of sources sharing a name, type and model, where it used to be compared between unrelated sources, so a forecaster at v9 no longer outranks a scheduler at v1; between families, the order in which a caller passed its sources decides, which is what the ``AggregatorReporter`` has documented all along; and a tie neither settles falls to the most recent belief and then the highest source id, rather than to the source whose name sorts first [see `PR #2494 `_] +* Where several data sources report the same event, which one a search keeps is now decided the same way every time: a source version only counts against other versions of that source, and a caller that lists its sources gets the order it asked for [see `PR #2494 `_] * A KPI on the asset page counted an event once per data source that reported it, so a total could come out higher than any source reported; it now reduces one value per event, from the latest source version and the most recent belief within it [see `PR #2472 `_] * ``flexmeasures add schedule --dry-run`` no longer saves a schedule when it is combined with ``--as-job``, where the flag used to be dropped without a word and the queued job stored its schedule anyway; that combination is now rejected, and a dry run says how many beliefs it would have saved and which events they cover [see `PR #2483 `_] * Upgrading a database old enough to still carry the pre-``GenericAsset``/``Sensor`` tables now works, where the v0.18.0 migration that removes them crashed twice over: once while checking whether those tables hold data, as soon as one of them held more than a single row, and once while dropping them, because it dropped each table before the ones referencing it [see `PR #2475 `_] diff --git a/flexmeasures/data/models/parsing_utils.py b/flexmeasures/data/models/parsing_utils.py index df58f70f0f..68b6d0e09c 100644 --- a/flexmeasures/data/models/parsing_utils.py +++ b/flexmeasures/data/models/parsing_utils.py @@ -9,7 +9,7 @@ from flexmeasures.data.models.data_sources import DataSource -def parse_source_arg( +def parse_source_arg_per_entry( source: ( DataSource | int @@ -19,38 +19,62 @@ def parse_source_arg( | Sequence[str] | None ), -) -> list[DataSource] | None: - """Parse the "source" argument by looking up DataSources corresponding to any given ids or names. +) -> list[list[DataSource]] | None: + """Parse the "source" argument, keeping the sources of each entry together. + + One entry can name more than one source, because a name is not unique, + and the sources it names arrive in no particular order. + Callers that care which source is preferred should treat one entry as one preference, + rather than reading an order into what a single name happened to match. Passes None as is (i.e. no source argument is given). - Accepts ids and names as list or tuples, always converting them to a list. """ if source is None: return source if isinstance(source, (DataSource, str, int)): - sources = [source] + entries: Sequence = [source] else: - sources = source - parsed_sources: list[DataSource] = [] - for source in sources: - if isinstance(source, int): - parsed_source = db.session.get(DataSource, source) + entries = source + parsed_entries: list[list[DataSource]] = [] + for entry in entries: + if isinstance(entry, int): + parsed_source = db.session.get(DataSource, entry) if parsed_source is None: current_app.logger.warning( - f"Beliefs searched for unknown source {source}" + f"Beliefs searched for unknown source {entry}" ) + parsed_entries.append([]) else: - parsed_sources.append(parsed_source) - elif isinstance(source, str): - _parsed_sources = db.session.scalars( - select(DataSource).filter_by(name=source) - ).all() - if _parsed_sources is []: + parsed_entries.append([parsed_source]) + elif isinstance(entry, str): + named = db.session.scalars(select(DataSource).filter_by(name=entry)).all() + if not named: current_app.logger.warning( - f"Beliefs searched for unknown source {source}" + f"Beliefs searched for unknown source {entry}" ) - else: - parsed_sources.extend(_parsed_sources) + parsed_entries.append(list(named)) else: - parsed_sources.append(source) - return parsed_sources + parsed_entries.append([entry]) + return parsed_entries + + +def parse_source_arg( + source: ( + DataSource + | int + | str + | Sequence[DataSource] + | Sequence[int] + | Sequence[str] + | None + ), +) -> list[DataSource] | None: + """Parse the "source" argument by looking up DataSources corresponding to any given ids or names. + + Passes None as is (i.e. no source argument is given). + Accepts ids and names as list or tuples, always converting them to a list. + """ + parsed_entries = parse_source_arg_per_entry(source) + if parsed_entries is None: + return None + return [parsed_source for entry in parsed_entries for parsed_source in entry] diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index 6ce67ae723..a01b797f54 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -23,7 +23,10 @@ from flexmeasures.data import db from flexmeasures.data.models.legacy_migration_utils import upgrade_value from flexmeasures.data.models.data_sources import keep_latest_version -from flexmeasures.data.models.parsing_utils import parse_source_arg +from flexmeasures.data.models.parsing_utils import ( + parse_source_arg, + parse_source_arg_per_entry, +) from flexmeasures.data.services.annotations import prepare_annotations_for_chart from flexmeasures.data.services.timerange import get_timerange from flexmeasures.data.queries.utils import get_source_criteria @@ -922,7 +925,7 @@ def _belief_recency(bdf: tb.BeliefsDataFrame) -> np.ndarray: def _select_latest_version_and_belief_per_event( bdf: tb.BeliefsDataFrame, - preferred_sources: list["DataSource"] | None = None, + preferred_sources: list | None = None, ) -> tb.BeliefsDataFrame: """Keep one belief per event, choosing between the sources that reported it. @@ -935,6 +938,8 @@ def _select_latest_version_and_belief_per_event( Between families, the order the caller named its sources in wins, for a caller that named any, then the most recent belief, then the highest source id. + One entry of `preferred_sources` is one preference, so a group of sources given together, + as a name that matched several of them does, shares a rank rather than being ordered among itself. That order is how a caller says which source it prefers where two of them report one event, which is what `AggregatorReporter` offers through its `sources` field. @@ -965,8 +970,10 @@ def _select_latest_version_and_belief_per_event( ) # Sources the caller did not name rank behind the ones it did, in the order it gave them. positions: dict = {} - for position, source in enumerate(preferred_sources or []): - positions.setdefault(source.id, position) + for position, entry in enumerate(preferred_sources or []): + group = entry if isinstance(entry, (list, tuple)) else [entry] + for source in group: + positions.setdefault(source.id, position) position_per_source = np.array( [positions.get(source.id, len(positions)) for source in unique_sources] ) @@ -1162,7 +1169,16 @@ def search( ).all() sensors.extend(sensors_from_names) - parsed_sources = parse_source_arg(source) + parsed_source_entries = parse_source_arg_per_entry(source) + parsed_sources = ( + None + if parsed_source_entries is None + else [ + parsed_source + for entry in parsed_source_entries + for parsed_source in entry + ] + ) source_criteria = get_source_criteria( cls=cls, user_source_ids=user_source_ids, @@ -1214,7 +1230,7 @@ def search( # Make the beliefs deterministic, so that one of them can be chosen. bdf = bdf.for_each_belief(get_median_belief) bdf = _select_latest_version_and_belief_per_event( - bdf, preferred_sources=parsed_sources + bdf, preferred_sources=parsed_source_entries ) elif one_deterministic_belief_per_event_per_source: if len(bdf) == 0 or bdf.lineage.probabilistic_depth == 1: diff --git a/flexmeasures/data/tests/test_search_postprocessing.py b/flexmeasures/data/tests/test_search_postprocessing.py index 3fa53637f8..dc54fa1ba8 100644 --- a/flexmeasures/data/tests/test_search_postprocessing.py +++ b/flexmeasures/data/tests/test_search_postprocessing.py @@ -189,6 +189,29 @@ def test_a_caller_that_names_its_sources_says_which_it_prefers(): assert _chosen_value(beliefs) == 1.0 +def test_sources_named_together_share_one_preference(): + """A name can match several sources, and they arrive in no particular order. + + Reading an order into that would let the database decide precedence, + so one entry is one preference, and the sources in it are told apart by belief time and id instead. + """ + # Two sources of one name, which is what `source="rep"` would match. + older = DataSource(id=1, name="rep", model="A", type="reporter") + newer = DataSource(id=2, name="rep", model="B", type="reporter") + other = DataSource(id=3, name="other", model="C", type="reporter") + beliefs = [ + (older, "2024-12-31T06:00+00:00", 1.0), + (newer, "2024-12-31T00:00+00:00", 2.0), + (other, "2024-12-31T12:00+00:00", 3.0), + ] + # Named first as one entry, the pair outranks the other source, and the fresher of the pair wins. + assert _chosen_value(beliefs, preferred_sources=[[older, newer], other]) == 1.0 + # Whichever way round that entry lists them, since one entry is one preference. + assert _chosen_value(beliefs, preferred_sources=[[newer, older], other]) == 1.0 + # Naming the other source first still puts it ahead of both. + assert _chosen_value(beliefs, preferred_sources=[other, [older, newer]]) == 3.0 + + def test_a_tie_no_one_broke_is_answered_the_same_way_every_time(): """Two sources that nothing else tells apart are settled by the highest id. From d7e29ec0864a341d37e3dc5794328ed712d40360 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 11:58:23 +0200 Subject: [PATCH 10/14] Let an entry that named nothing keep its place, without demoting the rest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A caller can name a source this database does not know, and the entry then matches nothing. Those entries still count when ranking the ones that did match, which is right, but the rank standing for "nobody named this" was the number of sources found rather than the number of entries given. Name two unknown sources and then a real one, and the real one ranked behind the sources nobody had named at all — the opposite of what naming it meant. The rank for the unnamed now clears every entry, matched or not. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- flexmeasures/data/models/time_series.py | 8 ++++++-- .../data/tests/test_search_postprocessing.py | 17 +++++++++++++++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index a01b797f54..89af1af81d 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -969,13 +969,17 @@ def _select_latest_version_and_belief_per_event( [source.id if source.id is not None else -1 for source in unique_sources] ) # Sources the caller did not name rank behind the ones it did, in the order it gave them. + # An entry that named nothing this database knows still holds its place, + # so the rank for the unnamed has to clear every entry, not merely the ones that matched. + entries = preferred_sources or [] positions: dict = {} - for position, entry in enumerate(preferred_sources or []): + for position, entry in enumerate(entries): group = entry if isinstance(entry, (list, tuple)) else [entry] for source in group: positions.setdefault(source.id, position) + unnamed_rank = len(entries) position_per_source = np.array( - [positions.get(source.id, len(positions)) for source in unique_sources] + [positions.get(source.id, unnamed_rank) for source in unique_sources] ) events = bdf.index.get_level_values("event_start").asi8 diff --git a/flexmeasures/data/tests/test_search_postprocessing.py b/flexmeasures/data/tests/test_search_postprocessing.py index dc54fa1ba8..0a94627d89 100644 --- a/flexmeasures/data/tests/test_search_postprocessing.py +++ b/flexmeasures/data/tests/test_search_postprocessing.py @@ -212,6 +212,23 @@ def test_sources_named_together_share_one_preference(): assert _chosen_value(beliefs, preferred_sources=[other, [older, newer]]) == 3.0 +def test_an_entry_that_named_nothing_does_not_demote_the_ones_that_did(): + """A caller can name a source this database does not know, and still be heard about the others. + + An unknown id or name leaves an entry that matched nothing, + and that entry still holds its place, so the sources named after it must keep outranking the unnamed. + """ + named = DataSource(id=1, name="scheduler", model="S", type="scheduler") + unnamed = DataSource(id=2, name="meter", model="M", type="other") + beliefs = [ + (named, "2024-12-31T00:00+00:00", 1.0), + # The source nobody named holds the fresher belief, so only the naming can decide this. + (unnamed, "2024-12-31T06:00+00:00", 2.0), + ] + # Two entries matched nothing, and the third named the scheduler. + assert _chosen_value(beliefs, preferred_sources=[[], [], named]) == 1.0 + + def test_a_tie_no_one_broke_is_answered_the_same_way_every_time(): """Two sources that nothing else tells apart are settled by the highest id. From f1526ba827fa0e81a9e31fb99f3bfa454d8db025 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 12:06:05 +0200 Subject: [PATCH 11/14] Say in the signature what a preference entry may be The parameter takes either a source or a group of them per entry, which `list | None` did not say, so a caller could pass the wrong thing and only find out when the ranking reached for an id. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- flexmeasures/data/models/time_series.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index 89af1af81d..c5071139a3 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Sequence from typing import Any, Type from datetime import datetime as datetime_type, timedelta from functools import cached_property @@ -925,7 +926,7 @@ def _belief_recency(bdf: tb.BeliefsDataFrame) -> np.ndarray: def _select_latest_version_and_belief_per_event( bdf: tb.BeliefsDataFrame, - preferred_sources: list | None = None, + preferred_sources: Sequence[DataSource | Sequence[DataSource]] | None = None, ) -> tb.BeliefsDataFrame: """Keep one belief per event, choosing between the sources that reported it. From cf3b0d56e284bff153a7e5ddf79d837ccca4488f Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 12:25:18 +0200 Subject: [PATCH 12/14] Hold the reference implementation to the same rule as the selector The plainly written reference still read each entry as one source and still took the rank for the unnamed from the sources it found, so it encoded both of the mistakes the selector has since stopped making. An oracle that agrees with the old rule cannot catch a return to it: the equivalence test passed only because its trials never named a group, nor anything this database does not know. The reference now follows the same two rules, and the trials include an entry naming two sources and an entry naming nothing. Reintroducing either mistake in the selector now fails the equivalence test, which is what it is there for. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- .../data/tests/test_search_postprocessing.py | 24 +++++++++++++++---- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/flexmeasures/data/tests/test_search_postprocessing.py b/flexmeasures/data/tests/test_search_postprocessing.py index 0a94627d89..0e9bed72e3 100644 --- a/flexmeasures/data/tests/test_search_postprocessing.py +++ b/flexmeasures/data/tests/test_search_postprocessing.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Sequence from datetime import timedelta import numpy as np @@ -36,7 +37,7 @@ def make_random_deterministic_bdf( def naive_select_latest_version_and_belief_per_event( bdf: tb.BeliefsDataFrame, - preferred_sources: list[DataSource] | None = None, + preferred_sources: Sequence[DataSource | Sequence[DataSource]] | None = None, ) -> tb.BeliefsDataFrame: """Reference implementation, written per row rather than vectorised. @@ -45,10 +46,15 @@ def naive_select_latest_version_and_belief_per_event( Then, among those, choose the source the caller named first, again falling back on the most recent belief and then the highest source id. """ + # One entry is one preference, and an entry that named nothing still holds its place, + # so the rank for the unnamed clears every entry rather than every source found. + entries = preferred_sources or [] positions: dict = {} - for position, source in enumerate(preferred_sources or []): - positions.setdefault(source.id, position) - unlisted = len(positions) + for position, entry in enumerate(entries): + group = entry if isinstance(entry, (list, tuple)) else [entry] + for source in group: + positions.setdefault(source.id, position) + unlisted = len(entries) per_family: dict = {} for i, (event_start, belief_time, source, _cp) in enumerate(bdf.index): @@ -95,7 +101,15 @@ def test_select_latest_version_and_belief_per_event_equivalence(): DataSource(id=5, name="s2", model="model 2", type="scheduler", version="1.0.0"), DataSource(id=6, name="s2", model="model 2", type="scheduler", version="9.0.0"), ] - for preference in (None, [sources[4], sources[0]], [sources[0], sources[5]]): + for preference in ( + None, + [sources[4], sources[0]], + [sources[0], sources[5]], + # A name that matched two sources, given as one entry, ahead of a single source. + [[sources[1], sources[2]], sources[4]], + # An entry that matched nothing, before the ones that did. + [[], sources[5], sources[0]], + ): for _ in range(10): bdf = make_random_deterministic_bdf( rng, sources, n_beliefs=int(rng.integers(2, 30)) From 28c4cb888db3af5f6f6e5c03d01d776c59e00b54 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 12:33:22 +0200 Subject: [PATCH 13/14] Test for the single source, rather than listing the containers The annotation says an entry is a source or any sequence of them, while the check asked whether it was a list or a tuple, so a sequence of another kind would have been read as one source and the ranking would have reached for an id it does not have. Asking whether the entry *is* a source cannot fall behind what the annotation allows. The reference implementation had the same check, and gets the same treatment. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- flexmeasures/data/models/time_series.py | 4 +++- flexmeasures/data/tests/test_search_postprocessing.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index c5071139a3..7bb6734596 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -975,7 +975,9 @@ def _select_latest_version_and_belief_per_event( entries = preferred_sources or [] positions: dict = {} for position, entry in enumerate(entries): - group = entry if isinstance(entry, (list, tuple)) else [entry] + # One source, or any sequence of them: test for the single case, + # so that the check cannot fall behind what the annotation allows. + group = [entry] if isinstance(entry, DataSource) else list(entry) for source in group: positions.setdefault(source.id, position) unnamed_rank = len(entries) diff --git a/flexmeasures/data/tests/test_search_postprocessing.py b/flexmeasures/data/tests/test_search_postprocessing.py index 0e9bed72e3..b0f88b277f 100644 --- a/flexmeasures/data/tests/test_search_postprocessing.py +++ b/flexmeasures/data/tests/test_search_postprocessing.py @@ -51,7 +51,9 @@ def naive_select_latest_version_and_belief_per_event( entries = preferred_sources or [] positions: dict = {} for position, entry in enumerate(entries): - group = entry if isinstance(entry, (list, tuple)) else [entry] + # One source, or any sequence of them: test for the single case, + # so that the check cannot fall behind what the annotation allows. + group = [entry] if isinstance(entry, DataSource) else list(entry) for source in group: positions.setdefault(source.id, position) unlisted = len(entries) From 28668b02b39c08256b7cc11302ee28f2d2b4cea6 Mon Sep 17 00:00:00 2001 From: "F.N. Claessen" Date: Tue, 8 Sep 2026 13:41:19 +0200 Subject: [PATCH 14/14] docs: say that the source order matters wherever one belief per event is asked for `Sensor.search_beliefs` said so, while `TimedBelief.search`, which takes the flag, and `Sensor.latest_state`, which sets it, still described the argument as a plain list. A caller reading either would have assumed the order was ignored, as it used to be. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WS6V8nZyRzMxsvqGnNpUTk Signed-off-by: F.N. Claessen --- flexmeasures/data/models/time_series.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py index 7bb6734596..fcfc28d45c 100644 --- a/flexmeasures/data/models/time_series.py +++ b/flexmeasures/data/models/time_series.py @@ -306,7 +306,7 @@ def latest_state( ) -> tb.BeliefsDataFrame: """Search the most recent event for this sensor, and return the most recent ex-post belief. - :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources + :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources. Where a list is given, its order says which source to prefer for an event that several of them report. """ return self.search_beliefs( horizons_at_most=timedelta(0), @@ -1135,7 +1135,7 @@ def search( :param beliefs_before: only return beliefs formed before this datetime (inclusive) :param horizons_at_least: only return beliefs with a belief horizon equal or greater than this timedelta (for example, use timedelta(0) to get ante knowledge time beliefs) :param horizons_at_most: only return beliefs with a belief horizon equal or less than this timedelta (for example, use timedelta(0) to get post knowledge time beliefs) - :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources + :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources. Where a list is given, its order says which source to prefer for an event that several of them report. :param user_source_ids: Optional list of user source ids to query only specific user sources :param source_account_ids: Optional account ID (or list thereof) to query only sources linked to specific accounts :param source_types: Optional list of source type names to query only specific source types *