flexmeasures.api.common.utils.api_utils

Functions

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.convert_asset_json_fields(asset_kwargs)

Convert string fields in asset_kwargs to JSON where needed.

flexmeasures.api.common.utils.api_utils.copy_asset(asset: GenericAsset, account=None, parent_asset=None) → GenericAsset

Copy a single asset to a target account and/or under a target parent asset.

Resolution rules:

  • If neither account nor parent_asset is given, the copy is placed in the same account and under the same parent as the original (i.e. a sibling).

  • If account is given but parent_asset is not, the copy becomes a top-level asset (no parent) in the given account.

  • If parent_asset is given but account is not, the copy is placed under the given parent and inherits that parent’s account.

  • If both are given, the copy belongs to the given account and is placed under the given parent. This allows creating a copy that belongs to a different account than its parent.

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_accessible_accounts() → list[Account]
flexmeasures.api.common.utils.api_utils.job_status_description(job: Job, extra_message: str | None = None)

Return a matching description for the job’s status.

Supports each rq.job.JobStatus (NB JobStatus.CREATED is deprecated).

Parameters:

  • job – The rq.Job.

  • extra_message – Optionally, append a message to the job status description.

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) → tuple[dict, int] | tuple[dict, int, dict]
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