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, serialized_data: list[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.

• serialized_data – Queue-safe payload containing only primitive types.

• 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, one of: - ‘success’ - ‘success_with_unchanged_beliefs_skipped’ - ‘success_but_nothing_new’

flexmeasures.data.services.data_ingestion.deserialize_ingestion_data(payload: Sequence[dict]) → list[BeliefsDataFrame]

Deserialize queue-safe ingestion payload back into BeliefsDataFrames.

flexmeasures.data.services.data_ingestion.serialize_ingestion_data(data: BeliefsDataFrame | list[BeliefsDataFrame]) → list[dict]

Serialize beliefs data to primitive types suitable for queue kwargs.

The returned payload intentionally avoids ORM instances (Sensor/DataSource), which can break across process boundaries when pickled by RQ.