Read csv with naive data, column filters and datetime/timedelta units

documentation/changelog.rst

@@ -12,31 +12,18 @@ New features
 * Ability to provide your own custom scheduling function [see `PR #505 <http://www.github.com/FlexMeasures/flexmeasures/pull/505>`_]
 * Visually distinguish forecasts/schedules (dashed lines) from measurements (solid lines), and expand the tooltip with timing info regarding the forecast/schedule horizon or measurement lag [see `PR #503 <http://www.github.com/FlexMeasures/flexmeasures/pull/503>`_]
 * The asset page also allows to show sensor data from other assets that belong to the same account [see `PR #500 <http://www.github.com/FlexMeasures/flexmeasures/pull/500>`_]
-* Improved import of time series data from CSV file: 1) drop duplicate records with warning, and 2) allow configuring which column contains explicit recording times for each data point (use case: import forecasts) [see `PR #501 <http://www.github.com/FlexMeasures/flexmeasures/pull/501>`_]
 * The CLI command ``flexmeasures show beliefs`` supports showing beliefs data in a custom resolution and/or timezone, and also saving the shown beliefs data to a CSV file [see `PR #519 <http://www.github.com/FlexMeasures/flexmeasures/pull/519>`_]
+* Improved import of time series data from CSV file: 1) drop duplicate records with warning, 2) allow configuring which column contains explicit recording times for each data point (use case: import forecasts) [see `PR #501 <http://www.github.com/FlexMeasures/flexmeasures/pull/501>`_], 3) localize timezone naive data, 4) support reading in datetime and timedelta values, 5) remove rows with NaN values, and 6) filter by values in specific columns [see `PR #521 <http://www.github.com/FlexMeasures/flexmeasures/pull/521>`_]
 
 Bugfixes
 -----------
 * The CLI command ``flexmeasures show beliefs`` now supports plotting time series data that includes NaN values, and provides better support for plotting multiple sensors that do not share the same unit [see `PR #516 <http://www.github.com/FlexMeasures/flexmeasures/pull/516>`_]
-* Consistent CLI/UI support for asset lat/lng positions up to 7 decimal places (previously the UI rounded to 4 decimal places, whereas the CLI allowed more than 4) [see `PR #522 <http://www.github.com/FlexMeasures/flexmeasures/pull/522>`_]
 
 Infrastructure / Support
 ----------------------
 
 * Reduce size of Docker image (from 2GB to 1.4GB) [see `PR #512 <http://www.github.com/FlexMeasures/flexmeasures/pull/512>`_]
 * Remove bokeh dependency and obsolete UI views [see `PR #476 <http://www.github.com/FlexMeasures/flexmeasures/pull/476>`_]
-* Fix ``flexmeasures db-ops dump`` and ``flexmeasures db-ops restore`` incorrectly reporting a success when `pg_dump` and `pg_restore` are not installed [see `PR #526 <http://www.github.com/FlexMeasures/flexmeasures/pull/526>`_]
-* Plugins can save BeliefsSeries, too, instead of just BeliefsDataFrames [see `PR #523 <http://www.github.com/FlexMeasures/flexmeasures/pull/523>`_]
-
-
-v0.11.3 | November 2, 2022
-============================
-
-Bugfixes
------------
-* Fix scheduling with imperfect efficiencies, which resulted in exceeding the device's lower SoC limit. [see `PR #520 <http://www.github.com/FlexMeasures/flexmeasures/pull/520>`_]
-* Fix scheduler for Charge Points when taking into account inflexible devices [see `PR #517 <http://www.github.com/FlexMeasures/flexmeasures/pull/517>`_]
-* Prevent rounding asset lat/long positions to 4 decimal places when editing an asset in the UI [see `PR #522 <http://www.github.com/FlexMeasures/flexmeasures/pull/522>`_]
 
 
 v0.11.2 | September 6, 2022

documentation/cli/change_log.rst

@@ -8,6 +8,7 @@ since v0.12.0 | November XX, 2022
 =================================
 
 * Add ``--resolution``, ``--timezone`` and ``--to-file`` options to ``flexmeasures show beliefs``, to show beliefs data in a custom resolution and/or timezone, and also to save shown beliefs data to a CSV file.
+* Add options to ``flexmeasures add beliefs`` to 1) read CSV data with timezone naive datetimes (use ``--timezone``to localize the data), 2) read CSV data with datetime/timedelta units (use ``--unit datetime`` or ``--unit timedelta`, 3) remove rows with NaN values, and 4) add filter read-in data by matching values in specific columns (use ``--filter-column`` and ``--filter-value`` together).
 * Fix ``flexmeasures db-ops dump`` and ``flexmeasures db-ops restore`` incorrectly reporting a success when `pg_dump` and `pg_restore` are not installed.
 
 since v0.11.0 | August 28, 2022

flexmeasures/api/v1_3/tests/test_api_v1_3.py

@@ -112,9 +112,7 @@ def test_post_udi_event_and_get_device_message(
     # check targets, if applicable
     if "targets" in message:
         start_soc = message["value"] / 1000  # in MWh
-        soc_schedule = integrate_time_series(
-            consumption_schedule, start_soc, decimal_precision=6
-        )
+        soc_schedule = integrate_time_series(consumption_schedule, start_soc, 6)
         print(consumption_schedule)
         print(soc_schedule)
         for target in message["targets"]:

flexmeasures/api/v2_0/tests/test_api_v2_0_assets.py

@@ -248,7 +248,7 @@ def test_post_an_asset_with_invalid_data(client, db):
         in post_asset_response.json["message"]["json"]["capacity_in_mw"][0]
     )
     assert (
-        "Longitude 300.9 exceeds the maximum longitude of 180 degrees."
+        "greater than or equal to -180 and less than or equal to 180"
         in post_asset_response.json["message"]["json"]["longitude"][0]
     )
     assert "required field" in post_asset_response.json["message"]["json"]["unit"][0]

flexmeasures/api/v3_0/tests/test_sensor_schedules.py

@@ -90,9 +90,7 @@ def test_trigger_and_get_schedule(
     # check targets, if applicable
     if "targets" in message:
         start_soc = message["soc-at-start"] / 1000  # in MWh
-        soc_schedule = integrate_time_series(
-            consumption_schedule, start_soc, decimal_precision=6
-        )
+        soc_schedule = integrate_time_series(consumption_schedule, start_soc, 6)
         print(consumption_schedule)
         print(soc_schedule)
         for target in message["targets"]:

flexmeasures/cli/data_add.py

@@ -38,13 +38,7 @@
     MissingAttributeException,
 )
 from flexmeasures.data.models.annotations import Annotation, get_or_create_annotation
-from flexmeasures.data.schemas import (
-    AwareDateTimeField,
-    DurationField,
-    LatitudeField,
-    LongitudeField,
-    SensorIdField,
-)
+from flexmeasures.data.schemas import AwareDateTimeField, DurationField, SensorIdField
 from flexmeasures.data.schemas.sensors import SensorSchema
 from flexmeasures.data.schemas.units import QuantityField
 from flexmeasures.data.schemas.generic_assets import (
@@ -248,12 +242,12 @@ def add_asset_type(**args):
 @click.option("--name", required=True)
 @click.option(
     "--latitude",
-    type=LatitudeField(),
+    type=float,
     help="Latitude of the asset's location",
 )
 @click.option(
     "--longitude",
-    type=LongitudeField(),
+    type=float,
     help="Longitude of the asset's location",
 )
 @click.option("--account-id", type=int, required=True)
@@ -286,8 +280,9 @@ def add_initial_structure():
 @click.argument("file", type=click.Path(exists=True))
 @click.option(
     "--sensor-id",
+    "sensor",
     required=True,
-    type=click.IntRange(min=1),
+    type=SensorIdField(),
     help="Sensor to which the beliefs pertain.",
 )
 @click.option(
@@ -301,6 +296,8 @@ def add_initial_structure():
     required=False,
     type=str,
     help="Unit of the data, for conversion to the sensor unit, if possible (a string unit such as 'kW' or 'm³/h').\n"
+    "Measurements of time itself that are formatted as a 'datetime' or 'timedelta' can be converted to a sensor unit representing time (such as 's' or 'h'),\n"
+    "where datetimes are represented as a duration with respect to the UNIX epoch."
     "Hint: to switch the sign of the data, prepend a minus sign.\n"
     "For example, when assigning kW consumption data to a kW production sensor, use '-kW'.",
 )
@@ -341,6 +338,12 @@ def add_initial_structure():
     multiple=True,
     help="Additional strings to recognize as NaN values. This argument can be given multiple times.",
 )
+@click.option(
+    "--keep-default-na",
+    default=False,
+    type=bool,
+    help="Whether or not to keep NaN values in the data.",
+)
 @click.option(
     "--nrows",
     required=False,
@@ -367,6 +370,24 @@ def add_initial_structure():
     type=int,
     help="Column number with datetimes",
 )
+@click.option(
+    "--timezone",
+    required=False,
+    default=None,
+    help="timezone as string, e.g. 'UTC' or 'Europe/Amsterdam'",
+)
+@click.option(
+    "--filter-column",
+    "filter_columns",
+    multiple=True,
+    help="Set a column number to filter data. Use together with --filter-value.",
+)
+@click.option(
+    "--filter-value",
+    "filter_values",
+    multiple=True,
+    help="Set a column value to filter data. Only rows with this value will be added. Use together with --filter-column.",
+)
 @click.option(
     "--delimiter",
     required=True,
@@ -396,19 +417,23 @@ def add_initial_structure():
 )
 def add_beliefs(
     file: str,
-    sensor_id: int,
+    sensor: Sensor,
     source: str,
+    filter_columns: List[int],
+    filter_values: List[int],
     unit: Optional[str] = None,
     horizon: Optional[int] = None,
     cp: Optional[float] = None,
     resample: bool = True,
     allow_overwrite: bool = False,
     skiprows: int = 1,
     na_values: list[str] | None = None,
+    keep_default_na: bool = False,
     nrows: Optional[int] = None,
     datecol: int = 0,
     valuecol: int = 1,
     beliefcol: Optional[int] = None,
+    timezone: Optional[str] = None,
     delimiter: str = ",",
     decimal: str = ".",
     thousands: Optional[str] = None,
@@ -433,17 +458,7 @@ def add_beliefs(
     In case no --horizon is specified and no beliefcol is specified,
     the moment of executing this CLI command is taken as the time at which the beliefs were recorded.
     """
-    sensor = Sensor.query.filter(Sensor.id == sensor_id).one_or_none()
-    if sensor is None:
-        print(f"Failed to create beliefs: no sensor found with ID {sensor_id}.")
-        return
-    if source.isdigit():
-        _source = get_source_or_none(int(source), source_type="CLI script")
-        if not _source:
-            print(f"Failed to find source {source}.")
-            return
-    else:
-        _source = get_or_create_source(source, source_type="CLI script")
+    _source = parse_source(source)
 
     # Set up optional parameters for read_csv
     if file.split(".")[-1].lower() == "csv":
@@ -458,6 +473,14 @@ def add_beliefs(
     elif beliefcol is None:
         kwargs["belief_time"] = server_now().astimezone(pytz.timezone(sensor.timezone))
 
+    # Set up optional filters:
+    if len(filter_columns) != len(filter_values):
+        raise ValueError(
+            "The number of filter columns and filter values should be the same."
+        )
+    filter_by_column = (
+        dict(zip(filter_columns, filter_values)) if filter_columns else None
+    )
     bdf = tb.read_csv(
         file,
         sensor,
@@ -472,6 +495,9 @@ def add_beliefs(
         else [datecol, beliefcol, valuecol],
         parse_dates=True,
         na_values=na_values,
+        keep_default_na=keep_default_na,
+        timezone=timezone,
+        filter_by_column=filter_by_column,
         **kwargs,
     )
     duplicate_rows = bdf.index.duplicated(keep="first")
@@ -1099,3 +1125,14 @@ def check_errors(errors: Dict[str, List[str]]):
             f"Please correct the following errors:\n{errors}.\n Use the --help flag to learn more."
         )
         raise click.Abort
+
+
+def parse_source(source):
+    if source.isdigit():
+        _source = get_source_or_none(int(source))
+        if not _source:
+            print(f"Failed to find source {source}.")
+            return
+    else:
+        _source = get_or_create_source(source, source_type="CLI script")
+    return _source

flexmeasures/cli/data_delete.py

@@ -12,7 +12,6 @@
 from flexmeasures.data.models.generic_assets import GenericAsset
 from flexmeasures.data.models.time_series import Sensor, TimedBelief
 from flexmeasures.data.schemas.generic_assets import GenericAssetIdField
-from flexmeasures.data.schemas.sensors import SensorIdField
 from flexmeasures.data.services.users import find_user_by_email, delete_user
 
 
@@ -124,7 +123,7 @@ def delete_asset_and_data(asset: GenericAsset, force: bool):
     Delete an asset & also its sensors and data.
     """
     if not force:
-        prompt = f"Delete {asset.__repr__()}, including all its sensors and data?"
+        prompt = f"Delete {asset}, including all its sensors and data?"
         click.confirm(prompt, abort=True)
     db.session.delete(asset)
     db.session.commit()
@@ -296,18 +295,21 @@ def delete_nan_beliefs(sensor_id: Optional[int] = None):
 @with_appcontext
 @click.option(
     "--id",
-    "sensor",
-    type=SensorIdField(),
+    "sensor_id",
+    type=int,
     required=True,
     help="Delete a single sensor and its (time series) data. Follow up with the sensor's ID.",
 )
 def delete_sensor(
-    sensor: Sensor,
+    sensor_id: int,
 ):
     """Delete a sensor and all beliefs about it."""
-    n = TimedBelief.query.filter(TimedBelief.sensor_id == sensor.id).delete()
+    sensor = Sensor.query.get(sensor_id)
+    n = TimedBelief.query.filter(TimedBelief.sensor_id == sensor_id).delete()
     db.session.delete(sensor)
-    click.confirm(f"Delete {sensor.__repr__()}, along with {n} beliefs?", abort=True)
+    click.confirm(
+        f"Really delete sensor {sensor_id}, along with {n} beliefs?", abort=True
+    )
     db.session.commit()
 
 

flexmeasures/cli/db_ops.py

@@ -95,7 +95,10 @@ def dump():
     dump_filename = f"pgbackup_{db_name}_{time_of_saving}.dump"
     command_for_dumping = f"pg_dump --no-privileges --no-owner --data-only --format=c --file={dump_filename} {db_uri}"
     try:
-        subprocess.check_output(command_for_dumping, shell=True)
+        proc = subprocess.Popen(command_for_dumping, shell=True)  # , env={
+        # 'PGPASSWORD': DB_PASSWORD
+        # })
+        proc.wait()
         click.echo(f"db dump successful: saved to {dump_filename}")
 
     except Exception as e:
@@ -122,7 +125,10 @@ def restore(file: str):
     click.echo(f"Restoring {db_host_and_db_name} database from file {file}")
     command_for_restoring = f"pg_restore -d {db_uri} {file}"
     try:
-        subprocess.check_output(command_for_restoring, shell=True)
+        proc = subprocess.Popen(command_for_restoring, shell=True)  # , env={
+        # 'PGPASSWORD': DB_PASSWORD
+        # })
+        proc.wait()
         click.echo("db restore successful")
 
     except Exception as e:

flexmeasures/data/models/planning/charging_station.py

@@ -110,10 +110,9 @@ def schedule_charging_station(
     ]
     if inflexible_device_sensors is None:
         inflexible_device_sensors = []
-    device_constraints = [
-        initialize_df(columns, start, end, resolution)
-        for i in range(1 + len(inflexible_device_sensors))
-    ]
+    device_constraints = [initialize_df(columns, start, end, resolution)] * (
+        1 + len(inflexible_device_sensors)
+    )
     for i, inflexible_sensor in enumerate(inflexible_device_sensors):
         device_constraints[i + 1]["derivative equals"] = get_power_values(
             query_window=(start, end),

flexmeasures/data/models/planning/solver.py

@@ -46,8 +46,8 @@ def device_scheduler(  # noqa C901
         derivative max: maximum flow (e.g. in MW or boxes/h)
         derivative min: minimum flow
         derivative equals: exact amount of flow (we do this by clamping derivative min and derivative max)
-        derivative down efficiency: conversion efficiency of flow out of a device (flow out : stock decrease)
-        derivative up efficiency: conversion efficiency of flow into a device (stock increase : flow in)
+        derivative down efficiency: ratio of downwards flows (flow into EMS : flow out of device)
+        derivative up efficiency: ratio of upwards flows (flow into device : flow out of EMS)
     EMS constraints are on an EMS level. Handled constraints (listed by column name):
         derivative max: maximum flow
         derivative min: minimum flow
@@ -228,12 +228,10 @@ def device_derivative_up_efficiency(m, d, j):
 
     # Add constraints as a tuple of (lower bound, value, upper bound)
     def device_bounds(m, d, j):
-        """Apply efficiencies to conversion from flow to stock change and vice versa."""
         return (
             m.device_min[d, j],
             sum(
-                m.device_power_down[d, k] / m.device_derivative_down_efficiency[d, k]
-                + m.device_power_up[d, k] * m.device_derivative_up_efficiency[d, k]
+                m.device_power_down[d, k] + m.device_power_up[d, k]
                 for k in range(0, j + 1)
             ),
             m.device_max[d, j],
@@ -277,10 +275,12 @@ def ems_flow_commitment_equalities(m, j):
         )
 
     def device_derivative_equalities(m, d, j):
-        """Couple device flows to EMS flows per device."""
+        """Couple device flows to EMS flows per device, applying efficiencies."""
         return (
             0,
-            m.device_power_up[d, j] + m.device_power_down[d, j] - m.ems_power[d, j],
+            m.device_power_up[d, j] / m.device_derivative_up_efficiency[d, j]
+            + m.device_power_down[d, j] * m.device_derivative_down_efficiency[d, j]
+            - m.ems_power[d, j],
             0,
         )
 
@@ -321,7 +321,10 @@ def cost_function(m):
     planned_costs = value(model.costs)
     planned_power_per_device = []
     for d in model.d:
-        planned_device_power = [model.ems_power[d, j].value for j in model.j]
+        planned_device_power = [
+            model.device_power_down[d, j].value + model.device_power_up[d, j].value
+            for j in model.j
+        ]
         planned_power_per_device.append(
             pd.Series(
                 index=pd.date_range(
@@ -332,7 +335,7 @@ def cost_function(m):
         )
 
     # model.pprint()
-    # model.display()
     # print(results.solver.termination_condition)
     # print(planned_costs)
+    # model.display()
     return planned_power_per_device, planned_costs, results