9170661259

Committed 21 May 2024 07:29AM UTC coverage: 66.737% (+0.4%) from 66.357%

Build # 9170661259

Build Type

Pull #1059

github

Committed by

nrozanov

Commit Message

Add comment to asset creation / editing form

Signed-off-by: Nikolai Rozanov <nickolay.rozanov@gmail.com>

Pull Request Pull Request #1059: consumption-price-sensor, production-price-sensor and inflexible-devi…

Run Details

5831 of 9614 branches covered (60.65%)

Branch coverage included in aggregate %.

173 of 198 new or added lines in 9 files covered. (87.37%)

1 existing line in 1 file now uncovered.

7441 of 10273 relevant lines covered (72.43%)

5.74 hits per line

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

66.89

/flexmeasures/data/services/sensors.py

from __future__ import annotations

from datetime import datetime, timedelta
from flask import current_app
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.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 account is None:
        account_ids = []
    elif isinstance(account, list):
        account_ids = [account.id for account in account]
    else:
        account_ids = [account.id]
    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:
    """Get bdf for a given sensor with given search parameters."""
    bdf = TimedBelief.search(
        sensors=sensor,
        most_recent_events_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 immediate staleness after knowledge time
            status_specs = {"staleness_search": {}, "max_staleness": "PT0H"}
    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 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
    """
    if not now:
        now = server_now()

    sensors = []
    for asset in (asset, *asset.child_assets):
        for sensor in asset.sensors:
            sensor_status = get_status(
                sensor=sensor,
                now=now,
            )
            sensor_status["name"] = sensor.name
            sensor_status["id"] = sensor.id
            sensor_status["asset_name"] = asset.name
            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
    """

    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,
            current_app.job_cache.get(asset.id, "scheduling", "asset"),
        )
    )

    for sensor in asset.sensors:
        jobs.append(
            (
                "scheduling",
                "sensor",
                sensor.id,
                current_app.job_cache.get(sensor.id, "scheduling", "sensor"),
            )
        )
        jobs.append(
            (
                "forecasting",
                "sensor",
                sensor.id,
                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, asset_id, 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
            )

            jobs_data.append(
                {
                    "job_id": job.id,
                    "queue": queue,
                    "asset_or_sensor_type": asset_or_sensor_type,
                    "asset_id": asset_id,
                    "status": job.get_status(),
                    "err": job_err,
                    "enqueued_at": job.enqueued_at,
                }
            )

    return jobs_data

1	from __future__ import annotations	8✔
2
3	from datetime import datetime, timedelta	8✔
4	from flask import current_app	8✔
5	from timely_beliefs import BeliefsDataFrame	8✔
6
7	from humanize.time import naturaldelta	8✔
8
9	from flexmeasures.data.models.time_series import TimedBelief	8✔
10
11
12	import sqlalchemy as sa	8✔
13
14	from flexmeasures.data import db	8✔
15	from flexmeasures import Sensor, Account, Asset	8✔
16	from flexmeasures.data.models.generic_assets import GenericAsset	8✔
17	from flexmeasures.data.schemas.reporting import StatusSchema	8✔
18	from flexmeasures.utils.time_utils import server_now	8✔
19
20
21	def get_sensors(	8✔
22	account: Account \| list[Account] \| None,
23	include_public_assets: bool = False,
24	sensor_id_allowlist: list[int] \| None = None,
25	sensor_name_allowlist: list[str] \| None = None,
26	) -> list[Sensor]:
27	"""Return a list of Sensor objects that belong to the given account, and/or public sensors.
28
29	:param account: select only sensors from this account (or list of accounts)
30	:param include_public_assets: if True, include sensors that belong to a public asset
31	:param sensor_id_allowlist: optionally, allow only sensors whose id is in this list
32	:param sensor_name_allowlist: optionally, allow only sensors whose name is in this list
33	"""
34	sensor_query = sa.select(Sensor)	×
35	if account is None:	×
36	account_ids = []	×
37	elif isinstance(account, list):	×
38	account_ids = [account.id for account in account]	×
39	else:
40	account_ids = [account.id]	×
NEW 41	sensor_query = sensor_query.join(	×
42	GenericAsset, GenericAsset.id == Sensor.generic_asset_id
43	).filter(Sensor.generic_asset_id == GenericAsset.id)
44	if include_public_assets:	×
45	sensor_query = sensor_query.filter(	×
46	sa.or_(
47	GenericAsset.account_id.in_(account_ids),
48	GenericAsset.account_id.is_(None),
49	)
50	)
51	else:
52	sensor_query = sensor_query.filter(GenericAsset.account_id.in_(account_ids))	×
53	if sensor_id_allowlist:	×
54	sensor_query = sensor_query.filter(Sensor.id.in_(sensor_id_allowlist))	×
55	if sensor_name_allowlist:	×
56	sensor_query = sensor_query.filter(Sensor.name.in_(sensor_name_allowlist))	×
57	return db.session.scalars(sensor_query).all()	×
58
59
60	def _get_sensor_bdf(sensor: Sensor, staleness_search: dict) -> BeliefsDataFrame \| None:	8✔
61	"""Get bdf for a given sensor with given search parameters."""
62	bdf = TimedBelief.search(	8✔
63	sensors=sensor,
64	most_recent_events_only=True,
65	**staleness_search,
66	)
67	if bdf.empty:	8✔
68	return None	8✔
69	return bdf	8✔
70
71
72	def get_most_recent_knowledge_time(	8✔
73	sensor: Sensor, staleness_search: dict
74	) -> datetime \| None:
75	"""Get the knowledge time of the sensor's most recent event.
76
77	This knowledge time represents when you could have known about the event
78	(specifically, when you could have formed an ex-post belief about it).
79	"""
80	staleness_bdf = _get_sensor_bdf(sensor=sensor, staleness_search=staleness_search)	8✔
81	return None if staleness_bdf is None else staleness_bdf.knowledge_times[-1]	8✔
82
83
84	def get_staleness(	8✔
85	sensor: Sensor, staleness_search: dict, now: datetime
86	) -> timedelta \| None:
87	"""Get the staleness of the sensor.
88
89	The staleness is defined relative to the knowledge time of the most recent event, rather than to its belief time.
90	Basically, that means that we don't really care when the data arrived,
91	as long as the available data is about what we should be able to know by now.
92
93	:param sensor: The sensor to compute the staleness for.
94	:param staleness_search: Deserialized keyword arguments to `TimedBelief.search`.
95	:param now: Datetime representing now, used both to mask future beliefs,
96	and to measures staleness against.
97	"""
98
99	# Mask beliefs before now
100	staleness_search = staleness_search.copy() # no inplace operations	8✔
101	beliefs_before = staleness_search.get("beliefs_before")	8✔
102	if beliefs_before is not None:	8✔
103	staleness_search["beliefs_before"] = min(beliefs_before, now)	×
104	else:
105	staleness_search["beliefs_before"] = now	8✔
106
107	staleness_start_time = get_most_recent_knowledge_time(	8✔
108	sensor=sensor, staleness_search=staleness_search
109	)
110	if staleness_start_time is not None:	8✔
111	staleness = now - staleness_start_time	8✔
112	else:
113	staleness = None	8✔
114
115	return staleness	8✔
116
117
118	def get_status_specs(sensor: Sensor) -> dict:	8✔
119	"""Get status specs from a given sensor."""
120
121	# Check for explicitly defined status specs
122	status_specs = sensor.attributes.get("status_specs")	×
123	if status_specs is None:	×
124	# Default to status specs for economical sensors with daily updates
125	if sensor.knowledge_horizon_fnc == "x_days_ago_at_y_oclock":	×
126	status_specs = {"staleness_search": {}, "max_staleness": "P1D"}	×
127	else:
128	# Default to status specs indicating immediate staleness after knowledge time
129	status_specs = {"staleness_search": {}, "max_staleness": "PT0H"}	×
130	return status_specs	×
131
132
133	def get_status(	8✔
134	sensor: Sensor,
135	now: datetime,
136	status_specs: dict \| None = None,
137	) -> dict:
138	"""Get the status of the sensor
139	Main part of result here is a stale value, which is True if the sensor is stale, False otherwise.
140	Other values are just context information for the stale value.
141	"""
142	if status_specs is None:	8✔
143	status_specs = get_status_specs(sensor=sensor)	×
144	status_specs = StatusSchema().load(status_specs)	8✔
145	max_staleness = status_specs.pop("max_staleness")	8✔
146	staleness_search = status_specs.pop("staleness_search")	8✔
147	staleness: timedelta = get_staleness(	8✔
148	sensor=sensor,
149	staleness_search=staleness_search,
150	now=now,
151	)
152	if staleness is not None:	8✔
153	staleness_since = now - staleness	8✔
154	stale = staleness > max_staleness	8✔
155	reason = (	8✔
156	"" if stale else "not "
157	) + f"more than {naturaldelta(max_staleness)} old"
158	staleness = staleness if staleness > timedelta(0) else -staleness	8✔
159	else:
160	staleness_since = None	8✔
161	stale = True	8✔
162	reason = "no data recorded"	8✔
163	status = dict(	8✔
164	staleness=staleness,
165	stale=stale,
166	staleness_since=staleness_since,
167	reason=reason,
168	)
169	return status	8✔
170
171
172	def build_sensor_status_data(	8✔
173	asset: Asset,
174	now: datetime = None,
175	) -> list[dict]:
176	"""Get data connectivity status information for each sensor in given asset and its children
177	Returns a list of dictionaries, each containing the following keys:
178	- id: sensor id
179	- name: sensor name
180	- asset_name: asset name
181	- staleness: staleness of the sensor (for how long the sensor data is stale)
182	- stale: whether the sensor is stale
183	- staleness_since: time since sensor data is considered stale
184	- reason: reason for staleness
185	"""
186	if not now:	8✔
187	now = server_now()	8✔
188
189	sensors = []	8!
190	for asset in (asset, *asset.child_assets):	8✔
191	for sensor in asset.sensors:	8✔
192	sensor_status = get_status(	8✔
193	sensor=sensor,
194	now=now,
195	)
196	sensor_status["name"] = sensor.name	8✔
197	sensor_status["id"] = sensor.id	8✔
198	sensor_status["asset_name"] = asset.name	8✔
199	sensors.append(sensor_status)	8✔
200	return sensors	8✔
201
202
203	def build_asset_jobs_data(	8✔
204	asset: Asset,
205	) -> list[dict]:
206	"""Get all jobs data for an asset
207	Returns a list of dictionaries, each containing the following keys:
208	- job_id: id of a job
209	- queue: job queue (scheduling or forecasting)
210	- asset_or_sensor_type: type of an asset that is linked to the job (asset or sensor)
211	- asset_id: id of sensor or asset
212	- status: job status (e.g finished, failed, etc)
213	- err: job error (equals to None when there was no error for a job)
214	- enqueued_at: time when the job was enqueued
215	"""
216
217	jobs = list()	8✔
218
219	# try to get scheduling jobs for asset first (only scheduling jobs can be stored by asset id)
220	jobs.append(	8✔
221	(
222	"scheduling",
223	"asset",
224	asset.id,
225	current_app.job_cache.get(asset.id, "scheduling", "asset"),
226	)
227	)
228
229	for sensor in asset.sensors:	8✔
230	jobs.append(	8✔
231	(
232	"scheduling",
233	"sensor",
234	sensor.id,
235	current_app.job_cache.get(sensor.id, "scheduling", "sensor"),
236	)
237	)
238	jobs.append(	8✔
239	(
240	"forecasting",
241	"sensor",
242	sensor.id,
243	current_app.job_cache.get(sensor.id, "forecasting", "sensor"),
244	)
245	)
246
247	jobs_data = list()	8✔
248	# Building the actual return list - we also unpack lists of jobs, each to its own entry, and we add error info
249	for queue, asset_or_sensor_type, asset_id, jobs in jobs:	8✔
250	for job in jobs:	8✔
251	e = job.meta.get(	8✔
252	"exception",
253	Exception(
254	"The job does not state why it failed. "
255	"The worker may be missing an exception handler, "
256	"or its exception handler is not storing the exception as job meta data."
257	),
258	)
259	job_err = (	8✔
260	f"Scheduling job failed with {type(e).__name__}: {e}"
261	if job.is_failed
262	else None
263	)
264
265	jobs_data.append(	8✔
266	{
267	"job_id": job.id,
268	"queue": queue,
269	"asset_or_sensor_type": asset_or_sensor_type,
270	"asset_id": asset_id,
271	"status": job.get_status(),
272	"err": job_err,
273	"enqueued_at": job.enqueued_at,
274	}
275	)
276
277	return jobs_data	8✔

FlexMeasures / flexmeasures / 9170661259

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