Repository: FlexMeasures/flexmeasures (language: python)

Issue:
Move sensor data from Power/Price/Weather to TimedBelief
- [x] Initialize TimedBelief when initializing Power/Price/Weather
- [x] Query TimedBelief rather than Power/Price/Weather
- [x] Migrate Power/Price/Weather data to TimedBelief
- [x] Stop saving to Power/Price/Weather tables


Discussion / hints:

Fix patch:
diff --git a/documentation/changelog.rst b/documentation/changelog.rst
index 2844d3399..4015730cc 100644
--- a/documentation/changelog.rst
+++ b/documentation/changelog.rst
@@ -23,6 +23,7 @@ Infrastructure / Support
 * Allow plugins to register their custom config settings, so that FlexMeasures can check whether they are set up correctly [see `PR #230 <http://www.github.com/FlexMeasures/flexmeasures/pull/230>`_ and `PR #237 <http://www.github.com/FlexMeasures/flexmeasures/pull/237>`_]
 * Add sensor method to obtain just its latest state (excl. forecasts) [see `PR #235 <http://www.github.com/FlexMeasures/flexmeasures/pull/235>`_]
 * Migrate attributes of assets, markets and weather sensors to our new sensor model [see `PR #254 <http://www.github.com/FlexMeasures/flexmeasures/pull/254>`_ and `project 9 <http://www.github.com/FlexMeasures/flexmeasures/projects/9>`_]
+* Migrate all time series data to our new sensor data model based on the `timely beliefs <https://github.com/SeitaBV/timely-beliefs>`_ lib [see `PR #286 <http://www.github.com/FlexMeasures/flexmeasures/pull/286>`_ and `project 9 <http://www.github.com/FlexMeasures/flexmeasures/projects/9>`_]
 
 
 v0.7.1 | November 08, 2021
diff --git a/flexmeasures/data/migrations/versions/e690d373a3d9_copy_Power_Price_Weather_time_series_data_to_TimedBeliefs_table.py b/flexmeasures/data/migrations/versions/e690d373a3d9_copy_Power_Price_Weather_time_series_data_to_TimedBeliefs_table.py
new file mode 100644
index 000000000..c44782025
--- /dev/null
+++ b/flexmeasures/data/migrations/versions/e690d373a3d9_copy_Power_Price_Weather_time_series_data_to_TimedBeliefs_table.py
@@ -0,0 +1,122 @@
+"""Copy Power/Price/Weather time series data to TimedBeliefs table
+
+Revision ID: e690d373a3d9
+Revises: 830e72a8b218
+Create Date: 2021-12-27 15:01:38.967237
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "e690d373a3d9"
+down_revision = "830e72a8b218"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    print("Database migration started")
+    print("- preparing to copy time series data...")
+
+    # Declare ORM table views
+    t_power = sa.Table(
+        "power",
+        sa.MetaData(),
+        sa.Column("sensor_id"),
+        sa.Column("datetime"),
+        sa.Column("horizon"),
+        sa.Column("value"),
+        sa.Column("data_source_id"),
+    )
+    t_price = sa.Table(
+        "price",
+        sa.MetaData(),
+        sa.Column("sensor_id"),
+        sa.Column("datetime"),
+        sa.Column("horizon"),
+        sa.Column("value"),
+        sa.Column("data_source_id"),
+    )
+    t_weather = sa.Table(
+        "weather",
+        sa.MetaData(),
+        sa.Column("sensor_id"),
+        sa.Column("datetime"),
+        sa.Column("horizon"),
+        sa.Column("value"),
+        sa.Column("data_source_id"),
+    )
+    t_timed_belief = sa.Table(
+        "timed_belief",
+        sa.MetaData(),
+        sa.Column("sensor_id"),
+        sa.Column("event_start"),
+        sa.Column("belief_horizon"),
+        sa.Column("event_value"),
+        sa.Column("cumulative_probability"),
+        sa.Column("source_id"),
+    )
+
+    # Use SQLAlchemy's connection and transaction to go through the data
+    connection = op.get_bind()
+
+    copy_time_series_data(
+        connection,
+        t_price,
+        t_timed_belief,
+    )
+    copy_time_series_data(
+        connection,
+        t_power,
+        t_timed_belief,
+    )
+    copy_time_series_data(
+        connection,
+        t_weather,
+        t_timed_belief,
+    )
+    print("- finished copying time series data...")
+
+
+def downgrade():
+    pass
+
+
+def copy_time_series_data(
+    connection,
+    t_old_data_model,
+    t_timed_belief,
+    batch_size: int = 100000,
+):
+    mapping = {
+        "value": "event_value",
+        "data_source_id": "source_id",
+        "datetime": "event_start",
+        "horizon": "belief_horizon",
+        "sensor_id": "sensor_id",
+    }
+
+    # Get data from old data model
+    results = connection.execute(
+        sa.select([getattr(t_old_data_model.c, a) for a in mapping.keys()])
+    ).fetchall()
+
+    print(
+        f"- copying {len(results)} rows from the {t_old_data_model.name} table to the {t_timed_belief.name} table..."
+    )
+
+    # Copy in batches and report on progress
+    for i in range(len(results) // batch_size + 1):
+        if i > 0:
+            print(f"  - done copying {i*batch_size} rows...")
+
+        insert_values = []
+        for values in results[i * batch_size : (i + 1) * batch_size]:
+            d = {k: v for k, v in zip(mapping.values(), values)}
+            d["cumulative_probability"] = 0.5
+            insert_values.append(d)
+        op.bulk_insert(t_timed_belief, insert_values)
+
+    print(f"  - finished copying {len(results)} rows...")


Test patch:
diff --git a/flexmeasures/data/tests/test_queries.py b/flexmeasures/data/tests/test_queries.py
index bf9696c5f..1f4232753 100644
--- a/flexmeasures/data/tests/test_queries.py
+++ b/flexmeasures/data/tests/test_queries.py
@@ -90,7 +90,7 @@ def test_collect_power(db, app, query_start, query_end, num_values, setup_test_d
         ),
     ],
 )
-def test_collect_power_resampled(
+def tesfijfijft_collect_power_resampled(
     db, app, query_start, query_end, resolution, num_values, setup_test_data
 ):
     wind_device_1 = Sensor.query.filter_by(name="wind-asset-1").one_or_none()
@@ -220,7 +220,7 @@ def test_simplify_index(setup_test_data, check_empty_frame):
         event_starts_after=datetime(2015, 1, 1, tzinfo=pytz.utc),
         event_ends_before=datetime(2015, 1, 2, tzinfo=pytz.utc),
         resolution=timedelta(minutes=15),
-    )
+    ).convert_index_from_belief_time_to_horizon()
     if check_empty_frame:
         # We empty the BeliefsDataFrame, which retains the metadata such as sensor and resolution
         bdf = bdf.iloc[0:0, :]