Settle which source wins an event with one rule, in two steps

documentation/changelog.rst

@@ -38,6 +38,7 @@ Infrastructure / Support
 Bugfixes
 -----------
 
+* 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 <https://www.github.com/FlexMeasures/flexmeasures/pull/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 <https://www.github.com/FlexMeasures/flexmeasures/pull/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 <https://www.github.com/FlexMeasures/flexmeasures/pull/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 <https://www.github.com/FlexMeasures/flexmeasures/pull/2475>`_]

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]

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)),