Repository: FlexMeasures/flexmeasures
Pull request #1022 (2024-03-26 12:37:11)

Status page: show by source type
## Description

1. Split status by data source type
2. Change staleness calculation for Forecaster / Scheduler sources (staleness should be calculated as difference between now and max event start time)

## Look & Feel

1. "/<id>/status/" page should return source key as well
2. UI status page should include source type

![asset-status-source-type](https://github.com/user-attachments/assets/003576ae-4692-4e0c-85c5-0cd0b6996c4e)

![Screenshot from 2025-03-17 17-18-33](https://github.com/user-attachments/assets/816df739-8f28-4fa5-bed5-fdbe46ec4339)


## How to test

Is tested in test_sensor_data_schema.py (test_get_status_multi_source)
Can be tested the following way - add data from several sources to some sensor. Go to its status page and check that statuses are split by this source column + scheduler/forecaster statuses are calculated differently 

## Further Improvements

-

## Related Items

Follow up for https://github.com/FlexMeasures/flexmeasures/pull/941

---

- [x] I agree to contribute to the project under Apache 2 License. 
- [x] To the best of my knowledge, the proposed patch is not based on code under GPL or other license that is incompatible with FlexMeasures


Flix6x: @nrozanov if you edit this PR and set its base to 927-status-page, the diff should become a lot more to the point, making a review easier. (Once that base branch gets merged, and that branch will automatically get closed, this PR will get automatically rebased on main again.)
nhoening: Maybe that was based on my bad advice, forgot that the original branch is still open.
I'm releasing now, so we can merge the other one in a bit.
Flix6x: I encountered the issue again where the status page falsely reports that no data is available from a `forecaster` data source type (see bottom right below). There is data from this forecaster for 22 June 2023.

![image](https://github.com/FlexMeasures/flexmeasures/assets/30658763/b90cbf8d-c081-4e9e-a926-7094b5fb2178)

Likewise for `scheduler` data source types (data *is* available for 18 June 2023):

![image](https://github.com/FlexMeasures/flexmeasures/assets/30658763/e0b082c5-31ac-4078-bb35-1bc092df8148)

Maybe we can take a look together tomorrow.
nhoening: Comment by @nrozanov which is relevant for picking this up, probably:

```
Now timely-beliefs TimedBeliefDBMixin search_session method returns df with source column which is in fact source name. For https://github.com/FlexMeasures/flexmeasures/pull/1022 we need a possibility to group these results by source type. Need to consider what is the best way to do it (add new search_session argument that results in df splitting by source_type / adding source type to df / adding source id to df)
```

Might necessitate to ask @Flix6x for a recommendation on this.

Patch:
diff --git a/documentation/changelog.rst b/documentation/changelog.rst
index 2e3a7105..d703c2e1 100644
--- a/documentation/changelog.rst
+++ b/documentation/changelog.rst
@@ -12,9 +12,11 @@ v0.25.0 | February XX, 2025
 New features
 -------------
 * Added form modal to edit an asset's ``flex_context`` [see `PR #1320 <https://github.com/FlexMeasures/flexmeasures/pull/1320>`_, `PR #1365 <https://github.com/FlexMeasures/flexmeasures/pull/1365>`_ and `PR #1364 <https://github.com/FlexMeasures/flexmeasures/pull/1364>`_]
+* Improve asset status page - distinguish better by data source type [see `PR #1022 <https://github.com/FlexMeasures/flexmeasures/pull/1022/>`_]
 * Better y-axis titles for charts that show multiple sensors with a shared unit [see `PR #1346 <https://github.com/FlexMeasures/flexmeasures/pull/1346>`_]
 * Add CLI command ``flexmeasures jobs save-last-failed`` for saving the last failed jobs [see `PR #1342 <https://www.github.com/FlexMeasures/flexmeasures/pull/1342>`_ and `PR #1359 <https://github.com/FlexMeasures/flexmeasures/pull/1359>`_]
 
+
 Infrastructure / Support
 ----------------------
 * Support multi-asset scheduling in the ``StorageScheduler`` and job queueing - functionality for (plugin) developers (incl. prep work for future API endpoint for multi-asset scheduling) [see `PR #1313 <https://github.com/FlexMeasures/flexmeasures/pull/1313>`_]
diff --git a/documentation/concepts/data-model.rst b/documentation/concepts/data-model.rst
index 752aa700..f2d9be63 100644
--- a/documentation/concepts/data-model.rst
+++ b/documentation/concepts/data-model.rst
@@ -23,6 +23,8 @@ In essence, an asset is anything on which you collect data.
 Assets can also have a parent-child relationship with other assets.
 So, you could model a building that contains assets like solar panels, a heat pump and EV chargers.
 
+We model asset types explicitly. None are required for running FlexMeasures. Some asset types have support in the UI (for icons, like a sun for ``"solar"``), and in the toy tutorial and test. Some are used to select the scheduler (e.g. using ``"battery"`` or ``"one-way_evse"`` leads to using the storage scheduler). You can add your own types, which is useful for plugin logic (an example is the ``"weather station"`` type for a plugin that reads in weather forecasts).
+
 
 Sensors
 ---------
@@ -42,9 +44,12 @@ A market might have a publication date you want to adhere to. More information `
 Data sources
 ------------
 
-We keep track of where data comes from, for better reporting (this is also an aspect of the timely-beliefs package).
+We keep track of where data comes from, for better reporting, graphing and the status page (this is also an aspect of the timely-beliefs package).
 A data source can be a FlexMeasures user, but also simply a named source from outside, e.g. a third-party API, where weather forecasts are collected from.
 
+In FlexMeasures, data sources have a type. It is just a string which you can freely choose (we do not model them explicitly im the data model like Asset types).
+We do support some types out of the box: "scheduler", "forecaster" "reporter", "demo script" and "user".  
+
 
 Beliefs
 ---------
@@ -74,4 +79,4 @@ Accounts "own" assets, and data of these assets are protected against anyone fro
 
 Accounts can "consult" other accounts. This depicts the real situation that some organizations are the consultants or advisors to many others.
 They have certain rights, e.g. to read the data of their clients. That is useful for serving them.
-If you are hosting FlexMeasures, and the organizations you serve with it use this feature, you are effectively running a B2B2B setup :)
\ No newline at end of file
+If you are hosting FlexMeasures, and the organizations you serve with it use this feature, you are effectively running a B2B2B setup :)
diff --git a/documentation/views/asset-data.rst b/documentation/views/asset-data.rst
index bbb9a919..ee7bbc65 100644
--- a/documentation/views/asset-data.rst
+++ b/documentation/views/asset-data.rst
@@ -81,15 +81,17 @@ Setting a Field Value
 Status page
 ^^^^^^^^^^^^
 
-For each asset, you can also visit a status page to see if your data connectivity and recent jobs are okay. This is how data connectivity status looks like on the building asset from our tutorial:
+For each asset, you can also visit a status page to see if your data connectivity and recent jobs are okay. At the moment, all sensors on the asset and from its flex context are tracked. Below is a fictious example, where the toy battery (from our tutorial) has schedules discharging data, but also some added by a user, and wind production data is part of the battery's flex context. There have been three succesful scheduling jobs.
 
 .. image:: https://github.com/FlexMeasures/screenshots/raw/main/tut/toy-schedule/screenshot_building_status.png
     :align: center
 ..    :scale: 40%
 
-|
-|
 
+Audit log 
+^^^^^^^^^
+
+The audit log lets you see who made what changes to the asset over time. 
 This is how the audit log looks for the history of actions taken on an asset:
 
 .. image:: https://github.com/FlexMeasures/screenshots/raw/main/screenshot-auditlog.PNG
diff --git a/flexmeasures/data/queries/utils.py b/flexmeasures/data/queries/utils.py
index 3044240d..badbd72a 100644
--- a/flexmeasures/data/queries/utils.py
+++ b/flexmeasures/data/queries/utils.py
@@ -275,6 +275,7 @@ def multiply_dataframe_with_deterministic_beliefs(
     df = (df1["event_value"] * df2["event_value"] * multiplication_factor).to_frame(
         name="event_value"
     )
+
     if "belief_horizon" in df1.columns and "belief_horizon" in df2.columns:
         df["belief_horizon"] = (
             df1["belief_horizon"]
@@ -283,7 +284,8 @@ def multiply_dataframe_with_deterministic_beliefs(
             .join(df2["belief_horizon"], how="outer")
             .min(axis=1)
             .rename("belief_horizon")
-        )  # Add existing belief_horizon information, keeping only the smaller horizon per row
+        )
+        # Add existing belief_horizon information, keeping only the smaller horizon per row
     if result_source is not None:
         df["source"] = result_source  # also for rows with nan event_value
     return df
diff --git a/flexmeasures/data/schemas/reporting/__init__.py b/flexmeasures/data/schemas/reporting/__init__.py
index 69829229..43a04124 100644
--- a/flexmeasures/data/schemas/reporting/__init__.py
+++ b/flexmeasures/data/schemas/reporting/__init__.py
@@ -69,5 +69,6 @@ class BeliefsSearchConfigSchema(Schema):
 
 class StatusSchema(Schema):
     max_staleness = DurationField(required=True)
+    max_future_staleness = DurationField(required=True)
 
     staleness_search = fields.Nested(BeliefsSearchConfigSchema(), required=True)
diff --git a/flexmeasures/data/services/sensors.py b/flexmeasures/data/services/sensors.py
index 2ca4631e..8d96a432 100644
--- a/flexmeasures/data/services/sensors.py
+++ b/flexmeasures/data/services/sensors.py
@@ -10,7 +10,7 @@ import time
 from timely_beliefs import BeliefsDataFrame
 import pandas as pd
 
-from humanize.time import naturaldelta
+from humanize.time import precisedelta
 
 from flexmeasures.data.models.time_series import TimedBelief
 
@@ -65,38 +65,67 @@ def get_sensors(
     return db.session.scalars(sensor_query).all()
 
 
-def _get_sensor_bdf(sensor: Sensor, staleness_search: dict) -> BeliefsDataFrame | None:
-    """
-    Retrieve the BeliefsDataFrame for a given sensor using the specified search parameters. The 'most_recent_only'
-    parameter is set to True, which ensures that only the most recent belief from the most recent event is returned.
+def _get_sensor_bdfs_by_source_type(
+    sensor: Sensor, staleness_search: dict
+) -> dict[str, BeliefsDataFrame] | None:
+    """Get latest event, split by source type for a given sensor with given search parameters.
+    For now we use 'demo script', 'user', 'forecaster', 'scheduler' and 'reporter' source types
     """
-    bdf = TimedBelief.search(
-        sensors=sensor,
-        most_recent_beliefs_only=False,
-        most_recent_only=True,
-        **staleness_search,
-    )
-    if bdf.empty:
-        return None
-    return bdf
+    bdfs_by_source = dict()
+    for source_type in ("demo script", "user", "forecaster", "scheduler", "reporter"):
+        bdf = TimedBelief.search(
+            sensors=sensor,
+            most_recent_events_only=True,
+            source_types=[source_type],
+            **staleness_search,
+        )
+        if not bdf.empty:
+            bdfs_by_source[source_type] = bdf
+    return None if not bdfs_by_source else bdfs_by_source
 
 
-def get_most_recent_knowledge_time(
-    sensor: Sensor, staleness_search: dict
-) -> datetime | None:
-    """Get the knowledge time of the sensor's most recent event.
+def get_staleness_start_times(
+    sensor: Sensor, staleness_search: dict, now: datetime
+) -> dict[str, timedelta] | None:
+    """Get staleness start times for a given sensor by source.
+    Also add whether there has any relevant data (for forecasters and schedulers this is future data).
+    For scheduler and forecaster sources staleness start is latest event start time.
 
+    For other sources staleness start is the knowledge time of the sensor's most recent event.
     This knowledge time represents when you could have known about the event
     (specifically, when you could have formed an ex-post belief about it).
     """
-    staleness_bdf = _get_sensor_bdf(sensor=sensor, staleness_search=staleness_search)
-    return None if staleness_bdf is None else staleness_bdf.knowledge_times[-1]
+    staleness_bdfs = _get_sensor_bdfs_by_source_type(
+        sensor=sensor, staleness_search=staleness_search
+    )
+    if staleness_bdfs is None:
+        return None
+
+    start_times = dict()
+    for source_type, bdf in staleness_bdfs.items():
+        time_column = "knowledge_times"
+        source_type = str(source_type)
+        has_relevant_data = True
+        if source_type in ("scheduler", "forecaster"):
+            # filter to get only future events
+            bdf_filtered = bdf[bdf.event_starts > now]
+            time_column = "event_starts"
+            if bdf_filtered.empty:
+                has_relevant_data = False
+                bdf_filtered = bdf
+            bdf = bdf_filtered
+        start_times[source_type] = (
+            has_relevant_data,
+            getattr(bdf, time_column)[-1] if not bdf.empty else None,
+        )
 
+    return start_times
 
-def get_staleness(
+
+def get_stalenesses(
     sensor: Sensor, staleness_search: dict, now: datetime
-) -> timedelta | None:
-    """Get the staleness of the sensor.
+) -> dict[str, timedelta] | None:
+    """Get the staleness of the sensor split by source.
 
     The staleness is defined relative to the knowledge time of the most recent event, rather than to its belief time.
     Basically, that means that we don't really care when the data arrived,
@@ -110,47 +139,55 @@ def get_staleness(
 
     # Mask beliefs before now
     staleness_search = staleness_search.copy()  # no inplace operations
-    beliefs_before = staleness_search.get("beliefs_before")
-    if beliefs_before is not None:
-        staleness_search["beliefs_before"] = min(beliefs_before, now)
-    else:
-        staleness_search["beliefs_before"] = now
+    staleness_search["beliefs_before"] = min(
+        now, staleness_search.get("beliefs_before", now)
+    )
 
-    staleness_start_time = get_most_recent_knowledge_time(
-        sensor=sensor, staleness_search=staleness_search
+    staleness_start_times = get_staleness_start_times(
+        sensor=sensor, staleness_search=staleness_search, now=now
     )
-    if staleness_start_time is not None:
-        staleness = now - staleness_start_time
-    else:
-        staleness = None
+    if staleness_start_times is None:
+        return None
 
-    return staleness
+    stalenesses = dict()
+    for source_type, (has_relevant_data, start_time) in staleness_start_times.items():
+        stalenesses[str(source_type)] = (
+            has_relevant_data,
+            None if start_time is None else now - start_time,
+        )
+
+    return stalenesses
 
 
 def get_status_specs(sensor: Sensor) -> dict:
     """Get status specs from a given sensor."""
 
     # Check for explicitly defined status specs
-    status_specs = sensor.attributes.get("status_specs")
-    if status_specs is None:
-        # Default to status specs for economical sensors with daily updates
-        if sensor.knowledge_horizon_fnc == "x_days_ago_at_y_oclock":
-            status_specs = {"staleness_search": {}, "max_staleness": "P1D"}
-        else:
-            # Default to status specs indicating staleness after knowledge time + 2 sensor resolutions
-            status_specs = {
-                "staleness_search": {},
-                "max_staleness": duration_isoformat(sensor.event_resolution * 2),
-            }
+    status_specs = sensor.attributes.get("status_specs", dict())
+    if status_specs:
+        return status_specs
+
+    status_specs["staleness_search"] = {}
+    # Consider forecast or schedule data stale if it is less than 12 hours in the future
+    status_specs["max_future_staleness"] = "-PT12H"
+
+    # Default to status specs for economical sensors with daily updates
+    if sensor.knowledge_horizon_fnc == "x_days_ago_at_y_oclock":
+        status_specs["max_staleness"] = "P1D"
+        status_specs["staleness_search"] = {}
+    else:
+        # Default to status specs indicating staleness after knowledge time + 2 sensor resolutions
+        status_specs["staleness_search"] = {}
+        status_specs["max_staleness"] = duration_isoformat(sensor.event_resolution * 2)
     return status_specs
 
 
-def get_status(
+def get_statuses(
     sensor: Sensor,
     now: datetime,
     status_specs: dict | None = None,
-) -> dict:
-    """Get the status of the sensor
+) -> list[dict]:
+    """Get the status of the sensor by source type.
     Main part of result here is a stale value, which is True if the sensor is stale, False otherwise.
     Other values are just context information for the stale value.
     """
@@ -158,54 +195,73 @@ def get_status(
         status_specs = get_status_specs(sensor=sensor)
     status_specs = StatusSchema().load(status_specs)
     max_staleness = status_specs.pop("max_staleness")
+    max_future_staleness = status_specs.pop("max_future_staleness")
     staleness_search = status_specs.pop("staleness_search")
-    staleness: timedelta = get_staleness(
+    stalenesses = get_stalenesses(
         sensor=sensor,
         staleness_search=staleness_search,
         now=now,
     )
-    if staleness is not None:
-        staleness_since = now - staleness
-        stale = staleness > max_staleness
-        reason = (
-            "" if stale else "not "
-        ) + f"more than {naturaldelta(max_staleness)} old"
-        staleness = staleness if staleness > timedelta(0) else -staleness
-    else:
-        staleness_since = None
-        stale = True
-        reason = "no data recorded"
-    status = dict(
-        staleness=staleness,
-        stale=stale,
-        staleness_since=staleness_since,
-        reason=reason,
-    )
-    return status
+
+    statuses = list()
+    for source_type, (has_relevant_data, staleness) in (
+        stalenesses or {None: (True, None)}
+    ).items():
+        if staleness is None or not has_relevant_data:
+            staleness_since = now - staleness if not has_relevant_data else None
+            stale = True
+            reason = (
+                "no data recorded"
+                if staleness is None
+                else "Found no future data which this source should have"
+            )
+            staleness = None
+        else:
+            max_source_staleness = (
+                max_staleness if staleness > timedelta(0) else max_future_staleness
+            )
+            staleness_since = now - staleness
+            stale = staleness > max_source_staleness
+            timeline = "old" if staleness > timedelta(0) else "in the future"
+            reason_part = ""
+            if staleness > timedelta(0):
+                reason_part = (
+                    "which is not more" if not stale else "but should not be more"
+                )
+            else:
+                reason_part = "which is not less" if not stale else "but should be more"
+            staleness = staleness if staleness > timedelta(0) else -staleness
+            reason = f"most recent data is {precisedelta(staleness)} {timeline}, {reason_part} than {precisedelta(max_source_staleness)} {timeline}"
+
+        statuses.append(
+            dict(
+                staleness=staleness,
+                stale=stale,
+                staleness_since=staleness_since,
+                reason=reason,
+                source_type=source_type,
+            )
+        )
+
+    return statuses
 
 
 def _get_sensor_asset_relation(
     asset: Asset,
     sensor: Sensor,
     inflexible_device_sensors: list[Sensor],
+    context_sensors: dict[str, Sensor],
 ) -> str:
     """Get the relation of a sensor to an asset."""
     relations = list()
-    consumption_price_sensor = asset.get_consumption_price_sensor()
-    production_price_sensor = asset.get_production_price_sensor()
     if sensor.generic_asset_id == asset.id:
-        relations.append("included device")
-    if (
-        consumption_price_sensor is not None
-        and consumption_price_sensor.id == sensor.id
-    ):
-        relations.append("consumption price")
-    if production_price_sensor is not None and production_price_sensor.id == sensor.id:
-        relations.append("production price")
+        relations.append("sensor belongs to this asset")
     inflexible_device_sensors_ids = {sensor.id for sensor in inflexible_device_sensors}
     if sensor.id in inflexible_device_sensors_ids:
-        relations.append("inflexible device")
-
+        relations.append("flex context (inflexible device)")
+    for field, ctxt_sensor in context_sensors.items():
+        if sensor.id == ctxt_sensor.id:
+            relations.append(f"flex context ({field})")
     return ";".join(relations)
 
 
@@ -213,15 +269,17 @@ def build_sensor_status_data(
     asset: Asset,
     now: datetime = None,
 ) -> list[dict]:
-    """Get data connectivity status information for each sensor in given asset and its children
+    """Get data connectivity status information for each sensor split by source in given asset and its children
     Returns a list of dictionaries, each containing the following keys:
     - id: sensor id
     - name: sensor name
+    - resolution: sensor resolution
     - asset_name: asset name
     - staleness: staleness of the sensor (for how long the sensor data is stale)
     - stale: whether the sensor is stale
     - staleness_since: time since sensor data is considered stale
     - reason: reason for staleness
+    - source: source of the sensor data
     - relation: relation of the sensor to the asset
     """
     if not now:
@@ -229,9 +287,13 @@ def build_sensor_status_data(
 
     sensors = []
     sensor_ids = set()
-    production_price_sensor = asset.get_production_price_sensor()
-    consumption_price_sensor = asset.get_consumption_price_sensor()
     inflexible_device_sensors = asset.get_inflexible_device_sensors()
+    context_sensors = {
+        field: Sensor.query.get(asset.flex_context[field]["sensor"])
+        for field in asset.flex_context
+        if isinstance(asset.flex_context[field], dict)
+        and field != "inflexible-device-sensors"
+    }
     for asset, is_child_asset in (
         (asset, False),
         *[(child_asset, True) for child_asset in asset.child_assets],
@@ -239,25 +301,26 @@ def build_sensor_status_data(
         sensors_list = list(asset.sensors)
         if not is_child_asset:
             sensors_list += [
-                *asset.get_inflexible_device_sensors(),
-                production_price_sensor,
-                consumption_price_sensor,
+                *inflexible_device_sensors,
+                *context_sensors.values(),
             ]
         for sensor in sensors_list:
             if sensor is None or sensor.id in sensor_ids:
                 continue
-            sensor_status = get_status(
+            sensor_statuses = get_statuses(
                 sensor=sensor,
                 now=now,
             )
-            sensor_status["name"] = sensor.name
-            sensor_status["id"] = sensor.id
-            sensor_status["asset_name"] = sensor.generic_asset.name
-            sensor_status["relation"] = _get_sensor_asset_relation(
-                asset, sensor, inflexible_device_sensors
-            )
-            sensor_ids.add(sensor.id)
-            sensors.append(sensor_status)
+            for sensor_status in sensor_statuses:
+                sensor_status["id"] = sensor.id
+                sensor_status["name"] = sensor.name
+                sensor_status["resolution"] = sensor.event_resolution
+                sensor_status["asset_name"] = sensor.generic_asset.name
+                sensor_status["relation"] = _get_sensor_asset_relation(
+                    asset, sensor, inflexible_device_sensors, context_sensors
+                )
+                sensor_ids.add(sensor.id)
+                sensors.append(sensor_status)
     return sensors
 
 
@@ -330,6 +393,7 @@ def build_asset_jobs_data(
             metadata = json.dumps({**job.meta, "job_id": job.id}, default=str, indent=4)
             jobs_data.append(
                 {
+                    "job_id": job.id,
                     "metadata": metadata,
                     "queue": queue,
                     "asset_or_sensor_type": asset_or_sensor_type,
diff --git a/flexmeasures/ui/crud/assets/views.py b/flexmeasures/ui/crud/assets/views.py
index 6c03c4d6..5a69e0d0 100644
--- a/flexmeasures/ui/crud/assets/views.py
+++ b/flexmeasures/ui/crud/assets/views.py
@@ -4,9 +4,11 @@ from flask import url_for, current_app, request
 from flask_classful import FlaskView, route
 from flask_security import login_required, current_user
 from webargs.flaskparser import use_kwargs
+from werkzeug.exceptions import NotFound
 
 from flexmeasures.data import db
 from flexmeasures.auth.error_handling import unauthorized_handler
+from flexmeasures.auth.policy import check_access
 from flexmeasures.data.schemas import StartEndTimeSchema
 from flexmeasures.data.services.job_cache import NoRedisConfigured
 from flexmeasures.data.models.generic_assets import (
@@ -143,10 +145,11 @@ class AssetCrudUI(FlaskView):
     def status(self, id: str):
         """GET from /assets/<id>/status to show the staleness of the asset's sensors."""
 
-        get_asset_response = InternalApi().get(url_for("AssetAPI:fetch_one", id=id))
-        asset_dict = get_asset_response.json()
+        asset = GenericAsset.query.get(id)
+        if asset is None:
+            raise NotFound
+        check_access(asset, "read")
 
-        asset = process_internal_api_response(asset_dict, int(id), make_obj=True)
         status_data = build_sensor_status_data(asset)
 
         # add data about forecasting and scheduling jobs
diff --git a/flexmeasures/ui/templates/views/status.html b/flexmeasures/ui/templates/views/status.html
index edc0fb3f..a362f02d 100644
--- a/flexmeasures/ui/templates/views/status.html
+++ b/flexmeasures/ui/templates/views/status.html
@@ -18,15 +18,20 @@
     <div class="row">
         <div class="col-sm-2"></div>
         <div class="col-sm-8">
-            <div class="sensors-asset card">
-                <h3>Data connectivity for sensors of {{ asset.name }}</h3>
-                <table class="table table-striped paginate nav-on-click">
+            <div class="card">
+                <h4>Data connectivity for sensors of {{ asset.name }}
+                  <span class="fa fa-info-circle" title="This table shows the connectivity status for all sensors under or relevant to this asset. Click the table to see the actual data. Per sensor, we report the different data source types. We only explicitly support source types 'demo script', 'user', 'forecaster', 'scheduler' and 'reporter'.
+
+A red traffic light indicates that the last time a record was made was too long ago (it is 'stale'). Hover the light to learn more about why it is displayed red. It could be a source for errors down the line if that sensor data is necessary for forecasts and schedules to be made.
+
+How long ago is considered too long (stale) depends on the sensor, usually we use the sensor's resolution times two. If the source type means we expect future data ('forecaster', 'scheduler'), data should extend 12 hours into the future." style="margin-left: 10px;"></span>
+                </h4>
+                <table id="sensor_statuses" class="table table-striped paginate nav-on-click">
                     <thead>
                         <tr>
-                            <th>Name</th>
-                            <th>Asset name</th>
-                            <th>Relation to asset</th>
-                            <th class="no-sort" title="This is the knowledge time of the most recent event recorded">Time of last value</th>
+                            <th>Sensor</th>
+                            <th>Data source type</th>
+                            <th class="no-sort" title="This is the knowledge time of the most recent event recorded (potentially in the future, for forecasts and schedules)">Time of latest record</th>
                             <th class="text-right no-sort">Status</th>
                             <th class="d-none">URL</th>
                         </tr>
@@ -35,13 +40,15 @@
                         {% for sensor in sensors: %}
                         <tr title="View data">
                             <td>
-                                {{ sensor.name }}
+                                {% if loop.first or sensor.name != loop.previtem.name %}
+                                  {{ sensor.name }} (<a href="/sensors/{{ sensor.id }}">{{ sensor.id }}</a>)
+                                  {%- set asset_info = [ sensor.name, "(Asset: '", sensor.asset_name, "')" ]-%}
+                                  <span class="fa fa-info" title="Resolution: {{sensor.resolution | naturalized_timedelta }}, Asset: '{{sensor.asset_name}}', reason for listing here: {{sensor.relation}}" style="margin-left: 10px;"></span>               
+                                {% endif %}
                             </td>
                             <td>
-                                {{ sensor.asset_name }}
+                                {{ sensor.source_type }} 
                             </td>
-                            <td>
-                                {{ sensor.relation }}
                             <td>
                                 <span title="{{ sensor['staleness_since'] }}">{{ sensor["staleness_since"] | naturalized_datetime }}</span>
                             </td>
@@ -61,7 +68,13 @@
                         {% endfor %}
                     </tbody>
                 </table>
-                <h3>Latest jobs of {{ asset.name }}</h3>
+                <h4>Latest jobs of {{ asset.name }}
+                  <span class="fa fa-info-circle" title="This table shows forecasting or scheduling jobs for this asset.
+A red traffic light indicates something went wrong and should be reported to admins.
+
+Note that jobs do not live forever, so only rather recent jobs (usually younger than a day) are shown at all.
+                  " style="margin-left: 10px;"></span>
+                </h4>
                 <table id="scheduling_forecasting_jobs" class="table table-striped paginate nav-on-click">
                     <thead>
                         <tr>
@@ -130,14 +143,26 @@
                     </tbody>
                 </table>
                 {% if redis_connection_err is not none %}
-                    <div class="warning">{{ redis_connection_err }}</div>
+                    <div class="alert alert-warning">Warning: {{ redis_connection_err }}</div>
                 {% endif %}
             </div>
         </div>
     </div>
 </div>
 
-<!-- sort scheduling and forecasting jobs by hidden 'Created at Timestamp' column -->
+<!-- hide control elements -->
+<script>
+    $(document).ready(function() {
+        $('#sensor_statuses').DataTable({
+        "searching": false,
+        "paging": false,
+        "info": false,
+        "ordering": false
+    });
+});
+</script>
+
+<!-- sort scheduling and forecasting jobs by hidden 'Created at Timestamp' column, hide control elements -->
 <script>
     $(document).ready(function() {
         $('#scheduling_forecasting_jobs').DataTable({
@@ -147,7 +172,10 @@
                 "targets": 1,  // Target the visible "Created at" column
                 "orderData": 0  // Use data from the first column (UTC Timestamp) for sorting
             }
-        ]
+        ],
+        "searching": false,
+        "paging": false,
+        "info": false
     });
 });
 </script>


Test patch:
diff --git a/flexmeasures/api/common/schemas/tests/test_sensor_data_schema.py b/flexmeasures/api/common/schemas/tests/test_sensor_data_schema.py
index 4845309c..fe71e850 100644
--- a/flexmeasures/api/common/schemas/tests/test_sensor_data_schema.py
+++ b/flexmeasures/api/common/schemas/tests/test_sensor_data_schema.py
@@ -5,6 +5,7 @@ import pytz
 
 from marshmallow import ValidationError
 import pandas as pd
+from unittest import mock
 
 from flexmeasures.api.common.schemas.sensor_data import (
     SingleValueField,
@@ -15,8 +16,8 @@ from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.services.forecasting import create_forecasting_jobs
 from flexmeasures.data.services.scheduling import create_scheduling_job
 from flexmeasures.data.services.sensors import (
-    get_staleness,
-    get_status,
+    get_stalenesses,
+    get_statuses,
     build_sensor_status_data,
     build_asset_jobs_data,
 )
@@ -138,149 +139,207 @@ def test_value_field_invalid(deserialization_input, error_msg):
     assert error_msg in str(ve)
 
 
+# knowledge time 2016-01-01T12:00+01
 @pytest.mark.parametrize(
-    "now, sensor_type, source_name, expected_staleness, expected_stale",
+    "now, expected_staleness, expected_stale",
     [
         (
-            # Last event start at 2016-01-01T23:00+01, with knowledge time 2016-01-01T12:00+01, 12 hours from now
+            # Knowledge time 12 hours from now
             "2016-01-01T00:00+01",
-            "market",
-            None,
             None,  # Not known yet
             True,
         ),
         (
-            # Last event start at 2016-01-01T23:00+01, with knowledge time 2016-01-01T12:00+01, 12 hours and 18 minutes ago
+            # Knowledge time 12 hours and 18 minutes ago
             "2016-01-02T00:18+01",
-            "market",
-            None,
             timedelta(hours=12, minutes=18),
             True,
         ),
         (
-            # Last event start at 2016-01-01T23:00+01, with knowledge time 2016-01-01T12:00+01, 1 day and 12 hours ago
+            # Knowledge time 1 day and 12 hours ago
             "2016-01-03T00:00+01",
-            "market",
-            None,
             timedelta(days=1, hours=12),
             True,
         ),
         (
-            # Last event start at 2015-01-01T23:00:00+00:00, with knowledge time 2015-01-01T23:15:00+00:00, 40 minutes from now
-            "2015-01-01T22:35+01",
-            "production",
-            "Seita",
-            None,  # Not known yet
-            True,
-        ),
-        (
-            # Last event start at 2015-01-02T07:45+01, with knowledge time 2015-01-02T08:00+01, 40 minutes ago (but still less than max PT1H allowed)
-            "2015-01-02T08:40+01",
-            "production",
-            "Seita",
-            timedelta(minutes=40),
+            # Knowledge time 1 min ago
+            "2016-01-01T12:01+01",
+            timedelta(minutes=1),
             False,
         ),
+    ],
+)
+def test_get_status_single_source(
+    add_market_prices,
+    now,
+    expected_staleness,
+    expected_stale,
+):
+    sensor = add_market_prices["epex_da"]
+    staleness_search = dict()
+
+    now = pd.Timestamp(now)
+    stalenesses = get_stalenesses(
+        sensor=sensor, staleness_search=staleness_search, now=now
+    )
+    if stalenesses is not None:
+        stalenesses.pop("forecaster", None)
+
+    source_type_of_interest = "reporter"
+
+    if expected_staleness is None:
+        assert stalenesses is None
+    else:
+        assert stalenesses[source_type_of_interest] == (mock.ANY, expected_staleness)
+
+    status_specs = {
+        "staleness_search": staleness_search,
+        "max_staleness": "PT1H",
+        "max_future_staleness": "-PT12H",
+    }
+    assert StatusSchema().load(status_specs)
+
+    sensor_statuses = get_statuses(
+        sensor=sensor,
+        status_specs=status_specs,
+        now=now,
+    )
+
+    if not expected_staleness:
+        return  # the following
+
+    sensor_statuses = [
+        status
+        for status in sensor_statuses
+        if status["source_type"] == source_type_of_interest
+    ]
+    sensor_status = sensor_statuses[0]
+
+    assert sensor_status["staleness"] == expected_staleness
+    assert sensor_status["stale"] == expected_stale
+    if stalenesses is None:
+        assert sensor_status["source_type"] is None
+    else:
+        assert sensor_status["source_type"] == source_type_of_interest
+
+
+# both sources have the same data
+# max_staleness for forecaster is 12 hours
+# max_staleness for reporter is 1 day
+@pytest.mark.parametrize(
+    "now, expected_forecaster_staleness, expected_forecaster_stale, expect_forecaster_reason, expected_reporter_staleness, expected_reporter_stale, expect_reporter_reason",
+    [
         (
-            # Last event start of Seita's belief at 2015-01-01T23:00:00+00:00, with knowledge time 2015-01-01T23:15:00+00:00, 2 minutes from now
-            "2015-01-01T23:13+01",
-            "production",
-            "Seita",
-            None,  # Not known yet
+            # Both stale
+            # Last event start at 2016-01-02T23:00+01 10 hours from now,
+            # with knowledge time 2016-01-01T12:00+01, 1 day 1 hour ago
+            "2016-01-02T13:00+01",
+            timedelta(hours=10),
             True,
-        ),
-        (
-            # Last event start of Seita's belief at 2015-01-02T07:45+01, with knowledge time 2015-01-02T08:00+01, 4 hours and 42 minutes ago
-            "2015-01-02T12:42+01",
-            "production",
-            "Seita",
-            timedelta(hours=4, minutes=42),
+            "most recent data is 10 hours in the future, but should be more than 12 hours in the future",
+            timedelta(days=1, hours=1),
             True,
+            "most recent data is 1 day and 1 hour old, but should not be more than 1 day old",
         ),
         (
-            # Last event start of DummyScheduler's belief at 2016-01-02T07:45+01, with knowledge time 2016-01-02T08:00+01, 13 hours ago
-            "2016-01-02T21:00+01",
-            "production",
-            "DummySchedule",
+            # Both not stale
+            # Last event start at 2016-01-02T23:00+01 13 hours from now,
+            # with knowledge time 2016-01-01T12:00+01, 22 hours ago
+            "2016-01-02T10:00+01",
             timedelta(hours=13),
+            False,
+            "most recent data is 13 hours in the future, which is not less than 12 hours in the future",
+            timedelta(hours=22),
+            False,
+            "most recent data is 22 hours old, which is not more than 1 day old",
+        ),
+        (
+            # Reporter not stale, forecaster stale
+            # Last event start at 2016-01-02T23:00+01,
+            # with knowledge time 2016-01-01T12:00+01, 1 day ago
+            "2016-01-02T12:00+01",
+            timedelta(hours=11),
             True,
+            "most recent data is 11 hours in the future, but should be more than 12 hours in the future",
+            timedelta(days=1),
+            False,
+            "most recent data is 1 day old, which is not more than 1 day old",
         ),
         (
-            # Last event start at 2016-01-02T07:45+01, with knowledge time 2016-01-02T08:00+01, 13 hours ago
-            "2016-01-02T21:00+01",
-            "production",
+            # Both stale, no data in the future
+            # Last event start at 2016-01-02T23:00+01,
+            # with knowledge time 2016-01-01T12:00+01, 2 days ago
+            "2016-01-03T12:00+01",
             None,
-            timedelta(hours=13),
             True,
+            "Found no future data which this source should have",
+            timedelta(days=2),
+            True,
+            "most recent data is 2 days old, but should not be more than 1 day old",
         ),
     ],
 )
-def test_get_status(
+def test_get_status_multi_source(
     add_market_prices,
-    capacity_sensors,
     now,
-    sensor_type,
-    source_name,
-    expected_staleness,
-    expected_stale,
+    expected_forecaster_staleness,
+    expected_forecaster_stale,
+    expect_forecaster_reason,
+    expected_reporter_staleness,
+    expected_reporter_stale,
+    expect_reporter_reason,
 ):
-    if sensor_type == "market":
-        sensor = add_market_prices["epex_da"]
-        deserialized_staleness_search = dict()
-        serialized_staleness_search = {}
-    elif sensor_type == "production":
-        sensor = capacity_sensors["production"]
-        deserialized_staleness_search = dict()
-        serialized_staleness_search = {}
-        for source in sensor.data_sources:
-            print(source.name)
-            if source.name == source_name:
-                deserialized_staleness_search = dict(source=source)
-                serialized_staleness_search = {"source": source.id}
-
-    print(deserialized_staleness_search)
+    sensor = add_market_prices["epex_da"]
     now = pd.Timestamp(now)
-    staleness = get_staleness(
-        sensor=sensor, staleness_search=deserialized_staleness_search, now=now
-    )
-    status_specs = {
-        "staleness_search": serialized_staleness_search,
-        "max_staleness": "PT1H",
-    }
-    sensor_status = get_status(
+
+    sensor_statuses = get_statuses(
         sensor=sensor,
-        status_specs=status_specs,
         now=now,
     )
-
-    assert StatusSchema().load(status_specs)
-    assert staleness == expected_staleness
-    assert sensor_status["staleness"] == expected_staleness
-    assert sensor_status["stale"] == expected_stale
+    for sensor_status in sensor_statuses:
+        if sensor_status["source_type"] == "reporter":
+            assert sensor_status["staleness"] == expected_reporter_staleness
+            assert sensor_status["stale"] == expected_reporter_stale
+            assert sensor_status["reason"] == expect_reporter_reason
+        if sensor_status["source_type"] == "forecaster":
+            assert sensor_status["staleness"] == expected_forecaster_staleness
+            assert sensor_status["stale"] == expected_forecaster_stale
+            assert sensor_status["reason"] == expect_forecaster_reason
 
 
 @pytest.mark.parametrize(
-    "now, expected_staleness, expected_stale, expected_stale_reason",
+    "source_type, now, expected_staleness, expected_stale, expected_stale_reason",
     [
         # sensor resolution is 15 min
         (
-            # Last event start at 2016-01-02T07:45+01, with knowledge time 2016-01-02T08:00+01, 29 minutes ago
-            "2016-01-02T08:29+01",
+            "demo script",
+            # Last event start (in the past) at 2015-01-02T07:45+01, with knowledge time 2015-01-02T08:00+01, 29 minutes ago
+            "2015-01-02T08:29+01",
             timedelta(minutes=29),
             False,
             "not more than 30 minutes old",
         ),
         (
-            # Last event start at 2016-01-02T07:45+01, with knowledge time 2016-01-02T08:00+01, 31 minutes ago
-            "2016-01-02T08:31+01",
+            "demo script",
+            # Last event start (in the past) at 2015-01-02T07:45+01, with knowledge time 2015-01-02T08:00+01, 31 minutes ago
+            "2015-01-02T08:31+01",
             timedelta(minutes=31),
             True,
             "more than 30 minutes old",
         ),
+        (
+            "scheduler",
+            # Last event start (in the future) at 2016-01-02T07:45+01, in 24 hours 45 minutes
+            "2016-01-01T07:00+01",
+            timedelta(minutes=24 * 60 + 45),
+            False,
+            "not less than 12 hours in the future",
+        ),
     ],
 )
 def test_get_status_no_status_specs(
     capacity_sensors,
+    source_type,
     now,
     expected_staleness,
     expected_stale,
@@ -288,20 +347,30 @@ def test_get_status_no_status_specs(
 ):
     sensor = capacity_sensors["production"]
     now = pd.Timestamp(now)
-    sensor_status = get_status(
+    sensor_statuses = get_statuses(
         sensor=sensor,
         status_specs=None,
         now=now,
     )
 
-    assert sensor_status["staleness"] == expected_staleness
-    assert sensor_status["stale"] == expected_stale
-    assert sensor_status["reason"] == expected_stale_reason
+    assert source_type in [ss["source_type"] for ss in sensor_statuses]
+    for sensor_status in sensor_statuses:
+        if sensor_status["source_type"] == source_type:
+            assert sensor_status["staleness"] == expected_staleness
+            assert sensor_status["stale"] == expected_stale
+            assert expected_stale_reason in sensor_status["reason"]
 
 
 def test_build_asset_status_data(
-    db, mock_get_status, add_weather_sensors, add_battery_assets
+    db, mock_get_statuses, add_weather_sensors, add_battery_assets
 ):
+    """
+    Test the function to build status data structure, using a weather station asset.
+    We include the sensor of a different asset (a battery) via the flex context
+    (as production price, does not make too much sense actually).
+    One sensor which the asset already includes is also set in the context as inflexible device,
+    so we can test if the relationship tagging works for that as well.
+    """
     asset = add_weather_sensors["asset"]
     battery_asset = add_battery_assets["Test battery"]
     wind_sensor, temperature_sensor = (
@@ -318,41 +387,41 @@ def test_build_asset_status_data(
     db.session.add(production_price_sensor)
     db.session.flush()
 
-    asset.flex_context["consumption-price-sensor"] = wind_sensor.id
-    asset.flex_context["production-price-sensor"] = production_price_sensor.id
+    asset.flex_context["production-price"] = {"sensor": production_price_sensor.id}
     asset.flex_context["inflexible-device-sensors"] = [temperature_sensor.id]
     db.session.add(asset)
 
     wind_speed_res, temperature_res = {"staleness": True}, {"staleness": False}
     production_price_res = {"staleness": True}
-    mock_get_status.side_effect = (
-        wind_speed_res,
-        temperature_res,
-        production_price_res,
+    mock_get_statuses.side_effect = (
+        [wind_speed_res],
+        [temperature_res],
+        [production_price_res],
     )
 
     status_data = build_sensor_status_data(asset=asset)
+
     assert status_data == [
         {
             **wind_speed_res,
             "name": "wind speed",
             "id": wind_sensor.id,
             "asset_name": asset.name,
-            "relation": "included device;consumption price",
+            "relation": "sensor belongs to this asset",
         },
         {
             **temperature_res,
             "name": "temperature",
             "id": temperature_sensor.id,
             "asset_name": asset.name,
-            "relation": "included device;inflexible device",
+            "relation": "sensor belongs to this asset;flex context (inflexible device)",
         },
         {
             **production_price_res,
             "name": "production price",
             "id": production_price_sensor.id,
             "asset_name": battery_asset.name,
-            "relation": "production price",
+            "relation": "flex context (production-price)",
         },
     ]
 
diff --git a/flexmeasures/api/conftest.py b/flexmeasures/api/conftest.py
index 27724558..cb62f3f3 100644
--- a/flexmeasures/api/conftest.py
+++ b/flexmeasures/api/conftest.py
@@ -1,5 +1,5 @@
 import pytest
-
+from pytest_mock import MockerFixture
 from flask_login import login_user, logout_user
 
 from flexmeasures.api.tests.utils import UserContext
@@ -37,3 +37,10 @@ def requesting_user(request):
             logout_user()
     else:
         yield
+
+
+@pytest.fixture
+def mock_get_statuses(mocker: MockerFixture):
+    return mocker.patch(
+        "flexmeasures.data.services.sensors.get_statuses", autospec=True
+    )
diff --git a/flexmeasures/conftest.py b/flexmeasures/conftest.py
index 2593ba5c..cb551dbc 100644
--- a/flexmeasures/conftest.py
+++ b/flexmeasures/conftest.py
@@ -11,7 +11,6 @@ import numpy as np
 from flask import request, jsonify
 from flask_sqlalchemy import SQLAlchemy
 from flask_security import roles_accepted
-from pytest_mock import MockerFixture
 from timely_beliefs.sensors.func_store.knowledge_horizons import x_days_ago_at_y_oclock
 
 from werkzeug.exceptions import (
@@ -367,12 +366,18 @@ def create_sources(db) -> dict[str, DataSource]:
     db.session.add(seita_source)
     entsoe_source = DataSource(name="ENTSO-E", type="demo script")
     db.session.add(entsoe_source)
-    dummy_schedule_source = DataSource(name="DummySchedule", type="demo script")
+    dummy_schedule_source = DataSource(name="DummySchedule", type="scheduler")
     db.session.add(dummy_schedule_source)
+    forecaster_source = DataSource(name="forecaster name", type="forecaster")
+    db.session.add(forecaster_source)
+    reporter_source = DataSource(name="reporter name", type="reporter")
+    db.session.add(reporter_source)
     return {
         "Seita": seita_source,
         "ENTSO-E": entsoe_source,
         "DummySchedule": dummy_schedule_source,
+        "forecaster": forecaster_source,
+        "reporter": reporter_source,
     }
 
 
@@ -733,6 +738,30 @@ def add_market_prices_common(
     ]
     db.session.add_all(today_beliefs)
 
+    today_forecaster_beliefs = [
+        TimedBelief(
+            event_start=dt,
+            belief_horizon=timedelta(hours=0),
+            event_value=val,
+            source=setup_sources["forecaster"],
+            sensor=setup_markets["epex_da"],
+        )
+        for dt, val in zip(time_slots, values_today)
+    ]
+    db.session.add_all(today_forecaster_beliefs)
+
+    today_reporter_beliefs = [
+        TimedBelief(
+            event_start=dt,
+            belief_horizon=timedelta(hours=0),
+            event_value=val,
+            source=setup_sources["reporter"],
+            sensor=setup_markets["epex_da"],
+        )
+        for dt, val in zip(time_slots, values_today)
+    ]
+    db.session.add_all(today_reporter_beliefs)
+
     return {
         "epex_da": setup_markets["epex_da"],
         "epex_da_production": setup_markets["epex_da_production"],
@@ -1455,8 +1484,3 @@ def add_beliefs(
         for dt, val in zip(time_slots, values)
     ]
     db.session.add_all(beliefs)
-
-
-@pytest.fixture
-def mock_get_status(mocker: MockerFixture):
-    return mocker.patch("flexmeasures.data.services.sensors.get_status", autospec=True)