diff --git a/CHANGELOG-S2.rst b/CHANGELOG-S2.rst index 12887b18..49e13f7d 100644 --- a/CHANGELOG-S2.rst +++ b/CHANGELOG-S2.rst @@ -19,12 +19,13 @@ Added * ``FRBCSimple`` can record what a resource manager declares about its storage, so it can be seen and planned with. Give it any of the new ``soc_minima_sensor_id``, ``soc_maxima_sensor_id``, ``usage_forecast_sensor_id`` and ``leakage_behaviour_sensor_id``; a declaration without a sensor is not recorded. - A fill-level target profile is recorded on the minima and maxima sensors, if both are given. It is recorded on the handler's ``resolution`` steps, starting from the step its start falls in. It is kept inside the declared fill-level range: a target past either end, such as a night setback below the range, is cut back to the range, and a target band wholly outside the range is recorded at its nearest end, with a warning. If one of the two bounds cannot be recorded, this is logged and the other is still recorded. - - A usage forecast is recorded in ``/s``. A negative expected usage is recorded as zero, with a warning. - - A leakage behaviour is recorded as a storage efficiency, each time a schedule is requested (see Changed). + - A usage forecast is recorded in ``fill_rate_unit`` (see below). A negative expected usage, which S2 allows, is recorded as stated. Without a usage correction (see below), a schedule request carries the latest forecast without its negative values, since FlexMeasures would plan them as a gain the storage cannot refuse. A FlexMeasures server before v0.23.0 only plans with what is recorded, so there a negative expected usage is recorded as zero, with a warning [`#286 `__]. + - A leakage behaviour is recorded as the leakage rate it states, in ``fill_rate_unit``, each time a schedule is requested (see Changed) [`#289 `__]. [`#244 `__, `#259 `__, `#271 `__, `#274 `__] -* ``FRBCSimple`` takes an ``energy_unit`` (default ``kWh``). The state of charge, the fill-level band, the fill rate and the usage forecast are recorded in this unit (the last two per second). Allowed units are ``J``, ``kJ``, ``MJ``, ``Wh``, ``kWh`` and ``MWh``; any other unit is refused when the handler is created. In a flex model, the state of charge is stated in ``kWh`` or ``MWh``, converted from the other units, because FlexMeasures accepts no others [`#244 `__, `#260 `__] -* ``FRBCSimple`` takes a ``fill_level_scale`` (default 1). Every fill level and fill rate sent to FlexMeasures is multiplied by it: the state of charge, the flex model, the fill-level band, the fill rate, the usage forecast and the storage efficiency. Translating a schedule into instructions keeps the resource manager's own units, because it matches fill levels against the operation modes, which are stated in those units. A scale of zero or below is refused. ``FRBCSimple`` also takes a ``power_unit`` (default ``W``), which an actuator's power is converted into and posted in [`#236 `__, `#244 `__, `#245 `__, `#252 `__] +* ``FRBCSimple`` takes an ``energy_unit`` (default ``kWh``). The state of charge and the fill-level band are recorded in this unit. Allowed units are ``J``, ``kJ``, ``MJ``, ``Wh``, ``kWh`` and ``MWh``; any other unit is refused when the handler is created. In a flex model, the state of charge is stated in ``kWh`` or ``MWh``, converted from the other units, because FlexMeasures accepts no others [`#244 `__, `#260 `__] +* ``FRBCSimple`` takes a ``fill_rate_unit`` (default ``W``; also ``kW`` or ``MW``). Rates of fill level, which S2 states per second, are recorded in it: the fill rate, the usage forecast and the leakage. In ``kWh/s``, a heat buffer's leakage of some tens of watts would read about 0.00001. Any other unit is refused when the handler is created [`#289 `__] +* ``FRBCSimple`` takes a ``fill_level_scale`` (default 1). Every fill level and fill rate sent to FlexMeasures is multiplied by it: the state of charge, the flex model, the fill-level band, the fill rate, the usage forecast and the leakage. Translating a schedule into instructions keeps the resource manager's own units, because it matches fill levels against the operation modes, which are stated in those units. A scale of zero or below is refused. ``FRBCSimple`` also takes a ``power_unit`` (default ``W``), which an actuator's power is converted into and posted in [`#236 `__, `#244 `__, `#245 `__, `#252 `__] * ``FRBCSimple`` takes a ``hard_bound_factor`` (default 1.5). It sets how far above the declared fill-level range a schedule's hard upper bound lies. If a schedule starts from a state of charge above that bound, the bound is raised to include it. A factor below 1 is refused: FlexMeasures clips the band to the hard bounds, so a hard upper bound inside the declared range would narrow the band to something the resource manager never asked for [`#245 `__] * ``FRBCSimple`` can take the actuator's charging efficiency into account. Give it the new ``charging_efficiency_sensor_id``: it records there what the operation modes say about converting drawn power into stored energy, and requests a schedule with it as ``charging-efficiency``. For a heat pump, this ratio is its coefficient of performance. Without it, the scheduler assumes power converts into stored energy one to one, and plans to reach fill-level targets at a power the actuator cannot deliver. @@ -48,13 +49,22 @@ Added * An FRBC handler tracks the instruction batch the resource manager is working from, and can revoke instructions it no longer wants carried out. Set its ``send_revocations`` attribute to ``True``. When a new batch replaces that one, the handler revokes the old batch's pending instructions: those without a reported status, or reported as ``NEW`` or ``ACCEPTED``. This is off by default, because a resource manager does not have to honour revocations [`#237 `__] * New ``FRBC`` methods for handlers. ``filter_instruction_transitions()`` drops instructions that repeat the previous one. ``send_instruction_batch()`` sends a batch that replaces the previous one, unless a newer request has arrived in the meantime. ``system_description_fingerprint()`` tells a changed system description from a re-sent one [`#237 `__] * ``ControlTypeHandler.send_message()``. Handlers now use it to send messages, instead of putting them on the CEM's sending queue themselves [`#237 `__] -* ``FRBCSimple`` takes an optional ``fill_rate_sensor_id``, and records the actuator's fill rate there. The fill rate is recorded in ``/s``, not as a power: S2 states a fill rate as fill level per second, which for a heat buffer is not electricity at all [`#236 `__, `#244 `__] +* ``FRBCSimple`` takes an optional ``fill_rate_sensor_id``, and records the actuator's fill rate there. The fill rate is recorded in ``fill_rate_unit``, not in ``power_unit``: a fill rate is what the actuator delivers into its buffer, which for a heat buffer is not electricity at all [`#236 `__, `#244 `__] * ``flexmeasures_client.s2.clock``, with a ``WallClock`` and a ``SimulatedClock``, and ``CEM(clock=...)``. The clock sets the time for the CEM and for every handler it registers. That includes their timers, and the time a recorded state of charge is known from [`#264 `__, `#268 `__] * ``CEM(accept_simulation_metadata=True)`` accepts messages from a co-simulation that are wrapped with the simulated time (``S2Wrapper``), and sets the CEM's clock from them. It is off by default, and then a wrapped message is refused as invalid. The example server turns it on with ``S2_ACCEPT_SIMULATION_METADATA`` [`#264 `__] -* ``flexmeasures_client.s2.site``, with ``FRBC_SITE_SENSORS``, ``ensure_site`` and ``FRBCSiteConfigurator``. It gives each resource manager its own FlexMeasures site, named after it, with an FRBC handler connected to that site's sensors [`#265 `__] +* ``flexmeasures_client.s2.site``, with ``FRBC_SITE_SENSORS``, ``ensure_site`` and ``FRBCSiteConfigurator``. It gives each resource manager its own FlexMeasures site, named after it, with an FRBC handler connected to that site's sensors. The sensors' units suit a residential site and match the handler's defaults: powers and rates of fill level in ``W``, energy and fill levels in ``kWh``, and prices per ``kWh``. For sites at a larger scale, such as charge points, pass ``sensors`` with other units, and the handler's ``power_unit`` and ``fill_rate_unit`` in ``handler_options``. [`#265 `__, `#289 `__] * ``CEM(site_configurator=...)`` sets up a resource manager's site once its details arrive. Until then, the CEM answers its messages with a temporary error. The example server enables this with ``S2_SITE_PER_RESOURCE_MANAGER`` [`#265 `__] * ``examples/S2-co-simulation``: a FlexMeasures stack with one CEM per simulated home, for co-simulations whose resource managers connect over websockets [`#265 `__, `#267 `__] * ``flexmeasures_client.s2.measurements.MeasurementRecorder`` and ``CEM(measurement_recorder=...)`` record a resource manager's measured power. Use one recorder per server, so that a resource manager that reconnects is not recorded twice [`#266 `__] +* ``FRBCSimple`` can correct a usage forecast that no plan can absorb, instead of leaving the storage without a plan. Give it the new ``usage_correction_sensor_id``; it needs FlexMeasures v1.1.0 or newer. A forced gain at a full storage, or a draw at an empty one that the actuator cannot keep up with, used to make every plan infeasible. + + - The plan may move each step's expected usage towards zero, never past it, and only where nothing physical keeps the storage within its hard bounds. It is recorded on the new sensor. + - A negative expected usage is then planned as stated, instead of as zero. + - The request sets the breach price of the fill-level band relative to the highest market price over the window (``soc_breach_price_factor``, default 10), instead of leaving it to FlexMeasures. The correction is priced above anything else a unit of stock can be worth, including the site's own stored peak and commitment prices, except a breach of the site's capacity. The site's breach prices are kept from the asset, or else set at FlexMeasures' default, which FlexMeasures itself only fills in for a flex context without commodities, so a site keeps to the capacity it is steered with. + - Where a correction cannot be planned, for example without readable prices, the forecast is planned without its negative values, and the log says why. + - ``FRBCSiteConfigurator`` takes ``correct_usage`` to give each site the sensor, and the example server does so with ``S2_CORRECT_USAGE``. Both are off by default. + + [`#286 `__] Changed ------- @@ -84,7 +94,7 @@ Changed - The hard bounds ``soc-min`` and ``soc-max`` are now zero and ``hard_bound_factor`` times the top of the range, raised to include a state of charge above it. In v0.9.0, the declared range itself was the hard bound. So a storage that really was outside it, for example floating above the top or left below the bottom, made the problem infeasible: there was no schedule at all, or one from a fallback scheduler. A storage reported below zero is planned from zero, with a warning, because FlexMeasures refuses a negative ``soc-min``. - FlexMeasures older than v0.26.0 cannot relax the band. Then the band is left out, and this is logged, so the storage is planned anywhere between the wide hard bounds. - The usage forecast is sent as ``soc-usage`` if its sensor is given and the resource manager provides a forecast (or has sent one anyway). - - The leakage behaviour is recorded as a ``storage-efficiency`` with each schedule request, over the window of the schedule. It uses the leakage at the starting fill level, as a share of that fill level (or of the middle of the range, if the storage starts below that). It is only named in the flex model if FlexMeasures confirmed storing it. FlexMeasures takes a storage efficiency as a share of the stock, whereas S2 states a rate, so the loss is exact only at that one fill level. + - The leakage behaviour is planned as part of the usage: the leakage rate of the fill-level range the storage starts in is added to ``soc-usage`` over the whole window, and sent with the request. S2 states leakage as a rate, so it is not planned as a ``storage-efficiency``, which FlexMeasures takes as a share of the stock, and which would make the storage leak in proportion to how full it is. A flat rate keeps draining an empty storage, as any forced usage does; with a usage correction, the plan can take that away. It needs FlexMeasures v0.23.0 or newer, which takes a usage with the request; an older server is planned without leakage, and this is logged [`#289 `__]. [`#244 `__, `#245 `__, `#259 `__, `#271 `__, `#273 `__] * ``FRBCSimple`` posts the state of charge in its ``energy_unit`` (``kWh`` by default), and sets the flex model's ``soc-unit`` to match (to ``kWh`` if ``energy_unit`` is neither ``kWh`` nor ``MWh``). In v0.9.0, every state of charge and every schedule's ``soc-unit`` was labelled ``MWh``, whatever unit the resource manager reported in. On the example server's ``kWh`` sensor, a fill level of 10 was recorded as 10,000. A resource manager that reports in megawatt hours now needs ``energy_unit="MWh"`` [`#244 `__] diff --git a/examples/S2-co-simulation/README.md b/examples/S2-co-simulation/README.md index 6e95fa06..ee2313c0 100644 --- a/examples/S2-co-simulation/README.md +++ b/examples/S2-co-simulation/README.md @@ -18,6 +18,16 @@ with two switches on: resource manager retries. Its schedules start from the simulated now and cover 24 hours. The site's prices are for whoever runs the simulation to post. +A third switch is off by default: + +- `S2_CORRECT_USAGE`: a usage forecast that no plan can absorb within the storage's + hard bounds is corrected towards zero, instead of leaving the home without a plan. + Each site gets a `usage correction` sensor, which records where the plan corrected + the forecast. A negative expected usage is then planned as the gain it states, + rather than as none. The request also sets the breach prices of the fill-level band + and the site's capacity relative to the market, rather than leaving them to + FlexMeasures' defaults. It needs FlexMeasures v1.1.0 or newer. + FlexMeasures reads three settings a co-simulation needs from its instance config (`flexmeasures-instance/flexmeasures.cfg` in the flexmeasures checkout), not from the environment: diff --git a/examples/S2-co-simulation/docker-compose.cem.yml b/examples/S2-co-simulation/docker-compose.cem.yml index 9304b049..d57c2d58 100644 --- a/examples/S2-co-simulation/docker-compose.cem.yml +++ b/examples/S2-co-simulation/docker-compose.cem.yml @@ -37,6 +37,8 @@ x-cem: &cem S2_ACCEPT_SIMULATION_METADATA: "1" # And each home is a site of its own, rather than all of them sharing one. S2_SITE_PER_RESOURCE_MANAGER: "1" + # Uncomment to correct a usage forecast that no plan can absorb, instead of leaving the home without a plan (see README.md). + # S2_CORRECT_USAGE: "1" volumes: - ../flexmeasures-client:/app/flexmeasures-client:rw entrypoint: ["/bin/sh", "-c"] diff --git a/src/flexmeasures_client/s2/control_types/FRBC/frbc_simple.py b/src/flexmeasures_client/s2/control_types/FRBC/frbc_simple.py index 748c6702..dc9262fe 100644 --- a/src/flexmeasures_client/s2/control_types/FRBC/frbc_simple.py +++ b/src/flexmeasures_client/s2/control_types/FRBC/frbc_simple.py @@ -36,6 +36,7 @@ from flexmeasures_client.s2.control_types.FRBC import FRBC, STORAGE_STATUS_REQUESTS from flexmeasures_client.s2.control_types.FRBC.utils import ( WATTS_PER_ENERGY_UNIT_PER_SECOND, + WATTS_PER_FILL_RATE_UNIT, clip_fill_level_target_profile, element_for_fill_level, fm_schedule_to_instructions, @@ -45,7 +46,6 @@ operation_mode_conversion_efficiency, ) from flexmeasures_client.s2.control_types.translations import ( - leakage_behaviour_to_storage_efficiency, translate_fill_level_target_profile, translate_usage_forecast_to_fm, ) @@ -111,6 +111,17 @@ def _merged(ranges: list[tuple[float, float]]) -> list[tuple[float, float]]: return merged +def _added_up(*usages: pd.Series | None) -> pd.Series | None: + """Return the sum of the usages given, per step, counting a step a usage does not cover as zero, or None if none is given.""" + given = [usage for usage in usages if usage is not None] + if not given: + return None + total = given[0] + for usage in given[1:]: + total = total.add(usage, fill_value=0) + return total + + def _without_negative_usage(usage_forecast): """Return a usage forecast in which each negative expected rate is replaced by zero. @@ -134,6 +145,179 @@ def _without_negative_usage(usage_forecast): ) +#: The FlexMeasures version from which a usage forecast is planned with a correction, instead of being clipped. +#: The correction is a second device sharing the storage's stock, priced on a commodity of its own. +USAGE_CORRECTION_SINCE = "1.1.0" + +#: The FlexMeasures version from which a flex model takes a usage as a time series, rather than only as a sensor. +#: From then on, a usage forecast is recorded as stated, and a schedule request carries the forecast it plans with. +INLINE_USAGE_SINCE = "0.23.0" +#: The commodity that a usage correction is planned on, apart from the site's electricity. +USAGE_CORRECTION_COMMODITY = "usage correction" + + +def _usage_correction_limits( + usage_per_hour: pd.Series, + start: datetime, + steps: int, + resolution: timedelta, + unit: str, +) -> tuple[list[dict], list[dict]]: + """Return the correction's consumption and production capacities per step, as FlexMeasures time series. + + The correction may only move the expected usage towards zero. + Where usage is expected, it may give up to that usage back to the stock: that is its consumption capacity. + Where a gain is expected, it may take up to that gain away: that is its production capacity. + A step without a forecast allows no correction. + + :param usage_per_hour: expected usage per step, as a rate of fill level, indexed by the start of each step. + :param unit: the unit of that rate, such as kW (see fill_rate_unit). + """ + + def segments(values: list[float]) -> list[dict]: + merged: list[dict] = [] + for i, value in enumerate(values): + if merged and merged[-1]["_value"] == value: + merged[-1]["_steps"] += 1 + continue + merged.append( + {"_start": start + i * resolution, "_steps": 1, "_value": value} + ) + return [ + { + "start": pd.Timestamp(m["_start"]).isoformat(), + "duration": pd.Timedelta(resolution * m["_steps"]).isoformat(), + "value": f"{m['_value']} {unit}", + } + for m in merged + ] + + index = pd.date_range(start, periods=steps, freq=resolution) + usage = usage_per_hour.reindex(index).fillna(0.0) + return ( + segments([max(0.0, float(u)) for u in usage]), + segments([max(0.0, -float(u)) for u in usage]), + ) + + +#: Flex-context fields that FlexMeasures reads at the top level, also in a flex context with commodities. +TOP_LEVEL_FLEX_CONTEXT_FIELDS = ( + "relax-constraints", + "relax-soc-constraints", + "relax-capacity-constraints", + "relax-site-capacity-constraints", + "soc-minima-breach-price", + "soc-maxima-breach-price", + "consumption-breach-price", + "production-breach-price", +) + + +#: The breach price per kW that FlexMeasures fills in for a site's capacity, where that capacity may be breached. +#: It only does so for a flex context without commodities, so a request with commodities states it itself. +FLEXMEASURES_SITE_BREACH_PRICE_PER_KW = 10000 + +#: The flex-context fields that name inflexible devices, which FlexMeasures treats as one family. +#: Where a request names any of them at the top level, FlexMeasures does not merge in the stored ones. +INFLEXIBLE_DEVICE_FIELDS = ( + "inflexible-device-sensors", + "inflexible-consumption", + "inflexible-production", +) + +#: Flex-context fields that a commodity's own entry accepts, as FlexMeasures' commodity flex-context schema lists them. +#: Other stored fields, such as aggregate-power, are only valid at the top level, where FlexMeasures merges them in itself. +COMMODITY_FLEX_CONTEXT_FIELDS = ( + "consumption-price", + "production-price", + "site-power-capacity", + "site-consumption-capacity", + "site-production-capacity", + "site-consumption-breach-price", + "site-production-breach-price", + "site-peak-consumption", + "site-peak-consumption-price", + "site-peak-production", + "site-peak-production-price", + "consumption-breach-price", + "production-breach-price", + "soc-minima-breach-price", + "soc-maxima-breach-price", + "relax-constraints", + "relax-soc-constraints", + "relax-capacity-constraints", + "relax-site-capacity-constraints", + "commitments", + "inflexible-device-sensors", + "inflexible-consumption", + "inflexible-production", + "aggregate-consumption", + "aggregate-production", +) + + +def _price_per_mw(value, currency: str) -> float | None: + """Return a fixed price per MW, such as "260 EUR/MW" or "0.26 EUR/kW", in the given currency, or None if it is not one.""" + if not isinstance(value, str): + return None + parts = value.split() + if len(parts) != 2 or "/" not in parts[1]: + return None + price_currency, per = parts[1].split("/", 1) + factor = {"MW": 1.0, "kW": 1e3, "W": 1e6}.get(per) + if price_currency != currency or factor is None: + return None + try: + return abs(float(parts[0])) * factor + except ValueError: + return None + + +def _usage_correction_price( + max_abs_price: float, + soc_breach_price: float, + charging_efficiency: float, + storage_efficiency: float, + steps: int, + step_hours: float, + other_prices_per_mw: float = 0.0, +) -> float: + """Return a price per MWh of stock at which correcting the usage forecast is always the last resort. + + It is twice an upper bound on what one MWh of stock can be worth to the scheduler, short of a breach of the site's capacity. + Below that worth, the scheduler would rather correct the forecast than buy energy or breach the band. + The bound adds two ways a MWh of stock can be worth something. + + - Through the actuator: buying it, or disposing of it, at the highest absolute market price, + plus any other price per MW on the site's power. + This is divided by the charging efficiency, and by the storage losses of every step it may have to be held. + - Through the fill-level band: avoiding a breach on either side, + priced once for the largest breach and once for every step a breach lasts. + + :param max_abs_price: the highest absolute market price over the window, per MWh of electricity. + :param soc_breach_price: the breach price of the fill-level band, per MWh of stock, on each side. + :param charging_efficiency: the stock gained per unit of electricity, which a heat pump's COP puts above one. + :param storage_efficiency: the share of the stock that remains after one step. + :param other_prices_per_mw: the sum of any other prices per MW on the site's power, such as peak prices and commitments, + each of which one step's MWh can cost at most the price divided by the step's length. + + An efficiency above one is counted as one. + A heat pump's COP makes buying stock cheaper, but feeding it back is then worth up to the price itself, + and a storage that gains by itself makes holding stock no costlier. + + A breach of the site's capacity is left out of the bound, so a correction can be cheaper than one. + Pricing the correction above it would take a price far beyond what the solver handles reliably at FlexMeasures' default site breach price, + and lowering the site breach price to fit makes a site ignore the capacity it is steered with. + The correction stays dearer than a breach of the band, so a plan still breaches the band before it corrects the forecast. + """ + per_breach = 1 / step_hours + steps + via_actuator = (max_abs_price + other_prices_per_mw / step_hours) / ( + min(charging_efficiency, 1.0) * min(storage_efficiency, 1.0) ** steps + ) + via_band = 2 * soc_breach_price * per_breach + return 2 * (via_actuator + via_band) + + def _operation_mode_power_bands( power_ranges: list[tuple[float, float]], power_unit: str ) -> list[dict[str, list[str]]]: @@ -238,14 +422,24 @@ def __init__( soc_minima_sensor_id: int | None = None, soc_maxima_sensor_id: int | None = None, usage_forecast_sensor_id: int | None = None, - # The resolution of this sensor must equal the handler's resolution. - # A storage efficiency is what remains after one step, and this handler computes it for a step of its own resolution. - # FlexMeasures reads it as one step of the sensor's resolution. + # Also optional: a sensor to record the leakage rate on, in the fill-rate unit. + # With it, and FlexMeasures v0.23.0 or newer, the leakage the resource manager declares is planned as part of the usage. + # Without it, the storage is planned as if it lost nothing by itself. leakage_behaviour_sensor_id: int | None = None, # Also optional. # Without a sensor to record it on, no charging efficiency is declared. # Then the scheduler plans the storage as if the actuator converted power into stored energy one to one. charging_efficiency_sensor_id: int | None = None, + # Also optional: a power sensor on the site's asset, to plan a correction of the usage forecast on. + # With it, and FlexMeasures v1.1.0 or newer, an expected usage or gain that no plan can absorb is corrected towards zero, + # instead of making every plan infeasible. + # A negative expected usage is then planned as the gain it states, instead of as no usage. + # Without it, a negative expected usage is recorded and planned as no usage, as before. + usage_correction_sensor_id: int | None = None, + # The breach price of the fill-level band sent with a corrected request, per MWh of stock, + # as a multiple of the highest absolute market price over the window. + # The correction is priced from it, so it is only sent where a correction is planned. + soc_breach_price_factor: float = 10.0, timezone: str = "UTC", schedule_duration: timedelta = timedelta(hours=12), # How many times to request a schedule before leaving the storage without an instruction, @@ -271,6 +465,10 @@ def __init__( # The default is kilowatt-hours, because S2 is a residential protocol, and a home battery or heat buffer is of that order. # A deployment at another scale can set a different unit here. energy_unit: str = "kWh", + # The unit that rates of fill level are recorded in: the fill rate, the usage forecast and the leakage. + # S2 states them per second, which in kWh/s makes tiny numbers, so they are recorded in a power unit: W, kW or MW. + # The default is watts, in which a residential heat buffer's leakage of some tens of watts reads as such. + fill_rate_unit: str = "W", ) -> None: super().__init__(max_size) self._power_sensor_id = power_sensor_id @@ -291,8 +489,20 @@ def __init__( self._usage_forecast_sensor_id = usage_forecast_sensor_id self._leakage_behaviour_sensor_id = leakage_behaviour_sensor_id self._charging_efficiency_sensor_id = charging_efficiency_sensor_id + self._usage_correction_sensor_id = usage_correction_sensor_id + if soc_breach_price_factor <= 0: + # A breach price of zero makes breaching free, so the band would mean nothing. + raise ValueError( + f"soc_breach_price_factor has to be greater than zero, but {soc_breach_price_factor} was given." + ) + self._soc_breach_price_factor = soc_breach_price_factor + #: The charging efficiency that the last plan was asked for with, which the correction price depends on. + self._planned_charging_efficiency: float | None = None + #: The currency of each price sensor, read once, since it is fixed when the sensor is set up. + self._price_currency: dict[int, str] = {} self._charging_efficiency_recorded = None - self._storage_efficiency_recorded: tuple[datetime, float] | None = None + #: The window and rate of the leakage last recorded, so that the same one is not recorded again. + self._leakage_recorded: tuple[datetime, float] | None = None if resolution <= timedelta(0) or schedule_duration <= timedelta(0): raise ValueError( f"resolution and schedule_duration have to be positive, but {resolution} and {schedule_duration} were given." @@ -340,24 +550,34 @@ def __init__( f"energy_unit has to be one of {', '.join(WATTS_PER_ENERGY_UNIT_PER_SECOND)}, but {energy_unit!r} was given." ) self.energy_unit = energy_unit + if fill_rate_unit not in WATTS_PER_FILL_RATE_UNIT: + raise ValueError( + f"fill_rate_unit has to be one of {', '.join(WATTS_PER_FILL_RATE_UNIT)}, but {fill_rate_unit!r} was given." + ) + self._fill_rate_unit = fill_rate_unit # delay the start of the schedule from the time `valid_from` # of the FRBC.SystemDescription. self._valid_from_shift = valid_from_shift def fill_rate_unit(self) -> str: - """The unit a rate of change of the fill level is recorded in. + """The unit a rate of change of the fill level is recorded in: the fill_rate_unit given, by default W. - S2 states a fill rate, and a usage forecast, as fill level per second, so this follows - whatever unit the fill level itself is recorded in. It is not the power unit: a fill rate - is what an actuator delivers into its buffer, which for a heat buffer is not electricity - at all, and which even for a battery differs from kilowatts by the seconds in an hour. + S2 states a fill rate, a usage forecast and a leakage as fill level per second. + Per second, a residential heat buffer's rates are tiny numbers in kWh/s, so they are recorded in a power unit instead (see _per_second_in_fill_rate_unit). + It is not the power unit: a fill rate is what an actuator delivers into its buffer, which for a heat buffer is not electricity at all. - A method rather than a property, because S2 Handler's discover() walks dir(self) and - getattrs every name during __init__, before a subclass has assigned its own attributes. + A method rather than a property, because S2 Handler's discover() walks dir(self) and getattrs every name during __init__, before a subclass has assigned its own attributes. A property would be evaluated there and raise on the energy_unit it reads. """ - return f"{self.energy_unit}/s" + return self._fill_rate_unit + + def _per_second_in_fill_rate_unit(self, rate): + """Return a rate of fill level per second, as S2 states it, in the fill-rate unit; a number or a series.""" + return rate * ( + WATTS_PER_ENERGY_UNIT_PER_SECOND[self.energy_unit] + / WATTS_PER_FILL_RATE_UNIT[self._fill_rate_unit] + ) def align_to_resolution(self, moment: datetime) -> datetime: """The last resolution boundary at or before that moment. @@ -808,10 +1028,13 @@ async def send_actuator_status(self, status: FRBCActuatorStatus): # useful quantity - what the actuator delivers into its buffer - it simply is not power. if self._fill_rate_sensor_id is not None: fill_rate = element.fill_rate - rate = ( - fill_rate.start_of_range - + (fill_rate.end_of_range - fill_rate.start_of_range) * factor - ) * self._fill_level_scale + rate = self._per_second_in_fill_rate_unit( + ( + fill_rate.start_of_range + + (fill_rate.end_of_range - fill_rate.start_of_range) * factor + ) + * self._fill_level_scale + ) # Catch failures, like for the power above, because this runs in a task that nobody awaits. try: await self._fm_client.post_sensor_data( @@ -873,6 +1096,7 @@ async def _record_charging_efficiency( return False efficiency = min(efficiencies.values()) + self._planned_charging_efficiency = efficiency if len(set(efficiencies.values())) > 1: # FlexMeasures takes one charging efficiency for the whole device, and the modes disagree. # The lowest goes out: a plan that converts less than the actuator really does reaches its target early, @@ -969,7 +1193,7 @@ async def _ask_for_a_schedule( self, start: datetime, flex_context: dict, - flex_model: dict, + flex_model: dict | list[dict], request_number: int | None = None, ) -> dict | list[dict] | None: """Ask FlexMeasures for a schedule, and retry if the failure is worth retrying. @@ -1183,7 +1407,7 @@ async def _plan( # FlexMeasures only accepts a state of charge in kWh or MWh. # So a fill level recorded in another energy unit is converted here, at the edge, instead of being refused. # The bounds used for translating the schedule below stay in the recorded unit. - # The usage and the leakage are read from sensors, which FlexMeasures converts itself. + # The usage and the leakage are given in fill-rate units, which FlexMeasures converts itself. # The band is converted here. soc_unit, to_soc_unit = self._soc_unit() flex_model: dict = { @@ -1221,10 +1445,8 @@ async def _plan( self._logger.info( "Planning without usage: the resource manager says it provides no usage forecast, and has sent none." ) - if await self._record_storage_efficiency(start, fill_level_at_start): - flex_model["storage-efficiency"] = { - "sensor": self._leakage_behaviour_sensor_id - } + # What the storage loses by itself, planned as part of the usage (see _with_usage_correction). + leakage = await self._planned_leakage(start, fill_level_at_start) # What the operation modes say about converting drawn power into stored energy. # Recorded first, since FlexMeasures reads it from the sensor named here and not from the flex model, # and named only where it was recorded. @@ -1287,8 +1509,17 @@ async def _plan( if self._relax_site_capacity is not None: flex_context["relax-site-capacity-constraints"] = self._relax_site_capacity + # Plan a correction of the usage forecast where one is configured and the server supports it. + # Then an expected usage or gain that no plan can absorb is corrected towards zero, instead of making every plan infeasible. + corrected = await self._with_usage_correction( + start, flex_model, flex_context, leakage + ) + request_model: dict | list[dict] = flex_model + if corrected is not None: + request_model, flex_context = corrected + schedule = await self._ask_for_a_schedule( - start, flex_context, flex_model, request_number=request_number + start, flex_context, request_model, request_number=request_number ) if schedule is None: return @@ -1321,9 +1552,14 @@ async def _plan( # Pass what the storage loses by itself and what is drawn from it, as the resource manager last reported. # Then the fill level used for each step can fall as well as rise. leakage_behaviour=self._latest(self._leakage_behaviour_history), - # Leave out negative usage, as when the forecast was recorded and planned against (see send_usage_forecast). - usage_forecast=_without_negative_usage( + # Use the usage that the schedule was planned against (see send_usage_forecast). + # With a correction, that is the forecast as stated: the plan corrects it only at the hard bounds, + # where this translation holds the fill level as well. + # Without one, negative usage is left out. + usage_forecast=( self._latest(self._usage_forecast_history) + if corrected is not None + else _without_negative_usage(self._latest(self._usage_forecast_history)) ), ) @@ -1404,6 +1640,391 @@ async def send_fill_level_target_profile( start, ) + async def _corrects_usage(self) -> bool: + """Return whether a usage forecast is planned with a correction. + + That needs a sensor to plan the correction on, and a FlexMeasures that can share a stock between devices. + If the server's version cannot be read, the forecast is treated as before. + """ + if self._usage_correction_sensor_id is None: + return False + try: + return await self._fm_client.server_version_at_least(USAGE_CORRECTION_SINCE) + except REQUEST_FAILURES: + return False + + async def _plans_usage_inline(self) -> bool: + """Whether a schedule request carries the usage forecast it plans with, rather than pointing at the usage sensor.""" + try: + return await self._fm_client.server_version_at_least(INLINE_USAGE_SINCE) + except REQUEST_FAILURES: + return False + + def _aligned_usage(self) -> pd.Series | None: + """Return the latest usage forecast per step, in the fill-rate unit, indexed by step start, as it is recorded.""" + usage_forecast = self._latest(self._usage_forecast_history) + if usage_forecast is None: + return None + translated = self._per_second_in_fill_rate_unit( + translate_usage_forecast_to_fm( + usage_forecast, + self._resolution, + strategy="mean", + fill_level_scale=self._fill_level_scale, + ) + ) + return pd.Series( + translated.values, + index=pd.date_range( + self.align_to_resolution(usage_forecast.start_time), + periods=len(translated), + freq=self._resolution, + ), + ) + + def _as_time_series(self, usage: pd.Series) -> list[dict]: + """Return a usage per step as a FlexMeasures time series, in the fill-rate unit.""" + return [ + { + "start": pd.Timestamp(moment).isoformat(), + "duration": pd.Timedelta(self._resolution).isoformat(), + "value": f"{float(value)} {self.fill_rate_unit()}", + } + for moment, value in usage.items() + ] + + async def _max_abs_price(self, start: datetime) -> tuple[float, str] | None: + """Return the highest absolute market price over the window, per MWh, and its currency. + + Both price sensors are read, for drawing and for feeding back. + None is returned if either cannot be read, if they use different currencies, or if neither holds a price for the window. + """ + highest: float | None = None + currency: str | None = None + for sensor_id in dict.fromkeys( + (self._price_sensor_id, self._production_price_sensor_id) + ): + try: + if sensor_id not in self._price_currency: + sensor = await self._fm_client.get_sensor(sensor_id) + unit = sensor.get("unit", "") if isinstance(sensor, dict) else "" + self._price_currency[sensor_id] = unit.split("/")[0].strip() + sensor_currency = self._price_currency[sensor_id] + data = await self._fm_client.get_sensor_data( + sensor_id, + start=start, + duration=self._schedule_duration, + unit=f"{sensor_currency}/MWh", + resolution=self._resolution, + ) + except REQUEST_FAILURES as exc: + self._logger.warning( + "Could not read the prices on sensor %s: %s", sensor_id, exc + ) + return None + if currency is not None and sensor_currency != currency: + self._logger.warning( + "The price sensors use different currencies, %s and %s.", + currency, + sensor_currency, + ) + return None + currency = sensor_currency + values = data.get("values", []) if isinstance(data, dict) else [] + for value in values: + if value is not None and not math.isnan(value): + highest = ( + abs(value) if highest is None else max(highest, abs(value)) + ) + if highest is None or currency is None: + return None + return highest, currency + + async def _stored_flex_context(self) -> dict: + """Return the flex context stored on the site's asset, which FlexMeasures merges into each request. + + :raises: what the client raises where the asset cannot be read. + """ + sensor = await self._fm_client.get_sensor(self._power_sensor_id) + asset_id = sensor.get("generic_asset_id") if isinstance(sensor, dict) else None + if asset_id is None: + raise ValueError(f"power sensor {self._power_sensor_id} states no asset") + asset = await self._fm_client.get_asset(asset_id, parse_json_fields=True) + stored = asset.get("flex_context") if isinstance(asset, dict) else None + return stored if isinstance(stored, dict) else {} + + async def _highest_per_mw_on_sensor( + self, sensor_id: int, start: datetime, currency: str + ) -> float | None: + """Return the highest absolute price per MW on a sensor over the window, zero if it holds none, or None if it cannot be read.""" + try: + data = await self._fm_client.get_sensor_data( + sensor_id, + start=start, + duration=self._schedule_duration, + unit=f"{currency}/MW", + resolution=self._resolution, + ) + except REQUEST_FAILURES as exc: + self._logger.warning( + "Could not read the prices on sensor %s: %s", sensor_id, exc + ) + return None + values = data.get("values", []) if isinstance(data, dict) else [] + return max( + ( + abs(value) + for value in values + if value is not None and not math.isnan(value) + ), + default=0.0, + ) + + async def _per_mw(self, value, start: datetime, currency: str) -> float | None: + """Return the highest price per MW a stored price can take over the window, or None if it cannot be bounded. + + A fixed price is read as it is; a price on a sensor is read from the sensor over the window, + as a co-simulation's controller sets its steering prices. + """ + if isinstance(value, dict) and set(value) == {"sensor"}: + return await self._highest_per_mw_on_sensor( + value["sensor"], start, currency + ) + return _price_per_mw(value, currency) + + async def _bound_stored_prices( + self, stored: dict, skipped_fields, start: datetime, currency: str + ) -> tuple[float, str | None]: + """Return the sum of the stored per-MW prices that count towards the correction price, or a reason why one cannot be bounded. + + The skipped fields do not count: the request overrides some, and the site's breach prices are left out of the bound. + A price that FlexMeasures reads at the top level is set by the request, so it does not count either. + Every other price is part of what a unit of stock can be worth. + So it has to be a fixed price per MW, or a sensor of prices per MW, to be bounded. + """ + total = 0.0 + for key, value in stored.items(): + if key in skipped_fields or key in TOP_LEVEL_FLEX_CONTEXT_FIELDS: + continue + if key == "commitments": + for commitment in value or []: + for side in ("up-price", "down-price"): + if side not in commitment: + continue + price = await self._per_mw(commitment[side], start, currency) + if price is None: + return ( + 0.0, + f"the stored commitment {commitment.get('name')!r} has a {side} that is not a price per MW in {currency}", + ) + total += price + elif key.endswith("-price"): + price = await self._per_mw(value, start, currency) + if price is None: + return ( + 0.0, + f"the stored {key} is not a price per MW in {currency}", + ) + total += price + return total, None + + async def _with_usage_correction( + self, + start: datetime, + flex_model: dict, + flex_context: dict, + leakage: pd.Series | None = None, + ) -> tuple[list[dict], dict] | None: + """Return the request for a schedule with a correction of the usage forecast, or None to plan without one. + + A forecast may expect more usage, or more gain, than any plan can absorb within the hard bounds. + S2 does not say what happens then, so the expectation is corrected towards zero, by as little as needed. + The correction is a second device on the storage's stock, on a commodity of its own, so it never touches the site's electricity. + It may only move each step's expected usage towards zero, and is priced above anything else a unit of stock can be worth to the scheduler. + So it is only used where nothing physical can keep the plan within the hard bounds. + + The request carries a coherent set of prices to make that hold. + The breach price of the fill-level band is set relative to the market price. + The correction price follows from it, and from any other per-MW price stored on the site's asset, except a breach of the site's capacity (see _usage_correction_price). + The site's breach prices are left to the asset, or else set at FlexMeasures' default, so that the site keeps to the capacity it is steered with. + Device capacities are kept hard, so that the correction cannot move a usage past zero. + + The usage is sent with the request, from the same forecast as the correction's limits and the translation back into instructions. + The usage sensor can hold other forecasts, including negative ones recorded as stated, which the correction would not cover. + + Where a correction is configured but cannot be planned, the forecast is sent without its negative values. + Without a correction, it is sent the same way, from a server that takes a usage with the request on; + an older server plans with what is recorded, which is then recorded without its negative values. + + The leakage, if planned (see _planned_leakage), is added to the forecast before any of this. + So the correction can also take away leakage that an empty storage cannot give. + """ + forecast_named = "soc-usage" in flex_model + corrects = await self._corrects_usage() + if not corrects and not await self._plans_usage_inline(): + # This server plans with what is recorded, and no negative usage is recorded for it. + # No leakage is planned for it either (see _planned_leakage). + return None + forecast = self._aligned_usage() if forecast_named else None + if forecast_named and forecast is None: + # The sensor may hold negative usage recorded as stated, from before this handler started, + # which a plan without a correction cannot absorb. A plan without usage can. + self._logger.info( + "Planning without a usage forecast: no FRBC.UsageForecast has arrived since this handler started." + ) + usage = _added_up(forecast, leakage) + if usage is None: + flex_model.pop("soc-usage", None) + return None + if not corrects: + if (usage < 0).any(): + self._logger.warning( + "Planning the usage forecast without its negative values, as no usage there: without a usage correction, FlexMeasures would plan them as a gain the storage cannot refuse." + ) + flex_model["soc-usage"] = [self._as_time_series(usage.clip(lower=0))] + return None + reason = None + prices = None + stored: dict = {} + site_fields = ("consumption-price", "production-price", "site-power-capacity") + # A stored breach price of the site's capacity is carried over, but left out of the correction price. + site_breach_fields = ( + "site-consumption-breach-price", + "site-production-breach-price", + ) + other_prices_per_mw = 0.0 + if await self._soc_duration() != timedelta(0): + reason = f"state-of-charge sensor {self._soc_sensor_id} does not record instants, which FlexMeasures needs to share a stock between devices" + else: + prices = await self._max_abs_price(start) + if prices is None: + reason = "the correction is priced from the market prices over the window, which could not be read" + else: + try: + stored = await self._stored_flex_context() + except REQUEST_FAILURES as exc: + reason = f"the flex context stored on the site's asset could not be read: {exc}" + else: + if "commodities" in stored: + reason = "the site's asset stores a flex context with commodities, which this request would replace" + else: + other_prices_per_mw, reason = await self._bound_stored_prices( + stored, site_fields + site_breach_fields, start, prices[1] + ) + if prices is None or reason is not None: + self._logger.warning( + "Planning the usage forecast without a correction, and without its negative values: %s.", + reason, + ) + flex_model["soc-usage"] = [self._as_time_series(usage.clip(lower=0))] + return None + max_abs_price, currency = prices + # A market without prices would make every breach free, so a floor of one unit of currency per MWh is kept. + reference_price = max(max_abs_price, 1.0) + + steps = int(self._schedule_duration / self._resolution) + step_hours = self._resolution / timedelta(hours=1) + soc_breach_price = self._soc_breach_price_factor * reference_price + charging_efficiency = ( + self._planned_charging_efficiency + if "charging-efficiency" in flex_model and self._planned_charging_efficiency + else 1.0 + ) + correction_price = _usage_correction_price( + max_abs_price=max_abs_price, + soc_breach_price=soc_breach_price, + charging_efficiency=charging_efficiency, + # The leakage is planned as a usage, which does not grow with the stock, so holding stock loses nothing extra. + storage_efficiency=1.0, + steps=steps, + step_hours=step_hours, + other_prices_per_mw=other_prices_per_mw, + ) + consumption_limits, production_limits = _usage_correction_limits( + usage, start, steps, self._resolution, self.fill_rate_unit() + ) + + stock_fields = ( + "soc-unit", + "soc-at-start", + "soc-min", + "soc-max", + "soc-minima", + "soc-maxima", + "soc-usage", + ) + state_of_charge = {"sensor": self._soc_sensor_id} + stock: dict = {"state-of-charge": state_of_charge} + stock.update({k: v for k, v in flex_model.items() if k in stock_fields}) + # The same usage as the correction's limits, rather than whatever the sensor holds. + stock["soc-usage"] = [self._as_time_series(usage)] + actuator = { + "sensor": self._power_sensor_id, + "state-of-charge": state_of_charge, + "commodity": "electricity", + } + actuator.update({k: v for k, v in flex_model.items() if k not in stock_fields}) + correction = { + "sensor": self._usage_correction_sensor_id, + "state-of-charge": state_of_charge, + "commodity": USAGE_CORRECTION_COMMODITY, + "power-capacity": f"{float(usage.abs().max())} {self.fill_rate_unit()}", + "consumption-capacity": consumption_limits, + "production-capacity": production_limits, + # It changes the stock directly, so one for one. + "charging-efficiency": 1, + "discharging-efficiency": 1, + } + + # FlexMeasures reads the asset's stored electricity fields only where a request names no electricity commodity, + # and this one does, so they are carried over here, under what the request sets itself. + electricity = {"commodity": "electricity"} + electricity.update( + { + k: v + for k, v in stored.items() + if k in COMMODITY_FLEX_CONTEXT_FIELDS + and k not in TOP_LEVEL_FLEX_CONTEXT_FIELDS + } + ) + electricity.update({k: v for k, v in flex_context.items() if k in site_fields}) + if self._relax_site_capacity is not False: + # FlexMeasures leaves a commodity's site capacity hard where no breach price is given, + # unlike the same capacity in a flex context without commodities, so its default is stated here. + for field in ( + "site-consumption-breach-price", + "site-production-breach-price", + ): + electricity.setdefault( + field, f"{FLEXMEASURES_SITE_BREACH_PRICE_PER_KW} {currency}/kW" + ) + context = {k: v for k, v in flex_context.items() if k not in site_fields} + # FlexMeasures also merges the stored flex context in at the top level, and counts inflexible devices from there as well as from the electricity entry, + # so each would be counted twice (FlexMeasures/flexmeasures#2848). + # Naming the same fields empty at the top level keeps the stored ones out of it. + for field in INFLEXIBLE_DEVICE_FIELDS: + if field in electricity: + context[field] = [] + context["commodities"] = [ + electricity, + { + "commodity": USAGE_CORRECTION_COMMODITY, + # Correcting in either direction costs the same: drawing pays it, and feeding back earns minus it. + "consumption-price": f"{correction_price} {currency}/MWh", + "production-price": f"{-correction_price} {currency}/MWh", + }, + ] + context["soc-minima-breach-price"] = f"{soc_breach_price} {currency}/MWh" + context["soc-maxima-breach-price"] = f"{soc_breach_price} {currency}/MWh" + # Soft device capacities would let the correction move a usage past zero, and the actuator run beyond what it can. + context["relax-capacity-constraints"] = False + self._logger.debug( + "Planning with a usage correction, priced at %s %s/MWh.", + correction_price, + currency, + ) + return [actuator, stock, correction], context + async def send_usage_forecast(self, usage_forecast: FRBCUsageForecast): """Record what a resource manager expects its storage to be drawn down by.""" if self._usage_forecast_sensor_id is None: @@ -1412,20 +2033,29 @@ async def send_usage_forecast(self, usage_forecast: FRBCUsageForecast): ) return - translated = translate_usage_forecast_to_fm( - usage_forecast, - self._resolution, - strategy="mean", - fill_level_scale=self._fill_level_scale, + translated = self._per_second_in_fill_rate_unit( + translate_usage_forecast_to_fm( + usage_forecast, + self._resolution, + strategy="mean", + fill_level_scale=self._fill_level_scale, + ) ) - # A negative usage is a gain that the storage is forced to take, and FlexMeasures cannot relax it. - # Above the hard ceiling, it makes every schedule for the window infeasible, so the storage gets no instruction at all. - # S2 allows the negative sign, but no storage that this handler controls fills up by being drawn from. - # So a negative value is treated as an artefact of the forecast: it is counted, and recorded as no usage. + # S2 allows a negative usage, a gain, so it is recorded as stated. + # What to plan with is decided when a schedule is requested (see _with_usage_correction). + # Only a server that takes no usage with the request plans with what is recorded, + # and there a negative usage would be a gain the storage is forced to take, which makes every schedule above the hard ceiling infeasible. negative = translated < 0 - if negative.any(): + if negative.any() and await self._plans_usage_inline(): + self._logger.info( + "An FRBC.UsageForecast expects negative usage on %s of %s steps, down to %s; recorded as stated.", + int(negative.sum()), + len(translated), + float(translated.min()), + ) + elif negative.any(): self._logger.warning( - "An FRBC.UsageForecast expects negative usage on %s of %s steps, down to %s; recorded as no usage there, since FlexMeasures would plan it as a gain the storage cannot refuse.", + "An FRBC.UsageForecast expects negative usage on %s of %s steps, down to %s; recorded as no usage there, since this server plans with what is recorded, and would plan it as a gain the storage cannot refuse.", int(negative.sum()), len(translated), float(translated.min()), @@ -1439,7 +2069,6 @@ async def send_usage_forecast(self, usage_forecast: FRBCUsageForecast): start=self.align_to_resolution(usage_forecast.start_time), prior=self.now(), values=translated.tolist(), - # translate_usage_forecast_to_fm returns fill level per second, scaled. unit=self.fill_rate_unit(), duration=pd.Timedelta(self._resolution) * len(translated), ) @@ -1457,60 +2086,57 @@ async def send_leakage_behaviour(self, leakage_behaviour: FRBCLeakageBehaviour): "Not recording an FRBC.LeakageBehaviour: no leakage-behaviour sensor is configured." ) - async def _record_storage_efficiency( + async def _planned_leakage( self, start: datetime, fill_level: float - ) -> bool: - """Record the share of its stock that the storage keeps per step, over the window a schedule covers, from the schedule's start. + ) -> pd.Series | None: + """Return the leakage to plan with, per step of the window a schedule covers, in the fill-rate unit, or None to plan without. + + S2 states leakage as a rate, per fill-level range, so it is planned as a usage, rather than as a share of the stock the storage keeps. + A share would make the storage leak in proportion to how full it is, which S2 does not say. + The rate is that of the range the schedule starts in, constant over the window, since the plan moves the fill level. + A flat rate keeps draining an empty storage, as any forced usage does, which a usage correction can take away. - :returns: whether the schedule may be planned against the sensor. It may not, if nothing is recorded on the sensor. + The rate is also recorded on the leakage sensor, over the same window, so that it can be seen next to the schedule. """ leakage_behaviour = self._latest(self._leakage_behaviour_history) - if ( - self._leakage_behaviour_sensor_id is None - or leakage_behaviour is None - or not self._system_description_history - ): - return False - # S2 gives leakage as a rate, but FlexMeasures takes a share of the stock. - # So one share matches the rate at only one fill level; at other fill levels, the loss scales with the stock. - # That fill level is where the schedule starts, as long as that is at least the middle of the storage's range. - # From a near-empty start, a share of the start level would let almost nothing remain per step. - # The loss would then be overstated many times over, as soon as the plan filled the storage. - # Below the middle, the loss is understated instead, by at most the ratio of the middle to the stock. - # That is the milder error. - storage_range = list(self._system_description_history.values())[ - -1 - ].storage.fill_level_range - middle = (storage_range.start_of_range + storage_range.end_of_range) / 2 - # The share that remains after one step of this handler's resolution. - # FlexMeasures reads it as one step of the leakage sensor's resolution, so the two resolutions must be equal. - efficiency = leakage_behaviour_to_storage_efficiency( - message=leakage_behaviour, - resolution=self._resolution, - fill_level_scale=self._fill_level_scale, - fill_level=fill_level, - share_of=max(fill_level, middle), - ) - if self._storage_efficiency_recorded == (start, efficiency): - return True - try: - await self._fm_client.post_sensor_data( - sensor_id=self._leakage_behaviour_sensor_id, - start=start, - prior=self.now(), - values=[efficiency], - unit="%", - duration=self._schedule_duration, - # As for the charging efficiency: the value is remembered below as recorded, so it must really be recorded. - confirm_ingestion=True, - ) - except REQUEST_FAILURES: - self._logger.exception( - "Not asking for a schedule against a storage efficiency: could not record %s %% on sensor %s for %s.", - efficiency, - self._leakage_behaviour_sensor_id, - start, + if self._leakage_behaviour_sensor_id is None or leakage_behaviour is None: + return None + if not await self._plans_usage_inline(): + # Only a usage given with the request can carry the leakage. + self._logger.info( + "Planning without leakage: FlexMeasures takes a usage with the request from v%s, and this server reports %s.", + INLINE_USAGE_SINCE, + self._fm_client.server_version, ) - return False - self._storage_efficiency_recorded = (start, efficiency) - return True + return None + element = element_for_fill_level(leakage_behaviour.elements, fill_level) + rate = self._per_second_in_fill_rate_unit( + element.leakage_rate * self._fill_level_scale + ) + steps = int(self._schedule_duration / self._resolution) + leakage = pd.Series( + float(rate), + index=pd.date_range(start, periods=steps, freq=self._resolution), + ) + if self._leakage_recorded != (start, rate): + try: + await self._fm_client.post_sensor_data( + sensor_id=self._leakage_behaviour_sensor_id, + start=start, + prior=self.now(), + values=[rate], + unit=self.fill_rate_unit(), + duration=self._schedule_duration, + ) + except REQUEST_FAILURES: + # The leakage goes with the request, so the plan does not depend on it being recorded. + self._logger.exception( + "Could not record the leakage rate %s %s on sensor %s for %s; planning with it all the same.", + rate, + self.fill_rate_unit(), + self._leakage_behaviour_sensor_id, + start, + ) + else: + self._leakage_recorded = (start, rate) + return leakage diff --git a/src/flexmeasures_client/s2/control_types/FRBC/utils.py b/src/flexmeasures_client/s2/control_types/FRBC/utils.py index 4e8fb54f..bfbbc6a6 100644 --- a/src/flexmeasures_client/s2/control_types/FRBC/utils.py +++ b/src/flexmeasures_client/s2/control_types/FRBC/utils.py @@ -451,6 +451,11 @@ def get_soc_min_max( } +#: The units a rate of fill level can be recorded in, and the number of watts in one of each. +#: S2 states rates per second, which in kWh/s puts a heat buffer's leakage of some tens of watts near 1e-5, so they are recorded in a power unit instead, by default W. +WATTS_PER_FILL_RATE_UNIT: dict[str, float] = {"W": 1.0, "kW": 1e3, "MW": 1e6} + + def fill_rate_in_watts(fill_rate: float, energy_unit: str) -> float: """Convert a fill rate to watts, given the unit that the fill level is recorded in. diff --git a/src/flexmeasures_client/s2/script/websockets_server.py b/src/flexmeasures_client/s2/script/websockets_server.py index 80d736ec..7a39d403 100644 --- a/src/flexmeasures_client/s2/script/websockets_server.py +++ b/src/flexmeasures_client/s2/script/websockets_server.py @@ -22,6 +22,7 @@ from flexmeasures_client.s2.measurements import MeasurementRecorder from flexmeasures_client.s2.site import ( FRBC_SITE_SENSORS, + USAGE_CORRECTION_SENSORS, FRBCSiteConfigurator, ensure_site, ) @@ -47,8 +48,15 @@ "yes", ) +#: Whether a usage forecast that no plan can absorb is corrected towards zero, instead of being planned without its negative values. +#: This also sets the breach prices of the fill-level band and the site's capacity relative to the market. +CORRECT_USAGE = os.getenv("S2_CORRECT_USAGE", "").lower() in ("1", "true", "yes") + #: The sensors the site needs, taken from the library's table of sensors for an FRBC site. -SITE_SENSORS = FRBC_SITE_SENSORS +SITE_SENSORS = { + **FRBC_SITE_SENSORS, + **(USAGE_CORRECTION_SENSORS if CORRECT_USAGE else {}), +} #: The specs to create every site asset with: a building with a three-phase connection of 25 A per phase at 230 V. SITE_ASSET_SPECS: dict[str, Any] = dict( @@ -203,8 +211,10 @@ def fixed_site_handler(sensors: dict[str, dict]) -> FRBCSimple: soc_minima_sensor_id=sensors["soc minima"]["id"], soc_maxima_sensor_id=sensors["soc maxima"]["id"], usage_forecast_sensor_id=sensors["usage forecast"]["id"], - leakage_behaviour_sensor_id=sensors["storage efficiency"]["id"], + leakage_behaviour_sensor_id=sensors["leakage"]["id"], charging_efficiency_sensor_id=sensors["charging efficiency"]["id"], + # Only where S2_CORRECT_USAGE asked for a correction (see SITE_SENSORS). + usage_correction_sensor_id=sensors.get("usage correction", {}).get("id"), timezone=SITE_TIMEZONE, # Plan from now, not from a day after the description, so that a demo shows its instructions being followed. valid_from_shift=timedelta(0), @@ -337,6 +347,7 @@ async def configure_site_once(app: web.Application) -> None: power_sensor_attributes={"is_strictly_non_positive": True}, # Each member's site capacity is a target set by the community that steers it, so it may be breached at a price, as #188 asked for. handler_options={"relax_site_capacity": True}, + correct_usage=CORRECT_USAGE, logger=LOGGER, ) LOGGER.info("Each resource manager gets a site of its own; ready for them.") diff --git a/src/flexmeasures_client/s2/site.py b/src/flexmeasures_client/s2/site.py index 56279f9f..c2d328c2 100644 --- a/src/flexmeasures_client/s2/site.py +++ b/src/flexmeasures_client/s2/site.py @@ -38,7 +38,7 @@ class SiteSensor: #: The sensors an FRBC site needs, by name. -#: Their units are the units the CEM records in: fill levels in kWh, and fill rates per second, as in S2. +#: Their units suit a residential site, and match the handler's defaults: energy and fill levels in kWh, prices per kWh, and powers and rates of fill level in W. #: The names are a contract with any reader of the site, such as a co-simulation. #: A co-simulation looks up "price", "production price", "power", "state of charge" and "measured-power" by name. FRBC_SITE_SENSORS: dict[str, SiteSensor] = { @@ -48,19 +48,27 @@ class SiteSensor: # This sensor holds both an actuator's power and the schedule planned for it. # Without this attribute, FlexMeasures treats production as positive on a power sensor. # It would then flip the sign of an incoming schedule, so the power and the schedule would point in opposite directions. - "power": SiteSensor("kW", "PT15M", {"consumption_is_positive": True}), + "power": SiteSensor("W", "PT15M", {"consumption_is_positive": True}), # The power the resource manager measures its site drawing, kept separate from the schedule on "power". - "measured-power": SiteSensor("kW", "PT15M", {"consumption_is_positive": True}), + "measured-power": SiteSensor("W", "PT15M", {"consumption_is_positive": True}), "state of charge": SiteSensor("kWh", "PT0M"), "RM discharge": SiteSensor("dimensionless", "PT15M"), - "fill rate": SiteSensor("kWh/s", "PT15M"), + "fill rate": SiteSensor("W", "PT15M"), "soc minima": SiteSensor("kWh", "PT15M"), "soc maxima": SiteSensor("kWh", "PT15M"), - "usage forecast": SiteSensor("kWh/s", "PT15M"), - "storage efficiency": SiteSensor("%", "PT15M"), + "usage forecast": SiteSensor("W", "PT15M"), + # The leakage rate the resource manager declares, as it is planned with, in the same unit as the usage forecast. + "leakage": SiteSensor("W", "PT15M"), "charging efficiency": SiteSensor("dimensionless", "PT15M"), } +#: The sensor a site needs for the handler to correct a usage forecast that no plan can absorb (see FRBCSimple's usage_correction_sensor_id). +#: It holds the rate of fill level the correction adds to the stock: positive where it gives back usage, negative where it takes away a gain. +#: Kept apart from FRBC_SITE_SENSORS, since a site only gets it where the deployment asks for a correction. +USAGE_CORRECTION_SENSORS: dict[str, SiteSensor] = { + "usage correction": SiteSensor("W", "PT15M", {"consumption_is_positive": True}), +} + def frbc_site_graphs(sensors: dict[str, dict]) -> list[dict]: """Return a site's graphs, top to bottom, in the format of an asset's sensors_to_show in FlexMeasures. @@ -193,6 +201,7 @@ def __init__( power_sensor_attributes: dict[str, Any] | None = None, handler_options: dict[str, Any] | None = None, send_revocations: bool = True, + correct_usage: bool = False, logger: Logger | None = None, ) -> None: """ @@ -205,6 +214,9 @@ def __init__( :param handler_options: Keyword arguments for each FRBCSimple, other than its sensors. By default, it plans from now over 24 hours, because a co-simulation steps through days. :param send_revocations: Whether a newer plan revokes the remainder of the previous plan. + :param correct_usage: Whether each site gets a sensor to correct a usage forecast that no plan can absorb. + With a correction, the handler also sets the breach prices of the fill-level band and the site's capacity, + relative to the market, rather than leaving them to FlexMeasures (see FRBCSimple's usage_correction_sensor_id). """ self._fm_client = fm_client self._timezone = timezone @@ -213,6 +225,8 @@ def __init__( **(asset_specs or {}), } sensors = dict(sensors if sensors is not None else FRBC_SITE_SENSORS) + if correct_usage: + sensors.update(USAGE_CORRECTION_SENSORS) if power_sensor_attributes: power = sensors["power"] sensors["power"] = SiteSensor( @@ -261,8 +275,9 @@ def sensor_id(sensor_name: str) -> int | None: soc_minima_sensor_id=sensor_id("soc minima"), soc_maxima_sensor_id=sensor_id("soc maxima"), usage_forecast_sensor_id=sensor_id("usage forecast"), - leakage_behaviour_sensor_id=sensor_id("storage efficiency"), + leakage_behaviour_sensor_id=sensor_id("leakage"), charging_efficiency_sensor_id=sensor_id("charging efficiency"), + usage_correction_sensor_id=sensor_id("usage correction"), timezone=self._timezone, **self._handler_options, ) diff --git a/tests/s2/test_example_end_to_end.py b/tests/s2/test_example_end_to_end.py index 12a64aa3..1ea7e66f 100644 --- a/tests/s2/test_example_end_to_end.py +++ b/tests/s2/test_example_end_to_end.py @@ -178,7 +178,7 @@ async def test_the_shipped_resource_manager_is_planned_for_and_instructed(): "soc minima", "soc maxima", "usage forecast", - "storage efficiency", + "leakage", "charging efficiency", ): assert sensors[name]["id"] in posted_to, name @@ -326,7 +326,7 @@ async def test_the_site_gets_a_sensor_for_every_declaration_and_prices_from_toda made = {call.kwargs["name"]: call.kwargs for call in fm.add_sensor.await_args_list} assert "price" not in made assert made["power"]["attributes"] == {"consumption_is_positive": True} - assert made["fill rate"]["unit"] == "kWh/s" + assert made["fill rate"]["unit"] == "W" assert made["soc minima"]["unit"] == "kWh" prices = fm.post_sensor_data.await_args.kwargs assert prices["sensor_id"] == 50 @@ -538,8 +538,8 @@ def test_the_site_sensors_are_what_the_cem_records_in(): server.SITE_SENSORS["state of charge"].unit, server.SITE_SENSORS["state of charge"].resolution, ) == ("kWh", "PT0M") - assert server.SITE_SENSORS["storage efficiency"].unit == "%" - assert server.SITE_SENSORS["usage forecast"].unit == "kWh/s" + assert server.SITE_SENSORS["leakage"].unit == "W" + assert server.SITE_SENSORS["usage forecast"].unit == "W" @pytest.mark.asyncio @@ -561,18 +561,23 @@ async def test_the_demo_prices_are_seeded_again_while_the_server_runs(): assert fm.post_sensor_data.await_args.kwargs["sensor_id"] == 10 -def test_the_loss_is_a_share_of_the_fill_level_a_schedule_starts_from(): - """1.5 kWh a day from a 5 kWh buffer is a larger share of it than of a 12 kWh one.""" - from flexmeasures_client.s2.control_types.translations import ( - leakage_behaviour_to_storage_efficiency, +@pytest.mark.asyncio +async def test_the_loss_is_planned_at_the_rate_of_the_range_a_schedule_starts_in(): + """1.5 kWh a day once the buffer is warm, whatever its fill level within that range.""" + frbc = server.fixed_site_handler( + {name: {"id": i} for i, name in enumerate(server.SITE_SENSORS, start=1)} ) - + frbc._fm_client = AsyncMock() + frbc._fm_client.server_version_at_least = AsyncMock(return_value=True) leakage = leakage_behaviour(NOON) - at_five = leakage_behaviour_to_storage_efficiency( - leakage, resolution=timedelta(minutes=15), fill_level=7.0, share_of=5.0 - ) - rate = per_second(1.5 / 24) # the warm element, which applies at 7 - assert at_five == pytest.approx(100 * (1 - rate / 5.0) ** 900, rel=1e-12) + frbc._leakage_behaviour_history[str(leakage.message_id)] = leakage + + planned = await frbc._planned_leakage(NOON, 7.0) + + assert list(planned) == pytest.approx( + [1500 / 24] * len(planned) + ) # 1.5 kWh a day, in W + assert len(planned) > 0 @pytest.mark.asyncio diff --git a/tests/s2/test_frbc_declared_profiles.py b/tests/s2/test_frbc_declared_profiles.py index a1489358..70e1bc61 100644 --- a/tests/s2/test_frbc_declared_profiles.py +++ b/tests/s2/test_frbc_declared_profiles.py @@ -40,7 +40,6 @@ ) from flexmeasures_client.client import FlexMeasuresClient -from flexmeasures_client.exceptions import JobTimeoutError from flexmeasures_client.s2.control_types.FRBC.frbc_simple import FRBCSimple from flexmeasures_client.s2.control_types.FRBC.utils import ( clip_fill_level_target_profile, @@ -260,11 +259,11 @@ async def test_usage_forecast_is_recorded_from_a_resolution_boundary( assert posted["start"] == frbc.align_to_resolution(usage_forecast.start_time) assert posted["start"].minute % 15 == 0 and posted["start"].second == 0 # A usage forecast is a rate of fill level, not a power. - assert posted["unit"] == frbc.fill_rate_unit() == "kWh/s" + assert posted["unit"] == frbc.fill_rate_unit() == "W" @pytest.mark.asyncio -async def test_leakage_behaviour_is_recorded_as_a_storage_efficiency( +async def test_leakage_behaviour_is_recorded_as_a_rate( system_description, ): """Recorded when a schedule is asked for, over the window it covers, rather than on arrival.""" @@ -291,11 +290,15 @@ async def test_leakage_behaviour_is_recorded_as_a_storage_efficiency( ] # nothing on arrival start = datetime(2024, 1, 1, 12, 15, tzinfo=timezone.utc) - assert await frbc._record_storage_efficiency(start, (RANGE_BOTTOM + RANGE_TOP) / 2) + planned = await frbc._planned_leakage(start, (RANGE_BOTTOM + RANGE_TOP) / 2) + # Planned as a usage over the window, at the rate S2 states, rather than as a share of the stock. + assert planned.index[0] == start + assert len(planned) == frbc._schedule_duration / frbc._resolution + assert set(planned) == {36000.0} # 0.01 kWh/s, in W posted = _posted(frbc, LEAKAGE_SENSOR_ID) - assert posted["unit"] == "%" - assert len(posted["values"]) == 1 + assert posted["unit"] == frbc.fill_rate_unit() + assert posted["values"] == [36000.0] assert posted["start"] == start assert posted["duration"] == frbc._schedule_duration @@ -546,31 +549,44 @@ async def test_the_instruction_translation_stays_in_the_resource_managers_units( assert translate.call_args.kwargs["initial_fill_level"] == 30.0 -def test_a_fill_rate_is_not_recorded_as_a_power(): - """S2 states a fill rate as fill level per second, which is not the power unit. - - For a heat buffer the quantity is not electricity at all, and even for a battery kWh/s - differs from kW by the number of seconds in an hour. - """ - frbc = FRBCSimple( +def _handler(**units) -> FRBCSimple: + return FRBCSimple( power_sensor_id=1, price_sensor_id=2, soc_sensor_id=3, rm_discharge_sensor_id=4, - energy_unit="kWh", - power_unit="W", + **units, ) - assert frbc.fill_rate_unit() == "kWh/s" - assert frbc.fill_rate_unit() != frbc.power_unit - megawatt_hours = FRBCSimple( - power_sensor_id=1, - price_sensor_id=2, - soc_sensor_id=3, - rm_discharge_sensor_id=4, - energy_unit="MWh", - ) - assert megawatt_hours.fill_rate_unit() == "MWh/s" + +@pytest.mark.parametrize( + "units, fill_rate_unit, recorded", + [ + # 0.01 kWh/s is 36 kW. + ({}, "W", 36000.0), + ({"fill_rate_unit": "kW"}, "kW", 36.0), + ({"energy_unit": "MWh", "fill_rate_unit": "MW"}, "MW", 36.0), + # A joule per second is a watt. + ({"energy_unit": "kJ"}, "W", 10.0), + ({"energy_unit": "J", "fill_rate_unit": "kW"}, "kW", 0.00001), + ], +) +def test_a_fill_rate_is_recorded_in_the_fill_rate_unit_whatever_the_power_unit( + units, fill_rate_unit, recorded +): + """S2 states a fill rate as fill level per second, which in kWh/s makes tiny numbers, so it is recorded in a power unit, by default W. + + That is not the power unit, which says how an actuator's electric power is posted: for a heat buffer, a fill rate is not electricity at all. + """ + frbc = _handler(power_unit="kW", **units) + assert frbc.fill_rate_unit() == fill_rate_unit + assert frbc._per_second_in_fill_rate_unit(0.01) == pytest.approx(recorded) + + +@pytest.mark.parametrize("fill_rate_unit", ["kWh/s", "kWh", "GW"]) +def test_a_fill_rate_unit_that_is_no_power_unit_is_refused(fill_rate_unit): + with pytest.raises(ValueError, match="fill_rate_unit has to be one of W, kW, MW"): + _handler(fill_rate_unit=fill_rate_unit) @pytest.mark.asyncio @@ -609,15 +625,20 @@ async def test_an_actuator_s_fill_rate_is_posted_in_the_fill_rate_unit( ) ) - assert ( - _posted(frbc, fill_rate_sensor_id)["unit"] == frbc.fill_rate_unit() == "kWh/s" + assert _posted(frbc, fill_rate_sensor_id)["unit"] == frbc.fill_rate_unit() == "W" + # Halfway through the mode's fill rate, from kWh/s into W. + fill_rate = actuator.operation_modes[0].elements[0].fill_rate + halfway = (fill_rate.start_of_range + fill_rate.end_of_range) / 2 + assert halfway > 0 + assert _posted(frbc, fill_rate_sensor_id)["values"] == pytest.approx( + [halfway * 3.6e6] ) # The power sensor really does carry a power, and is untouched by this. assert _posted(frbc, 1)["unit"] == frbc.power_unit @pytest.mark.asyncio -async def test_a_storage_efficiency_already_recorded_for_the_window_is_not_recorded_again( +async def test_a_leakage_already_recorded_for_the_window_is_not_recorded_again( system_description, ): frbc = _frbc(system_description, with_sensors=True) @@ -635,8 +656,8 @@ async def test_a_storage_efficiency_already_recorded_for_the_window_is_not_recor ) frbc._leakage_behaviour_history[str(leakage.message_id)] = leakage start = datetime(2024, 1, 1, 12, 15, tzinfo=timezone.utc) - assert await frbc._record_storage_efficiency(start, 50.0) - assert await frbc._record_storage_efficiency(start, 50.0) + assert await frbc._planned_leakage(start, 50.0) is not None + assert await frbc._planned_leakage(start, 50.0) is not None posts = [ c for c in frbc._fm_client.post_sensor_data.await_args_list @@ -646,10 +667,10 @@ async def test_a_storage_efficiency_already_recorded_for_the_window_is_not_recor @pytest.mark.asyncio -async def test_a_storage_efficiency_that_could_not_be_recorded_is_not_planned_against( +async def test_a_leakage_that_could_not_be_recorded_is_planned_all_the_same( system_description, caplog ): - """And it is tried again next time, rather than remembered as recorded.""" + """It goes with the request, so the plan does not need it recorded; it is tried again next time, rather than remembered as recorded.""" frbc = _frbc(system_description, with_sensors=True) leakage = FRBCLeakageBehaviour( message_id=get_unique_id(), @@ -666,15 +687,15 @@ async def test_a_storage_efficiency_that_could_not_be_recorded_is_not_planned_ag frbc._leakage_behaviour_history[str(leakage.message_id)] = leakage frbc._fm_client.post_sensor_data.side_effect = ConnectionError("unreachable") start = datetime(2024, 1, 1, 12, 15, tzinfo=timezone.utc) - assert await frbc._record_storage_efficiency(start, 50.0) is False - assert frbc._storage_efficiency_recorded is None - assert "could not record" in caplog.text + assert set(await frbc._planned_leakage(start, 50.0)) == {36000.0} + assert frbc._leakage_recorded is None + assert "could not record" in caplog.text.lower() @pytest.mark.asyncio -async def test_a_storage_efficiency_whose_ingestion_is_unconfirmed_is_not_remembered( - system_description, -): +async def test_an_older_server_is_planned_without_leakage(system_description, caplog): + """Before FlexMeasures v0.23.0, a usage only comes from a sensor, so the leakage cannot go with the request.""" + caplog.set_level(logging.INFO) frbc = _frbc(system_description, with_sensors=True) leakage = FRBCLeakageBehaviour( message_id=get_unique_id(), @@ -689,13 +710,11 @@ async def test_a_storage_efficiency_whose_ingestion_is_unconfirmed_is_not_rememb ], ) frbc._leakage_behaviour_history[str(leakage.message_id)] = leakage - frbc._fm_client.post_sensor_data.side_effect = JobTimeoutError("still pending") + frbc._fm_client.server_version_at_least = AsyncMock(return_value=False) start = datetime(2024, 1, 1, 12, 15, tzinfo=timezone.utc) - assert await frbc._record_storage_efficiency(start, 50.0) is False - assert frbc._storage_efficiency_recorded is None - assert ( - frbc._fm_client.post_sensor_data.await_args.kwargs["confirm_ingestion"] is True - ) + assert await frbc._planned_leakage(start, 50.0) is None + assert _posted_sensor_ids(frbc) == [] + assert "Planning without leakage" in caplog.text @pytest.mark.asyncio diff --git a/tests/s2/test_frbc_element_selection.py b/tests/s2/test_frbc_element_selection.py index 992df47e..87ef5035 100644 --- a/tests/s2/test_frbc_element_selection.py +++ b/tests/s2/test_frbc_element_selection.py @@ -522,78 +522,30 @@ def test_the_storage_efficiency_at_a_fill_level_is_that_elements(): @pytest.mark.asyncio -async def test_the_handler_records_leakage_at_the_latest_fill_level(): +@pytest.mark.parametrize( + "fill_level, rate", + [ + (20.0, LEAKS_LITTLE.leakage_rate), + (50.0, LEAKS_MOST.leakage_rate), + (70.0, LEAKS_SOME.leakage_rate), + ], +) +async def test_the_handler_plans_and_records_the_leakage_of_the_range_it_starts_in( + fill_level, rate +): + """S2 states a rate per fill-level range, and the plan starts in one of them; not the leakiest, nor the first listed.""" leakage = _leakage(LEAKS_LITTLE, LEAKS_MOST, LEAKS_SOME) frbc = _frbc(_system_description(_mode(COLD, WARM)), leakage_behaviour_sensor_id=9) frbc._leakage_behaviour_history[str(leakage.message_id)] = leakage - await frbc._record_storage_efficiency( - datetime(2024, 1, 2, tzinfo=timezone.utc), 20.0 - ) - posted = [ - call.kwargs["values"] - for call in frbc._fm_client.post_sensor_data.await_args_list - if call.kwargs.get("sensor_id") == 9 - ] - assert posted == [ - [ - leakage_behaviour_to_storage_efficiency( - leakage, - resolution=timedelta(minutes=15), - fill_level=20.0, - share_of=max(20.0, _middle_of_the_range(frbc)), - ) - ] - ] - -def _middle_of_the_range(frbc) -> float: - storage_range = list(frbc._system_description_history.values())[ - -1 - ].storage.fill_level_range - return (storage_range.start_of_range + storage_range.end_of_range) / 2 - - -@pytest.mark.asyncio -async def test_a_near_empty_start_takes_the_loss_as_a_share_of_the_middle_of_the_range(): - """A share of a near-empty start would leave almost nothing per step, and overstate the loss once the plan fills up.""" - leakage = _leakage(LEAKS_LITTLE, LEAKS_MOST, LEAKS_SOME) - frbc = _frbc(_system_description(_mode(COLD, WARM)), leakage_behaviour_sensor_id=9) - frbc._leakage_behaviour_history[str(leakage.message_id)] = leakage - start = datetime(2024, 1, 2, tzinfo=timezone.utc) - await frbc._record_storage_efficiency(start, 0.01) - (posted,) = [ - call.kwargs["values"][0] - for call in frbc._fm_client.post_sensor_data.await_args_list - if call.kwargs.get("sensor_id") == 9 - ] - middle = _middle_of_the_range(frbc) - assert middle > 0.01 - assert posted == leakage_behaviour_to_storage_efficiency( - leakage, - resolution=timedelta(minutes=15), - fill_level=0.01, - share_of=middle, + planned = await frbc._planned_leakage( + datetime(2024, 1, 2, tzinfo=timezone.utc), fill_level ) - assert posted > 99 - -@pytest.mark.asyncio -async def test_a_start_above_the_middle_takes_the_loss_as_a_share_of_the_start(): - leakage = _leakage(LEAKS_LITTLE, LEAKS_MOST, LEAKS_SOME) - frbc = _frbc(_system_description(_mode(COLD, WARM)), leakage_behaviour_sensor_id=9) - frbc._leakage_behaviour_history[str(leakage.message_id)] = leakage - storage_range = list(frbc._system_description_history.values())[ - -1 - ].storage.fill_level_range - high = (storage_range.end_of_range + _middle_of_the_range(frbc)) / 2 - await frbc._record_storage_efficiency( - datetime(2024, 1, 2, tzinfo=timezone.utc), high - ) - (posted,) = [ - call.kwargs["values"][0] + assert set(planned) == {rate * 3.6e6} # in W + posted = [ + call.kwargs["values"] for call in frbc._fm_client.post_sensor_data.await_args_list if call.kwargs.get("sensor_id") == 9 ] - assert posted == leakage_behaviour_to_storage_efficiency( - leakage, resolution=timedelta(minutes=15), fill_level=high, share_of=high - ) + assert posted == [[rate * 3.6e6]] diff --git a/tests/s2/test_frbc_soft_band.py b/tests/s2/test_frbc_soft_band.py index f160d3cc..b302cd79 100644 --- a/tests/s2/test_frbc_soft_band.py +++ b/tests/s2/test_frbc_soft_band.py @@ -182,7 +182,15 @@ async def test_the_usage_and_the_leakage_a_resource_manager_declares_are_planned system_description, ): """A storage that is drawn from and leaks is not planned for by its prices alone.""" - from s2python.frbc import FRBCLeakageBehaviour, FRBCLeakageBehaviourElement + from datetime import timedelta + + from s2python.common import Duration + from s2python.frbc import ( + FRBCLeakageBehaviour, + FRBCLeakageBehaviourElement, + FRBCUsageForecast, + FRBCUsageForecastElement, + ) frbc = _frbc(system_description) leakage = FRBCLeakageBehaviour( @@ -196,11 +204,26 @@ async def test_the_usage_and_the_leakage_a_resource_manager_declares_are_planned ], ) frbc._leakage_behaviour_history[str(leakage.message_id)] = leakage + frbc._usage_forecast_history["f"] = FRBCUsageForecast( + message_id=get_unique_id(), + start_time=frbc.schedule_start(system_description.valid_from), + elements=[ + FRBCUsageForecastElement( + duration=Duration.from_timedelta(timedelta(minutes=15)), + usage_rate_expected=0.01, + ) + ], + ) flex_model, _ = await _asked_for(frbc, system_description) - assert flex_model["soc-usage"] == [{"sensor": USAGE_FORECAST_SENSOR_ID}] - assert flex_model["storage-efficiency"] == {"sensor": LEAKAGE_SENSOR_ID} + # The forecast goes with the request, with the leakage added to it as a rate over the whole window, + # rather than as a share of the stock the storage keeps. + planned = [float(step["value"].split()[0]) for step in flex_model["soc-usage"][0]] + assert planned[0] == pytest.approx((0.01 + 0.001) * 3.6e6) # in W + assert planned[1:] == pytest.approx([0.001 * 3.6e6] * (len(planned) - 1)) + assert len(planned) == frbc._schedule_duration / frbc._resolution + assert "storage-efficiency" not in flex_model @pytest.mark.asyncio @@ -533,15 +556,16 @@ def test_bounds_that_run_backwards_are_refused(): @pytest.mark.asyncio -async def test_a_negative_usage_forecast_is_recorded_as_no_usage_and_said_so( +async def test_a_negative_usage_forecast_is_recorded_as_stated_and_said_so( system_description, caplog ): - """A forced gain FlexMeasures cannot relax made every schedule for the window infeasible.""" + """S2 allows a negative usage, so it is recorded; whether a plan can absorb it is decided when a schedule is requested.""" from datetime import timedelta from s2python.common import Duration from s2python.frbc import FRBCUsageForecast, FRBCUsageForecastElement + caplog.set_level("INFO") frbc = _frbc(system_description) forecast = FRBCUsageForecast( message_id=get_unique_id(), @@ -556,8 +580,7 @@ async def test_a_negative_usage_forecast_is_recorded_as_no_usage_and_said_so( ) await frbc.send_usage_forecast(forecast) posted = frbc._fm_client.post_sensor_data.await_args.kwargs - assert min(posted["values"]) == 0.0 - assert posted["values"][0] == pytest.approx(0.01) + assert posted["values"] == pytest.approx([36000.0, -72000.0, 0.0, -18000.0]) # in W assert "negative usage on 2 of 4 steps" in caplog.text diff --git a/tests/s2/test_frbc_unsupported_is_said.py b/tests/s2/test_frbc_unsupported_is_said.py index ad71e746..b1b4c164 100644 --- a/tests/s2/test_frbc_unsupported_is_said.py +++ b/tests/s2/test_frbc_unsupported_is_said.py @@ -4,6 +4,7 @@ import logging from datetime import datetime, timezone +from unittest.mock import AsyncMock import pytest from s2python.common import Duration, ReceptionStatusValues, Timer @@ -111,19 +112,41 @@ async def _flex_model(frbc, system_description) -> dict: async def test_usage_is_planned_with_only_where_the_resource_manager_says_it_sends_it( provided, ): - """Named regardless, the scheduler read a sensor nothing was ever recorded on.""" + """Named regardless, the scheduler read a sensor nothing was ever recorded on. + + Only a server that takes no usage with the request reads the sensor (before FlexMeasures v0.23.0). + """ system_description = _heater() storage = system_description.storage.model_copy( update={"provides_usage_forecast": provided} ) system_description = system_description.model_copy(update={"storage": storage}) frbc = _frbc(system_description, usage_forecast_sensor_id=7) + frbc._fm_client.server_version_at_least = AsyncMock(return_value=False) flex_model = await _flex_model(frbc, system_description) assert ("soc-usage" in flex_model) is provided +@pytest.mark.asyncio +async def test_no_usage_is_planned_before_a_forecast_arrives(caplog): + """The sensor may hold negative usage recorded as stated, which a plan without a correction cannot absorb.""" + system_description = _heater() + storage = system_description.storage.model_copy( + update={"provides_usage_forecast": True} + ) + system_description = system_description.model_copy(update={"storage": storage}) + frbc = _frbc(system_description, usage_forecast_sensor_id=7) + caplog.set_level("INFO") + frbc._fm_client.server_version_at_least = AsyncMock(return_value=True) + + flex_model = await _flex_model(frbc, system_description) + + assert "soc-usage" not in flex_model + assert "no FRBC.UsageForecast has arrived" in caplog.text + + @pytest.mark.asyncio async def test_a_usage_forecast_sent_despite_the_flag_is_planned_with(): """What a resource manager sends says more than the flag it set once.""" diff --git a/tests/s2/test_frbc_usage_correction.py b/tests/s2/test_frbc_usage_correction.py new file mode 100644 index 00000000..841ec655 --- /dev/null +++ b/tests/s2/test_frbc_usage_correction.py @@ -0,0 +1,987 @@ +"""A usage forecast that no plan can absorb is corrected towards zero, instead of making every plan infeasible. + +S2 lets a resource manager expect any usage, negative included, but says nothing about what happens when the storage cannot take it. +FlexMeasures plans a forecast usage as forced, so a gain at a full storage, or a draw at an empty one that the actuator cannot keep up with, +left no feasible plan at all, and the storage without an instruction. +The handler now plans a correction: a second device on the storage's stock, which may only move each step's usage towards zero, +priced so that it is used only where nothing physical can keep the plan within the hard bounds. +""" + +from __future__ import annotations + +import logging +from datetime import timedelta +from unittest.mock import AsyncMock, patch + +import pandas as pd +import pytest +from s2python.common import ( + Commodity, + CommodityQuantity, + Duration, + NumberRange, + PowerRange, +) +from s2python.frbc import ( + FRBCActuatorDescription, + FRBCOperationMode, + FRBCOperationModeElement, + FRBCStorageDescription, + FRBCStorageStatus, + FRBCSystemDescription, + FRBCUsageForecast, + FRBCUsageForecastElement, +) + +from flexmeasures_client.client import FlexMeasuresClient +from flexmeasures_client.s2.control_types.FRBC.frbc_simple import ( + USAGE_CORRECTION_COMMODITY, + FRBCSimple, + _usage_correction_limits, + _usage_correction_price, +) +from flexmeasures_client.s2.utils import get_unique_id + +USAGE_FORECAST_SENSOR_ID = 12 +CORRECTION_SENSOR_ID = 20 +ASSET_ID = 7 +SOC_SENSOR_ID = 3 +PRICE_SENSOR_ID = 2 +START = pd.Timestamp("2024-01-01T00:00:00+00:00") +STEP = timedelta(minutes=15) + + +@pytest.fixture +def system_description() -> FRBCSystemDescription: + """A storage declaring a fill-level range of [20, 80], with one operation mode from 0 to 1000 W.""" + operation_mode = FRBCOperationMode( + id=get_unique_id(), + diagnostic_label="charge", + elements=[ + FRBCOperationModeElement( + fill_level_range=NumberRange(start_of_range=20, end_of_range=80), + fill_rate=NumberRange(start_of_range=0, end_of_range=1), + power_ranges=[ + PowerRange( + start_of_range=0, + end_of_range=1000, + commodity_quantity=CommodityQuantity.ELECTRIC_POWER_L1, + ) + ], + ) + ], + abnormal_condition_only=False, + ) + return FRBCSystemDescription( + message_id=get_unique_id(), + valid_from="2024-01-01T00:00:00+00:00", + actuators=[ + FRBCActuatorDescription( + id=get_unique_id(), + supported_commodities=[Commodity.ELECTRICITY], + operation_modes=[operation_mode], + transitions=[], + timers=[], + ) + ], + storage=FRBCStorageDescription( + provides_leakage_behaviour=True, + provides_fill_level_target_profile=True, + provides_usage_forecast=True, + fill_level_range=NumberRange(start_of_range=20, end_of_range=80), + ), + ) + + +def _frbc(system_description, **kwargs) -> FRBCSimple: + """A handler that has had a system description and a storage status, ready to plan.""" + frbc = FRBCSimple( + power_sensor_id=1, + price_sensor_id=PRICE_SENSOR_ID, + soc_sensor_id=SOC_SENSOR_ID, + rm_discharge_sensor_id=4, + **{"usage_forecast_sensor_id": USAGE_FORECAST_SENSOR_ID, **kwargs}, + ) + frbc._fm_client = AsyncMock(FlexMeasuresClient) + frbc._logger = logging.getLogger(__name__) + frbc._system_description_history = { + system_description.message_id: system_description + } + frbc._storage_status_history[get_unique_id()] = FRBCStorageStatus( + message_id=get_unique_id(), present_fill_level=30.0 + ) + frbc._fm_client.trigger_and_get_schedule = AsyncMock( + return_value={ + "values": [0.0] * 48, + "start": "2024-01-01T00:00:00+00:00", + "duration": "PT12H", + } + ) + frbc.send_instruction_batch = AsyncMock(return_value=0) + return frbc + + +def _forecast(*rates: float) -> FRBCUsageForecast: + return FRBCUsageForecast( + message_id=get_unique_id(), + start_time=START.isoformat(), + elements=[ + FRBCUsageForecastElement( + duration=Duration.from_timedelta(STEP), usage_rate_expected=rate + ) + for rate in rates + ], + ) + + +def _correcting( + system_description, + *, + prices=(100.0, -40.0, 450.0), + soc_resolution="PT0M", + stored=None, + **kwargs, +): + """A handler with a correction sensor, against a server that can plan one, with readable prices.""" + frbc = _frbc( + system_description, usage_correction_sensor_id=CORRECTION_SENSOR_ID, **kwargs + ) + frbc._fm_client.server_version_at_least = AsyncMock(return_value=True) + + async def get_sensor(sensor_id, **_): + if sensor_id == SOC_SENSOR_ID: + return {"event_resolution": soc_resolution, "unit": "kWh"} + return { + "event_resolution": "PT15M", + "unit": "EUR/kWh", + "generic_asset_id": ASSET_ID, + } + + frbc._fm_client.get_sensor = AsyncMock(side_effect=get_sensor) + frbc._fm_client.get_asset = AsyncMock( + return_value={"id": ASSET_ID, "flex_context": stored or {}} + ) + frbc._fm_client.get_sensor_data = AsyncMock(return_value={"values": list(prices)}) + # Pin the window, so that it starts where the forecast does. + frbc.schedule_start = lambda valid_from: START.to_pydatetime() + return frbc + + +async def _request(frbc, system_description): + await frbc.trigger_schedule(system_description.message_id) + kwargs = frbc._fm_client.trigger_and_get_schedule.await_args.kwargs + return kwargs["flex_model"], kwargs["flex_context"] + + +def test_the_correction_may_only_move_each_step_towards_zero(): + usage = pd.Series( + [2.0, -1.0, 0.0, -1.0], index=pd.date_range(START, periods=4, freq=STEP) + ) + + up, down = _usage_correction_limits(usage, START, 6, STEP, "kW") + + # Where usage is expected, it may give that much back; where a gain is, it may take that much away; nothing else. + assert [(s["start"], pd.Timedelta(s["duration"]), s["value"]) for s in up] == [ + (START.isoformat(), STEP, "2.0 kW"), + ((START + STEP).isoformat(), 5 * STEP, "0.0 kW"), + ] + assert [s["value"] for s in down] == [ + "0.0 kW", + "1.0 kW", + "0.0 kW", + "1.0 kW", + "0.0 kW", + ] + assert ( + pd.Timedelta(down[-1]["duration"]) == 2 * STEP + ), "steps the forecast does not cover allow no correction" + + +def test_the_correction_price_bounds_every_worth_of_a_unit_of_stock(): + # The probe's case: market prices up to 450, band breaches at ten times that, a day of quarter hours. + price = _usage_correction_price( + max_abs_price=450, + soc_breach_price=4500, + charging_efficiency=1, + storage_efficiency=1, + steps=96, + step_hours=0.25, + ) + assert price == pytest.approx(2 * (450 + 2 * 4500 * (4 + 96))) + # A unit of stock is worth more where buying it loses more on the way in, or leaks while it waits. + lossy = _usage_correction_price(450, 4500, 0.8, 0.99, 96, 0.25) + assert lossy - price == pytest.approx(2 * 450 * (1 / (0.8 * 0.99**96) - 1)) + + +@pytest.mark.asyncio +async def test_a_schedule_is_asked_for_with_a_correction_on_the_storages_stock( + system_description, +): + frbc = _correcting(system_description) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, flex_context = await _request(frbc, system_description) + + actuator, stock, correction = flex_model + assert actuator["sensor"] == 1 and actuator["commodity"] == "electricity" + assert "consumption-capacity" in actuator + assert "soc-at-start" not in actuator and "soc-usage" not in actuator + assert ( + stock["state-of-charge"] + == {"sensor": SOC_SENSOR_ID} + == actuator["state-of-charge"] + ) + assert "sensor" not in stock and "soc-minima" in stock + # The usage planned is the forecast the limits come from, not whatever the sensor holds. + assert [s["value"] for s in stock["soc-usage"][0]] == [ + "36000.0 W", + "-72000.0 W", + ] + assert correction["sensor"] == CORRECTION_SENSOR_ID + assert correction["commodity"] == USAGE_CORRECTION_COMMODITY + assert correction["state-of-charge"] == {"sensor": SOC_SENSOR_ID} + # Usage of 0.01 kWh/s (36000 W), then a gain of 0.02 kWh/s (72000 W): each may only be moved towards zero. + assert correction["consumption-capacity"][0]["value"] == "36000.0 W" + assert correction["consumption-capacity"][1]["value"] == "0.0 W" + assert correction["production-capacity"][0]["value"] == "0.0 W" + assert correction["production-capacity"][1]["value"] == "72000.0 W" + + +@pytest.mark.asyncio +async def test_the_request_carries_a_coherent_set_of_prices(system_description): + frbc = _correcting(system_description) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + electricity, correction = flex_context["commodities"] + assert electricity["consumption-price"] == {"sensor": PRICE_SENSOR_ID} + assert "site-power-capacity" in electricity + # Breach prices relative to the highest absolute market price, 450. + assert flex_context["soc-minima-breach-price"] == "4500.0 EUR/MWh" + assert flex_context["soc-maxima-breach-price"] == "4500.0 EUR/MWh" + # FlexMeasures' default site breach price, which it fills in only without commodities, + # so the site keeps to the capacity it is steered with, softly as without a correction. + assert electricity["site-consumption-breach-price"] == "10000 EUR/kW" + assert electricity["site-production-breach-price"] == "10000 EUR/kW" + steps = 48 # the handler's default twelve hours + expected = _usage_correction_price(450.0, 4500.0, 1.0, 1.0, steps, 0.25) + assert correction == { + "commodity": USAGE_CORRECTION_COMMODITY, + "consumption-price": f"{expected} EUR/MWh", + "production-price": f"{-expected} EUR/MWh", + } + + +@pytest.mark.asyncio +async def test_a_hard_site_capacity_has_no_breach_price(system_description): + frbc = _correcting(system_description, relax_site_capacity=False) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + electricity, _ = flex_context["commodities"] + assert "site-consumption-breach-price" not in electricity + assert "site-production-breach-price" not in electricity + assert flex_context["relax-site-capacity-constraints"] is False + + +@pytest.mark.asyncio +async def test_a_stored_site_breach_price_is_kept_but_does_not_raise_the_correction_price( + system_description, +): + """A correction may be cheaper than a breach of the site's capacity, so a steering site keeps its breach price.""" + stored = { + "site-consumption-breach-price": "5000 EUR/kW", + "site-production-breach-price": "5000 EUR/kW", + } + frbc = _correcting(system_description, stored=stored) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + electricity, correction = flex_context["commodities"] + assert electricity["site-consumption-breach-price"] == "5000 EUR/kW" + assert electricity["site-production-breach-price"] == "5000 EUR/kW" + expected = _usage_correction_price(450.0, 4500.0, 1.0, 1.0, 48, 0.25) + assert correction["consumption-price"] == f"{expected} EUR/MWh" + + +@pytest.mark.asyncio +async def test_a_negative_usage_is_recorded_as_stated_where_it_is_corrected( + system_description, +): + frbc = _correcting(system_description) + + await frbc.send_usage_forecast(_forecast(0.01, -0.02)) + + posted = frbc._fm_client.post_sensor_data.await_args.kwargs + assert posted["values"] == pytest.approx([36000.0, -72000.0]) # in W + + +@pytest.mark.asyncio +async def test_the_schedule_is_read_back_with_the_usage_it_was_planned_with( + system_description, +): + frbc = _correcting(system_description) + frbc._usage_forecast_history["f"] = _forecast(-0.5) + + with patch( + "flexmeasures_client.s2.control_types.FRBC.frbc_simple.fm_schedule_to_instructions", + return_value=[], + ) as translate: + await frbc.trigger_schedule(system_description.message_id) + + read_back = translate.call_args.kwargs["usage_forecast"] + assert [e.usage_rate_expected for e in read_back.elements] == [-0.5] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "why, setup, said", + [ + ( + "no prices", + dict(prices=()), + "market prices over the window, which could not be read", + ), + ( + "a state-of-charge sensor with a resolution", + dict(soc_resolution="PT1M"), + "does not record instants", + ), + ], +) +async def test_without_what_a_correction_needs_negative_usage_is_planned_as_none( + system_description, caplog, why, setup, said +): + frbc = _correcting(system_description, **setup) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, flex_context = await _request(frbc, system_description) + + assert isinstance(flex_model, dict), why + assert "commodities" not in flex_context + assert [s["value"] for s in flex_model["soc-usage"][0]] == [ + "36000.0 W", + "0.0 W", + ] + assert said in caplog.text + + +@pytest.mark.asyncio +async def test_a_server_before_the_correction_plans_without_one(system_description): + frbc = _correcting(system_description) + frbc._fm_client.server_version_at_least = AsyncMock( + side_effect=lambda version: version != "1.1.0" + ) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, _ = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + # Planned without a correction: the forecast goes with the request, without its negative values. + assert [step["value"] for step in flex_model["soc-usage"][0]] == [ + "36000.0 W", + "0.0 W", + ] + + +@pytest.mark.asyncio +async def test_without_a_correction_sensor_the_request_carries_the_clipped_forecast( + system_description, +): + frbc = _frbc(system_description) + frbc._fm_client.server_version_at_least = AsyncMock(return_value=True) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, flex_context = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + # Planned without a correction: the forecast goes with the request, without its negative values. + assert [step["value"] for step in flex_model["soc-usage"][0]] == [ + "36000.0 W", + "0.0 W", + ] + assert ( + "commodities" not in flex_context + and "soc-minima-breach-price" not in flex_context + ) + + +@pytest.mark.asyncio +async def test_a_server_that_takes_no_usage_with_the_request_plans_with_what_is_recorded( + system_description, +): + """Before FlexMeasures v0.23.0, a flex model takes a usage only as a sensor, so the sensor holds no negative usage.""" + frbc = _correcting(system_description) + frbc._fm_client.server_version_at_least = AsyncMock(return_value=False) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, _ = await _request(frbc, system_description) + await frbc.send_usage_forecast(_forecast(0.01, -0.02)) + + assert flex_model["soc-usage"] == [{"sensor": USAGE_FORECAST_SENSOR_ID}] + assert min(frbc._fm_client.post_sensor_data.await_args.kwargs["values"]) == 0.0 + + +@pytest.mark.asyncio +async def test_a_server_whose_version_cannot_be_read_plans_with_what_is_recorded( + system_description, +): + """Not knowing whether the server takes a usage with the request, the handler treats it as one that does not.""" + frbc = _correcting(system_description) + + async def version_at_least(version): + if version in ("0.23.0", "1.1.0"): + raise ConnectionError("unreachable") + return True + + frbc._fm_client.server_version_at_least = AsyncMock(side_effect=version_at_least) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, _ = await _request(frbc, system_description) + await frbc.send_usage_forecast(_forecast(0.01, -0.02)) + + assert flex_model["soc-usage"] == [{"sensor": USAGE_FORECAST_SENSOR_ID}] + assert min(frbc._fm_client.post_sensor_data.await_args.kwargs["values"]) == 0.0 + + +@pytest.mark.asyncio +async def test_a_negative_usage_is_recorded_as_stated_without_a_correction( + system_description, +): + frbc = _frbc(system_description) + frbc._fm_client.server_version_at_least = AsyncMock(return_value=True) + + await frbc.send_usage_forecast(_forecast(0.01, -0.02)) + + assert frbc._fm_client.post_sensor_data.await_args.kwargs[ + "values" + ] == pytest.approx([36000.0, -72000.0]) + + +def test_a_breach_price_factor_of_zero_is_refused(system_description): + with pytest.raises(ValueError, match="greater than zero"): + _frbc(system_description, soc_breach_price_factor=0) + + +@pytest.mark.asyncio +async def test_the_correction_price_counts_the_charging_efficiency_and_not_the_leakage( + system_description, +): + """Buying a unit of stock through a lossy actuator makes it worth more. + + Leakage planned as a usage drains the same whatever the stock, so holding stock costs nothing extra. + """ + frbc = _correcting( + system_description, + charging_efficiency_sensor_id=30, + leakage_behaviour_sensor_id=31, + ) + frbc._usage_forecast_history["f"] = _forecast(0.01) + frbc._leakage_behaviour_history["l"] = _leakage(0.001) + frbc._record_charging_efficiency = AsyncMock(return_value=True) + frbc._planned_charging_efficiency = 0.8 + + _, flex_context = await _request(frbc, system_description) + + expected = _usage_correction_price(450.0, 4500.0, 0.8, 1.0, 48, 0.25) + assert flex_context["commodities"][1]["consumption-price"] == f"{expected} EUR/MWh" + + +@pytest.mark.asyncio +async def test_a_market_without_prices_still_prices_a_breach(system_description): + frbc = _correcting(system_description, prices=(0.0, 0.0)) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + assert flex_context["soc-minima-breach-price"] == "10.0 EUR/MWh" + + +@pytest.mark.asyncio +async def test_price_sensors_in_different_currencies_are_not_combined( + system_description, caplog +): + frbc = _correcting(system_description, production_price_sensor_id=5) + + async def get_sensor(sensor_id, **_): + if sensor_id == SOC_SENSOR_ID: + return {"event_resolution": "PT0M", "unit": "kWh"} + return { + "event_resolution": "PT15M", + "unit": "EUR/kWh" if sensor_id == PRICE_SENSOR_ID else "SEK/kWh", + } + + frbc._fm_client.get_sensor = AsyncMock(side_effect=get_sensor) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + flex_model, _ = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + assert "different currencies, EUR and SEK" in caplog.text + + +@pytest.mark.asyncio +async def test_the_charging_efficiency_recorded_for_a_schedule_is_the_one_it_is_priced_with( + system_description, +): + """The correction price uses the very value recorded for the window.""" + frbc = _correcting(system_description, charging_efficiency_sensor_id=30) + + assert await frbc._record_charging_efficiency( + system_description, START.to_pydatetime(), 30.0 + ) + charging_posted = frbc._fm_client.post_sensor_data.await_args.kwargs["values"][0] + + assert frbc._planned_charging_efficiency == charging_posted + + +def _leakage(rate: float): + from s2python.frbc import FRBCLeakageBehaviour, FRBCLeakageBehaviourElement + + return FRBCLeakageBehaviour( + message_id=get_unique_id(), + valid_from=START.isoformat(), + elements=[ + FRBCLeakageBehaviourElement( + fill_level_range=NumberRange(start_of_range=0, end_of_range=100), + leakage_rate=rate, + ) + ], + ) + + +def _values(series: list[dict]) -> list[float]: + return [float(step["value"].split()[0]) for step in series] + + +@pytest.mark.asyncio +async def test_the_correction_covers_the_forecast_and_the_leakage_together( + system_description, +): + """An empty storage cannot give its leakage either, so the correction may take that away too.""" + frbc = _correcting(system_description, leakage_behaviour_sensor_id=31) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + frbc._leakage_behaviour_history["l"] = _leakage(0.001) + + flex_model, _ = await _request(frbc, system_description) + + _, stock, correction = flex_model + usage = _values(stock["soc-usage"][0]) + assert usage[:3] == pytest.approx([39600.0, -68400.0, 3600.0]) # in W + assert len(usage) == 48 # the whole window, beyond the forecast + assert "storage-efficiency" not in stock + # Each step may be moved towards zero by as much as the usage and the leakage there add up to. + up = { + s["start"]: float(s["value"].split()[0]) + for s in correction["consumption-capacity"] + } + down = { + s["start"]: float(s["value"].split()[0]) + for s in correction["production-capacity"] + } + assert up[START.isoformat()] == pytest.approx(39600.0) + assert down[(START + STEP).isoformat()] == pytest.approx(68400.0) + assert up[(START + 2 * STEP).isoformat()] == pytest.approx(3600.0) + + +@pytest.mark.asyncio +async def test_leakage_alone_is_planned_and_corrected_without_a_forecast( + system_description, +): + frbc = _correcting(system_description, leakage_behaviour_sensor_id=31) + frbc._leakage_behaviour_history["l"] = _leakage(0.001) + + flex_model, _ = await _request(frbc, system_description) + + _, stock, correction = flex_model + assert _values(stock["soc-usage"][0]) == pytest.approx([3600.0] * 48) + assert correction["sensor"] == CORRECTION_SENSOR_ID + + +@pytest.mark.asyncio +async def test_leakage_is_planned_without_a_usage_forecast_sensor(system_description): + """Leakage is what the storage loses by itself, whether or not a usage forecast is planned with.""" + frbc = _correcting( + system_description, + usage_forecast_sensor_id=None, + leakage_behaviour_sensor_id=31, + ) + frbc._leakage_behaviour_history["l"] = _leakage(0.001) + + flex_model, _ = await _request(frbc, system_description) + + _, stock, _ = flex_model + assert _values(stock["soc-usage"][0]) == pytest.approx([3600.0] * 48) + + +@pytest.mark.asyncio +async def test_the_leakage_is_scaled_like_every_other_fill_rate(system_description): + frbc = _correcting( + system_description, leakage_behaviour_sensor_id=31, fill_level_scale=2 + ) + frbc._leakage_behaviour_history["l"] = _leakage(0.001) + + flex_model, _ = await _request(frbc, system_description) + + _, stock, _ = flex_model + assert _values(stock["soc-usage"][0]) == pytest.approx([7200.0] * 48) + + +@pytest.mark.asyncio +async def test_no_leakage_is_planned_without_a_sensor_to_record_it_on( + system_description, +): + """The leakage sensor is what opts a handler in, as it was for the storage efficiency.""" + frbc = _correcting(system_description) + frbc._usage_forecast_history["f"] = _forecast(0.01) + frbc._leakage_behaviour_history["l"] = _leakage(0.001) + + flex_model, _ = await _request(frbc, system_description) + + _, stock, _ = flex_model + assert _values(stock["soc-usage"][0]) == pytest.approx([36000.0]) + + +@pytest.mark.asyncio +async def test_without_a_correction_the_forecast_and_the_leakage_are_clipped_together( + system_description, +): + frbc = _frbc(system_description, leakage_behaviour_sensor_id=31) + frbc._fm_client.server_version_at_least = AsyncMock(return_value=True) + # Pin the window, so that it starts where the forecast does. + frbc.schedule_start = lambda valid_from: START.to_pydatetime() + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + frbc._leakage_behaviour_history["l"] = _leakage(0.001) + + flex_model, _ = await _request(frbc, system_description) + + usage = _values(flex_model["soc-usage"][0]) + assert usage[:3] == pytest.approx([39600.0, 0.0, 3600.0]) + assert "storage-efficiency" not in flex_model + + +@pytest.mark.asyncio +async def test_a_forecast_that_is_not_planned_with_is_not_corrected(system_description): + """Without a usage-forecast sensor, the plan has no usage to correct, even if a forecast arrived.""" + frbc = _correcting(system_description, usage_forecast_sensor_id=None) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, flex_context = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) and "soc-usage" not in flex_model + assert "commodities" not in flex_context + + +@pytest.mark.asyncio +async def test_a_server_whose_version_cannot_be_read_plans_without_a_correction( + system_description, +): + frbc = _correcting(system_description) + + async def version_at_least(version): + if version == "1.1.0": + raise ConnectionError("unreachable") + return True + + frbc._fm_client.server_version_at_least = AsyncMock(side_effect=version_at_least) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, _ = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + # Planned without a correction: the forecast goes with the request, without its negative values. + assert [step["value"] for step in flex_model["soc-usage"][0]] == [ + "36000.0 W", + "0.0 W", + ] + + +@pytest.mark.asyncio +async def test_prices_that_cannot_be_read_leave_the_plan_without_a_correction( + system_description, caplog +): + frbc = _correcting(system_description) + frbc._fm_client.get_sensor_data = AsyncMock( + side_effect=ConnectionError("unreachable") + ) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, _ = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + assert "Could not read the prices on sensor 2: unreachable" in caplog.text + + +@pytest.mark.asyncio +async def test_before_any_forecast_arrives_the_plan_goes_without_usage( + system_description, +): + """The sensor may hold negative usage recorded as stated before a restart, which no plan without a correction can absorb.""" + frbc = _correcting(system_description) + + flex_model, flex_context = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + assert "soc-usage" not in flex_model + assert "commodities" not in flex_context + + +@pytest.mark.parametrize( + "stored", + [ + {"inflexible-device-sensors": [827, 829]}, + { + "inflexible-consumption": [{"sensor": 827}], + "inflexible-production": [{"sensor": 829}], + }, + ], +) +@pytest.mark.asyncio +async def test_stored_inflexible_devices_are_counted_once(system_description, stored): + """FlexMeasures counts inflexible devices from the top level, where it merges the stored context, and from the electricity entry.""" + frbc = _correcting(system_description, stored=stored) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + electricity, _ = flex_context["commodities"] + for field, devices in stored.items(): + # Counted from the electricity entry, + assert electricity[field] == devices + # and named empty at the top level, which keeps FlexMeasures from merging the stored ones in there. + assert flex_context[field] == [] + + +@pytest.mark.asyncio +async def test_no_inflexible_devices_are_named_where_none_are_stored( + system_description, +): + frbc = _correcting( + system_description, stored={"site-consumption-capacity": "10 kW"} + ) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + for field in ( + "inflexible-device-sensors", + "inflexible-consumption", + "inflexible-production", + ): + assert field not in flex_context + + +@pytest.mark.asyncio +async def test_the_sites_stored_electricity_context_is_kept_and_its_prices_bounded( + system_description, +): + """FlexMeasures reads stored electricity fields only where a request names no electricity commodity, so they are carried over.""" + stored = { + "site-consumption-capacity": "10 kW", + "site-peak-consumption": "5 kW", + "site-peak-consumption-price": "260 EUR/MW", + "commitments": [ + { + "name": "contract", + "baseline": "0 kW", + "up-price": "0.5 EUR/kW", + "down-price": "0 EUR/kW", + } + ], + "soc-minima-breach-price": "1000 EUR/kWh", + "consumption-price": {"sensor": 99}, + "aggregate-power": {"sensor": 709}, + } + frbc = _correcting(system_description, stored=stored) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + electricity, correction = flex_context["commodities"] + assert electricity["site-consumption-capacity"] == "10 kW" + assert electricity["site-peak-consumption-price"] == "260 EUR/MW" + assert electricity["commitments"] == stored["commitments"] + # What the request sets itself wins, and top-level fields stay at the top. + assert electricity["consumption-price"] == {"sensor": PRICE_SENSOR_ID} + assert "soc-minima-breach-price" not in electricity + # A field a commodity entry does not accept stays where FlexMeasures merges it in: at the top level. + assert ( + "aggregate-power" not in electricity and "aggregate-power" not in flex_context + ) + assert flex_context["relax-capacity-constraints"] is False + expected = _usage_correction_price( + 450.0, 4500.0, 1.0, 1.0, 48, 0.25, other_prices_per_mw=260 + 500 + ) + assert correction["consumption-price"] == f"{expected} EUR/MWh" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "stored, said", + [ + ( + {"site-peak-consumption-price": "lots"}, + "stored site-peak-consumption-price is not a price per MW in EUR", + ), + ( + { + "commitments": [ + {"name": "c", "baseline": "0 kW", "up-price": "1 SEK/kW"} + ] + }, + "stored commitment 'c' has a up-price that is not a price per MW", + ), + ({"commodities": []}, "stores a flex context with commodities"), + ], +) +async def test_a_stored_price_that_cannot_be_bounded_leaves_the_plan_without_a_correction( + system_description, caplog, stored, said +): + frbc = _correcting(system_description, stored=stored) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, _ = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + assert said in caplog.text + + +def test_efficiencies_above_one_do_not_lower_the_bound(): + """A heat pump's COP, or a storage that gains by itself, makes neither buying nor holding stock cheaper than one for one.""" + one = _usage_correction_price(450, 4500, 1.0, 1.0, 48, 0.25) + assert _usage_correction_price(450, 4500, 4.0, 1.01, 48, 0.25) == one + + +def test_other_per_mw_prices_raise_the_bound_by_their_worth_per_mwh_in_one_step(): + """A peak or commitment price per MW costs at most that price over one step's length per MWh.""" + without = _usage_correction_price(450, 4500, 1.0, 1.0, 48, 0.25) + with_peak = _usage_correction_price( + 450, 4500, 1.0, 1.0, 48, 0.25, other_prices_per_mw=760 + ) + assert with_peak - without == pytest.approx(2 * 760 / 0.25) + + +@pytest.mark.parametrize( + "value, per_mw", + [ + ("260 EUR/MW", 260.0), + ("0.26 EUR/kW", 260.0), + ("-0.00026 EUR/W", 260.0), + ("260 SEK/MW", None), + ("260 EUR/MWh", None), + ("lots EUR/MW", None), + ("260", None), + ({"sensor": 5}, None), + ], +) +def test_only_a_fixed_price_per_mw_in_the_currency_can_be_bounded(value, per_mw): + from flexmeasures_client.s2.control_types.FRBC.frbc_simple import _price_per_mw + + assert _price_per_mw(value, "EUR") == per_mw + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "power_sensor, said", + [ + (ConnectionError("unreachable"), "could not be read: unreachable"), + ({"unit": "W"}, "could not be read: power sensor 1 states no asset"), + ], +) +async def test_a_stored_flex_context_that_cannot_be_read_leaves_the_plan_without_a_correction( + system_description, caplog, power_sensor, said +): + frbc = _correcting(system_description) + lookup = frbc._fm_client.get_sensor.side_effect + + async def get_sensor(sensor_id, **kwargs): + if sensor_id == 1: + if isinstance(power_sensor, Exception): + raise power_sensor + return power_sensor + return await lookup(sensor_id, **kwargs) + + frbc._fm_client.get_sensor = AsyncMock(side_effect=get_sensor) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, _ = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + assert said in caplog.text + + +@pytest.mark.asyncio +async def test_a_commitment_with_one_price_is_bounded_by_that_price(system_description): + stored = { + "commitments": [{"name": "c", "baseline": "0 kW", "up-price": "100 EUR/MW"}] + } + frbc = _correcting(system_description, stored=stored) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + expected = _usage_correction_price( + 450.0, 4500.0, 1.0, 1.0, 48, 0.25, other_prices_per_mw=100 + ) + assert flex_context["commodities"][1]["consumption-price"] == f"{expected} EUR/MWh" + + +@pytest.mark.asyncio +async def test_a_stored_price_on_a_sensor_is_bounded_by_its_highest_value_over_the_window( + system_description, +): + """A co-simulation's controller steers each home with peak prices on sensors, which the bound reads like the market prices.""" + stored = { + "site-peak-consumption-price": {"sensor": 705}, + "site-peak-production-price": {"sensor": 706}, + "site-consumption-capacity": {"sensor": 707}, + } + frbc = _correcting(system_description, stored=stored) + market = frbc._fm_client.get_sensor_data.return_value + + async def get_sensor_data(sensor_id, **kwargs): + if sensor_id == 705: + assert kwargs["unit"] == "EUR/MW" + return {"values": [None, 300.0, -800.0]} + if sensor_id == 706: + return {"values": []} + return market + + frbc._fm_client.get_sensor_data = AsyncMock(side_effect=get_sensor_data) + frbc._usage_forecast_history["f"] = _forecast(0.01) + + _, flex_context = await _request(frbc, system_description) + + electricity, correction = flex_context["commodities"] + assert electricity["site-peak-consumption-price"] == {"sensor": 705} + assert electricity["site-consumption-capacity"] == {"sensor": 707} + expected = _usage_correction_price( + 450.0, 4500.0, 1.0, 1.0, 48, 0.25, other_prices_per_mw=800.0 + ) + assert correction["consumption-price"] == f"{expected} EUR/MWh" + + +@pytest.mark.asyncio +async def test_a_stored_price_sensor_that_cannot_be_read_leaves_the_plan_without_a_correction( + system_description, caplog +): + frbc = _correcting( + system_description, stored={"site-peak-consumption-price": {"sensor": 705}} + ) + market = frbc._fm_client.get_sensor_data.return_value + + async def get_sensor_data(sensor_id, **kwargs): + if sensor_id == 705: + raise ConnectionError("unreachable") + return market + + frbc._fm_client.get_sensor_data = AsyncMock(side_effect=get_sensor_data) + frbc._usage_forecast_history["f"] = _forecast(0.01, -0.02) + + flex_model, _ = await _request(frbc, system_description) + + assert isinstance(flex_model, dict) + assert "Could not read the prices on sensor 705: unreachable" in caplog.text diff --git a/tests/s2/test_site.py b/tests/s2/test_site.py index 69676bb3..c38d5c25 100644 --- a/tests/s2/test_site.py +++ b/tests/s2/test_site.py @@ -14,6 +14,7 @@ from flexmeasures_client.s2.cem import CEM from flexmeasures_client.s2.site import ( FRBC_SITE_SENSORS, + USAGE_CORRECTION_SENSORS, FRBCSiteConfigurator, SiteSensor, ensure_site, @@ -84,6 +85,29 @@ async def test_an_existing_site_keeps_its_sensors_and_gets_the_missing_ones(): assert specs_update.args == (7, {"generic_asset_type_id": 6}) +def test_a_site_s_sensors_suit_a_residential_site(): + """Powers and rates of fill level in W, energy in kWh: in kWh/s, a heat buffer's leakage of some tens of watts would read about 0.00001.""" + units = { + name: spec.unit + for name, spec in {**FRBC_SITE_SENSORS, **USAGE_CORRECTION_SENSORS}.items() + } + assert units == { + "price": "EUR/kWh", + "production price": "EUR/kWh", + "power": "W", + "measured-power": "W", + "state of charge": "kWh", + "RM discharge": "dimensionless", + "fill rate": "W", + "soc minima": "kWh", + "soc maxima": "kWh", + "usage forecast": "W", + "leakage": "W", + "charging efficiency": "dimensionless", + "usage correction": "W", + } + + @pytest.mark.asyncio async def test_a_known_site_is_fetched_by_id_rather_than_found_in_a_listing(): """A resource manager reconnects every round of a co-simulation; a listing of every asset each time is slow.""" @@ -148,6 +172,8 @@ async def test_each_resource_manager_gets_a_site_named_after_it_and_a_handler_wi assert handler._price_sensor_id == made["price"]["id"] assert handler._production_price_sensor_id == made["production price"]["id"] assert handler._soc_sensor_id == made["state of charge"]["id"] + assert handler._usage_correction_sensor_id is None + assert "usage correction" not in made, "a correction is the deployment's to ask for" assert handler._valid_from_shift == timedelta(0) assert handler._schedule_duration == timedelta(hours=24) assert handler.send_revocations is True @@ -405,3 +431,15 @@ async def test_of_several_sites_with_one_name_the_newest_is_taken(): fm, "home A", asset_specs={}, sensors={}, timezone="UTC" ) assert asset["id"] == 12 + + +@pytest.mark.asyncio +async def test_a_site_gets_a_usage_correction_where_the_deployment_asks_for_one( + named_details, +): + fm = _flexmeasures() + configurator = FRBCSiteConfigurator(fm, asset_type_id=6, correct_usage=True) + site = await configurator(named_details) + made = fm.made + assert site.handler._usage_correction_sensor_id == made["usage correction"]["id"] + assert made["usage correction"]["attributes"] == {"consumption_is_positive": True}