flexmeasures.data.services.resources
Generic services for accessing asset data.
- TODO: This works with the legacy data model (esp. Assets), so it is marked for deprecation.
We are building data.services.asset_grouping, porting much of the code here. The data access logic here might also be useful for sensor data access logic we’ll build elsewhere, but that’s not quite certain at this point in time.
Functions
- flexmeasures.data.services.resources.can_access_asset(asset_or_sensor: Asset | Sensor) bool
Return True if: - the current user is an admin, or - the current user is the owner of the asset, or - the current user’s organisation account owns the corresponding generic asset, or - the corresponding generic asset is public
todo: refactor to def can_access_sensor(sensor: Sensor) -> bool once ui.views.state.state_view stops calling it with an Asset todo: let this function use our new auth model (row-level authorization) todo: deprecate this function in favor of an authz decorator on the API route
- flexmeasures.data.services.resources.check_cache(attribute)
Decorator for Resource class attributes to check if the resource has cached the attribute.
Example usage: @check_cache(“cached_data”) def some_property(self):
return self.cached_data
- flexmeasures.data.services.resources.get_asset_group_queries(custom_additional_groups: List[str] | None = None, all_users: bool = False) Dict[str, Query]
An asset group is defined by Asset queries. Each query has a name, and we prefer pluralised display names. They still need an executive call, like all(), count() or first().
- Parameters:
custom_additional_groups –
list of additional groups next to groups that represent unique asset types. Valid names are: - “renewables”, to query all solar and wind assets - “EVSE”, to query all Electric Vehicle Supply Equipment - “location”, to query each individual location with assets
(i.e. all EVSE at 1 location or each household)
all_users – if True, do not filter out assets that do not belong to the user (use with care)
- flexmeasures.data.services.resources.get_assets(owner_id: int | None = None, order_by_asset_attribute: str = 'id', order_direction: str = 'desc') List[Asset]
Return a list of all Asset objects owned by current_user (or all users or a specific user - for this, admins can set an owner_id).
- flexmeasures.data.services.resources.get_center_location(user: User | None) Tuple[float, float]
Find the center position between all assets. If user is passed and not admin then we only consider assets owned by the user. TODO: if we introduce accounts, this logic should look for these assets.
- flexmeasures.data.services.resources.get_demand_from_bdf(bdf: DataFrame | BeliefsDataFrame) DataFrame | BeliefsDataFrame
Positive values become 0 and negative values become positive values.
- flexmeasures.data.services.resources.get_location_queries() Dict[str, Query]
We group EVSE assets by location (if they share a location, they belong to the same Charge Point) Like get_asset_group_queries, the values in the returned dict still need an executive call, like all(), count() or first().
The Charge Points are named on the basis of the first EVSE in their list, using either the whole EVSE display name or that part that comes before a ” -” delimiter. For example: If:
evse_display_name = “Seoul Hilton - charger 1”
- Then:
charge_point_display_name = “Seoul Hilton (Charge Point)”
A Charge Point is a special case. If all assets on a location are of type EVSE, we can call the location a “Charge Point”.
- flexmeasures.data.services.resources.get_markets() List[Market]
Return a list of all Market objects.
- flexmeasures.data.services.resources.get_sensor_types(resource: Resource) List[WeatherSensorType]
Return a list of WeatherSensorType objects applicable to the given resource.
- flexmeasures.data.services.resources.get_sensors(owner_id: int | None = None, order_by_asset_attribute: str = 'id', order_direction: str = 'desc') List[Sensor]
Return a list of all Sensor objects owned by current_user’s organisation account (or all users or a specific user - for this, admins can set an owner_id).
- flexmeasures.data.services.resources.get_supply_from_bdf(bdf: DataFrame | BeliefsDataFrame) DataFrame | BeliefsDataFrame
Negative values become 0.
- flexmeasures.data.services.resources.group_assets_by_location(asset_list: List[Asset]) List[List[Asset]]
- flexmeasures.data.services.resources.has_assets(owner_id: int | None = None) bool
Return True if the current user owns any assets. (or all users or a specific user - for this, admins can set an owner_id).
- flexmeasures.data.services.resources.mask_inaccessible_assets(asset_queries: Query | Dict[str, Query]) Query | Dict[str, Query]
Filter out any assets that the user should not be able to access.
We do not explicitly check user authentication here, because non-authenticated users are not admins and have no asset ownership, so applying this filter for non-admins masks all assets.
Classes
- class flexmeasures.data.services.resources.Resource(name: str)
This class represents a group of assets of the same type, and provides helpful functions to retrieve their time series data and derived statistics.
Resolving asset type names
When initialised with a plural asset type name, the resource will contain all assets of the given type that are accessible to the user. When initialised with just one asset name, the resource will list only that asset.
Loading structure
Initialization only loads structural information from the database (which assets the resource groups).
Loading and caching time series
To load time series data for a certain time window, use the load_sensor_data() method. This loads beliefs data from the database and caches the results (as a named attribute). Caches are cleared when new time series data is loaded (or when the Resource instance seizes to exist).
Loading and caching derived statistics
Cached time series data is used to compute derived statistics, such as aggregates and scores. More specifically: - demand and supply - aggregated values (summed over assets) - total values (summed over time) - mean values (averaged over time) (todo: add this property) - revenue and cost - profit/loss When a derived statistic is called for, the results are also cached (using @functools.cached_property).
Resource(session[“resource”]).assets
Resource(session[“resource”]).display_name
Resource(session[“resource”]).get_data()
Usage
>>> from flask import session >>> resource = Resource(session["resource"]) >>> resource.assets >>> resource.display_name >>> resource.load_sensor_data(Power) >>> resource.cached_power_data >>> resource.load_sensor_data(Price, sensor_key_attribute="market.name") >>> resource.cached_price_data
- property aggregate_demand: BeliefsDataFrame
Returns aggregate demand as positive values.
- property aggregate_supply: BeliefsDataFrame
Returns aggregate supply (as positive values).
- is_eligible_for_comparing_individual_traces(max_traces: int = 7) bool
Decide whether comparing individual traces for assets in this resource is a useful feature. The number of assets that can be compared is parametrizable with max_traces. Plot colors are reused if max_traces > 7, and run out if max_traces > 105.
- load_sensor_data(sensor_types: List[SensorType] | None = None, start: datetime | None = None, end: datetime | None = None, resolution: str | None = None, belief_horizon_window=(None, None), belief_time_window=(None, None), source_types: List[str] | None = None, exclude_source_types: List[str] | None = None) Resource
Load data for one or more assets and cache the results. If the time range parameters are None, they will be gotten from the session. The horizon window will default to the latest measurement (anything more in the future than the end of the time interval. To load data for a specific source, pass a source id.
- Returns:
self (to allow piping)
Usage
>>> resource = Resource() >>> resource.load_sensor_data([Power], start=datetime(2014, 3, 1), end=datetime(2014, 3, 1)) >>> resource.cached_power_data >>> resource.load_sensor_data([Power, Price], start=datetime(2014, 3, 1), end=datetime(2014, 3, 1)).cached_price_data