flexmeasures.api.common.utils.api_utils

Functions

flexmeasures.api.common.utils.api_utils.append_doc_of(fun)

flexmeasures.api.common.utils.api_utils.asset_replace_name_with_id(connections_as_name: List[str]) → List[str]: Look up the owner and id given the asset name and construct a type 1 USEF entity address.

flexmeasures.api.common.utils.api_utils.catch_timed_belief_replacements(error: IntegrityError)

Catch IntegrityErrors due to a UniqueViolation on the TimedBelief primary key.

Return a more informative message.

flexmeasures.api.common.utils.api_utils.contains_empty_items(groups: List[List[str]]): Return True if any of the items in the groups is empty.

flexmeasures.api.common.utils.api_utils.determine_belief_timing(event_values: list, start: datetime, resolution: timedelta, horizon: timedelta, prior: datetime, sensor: Sensor) → Tuple[List[datetime], List[timedelta]]

Determine event starts from start, resolution and len(event_values), and belief horizons from horizon, prior, or both, taking into account the sensor’s knowledge horizon function.

In case both horizon and prior is set, we take the greatest belief horizon, which represents the earliest belief time.

flexmeasures.api.common.utils.api_utils.enqueue_forecasting_jobs(forecasting_jobs: list[Job] | None = None)

Enqueue forecasting jobs.

Parameters:: forecasting_jobs – list of forecasting Jobs for redis queues.

flexmeasures.api.common.utils.api_utils.get_form_from_request(_request) → dict | None

flexmeasures.api.common.utils.api_utils.get_sensor_by_generic_asset_type_and_location(generic_asset_type_name: str, latitude: float = 0, longitude: float = 0) → Sensor | Tuple[dict, int]: Search a sensor by generic asset type and location. Can create a sensor if needed (depends on API mode) and then inform the requesting user which one to use.

flexmeasures.api.common.utils.api_utils.groups_to_dict(connection_groups: List[str], value_groups: List[List[str]], generic_asset_type_name: str, plural_name: str | None = None, groups_name='groups') → dict

Put the connections and values in a dictionary and simplify if groups have identical values and/or if there is only one group.

Examples:

>> connection_groups = [[1]] >> value_groups = [[300, 300, 300]] >> response_dict = groups_to_dict(connection_groups, value_groups, “connection”) >> print(response_dict) << {

“connection”: 1, “values”: [300, 300, 300]

}

>> connection_groups = [[1], [2]] >> value_groups = [[300, 300, 300], [300, 300, 300]] >> response_dict = groups_to_dict(connection_groups, value_groups, “connection”) >> print(response_dict) << {

“connections”: [1, 2], “values”: [300, 300, 300]

}

>> connection_groups = [[1], [2]] >> value_groups = [[300, 300, 300], [400, 400, 400]] >> response_dict = groups_to_dict(connection_groups, value_groups, “connection”) >> print(response_dict) << {

“groups”: [

{
“connection”: 1, “values”: [300, 300, 300]

}, {

“connection”: 2, “values”: [400, 400, 400]

}

]

}

flexmeasures.api.common.utils.api_utils.list_access(service_listing, service_name): For a given USEF service name (API endpoint) in a service listing, return the list of USEF roles that are allowed to access the service.

flexmeasures.api.common.utils.api_utils.message_replace_name_with_ea(message_with_connections_as_asset_names: dict) → dict: For each connection in the message specified by a name, replace that name with the correct entity address. TODO: Deprecated. This function is now only used in tests of deprecated API versions and should go (also asset_replace_name_with_id)

flexmeasures.api.common.utils.api_utils.parse_as_list(connection: str | float | Sequence[str | float], of_type: type | None = None) → Sequence[str | float | None]: Return a list of connections (or values), even if it’s just one connection (or value)

flexmeasures.api.common.utils.api_utils.save_and_enqueue(data: BeliefsDataFrame | List[BeliefsDataFrame], forecasting_jobs: list[Job] | None = None, save_changed_beliefs_only: bool = True) → ResponseTuple

flexmeasures.api.common.utils.api_utils.save_to_db(timed_values: BeliefsDataFrame | List[Power | Price | Weather], forecasting_jobs: List[Job] = [], save_changed_beliefs_only: bool = True) → Tuple[dict, int]

Put the timed values into the database and enqueue forecasting jobs.

Data can only be replaced on servers in play mode.

TODO: remove this legacy function in its entirety (announced v0.8.0)

Parameters:

timed_values – BeliefsDataFrame or a list of Power, Price or Weather values to be saved
forecasting_jobs – list of forecasting Jobs for redis queues.
save_changed_beliefs_only – if True, beliefs that are already stored in the database with an earlier belief time are dropped.

Returns:

ResponseTuple

flexmeasures.api.common.utils.api_utils.unique_ever_seen(iterable: Sequence, selector: Sequence)

Return unique iterable elements with corresponding lists of selector elements, preserving order.

>>> a, b = unique_ever_seen([[10, 20], [10, 20], [20, 40]], [1, 2, 3])
>>> print(a)
[[10, 20], [20, 40]]
>>> print(b)
[[1, 2], 3]

flexmeasures.api.common.utils.api_utils.upsample_values(value_groups: List[List[float]] | List[float], from_resolution: timedelta, to_resolution: timedelta) → List[List[float]] | List[float]: Upsample the values (in value groups) to a smaller resolution. from_resolution has to be a multiple of to_resolution