Cli resample sensor

documentation/changelog.rst

@@ -15,6 +15,7 @@ New features
 * Add CLI commands ``flexmeasures add sensor``, ``flexmeasures add asset-type``, ``flexmeasures add beliefs`` (which were experimental features before). [see `PR #337 <http://www.github.com/FlexMeasures/flexmeasures/pull/337>`_]
 * Add CLI commands for showing data [see `PR #339 <http://www.github.com/FlexMeasures/flexmeasures/pull/339>`_]
 * Add CLI command for attaching annotations to assets: ``flexmeasures add holidays`` adds public holidays [see `PR #343 <http://www.github.com/FlexMeasures/flexmeasures/pull/343>`_]
+* Add CLI command for resampling existing sensor data to new resolution [see `PR #360 <http://www.github.com/FlexMeasures/flexmeasures/pull/360>`_]
 
 Bugfixes
 -----------

documentation/cli/change_log.rst

@@ -8,6 +8,7 @@ since v0.9.0 | January 26, 2022
 =====================
 
 * add CLI comands for showing data ``flexmeasures show accounts``, ``flexmeasures show account``, ``flexmeasures show roles``, ``flexmeasures show asset-types``, ``flexmeasures show asset`` and ``flexmeasures show data-sources``.
+* Add ``flexmeasures db-ops resample-data`` CLI command to resample sensor data to a different resolution.
 
 
 since v0.9.0 | January 26, 2022

flexmeasures/cli/db_ops.py

@@ -1,15 +1,20 @@
 """CLI Tasks for saving, resetting, etc of the database"""
 
-from datetime import datetime
+from datetime import datetime, timedelta
 import subprocess
+from typing import List
 
 from flask import current_app as app
 from flask.cli import with_appcontext
 import flask_migrate as migrate
 import click
 
+from flexmeasures.data import db
+from flexmeasures.data.models.time_series import Sensor, TimedBelief
+from flexmeasures.data.utils import save_to_db
 
-BACKUP_PATH = app.config.get("FLEXMEASURES_DB_BACKUP_PATH")
+
+BACKUP_PATH: str = app.config.get("FLEXMEASURES_DB_BACKUP_PATH")  # type: ignore
 
 
 @click.group("db-ops")
@@ -121,7 +126,7 @@ def restore(file: str):
 
     """
 
-    db_uri = app.config.get("SQLALCHEMY_DATABASE_URI")
+    db_uri: str = app.config.get("SQLALCHEMY_DATABASE_URI")  # type: ignore
     db_host_and_db_name = db_uri.split("@")[-1]
     click.echo(f"Restoring {db_host_and_db_name} database from file {file}")
     command_for_restoring = f"pg_restore -d {db_uri} {file}"
@@ -137,4 +142,63 @@ def restore(file: str):
         click.echo("db restore unsuccessful")
 
 
+@fm_db_ops.command("resample-data")
+@with_appcontext
+@click.option(
+    "--sensor-id",
+    "sensor_ids",
+    multiple=True,
+    required=True,
+    help="Resample data for this sensor. Follow up with the sensor's ID. This argument can be given multiple times.",
+)
+@click.option(
+    "--event-resolution",
+    "event_resolution_in_minutes",
+    type=int,
+    required=True,
+    help="New event resolution as an integer number of minutes.",
+)
+@click.option(
+    "--skip-integrity-check",
+    is_flag=True,
+    help="Whether to skip checking the resampled time series data for each sensor."
+    " By default, an excerpt and the mean value of the original"
+    " and resampled data will be shown for manual approval.",
+)
+def resample_sensor_data(
+    sensor_ids: List[int],
+    event_resolution_in_minutes: int,
+    skip_integrity_check: bool = False,
+):
+    """Assign a new event resolution to an existing sensor and resample its data accordingly."""
+    event_resolution = timedelta(minutes=event_resolution_in_minutes)
+    for sensor_id in sensor_ids:
+        sensor = Sensor.query.get(sensor_id)
+        if sensor.event_resolution == event_resolution:
+            print(f"{sensor} already has the desired event resolution.")
+            continue
+        df_original = sensor.search_beliefs().sort_values("event_start")
+        df_resampled = df_original.resample_events(event_resolution).sort_values(
+            "event_start"
+        )
+        if not skip_integrity_check:
+            message = ""
+            if sensor.event_resolution < event_resolution:
+                message += f"Downsampling {sensor} to {event_resolution} will result in a loss of data. "
+            click.confirm(
+                message
+                + f"Data before:\n{df_original}\nData after:\n{df_resampled}\nMean before: {df_original['event_value'].mean()}\nMean after: {df_resampled['event_value'].mean()}\nContinue?",
+                abort=True,
+            )
+
+        # Update sensor
+        sensor.event_resolution = event_resolution
+        db.session.add(sensor)
+
+        # Update sensor data
+        TimedBelief.query.filter(TimedBelief.sensor == sensor).delete()
+        save_to_db(df_resampled, bulk_save_objects=True)
+    db.session.commit()
+
+
 app.cli.add_command(fm_db_ops)

flexmeasures/data/utils.py

@@ -51,6 +51,7 @@ def get_data_source(
 
 def save_to_db(
     data: Union[BeliefsDataFrame, List[BeliefsDataFrame]],
+    bulk_save_objects: bool = False,
     save_changed_beliefs_only: bool = True,
 ) -> str:
     """Save the timed beliefs to the database.
@@ -75,6 +76,9 @@ def save_to_db(
     Servers in 'play' mode are exempt from this rule, to facilitate replaying simulations.
 
     :param data: BeliefsDataFrame (or a list thereof) to be saved
+    :param bulk_save_objects: if True, objects are bulk saved with session.bulk_save_objects(),
+                              which is quite fast but has several caveats, see:
+                              https://docs.sqlalchemy.org/orm/persistence_techniques.html#bulk-operations-caveats
     :param save_changed_beliefs_only: if True, unchanged beliefs are skipped (updated beliefs are only stored if they represent changed beliefs)
                                       if False, all updated beliefs are stored
     :returns: status string, one of the following:
@@ -122,6 +126,7 @@ def save_to_db(
         TimedBelief.add_to_session(
             session=db.session,
             beliefs_data_frame=timed_values,
+            bulk_save_objects=bulk_save_objects,
             allow_overwrite=current_app.config.get(
                 "FLEXMEASURES_ALLOW_DATA_OVERWRITE", False
             ),

requirements/app.in

@@ -36,7 +36,7 @@ siphon
 tables
 tabulate
 timetomodel>=0.7.1
-timely-beliefs>=1.11.0
+timely-beliefs>=1.11.1
 python-dotenv
 # a backport, not needed in Python3.8
 importlib_metadata

requirements/app.txt

@@ -367,7 +367,7 @@ tabulate==0.8.9
     # via -r requirements/app.in
 threadpoolctl==3.0.0
     # via scikit-learn
-timely-beliefs==1.11.0
+timely-beliefs==1.11.1
     # via -r requirements/app.in
 timetomodel==0.7.1
     # via -r requirements/app.in