flexmeasures.data.services.data_ingestion

Logic around data ingestion (jobs)

Functions

flexmeasures.data.services.data_ingestion.add_beliefs_to_db_and_enqueue_forecasting_jobs(data: BeliefsDataFrame | list[BeliefsDataFrame] | None = None, sensor_id: int | None = None, user_id: int | None = None, sensor_data: dict | None = None, uploaded_files: list[dict] | None = None, upload_data: dict | None = None, forecasting_jobs: list[Job] | None = None, forecasting_job_ids: list[str] | None = None, save_changed_beliefs_only: bool = True) → str

Save sensor data to the database and optionally enqueue forecasting jobs.

This function is intended to be called as an RQ job by an ingestion queue worker, but can also be called directly (e.g. as a fallback when no workers are available).

Parameters:

  • data – BeliefsDataFrame (or list thereof) to be saved.

  • sensor_id – Sensor ID for raw JSON or file ingestion.

  • user_id – User ID used to resolve the source of raw ingested data.

  • sensor_data – Raw JSON payload from the sensor data endpoint.

  • uploaded_files – Uploaded file contents and metadata.

  • upload_data – Raw form payload from the sensor data upload endpoint.

  • forecasting_jobs – Optional list of forecasting Jobs to enqueue after saving.

  • forecasting_job_ids – Optional list of forecasting Job ids to enqueue after saving.

  • save_changed_beliefs_only – If True, skip saving beliefs whose value hasn’t changed.

Returns:

Status string as returned by save_to_db.

Raises:

BeliefReplacementNotAllowed – if saving would change existing beliefs, while FLEXMEASURES_ALLOW_DATA_OVERWRITE is off.

flexmeasures.data.services.data_ingestion.is_belief_replacement(error: IntegrityError) → bool

Tell whether an IntegrityError is a UniqueViolation on the TimedBelief primary key.

The error is recognised by its SQLSTATE rather than by its class, because the class depends on the database driver: psycopg2 exposes the SQLSTATE as pgcode, and psycopg 3 as sqlstate.

Exceptions

exception flexmeasures.data.services.data_ingestion.BeliefReplacementNotAllowed(message: str = 'Some of the data represents a replacement: existing values would be changed for the same sensor, source, event timestamp and belief time. This is reserved for customized servers. If you are hosting FlexMeasures, you can enable replacements by setting FLEXMEASURES_ALLOW_DATA_OVERWRITE=True in the configuration settings. Alternatively, submit the data with a different prior, or check whether the same file was already uploaded with different values.')

Saving would change existing beliefs, which this server does not allow.