/
sensors.py
43 lines (38 loc) · 1.68 KB
/
sensors.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from __future__ import annotations
import sqlalchemy as sa
from flexmeasures import Sensor, Account
from flexmeasures.data.models.generic_assets import GenericAsset
def get_sensors(
account: Account | list[Account],
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 = Sensor.query
if isinstance(account, list):
account_ids = [account.id for account in account]
else:
account_ids = [account.id]
sensor_query = sensor_query.join(GenericAsset).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 sensor_query.all()