HydSim¶
The Mesh Python SDK includes experimental functionality to run hydro simulations, run inflow calculations and generate inputs for Marginal Cost using HydSim on the Mesh server. This functionality is under development. Therefore, it is subject to change in future releases.
All methods stream log messages from the server while the operation runs. Responses that this version of the Python SDK does not know are skipped, so newer Mesh servers may send additional response kinds without breaking existing clients.
The functionality may be used both through a synchronous
volue.mesh.Connection.Session and through an asynchronous
volue.mesh.aio.Connection.Session using the following methods.
Further below you’ll find synchronous and asynchronous examples.
API documentation¶
- abstract Session.run_simulation(model: str, case: str, start_time: datetime, end_time: datetime, *, resolution: timedelta = None, scenario: int = None) Union[Iterator[LogMessage], AsyncIterator[LogMessage]][source]
Run a hydro simulation using HydSim on the Mesh server.
This function is experimental and subject to larger changes.
- Parameters
model – The name of the Mesh model in which the simulation case exists.
case – The names of the case group and simulation case in the form ‘CaseGroup/CaseName’.
start_time – The (inclusive) start of the simulation interval.
end_time – The (exclusive) end of the simulation interval.
resolution – The resolution of the simulation. The default resolution of the simulation case is used if this is left as None. Officially supported resolutions are 5, 10, 15, and 60 minutes.
scenario – The scenario(s) to run. All scenarios are run if left as None, no scenarios are run if set as -1, and a specific numbered scenario is run if set as the number of that scenario.
- Returns
An iterator of LogMessage objects. The simulation is done when the iterator is exhausted.
Exhausting the iterator without an exception does not guarantee that the simulation completed successfully. To determine that you must analyze the simulation’s result time series and the log messages from the server.
- Raises
- abstract Session.run_inflow_calculation(model: str, area: str, water_course: str, start_time: datetime, end_time: datetime, *, resolution: timedelta = None) Union[Iterator[LogMessage], AsyncIterator[LogMessage]][source]
Run an inflow calculation using HydSim on the Mesh server.
This function is experimental and subject to larger changes.
- Parameters
model – The name of the Mesh model in which the inflow calculation exists.
area – The area of the water course to calculate.
water_course – The water course to calculate.
start_time – The (inclusive) start of the calculation interval.
end_time – The (exclusive) end of the calculation interval.
resolution – The resolution of the simulation. The default resolution of the inflow calculation case is used if this is left as None. Officially supported resolutions are 5, 10, 15, and 60 minutes.
- Returns
An iterator of LogMessage objects. The calculation is done when the iterator is exhausted.
Exhausting the iterator without an exception does not guarantee that the calculation completed successfully. To determine that you must analyze the calculation’s result time series and the log messages from the server.
- Raises
- abstract Session.get_mc_file(model: str, case: str, start_time: datetime, end_time: datetime) Union[Iterator[volue.mesh._common.LogMessage | str], AsyncIterator[volue.mesh._common.LogMessage | str]][source]
Generate Marginal Cost input using HydSim on the Mesh server.
- Parameters
model – The name of the Mesh model in which the simulation case exists.
case – The names of the case group and simulation case in the form ‘CaseGroup/CaseName’.
start_time – The (inclusive) start of the simulation interval.
end_time – The (exclusive) end of the simulation interval.
- Returns
An iterator of LogMessage objects followed by a single str. The final string is the Marginal Cost input.
- Raises
Example of simulations¶
import asyncio
import logging
from datetime import datetime, timedelta
import helpers
import volue.mesh.aio
from volue import mesh
def sync_run_simulation(address, tls_root_pem_cert):
print("connecting...")
# For production environments create connection using: with_tls, with_kerberos, or with_external_access_token, e.g.:
# connection = mesh.Connection.with_tls(
# address,
# tls_root_pem_cert
# )
connection = mesh.Connection.insecure(address)
with connection.create_session() as session:
start_time = datetime(2023, 11, 1)
end_time = datetime(2023, 11, 2)
print("running simulation...")
try:
for response in session.run_simulation(
"Mesh",
"Cases/Demo",
start_time,
end_time,
resolution=timedelta(minutes=5),
):
if isinstance(response, mesh.LogMessage):
print(
f"[{logging.getLevelName(response.level)}] {response.message}"
)
print("done")
except Exception as e:
print(f"failed to run simulation: {e}")
async def async_run_simulation(address, tls_root_pem_cert):
print("connecting...")
# For production environments create connection using: with_tls, with_kerberos, or with_external_access_token, e.g.:
# connection = mesh.aio.Connection.with_tls(
# address,
# tls_root_pem_cert
# )
connection = mesh.aio.Connection.insecure(address)
async with connection.create_session() as session:
start_time = datetime(2023, 11, 1)
end_time = datetime(2023, 11, 2)
print("running simulation...")
try:
async for response in session.run_simulation(
"Mesh",
"Cases/Demo",
start_time,
end_time,
resolution=timedelta(minutes=5),
):
if isinstance(response, mesh.LogMessage):
print(
f"[{logging.getLevelName(response.level)}] {response.message}"
)
print("done")
except Exception as e:
print(f"failed to run simulation: {e}")
if __name__ == "__main__":
address, tls_root_pem_cert = helpers.get_connection_info()
sync_run_simulation(address, tls_root_pem_cert)
asyncio.run(async_run_simulation(address, tls_root_pem_cert))
Example of inflow calculations¶
import asyncio
import logging
from datetime import datetime, timedelta
import helpers
import volue.mesh.aio
from volue import mesh
def sync_run_inflow_calculation(address, tls_root_pem_cert):
print("connecting...")
# For production environments create connection using: with_tls, with_kerberos, or with_external_access_token, e.g.:
# connection = mesh.Connection.with_tls(
# address,
# tls_root_pem_cert
# )
connection = mesh.Connection.insecure(address)
with connection.create_session() as session:
start_time = datetime(2021, 1, 1)
end_time = datetime(2021, 1, 2)
print("running inflow calculation...")
try:
for response in session.run_inflow_calculation(
"Mesh",
"Area",
"WaterCourse",
start_time,
end_time,
resolution=timedelta(minutes=5),
):
if isinstance(response, mesh.LogMessage):
print(
f"[{logging.getLevelName(response.level)}] {response.message}"
)
print("done")
except Exception as e:
print(f"failed to run inflow calculation: {e}")
async def async_run_inflow_calculation(address, tls_root_pem_cert):
print("connecting...")
# For production environments create connection using: with_tls, with_kerberos, or with_external_access_token, e.g.:
# connection = mesh.aio.Connection.with_tls(
# address,
# tls_root_pem_cert
# )
connection = mesh.aio.Connection.insecure(address)
async with connection.create_session() as session:
start_time = datetime(2021, 1, 1)
end_time = datetime(2021, 1, 2)
print("running inflow calculation...")
try:
async for response in session.run_inflow_calculation(
"Mesh",
"Area",
"WaterCourse",
start_time,
end_time,
resolution=timedelta(minutes=5),
):
if isinstance(response, mesh.LogMessage):
print(
f"[{logging.getLevelName(response.level)}] {response.message}"
)
print("done")
except Exception as e:
print(f"failed to run inflow calculation: {e}")
if __name__ == "__main__":
address, tls_root_pem_cert = helpers.get_connection_info()
sync_run_inflow_calculation(address, tls_root_pem_cert)
asyncio.run(async_run_inflow_calculation(address, tls_root_pem_cert))