11295014679

Committed 11 Oct 2024 03:24PM UTC coverage: 67.508% (+0.2%) from 67.29%

Build # 11295014679

Build Type

push

github

Committed by

web-flow

Commit Message

feat: search filter on sensor API (#1191)

* feat: search filter on sensor API

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* feat: sensor API pagintion and search

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* refactor: reduce code repitition and code size

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* refactor: request changes

- relocated SearchField to dedicated file
- worked on event resolution formating

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* feat: api to fetch all sensors under an asset

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* feat: added filter feature for sensor table, both name and unit filtering

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* feat: unit filter features on get sensors API

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* refactor: asset sensor and sensors API changes

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* feat: test case for asset_sensors API

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* chore: few changes variable namings and import arrangement

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* refactor: updated testcase with more checks and removed redundant code

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* chore: little changes on test(waiting for comment resolution)

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* chore: moved formatResolution func to base.html

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* refactor: changes to asset sensors API and ISO formatter logic on forntend

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* chore: removed hard coded asset ID in asset sensors test case

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* feat: test to get all sensors

Signed-off-by: joshuaunity <oghenerobojosh01@gmail.com>

* fix: fix wrong imports

Si... (continued)

Run Details

6333 of 10209 branches covered (62.03%)

Branch coverage included in aggregate %.

72 of 82 new or added lines in 9 files covered. (87.8%)

3 existing lines in 2 files now uncovered.

7955 of 10956 relevant lines covered (72.61%)

5.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.93

/flexmeasures/data/services/sensors.py

from __future__ import annotations

import json
import hashlib
from datetime import datetime, timedelta
from flask import current_app
from functools import lru_cache
from isodate import duration_isoformat
import time
from timely_beliefs import BeliefsDataFrame

from humanize.time import naturaldelta

from flexmeasures.data.models.time_series import TimedBelief


import sqlalchemy as sa

from flexmeasures.data import db
from flexmeasures import Sensor, Account, Asset
from flexmeasures.data.models.data_sources import DataSource
from flexmeasures.data.models.generic_assets import GenericAsset
from flexmeasures.data.schemas.reporting import StatusSchema
from flexmeasures.utils.time_utils import server_now


def get_sensors(
    account: Account | list[Account] | None,
    include_public_assets: bool = False,
    sensor_id_allowlist: list[int] | None = None,
    sensor_name_allowlist: list[str] | None = None,
) -> list[Sensor]:
    """Return a list of Sensor objects that belong to the given account, and/or public sensors.

    :param account:                 select only sensors from this account (or list of accounts)
    :param include_public_assets:   if True, include sensors that belong to a public asset
    :param sensor_id_allowlist:     optionally, allow only sensors whose id is in this list
    :param sensor_name_allowlist:   optionally, allow only sensors whose name is in this list
    """
    sensor_query = sa.select(Sensor)
    if isinstance(account, list):
        accounts = account
    else:
        accounts: list = [account] if account else []
    account_ids: list = [acc.id for acc in accounts]

    sensor_query = sensor_query.join(
        GenericAsset, GenericAsset.id == Sensor.generic_asset_id
    ).filter(Sensor.generic_asset_id == GenericAsset.id)
    if include_public_assets:
        sensor_query = sensor_query.filter(
            sa.or_(
                GenericAsset.account_id.in_(account_ids),
                GenericAsset.account_id.is_(None),
            )
        )
    else:
        sensor_query = sensor_query.filter(GenericAsset.account_id.in_(account_ids))
    if sensor_id_allowlist:
        sensor_query = sensor_query.filter(Sensor.id.in_(sensor_id_allowlist))
    if sensor_name_allowlist:
        sensor_query = sensor_query.filter(Sensor.name.in_(sensor_name_allowlist))

    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.
    """
    bdf = TimedBelief.search(
        sensors=sensor,
        most_recent_beliefs_only=False,
        most_recent_only=True,
        **staleness_search,
    )
    if bdf.empty:
        return None
    return bdf


def get_most_recent_knowledge_time(
    sensor: Sensor, staleness_search: dict
) -> datetime | None:
    """Get 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]


def get_staleness(
    sensor: Sensor, staleness_search: dict, now: datetime
) -> timedelta | None:
    """Get the staleness of the sensor.

    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,
    as long as the available data is about what we should be able to know by now.

    :param sensor:              The sensor to compute the staleness for.
    :param staleness_search:    Deserialized keyword arguments to `TimedBelief.search`.
    :param now:                 Datetime representing now, used both to mask future beliefs,
                                and to measures staleness against.
    """

    # 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_start_time = get_most_recent_knowledge_time(
        sensor=sensor, staleness_search=staleness_search
    )
    if staleness_start_time is not None:
        staleness = now - staleness_start_time
    else:
        staleness = None

    return staleness


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),
            }
    return status_specs


def get_status(
    sensor: Sensor,
    now: datetime,
    status_specs: dict | None = None,
) -> dict:
    """Get the status of the sensor
    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.
    """
    if status_specs is None:
        status_specs = get_status_specs(sensor=sensor)
    status_specs = StatusSchema().load(status_specs)
    max_staleness = status_specs.pop("max_staleness")
    staleness_search = status_specs.pop("staleness_search")
    staleness: timedelta = get_staleness(
        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


def _get_sensor_asset_relation(
    asset: Asset,
    sensor: Sensor,
    inflexible_device_sensors: list[Sensor],
) -> str:
    """Get the relation of a sensor to an asset."""
    relations = list()
    if sensor.generic_asset_id == asset.id:
        relations.append("included device")
    if asset.consumption_price_sensor_id == sensor.id:
        relations.append("consumption price")
    if asset.production_price_sensor_id == sensor.id:
        relations.append("production price")
    inflexible_device_sensors_ids = {sensor.id for sensor in inflexible_device_sensors}
    if sensor.id in inflexible_device_sensors_ids:
        relations.append("inflexible device")

    return ";".join(relations)


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
    Returns a list of dictionaries, each containing the following keys:
    - id: sensor id
    - name: sensor name
    - 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
    - relation: relation of the sensor to the asset
    """
    if not now:
        now = server_now()

    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()
    for asset, is_child_asset in (
        (asset, False),
        *[(child_asset, True) for child_asset in asset.child_assets],
    ):
        sensors_list = list(asset.sensors)
        if not is_child_asset:
            sensors_list += [
                *asset.inflexible_device_sensors,
                production_price_sensor,
                consumption_price_sensor,
            ]
        for sensor in sensors_list:
            if sensor is None or sensor.id in sensor_ids:
                continue
            sensor_status = get_status(
                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)
    return sensors


def build_asset_jobs_data(
    asset: Asset,
) -> list[dict]:
    """Get all jobs data for an asset
    Returns a list of dictionaries, each containing the following keys:
    - job_id: id of a job
    - queue: job queue (scheduling or forecasting)
    - asset_or_sensor_type: type of an asset that is linked to the job (asset or sensor)
    - asset_id: id of sensor or asset
    - status: job status (e.g finished, failed, etc)
    - err: job error (equals to None when there was no error for a job)
    - enqueued_at: time when the job was enqueued
    - metadata_hash: hash of job metadata (internal field)
    """

    jobs = list()

    # try to get scheduling jobs for asset first (only scheduling jobs can be stored by asset id)
    jobs.append(
        (
            "scheduling",
            "asset",
            asset.id,
            asset.name,
            current_app.job_cache.get(asset.id, "scheduling", "asset"),
        )
    )

    for sensor in asset.sensors:
        jobs.append(
            (
                "scheduling",
                "sensor",
                sensor.id,
                sensor.name,
                current_app.job_cache.get(sensor.id, "scheduling", "sensor"),
            )
        )
        jobs.append(
            (
                "forecasting",
                "sensor",
                sensor.id,
                sensor.name,
                current_app.job_cache.get(sensor.id, "forecasting", "sensor"),
            )
        )

    jobs_data = list()
    # Building the actual return list - we also unpack lists of jobs, each to its own entry, and we add error info
    for queue, asset_or_sensor_type, entity_id, entity_name, jobs in jobs:
        for job in jobs:
            e = job.meta.get(
                "exception",
                Exception(
                    "The job does not state why it failed. "
                    "The worker may be missing an exception handler, "
                    "or its exception handler is not storing the exception as job meta data."
                ),
            )
            job_err = (
                f"Scheduling job failed with {type(e).__name__}: {e}"
                if job.is_failed
                else None
            )

            metadata = json.dumps({**job.meta, "job_id": job.id}, default=str, indent=4)
            jobs_data.append(
                {
                    "metadata": metadata,
                    "queue": queue,
                    "asset_or_sensor_type": asset_or_sensor_type,
                    "entity": f"{asset_or_sensor_type}: {entity_name} (Id: {entity_id})",
                    "status": job.get_status(),
                    "err": job_err,
                    "enqueued_at": job.enqueued_at,
                    "metadata_hash": hashlib.sha256(metadata.encode()).hexdigest(),
                }
            )

    return jobs_data


@lru_cache()
def _get_sensor_stats(sensor: Sensor, ttl_hash=None) -> dict:
    # Subquery for filtered aggregates
    subquery_for_filtered_aggregates = (
        sa.select(
            TimedBelief.source_id,
            sa.func.max(TimedBelief.event_value).label("max_event_value"),
            sa.func.avg(TimedBelief.event_value).label("avg_event_value"),
            sa.func.sum(TimedBelief.event_value).label("sum_event_value"),
            sa.func.min(TimedBelief.event_value).label("min_event_value"),
        )
        .filter(TimedBelief.event_value != float("NaN"))
        .filter(TimedBelief.sensor_id == sensor.id)
        .group_by(TimedBelief.source_id)
        .subquery()
    )

    raw_stats = db.session.execute(
        sa.select(
            DataSource.name,
            sa.func.min(TimedBelief.event_start).label("min_event_start"),
            sa.func.max(TimedBelief.event_start).label("max_event_start"),
            subquery_for_filtered_aggregates.c.min_event_value,
            subquery_for_filtered_aggregates.c.max_event_value,
            subquery_for_filtered_aggregates.c.avg_event_value,
            subquery_for_filtered_aggregates.c.sum_event_value,
            sa.func.count(TimedBelief.event_value).label("count_event_value"),
        )
        .select_from(TimedBelief)
        .join(DataSource, DataSource.id == TimedBelief.source_id)
        .join(
            subquery_for_filtered_aggregates,
            subquery_for_filtered_aggregates.c.source_id == TimedBelief.source_id,
        )
        .filter(TimedBelief.sensor_id == sensor.id)
        .group_by(
            DataSource.name,
            subquery_for_filtered_aggregates.c.min_event_value,
            subquery_for_filtered_aggregates.c.max_event_value,
            subquery_for_filtered_aggregates.c.avg_event_value,
            subquery_for_filtered_aggregates.c.sum_event_value,
        )
    ).fetchall()

    stats = dict()
    for row in raw_stats:
        (
            data_source,
            min_event_start,
            max_event_start,
            min_value,
            max_value,
            mean_value,
            sum_values,
            count_values,
        ) = row
        stats[data_source] = {
            "min_event_start": min_event_start,
            "max_event_start": max_event_start,
            "min_value": min_value,
            "max_value": max_value,
            "mean_value": mean_value,
            "sum_values": sum_values,
            "count_values": count_values,
        }
    return stats


def _get_ttl_hash(seconds=120) -> int:
    """Returns the same value within "seconds" time period
    Is needed to make LRU cache a TTL one
    (lru_cache is used when call arguments are the same,
    here we ensure that call arguments are the same in "seconds" period of time).
    """
    return round(time.time() / seconds)


def get_sensor_stats(sensor: Sensor) -> dict:
    """Get stats for a sensor"""
    return _get_sensor_stats(sensor, ttl_hash=_get_ttl_hash())

1	from __future__ import annotations	8✔
2
3	import json	8✔
4	import hashlib	8✔
5	from datetime import datetime, timedelta	8✔
6	from flask import current_app	8✔
7	from functools import lru_cache	8✔
8	from isodate import duration_isoformat	8✔
9	import time	8✔
10	from timely_beliefs import BeliefsDataFrame	8✔
11
12	from humanize.time import naturaldelta	8✔
13
14	from flexmeasures.data.models.time_series import TimedBelief	8✔
15
16
17	import sqlalchemy as sa	8✔
18
19	from flexmeasures.data import db	8✔
20	from flexmeasures import Sensor, Account, Asset	8✔
21	from flexmeasures.data.models.data_sources import DataSource	8✔
22	from flexmeasures.data.models.generic_assets import GenericAsset	8✔
23	from flexmeasures.data.schemas.reporting import StatusSchema	8✔
24	from flexmeasures.utils.time_utils import server_now	8✔
25
26
27	def get_sensors(	8✔
28	account: Account \| list[Account] \| None,
29	include_public_assets: bool = False,
30	sensor_id_allowlist: list[int] \| None = None,
31	sensor_name_allowlist: list[str] \| None = None,
32	) -> list[Sensor]:
33	"""Return a list of Sensor objects that belong to the given account, and/or public sensors.
34
35	:param account: select only sensors from this account (or list of accounts)
36	:param include_public_assets: if True, include sensors that belong to a public asset
37	:param sensor_id_allowlist: optionally, allow only sensors whose id is in this list
38	:param sensor_name_allowlist: optionally, allow only sensors whose name is in this list
39	"""
40	sensor_query = sa.select(Sensor)	×
NEW 41	if isinstance(account, list):	×
NEW 42	accounts = account	×
43	else:
NEW 44	accounts: list = [account] if account else []	×
NEW 45	account_ids: list = [acc.id for acc in accounts]	×
46
UNCOV 47	sensor_query = sensor_query.join(	×
48	GenericAsset, GenericAsset.id == Sensor.generic_asset_id
49	).filter(Sensor.generic_asset_id == GenericAsset.id)
50	if include_public_assets:	×
51	sensor_query = sensor_query.filter(	×
52	sa.or_(
53	GenericAsset.account_id.in_(account_ids),
54	GenericAsset.account_id.is_(None),
55	)
56	)
57	else:
58	sensor_query = sensor_query.filter(GenericAsset.account_id.in_(account_ids))	×
59	if sensor_id_allowlist:	×
60	sensor_query = sensor_query.filter(Sensor.id.in_(sensor_id_allowlist))	×
61	if sensor_name_allowlist:	×
62	sensor_query = sensor_query.filter(Sensor.name.in_(sensor_name_allowlist))	×
63
UNCOV 64	return db.session.scalars(sensor_query).all()	×
65
66
67	def _get_sensor_bdf(sensor: Sensor, staleness_search: dict) -> BeliefsDataFrame \| None:	8✔
68	"""
69	Retrieve the BeliefsDataFrame for a given sensor using the specified search parameters. The 'most_recent_only'
70	parameter is set to True, which ensures that only the most recent belief from the most recent event is returned.
71	"""
72	bdf = TimedBelief.search(	8✔
73	sensors=sensor,
74	most_recent_beliefs_only=False,
75	most_recent_only=True,
76	**staleness_search,
77	)
78	if bdf.empty:	8✔
79	return None	8✔
80	return bdf	8✔
81
82
83	def get_most_recent_knowledge_time(	8✔
84	sensor: Sensor, staleness_search: dict
85	) -> datetime \| None:
86	"""Get the knowledge time of the sensor's most recent event.
87
88	This knowledge time represents when you could have known about the event
89	(specifically, when you could have formed an ex-post belief about it).
90	"""
91	staleness_bdf = _get_sensor_bdf(sensor=sensor, staleness_search=staleness_search)	8✔
92	return None if staleness_bdf is None else staleness_bdf.knowledge_times[-1]	8✔
93
94
95	def get_staleness(	8✔
96	sensor: Sensor, staleness_search: dict, now: datetime
97	) -> timedelta \| None:
98	"""Get the staleness of the sensor.
99
100	The staleness is defined relative to the knowledge time of the most recent event, rather than to its belief time.
101	Basically, that means that we don't really care when the data arrived,
102	as long as the available data is about what we should be able to know by now.
103
104	:param sensor: The sensor to compute the staleness for.
105	:param staleness_search: Deserialized keyword arguments to `TimedBelief.search`.
106	:param now: Datetime representing now, used both to mask future beliefs,
107	and to measures staleness against.
108	"""
109
110	# Mask beliefs before now
111	staleness_search = staleness_search.copy() # no inplace operations	8✔
112	beliefs_before = staleness_search.get("beliefs_before")	8✔
113	if beliefs_before is not None:	8✔
114	staleness_search["beliefs_before"] = min(beliefs_before, now)	×
115	else:
116	staleness_search["beliefs_before"] = now	8✔
117
118	staleness_start_time = get_most_recent_knowledge_time(	8✔
119	sensor=sensor, staleness_search=staleness_search
120	)
121	if staleness_start_time is not None:	8✔
122	staleness = now - staleness_start_time	8✔
123	else:
124	staleness = None	8✔
125
126	return staleness	8✔
127
128
129	def get_status_specs(sensor: Sensor) -> dict:	8✔
130	"""Get status specs from a given sensor."""
131
132	# Check for explicitly defined status specs
133	status_specs = sensor.attributes.get("status_specs")	8✔
134	if status_specs is None:	8✔
135	# Default to status specs for economical sensors with daily updates
136	if sensor.knowledge_horizon_fnc == "x_days_ago_at_y_oclock":	8✔
137	status_specs = {"staleness_search": {}, "max_staleness": "P1D"}	×
138	else:
139	# Default to status specs indicating staleness after knowledge time + 2 sensor resolutions
140	status_specs = {	8✔
141	"staleness_search": {},
142	"max_staleness": duration_isoformat(sensor.event_resolution * 2),
143	}
144	return status_specs	8!
145
146
147	def get_status(	8✔
148	sensor: Sensor,
149	now: datetime,
150	status_specs: dict \| None = None,
151	) -> dict:
152	"""Get the status of the sensor
153	Main part of result here is a stale value, which is True if the sensor is stale, False otherwise.
154	Other values are just context information for the stale value.
155	"""
156	if status_specs is None:	8✔
157	status_specs = get_status_specs(sensor=sensor)	8✔
158	status_specs = StatusSchema().load(status_specs)	8✔
159	max_staleness = status_specs.pop("max_staleness")	8✔
160	staleness_search = status_specs.pop("staleness_search")	8✔
161	staleness: timedelta = get_staleness(	8✔
162	sensor=sensor,
163	staleness_search=staleness_search,
164	now=now,
165	)
166	if staleness is not None:	8✔
167	staleness_since = now - staleness	8✔
168	stale = staleness > max_staleness	8✔
169	reason = (	8✔
170	"" if stale else "not "
171	) + f"more than {naturaldelta(max_staleness)} old"
172	staleness = staleness if staleness > timedelta(0) else -staleness	8✔
173	else:
174	staleness_since = None	8✔
175	stale = True	8✔
176	reason = "no data recorded"	8✔
177	status = dict(	8✔
178	staleness=staleness,
179	stale=stale,
180	staleness_since=staleness_since,
181	reason=reason,
182	)
183	return status	8✔
184
185
186	def _get_sensor_asset_relation(	8✔
187	asset: Asset,
188	sensor: Sensor,
189	inflexible_device_sensors: list[Sensor],
190	) -> str:
191	"""Get the relation of a sensor to an asset."""
192	relations = list()	8✔
193	if sensor.generic_asset_id == asset.id:	8✔
194	relations.append("included device")	8✔
195	if asset.consumption_price_sensor_id == sensor.id:	8✔
196	relations.append("consumption price")	8✔
197	if asset.production_price_sensor_id == sensor.id:	8✔
198	relations.append("production price")	8✔
199	inflexible_device_sensors_ids = {sensor.id for sensor in inflexible_device_sensors}	8✔
200	if sensor.id in inflexible_device_sensors_ids:	8✔
201	relations.append("inflexible device")	8✔
202
203	return ";".join(relations)	8✔
204
205
206	def build_sensor_status_data(	8✔
207	asset: Asset,
208	now: datetime = None,
209	) -> list[dict]:
210	"""Get data connectivity status information for each sensor in given asset and its children
211	Returns a list of dictionaries, each containing the following keys:
212	- id: sensor id
213	- name: sensor name
214	- asset_name: asset name
215	- staleness: staleness of the sensor (for how long the sensor data is stale)
216	- stale: whether the sensor is stale
217	- staleness_since: time since sensor data is considered stale
218	- reason: reason for staleness
219	- relation: relation of the sensor to the asset
220	"""
221	if not now:	8✔
222	now = server_now()	8✔
223
224	sensors = []	8!
225	sensor_ids = set()	8✔
226	production_price_sensor = asset.get_production_price_sensor()	8✔
227	consumption_price_sensor = asset.get_consumption_price_sensor()	8✔
228	inflexible_device_sensors = asset.get_inflexible_device_sensors()	8✔
229	for asset, is_child_asset in (	8✔
230	(asset, False),
231	*[(child_asset, True) for child_asset in asset.child_assets],
232	):
233	sensors_list = list(asset.sensors)	8✔
234	if not is_child_asset:	8✔
235	sensors_list += [	8✔
236	*asset.inflexible_device_sensors,
237	production_price_sensor,
238	consumption_price_sensor,
239	]
240	for sensor in sensors_list:	8!
241	if sensor is None or sensor.id in sensor_ids:	8✔
242	continue	8✔
243	sensor_status = get_status(	8✔
244	sensor=sensor,
245	now=now,
246	)
247	sensor_status["name"] = sensor.name	8✔
248	sensor_status["id"] = sensor.id	8✔
249	sensor_status["asset_name"] = sensor.generic_asset.name	8✔
250	sensor_status["relation"] = _get_sensor_asset_relation(	8✔
251	asset, sensor, inflexible_device_sensors
252	)
253	sensor_ids.add(sensor.id)	8✔
254	sensors.append(sensor_status)	8✔
255	return sensors	8✔
256
257
258	def build_asset_jobs_data(	8✔
259	asset: Asset,
260	) -> list[dict]:
261	"""Get all jobs data for an asset
262	Returns a list of dictionaries, each containing the following keys:
263	- job_id: id of a job
264	- queue: job queue (scheduling or forecasting)
265	- asset_or_sensor_type: type of an asset that is linked to the job (asset or sensor)
266	- asset_id: id of sensor or asset
267	- status: job status (e.g finished, failed, etc)
268	- err: job error (equals to None when there was no error for a job)
269	- enqueued_at: time when the job was enqueued
270	- metadata_hash: hash of job metadata (internal field)
271	"""
272
273	jobs = list()	8✔
274
275	# try to get scheduling jobs for asset first (only scheduling jobs can be stored by asset id)
276	jobs.append(	8✔
277	(
278	"scheduling",
279	"asset",
280	asset.id,
281	asset.name,
282	current_app.job_cache.get(asset.id, "scheduling", "asset"),
283	)
284	)
285
286	for sensor in asset.sensors:	8✔
287	jobs.append(	8✔
288	(
289	"scheduling",
290	"sensor",
291	sensor.id,
292	sensor.name,
293	current_app.job_cache.get(sensor.id, "scheduling", "sensor"),
294	)
295	)
296	jobs.append(	8✔
297	(
298	"forecasting",
299	"sensor",
300	sensor.id,
301	sensor.name,
302	current_app.job_cache.get(sensor.id, "forecasting", "sensor"),
303	)
304	)
305
306	jobs_data = list()	8✔
307	# Building the actual return list - we also unpack lists of jobs, each to its own entry, and we add error info
308	for queue, asset_or_sensor_type, entity_id, entity_name, jobs in jobs:	8✔
309	for job in jobs:	8✔
310	e = job.meta.get(	8✔
311	"exception",
312	Exception(
313	"The job does not state why it failed. "
314	"The worker may be missing an exception handler, "
315	"or its exception handler is not storing the exception as job meta data."
316	),
317	)
318	job_err = (	8✔
319	f"Scheduling job failed with {type(e).__name__}: {e}"
320	if job.is_failed
321	else None
322	)
323
324	metadata = json.dumps({**job.meta, "job_id": job.id}, default=str, indent=4)	8✔
325	jobs_data.append(	8✔
326	{
327	"metadata": metadata,
328	"queue": queue,
329	"asset_or_sensor_type": asset_or_sensor_type,
330	"entity": f"{asset_or_sensor_type}: {entity_name} (Id: {entity_id})",
331	"status": job.get_status(),
332	"err": job_err,
333	"enqueued_at": job.enqueued_at,
334	"metadata_hash": hashlib.sha256(metadata.encode()).hexdigest(),
335	}
336	)
337
338	return jobs_data	8✔
339
340
341	@lru_cache()	8✔
342	def _get_sensor_stats(sensor: Sensor, ttl_hash=None) -> dict:	8✔
343	# Subquery for filtered aggregates
344	subquery_for_filtered_aggregates = (	8✔
345	sa.select(
346	TimedBelief.source_id,
347	sa.func.max(TimedBelief.event_value).label("max_event_value"),
348	sa.func.avg(TimedBelief.event_value).label("avg_event_value"),
349	sa.func.sum(TimedBelief.event_value).label("sum_event_value"),
350	sa.func.min(TimedBelief.event_value).label("min_event_value"),
351	)
352	.filter(TimedBelief.event_value != float("NaN"))
353	.filter(TimedBelief.sensor_id == sensor.id)
354	.group_by(TimedBelief.source_id)
355	.subquery()
356	)
357
358	raw_stats = db.session.execute(	8✔
359	sa.select(
360	DataSource.name,
361	sa.func.min(TimedBelief.event_start).label("min_event_start"),
362	sa.func.max(TimedBelief.event_start).label("max_event_start"),
363	subquery_for_filtered_aggregates.c.min_event_value,
364	subquery_for_filtered_aggregates.c.max_event_value,
365	subquery_for_filtered_aggregates.c.avg_event_value,
366	subquery_for_filtered_aggregates.c.sum_event_value,
367	sa.func.count(TimedBelief.event_value).label("count_event_value"),
368	)
369	.select_from(TimedBelief)
370	.join(DataSource, DataSource.id == TimedBelief.source_id)
371	.join(
372	subquery_for_filtered_aggregates,
373	subquery_for_filtered_aggregates.c.source_id == TimedBelief.source_id,
374	)
375	.filter(TimedBelief.sensor_id == sensor.id)
376	.group_by(
377	DataSource.name,
378	subquery_for_filtered_aggregates.c.min_event_value,
379	subquery_for_filtered_aggregates.c.max_event_value,
380	subquery_for_filtered_aggregates.c.avg_event_value,
381	subquery_for_filtered_aggregates.c.sum_event_value,
382	)
383	).fetchall()
384
385	stats = dict()	8✔
386	for row in raw_stats:	8✔
387	(	8✔
388	data_source,
389	min_event_start,
390	max_event_start,
391	min_value,
392	max_value,
393	mean_value,
394	sum_values,
395	count_values,
396	) = row
397	stats[data_source] = {	8✔
398	"min_event_start": min_event_start,
399	"max_event_start": max_event_start,
400	"min_value": min_value,
401	"max_value": max_value,
402	"mean_value": mean_value,
403	"sum_values": sum_values,
404	"count_values": count_values,
405	}
406	return stats	8✔
407
408
409	def _get_ttl_hash(seconds=120) -> int:	8✔
410	"""Returns the same value within "seconds" time period
411	Is needed to make LRU cache a TTL one
412	(lru_cache is used when call arguments are the same,
413	here we ensure that call arguments are the same in "seconds" period of time).
414	"""
415	return round(time.time() / seconds)	8✔
416
417
418	def get_sensor_stats(sensor: Sensor) -> dict:	8✔
419	"""Get stats for a sensor"""
420	return _get_sensor_stats(sensor, ttl_hash=_get_ttl_hash())	8✔

FlexMeasures / flexmeasures / 11295014679

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous