diff --git a/docs/HEMS.rst b/docs/HEMS.rst
index ed24a9af..dc590519 100644
--- a/docs/HEMS.rst
+++ b/docs/HEMS.rst
@@ -22,6 +22,7 @@ Set up your environment
========================
To run the HEMS example (``HEMS_setup.py``), you'll need an environment in which both ``flexmeasures`` (the server) and ``flexmeasures-client`` is installed.
+The example requires FlexMeasures 1.0 or newer.
We use `uv `_ to manage dependencies. First, `install uv `_.
@@ -43,43 +44,170 @@ Or, alternatively, to install released versions into a fresh project:
Next steps:
- Follow instructions to set up flexmeasures (fresh database, etc).
-- Create an organisation account and an admin with:
+- Create an account and a user with the ``account-admin`` role. The quickest way
+ is FlexMeasures' toy account, which creates both in one step:
.. code-block:: bash
- flexmeasures add account
- flexmeasures add user --roles admin
+ flexmeasures add toy-account
-- Update the credentials in the ``examples/HEMS/const.py`` script accordingly.
+This is what ``examples/HEMS/const.py`` expects out of the box, so you can run
+the tutorial without editing it:
+
+.. code-block:: python
+
+ usr = "toy-user@flexmeasures.io"
+ pwd = "toy-password"
+
+The toy account also adds a few unrelated demo assets (``toy-building`` and its
+children). The tutorial ignores them and never deletes them.
+
+To use your own account instead, create it and give its user the
+``account-admin`` role:
+
+.. code-block:: bash
+
+ flexmeasures add account --name "HEMS tutorial"
+ flexmeasures add user --username hems-admin --email hems-admin@example.com \
+ --account 2 --roles account-admin
+
+Replace ``2`` with the account ID printed by the first command, and update
+``usr`` and ``pwd`` in ``examples/HEMS/const.py`` to match.
+
+Either way, the tutorial creates all assets and sensors in that one account. It
+does not create public assets, so a site-wide ``admin`` role is not required.
Run the tutorial script
=======================
-Before running the tutorial, make sure to update the connection details and other relevant settings (e.g., host, port, credentials) in examples/HEMS/const.py to match your local FlexMeasures setup.
+Before running the tutorial, update the connection details and other relevant
+settings in ``examples/HEMS/const.py``. Specify the host without an ``http://``
+or ``https://`` prefix, and set ``ssl = True`` when connecting over HTTPS. For
+example:
+
+.. code-block:: python
+
+ host = "127.0.0.1:5000"
+ ssl = False
+
+For an HTTPS deployment, use its host name and set ``ssl = True``.
+
+PV is inflexible by default: all available production is delivered and any
+surplus is treated as grid feed-in. Set ``PV_MODE = "curtailable"`` when the PV
+gateway can reduce production, for example at a site whose grid-production
+capacity is zero. In that mode the simulated gateway treats the PV schedule as
+a maximum setpoint; it can reduce available production but cannot increase it.
+Recreate an existing tutorial structure after changing this setting so its flex
+context and PV sensors match the selected mode.
+
+The PV chart distinguishes available production, delivered production,
+self-consumption, grid feed-in, and curtailment. The reporter calculates the
+latter two after realization as ``max(delivered PV - local load, 0)`` and
+``max(available PV - delivered PV, 0)``. Consequently, the daily
+self-consumption percentage uses delivered rather than merely available PV as
+its denominator.
+
Open three terminals. In the first terminal, run the server:
.. code-block:: bash
flexmeasures run
-In the second terminal, run a flexmeasures worker that listens to both the scheduling and forecasting queues:
+In the second terminal, run a flexmeasures worker that listens to the
+forecasting, scheduling, and ingestion queues:
.. code-block:: bash
- flexmeasures jobs run-worker --queue "forecasting|scheduling"
+ flexmeasures jobs run-worker --queue "forecasting|scheduling|ingestion"
Note: you can run the same command in two terminals (2 workers), to speed up the computation!
-In the third terminal, run the client script using the `/examples/HEMS` folder as the current working directory:
+In the third terminal, go to the HEMS directory:
.. code-block:: bash
cd examples/HEMS
- python3 HEMS_setup.py
.. note::
- Report generation (see :ref:`hems-tutorial` note above) shells out to a ``flexmeasures`` CLI process, which by default is expected on ``PATH`` and configured against the same database as the server. If your FlexMeasures server runs elsewhere (e.g. inside a Docker Compose service), point report generation at it instead via two environment variables:
+ For the time being, report generation (see :ref:`hems-tutorial` note above) shells out to a ``flexmeasures`` CLI process, which by default is expected on ``PATH`` and configured against the same database as the server. If your FlexMeasures server runs elsewhere (e.g. inside a Docker Compose service), point report generation at it instead via two environment variables:
+
+ - ``FLEXMEASURES_CLI_CMD``: the command used to invoke the CLI
+ - ``FLEXMEASURES_CLI_CONFIG_DIR``: the directory the CLI process sees the ``examples/HEMS/configs/`` files at, if different from their local path
+
+ Here are steps if you use FlexMeasures' docker-compose:
+ - ``export FLEXMEASURES_CLI_CMD="docker compose -f full/path/to/docker-compose.yml exec -T server flexmeasures"``.
+ - Add this mount in docker-compose.yml under server.volumes, and restart it: ``- /full/path/to/flexmeasures-client/examples/HEMS/configs:/app/hems-configs:ro``
+ - ``export FLEXMEASURES_CLI_CONFIG_DIR="/app/hems-configs"``
+
+Another caveat is rate-limiting. Since v1.0, FlexMeasures only allows a limited number of schedule and forecasts per 5 minute interval.
+Either give your account a generous plan (see the docs), or simply set ``FLEXMEASURES_MODE="play"`` and restart the server.
+If you use docker-compose, you could do that like this:
+
+Add ``FLEXMEASURES_MODE = "play"`` to the existing
+``/full/path/to/flexmeasures-instance/flexmeasures.cfg`` file without replacing
+its other settings, then restart the server container:
+
+.. code-block:: bash
+
+ docker compose restart name-of-flexmeasures-server-container
+
+Now run the client script using the `/examples/HEMS` folder as the current working directory:
+
+.. code-block:: bash
+
+ python3 HEMS_setup.py
+
+Rerunning or resuming the tutorial
+==================================
+
+The setup script records completed phases in a namespaced attribute on the
+community asset. If a tracked community already exists, the script shows which
+phases are complete and offers four choices:
+
+- ``y`` recreates the HEMS assets. This deletes their sensors, IDs, and data,
+ including the HEMS energy market and weather station, before creating
+ replacements with new IDs. You must confirm this by typing ``RECREATE``.
+- ``w`` preserves the asset and sensor structure and IDs, but permanently
+ deletes all HEMS time-series data before restarting at data upload. This
+ includes uploads, forecasts, schedules, simulated measurements, and report
+ outputs. You must confirm this by typing ``WIPE``.
+- ``n`` (the default) preserves everything and resumes at the first unfinished
+ phase. Completed phases are skipped.
+- ``q`` exits without changing the setup.
+
+If an earlier data wipe was interrupted, normal resume is disabled because
+some sensors may already be empty while others still contain old data. The
+script instead offers to continue the wipe, recreate the setup, or exit.
+
+If asset creation was interrupted before the setup marker was saved, the
+script offers to complete missing assets and sensors while preserving existing
+IDs, recreate the setup, or exit. For an older setup with different site names,
+it also offers to keep those names or rename the sites to the names configured
+in ``const.py``.
+
+The workflow marker stores the sensor IDs in the HEMS structure when the marker
+is created, so a data wipe remains limited to that recorded set. If an existing
+setup predates workflow markers, safe resume is unavailable because its
+completed phases are unknown. Choose the offered repair option to preserve IDs
+and complete missing structure, or choose recreation to replace the setup.
+
+
+Delete the tutorial assets and data
+===================================
+
+To remove the HEMS setup from the configured account, run the cleanup script
+from the same directory:
+
+.. code-block:: bash
+
+ python3 HEMS_cleanup.py
+
+The script shows the matching top-level assets and asks for confirmation. It
+deletes the community asset, the energy market, and the weather station. Asset
+deletion also removes their child assets, sensors, and time-series data.
- - ``FLEXMEASURES_CLI_CMD``: the command used to invoke the CLI, e.g. ``"docker compose exec -T server flexmeasures"``.
- - ``FLEXMEASURES_CLI_CONFIG_DIR``: the directory the CLI process sees the ``examples/HEMS/configs/`` files at, if different from their local path (e.g. because that directory is bind-mounted into a container at a different path).
+.. warning::
+ Deletion is permanent. The energy market and weather station are separate
+ top-level assets; do not continue if other systems in the account share them.
+ The configured user needs the ``account-admin`` role to delete assets.
diff --git a/examples/HEMS/HEMS_cleanup.py b/examples/HEMS/HEMS_cleanup.py
new file mode 100644
index 00000000..5543db7d
--- /dev/null
+++ b/examples/HEMS/HEMS_cleanup.py
@@ -0,0 +1,28 @@
+"""Delete the assets and data created by the HEMS tutorial."""
+
+import asyncio
+
+from const import COMMUNITY_NAME, host, pwd, ssl, usr
+from utils.asset_utils import delete_hems_assets
+
+from flexmeasures_client import FlexMeasuresClient
+
+
+async def main() -> None:
+ client = FlexMeasuresClient(email=usr, password=pwd, host=host, ssl=ssl)
+ try:
+ account = await client.get_account()
+ if not account:
+ raise RuntimeError("No account found for the configured user.")
+ print(f"Connected to account: {account['name']} (ID: {account['id']})")
+ await delete_hems_assets(
+ client=client,
+ account_id=account["id"],
+ community_name=COMMUNITY_NAME,
+ )
+ finally:
+ await client.close()
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/HEMS/HEMS_setup.py b/examples/HEMS/HEMS_setup.py
index 7a1bcf07..906bd6af 100644
--- a/examples/HEMS/HEMS_setup.py
+++ b/examples/HEMS/HEMS_setup.py
@@ -8,15 +8,241 @@
from typing import Callable
from assets_setup import create_community_asset
-from const import COMMUNITY_NAME, SITE_NAMES, host, pwd, usr
+from const import COMMUNITY_NAME, PV_MODE, SITE_NAMES, host, pwd, ssl, usr
from forecasting import generate_forecasts
from reporters import create_reports
from scheduling import just_continue, run_scheduling_simulation
-from utils.asset_utils import cleanup_existing_assets, upload_data_for_first_two_weeks
+from utils.asset_utils import delete_hems_assets, upload_data_for_first_two_weeks
+from utils.workflow_utils import (
+ DATA_UPLOAD_PHASE,
+ FORECASTING_PHASE,
+ PHASE_LABELS,
+ REPORTING_PHASE,
+ SCHEDULING_PHASE,
+ WORKFLOW_VERSION,
+ get_site_assets,
+ get_workflow_state,
+ initialize_workflow_state,
+ mark_phase_complete,
+ phase_is_complete,
+ rename_site_assets,
+ state_needs_upgrade,
+ upgrade_workflow_state,
+ wipe_hems_sensor_data,
+)
from flexmeasures_client import FlexMeasuresClient
+def print_workflow_summary(state: dict) -> None:
+ """Show which phases resume mode will skip and run."""
+ completed = set(state["completed-phases"])
+ print("\nCompleted phases:")
+ for phase, label in PHASE_LABELS.items():
+ if phase in completed:
+ print(f"- {label}")
+ print("Still to run:")
+ remaining = [
+ label for phase, label in PHASE_LABELS.items() if phase not in completed
+ ]
+ if remaining:
+ for label in remaining:
+ print(f"- {label}")
+ else:
+ print("- Nothing; the tutorial is already complete.")
+
+
+def prompt_for_existing_setup(account: dict, community_name: str, state: dict) -> str:
+ """Ask whether to recreate, wipe data, or resume an existing setup."""
+ print(
+ f"Asset '{community_name}' already exists in account "
+ f"'{account['name']}' (ID: {account['id']})."
+ )
+ print_workflow_summary(state)
+ while True:
+ answer = (
+ input(
+ "\nChoose how to continue:\n"
+ " [y] Recreate assets — delete assets, sensors, IDs, and data "
+ "(requires typing RECREATE).\n"
+ " [w] Wipe data — preserve the asset and sensor structure and IDs, "
+ "delete all HEMS time-series data, then restart at data upload "
+ "(requires typing WIPE).\n"
+ " [n] Resume — preserve everything and continue from the first "
+ "unfinished phase.\n"
+ " [q] Exit without making changes.\n"
+ "Choice [y/w/N/q]: "
+ )
+ .strip()
+ .lower()
+ )
+ if answer in {"y", "yes"}:
+ return "recreate"
+ if answer in {"w", "wipe"}:
+ return "wipe"
+ if answer in {"q", "quit", "exit"}:
+ return "exit"
+ if answer in {"", "n", "no"}:
+ return "resume"
+ print("Please choose 'y', 'w', 'n', or 'q'.")
+
+
+def prompt_for_interrupted_wipe() -> str:
+ """Require an explicit recovery choice after a partial data wipe."""
+ print(
+ "A previous data wipe was interrupted. Some sensor data may already "
+ "be deleted, so normal resume is not safe."
+ )
+ while True:
+ answer = (
+ input(
+ "\nChoose how to recover:\n"
+ " [c] Continue the interrupted wipe and preserve all IDs.\n"
+ " [y] Recreate assets, sensors, IDs, and data "
+ "(requires typing RECREATE).\n"
+ " [q] Exit without deleting anything else.\n"
+ "Choice [c/y/Q]: "
+ )
+ .strip()
+ .lower()
+ )
+ if answer in {"c", "continue"}:
+ return "continue-wipe"
+ if answer in {"y", "yes", "recreate"}:
+ return "recreate"
+ if answer in {"", "q", "quit", "exit", "n", "no"}:
+ return "exit"
+ print("Please choose 'c', 'y', or 'q'.")
+
+
+def prompt_for_untracked_setup(
+ existing_site_names: list[str], configured_site_names: list[str]
+) -> str:
+ """Choose how to recover a setup whose asset phase was not recorded."""
+ legacy_names = any(
+ name not in configured_site_names for name in existing_site_names
+ )
+ if legacy_names:
+ print(
+ "This setup uses different site names than the current tutorial:\n"
+ f"- Existing: {existing_site_names}\n"
+ f"- Configured: {configured_site_names}"
+ )
+ while True:
+ answer = (
+ input(
+ "\nChoose how to recover the asset structure:\n"
+ " [k] Keep the existing site names and complete missing items.\n"
+ " [m] Rename existing sites to the configured names and "
+ "complete missing items.\n"
+ " [y] Recreate the complete setup with new IDs "
+ "(requires typing RECREATE).\n"
+ " [q] Exit without making changes.\n"
+ "Choice [k/m/y/Q]: "
+ )
+ .strip()
+ .lower()
+ )
+ if answer in {"k", "keep"}:
+ return "keep-names"
+ if answer in {"m", "migrate", "rename"}:
+ return "rename-sites"
+ if answer in {"y", "yes", "recreate"}:
+ return "recreate"
+ if answer in {"", "q", "quit", "exit"}:
+ return "exit"
+ print("Please choose 'k', 'm', 'y', or 'q'.")
+
+ while True:
+ answer = (
+ input(
+ "The existing setup has no completed asset-setup marker and may "
+ "be incomplete.\n"
+ " [c] Complete missing assets and sensors, preserving existing IDs.\n"
+ " [y] Recreate the complete setup with new IDs "
+ "(requires typing RECREATE).\n"
+ " [q] Exit without making changes.\n"
+ "Choice [c/y/Q]: "
+ )
+ .strip()
+ .lower()
+ )
+ if answer in {"c", "continue", "complete", "repair"}:
+ return "repair"
+ if answer in {"y", "yes", "recreate"}:
+ return "recreate"
+ if answer in {"", "q", "quit", "exit"}:
+ return "exit"
+ print("Please choose 'c', 'y', or 'q'.")
+
+
+def confirm_recreation(account: dict, community_name: str) -> bool:
+ """Require explicit confirmation before replacing the HEMS structure."""
+ answer = input(
+ f"This permanently deletes the HEMS setup '{community_name}' from account "
+ f"'{account['name']}' (ID: {account['id']}), including its assets, sensors, "
+ "IDs, and all time-series data. The replacement assets and sensors will "
+ "receive new IDs. The HEMS energy market and weather station in this "
+ "account will also be replaced.\n"
+ "Type RECREATE to continue: "
+ )
+ return answer == "RECREATE"
+
+
+def confirm_data_wipe(state: dict) -> bool:
+ """Require explicit confirmation before deleting HEMS sensor data."""
+ sensor_count = len(state["sensor-ids"])
+ answer = input(
+ f"This permanently deletes all time-series data from {sensor_count} "
+ "HEMS sensors, including uploads, forecasts, schedules, simulated "
+ "measurements, and report outputs. Asset and sensor IDs are preserved.\n"
+ "Type WIPE to continue: "
+ )
+ return answer == "WIPE"
+
+
+async def upgrade_existing_setup(
+ client: FlexMeasuresClient,
+ account: dict,
+ community_asset: dict,
+ community_name: str,
+ site_names: list[str],
+ state: dict,
+) -> dict:
+ """Bring an older setup up to the current tutorial version, in place.
+
+ Asset setup is idempotent, so re-running it adds whatever the newer tutorial
+ version introduced (sensors, flex-context fields, dashboard panels) while
+ every existing asset and sensor keeps its ID and its data.
+
+ Only report generation is re-run afterwards: the reports are what write the
+ sensors an upgrade is most likely to have added, while the uploaded data,
+ forecasts and schedules remain valid.
+ """
+ print(
+ f"Upgrading this setup from tutorial version {state['workflow-version']} "
+ f"to {WORKFLOW_VERSION}."
+ )
+ print(
+ "Existing assets, sensors and data are preserved; missing structure is "
+ "added and reports are regenerated."
+ )
+ community_asset = await create_community_asset(
+ client,
+ account,
+ community_name=community_name,
+ site_names=site_names,
+ community_asset=community_asset,
+ )
+ return await upgrade_workflow_state(
+ client=client,
+ community_asset=community_asset,
+ account_id=account["id"],
+ state=state,
+ phases_to_rerun=(REPORTING_PHASE,),
+ )
+
+
async def main(
community_name: str, site_names: list[str], callback: Callable = just_continue
):
@@ -24,7 +250,7 @@ async def main(
Complete HEMS setup using FlexMeasures client.
Creates a comprehensive home energy management structure including:
- - Public price sensor for electricity costs
+ - Price sensor for electricity costs
- Building asset with consumption and energy cost KPI sensors
- PV asset (child of building) with production sensor
- Battery asset (child of building) with power and SoC sensors + settings
@@ -36,88 +262,235 @@ async def main(
print("Starting FlexMeasures HEMS")
print("=" * 50)
- # NOTE: Account and admin user creation must be done via FlexMeasures CLI first:
+ if PV_MODE not in {"inflexible", "curtailable"}:
+ raise ValueError(
+ f"Unsupported PV_MODE {PV_MODE!r}; choose 'inflexible' or 'curtailable'."
+ )
+
+ # NOTE: Create an account and an account-admin user via the FlexMeasures CLI
+ # first. The credentials in const.py default to FlexMeasures' toy account:
+ # flexmeasures add toy-account
+ # To use your own account instead, create it and update const.py to match:
# flexmeasures add account --name "MyCompany"
- # flexmeasures add user --username admin --email admin@admin.com --account-id 2 --roles admin
+ # flexmeasures add user --username hems-admin --email hems-admin@example.com \
+ # --account 2 --roles account-admin
- client = FlexMeasuresClient(email=usr, password=pwd, host=host)
+ client = FlexMeasuresClient(email=usr, password=pwd, host=host, ssl=ssl)
try:
+ print(
+ f"Checking server is up and on supported version ... connecting to {host} (ssl: {ssl})"
+ )
+ # The sign-explicit ``inflexible-consumption`` and
+ # ``inflexible-production`` flex-context fields were introduced for
+ # FlexMeasures 1.0. Accept its development releases for testing, too.
await client.ensure_minimum_server_version(
- "0.31.0",
- "The HEMS example requires a FlexMeasures server of v0.31.0 or above.",
+ "1.0.0.dev0",
+ "The HEMS example requires a FlexMeasures server from the v1.0 "
+ "series or above.",
)
# Get user account information
+ print(f"Logging in as {usr} ...")
account = await client.get_account()
if not account:
raise Exception("No account found. Please create an account first.")
account_id = account["id"]
- print(f" Connected to account: {account['name']} (ID: {account_id})")
+ print(f"Connected to account: {account['name']} (ID: {account_id})")
- asset = None # Initialize asset variable
- assets = await client.get_assets(parse_json_fields=True)
- for sst in assets:
- if sst["name"] == community_name:
- asset = sst
- break
+ active_site_names = list(site_names)
+ top_level_assets = await client.get_assets(
+ account_id=account_id,
+ depth=0,
+ fields=["id", "name", "account_id", "parent_asset_id", "attributes"],
+ parse_json_fields=True,
+ )
+ matching_communities = [
+ candidate
+ for candidate in top_level_assets
+ if candidate.get("name") == community_name
+ and candidate.get("account_id") == account_id
+ ]
+ if len(matching_communities) > 1:
+ raise LookupError(
+ f"Expected at most one top-level asset named '{community_name}', "
+ f"found {len(matching_communities)}."
+ )
+ asset = matching_communities[0] if matching_communities else None
if not asset:
print(
"Creating community Site asset with 2 building assets, each with PV and battery sensors, and weather station"
)
- await create_community_asset(
- client, account, community_name=community_name, site_names=site_names
+ asset = await create_community_asset(
+ client,
+ account,
+ community_name=community_name,
+ site_names=active_site_names,
+ )
+ state = await initialize_workflow_state(
+ client, asset, account_id, active_site_names
)
else:
- answer = input(f"Asset '{community_name}' already exists. Re-create?")
- if answer.lower() in ["y", "yes"]:
- await cleanup_existing_assets(
- client=client,
- account_id=account["id"],
- site_names=[community_name],
- )
- await create_community_asset(
- client,
- account,
- community_name=community_name,
- site_names=site_names,
+ existing_site_assets = await get_site_assets(
+ client, asset["id"], account_id
+ )
+ existing_site_names = [site["name"] for site in existing_site_assets]
+ state = get_workflow_state(asset)
+
+ if state is None or state.get("status") == "untracked":
+ action = prompt_for_untracked_setup(existing_site_names, site_names)
+ if action == "exit":
+ print("Exiting without making changes.")
+ return
+ if action == "recreate":
+ if not confirm_recreation(account, community_name):
+ print("Recreation cancelled. No assets or data were deleted.")
+ return
+ await delete_hems_assets(
+ client=client,
+ account_id=account["id"],
+ community_name=community_name,
+ confirm_first=False,
+ )
+ asset = await create_community_asset(
+ client,
+ account,
+ community_name=community_name,
+ site_names=active_site_names,
+ )
+ else:
+ if action == "keep-names":
+ active_site_names = existing_site_names
+ elif action == "rename-sites":
+ await rename_site_assets(
+ client, existing_site_assets, active_site_names
+ )
+ asset = await create_community_asset(
+ client,
+ account,
+ community_name=community_name,
+ site_names=active_site_names,
+ community_asset=asset,
+ )
+ state = await initialize_workflow_state(
+ client, asset, account_id, active_site_names
)
else:
- print("Assets already exist, skipping to data upload")
+ active_site_names = list(
+ state.get("site-names") or existing_site_names or site_names
+ )
+ if "site-names" not in state:
+ state = {**state, "site-names": active_site_names}
+
+ if state.get("status") == "wiping":
+ action = prompt_for_interrupted_wipe()
+ else:
+ action = prompt_for_existing_setup(account, community_name, state)
+
+ if action == "exit":
+ print("Exiting without making changes.")
+ return
+ if action == "recreate":
+ if not confirm_recreation(account, community_name):
+ print("Recreation cancelled. No assets or data were deleted.")
+ return
+ active_site_names = list(site_names)
+ await delete_hems_assets(
+ client=client,
+ account_id=account["id"],
+ community_name=community_name,
+ confirm_first=False,
+ )
+ asset = await create_community_asset(
+ client,
+ account,
+ community_name=community_name,
+ site_names=active_site_names,
+ )
+ state = await initialize_workflow_state(
+ client, asset, account_id, active_site_names
+ )
+ elif action in {"wipe", "continue-wipe"}:
+ if not confirm_data_wipe(state):
+ print("Data wipe cancelled. No additional data was deleted.")
+ return
+ state = await wipe_hems_sensor_data(client, asset["id"], state)
+ else:
+ print("Resuming the existing HEMS setup.")
+
+ # Run last, so that an interrupted wipe is recovered first and
+ # a recreation has already rebuilt the structure from scratch.
+ if state_needs_upgrade(state):
+ state = await upgrade_existing_setup(
+ client=client,
+ account=account,
+ community_asset=asset,
+ community_name=community_name,
+ site_names=active_site_names,
+ state=state,
+ )
# Part 2: Upload data for first two weeks
print("\n" + "=" * 50)
- print("PART 2: UPLOADING DATA")
- await upload_data_for_first_two_weeks(
- client, community_name=community_name, site_names=site_names
- )
+ if phase_is_complete(state, DATA_UPLOAD_PHASE):
+ print("PART 2: UPLOADING DATA (already complete; skipping)")
+ else:
+ print("PART 2: UPLOADING DATA")
+ await upload_data_for_first_two_weeks(
+ client, community_name=community_name, site_names=active_site_names
+ )
+ state = await mark_phase_complete(
+ client, asset["id"], state, DATA_UPLOAD_PHASE
+ )
# Part 3: Generate PV forecasts for second week
print("\n" + "=" * 50)
- print("PART 3: GENERATING PV FORECASTS")
- await generate_forecasts(
- client, community_name=community_name, site_names=site_names
- )
+ if phase_is_complete(state, FORECASTING_PHASE):
+ print("PART 3: GENERATING PV FORECASTS (already complete; skipping)")
+ else:
+ print("PART 3: GENERATING PV FORECASTS")
+ await generate_forecasts(
+ client, community_name=community_name, site_names=active_site_names
+ )
+ state = await mark_phase_complete(
+ client, asset["id"], state, FORECASTING_PHASE
+ )
# Part 4: Run scheduling simulation for third week
print("\n" + "=" * 50)
- print("PART 4: SCHEDULING SIMULATION")
- await run_scheduling_simulation(
- client,
- community_name=community_name,
- site_names=site_names,
- callback=callback,
- )
+ if phase_is_complete(state, SCHEDULING_PHASE):
+ print("PART 4: SCHEDULING SIMULATION (already complete; skipping)")
+ else:
+ print("PART 4: SCHEDULING SIMULATION")
+ scheduling_succeeded = await run_scheduling_simulation(
+ client,
+ community_name=community_name,
+ site_names=active_site_names,
+ callback=callback,
+ )
+ if not scheduling_succeeded:
+ raise RuntimeError("Scheduling simulation did not complete.")
+ state = await mark_phase_complete(
+ client, asset["id"], state, SCHEDULING_PHASE
+ )
# Part 5 : Create reports
print("\n" + "=" * 50)
- print("PART 5: CREATING REPORTS")
- # todo B2: compute aggregate power flow for the community asset's power sensor
- await create_reports(
- client, community_name=community_name, site_names=site_names
- )
+ if phase_is_complete(state, REPORTING_PHASE):
+ print("PART 5: CREATING REPORTS (already complete; skipping)")
+ else:
+ print("PART 5: CREATING REPORTS")
+ # todo B2: compute aggregate power flow for the community asset's power sensor
+ reports_succeeded = await create_reports(
+ client, community_name=community_name, site_names=active_site_names
+ )
+ if not reports_succeeded:
+ raise RuntimeError("Report generation did not complete.")
+ state = await mark_phase_complete(
+ client, asset["id"], state, REPORTING_PHASE
+ )
print("\n" + "=" * 50)
print("HEMS Tutorial completed successfully!")
diff --git a/examples/HEMS/assets_setup.py b/examples/HEMS/assets_setup.py
index 84d21650..64fb00bc 100644
--- a/examples/HEMS/assets_setup.py
+++ b/examples/HEMS/assets_setup.py
@@ -2,6 +2,7 @@
BATTERY_CONFIG,
EV_CONFIG,
HEATING_CONFIG,
+ PV_MODE,
battery_name,
evse1_name,
evse2_name,
@@ -12,112 +13,143 @@
pv_name,
weather_station_name,
)
-from utils.asset_utils import get_first_asset_by_name
from flexmeasures_client import FlexMeasuresClient
-async def create_public_price_sensor(client: FlexMeasuresClient):
- """Create a public price sensor (1h, EUR/kWh).
+async def get_or_create_asset(
+ client: FlexMeasuresClient,
+ *,
+ name: str,
+ account_id: int,
+ generic_asset_type_id: int,
+ parent_asset_id: int | None = None,
+) -> dict:
+ """Return one exact asset or create it in the requested hierarchy position."""
+ assets = await client.get_assets(
+ account_id=account_id,
+ fields=["id", "name", "account_id", "parent_asset_id"],
+ parse_json_fields=False,
+ )
+ matches = [
+ asset
+ for asset in assets
+ if asset.get("name") == name
+ and asset.get("account_id") == account_id
+ and asset.get("parent_asset_id") == parent_asset_id
+ ]
+ if len(matches) > 1:
+ raise LookupError(
+ f"Expected at most one asset named '{name}' under parent "
+ f"{parent_asset_id}, found {len(matches)}."
+ )
+ if matches:
+ print(f"Reusing asset '{name}' with ID {matches[0]['id']}")
+ return matches[0]
+ return await client.add_asset(
+ name=name,
+ latitude=latitude,
+ longitude=longitude,
+ generic_asset_type_id=generic_asset_type_id,
+ account_id=account_id,
+ parent_asset_id=parent_asset_id,
+ )
+
+
+async def get_or_create_sensor(
+ client: FlexMeasuresClient,
+ *,
+ name: str,
+ event_resolution: str,
+ unit: str,
+ generic_asset_id: int,
+ timezone: str | None = "Europe/Amsterdam",
+ attributes: dict | None = None,
+) -> dict:
+ """Return one exact sensor on an asset or create the missing sensor."""
+ sensors = await client.get_sensors(
+ asset_id=generic_asset_id, parse_json_fields=False
+ )
+ # FlexMeasures 0.33 may include sensors on descendant assets here.
+ matches = [
+ sensor
+ for sensor in sensors
+ if sensor.get("name") == name
+ and sensor.get("generic_asset_id") == generic_asset_id
+ ]
+ if len(matches) > 1:
+ raise LookupError(
+ f"Expected at most one sensor named '{name}' on asset "
+ f"{generic_asset_id}, found {len(matches)}."
+ )
+ if matches:
+ print(f"Reusing sensor '{name}' with ID {matches[0]['id']}")
+ return matches[0]
+ return await client.add_sensor(
+ name=name,
+ event_resolution=event_resolution,
+ unit=unit,
+ generic_asset_id=generic_asset_id,
+ timezone=timezone,
+ attributes=attributes,
+ )
+
+
+async def get_or_create_price_sensor(client: FlexMeasuresClient):
+ """Get or create an account-owned price sensor (1h, EUR/kWh).
Returns the price sensor for use in flex-context.
"""
- print("Creating public price sensor...")
+ print("Getting or creating price sensor...")
# Get the client account id
account = await client.get_account()
account_id = account["id"]
print(f"Account ID: {account_id}")
- # Create top-level market asset (not public, but still under the toy account)
- # Generic asset type 8 is typically used for market/price assets
- all_top_level_assets = await client.get_assets(
- include_public=True,
- depth=0,
- fields=["id", "name", "account_id", "sensors"],
- )
- price_market_asset = get_first_asset_by_name(
- assets=all_top_level_assets, name=price_market_name, account_id=account_id
- )
- if price_market_asset is None:
- price_market_asset = await client.add_asset(
- name=price_market_name,
- latitude=latitude,
- longitude=longitude,
- generic_asset_type_id=8, # Transmission zone A grid regulated & balanced as a whole, usually a national grid.
- account_id=account_id,
- )
-
- # Create price sensor with 1-hour resolution
- price_sensor = await client.add_sensor(
- name="electricity-price",
- event_resolution="PT1H",
- unit="EUR/kWh",
- generic_asset_id=price_market_asset["id"],
- timezone="Europe/Amsterdam",
- )
- else:
- price_sensor = price_market_asset["sensors"][0]
+ price_market_asset = await get_or_create_asset(
+ client,
+ name=price_market_name,
+ account_id=account_id,
+ generic_asset_type_id=8,
+ )
+ price_sensor = await get_or_create_sensor(
+ client,
+ name="electricity-price",
+ event_resolution="PT1H",
+ unit="EUR/kWh",
+ generic_asset_id=price_market_asset["id"],
+ )
- print(f"Created public price sensor with ID: {price_sensor['id']}")
+ print(f"Price sensor ID: {price_sensor['id']}")
return price_sensor
-async def create_weather_station(client: FlexMeasuresClient):
- """Create a public weather station with irradiation and cloud coverage sensors."""
- print("Creating weather station...")
+async def get_or_create_weather_station(client: FlexMeasuresClient):
+ """Get or create an account-owned weather station and its sensors."""
+ print("Getting or creating weather station...")
# Get the client account id
account = await client.get_account()
account_id = account["id"]
print(f"Account ID: {account_id}")
- # Create top-level weather station asset (not public, but still under the toy account)
- # Generic asset type 7 (process) used for weather stations since no dedicated type exists
- # TODO: remove hard-coded ID, we should actually create a weather station type somehow
- all_top_level_assets = await client.get_assets(
- include_public=True,
- depth=0,
- fields=["id", "name", "account_id", "sensors"],
- )
- weather_asset = get_first_asset_by_name(
- assets=all_top_level_assets, name=weather_station_name, account_id=account_id
- )
- if weather_asset is None:
- weather_asset = await client.add_asset(
- name=weather_station_name,
- latitude=latitude,
- longitude=longitude,
- generic_asset_type_id=7, # Process asset type (for weather station)
- account_id=account_id, # Public account ID
- )
-
- # Create irradiation sensor (1H, W/m²)
- irradiation_sensor = await client.add_sensor(
- name="irradiation",
- event_resolution="PT1H",
- unit="W/m²",
- generic_asset_id=weather_asset["id"],
- timezone="Europe/Amsterdam",
- )
-
- # Create cloud coverage sensor (1H, %)
- cloud_coverage_sensor = await client.add_sensor(
- name="cloud-coverage",
- event_resolution="PT1H",
- unit="%",
- generic_asset_id=weather_asset["id"],
- timezone="Europe/Amsterdam",
- )
- else:
- sensors = weather_asset["sensors"]
- cloud_coverage_sensor = None
- irradiation_sensor = None
- for sensor in sensors:
- if sensor["name"] == "cloud-coverage":
- cloud_coverage_sensor = sensor
- elif sensor["name"] == "irradiation":
- irradiation_sensor = sensor
- if cloud_coverage_sensor is None or irradiation_sensor is None:
- raise ValueError(
- "Could not identify cloud_coverage_sensor or irradiation_sensor. Maybe a name changed?"
- )
+ weather_asset = await get_or_create_asset(
+ client,
+ name=weather_station_name,
+ account_id=account_id,
+ generic_asset_type_id=7,
+ )
+ irradiation_sensor = await get_or_create_sensor(
+ client,
+ name="irradiation",
+ event_resolution="PT1H",
+ unit="W/m²",
+ generic_asset_id=weather_asset["id"],
+ )
+ cloud_coverage_sensor = await get_or_create_sensor(
+ client,
+ name="cloud-coverage",
+ event_resolution="PT1H",
+ unit="%",
+ generic_asset_id=weather_asset["id"],
+ )
print(f"Created weather station with ID: {weather_asset['id']}")
return weather_asset, irradiation_sensor, cloud_coverage_sensor
@@ -134,17 +166,17 @@ async def create_site_asset(
print("Creating Site asset...")
# Create site asset (generic_asset_type_id=6 for building)
- site_asset = await client.add_asset(
+ site_asset = await get_or_create_asset(
+ client,
name=site_name,
- latitude=latitude,
- longitude=longitude,
parent_asset_id=site_asset_id,
- generic_asset_type_id=6, # Building asset type
+ generic_asset_type_id=6,
account_id=account_id,
)
# Create general consumption sensor (15min resolution, kW)
- consumption_sensor = await client.add_sensor(
+ consumption_sensor = await get_or_create_sensor(
+ client,
name="electricity-consumption",
event_resolution="PT15M",
unit="kW",
@@ -154,7 +186,8 @@ async def create_site_asset(
)
# Create energy costs KPI sensor (1D resolution, EUR)
- energy_costs_sensor = await client.add_sensor(
+ energy_costs_sensor = await get_or_create_sensor(
+ client,
name="energy-costs-kpi",
event_resolution="P1D",
unit="EUR",
@@ -163,7 +196,8 @@ async def create_site_asset(
)
# Create aggregate power sensor for the site
- aggregate_sensor = await client.add_sensor(
+ aggregate_sensor = await get_or_create_sensor(
+ client,
name="electricity-aggregate",
event_resolution="PT15M",
unit="kW",
@@ -173,7 +207,8 @@ async def create_site_asset(
)
# Create max production capacity sensor for the site
- max_production_sensor = await client.add_sensor(
+ max_production_sensor = await get_or_create_sensor(
+ client,
name="max-production-capacity",
event_resolution="PT1H",
unit="kW",
@@ -183,7 +218,8 @@ async def create_site_asset(
)
# Create max consumption capacity sensor for the site
- max_consumption_sensor = await client.add_sensor(
+ max_consumption_sensor = await get_or_create_sensor(
+ client,
name="max-consumption-capacity",
event_resolution="PT1H",
unit="kW",
@@ -193,7 +229,8 @@ async def create_site_asset(
)
# Create site-peak-consumption-price sensor (15min resolution, EUR/kW)
- site_peak_consumption_price_sensor = await client.add_sensor( # noqa: F841
+ site_peak_consumption_price_sensor = await get_or_create_sensor(
+ client,
name="site-peak-consumption-price",
event_resolution="PT15M",
unit="EUR/kW",
@@ -202,7 +239,8 @@ async def create_site_asset(
)
# Create site-peak-production-price sensor (15min resolution, EUR/kW)
- site_peak_production_price_sensor = await client.add_sensor( # noqa: F841
+ site_peak_production_price_sensor = await get_or_create_sensor(
+ client,
name="site-peak-production-price",
event_resolution="PT15M",
unit="EUR/kW",
@@ -211,7 +249,8 @@ async def create_site_asset(
)
# Create self-consumption sensor for the site
- self_consumption_sensor = await client.add_sensor(
+ self_consumption_sensor = await get_or_create_sensor(
+ client,
name="self-consumption",
event_resolution="PT15M",
unit="kW",
@@ -221,7 +260,8 @@ async def create_site_asset(
)
# Create total energy costs sensor for the site
- total_energy_costs_sensor = await client.add_sensor(
+ total_energy_costs_sensor = await get_or_create_sensor(
+ client,
name="total-energy-costs",
event_resolution="PT15M",
unit="EUR",
@@ -230,7 +270,8 @@ async def create_site_asset(
)
# Create daily total energy costs sensor for the site
- daily_total_energy_costs_sensor = await client.add_sensor(
+ daily_total_energy_costs_sensor = await get_or_create_sensor(
+ client,
name="daily-total-energy-costs",
event_resolution="P1D",
unit="EUR",
@@ -239,7 +280,8 @@ async def create_site_asset(
)
# Create daily share of self-consumption sensor for the site
- daily_share_of_self_consumption_sensor = await client.add_sensor(
+ daily_share_of_self_consumption_sensor = await get_or_create_sensor(
+ client,
name="daily-share-of-self-consumption",
event_resolution="P1D",
unit="%",
@@ -271,35 +313,64 @@ async def create_pv_asset(
print("Creating PV asset...")
# Create PV asset (generic_asset_type_id=1 for solar/PV)
- pv_asset = await client.add_asset(
+ pv_asset = await get_or_create_asset(
+ client,
name=pv_name,
- latitude=latitude,
- longitude=longitude,
- generic_asset_type_id=1, # Solar/PV asset type
+ generic_asset_type_id=1,
account_id=account_id,
- parent_asset_id=site_asset_id, # Child of site
+ parent_asset_id=site_asset_id,
)
# Create production sensor (15min, kW)
- pv_production_sensor = await client.add_sensor( # to store raw generation values
+ pv_production_sensor = await get_or_create_sensor(
+ client,
name="electricity-production",
event_resolution="PT15M",
unit="kW",
generic_asset_id=pv_asset["id"],
timezone="Europe/Amsterdam",
+ attributes=dict(consumption_is_positive=False),
)
# Create power sensor (15min, kW)
- pv_power_sensor = await client.add_sensor( # to store realized generation values
+ pv_power_sensor = await get_or_create_sensor(
+ client,
name="electricity-power",
event_resolution="PT15M",
unit="kW",
generic_asset_id=pv_asset["id"],
timezone="Europe/Amsterdam",
+ attributes=dict(consumption_is_positive=False),
+ )
+
+ pv_feed_in_sensor = await get_or_create_sensor(
+ client,
+ name="solar-feed-in",
+ event_resolution="PT15M",
+ unit="kW",
+ generic_asset_id=pv_asset["id"],
+ timezone="Europe/Amsterdam",
+ attributes=dict(consumption_is_positive=False),
+ )
+
+ pv_curtailment_sensor = await get_or_create_sensor(
+ client,
+ name="solar-curtailment",
+ event_resolution="PT15M",
+ unit="kW",
+ generic_asset_id=pv_asset["id"],
+ timezone="Europe/Amsterdam",
+ attributes=dict(consumption_is_positive=False),
)
print(f"Created PV asset with ID: {pv_asset['id']}")
- return pv_asset, pv_production_sensor, pv_power_sensor
+ return (
+ pv_asset,
+ pv_production_sensor,
+ pv_power_sensor,
+ pv_feed_in_sensor,
+ pv_curtailment_sensor,
+ )
async def create_battery_asset(
@@ -312,17 +383,17 @@ async def create_battery_asset(
print("Creating battery asset...")
# Create battery asset (generic_asset_type_id=5 for battery)
- battery_asset = await client.add_asset(
+ battery_asset = await get_or_create_asset(
+ client,
name=battery_name,
- latitude=latitude,
- longitude=longitude,
- generic_asset_type_id=5, # Battery asset type
+ generic_asset_type_id=5,
account_id=account_id,
- parent_asset_id=site_asset_id, # Child of site
+ parent_asset_id=site_asset_id,
)
# Create power sensor (15min, kW)
- battery_power_sensor = await client.add_sensor(
+ battery_power_sensor = await get_or_create_sensor(
+ client,
name="electricity-power",
event_resolution="PT15M",
unit="kW",
@@ -332,7 +403,8 @@ async def create_battery_asset(
)
# Create state-of-charge sensor (0min, kWh)
- battery_soc_sensor = await client.add_sensor(
+ battery_soc_sensor = await get_or_create_sensor(
+ client,
name="state-of-charge",
event_resolution="PT0M",
unit="kWh",
@@ -376,17 +448,17 @@ async def create_evse_asset(
# Create EVSE asset - using generic type 4 for one-way EVSE based on the codebase search
# Note: We'll use a basic asset type since one-way_evse might not be available by default
- evse_asset = await client.add_asset(
+ evse_asset = await get_or_create_asset(
+ client,
name=evse_name,
- latitude=latitude,
- longitude=longitude,
- generic_asset_type_id=4, # Using a generic type, could be EVSE specific if available
+ generic_asset_type_id=4,
account_id=account_id,
- parent_asset_id=site_asset_id, # Child of site
+ parent_asset_id=site_asset_id,
)
# Create power sensor (15min, kW)
- evse_power_sensor = await client.add_sensor(
+ evse_power_sensor = await get_or_create_sensor(
+ client,
name="electricity-power",
event_resolution="PT15M",
unit="kW",
@@ -396,7 +468,8 @@ async def create_evse_asset(
)
# Create state-of-charge sensor (instantaneous, kWh)
- evse_soc_sensor = await client.add_sensor(
+ evse_soc_sensor = await get_or_create_sensor(
+ client,
name="state-of-charge",
event_resolution="PT0M",
unit="kWh",
@@ -405,7 +478,8 @@ async def create_evse_asset(
)
# Create soc-min sensor (15min, kWh)
- evse_soc_min_sensor = await client.add_sensor(
+ evse_soc_min_sensor = await get_or_create_sensor(
+ client,
name="soc-min",
event_resolution="PT15M",
unit="kWh",
@@ -414,7 +488,8 @@ async def create_evse_asset(
)
# Create soc-max sensor (15min, kWh)
- evse_soc_max_sensor = await client.add_sensor(
+ evse_soc_max_sensor = await get_or_create_sensor(
+ client,
name="soc-max",
event_resolution="PT15M",
unit="kWh",
@@ -427,6 +502,7 @@ async def create_evse_asset(
capacity = EV_CONFIG["default_capacity_kwh"]
attributes_flex_model = {
"soc_at_start": capacity * EV_CONFIG["min_soc_percent"], # Start at minimum SoC
+ "capacity_kwh": capacity,
}
flex_model = {
@@ -490,17 +566,17 @@ async def create_heating_asset(
print(f"Creating heating asset: {heating_name}...")
# Create heating asset (generic asset type id = 5 if heating not defined in DB)
- heating_asset = await client.add_asset(
+ heating_asset = await get_or_create_asset(
+ client,
name=heating_name,
- latitude=latitude,
- longitude=longitude,
- generic_asset_type_id=5, # Using battery type as placeholder for heating asset
+ generic_asset_type_id=5,
account_id=account_id,
parent_asset_id=site_asset_id,
)
# Power sensors (15min, kW)
- heating_power_sensor = await client.add_sensor(
+ heating_power_sensor = await get_or_create_sensor(
+ client,
name="power",
event_resolution="PT15M",
unit="kW",
@@ -510,7 +586,8 @@ async def create_heating_asset(
)
# Soc usage sensor (15min, kW)
- heating_soc_usage_sensor = await client.add_sensor(
+ heating_soc_usage_sensor = await get_or_create_sensor(
+ client,
name="soc-usage",
event_resolution="PT15M",
unit="kW",
@@ -520,21 +597,24 @@ async def create_heating_asset(
)
# State of Charge sensors (15min, kWh)
- heating_soc_sensor = await client.add_sensor(
+ heating_soc_sensor = await get_or_create_sensor(
+ client,
name="state of charge",
event_resolution="PT0M",
unit="kWh",
generic_asset_id=heating_asset["id"],
timezone="Europe/Amsterdam",
)
- heating_min_soc_sensor = await client.add_sensor(
+ heating_min_soc_sensor = await get_or_create_sensor(
+ client,
name="min SoC",
event_resolution="PT15M",
unit="kWh",
generic_asset_id=heating_asset["id"],
timezone="Europe/Amsterdam",
)
- heating_max_soc_sensor = await client.add_sensor(
+ heating_max_soc_sensor = await get_or_create_sensor(
+ client,
name="max SoC",
event_resolution="PT15M",
unit="kWh",
@@ -543,7 +623,8 @@ async def create_heating_asset(
)
# COP (Coefficient of Performance)
- heating_COP = await client.add_sensor(
+ heating_COP = await get_or_create_sensor(
+ client,
name="COP",
event_resolution="PT15M",
unit="%",
@@ -654,12 +735,15 @@ async def configure_site_flex_context(
# "site-production-breach-price": "10000000 EUR/MW",
# "consumption-breach-price": "1000 EUR/MW",
# "production-breach-price": "1000 EUR/MW",
- # Add inflexible devices as requested
- "inflexible-device-sensors": [
- consumption_sensor["id"], # General consumption
- ],
+ "inflexible-consumption": [{"sensor": consumption_sensor["id"]}],
"aggregate-power": {"sensor": aggregate_sensor["id"]},
}
+ if PV_MODE == "inflexible":
+ flex_context["inflexible-production"] = [{"sensor": pv_production_sensor["id"]}]
+ elif PV_MODE != "curtailable":
+ raise ValueError(
+ f"Unsupported PV_MODE {PV_MODE!r}; choose 'inflexible' or 'curtailable'."
+ )
# Update site asset with flex-context
await client.update_asset(
@@ -675,10 +759,14 @@ async def configure_site_dashboard(
consumption_sensor,
pv_production_sensor,
pv_power_sensor,
+ pv_feed_in_sensor,
+ pv_curtailment_sensor,
battery_power_sensor,
battery_soc_sensor,
evse1_power_sensor,
+ evse1_soc_sensor,
evse2_power_sensor,
+ evse2_soc_sensor,
heating_power_sensor,
heating_soc_sensor,
aggregate_sensor,
@@ -707,11 +795,13 @@ async def configure_site_dashboard(
],
},
{
- "title": "Solar self-consumption",
+ "title": "PV production and use",
"sensors": [
self_consumption_sensor["id"],
pv_production_sensor["id"],
pv_power_sensor["id"],
+ pv_feed_in_sensor["id"],
+ pv_curtailment_sensor["id"],
],
},
{
@@ -728,7 +818,12 @@ async def configure_site_dashboard(
},
{
"title": "Storages SoC",
- "sensors": [battery_soc_sensor["id"], heating_soc_sensor["id"]],
+ "sensors": [
+ battery_soc_sensor["id"],
+ evse1_soc_sensor["id"],
+ evse2_soc_sensor["id"],
+ heating_soc_sensor["id"],
+ ],
},
{
"title": "Site capacity",
@@ -808,7 +903,13 @@ async def create_sites_assets_and_sensors(
print(f"Max consumption sensor ID: {max_consumption_sensor['id']}")
print(f"Self-consumption sensor ID: {self_consumption_sensor['id']}")
print("Creating PV asset with production sensor")
- pv_asset, pv_production_sensor, pv_power_sensor = await create_pv_asset(
+ (
+ pv_asset,
+ pv_production_sensor,
+ pv_power_sensor,
+ pv_feed_in_sensor,
+ pv_curtailment_sensor,
+ ) = await create_pv_asset(
client, account_id, site_asset["id"], pv_name=f"{pv_name} {site_index}"
)
print(f"PV asset ID: {pv_asset['id']}")
@@ -908,10 +1009,14 @@ async def create_sites_assets_and_sensors(
consumption_sensor=consumption_sensor,
pv_production_sensor=pv_production_sensor,
pv_power_sensor=pv_power_sensor,
+ pv_feed_in_sensor=pv_feed_in_sensor,
+ pv_curtailment_sensor=pv_curtailment_sensor,
battery_power_sensor=battery_power_sensor,
battery_soc_sensor=battery_soc_sensor,
evse1_power_sensor=evse1_power_sensor,
+ evse1_soc_sensor=evse1_soc_sensor,
evse2_power_sensor=evse2_power_sensor,
+ evse2_soc_sensor=evse2_soc_sensor,
heating_power_sensor=heating_power_sensor,
heating_soc_sensor=heating_soc_sensor,
aggregate_sensor=aggregate_sensor,
@@ -930,47 +1035,51 @@ async def create_community_asset(
account: dict,
community_name: str,
site_names: list[str],
+ community_asset: dict | None = None,
):
- """Create an asset representing a community, which will serve as the parent asset for all sites in the community."""
+ """Create or complete the HEMS asset structure without replacing existing IDs."""
# Get account id
account_id = account["id"]
- print("Creating price market asset and associated price sensor")
- price_sensor = await create_public_price_sensor(client=client)
+ print("Getting or creating price market asset and associated price sensor")
+ price_sensor = await get_or_create_price_sensor(client=client)
- print("Creating weather station with irradiation and cloud coverage sensors")
+ print("Getting or creating weather station and its sensors")
weather_asset, irradiation_sensor, cloud_coverage_sensor = (
- await create_weather_station(client=client)
+ await get_or_create_weather_station(client=client)
)
print(f"Weather station asset ID: {weather_asset['id']}")
print(f"Irradiation sensor ID: {irradiation_sensor['id']}")
print(f"Cloud coverage sensor ID: {cloud_coverage_sensor['id']}")
print("Creating community asset...")
- # Create Site asset (generic_asset_type_id=6 for building)
- site_asset = await client.add_asset(
- name=community_name,
- latitude=latitude,
- longitude=longitude,
- generic_asset_type_id=6, # Building asset type
- account_id=account_id,
- )
+ if community_asset is None:
+ community_asset = await get_or_create_asset(
+ client,
+ name=community_name,
+ generic_asset_type_id=6,
+ account_id=account_id,
+ )
+ elif community_asset.get("parent_asset_id") is not None:
+ raise ValueError("The HEMS community asset must be a top-level asset.")
# Create site power capacity sensor (15min resolution, kW)
- site_power_capacity_sensor = await client.add_sensor(
+ site_power_capacity_sensor = await get_or_create_sensor(
+ client,
name="site-power-capacity",
event_resolution="PT15M",
unit="kW",
- generic_asset_id=site_asset["id"],
+ generic_asset_id=community_asset["id"],
timezone="Europe/Amsterdam",
attributes=dict(consumption_is_positive=True),
)
# Create site power sensor (15min resolution, kW)
# this is used to store aggregate assets power measurements
- site_power_sensor = await client.add_sensor( # noqa: F841
+ site_power_sensor = await get_or_create_sensor( # noqa: F841
+ client,
name="power",
event_resolution="PT15M",
unit="kW",
- generic_asset_id=site_asset["id"],
+ generic_asset_id=community_asset["id"],
timezone="Europe/Amsterdam",
attributes=dict(consumption_is_positive=True),
)
@@ -979,18 +1088,19 @@ async def create_community_asset(
flex_context = {
"site-power-capacity": {"sensor": site_power_capacity_sensor["id"]},
}
- print(f"Created site asset with ID: {site_asset['id']}")
+ print(f"Community asset ID: {community_asset['id']}")
# Update site asset with flex-context
await client.update_asset(
- asset_id=site_asset["id"], updates={"flex_context": flex_context}
+ asset_id=community_asset["id"], updates={"flex_context": flex_context}
)
for i in range(len(site_names)):
await create_sites_assets_and_sensors(
client=client,
account=account,
- community_asset_id=site_asset["id"],
+ community_asset_id=community_asset["id"],
site_index=i + 1,
site_names=site_names,
price_sensor=price_sensor,
)
+ return community_asset
diff --git a/examples/HEMS/configs/self-consumption_reporter_config.json b/examples/HEMS/configs/self-consumption_reporter_config.json
index 0b24709a..46e58c27 100644
--- a/examples/HEMS/configs/self-consumption_reporter_config.json
+++ b/examples/HEMS/configs/self-consumption_reporter_config.json
@@ -1,6 +1,6 @@
{
"required_input" :[{"name":"production","unit": "kW"}, {"name":"pv-power","unit": "kW"}, {"name":"heating-power","unit": "kW"}, {"name":"evse1-consumption","unit": "kW"}, {"name":"evse2-consumption","unit": "kW"}, {"name":"building-consumption","unit": "kW"}, {"name":"battery-power","unit": "kW"}],
- "required_output" :[{"name":"self-consumption","unit": "kW"}, {"name":"daily-share-of-self-consumption", "unit": "%"}],
+ "required_output" :[{"name":"self-consumption","unit": "kW"}, {"name":"solar-feed-in","unit": "kW"}, {"name":"solar-curtailment","unit": "kW"}, {"name":"daily-share-of-self-consumption", "unit": "%"}],
"droplevels": true,
"transformations" : [
{
@@ -39,7 +39,7 @@
},
{
"df_input": "excess-pv",
- "df_output": "excess-pv",
+ "df_output": "solar-feed-in",
"method": "clip",
"kwargs": {"lower": 0}
},
@@ -47,7 +47,19 @@
"df_input": "pv-power",
"df_output": "self-consumption",
"method": "sub",
- "args": ["@excess-pv"]
+ "args": ["@solar-feed-in"]
+ },
+ {
+ "df_input": "production",
+ "df_output": "solar-curtailment",
+ "method": "sub",
+ "args": ["@pv-power"]
+ },
+ {
+ "df_input": "solar-curtailment",
+ "df_output": "solar-curtailment",
+ "method": "clip",
+ "kwargs": {"lower": 0}
},
{
"df_input": "self-consumption",
@@ -59,8 +71,8 @@
"method": "sum"
},
{
- "df_input": "production",
- "df_output": "daily solar production",
+ "df_input": "pv-power",
+ "df_output": "daily delivered solar production",
"method": "resample",
"args": ["1D"]
},
@@ -71,7 +83,7 @@
"df_input": "daily self-consumption",
"df_output": "daily-share-of-self-consumption",
"method": "divide",
- "args": ["@daily solar production"]
+ "args": ["@daily delivered solar production"]
},
{
"method": "multiply",
diff --git a/examples/HEMS/configs/total-energy-costs_reporter_config.json b/examples/HEMS/configs/total-energy-costs_reporter_config.json
index 52c5c98a..1dcadd7f 100644
--- a/examples/HEMS/configs/total-energy-costs_reporter_config.json
+++ b/examples/HEMS/configs/total-energy-costs_reporter_config.json
@@ -46,6 +46,12 @@
"args": ["@feed_in_revenue"],
"df_output": "total-energy-costs"
},
+ {
+ "df_input": "total-energy-costs",
+ "method": "multiply",
+ "args": [0.25],
+ "df_output": "total-energy-costs"
+ },
{
"df_input": "total-energy-costs",
"method": "resample",
diff --git a/examples/HEMS/const.py b/examples/HEMS/const.py
index abeb4de2..b8aa9a2d 100644
--- a/examples/HEMS/const.py
+++ b/examples/HEMS/const.py
@@ -4,13 +4,19 @@
Settings for the HEMS example script.
"""
# Connection details - UPDATE THESE FOR YOUR SETUP
-usr = "admin@admin.com" # Admin user email
-pwd = "admin" # Admin password
-host = "127.0.0.1:5000" # FlexMeasures host
+usr = "toy-user@flexmeasures.io" # Account-admin user email
+pwd = "toy-password" # Account-admin user password
+host = "127.0.0.1:5000" # FlexMeasures host, without http:// or https://
+ssl = False # Local development server uses HTTP; set True for HTTPS deployments
+
+# PV operation mode:
+# - "inflexible": all available PV is delivered; surplus production is fed in.
+# - "curtailable": FlexMeasures may schedule PV below its available production.
+PV_MODE = "inflexible"
# Asset and sensor names
COMMUNITY_NAME = "Community Site"
-SITE_NAMES = ["My Home 1", "My Home 2"]
+SITE_NAMES = ["Building A", "Building B"]
pv_name = "Rooftop PV"
battery_name = "Home Battery"
@@ -28,7 +34,7 @@
TUTORIAL_START_DATE = "2030-01-01T00:00:00+01:00"
FORECASTING_START = "2030-01-15T00:00:00+01:00"
SCHEDULING_START = "2030-01-15T00:00:00+01:00"
-SCHEDULING_END = "2030-01-16T00:00:00+01:00"
+SCHEDULING_END = "2030-01-17T00:00:00+01:00"
SIMULATION_STEP_HOURS = 4
FORECAST_HORIZON_HOURS = 24
MAX_RESCHEDULING_ITERATIONS = 2
@@ -37,13 +43,14 @@
# Configuration constants
EV_CONFIG = {
- "default_capacity_kwh": 60.0,
+ "default_capacity_kwh": 40.0,
"default_power_capacity_kw": 11.0,
"min_soc_percent": 0.20, # 20% minimum SoC
"roundtrip_efficiency": 0.85, # 85% efficiency
"random_trip_probability": 0.10, # 10% chance per step
"random_trip_consumption_range": (0.10, 0.20), # 10-20% consumption
"driving_consumption_kwh_per_hour": 7.5, # 15 kWh/100km at 50 km/h average
+ "one_way_commute_duration_hours": 0.5,
}
BATTERY_CONFIG = {
@@ -59,10 +66,10 @@
# Each entry represents: (needs_charging_overnight, departure_time, return_time, target_soc_percent)
# Index 0 = Monday, 1 = Tuesday, ..., 6 = Sunday
EV_WEEKLY_PATTERNS = [
- (False, None, None, 40), # Monday - Free day, keep at moderate charge
- (True, "07:00", "13:00", 80), # Tuesday - Work day, need 80% by 7am
- (True, "08:00", "13:00", 80), # Wednesday - Work day, need 80% by 8am
- (True, "07:00", "13:00", 80), # Thursday - Work day, need 80% by 7am
+ (True, "07:00", "13:00", 60), # Monday - Work day, need 60% by 7am
+ (True, "07:00", "13:00", 60), # Tuesday - Work day, need 60% by 7am
+ (True, "08:00", "13:00", 60), # Wednesday - Work day, need 60% by 8am
+ (True, "07:00", "13:00", 60), # Thursday - Work day, need 60% by 7am
(False, None, None, 60), # Friday - Free day, charge to 60% for weekend
(False, None, None, 40), # Saturday - Free day
(False, None, None, 40), # Sunday - Free day
diff --git a/examples/HEMS/forecasting.py b/examples/HEMS/forecasting.py
index f00cc0ad..9f3eab2b 100644
--- a/examples/HEMS/forecasting.py
+++ b/examples/HEMS/forecasting.py
@@ -73,11 +73,9 @@ async def generate_sensor_forecasts(
)
except Exception as exc:
job_id = forecast_id if forecast_id is not None else "unknown"
- print(f"Forecast job {job_id} failed for {sensor_name} on {asset_name}: {exc}")
- print(
- "Look up this job in the RQ dashboard for more details about the failure."
- )
- return None
+ raise RuntimeError(
+ f"Forecast job {job_id} failed for {sensor_name} on " f"{asset_name}: {exc}"
+ ) from exc
print(f"Forecast job completed for {sensor_name} on {asset_name}")
diff --git a/examples/HEMS/reporters.py b/examples/HEMS/reporters.py
index 958e387e..99e4431a 100644
--- a/examples/HEMS/reporters.py
+++ b/examples/HEMS/reporters.py
@@ -34,12 +34,15 @@ async def create_reports(
if check_result.returncode != 0:
print("FlexMeasures CLI not found. Skipping report generation.")
return False
+ all_reports_succeeded = True
for i, site_name in enumerate(site_names, start=1):
# Find all required sensors
sensor_mappings = [
("electricity-production", "electricity-production", f"{pv_name} {i}"),
("pv-power", "electricity-power", f"{pv_name} {i}"),
+ ("solar-feed-in", "solar-feed-in", f"{pv_name} {i}"),
+ ("solar-curtailment", "solar-curtailment", f"{pv_name} {i}"),
("electricity-consumption", "electricity-consumption", site_name),
("electricity-power", "electricity-power", f"{battery_name} {i}"),
("evse1-power", "electricity-power", f"{evse1_name} {i}"),
@@ -73,6 +76,8 @@ async def create_reports(
],
output_sensors=[
sensors["self-consumption"],
+ sensors["solar-feed-in"],
+ sensors["solar-curtailment"],
sensors["daily-share-of-self-consumption"],
],
start=SCHEDULING_START,
@@ -109,5 +114,10 @@ async def create_reports(
start=SCHEDULING_START,
end=SCHEDULING_END,
)
+ all_reports_succeeded = (
+ self_consumption_result
+ and total_energy_costs_result
+ and all_reports_succeeded
+ )
- return self_consumption_result and total_energy_costs_result
+ return all_reports_succeeded
diff --git a/examples/HEMS/scheduling.py b/examples/HEMS/scheduling.py
index d77c5b86..87c2c817 100644
--- a/examples/HEMS/scheduling.py
+++ b/examples/HEMS/scheduling.py
@@ -10,6 +10,7 @@
FORECAST_HORIZON_HOURS,
HEATING_CONFIG,
MAX_RESCHEDULING_ITERATIONS,
+ PV_MODE,
SCHEDULING_END,
SCHEDULING_START,
SIMULATION_STEP_HOURS,
@@ -25,6 +26,8 @@
find_sensor_by_name_and_asset,
find_top_level_asset_id,
load_and_align_csv_data,
+ post_sensor_data_and_track_ingestion,
+ wait_for_ingestion_jobs,
)
from utils.ev_utils import (
calculate_ev_soc_targets_and_constraints,
@@ -42,6 +45,28 @@ async def just_continue(*args, **kwargs):
return True
+def realize_pv_power(
+ available_power: list[float],
+ scheduled_power: list[float],
+ pv_mode: str,
+) -> list[float]:
+ """Return delivered PV power for the configured operating mode."""
+ if pv_mode == "inflexible":
+ return list(available_power)
+ if pv_mode != "curtailable":
+ raise ValueError(
+ f"Unsupported PV_MODE {pv_mode!r}; choose 'inflexible' or 'curtailable'."
+ )
+ if len(available_power) != len(scheduled_power):
+ raise ValueError(
+ "Available and scheduled PV power must contain the same number of values."
+ )
+ return [
+ max(min(available, scheduled), 0)
+ for available, scheduled in zip(available_power, scheduled_power)
+ ]
+
+
async def run_scheduling_simulation(
client: FlexMeasuresClient,
community_name: str,
@@ -179,6 +204,7 @@ async def run_scheduling_simulation(
# Stop rescheduling
break
+ pending_ingestion_jobs: list[str] = []
for index, site_name in enumerate(site_names, start=1):
# Extract scheduled power for all devices for the next 4 hours
# Update SoC for next step based on retrieved SoC schedules
@@ -201,20 +227,29 @@ async def run_scheduling_simulation(
heating_soc_schedule=heating_soc_schedules[index - 1],
evse1_flex_model=evse1_flex_models[index - 1],
evse2_flex_model=evse2_flex_models[index - 1],
+ pending_ingestion_jobs=pending_ingestion_jobs,
)
next_current_soc_dict[site_name]["battery"] = battery_next_current_soc
next_current_soc_dict[site_name]["evse1"] = evse1_next_current_soc
next_current_soc_dict[site_name]["evse2"] = evse2_next_current_soc
next_current_soc_dict[site_name]["heating"] = heating_next_current_soc
+ # Reporters read the measurements submitted above. Wait until the server
+ # has actually ingested them instead of treating HTTP 202 as completion.
+ await wait_for_ingestion_jobs(client, pending_ingestion_jobs)
+
# Run reporter to log community site aggregate power consumption each scheduling step
- run_community_aggregate(
+ aggregate_reports_succeeded = run_community_aggregate(
sensors=sensors,
current_time=current_time,
step_end_time=step_end_time,
community_asset=community_asset,
site_names=site_names,
)
+ if not aggregate_reports_succeeded:
+ raise RuntimeError(
+ f"Aggregate report generation failed for simulation step {step_num}."
+ )
# Move to next simulation step
current_time = step_end_time
@@ -290,33 +325,29 @@ async def compute_site_schedules(
evse2_constraints = calculate_ev_soc_targets_and_constraints(
current_time_ts, evse2_capacity, evse2_has_trip
)
- if not evse1_constraints.get("unavailable"):
-
- # Create flex models for EVSE 1
- if evse1_next_current_soc is None:
- # Use initial SoC for first step
- evse1_current_soc = evse1_flex_model.get("soc_at_start", 12.0)
- else:
- evse1_current_soc = evse1_next_current_soc
- # Create dynamic flex model for EVSE 1 (Current SoC updated each step)
- evse1_scheduling_dynamic_flex_model = create_dynamic_storage_flex_model(
- current_soc=evse1_current_soc,
- constraints=evse1_constraints,
+ # Keep EVs in the model while they are away. Their zero consumption-capacity
+ # prevents charging, while soc-usage continues to account for driving.
+ if evse1_next_current_soc is None:
+ evse1_current_soc = evse1_flex_model.get(
+ "soc_at_start", EV_CONFIG["min_soc_percent"] * evse1_capacity
)
+ else:
+ evse1_current_soc = evse1_next_current_soc
+ evse1_scheduling_dynamic_flex_model = create_dynamic_storage_flex_model(
+ current_soc=evse1_current_soc,
+ constraints=evse1_constraints,
+ )
- if not evse2_constraints.get("unavailable"):
-
- # Create flex models for EVSE 2 (similar pattern, could be different car)
- if evse2_next_current_soc is None:
- # Use initial SoC for first step
- evse2_current_soc = evse2_flex_model.get("soc_at_start", 12.0)
- else:
- evse2_current_soc = evse2_next_current_soc
- # Create dynamic flex model for EVSE 2 (Current SoC updated each step)
- evse2_scheduling_dynamic_flex_model = create_dynamic_storage_flex_model(
- current_soc=evse2_current_soc,
- constraints=evse2_constraints,
+ if evse2_next_current_soc is None:
+ evse2_current_soc = evse2_flex_model.get(
+ "soc_at_start", EV_CONFIG["min_soc_percent"] * evse2_capacity
)
+ else:
+ evse2_current_soc = evse2_next_current_soc
+ evse2_scheduling_dynamic_flex_model = create_dynamic_storage_flex_model(
+ current_soc=evse2_current_soc,
+ constraints=evse2_constraints,
+ )
if heating_next_current_soc is None:
# Use initial SoC for first step
@@ -330,7 +361,10 @@ async def compute_site_schedules(
current_soc=heating_current_soc,
)
- # Start with the battery and PV flex models
+ # Curtailable PV is modeled as production which the scheduler may reduce,
+ # but never increase above the available-production forecast. Be careful
+ # when using its realized schedule in reports: a forecast underestimate can
+ # look like deliberate curtailment if the schedule is treated as a hard cap.
curtailable_pv_flex_model = {
"power-capacity": "12 kW",
"consumption-capacity": "0 kW",
@@ -341,42 +375,47 @@ async def compute_site_schedules(
"sensor": sensors[f"battery-power-{index}"]["id"],
**battery_scheduling_dynamic_flex_model,
},
- {
- "sensor": sensors[f"pv-power-{index}"][
- "id"
- ], # use power sensor to store realized data
- **curtailable_pv_flex_model,
- },
{
"sensor": sensors[f"heating-power-{index}"]["id"],
**heating_scheduling_dynamic_flex_model,
},
]
+ if PV_MODE == "curtailable":
+ final_flex_models.insert(
+ 1,
+ {
+ "sensor": sensors[f"pv-power-{index}"]["id"],
+ **curtailable_pv_flex_model,
+ },
+ )
+ elif PV_MODE != "inflexible":
+ raise ValueError(
+ f"Unsupported PV_MODE {PV_MODE!r}; choose 'inflexible' or 'curtailable'."
+ )
- # Conditionally add EVSE flex models if they are not on a trip
- if not evse1_constraints.get("unavailable"):
- final_flex_models.append(
+ final_flex_models.extend(
+ [
{
"sensor": sensors[f"evse1-power-{index}"]["id"],
**evse1_scheduling_dynamic_flex_model,
- }
- )
- else:
- print("EVSE 1 is on a trip, skipping scheduling.")
-
- if not evse2_constraints.get("unavailable"):
- final_flex_models.append(
+ },
{
"sensor": sensors[f"evse2-power-{index}"]["id"],
**evse2_scheduling_dynamic_flex_model,
- }
- )
- else:
- print("EVSE 2 is on a trip, skipping scheduling.")
+ },
+ ]
+ )
print("[FLEX-MODEL-DEBUG] === FLEX MODELS SENT TO SCHEDULER ===")
- for i, model in enumerate(final_flex_models):
- device_name = ["Battery", "PV", "Heating", "EVSE-1", "EVSE-2"][i]
+ device_names_by_sensor_id = {
+ sensors[f"battery-power-{index}"]["id"]: "Battery",
+ sensors[f"pv-power-{index}"]["id"]: "PV",
+ sensors[f"heating-power-{index}"]["id"]: "Heating",
+ sensors[f"evse1-power-{index}"]["id"]: "EVSE-1",
+ sensors[f"evse2-power-{index}"]["id"]: "EVSE-2",
+ }
+ for model in final_flex_models:
+ device_name = device_names_by_sensor_id[model["sensor"]]
print(f"[FLEX-MODEL] {device_name}: {model}")
print()
@@ -482,6 +521,7 @@ async def compute_site_measurements(
evse1_flex_model: dict,
evse2_flex_model: dict,
index: int,
+ pending_ingestion_jobs: list[str],
):
# Initialize power schedules
@@ -533,7 +573,9 @@ async def compute_site_measurements(
# Upload battery power measurements
battery_power_duration = timedelta(hours=SIMULATION_STEP_HOURS)
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"battery-power-{index}"]["id"],
start=current_time,
duration=battery_power_duration,
@@ -555,11 +597,15 @@ async def compute_site_measurements(
f"Failed to fetch PV raw power from sensor {sensors[f'pv-production-{index}']['id']}"
)
- pv_realized_power = [
- min(raw, scheduled) for raw, scheduled in zip(pv_raw_power, pv_scheduled_power)
- ]
+ # In curtailable mode this emulates a gateway which can lower PV output to
+ # the scheduled setpoint. It cannot produce more power than is available.
+ pv_realized_power = realize_pv_power(
+ pv_raw_power, pv_scheduled_power, pv_mode=PV_MODE
+ )
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"pv-power-{index}"][
"id"
], # use power sensor to store realized data
@@ -571,7 +617,9 @@ async def compute_site_measurements(
)
# Upload EVSE 1 power measurements
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"evse1-power-{index}"]["id"],
start=current_time,
duration=battery_power_duration,
@@ -581,7 +629,9 @@ async def compute_site_measurements(
)
# Upload EVSE 2 power measurements
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"evse2-power-{index}"]["id"],
start=current_time,
duration=battery_power_duration,
@@ -591,7 +641,9 @@ async def compute_site_measurements(
)
# Upload heating power measurements
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"heating-power-{index}"]["id"],
start=current_time,
duration=battery_power_duration,
@@ -614,7 +666,9 @@ async def compute_site_measurements(
)
+ pd.Timedelta(minutes=15)
)
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"building-consumption-{index}"]["id"],
start=building_data_step["event_start"].iloc[0],
duration=step_duration,
@@ -688,7 +742,9 @@ async def compute_site_measurements(
# Upload battery SoC measurements (FlexMeasures computed)
if battery_soc_values:
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"battery-soc-{index}"]["id"],
start=current_time,
duration=pd.Timedelta(hours=SIMULATION_STEP_HOURS).isoformat(),
@@ -702,7 +758,9 @@ async def compute_site_measurements(
# Upload EVSE 1 SoC measurements (FlexMeasures computed)
if evse1_soc_values:
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"evse1-soc-{index}"]["id"],
start=current_time,
duration=pd.Timedelta(hours=SIMULATION_STEP_HOURS).isoformat(),
@@ -716,7 +774,9 @@ async def compute_site_measurements(
# Upload EVSE 2 SoC measurements (FlexMeasures computed)
if evse2_soc_values:
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"evse2-soc-{index}"]["id"],
start=current_time,
duration=pd.Timedelta(hours=SIMULATION_STEP_HOURS).isoformat(),
@@ -729,7 +789,9 @@ async def compute_site_measurements(
)
# Upload heating SoC measurements (FlexMeasures computed)
if heating_soc_values:
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensors[f"heating-soc-{index}"]["id"],
start=current_time,
duration=pd.Timedelta(hours=SIMULATION_STEP_HOURS).isoformat(),
@@ -827,12 +889,23 @@ async def get_site_assets(
index: int,
):
"""Get all assets in a site's child building."""
+ community_asset_id = await find_top_level_asset_id(client, community_name)
+ community_asset = await client.get_asset(community_asset_id, parse_json_fields=True)
assets = await client.get_assets(
- fields=["id", "name", "attributes", "sensors"], parse_json_fields=True
+ root=community_asset_id,
+ fields=["id", "name", "attributes", "sensors", "parent_asset_id"],
+ parse_json_fields=True,
)
- assets_by_name = {a["name"]: a for a in assets}
+ assets_by_name: dict[str, dict] = {}
+ for asset in [community_asset, *assets]:
+ existing_asset = assets_by_name.get(asset["name"])
+ if existing_asset is not None and existing_asset["id"] != asset["id"]:
+ raise LookupError(
+ f"Asset name '{asset['name']}' is ambiguous in community "
+ f"'{community_name}'."
+ )
+ assets_by_name[asset["name"]] = asset
- community_asset = assets_by_name.get(community_name)
site_asset = assets_by_name.get(site_name)
battery_asset = assets_by_name.get(f"{battery_name} {index}")
evse1_asset = assets_by_name.get(f"{evse1_name} {index}")
@@ -895,14 +968,13 @@ async def map_site_sensors(
for sensor_key, asset_name, sensor_name in sensor_mappings:
sensor = await find_sensor_by_name_and_asset(
- client, sensor_name, asset_name, top_level_asset_id=top_level_asset_id
+ client,
+ sensor_name,
+ asset_name,
+ top_level_asset_id=top_level_asset_id,
+ allow_top_level_asset=asset_name == price_market_name,
)
- if sensor:
- sensors[sensor_key] = sensor
- else:
- raise LookupError(
- f"Could not find sensor '{sensor_name}' in asset '{asset_name}'"
- )
+ sensors[sensor_key] = sensor
return sensors
@@ -919,6 +991,7 @@ def run_community_aggregate(
community_power_sensor = x
break
# Run each site's aggregate reporter
+ all_reports_succeeded = True
for index, site_name in enumerate(site_names, start=1):
# Fill reporter parameters for each site
fill_reporter_params(
@@ -936,11 +1009,12 @@ def run_community_aggregate(
reporter_type="aggregate",
)
# Run AggregatorReporter
- run_report_cmd(
+ report_succeeded = run_report_cmd(
reporter_map={"name": "aggregate", "reporter": "AggregatorReporter"},
start=current_time.isoformat(),
end=step_end_time.isoformat(),
)
+ all_reports_succeeded = report_succeeded and all_reports_succeeded
fill_reporter_params(
input_sensors=[
@@ -953,8 +1027,9 @@ def run_community_aggregate(
reporter_type="aggregate",
)
# Run AggregatorReporter
- run_report_cmd(
+ community_report_succeeded = run_report_cmd(
reporter_map={"name": "aggregate", "reporter": "AggregatorReporter"},
start=current_time.isoformat(),
end=step_end_time.isoformat(),
)
+ return community_report_succeeded and all_reports_succeeded
diff --git a/examples/HEMS/utils/asset_utils.py b/examples/HEMS/utils/asset_utils.py
index d217545e..728876e9 100644
--- a/examples/HEMS/utils/asset_utils.py
+++ b/examples/HEMS/utils/asset_utils.py
@@ -1,6 +1,7 @@
import asyncio
import os
from pathlib import Path
+from typing import Any
import pandas as pd
from const import heating_name, price_market_name, pv_name, weather_station_name
@@ -10,34 +11,134 @@
BASE_DIR = Path(__file__).parent.parent
+async def post_sensor_data_and_track_ingestion(
+ client: FlexMeasuresClient,
+ pending_ingestion_jobs: list[str],
+ **kwargs: Any,
+) -> None:
+ """Post sensor data and remember asynchronous ingestion jobs."""
+ response, status = await client.post_sensor_data(**kwargs)
+
+ if status != 202:
+ return
+
+ # FlexMeasures 0.33 calls this field ``job_id``; newer servers use ``job``.
+ job_id = None
+ if isinstance(response, dict):
+ job_id = response.get("job") or response.get("job_id")
+ if not job_id:
+ raise RuntimeError(
+ "The server accepted sensor data for asynchronous ingestion "
+ "but did not return a job ID."
+ )
+ pending_ingestion_jobs.append(job_id)
+
+
+async def wait_for_ingestion_jobs(
+ client: FlexMeasuresClient, pending_ingestion_jobs: list[str]
+) -> None:
+ """Wait until all tracked sensor-data ingestion jobs have finished."""
+ if not pending_ingestion_jobs:
+ return
+
+ print(f"Waiting for {len(pending_ingestion_jobs)} ingestion job(s)...")
+ for job_id in pending_ingestion_jobs:
+ deadline = asyncio.get_running_loop().time() + client.polling_timeout
+ polling_step = 0
+
+ while True:
+ # FlexMeasures 0.33 returns HTTP 200 even while a job is in
+ # progress. Newer versions return 202, which client.request polls
+ # internally. Inspecting the status field supports both versions.
+ job, _ = await client.request(
+ uri=f"jobs/{job_id}",
+ method="GET",
+ )
+ job_status = (
+ str(job.get("status", "")).upper() if isinstance(job, dict) else ""
+ )
+ if job_status == "FINISHED":
+ break
+ if job_status not in {"QUEUED", "STARTED", "DEFERRED", "SCHEDULED"}:
+ raise RuntimeError(
+ f"Ingestion job {job_id} did not finish successfully: {job}"
+ )
+
+ polling_step += 1
+ if polling_step >= client.max_polling_steps:
+ raise ConnectionError(
+ f"Max polling steps reached while waiting for ingestion job "
+ f"{job_id}. Last status: {job_status}"
+ )
+
+ remaining = deadline - asyncio.get_running_loop().time()
+ if remaining <= 0:
+ raise ConnectionError(
+ f"Client polling timeout while waiting for ingestion job "
+ f"{job_id}. Last status: {job_status}"
+ )
+ sleep_interval = min(
+ client.polling_interval * (2 ** (polling_step - 1)), remaining
+ )
+ await asyncio.sleep(sleep_interval)
+ pending_ingestion_jobs.clear()
+
+
async def find_sensor_by_name_and_asset(
client: FlexMeasuresClient,
sensor_name: str,
asset_name: str,
top_level_asset_id: int | None = None,
+ allow_top_level_asset: bool = False,
):
- """Find a sensor by name within a specific asset."""
- assets = await client.get_assets(
- root=top_level_asset_id
- ) # first list those that are part of the community
- assets += await client.get_assets(
- parse_json_fields=True
- ) # then list all accessible assets
- target_asset = None
- for asset in assets:
- if asset["name"] == asset_name:
- target_asset = asset
- break
-
- if not target_asset:
- raise LookupError(f"Asset '{asset_name}' not found")
+ """Find one sensor in the community tree or an explicitly allowed root."""
+ if top_level_asset_id is None:
+ raise ValueError("top_level_asset_id is required for scoped sensor lookup")
+
+ community_asset = await client.get_asset(top_level_asset_id, parse_json_fields=True)
+ account_id = community_asset.get("account_id")
+ if not isinstance(account_id, int):
+ account = await client.get_account()
+ account_id = account["id"]
+ assets = [community_asset]
+ assets.extend(
+ await client.get_assets(root=top_level_asset_id, parse_json_fields=True)
+ )
+ if allow_top_level_asset:
+ assets.extend(
+ await client.get_assets(
+ account_id=account_id, depth=0, parse_json_fields=True
+ )
+ )
- sensors = await client.get_sensors(asset_id=target_asset["id"])
- for sensor in sensors:
- if sensor["name"] == sensor_name:
- return sensor
+ assets_by_id = {asset["id"]: asset for asset in assets}
+ matches = [
+ asset for asset in assets_by_id.values() if asset.get("name") == asset_name
+ ]
+ if not matches:
+ raise LookupError(f"Asset '{asset_name}' not found")
+ if len(matches) > 1:
+ raise LookupError(
+ f"Asset name '{asset_name}' is ambiguous in the allowed HEMS scope."
+ )
+ target_asset = matches[0]
- raise LookupError(f"Sensor '{sensor_name}' not found in asset '{asset_name}'")
+ sensors = await client.get_sensors(
+ asset_id=target_asset["id"], parse_json_fields=True
+ )
+ matches = [
+ sensor
+ for sensor in sensors
+ if sensor.get("name") == sensor_name
+ and sensor.get("generic_asset_id") == target_asset["id"]
+ ]
+ if not matches:
+ raise LookupError(f"Sensor '{sensor_name}' not found in asset '{asset_name}'")
+ if len(matches) > 1:
+ raise LookupError(
+ f"Sensor name '{sensor_name}' is ambiguous on asset '{asset_name}'."
+ )
+ return matches[0]
async def upload_csv_file_to_sensor(
@@ -45,32 +146,41 @@ async def upload_csv_file_to_sensor(
sensor_id: int,
file_path: str,
belief_time_measured_instantly: bool,
+ pending_ingestion_jobs: list[str],
):
- """Upload CSV file directly to a sensor using file upload."""
+ """Upload a CSV file and track asynchronous ingestion."""
try:
full_path = os.path.join(BASE_DIR, file_path)
- await client.post_sensor_data(
+ await post_sensor_data_and_track_ingestion(
+ client=client,
+ pending_ingestion_jobs=pending_ingestion_jobs,
sensor_id=sensor_id,
file_path=full_path,
belief_time_measured_instantly=belief_time_measured_instantly, # Set belief_time immediately after event ends
)
- print(f"Uploaded {file_path} to sensor {sensor_id}")
- return True
+ print(f"Submitted {file_path} to sensor {sensor_id}")
except Exception as e:
print(f"Failed to upload {file_path} to sensor {sensor_id}: {e}")
- return False
+ raise
async def find_top_level_asset_id(
client: FlexMeasuresClient,
name: str,
) -> int:
+ account = await client.get_account()
top_level_assets = await client.get_assets(
- depth=0, fields=["id", "name"], parse_json_fields=True
+ account_id=account["id"],
+ depth=0,
+ fields=["id", "name"],
+ parse_json_fields=True,
)
- for asset in top_level_assets:
- if asset["name"] == name:
- return asset["id"]
+ matches = [asset for asset in top_level_assets if asset["name"] == name]
+ if len(matches) != 1:
+ raise LookupError(
+ f"Expected one top-level asset named '{name}', found {len(matches)}."
+ )
+ return matches[0]["id"]
async def find_sensors_by_asset(
@@ -86,14 +196,14 @@ async def find_sensors_by_asset(
sensors = {}
for key, sensor_name, asset_name in sensor_mappings:
sensor = await find_sensor_by_name_and_asset(
- client, sensor_name, asset_name, top_level_asset_id
+ client,
+ sensor_name,
+ asset_name,
+ top_level_asset_id,
+ allow_top_level_asset=asset_name
+ in {price_market_name, weather_station_name},
)
- if sensor:
- sensors[key] = sensor
- else:
- raise LookupError(
- f"Could not find sensor '{sensor_name}' in asset '{asset_name}'"
- )
+ sensors[key] = sensor
return sensors
@@ -102,6 +212,7 @@ async def upload_data_for_first_two_weeks(
):
"""Upload historical data for the first two weeks."""
print("Uploading data for first two weeks...")
+ pending_ingestion_jobs: list[str] = []
for i, site_name in enumerate(site_names, start=1):
# Find all required sensors
@@ -135,88 +246,75 @@ async def upload_data_for_first_two_weeks(
2:
] # Remove site power capacity and price datafiles to not fill them more than once
for file_path, sensor_key, belief_time_measured_instantly in data_files:
- if sensor_key not in sensors:
- print(f"Skipping {file_path} - sensor not found")
- continue
-
print(f"Processing {file_path}...")
# Upload CSV file directly
- success = await upload_csv_file_to_sensor(
+ await upload_csv_file_to_sensor(
client=client,
sensor_id=sensors[sensor_key]["id"],
file_path=file_path,
belief_time_measured_instantly=belief_time_measured_instantly,
+ pending_ingestion_jobs=pending_ingestion_jobs,
)
- if success:
- print(f"Successfully uploaded {sensor_key} data")
- else:
- print(f"Failed to upload {sensor_key} data")
+ print(f"Submitted {sensor_key} data for ingestion")
+
+ # File uploads may only have been accepted (HTTP 202), not processed yet.
+ # Forecasting must not start until all historical data is available.
+ await wait_for_ingestion_jobs(client, pending_ingestion_jobs)
return True
-async def cleanup_existing_assets(
- client: FlexMeasuresClient, account_id: int, site_names: list[str]
-):
- """Clean up existing HEMS assets to avoid naming conflicts."""
- print("Cleaning up existing assets...")
-
- for site_name in site_names:
- # Asset names to clean up
- asset_names_to_clean = [
- site_name, # Deleting this asset also deletes child assets (battery, PV, EVSEs)
- weather_station_name,
- price_market_name,
- ]
+async def delete_hems_assets(
+ client: FlexMeasuresClient,
+ account_id: int,
+ community_name: str,
+ confirm_first: bool = True,
+) -> int:
+ """Delete the top-level assets belonging to this HEMS example.
+
+ Deleting the community asset also deletes all child assets, sensors, and data.
+ The price market and weather station are separate top-level assets, so they
+ are deleted explicitly.
+ """
+ asset_names_to_delete = {
+ community_name,
+ weather_station_name,
+ price_market_name,
+ }
+ top_level_assets = await client.get_assets(
+ depth=0,
+ fields=["id", "name", "account_id"],
+ parse_json_fields=False,
+ )
+ assets_to_delete = [
+ asset
+ for asset in top_level_assets
+ if asset["name"] in asset_names_to_delete
+ and asset.get("account_id") == account_id
+ ]
- try:
- # Get all existing assets
- assets = await client.get_assets(parse_json_fields=True)
-
- # Find and delete assets that match our names
- deleted_count = 0
- for asset in assets:
- if asset["name"] in asset_names_to_clean:
- print(
- f"Deleting existing asset: {asset['name']} (ID: {asset['id']})"
- )
- try:
- if asset.get("account_id") != account_id:
- print(
- f"Warning: Asset {asset['name']} (ID: {asset['id']}) does not belong to the current account."
- )
- raise
- await client.delete_asset(
- asset_id=asset["id"], confirm_first=False
- )
- deleted_count += 1
- except Exception as delete_error:
- # Check if it's a 404 error (asset not found)
- if "404" in str(delete_error) or "NOT FOUND" in str(
- delete_error
- ):
- print(
- f"Asset {asset['name']} (ID: {asset['id']}) no longer exists, skipping..."
- )
- else:
- print(
- f"Warning: Could not delete asset {asset['name']}: {delete_error}"
- )
- # Continue with other assets
-
- if deleted_count > 0:
- print(f"Cleaned up {deleted_count} existing assets")
- else:
- print("No existing assets to clean up")
-
- # Wait a moment for deletions to complete
- await asyncio.sleep(1)
-
- except Exception as e:
- print(f"Warning: Error during cleanup: {e}")
- print("Continuing with setup...")
+ if not assets_to_delete:
+ print("No HEMS assets found in the current account.")
+ return 0
+
+ print("The following top-level HEMS assets will be deleted:")
+ for asset in assets_to_delete:
+ print(f"- {asset['name']} (ID: {asset['id']})")
+ print("Their child assets, sensors, and time-series data will also be deleted.")
+
+ if confirm_first:
+ answer = input("Permanently delete these assets and all their data? [yN] ")
+ if answer.lower() not in ["y", "yes"]:
+ print("Aborting ...")
+ return 0
+
+ for asset in assets_to_delete:
+ await client.delete_asset(asset_id=asset["id"], confirm_first=False)
+
+ print(f"Deleted {len(assets_to_delete)} top-level HEMS assets.")
+ return len(assets_to_delete)
def load_and_align_csv_data(
@@ -238,20 +336,3 @@ def load_and_align_csv_data(
print(f"Aligned {len(df)} records from {file_path}")
return aligned_df
-
-
-def get_first_asset_by_name(
- assets: list[dict], name: str, account_id: int | None = 0
-) -> dict | None:
- """
- :param assets: List of dictionaries describing assets, each with at least a "name".
- :param name: The asset name to find the first occurrence for.
- :param account_id: Optionally, filter by account_id (a positive integer, or None for a public account).
- To use this filter, each dictionary in `assets` should contain the "account_id", too.
- NB the 0 default is used to signal the argument is missing (real IDs are strictly positive).
- """
- for asset in assets:
- if asset["name"] == name:
- if account_id != 0 and asset["account_id"] != account_id:
- continue
- return asset
diff --git a/examples/HEMS/utils/ev_utils.py b/examples/HEMS/utils/ev_utils.py
index 6fd42431..6cfab8c9 100644
--- a/examples/HEMS/utils/ev_utils.py
+++ b/examples/HEMS/utils/ev_utils.py
@@ -32,122 +32,125 @@ def calculate_ev_soc_targets_and_constraints(
"""
Calculate dynamic SoC targets and availability constraints for EV charging.
- Returns a dict with:
- - soc_targets: List of target SoC values with datetimes
- - soc_minima: List of minimum SoC constraints during unavailable periods
- - consumption_capacity: Availability windows (0 during unavailable periods)
+ Build SoC and availability constraints for the next 24 hours.
+
+ Driving is represented by two explicit SoC-usage periods: one after departure
+ and one before returning home. Keeping these separate from the SoC minima makes
+ rolling rescheduling preserve the remaining part of a trip.
"""
if capacity_kwh is None:
capacity_kwh = EV_CONFIG["default_capacity_kwh"]
print(
- f"[EV-CALC] Calculating EV constraints for {current_time.strftime('%Y-%m-%d %H:%M')}"
+ "[EV-CALC] Calculating EV constraints for "
+ f"{current_time.strftime('%Y-%m-%d %H:%M')}"
)
print(f" [CAPACITY] Battery capacity: {capacity_kwh} kWh")
- needs_charging, departure_time_str, return_time_str, target_soc_percent = (
- get_day_pattern(current_time)
- )
-
- target_soc_kwh = (target_soc_percent / 100.0) * capacity_kwh
min_soc_kwh = EV_CONFIG["min_soc_percent"] * capacity_kwh
-
- print(f" [TARGET] Target SoC: {target_soc_percent}% = {target_soc_kwh:.1f} kWh")
print(
- f" [MINIMUM] Minimum SoC: {EV_CONFIG['min_soc_percent']*100:.0f}% = {min_soc_kwh:.1f} kWh"
+ f" [MINIMUM] Minimum SoC: {EV_CONFIG['min_soc_percent']*100:.0f}% "
+ f"= {min_soc_kwh:.1f} kWh"
)
constraints = {
- "soc_targets": [],
"soc_minima": [],
+ "soc_usage": [],
"consumption_capacity": [],
}
+ usage_segments = []
+ planning_end = current_time + pd.Timedelta(hours=24)
+ commute_duration = pd.Timedelta(hours=EV_CONFIG["one_way_commute_duration_hours"])
+
+ def add_usage_segment(start: pd.Timestamp, end: pd.Timestamp) -> None:
+ """Add the part of a driving period that remains in the planning window."""
+ start = max(start, current_time)
+ end = min(end, planning_end)
+ if start < end:
+ usage_segments.append(
+ {
+ "start": start.isoformat(),
+ "end": end.isoformat(),
+ "value": f'{EV_CONFIG["driving_consumption_kwh_per_hour"]} kW',
+ }
+ )
- if needs_charging and departure_time_str and return_time_str:
- # Work day - need to be charged by departure time
- print(
- f" [WORK-DAY] Departure at {departure_time_str}, return at {return_time_str}"
- )
- departure_hour, departure_minute = map(int, departure_time_str.split(":"))
- return_hour, return_minute = map(int, return_time_str.split(":"))
-
- # Target: charged to 80% by departure time
- departure_datetime = current_time.replace(
- hour=departure_hour, minute=departure_minute, second=0, microsecond=0
+ # Include both the remainder of today's pattern and tomorrow's pattern. This
+ # matters when replanning while an EV is away or shortly before tomorrow's
+ # departure.
+ for day_offset in (0, 1):
+ day = current_time.normalize() + pd.Timedelta(days=day_offset)
+ needs_charging, departure_time_str, return_time_str, target_soc_percent = (
+ get_day_pattern(day)
)
-
- # If departure is already past today, target tomorrow
- if departure_datetime <= current_time:
- departure_datetime += pd.Timedelta(days=1)
- print(
- f" [SCHEDULE] Departure time adjusted to next day: {departure_datetime.strftime('%Y-%m-%d %H:%M')}"
+ target_soc_kwh = target_soc_percent / 100 * capacity_kwh
+
+ if needs_charging and departure_time_str and return_time_str:
+ departure_hour, departure_minute = map(int, departure_time_str.split(":"))
+ return_hour, return_minute = map(int, return_time_str.split(":"))
+ departure_datetime = day.replace(
+ hour=departure_hour,
+ minute=departure_minute,
+ second=0,
+ microsecond=0,
)
- else:
- print(
- f" [SCHEDULE] Departure time: {departure_datetime.strftime('%Y-%m-%d %H:%M')}"
+ return_datetime = day.replace(
+ hour=return_hour,
+ minute=return_minute,
+ second=0,
+ microsecond=0,
)
- constraints["soc_minima"] = [
- {
- "datetime": departure_datetime.isoformat(),
- "value": f"{target_soc_kwh} kWh",
- }
- ]
- print(
- f" [MINIMUM-SET] SoC minimum set: {target_soc_kwh:.1f} kWh by {departure_datetime.strftime('%H:%M')}"
- )
-
- # Unavailable period: departure time to return time (same day as departure)
- return_datetime = departure_datetime.replace(
- hour=return_hour, minute=return_minute
- )
- unavailable_duration = return_datetime - departure_datetime
-
- # Check if we are currently in the unavailable period
- if current_time >= departure_datetime and current_time <= return_datetime:
- print(" [UNAVAILABLE] Currently in unavailable period")
- return_datetime += pd.Timedelta(days=1)
print(
- f" [UNAVAILABLE] Period: {departure_datetime.strftime('%H:%M')} - {return_datetime.strftime('%H:%M')} ({unavailable_duration})"
+ f" [WORK-DAY] {day.date()}: {target_soc_percent}% "
+ f"({target_soc_kwh:.1f} kWh) by {departure_time_str}; "
+ f"return at {return_time_str}"
)
- constraints["unavailable"] = True
+ if current_time <= departure_datetime <= planning_end:
+ constraints["soc_minima"].append(
+ {
+ "datetime": departure_datetime.isoformat(),
+ "value": f"{target_soc_kwh} kWh",
+ }
+ )
+
+ unavailable_start = max(departure_datetime, current_time)
+ unavailable_end = min(return_datetime, planning_end)
+ if unavailable_start < unavailable_end:
+ constraints["consumption_capacity"].append(
+ {
+ "start": unavailable_start.isoformat(),
+ "end": unavailable_end.isoformat(),
+ "value": "0 kW",
+ }
+ )
+ constraints["soc_minima"].append(
+ {
+ "start": unavailable_start.isoformat(),
+ "end": unavailable_end.isoformat(),
+ "value": f"{min_soc_kwh} kWh",
+ }
+ )
+
+ add_usage_segment(departure_datetime, departure_datetime + commute_duration)
+ add_usage_segment(return_datetime - commute_duration, return_datetime)
else:
+ end_of_day = day + pd.Timedelta(days=1)
+ if current_time <= end_of_day <= planning_end:
+ constraints["soc_minima"].append(
+ {
+ "datetime": end_of_day.isoformat(),
+ "value": f"{target_soc_kwh} kWh",
+ }
+ )
print(
- f" [UNAVAILABLE] Period: {departure_datetime.strftime('%H:%M')} - {return_datetime.strftime('%H:%M')} ({unavailable_duration})"
+ f" [FLEXIBLE-DAY] {day.date()}: maintain "
+ f"{target_soc_percent}% ({target_soc_kwh:.1f} kWh)"
)
- # Disable charging during unavailable period by setting consumption capacity to 0
- constraints["consumption_capacity"] = [
- {
- "start": departure_datetime.isoformat(),
- "end": return_datetime.isoformat(),
- "value": "0 kW",
- }
- ]
-
- # Extend minimum SoC constraint during unavailable period
- constraints["soc_minima"].append(
- {
- "start": departure_datetime.isoformat(),
- "end": return_datetime.isoformat(),
- "value": f"{min_soc_kwh} kWh",
- }
- )
- print(" [DISABLED] Charging disabled during unavailable period (0 kW)")
- print(
- f" [MIN-SOC] Minimum SoC maintained: {min_soc_kwh:.1f} kWh during unavailable period"
- )
- else:
- # Free day - just maintain minimum SoC by end of planning horizon
- print(f" [FREE-DAY] Flexible charging to {target_soc_percent}%")
- end_of_day = current_time.replace(hour=23, minute=59, second=59, microsecond=0)
- constraints["soc_minima"] = [
- {"datetime": end_of_day.isoformat(), "value": f"{target_soc_kwh} kWh"}
- ]
- print(
- f" [FLEXIBLE] Minimum: {target_soc_kwh:.1f} kWh by end of day ({end_of_day.strftime('%H:%M')})"
- )
- print(" [AVAILABLE] No availability restrictions - can charge anytime")
+ if usage_segments:
+ # soc-usage is a list of components; this component is a time series.
+ constraints["soc_usage"] = [usage_segments]
# Handle random trips - reduce SoC randomly to simulate unplanned usage
if has_random_trip:
@@ -158,11 +161,15 @@ def calculate_ev_soc_targets_and_constraints(
trip_consumption_kwh = trip_consumption_percent * capacity_kwh
print(
- f" [CONSUMPTION] Trip consumption: {trip_consumption_percent*100:.1f}% = {trip_consumption_kwh:.1f} kWh"
+ " [CONSUMPTION] Trip consumption: "
+ f"{trip_consumption_percent*100:.1f}% = "
+ f"{trip_consumption_kwh:.1f} kWh"
)
- # Adjust minima to account for trip consumption
+ # Adjust point-in-time targets to account for the unexpected trip.
for minimum in constraints["soc_minima"]:
+ if "datetime" not in minimum:
+ continue
original_minimum_kwh = float(minimum["value"].split()[0])
# Ensure we charge enough to cover the trip consumption
adjusted_minimum = min(
@@ -170,7 +177,9 @@ def calculate_ev_soc_targets_and_constraints(
)
minimum["value"] = f"{adjusted_minimum} kWh"
print(
- f" [ADJUSTED] Minimum: {original_minimum_kwh:.1f} kWh -> {adjusted_minimum:.1f} kWh (+{trip_consumption_kwh:.1f} kWh for trip)"
+ f" [ADJUSTED] Minimum: {original_minimum_kwh:.1f} kWh "
+ f"-> {adjusted_minimum:.1f} kWh "
+ f"(+{trip_consumption_kwh:.1f} kWh for trip)"
)
print(" [SUMMARY] Final constraints:")
@@ -187,15 +196,29 @@ def calculate_ev_soc_targets_and_constraints(
start_dt = pd.to_datetime(minima["start"])
end_dt = pd.to_datetime(minima["end"])
print(
- f" [MINIMUM] {minima['value']} from {start_dt.strftime('%H:%M')} to {end_dt.strftime('%H:%M')}"
+ f" [MINIMUM] {minima['value']} from "
+ f"{start_dt.strftime('%H:%M')} to {end_dt.strftime('%H:%M')}"
)
if constraints["consumption_capacity"]:
for capacity in constraints["consumption_capacity"]:
start_dt = pd.to_datetime(capacity["start"])
end_dt = pd.to_datetime(capacity["end"])
print(
- f" [DISABLED] Charging: {start_dt.strftime('%H:%M')} to {end_dt.strftime('%H:%M')} ({capacity['value']})"
+ f" [DISABLED] Charging: {start_dt.strftime('%H:%M')} "
+ f"to {end_dt.strftime('%H:%M')} ({capacity['value']})"
)
+ if usage_segments:
+ total_driving_hours = sum(
+ (
+ pd.Timestamp(segment["end"]) - pd.Timestamp(segment["start"])
+ ).total_seconds()
+ / 3600
+ for segment in usage_segments
+ )
+ print(
+ f" [DRIVING] {total_driving_hours:.1f} h at "
+ f'{EV_CONFIG["driving_consumption_kwh_per_hour"]:.1f} kW'
+ )
print()
diff --git a/examples/HEMS/utils/scheduling_utils.py b/examples/HEMS/utils/scheduling_utils.py
index 4b5fcef3..bec46b64 100644
--- a/examples/HEMS/utils/scheduling_utils.py
+++ b/examples/HEMS/utils/scheduling_utils.py
@@ -1,7 +1,5 @@
from typing import Any
-from const import EV_CONFIG
-
def create_dynamic_storage_flex_model(
current_soc: float,
@@ -23,13 +21,9 @@ def create_dynamic_storage_flex_model(
# Add dynamic constraints if provided
if constraints:
if constraints.get("soc_minima"):
- # todo: here we remove the last soc_minima constraint and set it up as a soc_usage component instead
- # this is a workaround; we should define the SoC drop during the trip as a soc_usage component straightaway
- soc_usage = constraints["soc_minima"].pop(-1)
flex_model["soc-minima"] = constraints["soc_minima"]
- soc_usage["value"] = f'{EV_CONFIG["driving_consumption_kwh_per_hour"]} kW'
- # add soc_usage as a component (soc-usage supports a list of usage components)
- flex_model["soc-usage"] = [[soc_usage]]
+ if constraints.get("soc_usage"):
+ flex_model["soc-usage"] = constraints["soc_usage"]
if constraints.get("consumption_capacity"):
flex_model["consumption-capacity"] = constraints["consumption_capacity"]
diff --git a/examples/HEMS/utils/workflow_utils.py b/examples/HEMS/utils/workflow_utils.py
new file mode 100644
index 00000000..a863f9c7
--- /dev/null
+++ b/examples/HEMS/utils/workflow_utils.py
@@ -0,0 +1,284 @@
+from __future__ import annotations
+
+import json
+
+from const import price_market_name, weather_station_name
+
+from flexmeasures_client import FlexMeasuresClient
+
+WORKFLOW_ATTRIBUTE = "hems_tutorial"
+#: Bump whenever the tutorial changes the asset/sensor structure, the site
+#: flex-context or the dashboard, so that setups created by an older version
+#: are upgraded in place instead of silently keeping the old structure.
+WORKFLOW_VERSION = 2
+
+ASSET_SETUP_PHASE = "asset-setup"
+DATA_UPLOAD_PHASE = "historical-data-upload"
+FORECASTING_PHASE = "forecasting"
+SCHEDULING_PHASE = "scheduling"
+REPORTING_PHASE = "reporting"
+
+PHASE_LABELS = {
+ ASSET_SETUP_PHASE: "Asset setup",
+ DATA_UPLOAD_PHASE: "Historical data upload",
+ FORECASTING_PHASE: "Forecast generation",
+ SCHEDULING_PHASE: "Scheduling simulation",
+ REPORTING_PHASE: "Report generation",
+}
+
+
+def get_workflow_state(community_asset: dict) -> dict | None:
+ """Read a valid HEMS workflow marker from a community asset."""
+ attributes = community_asset.get("attributes", {})
+ if isinstance(attributes, str):
+ try:
+ attributes = json.loads(attributes)
+ except json.JSONDecodeError:
+ return None
+ if not isinstance(attributes, dict):
+ return None
+
+ state = attributes.get(WORKFLOW_ATTRIBUTE)
+ if not isinstance(state, dict):
+ return None
+ # Older markers stay valid: they are upgraded in place, which preserves both
+ # the existing asset and sensor IDs and any interrupted-wipe status. Newer
+ # markers are rejected, since this script cannot know what they describe.
+ version = state.get("workflow-version")
+ if isinstance(version, bool) or not isinstance(version, int):
+ return None
+ if not 1 <= version <= WORKFLOW_VERSION:
+ return None
+ if not isinstance(state.get("completed-phases"), list):
+ return None
+ if not isinstance(state.get("sensor-ids"), list):
+ return None
+ if not all(isinstance(sensor_id, int) for sensor_id in state["sensor-ids"]):
+ return None
+ if not isinstance(state.get("top-level-asset-ids"), list):
+ return None
+ if not all(isinstance(asset_id, int) for asset_id in state["top-level-asset-ids"]):
+ return None
+ if "site-names" in state and not (
+ isinstance(state["site-names"], list)
+ and all(isinstance(name, str) for name in state["site-names"])
+ ):
+ return None
+ return state
+
+
+async def get_site_assets(
+ client: FlexMeasuresClient,
+ community_asset_id: int,
+ account_id: int,
+) -> list[dict]:
+ """Return the community's direct child sites in stable creation order."""
+ assets = await client.get_assets(
+ account_id=account_id,
+ fields=["id", "name", "account_id", "parent_asset_id"],
+ parse_json_fields=False,
+ )
+ return sorted(
+ (
+ asset
+ for asset in assets
+ if asset.get("account_id") == account_id
+ and asset.get("parent_asset_id") == community_asset_id
+ ),
+ key=lambda asset: asset["id"],
+ )
+
+
+async def rename_site_assets(
+ client: FlexMeasuresClient,
+ site_assets: list[dict],
+ site_names: list[str],
+) -> None:
+ """Rename existing sites in stable order while preserving their IDs."""
+ if len(site_assets) > len(site_names):
+ raise ValueError(
+ f"Cannot map {len(site_assets)} existing sites to "
+ f"{len(site_names)} configured site names."
+ )
+ for site_asset, site_name in zip(site_assets, site_names):
+ if site_asset["name"] != site_name:
+ await client.update_asset(
+ asset_id=site_asset["id"],
+ updates={"name": site_name},
+ parse_json_fields=False,
+ )
+
+
+async def save_workflow_state(
+ client: FlexMeasuresClient,
+ community_asset_id: int,
+ state: dict,
+) -> dict:
+ """Save workflow state without replacing unrelated asset attributes."""
+ community_asset = await client.get_asset(
+ asset_id=community_asset_id, parse_json_fields=True
+ )
+ attributes = community_asset.get("attributes", {})
+ if not isinstance(attributes, dict):
+ attributes = {}
+ attributes[WORKFLOW_ATTRIBUTE] = state
+ await client.update_asset(
+ asset_id=community_asset_id,
+ updates={"attributes": attributes},
+ parse_json_fields=False,
+ )
+ return state
+
+
+async def collect_hems_structure_ids(
+ client: FlexMeasuresClient,
+ community_asset: dict,
+ account_id: int,
+) -> tuple[list[int], list[int]]:
+ """Collect the asset and sensor IDs that belong to this HEMS tutorial."""
+ all_assets = await client.get_assets(
+ fields=["id", "name", "account_id", "parent_asset_id"],
+ parse_json_fields=False,
+ )
+ top_level_asset_ids = [community_asset["id"]]
+ for asset_name in (price_market_name, weather_station_name):
+ matching_assets = [
+ asset
+ for asset in all_assets
+ if asset.get("name") == asset_name
+ and asset.get("account_id") == account_id
+ and asset.get("parent_asset_id") is None
+ ]
+ if len(matching_assets) != 1:
+ raise LookupError(
+ f"Expected one top-level HEMS asset named '{asset_name}' in "
+ f"account {account_id}, found {len(matching_assets)}."
+ )
+ top_level_asset_ids.append(matching_assets[0]["id"])
+
+ hems_asset_ids: set[int] = set()
+ for root_id in top_level_asset_ids:
+ hems_asset_ids.add(root_id)
+ descendants = await client.get_assets(
+ root=root_id,
+ fields=["id"],
+ parse_json_fields=False,
+ )
+ hems_asset_ids.update(asset["id"] for asset in descendants)
+
+ sensor_ids: set[int] = set()
+ for asset_id in sorted(hems_asset_ids):
+ sensors = await client.get_sensors(
+ asset_id=asset_id,
+ parse_json_fields=False,
+ )
+ sensor_ids.update(sensor["id"] for sensor in sensors)
+
+ return sorted(top_level_asset_ids), sorted(sensor_ids)
+
+
+async def initialize_workflow_state(
+ client: FlexMeasuresClient,
+ community_asset: dict,
+ account_id: int,
+ site_names: list[str],
+ status: str = "ready",
+) -> dict:
+ """Create the workflow marker after the complete asset structure exists."""
+ top_level_asset_ids, sensor_ids = await collect_hems_structure_ids(
+ client=client,
+ community_asset=community_asset,
+ account_id=account_id,
+ )
+ state = {
+ "workflow-version": WORKFLOW_VERSION,
+ "status": status,
+ "completed-phases": [ASSET_SETUP_PHASE],
+ "top-level-asset-ids": top_level_asset_ids,
+ "sensor-ids": sensor_ids,
+ "site-names": list(site_names),
+ }
+ return await save_workflow_state(client, community_asset["id"], state)
+
+
+def state_needs_upgrade(state: dict) -> bool:
+ """Tell whether a valid marker was written by an older tutorial version."""
+ return state["workflow-version"] < WORKFLOW_VERSION
+
+
+async def upgrade_workflow_state(
+ client: FlexMeasuresClient,
+ community_asset: dict,
+ account_id: int,
+ state: dict,
+ phases_to_rerun: tuple[str, ...] = (),
+) -> dict:
+ """Re-record an upgraded setup without discarding unaffected progress.
+
+ Call this only after the idempotent asset setup has run again, so that the
+ structure on the server already matches the current tutorial version. The
+ asset and sensor IDs are re-collected as they are now, which is what makes a
+ later data wipe cover sensors that the upgrade added.
+
+ Everything the upgrade does not invalidate is preserved: the recorded site
+ names, the ``status`` (so an interrupted wipe stays recoverable) and every
+ completed phase except those the caller asks to re-run.
+ """
+ top_level_asset_ids, sensor_ids = await collect_hems_structure_ids(
+ client=client,
+ community_asset=community_asset,
+ account_id=account_id,
+ )
+ completed_phases = [
+ phase for phase in state["completed-phases"] if phase not in phases_to_rerun
+ ]
+ if ASSET_SETUP_PHASE not in completed_phases:
+ completed_phases.insert(0, ASSET_SETUP_PHASE)
+ state = {
+ **state,
+ "workflow-version": WORKFLOW_VERSION,
+ "completed-phases": completed_phases,
+ "top-level-asset-ids": top_level_asset_ids,
+ "sensor-ids": sensor_ids,
+ }
+ return await save_workflow_state(client, community_asset["id"], state)
+
+
+def phase_is_complete(state: dict, phase: str) -> bool:
+ return phase in state["completed-phases"]
+
+
+async def mark_phase_complete(
+ client: FlexMeasuresClient,
+ community_asset_id: int,
+ state: dict,
+ phase: str,
+) -> dict:
+ """Mark one successfully finished phase as complete."""
+ completed_phases = list(state["completed-phases"])
+ if phase not in completed_phases:
+ completed_phases.append(phase)
+ state = {**state, "status": "ready", "completed-phases": completed_phases}
+ return await save_workflow_state(client, community_asset_id, state)
+
+
+async def wipe_hems_sensor_data(
+ client: FlexMeasuresClient,
+ community_asset_id: int,
+ state: dict,
+) -> dict:
+ """Delete HEMS time-series data and reset all data phase markers."""
+ state = {
+ **state,
+ "status": "wiping",
+ "completed-phases": [ASSET_SETUP_PHASE],
+ }
+ await save_workflow_state(client, community_asset_id, state)
+
+ sensor_ids = state["sensor-ids"]
+ print(f"Deleting time-series data from {len(sensor_ids)} HEMS sensors...")
+ for sensor_id in sensor_ids:
+ await client.delete_sensor_data(sensor_id, confirm_first=False)
+
+ state = {**state, "status": "ready"}
+ return await save_workflow_state(client, community_asset_id, state)
diff --git a/src/flexmeasures_client/client.py b/src/flexmeasures_client/client.py
index 4d552278..ffa55f2f 100644
--- a/src/flexmeasures_client/client.py
+++ b/src/flexmeasures_client/client.py
@@ -564,7 +564,7 @@ async def _post_sensor_data_json(
if prior:
json_payload["prior"] = pd.Timestamp(prior).isoformat()
- _response, status = await self.request(
+ response, status = await self.request(
uri=f"sensors/{sensor_id}/data",
json_payload=json_payload,
minimum_server_version="0.28.0",
@@ -572,6 +572,7 @@ async def _post_sensor_data_json(
)
check_for_status(status, 200)
self.logger.info("Sensor data sent successfully via JSON.")
+ return response, status
async def _post_sensor_data_file(
self,
@@ -884,7 +885,7 @@ async def get_assets(
if account_id and isinstance(account_id, int):
uri += f"&account_id={account_id}"
- if root or depth or fields:
+ if root is not None or depth is not None or fields:
if self.server_version is not None and Version(
self.server_version
) < Version("0.31.0"):
@@ -892,9 +893,9 @@ async def get_assets(
"get_assets(): The 'root', 'depth' and 'fields' parameters require FlexMeasures server version 0.31.0 or above. "
f"These parameters will be ignored for server version {self.server_version}."
)
- if root and isinstance(root, int):
+ if root is not None and isinstance(root, int):
uri += f"&root={root}"
- if depth and isinstance(depth, int):
+ if depth is not None and isinstance(depth, int):
uri += f"&depth={depth}"
if fields and isinstance(fields, list):
fields_str = "|".join(fields)
@@ -1404,6 +1405,51 @@ async def delete_sensor(self, sensor_id: int, confirm_first: bool = True):
_, status = await self.request(uri=uri, method="DELETE")
check_for_status(status, 204)
+ async def delete_sensor_data(
+ self,
+ sensor_id: int,
+ confirm_first: bool = True,
+ source: int | None = None,
+ start: str | datetime | None = None,
+ until: str | datetime | None = None,
+ ) -> None:
+ """Delete sensor data while preserving the sensor itself.
+
+ Optionally limit deletion to one source and/or an event-time range.
+ """
+ if confirm_first:
+ deletion_scope = (
+ "all data"
+ if source is None and start is None and until is None
+ else "matching data"
+ )
+ answer = input(
+ f"Permanently delete {deletion_scope} from sensor "
+ f"{sensor_id}? [y/N] "
+ )
+ if answer.lower() not in ["y", "yes"]:
+ print("Aborting ...")
+ return
+ json_payload = {}
+ if source is not None:
+ json_payload["source"] = source
+ if start is not None:
+ json_payload["start"] = pd.Timestamp(start).isoformat()
+ if until is not None:
+ json_payload["until"] = pd.Timestamp(until).isoformat()
+
+ _, status = await self.request(
+ uri=f"sensors/{sensor_id}/data",
+ json_payload=json_payload,
+ method="DELETE",
+ minimum_server_version="0.33.0",
+ minimum_server_version_msg=(
+ "Deleting sensor data without deleting the sensor requires "
+ "FlexMeasures server v0.33.0 or above."
+ ),
+ )
+ check_for_status(status, 204)
+
async def trigger_schedule(
self,
start: str | datetime,
diff --git a/tests/client/test_asset.py b/tests/client/test_asset.py
index edb52ea4..c5ae7b9f 100644
--- a/tests/client/test_asset.py
+++ b/tests/client/test_asset.py
@@ -204,6 +204,25 @@ async def test_get_assets_root_depth_fields_new_server():
await client.close()
+@pytest.mark.asyncio
+async def test_get_assets_includes_zero_depth():
+ """depth=0 is meaningful and must be included in the URL."""
+ with aioresponses() as m:
+ client = FlexMeasuresClient(email="test@test.test", password="test")
+ client.access_token = "test-token"
+ client.server_version = "0.31.0"
+ m.get(
+ "http://localhost:5000/api/v3_0/assets?all_accessible=False&sort_by=id&sort_dir=asc&include_public=False&depth=0",
+ status=200,
+ payload=[{"id": 1, "name": "top-level"}],
+ )
+
+ assets = await client.get_assets(depth=0, parse_json_fields=False)
+
+ assert assets == [{"id": 1, "name": "top-level"}]
+ await client.close()
+
+
@pytest.mark.asyncio
async def test_get_assets_root_old_server_warning(caplog):
"""root param on server < 0.31.0 emits warning."""
diff --git a/tests/client/test_sensor.py b/tests/client/test_sensor.py
index 14fbafc1..1fd39fc9 100644
--- a/tests/client/test_sensor.py
+++ b/tests/client/test_sensor.py
@@ -2,7 +2,7 @@
import os
import re
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
from urllib.parse import unquote
import pandas as pd
@@ -327,6 +327,104 @@ async def test_delete_sensor_confirm_no():
await client.close()
+@pytest.mark.asyncio
+async def test_delete_sensor_data_preserves_sensor():
+ with aioresponses() as m:
+ client = FlexMeasuresClient(email="test@test.test", password="test")
+ client.access_token = "test-token"
+ client.server_version = "0.33.0"
+ m.delete(
+ "http://localhost:5000/api/v3_0/sensors/7/data",
+ status=204,
+ payload={},
+ )
+
+ await client.delete_sensor_data(sensor_id=7, confirm_first=False)
+
+ m.assert_called_once_with(
+ "http://localhost:5000/api/v3_0/sensors/7/data",
+ method="DELETE",
+ json={},
+ headers={
+ "Content-Type": "application/json",
+ "Authorization": "test-token",
+ },
+ params=None,
+ ssl=False,
+ allow_redirects=False,
+ )
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_delete_sensor_data_confirmation_declined():
+ client = FlexMeasuresClient(email="test@test.test", password="test")
+ client.access_token = "test-token"
+ with (
+ patch("builtins.input", return_value="n") as prompt,
+ patch.object(client, "request", new_callable=AsyncMock) as request,
+ ):
+ await client.delete_sensor_data(sensor_id=7)
+ prompt.assert_called_once_with("Permanently delete all data from sensor 7? [y/N] ")
+ request.assert_not_awaited()
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_delete_filtered_sensor_data_confirmation_is_scoped():
+ client = FlexMeasuresClient(email="test@test.test", password="test")
+ client.access_token = "test-token"
+ with (
+ patch("builtins.input", return_value="n") as prompt,
+ patch.object(client, "request", new_callable=AsyncMock) as request,
+ ):
+ await client.delete_sensor_data(sensor_id=7, source=3)
+ prompt.assert_called_once_with(
+ "Permanently delete matching data from sensor 7? [y/N] "
+ )
+ request.assert_not_awaited()
+ await client.close()
+
+
+@pytest.mark.asyncio
+async def test_delete_sensor_data_with_filters():
+ with aioresponses() as m:
+ client = FlexMeasuresClient(email="test@test.test", password="test")
+ client.access_token = "test-token"
+ client.server_version = "0.33.0"
+ m.delete(
+ "http://localhost:5000/api/v3_0/sensors/7/data",
+ status=204,
+ payload={},
+ )
+
+ await client.delete_sensor_data(
+ sensor_id=7,
+ confirm_first=False,
+ source=3,
+ start="2030-01-01T00:00:00+00:00",
+ until="2030-01-02T00:00:00+00:00",
+ )
+
+ m.assert_called_once_with(
+ "http://localhost:5000/api/v3_0/sensors/7/data",
+ method="DELETE",
+ json={
+ "source": 3,
+ "start": "2030-01-01T00:00:00+00:00",
+ "until": "2030-01-02T00:00:00+00:00",
+ },
+ headers={
+ "Content-Type": "application/json",
+ "Authorization": "test-token",
+ },
+ params=None,
+ ssl=False,
+ allow_redirects=False,
+ )
+ await client.close()
+
+
@pytest.mark.asyncio
async def test_post_sensor_data() -> None:
with aioresponses() as m:
@@ -347,13 +445,15 @@ async def test_post_sensor_data() -> None:
values = "test"
unit = "test"
- await flexmeasures_client.post_sensor_data(
+ response, status = await flexmeasures_client.post_sensor_data(
sensor_id=sensor_id,
start=start,
duration=duration,
values=values,
unit=unit,
)
+ assert response == {"test": "test"}
+ assert status == 200
m.assert_called_once_with(
f"http://localhost:5000/api/v3_0/sensors/{sensor_id}/data",
method="POST",
@@ -371,6 +471,33 @@ async def test_post_sensor_data() -> None:
await flexmeasures_client.close()
+@pytest.mark.asyncio
+async def test_post_sensor_data_json_accepted_returns_ingestion_job() -> None:
+ with aioresponses() as m:
+ client = FlexMeasuresClient(email="test@test.test", password="test")
+ client.access_token = "test-token"
+ m.post(
+ "http://localhost:5000/api/v3_0/sensors/5/data",
+ status=202,
+ payload={
+ "job": "ingestion-job-id",
+ "status": "ACCEPTED",
+ },
+ )
+
+ response, status = await client.post_sensor_data(
+ sensor_id=5,
+ start="2023-03-26T10:00+02:00",
+ duration="PT1H",
+ values=[1.0],
+ unit="kW",
+ )
+
+ assert response["job"] == "ingestion-job-id"
+ assert status == 202
+ await client.close()
+
+
@pytest.mark.asyncio
async def test_post_sensor_data_no_params():
"""No json params and no file_path raises ValueError."""
diff --git a/tests/examples/test_hems_workflow.py b/tests/examples/test_hems_workflow.py
new file mode 100644
index 00000000..8974bc72
--- /dev/null
+++ b/tests/examples/test_hems_workflow.py
@@ -0,0 +1,543 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, call, patch
+
+import pytest
+
+HEMS_DIR = Path(__file__).parents[2] / "examples" / "HEMS"
+sys.path.insert(0, str(HEMS_DIR))
+
+from assets_setup import ( # noqa: E402
+ create_community_asset,
+ get_or_create_asset,
+ get_or_create_sensor,
+)
+from HEMS_setup import ( # noqa: E402
+ prompt_for_interrupted_wipe,
+ prompt_for_untracked_setup,
+)
+from scheduling import get_site_assets as get_scheduling_site_assets # noqa: E402
+from utils.asset_utils import ( # noqa: E402
+ find_sensor_by_name_and_asset,
+ post_sensor_data_and_track_ingestion,
+)
+from utils.workflow_utils import ( # noqa: E402
+ ASSET_SETUP_PHASE,
+ DATA_UPLOAD_PHASE,
+ FORECASTING_PHASE,
+ REPORTING_PHASE,
+ SCHEDULING_PHASE,
+ WORKFLOW_VERSION,
+ get_site_assets,
+ get_workflow_state,
+ rename_site_assets,
+ state_needs_upgrade,
+ upgrade_workflow_state,
+ wipe_hems_sensor_data,
+)
+
+
+class InMemoryClient:
+ """Small API-shaped store for exercising idempotent HEMS asset repair."""
+
+ def __init__(self):
+ self.assets = [
+ {
+ "id": 1,
+ "name": "Community Site",
+ "account_id": 9,
+ "parent_asset_id": None,
+ },
+ {
+ "id": 2,
+ "name": "Building A",
+ "account_id": 9,
+ "parent_asset_id": 1,
+ },
+ ]
+ self.sensors = [
+ {
+ "id": 100,
+ "name": "electricity-consumption",
+ "generic_asset_id": 2,
+ }
+ ]
+ self.next_asset_id = 3
+ self.next_sensor_id = 101
+
+ async def get_account(self):
+ return {"id": 9, "name": "test"}
+
+ async def get_assets(self, **kwargs):
+ return [dict(asset) for asset in self.assets]
+
+ async def get_sensors(self, asset_id, **kwargs):
+ return [
+ dict(sensor)
+ for sensor in self.sensors
+ if sensor["generic_asset_id"] == asset_id
+ ]
+
+ async def add_asset(self, **asset):
+ asset = {
+ **asset,
+ "id": self.next_asset_id,
+ "parent_asset_id": asset.get("parent_asset_id"),
+ }
+ self.next_asset_id += 1
+ self.assets.append(asset)
+ return dict(asset)
+
+ async def add_sensor(self, **sensor):
+ sensor = {**sensor, "id": self.next_sensor_id}
+ self.next_sensor_id += 1
+ self.sensors.append(sensor)
+ return dict(sensor)
+
+ async def update_asset(self, asset_id, updates, **kwargs):
+ asset = next(asset for asset in self.assets if asset["id"] == asset_id)
+ asset.update(updates)
+ return dict(asset)
+
+
+def test_interrupted_wipe_requires_explicit_continue():
+ with patch("builtins.input", return_value="n"):
+ assert prompt_for_interrupted_wipe() == "exit"
+ with patch("builtins.input", return_value="c"):
+ assert prompt_for_interrupted_wipe() == "continue-wipe"
+
+
+def test_untracked_setup_offers_repair():
+ with patch("builtins.input", return_value="c"):
+ assert (
+ prompt_for_untracked_setup(["Building A"], ["Building A", "Building B"])
+ == "repair"
+ )
+
+
+@pytest.mark.parametrize(
+ ("answer", "expected"),
+ [("k", "keep-names"), ("m", "rename-sites"), ("y", "recreate")],
+)
+def test_legacy_names_offer_user_choice(answer: str, expected: str):
+ with patch("builtins.input", return_value=answer):
+ assert (
+ prompt_for_untracked_setup(
+ ["My Home 1", "My Home 2"], ["Building A", "Building B"]
+ )
+ == expected
+ )
+
+
+def test_workflow_state_validates_recorded_ids_and_site_names():
+ valid_state = {
+ "workflow-version": 1,
+ "status": "ready",
+ "completed-phases": [ASSET_SETUP_PHASE],
+ "top-level-asset-ids": [1, 2, 3],
+ "sensor-ids": [10, 11],
+ "site-names": ["Building A", "Building B"],
+ }
+ assert (
+ get_workflow_state({"attributes": {"hems_tutorial": valid_state}})
+ == valid_state
+ )
+ assert (
+ get_workflow_state(
+ {
+ "attributes": {
+ "hems_tutorial": {**valid_state, "top-level-asset-ids": ["1"]}
+ }
+ }
+ )
+ is None
+ )
+
+
+@pytest.mark.asyncio
+async def test_get_or_create_asset_reuses_exact_hierarchy_position():
+ client = AsyncMock()
+ client.get_assets.return_value = [
+ {"id": 1, "name": "Building A", "account_id": 9, "parent_asset_id": 4},
+ {"id": 2, "name": "Building A", "account_id": 9, "parent_asset_id": 8},
+ ]
+
+ asset = await get_or_create_asset(
+ client,
+ name="Building A",
+ account_id=9,
+ parent_asset_id=4,
+ generic_asset_type_id=6,
+ )
+
+ assert asset["id"] == 1
+ client.add_asset.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_get_or_create_sensor_only_creates_missing_sensor():
+ client = AsyncMock()
+ client.get_sensors.return_value = [{"id": 10, "name": "existing"}]
+ client.add_sensor.return_value = {"id": 11, "name": "missing"}
+
+ sensor = await get_or_create_sensor(
+ client,
+ name="missing",
+ event_resolution="PT15M",
+ unit="kW",
+ generic_asset_id=1,
+ )
+
+ assert sensor["id"] == 11
+ client.add_sensor.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_get_or_create_sensor_ignores_same_name_on_descendant_assets():
+ client = AsyncMock()
+ client.get_sensors.return_value = [
+ {"id": 10, "name": "power", "generic_asset_id": 2},
+ {"id": 11, "name": "power", "generic_asset_id": 1},
+ ]
+
+ sensor = await get_or_create_sensor(
+ client,
+ name="power",
+ event_resolution="PT15M",
+ unit="kW",
+ generic_asset_id=1,
+ )
+
+ assert sensor["id"] == 11
+ client.add_sensor.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_get_or_create_sensor_rejects_duplicates_on_same_asset():
+ client = AsyncMock()
+ client.get_sensors.return_value = [
+ {"id": 10, "name": "power", "generic_asset_id": 1},
+ {"id": 11, "name": "power", "generic_asset_id": 1},
+ ]
+
+ with pytest.raises(LookupError, match="found 2"):
+ await get_or_create_sensor(
+ client,
+ name="power",
+ event_resolution="PT15M",
+ unit="kW",
+ generic_asset_id=1,
+ )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("job_field", ["job", "job_id"])
+async def test_async_sensor_data_ingestion_tracks_supported_job_fields(job_field):
+ client = AsyncMock()
+ client.post_sensor_data.return_value = ({job_field: "job-123"}, 202)
+ pending_jobs = []
+
+ await post_sensor_data_and_track_ingestion(client, pending_jobs, sensor_id=1)
+
+ assert pending_jobs == ["job-123"]
+
+
+@pytest.mark.asyncio
+async def test_repair_completes_structure_without_replacing_existing_ids():
+ client = InMemoryClient()
+ existing_community = client.assets[0]
+
+ community = await create_community_asset(
+ client,
+ account={"id": 9},
+ community_name="Community Site",
+ site_names=["Building A", "Building B"],
+ community_asset=existing_community,
+ )
+ first_asset_count = len(client.assets)
+ first_sensor_count = len(client.sensors)
+
+ await create_community_asset(
+ client,
+ account={"id": 9},
+ community_name="Community Site",
+ site_names=["Building A", "Building B"],
+ community_asset=community,
+ )
+
+ assert community["id"] == 1
+ assert (
+ next(asset for asset in client.assets if asset["name"] == "Building A")["id"]
+ == 2
+ )
+ assert (
+ next(
+ sensor
+ for sensor in client.sensors
+ if sensor["name"] == "electricity-consumption"
+ and sensor["generic_asset_id"] == 2
+ )["id"]
+ == 100
+ )
+ assert len(client.assets) == first_asset_count
+ assert len(client.sensors) == first_sensor_count
+
+
+@pytest.mark.asyncio
+async def test_sensor_lookup_stays_inside_community_by_default():
+ client = AsyncMock()
+ client.get_asset.return_value = {
+ "id": 1,
+ "name": "Community Site",
+ "account_id": 9,
+ }
+ client.get_assets.return_value = [{"id": 2, "name": "Building A"}]
+ client.get_sensors.return_value = [
+ {"id": 19, "name": "power", "generic_asset_id": 3},
+ {"id": 20, "name": "power", "generic_asset_id": 2},
+ ]
+
+ sensor = await find_sensor_by_name_and_asset(
+ client,
+ sensor_name="power",
+ asset_name="Building A",
+ top_level_asset_id=1,
+ )
+
+ assert sensor["id"] == 20
+ client.get_assets.assert_awaited_once_with(root=1, parse_json_fields=True)
+
+
+@pytest.mark.asyncio
+async def test_explicit_top_level_sensor_lookup_stays_in_community_account():
+ client = AsyncMock()
+ client.get_asset.return_value = {
+ "id": 1,
+ "name": "Community Site",
+ "account_id": 9,
+ }
+ client.get_assets.side_effect = [
+ [{"id": 2, "name": "Building A", "account_id": 9}],
+ [{"id": 3, "name": "Price Market", "account_id": 9}],
+ ]
+ client.get_sensors.return_value = [
+ {"id": 30, "name": "electricity-price", "generic_asset_id": 3}
+ ]
+
+ sensor = await find_sensor_by_name_and_asset(
+ client,
+ sensor_name="electricity-price",
+ asset_name="Price Market",
+ top_level_asset_id=1,
+ allow_top_level_asset=True,
+ )
+
+ assert sensor["id"] == 30
+ assert client.get_assets.await_args_list == [
+ call(root=1, parse_json_fields=True),
+ call(account_id=9, depth=0, parse_json_fields=True),
+ ]
+
+
+@pytest.mark.asyncio
+async def test_get_site_assets_only_returns_direct_children():
+ client = AsyncMock()
+ client.get_assets.return_value = [
+ {"id": 3, "name": "Battery", "account_id": 9, "parent_asset_id": 2},
+ {"id": 2, "name": "Building A", "account_id": 9, "parent_asset_id": 1},
+ {"id": 4, "name": "Building B", "account_id": 9, "parent_asset_id": 1},
+ ]
+
+ sites = await get_site_assets(client, community_asset_id=1, account_id=9)
+
+ assert [site["id"] for site in sites] == [2, 4]
+
+
+@pytest.mark.asyncio
+async def test_scheduling_assets_accept_root_repeated_by_server():
+ client = AsyncMock()
+ community = {"id": 1, "name": "Community Site"}
+ client.get_asset.return_value = community
+ client.get_assets.return_value = [
+ community,
+ {"id": 2, "name": "Building A"},
+ {"id": 3, "name": "Home Battery 1"},
+ {"id": 4, "name": "EV Connector 1 1"},
+ {"id": 5, "name": "EV Connector 2 1"},
+ {"id": 6, "name": "Heat Pump 1"},
+ ]
+
+ with patch("scheduling.find_top_level_asset_id", AsyncMock(return_value=1)):
+ assets = await get_scheduling_site_assets(
+ client, "Building A", "Community Site", 1
+ )
+
+ assert [asset["id"] for asset in assets] == [1, 2, 3, 4, 5, 6]
+
+
+@pytest.mark.asyncio
+async def test_scheduling_assets_reject_different_assets_with_same_name():
+ client = AsyncMock()
+ client.get_asset.return_value = {"id": 1, "name": "Community Site"}
+ client.get_assets.return_value = [{"id": 99, "name": "Community Site"}]
+
+ with patch("scheduling.find_top_level_asset_id", AsyncMock(return_value=1)):
+ with pytest.raises(LookupError, match="ambiguous"):
+ await get_scheduling_site_assets(client, "Building A", "Community Site", 1)
+
+
+@pytest.mark.asyncio
+async def test_rename_site_assets_preserves_ids():
+ client = AsyncMock()
+ sites = [{"id": 20, "name": "My Home 1"}, {"id": 21, "name": "My Home 2"}]
+
+ await rename_site_assets(client, sites, ["Building A", "Building B"])
+
+ assert client.update_asset.await_args_list == [
+ call(asset_id=20, updates={"name": "Building A"}, parse_json_fields=False),
+ call(asset_id=21, updates={"name": "Building B"}, parse_json_fields=False),
+ ]
+
+
+@pytest.mark.asyncio
+async def test_wipe_can_be_repeated_and_finishes_ready():
+ client = AsyncMock()
+ client.get_asset.return_value = {"id": 1, "attributes": {}}
+ state = {
+ "workflow-version": 1,
+ "status": "wiping",
+ "completed-phases": [ASSET_SETUP_PHASE],
+ "top-level-asset-ids": [1, 2, 3],
+ "sensor-ids": [10, 11],
+ "site-names": ["Building A", "Building B"],
+ }
+
+ result = await wipe_hems_sensor_data(client, 1, state)
+
+ assert result["status"] == "ready"
+ assert client.delete_sensor_data.await_args_list == [
+ call(10, confirm_first=False),
+ call(11, confirm_first=False),
+ ]
+ assert client.update_asset.await_count == 2
+
+
+def _state(**overrides) -> dict:
+ """A valid workflow marker, by default written by the current version."""
+ return {
+ "workflow-version": WORKFLOW_VERSION,
+ "status": "ready",
+ "completed-phases": [ASSET_SETUP_PHASE],
+ "top-level-asset-ids": [1, 2, 3],
+ "sensor-ids": [10, 11],
+ "site-names": ["Building A", "Building B"],
+ **overrides,
+ }
+
+
+def test_older_workflow_state_stays_valid_so_it_can_be_upgraded():
+ """A setup from an older tutorial version must not be discarded.
+
+ Rejecting it would leave resume and wipe unable to add newly introduced
+ sensors, and recreation (losing every ID and all data) the only way out.
+ """
+ older = _state(**{"workflow-version": 1})
+
+ assert get_workflow_state({"attributes": {"hems_tutorial": older}}) == older
+ assert state_needs_upgrade(older) is True
+ assert state_needs_upgrade(_state()) is False
+
+
+def test_workflow_state_from_a_newer_version_is_rejected():
+ newer = _state(**{"workflow-version": WORKFLOW_VERSION + 1})
+
+ assert get_workflow_state({"attributes": {"hems_tutorial": newer}}) is None
+
+
+def test_workflow_state_rejects_a_non_integer_version():
+ for version in ("1", 1.5, True, None):
+ state = _state(**{"workflow-version": version})
+ assert get_workflow_state({"attributes": {"hems_tutorial": state}}) is None
+
+
+@pytest.mark.asyncio
+async def test_upgrade_preserves_ids_status_and_unaffected_phases():
+ """Upgrading must cost the user only the phases it actually invalidates."""
+ client = AsyncMock()
+ client.get_asset.return_value = {"id": 1, "attributes": {}}
+ client.get_assets.return_value = [
+ {"id": 1, "name": "Community Site", "account_id": 9, "parent_asset_id": None},
+ {"id": 2, "name": "Energy Market", "account_id": 9, "parent_asset_id": None},
+ {
+ "id": 3,
+ "name": "Local Weather Station",
+ "account_id": 9,
+ "parent_asset_id": None,
+ },
+ ]
+ # sensor 12 is new: added by the asset setup that the upgrade re-ran
+ client.get_sensors.return_value = [{"id": 10}, {"id": 11}, {"id": 12}]
+ state = _state(
+ **{
+ "workflow-version": 1,
+ "status": "wiping",
+ "completed-phases": [
+ ASSET_SETUP_PHASE,
+ DATA_UPLOAD_PHASE,
+ FORECASTING_PHASE,
+ SCHEDULING_PHASE,
+ REPORTING_PHASE,
+ ],
+ }
+ )
+
+ upgraded = await upgrade_workflow_state(
+ client=client,
+ community_asset={"id": 1},
+ account_id=9,
+ state=state,
+ phases_to_rerun=(REPORTING_PHASE,),
+ )
+
+ assert upgraded["workflow-version"] == WORKFLOW_VERSION
+ # only reporting is dropped; the slow phases are not re-run
+ assert upgraded["completed-phases"] == [
+ ASSET_SETUP_PHASE,
+ DATA_UPLOAD_PHASE,
+ FORECASTING_PHASE,
+ SCHEDULING_PHASE,
+ ]
+ # a newly added sensor joins the list, so a later wipe also covers it
+ assert upgraded["sensor-ids"] == [10, 11, 12]
+ # an interrupted wipe stays recoverable, and site names survive
+ assert upgraded["status"] == "wiping"
+ assert upgraded["site-names"] == ["Building A", "Building B"]
+
+
+@pytest.mark.asyncio
+async def test_upgrade_always_records_asset_setup_as_complete():
+ client = AsyncMock()
+ client.get_asset.return_value = {"id": 1, "attributes": {}}
+ client.get_assets.return_value = [
+ {"id": 1, "name": "Community Site", "account_id": 9, "parent_asset_id": None},
+ {"id": 2, "name": "Energy Market", "account_id": 9, "parent_asset_id": None},
+ {
+ "id": 3,
+ "name": "Local Weather Station",
+ "account_id": 9,
+ "parent_asset_id": None,
+ },
+ ]
+ client.get_sensors.return_value = [{"id": 10}]
+
+ upgraded = await upgrade_workflow_state(
+ client=client,
+ community_asset={"id": 1},
+ account_id=9,
+ state=_state(**{"workflow-version": 1, "completed-phases": []}),
+ phases_to_rerun=(REPORTING_PHASE,),
+ )
+
+ assert upgraded["completed-phases"] == [ASSET_SETUP_PHASE]