Python cookbook💡
Introduction💡
The aim of this cookbook is to provide threedi-api-client example code snippets for the most commonly executed tasks.
This Notebook is not intended to cover the full range of options of the 3Di API:
-
Not all endpoints are included here
-
All endpoints used in this notebook have further configuration options
For a full overview of the API, please refer to the Swagger page: https://api.3di.live/v3/swagger/
<> indicate placeholders and are to be removed; so where the code snippet says
name__icontains="<(part of) name of your organisation>"
and your organisation is United Nations, in your case you should fill in
name__icontains="United Nations"
Getting started💡
Authentication💡
from threedi_api_client.api import ThreediApi
from threedi_api_client.versions import V3BetaApi
API_HOST = "https://api.3di.live"
PERSONAL_API_KEY = "<your personal api key>" # You can get one from https://management.3di.live/personal_api_keys
config = {
"THREEDI_API_HOST": API_HOST,
"THREEDI_API_PERSONAL_API_TOKEN": PERSONAL_API_KEY
}
api_client: V3BetaApi = ThreediApi(config=config, version='v3-beta') # version=v3-beta to have access to all the latest features
Finding your organisation💡
organisation = api_client.organisations_list(name__icontains="<(part of) name of your organisation>" ).results[0]
Schematisations and 3Di models💡
Create a new schematisation💡
# requires: Authentication, Finding your organisation
schematisation = threedi_api.schematisations_create(
data={
"owner": organisation.unique_id,
"name": "<Your schematisation name>",
"tags": ["description:<Your description of the schematisation>", "project:<Your project nr.>", "<tag1>", "<tag2>"], # Optional
}
)
Finding your schematisation💡
# requires: Authentication, Finding your organisation
# assumes: There is only one schematisation with the name you are looking for
schematisation_name="<Name of your schematisation>"
schematisation = threedi_api.schematisations_list(name="<Name of your schematisation>").results[0]
Creating a new revision💡
This process consists of several steps: - Create a new (empty) revision - Upload data to the new revision (spatialite and rasters) - Commit revision
# requires: Authentication, Create a new schematisation or Finding your schematisation
# assumes: All rasters are in {spatialite parent directory}/rasters
from pathlib import Path
from zipfile import ZipFile
import hashlib
import urllib3
from threedi_api_client.files import upload_file
UPLOAD_TIMEOUT = urllib3.Timeout(connect=60, read=600)
# Create a new (empty) revision
revision = threedi_api.schematisations_revisions_create(schematisation.id, data={"empty": True})
# Upload data to the new revision.
sqlite_path = Path(sqlite_path)
local_schematisation_dir = sqlite_path.parent
## Upload spatialite
sqlite_zip_path = sqlite_path.with_suffix('.zip')
ZipFile(sqlite_zip_path, mode='w').write(str(sqlite_path), arcname=str(sqlite_path.name)) # Sqlite must be zipped before upload
upload = threedi_api.schematisations_revisions_sqlite_upload(
id=revision.id,
schematisation_pk=schematisation.id,
data={"filename": str(sqlite_zip_path.name)}
)
if upload.put_url is None:
print(f"Sqlite '{sqlite_path.name}' already existed, skipping upload.")
else:
print(f"Uploading '{sqlite_path.name}'...")
upload_file(url=upload.put_url, file_path=sqlite_zip_path, timeout=UPLOAD_TIMEOUT)
# # Rasters
def md5(fname):
"""stackoverflow.com/questions/3431825/generating-an-md5-checksum-of-a-file"""
hash_md5 = hashlib.md5()
with open(fname, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
raster_names = {
"dem_file": "<name of your dem>.tif", # add more rasters here if relevant.
# see the POST /v3/schematisations/{schematisation_pk}/revisions/{revision_pk}/rasters/ endpoint
# for a list of possible raster types
}
for raster_type, raster_file_name in raster_names.items():
raster_path = local_schematisation_dir / "rasters" / raster_file_name
md5sum = md5(str(raster_path))
raster_create = threedi_api.schematisations_revisions_rasters_create(
revision.id,
schematisation.id,
data={
"name": raster_file_name.name,
"type": raster_type,
"md5sum": md5sum
}
)
if raster_create.file:
if raster_create.file.state == "uploaded":
logging.info(f"Raster '{raster_path}' already exists, skipping upload.")
return
print(f"Uploading '{raster_path}'...")
upload = threedi_api.schematisations_revisions_rasters_upload(
raster_create.id,
revision.id,
schematisation.id,
data={"filename": raster_file_name.name}
)
upload_file(
upload.put_url,
raster_path,
timeout=UPLOAD_TIMEOUT
)
# Commit revision
# First wait for all files' statuses to become to 'uploaded'
max_retries = 900
wait_time = 1
for i in range(max_retries):
revision = threedi_api.schematisations_revisions_read(revision.id, schematisation.id)
states = [revision.sqlite.file.state]
states.extend([raster.file.state for raster in revision.rasters])
if all(state == "uploaded" for state in states):
break
elif any(state == "created" for state in states):
print(f"Sleeping {wait_time} seconds to wait for the files to become 'uploaded'...")
time.sleep(wait_time)
continue
else:
raise RuntimeError("One or more rasters have an unexpected state")
else:
raise RuntimeError("Some files are still in 'created' state")
schematisation_revision = threedi_api.schematisations_revisions_commit(
revision.id,
schematisation.id,
data={"commit_message": "<Your commit message>"})
print(f"Committed revision {revision.number}.")
Create a 3Di model and simulation template💡
# requires: Authentication, Creating a new revision
import time
max_retries_creation=60
wait_time_creation=5
for i in range(max_retries_creation):
try:
threedimodel = threedi_api.schematisations_revisions_create_threedimodel(
id=revision.id,
schematisation_pk=schematisation.id
)
print(f"Creating threedimodel with id {threedimodel.id}...")
break
except Exception as e:
print(f"Sleeping {wait_time_creation} seconds to retry creating threedimodel...")
time.sleep(wait_time_creation)
continue
max_retries_processing=60
wait_time_processing=30
if threedimodel:
for i in range(max_retries_processing):
threedimodel = threedi_api.threedimodels_read(threedimodel.id)
if threedimodel.is_valid:
print(f"Succesfully created threedimodel with id {threedimodel.id}")
break
else:
time.sleep(wait_time_processing)
if not threedimodel.is_valid:
raise RuntimeError(f"Failed to sucessfully process threedimodel with id {threedimodel.id}, as it is not valid")
else:
raise RuntimeError('Failed to create threedimodel')
Finding your 3Di model💡
# requires: Authentication
# assumes: the model you are looking for is the most recently created model that matches the search criteria
threedimodel = api_client.threedimodels_list(name__icontains="<Your model name, or a part of it>").results[0]
Running simulations💡
Running simulation - minimum example💡
This consists of four steps: - Find the simulation template - Create a simulation from the template - Add initial conditions, forcings and events - Start the simulation
See other code snippets for variations, e.g. adding specific kinds of initial conditions, forcings, and events
# requires: Authentication, Finding your organisation, Create a 3Di model or Finding your 3Di model
# assumes: There is only one simulation template for your 3Di model
# Find the simulation template
simulation_templates = api_client.simulation_templates_list(simulation__threedimodel__id=threedmodel.id)
assert simulation_templates.count > 0, f"No simulation templates found for threedimodel {threedimodel.name}"
simulation_template = simulation_templates.results[0]
# Create a simulation from the template
simulation = api_client.simulations_from_template(
data={
"template": simulation_template.id,
"name": "<Your simulation name>",
"organisation": organisation.unique_id,
"start_datetime": "2000-01-01T00:00:00", # Format: "YYYY-MM-DDThh:mm:ss"
"duration": 3600 # in seconds, so we simulate for 1 hour
}
)
# Add initial conditions, forcings and events
## Example: constant rain event of 20 mm/h for 1 hour
rain_intensity_m_s = 20/1000/3600 # 20 mm / h
rain = api_client.simulations_events_rain_constant_create(
simulation_pk=simulation.id,
data={
'offset': 0, # starting at the start of the simulation
'value': rain_intensity_m_s,
'duration': 3600, # in seconds
'units': 'm/s'
}
)
# Start the simulation
api_client.simulations_actions_create(my_simulation.id, data={"name": "queue"})
# Monitor the simulation status
previous_status = ""
while True:
status = api_client.simulations_status_list(simulation.id)
if status.name != previous_status:
print(status.name)
previous_status = status.name
if status.name in ['finished', 'crashed']: # created, starting, initialized, queued, ended, postprocessing, finished, crashed
print(status.name)
break
time.sleep(1)
Choosing a simulation template💡
If multiple simulation templates exist for your 3Di model, you need to choose the correct one
# variation on: Running simulation - minimum example
# list the available simulation templates
simulation_templates = api_client.simulation_templates_list(simulation__threedimodel__id=my_model.id)
for simulation_template in simulation_templates.results:
print(simulation_template.id, simulation_template.name)
simulation_template = api_client.simulation_templates_read(id=<fill in one of the ids chosen from the list>)
Create saved state - stable threshold💡
# TBD
# Note that this feature is currently broken
Create saved state - timed💡
# variation on: Running simulation - minimum example
create_saved_state = simulations_create_saved_states_timed_create(
simulation_pk=simulation.id,
data={
"name": "<How you want to name your saved state on the server>",
"time": 24*60*60, # Create saved state after 24 hours of simulation
}
)
Boundary conditions file💡
If the schematisation contains any boundary condition locations, the simulation template will have time series defined for them. You can only replace all boundary conditions at once. The example below includes code to download the existing boundary conditions file, so that you may change one or more of the time series in the file and than re-upload it.
This process consists of several steps: - Downloading the existing boundary conditions file - Deleting the existing boundary conditions file from the simulation - Uploading a new file
Download the existing boundary conditions file (optional)💡
# requires: Authentication, Running simulation - minimum example
from threedi_api_client.files import download_file
import urllib3
DOWNLOAD_TIMEOUT = urllib3.Timeout(connect=60, read=600)
original_boundary_conditions_file = api_client.simulations_events_boundaryconditions_file_list(
simulation_pk=simulation_template.simulation.id
).results[0]
original_boundary_conditions_file_download_url = api_client.simulations_events_boundaryconditions_file_download(
simulation_pk=simulation_template.simulation.id,
id = original_boundary_conditions_file.id
).get_url
download_file(
url=original_boundary_conditions_file_download_url,
target=r"<your/target/path/boundary_conditions.json>",
timeout=DOWNLOAD_TIMEOUT
)
Replace boundary conditions file💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
# Deleting the existing boundary conditions file from the simulation
boundary_conditions_files = api_client.simulations_events_boundaryconditions_file_list(simulation.id)
if boundary_conditions_files.count > 0:
boundary_conditions_file = boundary_conditions_files.results[0]
api_client.simulations_events_boundaryconditions_file_delete(boundary_conditions_file.id, simulation.id)
# Post new boundary conditions file
from threedi_api_client.files import upload_file
boundary_conditions_file = r"<path\to\your\new\boundary_conditions.json"
upload_object = api_client.simulations_events_boundaryconditions_file_create(
simulation.id,
data={"filename": "<How you want to name the file on the server>"}
)
res = upload_file(upload_object.put_url, boundary_conditions_file)
print("Processing boundary conditions file...")
max_retries = 900
wait_time = 1
for i in range(max_retries):
state = api_client.simulations_events_lateral_file_read(simulation_pk=simulation.id, id=lateral.id).file.state
if state not in ["created", "uploaded", "processed"]:
raise Exception("Something went wrong while processing")
if state == "processed":
break
time.sleep(wait_time)
print("Finished processing boundary conditions file")
Breaches💡
# variation on: Running simulation - minimum example
# List available potential breaches
potential_breaches = api_client.threedimodels_potentialbreaches_list(threedimodel.id)
print(potential_breaches)
# Example: using the first potential breach
breach = potential_breaches.results[0]
breach_event = api_client.simulations_events_breaches_create(
simulation.id, data={
"potential_breach": breach.id,
"duration_till_max_depth": 1800,
"initial_width": 0.5,
"offset": 1800
}
)
Lateral - constant💡
# variation on: Running simulation - minimum example
# 1D lateral
lateral = api_client.simulations_events_lateral_constant_create(
simulation.id, data={
"offset": 900,
"duration": 300,
"connection_node": 4,
"value": 300/1000, # 300 L/s
"units": "m3/s",
}
)
# 1D lateral with substance concentration
# Note that you have to define the substances first, see "Substances"
lateral = api_client.simulations_events_lateral_constant_create(
simulation.id, data={
"offset": 900,
"duration": 7200,
"connection_node": 4,
"value": 300/1000, # 300 L/s
"units": "m3/s",
"substances": [
{
"substance": substance.id,
"concentrations": [
[0, 20],
[7200, 20]
]
}
]
}
)
# 2D lateral
## Starting after 1 hour, for a duration of two hours
lateral = api_client.simulations_events_lateral_constant_create(
simulation_pk=simulation.id,
data={
"offset": 3600,
"duration": 7200,
"value": 1,
"units": "m3/s",
"point": {
"type": "Point",
"coordinates": [ # coordinates in WGS84, [lon, lat]
3.3126012,
47.9751043
],
}
)
# 2D Lateral with substance concentration
# Note that you have to define the substances first, see "Substances"
lateral = api_client.simulations_events_lateral_constant_create(
simulation_pk=simulation.id,
data={
"offset": 3600,
"duration": 7200,
"value": 1,
"units": "m3/s",
"point": {
"type": "Point",
"coordinates": [ # coordinates in WGS84, [lon, lat]
3.3126012,
47.9751043
],
},
"substances": [
{
"substance": substance.id,
"concentrations": [
[0, 20],
[7200, 20]
]
}
]
)
Lateral - time series💡
# variation on: Running simulation - minimum example
# 1D lateral
lateral = api_client.simulations_events_lateral_constant_create(
simulation.id, data={
"offset": 900,
"duration": 300,
"connection_node": 4,
"values": [[0, 0.005], [900, 0.002], [1800, 0.001], [2700, 0.0005], [3600, 0]],
"units": "m3/s",
}
)
# 1D time series lateral with substance concentrations
lateral = api_client.simulations_events_lateral_constant_create(
simulation.id, data={
"offset": 900,
"duration": 300,
"connection_node": 4,
"values": [[0, 0.005], [900, 0.002], [1800, 0.001], [2700, 0.0005], [3600, 0]],
"units": "m3/s",
"substances": [
{
"substance": substance.id,
"concentrations": [
[0, 10],
[900, 20],
[1800, 60],
[2700, 20],
[3600, 10]
]
}
]
}
)
# 2D lateral
lateral = api_client.simulations_events_lateral_timeseries_create(
simulation_pk=simulation.id,
data={
"offset": 3600,
"duration": 7200,
"values": [[0, 0.005], [900, 0.002], [1800, 0.001], [2700, 0.0005], [3600, 0]],
"units": "m3/s",
"point": {
"type": "Point",
"coordinates": [
3.3126012,
47.9751043
]
},
}
)
# 2D lateral with substance concentrations
lateral = api_client.simulations_events_lateral_timeseries_create(
simulation_pk=simulation.id,
data={
"offset": 3600,
"duration": 7200,
"values": [[0, 0.005], [900, 0.002], [1800, 0.001], [2700, 0.0005], [3600, 0]],
"units": "m3/s",
"point": {
"type": "Point",
"coordinates": [
3.3126012,
47.9751043
]
},
"substances": [
{
"substance": substance.id,
"concentrations": [
[0, 10],
[900, 20],
[1800, 60],
[2700, 20],
[3600, 10]
]
}
]
}
)
Laterals - file💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
laterals_file = "<path/to/your/laterals.json>"
upload_object = api_client.simulations_events_lateral_file_create(
simulation.id,
data={
"filename": "<How you want to name the file on the server>",
"offset": 0,
# "periodic": "daily" # if the laterals are dry weather flow laterals
}
)
res = upload_file(upload_object.put_url, laterals_file)
print("Processing laterals file...")
lateral = api_client.simulations_events_lateral_file_list(simulation_pk=simulation.id).results[0]
max_retries = 900
wait_time = 1
for i in range(max_retries):
state = api_client.simulations_events_lateral_file_read(simulation_pk=simulation.id, id=lateral.id).file.state
if state not in ["created", "uploaded", "processed"]:
raise Exception("Something went wrong while processing uploaded laterals")
if state == "processed":
break
time.sleep(wait_time)
print("Finished processing laterals file")
# Note: substances can also be included in the laterals file
Leakage - constant💡
# variation on: Running simulation - minimum example
## Example: constant leakage of 2 mm/d for 10 days
leakage_rate_m_s = 2/1000/60/60/24 # 2 mm/d
leakage = api_client.simulations_events_leakage_constant_create(
simulation_pk=simulation.id,
data={
'offset': 0, # starting at the start of the simulation
'value': leakage_rate_m_s,
'duration': 10*24*60*60, # 10 days
'units': 'm/s'
}
)
# Constant leakage with substance concentrations
# Note that you have to define the substances first, see "Substances"
leakage_rate_m_s = 2/1000/60/60/24 # 2 mm/d
leakage = api_client.simulations_events_leakage_constant_create(
simulation_pk=simulation.id,
data={
'offset': 0, # starting at the start of the simulation
'value': leakage_rate_m_s,
'duration': 10*24*60*60, # 10 days
'units': 'm/s',
'substances': [
{
"substance": substance.id,
"concentrations": [
[0, 10],
[10*24*60*60, 10]
]
}
]
}
)
Leakage - NetCDF💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
upload_object = THREEDI_API.simulations_events_leakage_rasters_netcdf_create(
simulation.id,
data={
"filename": "<How you want to name the file on the server>"
}
)
res = upload_file(upload_object.put_url, "<path/to/your/rain.nc>")
print("Processing NetCDF file...")
rain_event = api_client.simulations_events_leakage_rasters_netcdf_read(simulation_pk=simulation.id).results[0]
max_retries = 900
wait_time = 1
for i in range(max_retries):
state = api_client.simulations_events_leakage_rasters_netcdf_read(simulation_pk=simulation.id, id=lateral.id).file.state
if state not in ["created", "uploaded", "processed"]:
raise Exception("Something went wrong while processing...")
if state == "processed":
break
time.sleep(wait_time)
print("Finished processing rain file")
# Note: you cannot combine NetCDF leakage with substance concentrations
Leakage - time series💡
# variation on: Running simulation - minimum example
# Time series leakage
leakage = api_client.simulations_events_leakage_timeseries_create(
simulation.id,
{
"offset": 0,
"interpolate": True,
"values": [
[
0, # from day 0
2/1000/60/60/24 # 2 mm/d
],
[
3*60*60*24, # from day 3
10/1000/60/60/24 # 10 mm/d
],
[
6*60*60*24, # from day 6
5/1000/60/60/24 # 5 mm/d
]
],
"units": "m/s",
)
# Time series leakage with substance concentration
# Note that you have to define the substances first, see "Substances"
leakage = api_client.simulations_events_leakage_timeseries_create(
simulation.id,
{
"offset": 0,
"interpolate": True,
"values": [
[
0, # from day 0
2/1000/60/60/24 # 2 mm/d
],
[
3*60*60*24, # from day 3
10/1000/60/60/24 # 10 mm/d
],
[
6*60*60*24, # from day 6
5/1000/60/60/24 # 5 mm/d
]
],
"units": "m/s",
"substances": [
{
"substance": substance.id,
"concentrations": [
[
0, # from day 0
18.5
],
[
3*60*60*24, # from day 3
10.2
],
[
6*60*60*24, # from day 6
8.3
]
]
}
]
}
)
Rain - constant💡
# variation on: Running simulation - minimum example
## Example: constant rain event of 20 mm/h for 1 hour
rain_intensity_m_s = 20/1000/3600 # 20 mm / h
rain = api_client.simulations_events_rain_constant_create(
simulation_pk=simulation.id,
data={
'offset': 0, # starting at the start of the simulation
'value': rain_intensity_m_s,
'duration': 3600, # in seconds
'units': 'm/s'
}
)
# Constant rain with substance concentrations
# Note that you have to define the substances first, see "Substances"
rain_intensity_m_s = 20/1000/3600 # 20 mm / h
rain = api_client.simulations_events_rain_constant_create(
simulation_pk=simulation.id,
data={
'offset': 0, # starting at the start of the simulation
'value': rain_intensity_m_s,
'duration': 3600, # in seconds
'units': 'm/s',
'substances': [
{
"substance": substance.id,
"concentrations": [
[0, 45.8],
[3600, 45.8],
]
}
)
Rain - raster (Lizard)💡
# variation on: Running simulation - minimum example
rain = api_client.simulations_events_rain_rasters_lizard_create(
simulation.id,
{
"offset":0,
"duration":12*60*60, # 12 hours
"reference_uuid": "730d6675-35dd-4a35-aa9b-bfb8155f9ca7",
"start_datetime":"2020-09-21T00:00:00.000Z",
"units":"mm/h"
}
)
# Lizard raster rain with substance concentrations
# Note that you have to define the substances first, see "Substances"
rain = api_client.simulations_events_rain_rasters_lizard_create(
simulation.id,
{
"offset":0,
"duration":12*60*60, # 12 hours
"reference_uuid": "730d6675-35dd-4a35-aa9b-bfb8155f9ca7",
"start_datetime":"2020-09-21T00:00:00.000Z",
"units":"mm/h",
"substances": [
{
"substance": substance.id,
"concentrations": [
[0, 100],
[12*60*60, 100],
]
}
)
Rain - raster (NetCDF)💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
upload_object = THREEDI_API.simulations_events_rain_rasters_netcdf_create(
simulation.id,
data={
"filename": "<How you want to name the file on the server>"
}
)
res = upload_file(upload_object.put_url, "<path/to/your/rain.nc>")
print("Processing NetCDF file...")
rain_event = api_client.simulations_events_rain_rasters_netcdf_read(simulation_pk=simulation.id).results[0]
max_retries = 900
wait_time = 1
for i in range(max_retries):
state = api_client.simulations_events_rain_rasters_netcdf_read(simulation_pk=simulation.id, id=lateral.id).file.state
if state not in ["created", "uploaded", "processed"]:
raise Exception("Something went wrong while processing...")
if state == "processed":
break
time.sleep(wait_time)
print("Finished processing rain file")
Rain - time series💡
# variation on: Running simulation - minimum example
rain = api_client.simulations_events_rain_timeseries_create(
simulation_pk=simulation.id,
data={
'offset':0,
'values': [[0, 0.005], [900, 0.002], [1800, 0.001], [2700, 0.0005], [3600, 0]], # Rain intensity decreases every 15 minutes.
'units': 'm/s'}
)
# Time series rain with substance concentrations
# Two zones (North and South) are labelled with their own tracer substance
# Note that you have to define the substances first, see "Substances"
rain_intensity_m3_s = 20/1000/3600 # 20 mm / h
rain = api_client.simulations_events_rain_timeseries_create(
simulation_pk=simulation.id,
data={
'offset':0,
'values': [[0, rain_intensity_m3_s], [2*60*60, rain_intensity_m3_s]],
'units': 'm/s',
'substances': [
{
'substance': tracer_north.id,
'concentrations': [[0, 100], [2*60*60, 100]],
'zone': "Polygon ((187406 320804.5, 187616 320965.5, 187819 320993.5, 187826 321049.5, 187826 321147.5, 188078 321322.5, 188141 321238.5, 188239 321343.5, 188414 321420.5, 188694 321455.5, 189135 321497.5, 189450 321469.5, 189030 321105.5, 188876 321021.5, 188722 321021.5, 188582 320916.5, 188400 320797.5, 188071 320657.5, 187749 320601.5, 187497 320531.5, 187406 320804.5))"
},
{
'substance': tracer_south.id,
'concentrations': [[0, 100], [2*60*60, 100]],
'zone': "Polygon ((187567 320503.5, 188316 320720.5, 188316 320720.5, 188743 320993.5, 189002 321035.5, 189191 321161.5, 189471 321441.5, 189527 321231.5, 189478 321077.5, 189506 320874.5, 189443 320762.5, 189380 320580.5, 189226 320454.5, 189023 320286.5, 188967 320125.5, 188974 319985.5, 188862 319957.5, 188729 319929.5, 188435 319929.5, 188218 319999.5, 187987 320083.5, 187686 320111.5, 187567 320503.5))"
}
]
}
)
Rain - time series (Lizard)💡
# variation on: Running simulation - minimum example
rain = api_client.simulations_events_rain_timeseries_lizard_create(
simulation.id,
{
"offset": 0,
"duration": 12*60*60, # 12 hours
"multiplier": 1.1,
"interpolate": True,
"reference_uuid": "730d6675-35dd-4a35-aa9b-bfb8155f9ca7",
"start_datetime":"2020-09-21T00:00:00.000Z",
"units":"mm/h"
}
)
Rain - time series (NetCDF)💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
upload_object = THREEDI_API.simulations_events_rain_timeseries_netcdf_create(
simulation.id,
data={
"filename": "<How you want to name the file on the server>"
}
)
res = upload_file(upload_object.put_url, "<path/to/your/rain.nc>")
print("Processing NetCDF file...")
rain_event = api_client.simulations_events_rain_timeseries_netcdf_read(simulation_pk=simulation.id).results[0]
max_retries = 900
wait_time = 1
for i in range(max_retries):
state = api_client.simulations_events_rain_timeseries_netcdf_read(simulation_pk=simulation.id, id=lateral.id).file.state
if state not in ["created", "uploaded", "processed"]:
raise Exception("Something went wrong while processing...")
if state == "processed":
break
time.sleep(wait_time)
print("Finished processing rain file")
Raster edits (direct input of coordinates)💡
# variation on: Running simulation - minimum example
raster_id = api_client.threedimodels_rasters_list(threedimodel.id, type="dem_file").results[0].id
raster_edits = api_client.simulations_events_raster_edits_create(
simulation.id,
data = {
"offset": 0, # time in seconds since start of simulation
"value": 2.5, # new height in mMSL
"raster": raster_id,
"polygon": {
"type": "Polygon",
"coordinates": [ # in WGS84, [lon, lat]
[
[
4.699366986751557,
52.64294137346387
],
[
4.699687510728837,
52.64336289692115
],
[
4.700147509574891,
52.642926725857535
],
[
4.699698239564897,
52.64304716158594
],
[
4.699366986751557,
52.64294137346387
]
]
]
}
}
)
Raster edits (read from a shapefile)💡
# variation on: Running simulation - minimum example
# assumes: the shapefile has attributes 'offset' and 'elevation'
from osgeo import ogr
from osgeo import osr
polygons_fn = "dem_edits.shp"
datasource = ogr.Open(polygons_fn)
layer = datasource.GetLayer()
# Create a CoordinateTransformation to WGS84, as required by the API
source_srs = layer.GetSpatialRef()
target_srs = osr.SpatialReference()
target_srs.ImportFromEPSG(4326)
transform = osr.CoordinateTransformation(source_srs, target_srs)
raster_id = api_client.threedimodels_rasters_list(threedimodel.id, type="dem_file").results[0].id
for feature in layer:
geom = feature.GetGeometryRef()
geom.Transform(transform)
wkt = geom.ExportToWkt()
raster_edits = api_client.simulations_events_raster_edits_create(
my_simulation.id,
data={
"offset": feature['offset'], # time in seconds since start of simulation
"value": feature['elevation'], # new height in m MSL
"raster": raster_id,
"polygon": wkt
}
)
Sources & sinks - constant💡
# variation on: Running simulation - minimum example
## Example: evapotranspiration of 10 mm/d for 24 hours
evaoptranspiration_m_s = 10/1000/60/60/24 # 10 mm / d
sss = api_client.simulations_events_sources_sinks_constant_create(
simulation_pk=simulation.id,
data={
'offset': 0, # starting at the start of the simulation
'value': evaoptranspiration_m_s,
'duration': 24*60*60, # in seconds
'units': 'm/s'
}
)
Sources & sinks - raster (Lizard)💡
# variation on: Running simulation - minimum example
sss = api_client.simulations_events_sources_sinks_rasters_lizard_create(
simulation.id,
{
"offset":0,
"duration":12*60*60,
"reference_uuid": "730d6675-35dd-4a35-aa9b-bfb8155f9ca7",
"start_datetime":"2020-09-21T00:00:00.000Z",
"multiplier":-0.1,
"units":"mm/h"
}
)
Sources & sinks - raster (NetCDF)💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
upload_object = THREEDI_API.simulations_events_sources_sinks_rasters_netcdf_create(
simulation.id,
data={
"filename": "<How you want to name the file on the server>"
}
)
res = upload_file(upload_object.put_url, "<path/to/your/evapotranspiration.nc>")
print("Processing NetCDF file...")
sss = api_client.simulations_events_sources_sinks_rasters_netcdf_read(simulation_pk=simulation.id).results[0]
max_retries = 900
wait_time = 1
for i in range(max_retries):
state = api_client.simulations_events_sources_sinks_rasters_netcdf_read(simulation_pk=simulation.id, id=sss.id).file.state
if state not in ["created", "uploaded", "processed"]:
raise Exception("Something went wrong while processing...")
if state == "processed":
break
time.sleep(wait_time)
print("Finished processing surface sources and sinks file")
Sources & sinks - time series💡
# variation on: Running simulation - minimum example
## Example: evapotranspiration varying per day, over 5 days
sss = api_client.simulations_events_sources_sinks_constant_create(
simulation_pk=simulation.id,
data={
'offset': 0, # starting at the start of the simulation
'values': [
[0*24*60*60, 10/1000/60/60/24], # 10 mm / d
[1*24*60*60, 8/1000/60/60/24], # 8 mm / d
[2*24*60*60, 5/1000/60/60/24], # 5 mm / d
[3*24*60*60, 4/1000/60/60/24], # 4 mm / d
[4*24*60*60, 6/1000/60/60/24], # 6 mm / d
],
'interpolate': True,
'units': 'm/s'
}
)
Sources & sinks - time series (Lizard)💡
# variation on: Running simulation - minimum example
sss = api_client.simulations_events_sources_sinks_timeseries_lizard_create(
simulation.id,
{
"offset": 0,
"duration": 12*60*60, # 12 hours
"multiplier": -1,
"interpolate": True,
"reference_uuid": "730d6675-35dd-4a35-aa9b-bfb8155f9ca7",
"start_datetime":"2020-09-21T00:00:00.000Z",
"units":"mm/h"
}
)
Sources & sinks - time series (NetCDF)💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
upload_object = THREEDI_API.simulations_events_sources_sinks_timeseries_netcdf_create(
simulation.id,
data={
"filename": "<How you want to name the file on the server>"
}
)
res = upload_file(upload_object.put_url, "<path/to/your/evapotranspiration.nc>")
print("Processing NetCDF file...")
sss = api_client.simulations_events_sources_sinks_timeseries_netcdf_read(simulation_pk=simulation.id).results[0]
max_retries = 900
wait_time = 1
for i in range(max_retries):
state = api_client.simulations_events_sources_sinks_timeseries_netcdf_read(simulation_pk=simulation.id, id=sss.id).file.state
if state not in ["created", "uploaded", "processed"]:
raise Exception("Something went wrong while processing...")
if state == "processed":
break
time.sleep(wait_time)
print("Finished processing surface sources and sinks file")
Structure control - file💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
structure_control_file = "<path/to/your/structure_control_file.json>"
upload_object = api_client.simulations_events_structure_control_file_create(
simulation.id,
data={
"filename": "<How you want to name the file on the server>",
"offset": 0,
}
)
res = upload_file(upload_object.put_url, structure_control_file)
print("Processing file...")
structure_control = api_client.simulations_events_structure_control_file_list(simulation_pk=simulation.id).results[0]
max_retries = 900
wait_time = 1
for i in range(max_retries):
state = api_client.simulations_events_structure_control_file_read(simulation_pk=simulation.id, id=structure_control.id).file.state
if state not in ["created", "uploaded", "processed"]:
raise Exception("Something went wrong while processing")
if state == "processed":
break
time.sleep(wait_time)
print("Finished processing structure control file")
Structure control - memory💡
# variation on: Running simulation - minimum example
structure_control = api_client.simulations_events_structure_control_memory_create(
simulation_pk=simulation.id,
data={
"offset": 0,
"duration": 3600,
"measure_specification": {
"name": "Measurement location",
"locations": [
{
"weight": "1.00",
"content_type": "v2_connection_node",
"content_pk": 356,
}
],
"variable": "s1",
"operator": "<"
},
"structure_id": 10250,
"structure_type": "v2_weir",
"type": "set_crest_level",
"value": [
9.05
],
"upper_threshold": 0.3,
"lower_threshold": 0.1,
"is_active": false,
"is_inverse": false
}
)
Structure control - table💡
# variation on: Running simulation - minimum example
structure_control = api_client.simulations_events_structure_control_table_create(
simulation_pk=simulation.id,
data={
"offset": 0,
"duration": 3600,
"measure_specification": {
"name": "Measurement location",
"locations": [
{
"weight": "1.00",
"content_type": "v2_connection_node",
"content_pk": 356,
}
],
"variable": "s1",
"operator": "<"
},
"structure_id": 10249,
"structure_type": "v2_weir",
"type": "set_crest_level",
"values": [
[
9.05,
-1.45
],
]
}
)
Structure control - timed💡
# variation on: Running simulation - minimum example
structure_control = api_client.simulations_events_structure_control_timed_create(
simulation_pk=simulation.id,
data={
"offset": 0,
"duration": 100,
"value": [
0.4, 0.8
],
"type": "set_discharge_coefficients",
"structure_id": 21,
"structure_type": "v2_weir"
}
)
Substances💡
To use the water quality module or to label flows, you need to: - Define one or more substances - Add these substances to one or more forcings and/or initial conditions
The code snippet below illustrates how to create a substance. Examples of how to add the substances to forcings are included in the code snippets for those forcings
# variation on: Running simulation - minimum example
substance = api_client.simulations_substances_create(
simulation.id,
data={
"name": "Phosphate",
"units": "mg/L"
}
)
label = api_client.simulations_substances_create(
simulation.id,
data={
"name": "Rain in area A",
"units": "%"
}
)
Wind - constant💡
# variation on: Running simulation - minimum example
# /v3/simulations/{simulation_pk}/events/wind/constant/
wind = api_client.simulations_events_wind_constant_create(
simulation_pk=simulation.id,
data={
"offset": 100,
"duration": 400,
"units": "m/s",
"speed_value": 80,
"direction_value": 220
}
)
Wind - time series💡
# variation on: Running simulation - minimum example
wind = api_client.simulations_events_wind_timeseries_create(
simulation_pk=simulation.id,
data={
"offset": 0,
"values": [
[0, 40, 180], # [time in s, speed in m/s, direction in degrees clockwise from north]
[60, 35, 160]
],
"units": "m/s",
}
)
Initial 1D water level - constant💡
# variation on: Running simulation - minimum example
# Find & delete any existing 1D water level files or constant values from this simulation
simulation_initial_water_level_files = api_client.simulations_initial1d_water_level_file_list(simulation.id)
if simulation_initial_water_level_files.count > 0:
simulation_initial_water_level_file = simulation_initial_water_level_files.results[0]
api_client.simulations_initial1d_water_level_file_delete(id=simulation_initial_water_level_file.id, simulation_pk=simulation.id)
simulations_initial1d_water_level_constant = api_client.simulations_initial1d_water_level_constant_list(simulation.id)
if simulations_initial1d_water_level_constant.count > 0:
simulations_initial1d_water_level_constant = simulations_initial1d_water_level_constant.results[0]
api_client.simulations_initial1d_water_level_constant_delete(id=simulations_initial1d_water_level_constant.id, simulation_pk=simulation.id)
# Create a new 1D initial water level
initial_1d_water_level = api_client.simulations_initial1d_water_level_constant_create(
simulation_pk=simulation.id, data={"value": 0.5}
)
Initial 1D water level - file (uploading a new set of initial water levels to the 3Di model)💡
Initial 1D water levels are owned by a threedimodel. In this exmple, a new set of initial water levels is uploaded to the 3Di model and than used in the simulation.
Note that he IDs in the file are calculation node IDs, not connection node IDs!
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
import time
# Add the initial water levels to the threedimodel
initial_water_levels_1d_file = "<path/to/your/initial_water_levels_1d.json>"
initial_water_levels_1d = api_client.threedimodels_initial_waterlevels_create(
threedimodel.id,
data={"dimension": "one_d"}
)
upload_object = api_client.threedimodels_initial_waterlevels_upload(
id=initial_water_levels_1d.id, threedimodel_pk=threedimodel.id,
data={"filename": "<How you want to call the file on the server>"})
res = upload_file(upload_object.put_url, initial_water_levels_1d_file)
print("Processing file...")
max_retries = 900
wait_time = 1
for i in range(max_retries):
initial_water_levels_1d = api_client.threedimodels_initial_waterlevels_read(id=initial_water_levels_1d.id, threedimodel_pk=threedimodel.id)
if initial_water_levels_1d.state not in ["processing", "valid"]:
raise Exception(f"Something went wrong while processing. State: {initial_water_levels_1d.state}. State detail: {initial_water_levels_1d.state_detail}")
if initial_water_levels_1d.state == "valid":
break
time.sleep(wait_time)
print("Finished processing initial water levels 1D file")
# Add it to the simulation
## Find & delete any existing 1D water level files or constant values from this simulation
simulation_initial_water_level_files = api_client.simulations_initial1d_water_level_file_list(simulation.id)
if simulation_initial_water_level_files.count > 0:
simulation_initial_water_level_file = simulation_initial_water_level_files.results[0]
api_client.simulations_initial1d_water_level_file_delete(id=simulation_initial_water_level_file.id, simulation_pk=simulation.id)
simulations_initial1d_water_level_constant = api_client.simulations_initial1d_water_level_constant_list(simulation.id)
if simulations_initial1d_water_level_constant.count > 0:
simulations_initial1d_water_level_constant = simulations_initial1d_water_level_constant.results[0]
api_client.simulations_initial1d_water_level_constant_delete(id=simulations_initial1d_water_level_constant.id, simulation_pk=simulation.id)
## Post a new one, i.e. assign the set of initial water levels that is owned by the threedimodel to the simulation
api_client.simulations_initial1d_water_level_file_create(simulation.id, data={"initial_waterlevel": initial_water_levels_1d.id})
Initial 1D water levels - file (using previously uploaded 1D water levels)💡
# variation on: Initial 1D water level - file (uploading a new set of initial water levels to the 3Di model)
all_initial_water_levels = api_client.threedimodels_initial_waterlevels_list(threedimodel_pk=threedimodel.id).results
for initial_water_levels_1d in all_initial_water_levels:
if initial_water_levels_1d.dimension == "one_d" and hasattr(initial_water_levels_1d.file, "filename"):
print(f"{initial_water_levels_1d.file.filename}: {initial_water_levels_1d.id} ({initial_water_levels_1d.state})")
initial_water_levels_1d = api_client.threedimodels_initial_waterlevels_read(id=<choose an id from the list printed above>, threedimodel_pk=threedimodel.id)
Initial 2D water level - constant💡
# variation on: Running simulation - minimum example
# First find & delete any existing 2D water level rasters or constant values from this simulation
simulation_initial_water_level_rasters = api_client.simulations_initial2d_water_level_raster_list(simulation.id)
if simulation_initial_water_level_rasters.count > 0:
simulation_initial_water_level_raster = simulation_initial_water_level_rasters.results[0]
api_client.simulations_initial2d_water_level_raster_delete(id=simulation_initial_water_level_raster.id, simulation_pk=simulation.id)
simulations_initial2d_water_level_constant = api_client.simulations_initial2d_water_level_constant_list(simulation.id)
if simulations_initial2d_water_level_constant.count > 0:
simulations_initial2d_water_level_constant = simulations_initial2d_water_level_constant.results[0]
api_client.simulations_initial2d_water_level_constant_delete(id=simulations_initial2d_water_level_constant.id, simulation_pk=simulation.id)
# Create a new 2D initial water level
initial_2d_water_level = api_client.simulations_initial2d_water_level_constant_create(
simulation_pk=simulation.id, data={"value": 0.5}
)
Initial 2D water level - raster💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
import time
# Add the initial water levels to the threedimodel
initial_water_levels_raster_file = "<path/to/your/initial_water_level.tif>"
initial_water_levels_raster = api_client.threedimodels_rasters_create(
threedimodel.id,
data={
name="<How you want to call this raster on the server>"
"type": "initial_waterlevel_file"
}
)
upload_object = api_client.threedimodels_rasters_upload(
id=initial_water_levels_raster.id,
threedimodel_pk=threedimodel.id,
data={"filename": "<How you want to call the file on the server>"})
res = upload_file(upload_object.put_url, initial_water_levels_raster_file)
print("Processing file...")
max_retries = 900
wait_time = 1
for i in range(max_retries):
initial_water_levels_raster = api_client.threedimodels_initial_waterlevels_read(id=initial_water_levels_raster.id, threedimodel_pk=threedimodel.id)
if initial_water_levels_raster.file.state not in ["created", "uploaded", "processed"]:
raise Exception(
f"Something went wrong while processing. "
f"State: {initial_water_levels_raster.file.state}. "
f"State description: {initial_water_levels_raster.file.state_description}")
if initial_water_levels_raster.file.state == "valid":
break
time.sleep(wait_time)
print("Finished processing initial water levels raster")
# Add it to the simulation
## find & delete any existing 2D water level rasters or constant values from this simulation
simulation_initial_water_level_rasters = api_client.simulations_initial2d_water_level_raster_list(simulation.id)
if simulation_initial_water_level_rasters.count > 0:
simulation_initial_water_level_raster = simulation_initial_water_level_rasters.results[0]
api_client.simulations_initial2d_water_level_raster_delete(id=simulation_initial_water_level_raster.id, simulation_pk=simulation.id)
simulations_initial2d_water_level_constant = api_client.simulations_initial2d_water_level_constant_list(simulation.id)
if simulations_initial2d_water_level_constant.count > 0:
simulations_initial2d_water_level_constant = simulations_initial2d_water_level_constant.results[0]
api_client.simulations_initial2d_water_level_constant_delete(id=simulations_initial2d_water_level_constant.id, simulation_pk=simulation.id)
## Post a new one, i.e. assign the set of initial water levels that is owned by the threedimodel to the simulation
api_client.simulations_initial2d_water_level_raster_create(
simulation.id,
data={
"aggregation_method": "mean", # choose from: mean, max, min
"initial_waterlevel": initial_water_levels_raster.id
}
)
Initial groundwater level - constant💡
# variation on: Running simulation - minimum example
# First find & delete any existing groundwater level rasters or constant values from this simulation
simulation_initial_groundwater_level_rasters = api_client.simulations_initial_groundwater_level_raster_list(simulation.id)
if simulation_initial_groundwater_level_rasters.count > 0:
simulation_initial_groundwater_level_raster = simulation_initial_groundwater_level_rasters.results[0]
api_client.simulations_initial_groundwater_level_raster_delete(id=simulation_initial_water_level_raster.id, simulation_pk=simulation.id)
simulations_initial_groundwater_level_constant = api_client.simulations_initial_groundwater_level_constant_list(simulation.id)
if simulations_initial_groundwater_level_constant.count > 0:
simulations_initial_groundwater_level_constant = simulations_initial_groundwater_level_constant.results[0]
api_client.simulations_initial_groundwater_level_constant_delete(id=simulations_initial_groundwater_level_constant.id, simulation_pk=simulation.id)
# Create a new initial groundwater level
initial_groundwater_level = api_client.simulations_initial_groundwater_level_constant_create(
simulation_pk=simulation.id, data={"value": 0.5}
)
Initial groundwater level - raster💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
import time
# Add the initial water levels to the threedimodel
# Note that the initial water levels for the groundwater domain are the same type of objects (InitialWaterLevel) as for the surface water domain
initial_water_levels_raster_file = "<path/to/your/initial_water_level.tif>"
initial_water_levels_raster = api_client.threedimodels_rasters_create(
threedimodel.id,
data={
name="<How you want to call this raster on the server>"
"type": "initial_waterlevel_file"
}
)
upload_object = api_client.threedimodels_rasters_upload(
id=initial_water_levels_raster.id,
threedimodel_pk=threedimodel.id,
data={"filename": "<How you want to call the file on the server>"})
res = upload_file(upload_object.put_url, initial_water_levels_raster_file)
print("Processing file...")
max_retries = 900
wait_time = 1
for i in range(max_retries):
initial_water_levels_raster = api_client.threedimodels_initial_waterlevels_read(id=initial_water_levels_raster.id, threedimodel_pk=threedimodel.id)
if initial_water_levels_raster.file.state not in ["created", "uploaded", "processed"]:
raise Exception(
f"Something went wrong while processing. "
f"State: {initial_water_levels_raster.file.state}. "
f"State description: {initial_water_levels_raster.file.state_description}")
if initial_water_levels_raster.file.state == "valid":
break
time.sleep(wait_time)
print("Finished processing initial water levels raster")
# Add it to the simulation
## find & delete any existing ground water level rasters or constant values from this simulation
simulation_initial_groundwater_level_rasters = api_client.simulations_initial_groundwater_level_raster_list(simulation.id)
if simulation_initial_groundwater_level_rasters.count > 0:
simulation_initial_groundwater_level_raster = simulation_initial_groundwater_level_rasters.results[0]
api_client.simulations_initial_groundwater_level_raster_delete(id=simulation_initial_groundwater_level_raster.id, simulation_pk=simulation.id)
simulations_initial_groundwater_level_constant = api_client.simulations_initial_groundwater_level_constant_list(simulation.id)
if simulations_initial_groundwater_level_constant.count > 0:
simulations_initial_groundwater_level_constant = simulations_initial_groundwater_level_constant.results[0]
api_client.simulations_initial_groundwater_level_constant_delete(id=simulations_initial_groundwater_level_constant.id, simulation_pk=simulation.id)
## Post a new one, i.e. assign the set of initial water levels that is owned by the threedimodel to the simulation
api_client.simulations_initial_groundwater_level_raster_create(
simulation.id,
data={
"aggregation_method": "mean", # choose from: mean, max, min
"initial_waterlevel": initial_water_levels_raster.id
}
)
Initial 1D substance concentrations (uploading a new set of initial concentrations to the 3Di model)💡
Using initial 1D substance concentrations involves these steps: - Define a substance (not included in this example, see "Substances") - Find or upload an initial concentrations file to the threedimodel - Link the substance to the initial concentration and assign it to the simulation
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
import time
# Add the initial concentrations to the threedimodel
initial_concentrations_1d_file = "<path/to/your/initial_concentrations_1d.json>"
initial_concentrations_1d = api_client.threedimodels_initial_concentrations_create(
threedimodel.id,
data={"dimension": "one_d"}
)
upload_object = api_client.threedimodels_initial_concentrations_upload(
id=initial_concentrations_1d.id, threedimodel_pk=threedimodel.id,
data={"filename": "<How you want to call the file on the server>"})
res = upload_file(upload_object.put_url, initial_concentrations_1d_file)
print("Processing file...")
max_retries = 900
wait_time = 1
for i in range(max_retries):
initial_concentrations_1d = api_client.threedimodels_initial_concentrations_read(id=initial_concentrations_1d.id, threedimodel_pk=threedimodel.id)
if initial_concentrations_1d.state not in ["processing", "valid"]:
raise Exception(f"Something went wrong while processing. State: {initial_concentrations_1d.state}. State detail: {initial_concentrations_1d.state_detail}")
if initial_concentrations_1d.state == "valid":
break
time.sleep(wait_time)
print("Finished processing initial concentrations 1D file")
# Add it to the simulation
one_d_substance_concentration = api_client.simulations_initial1d_substance_concentrations_create(
simulation.id, {
"substance": substance.id,
"initial_concentration": initial_concentrations_1d.id}
)
Initial 1D substance concentrations (using previously uploaded 1D concentrations)💡
# variation on: Initial 1D substance concentrations (uploading a new set of initial substance concentrations to the 3Di model)
all_initial_concentrations = api_client.threedimodels_initial_concentrations_list(threedimodel_pk=threedimodel.id).results
for initial_concentrations_1d in all_initial_concentrations:
if initial_concentrations_1d.dimension == "one_d" and hasattr(initial_concentrations_1d.file, "filename"):
print(f"{initial_concentrations_1d.file.filename}: {initial_concentrations_1d.id} ({initial_concentrations_1d.state})")
initial_concentrations_1d = api_client.threedimodels_initial_concentrations_read(id=<choose an id from the list printed above>, threedimodel_pk=threedimodel.id)
Initial 2D substance concentrations (uploading a new initial concentrations raster)💡
# variation on: Running simulation - minimum example
from threedi_api_client.files import upload_file
import time
# Add the initial water levels to the threedimodel
initial_concentrations_raster_file = "<path/to/your/initial_concentrations.tif>"
initial_concentrations_raster = api_client.threedimodels_rasters_create(
threedimodel.id,
data={
name="<How you want to call this raster on the server>"
"type": "initial_concentrations_file"
}
)
upload_object = api_client.threedimodels_rasters_upload(
id=initial_concentrations_raster.id,
threedimodel_pk=threedimodel.id,
data={"filename": "<How you want to call the file on the server>"})
res = upload_file(upload_object.put_url, initial_concentrations_raster_file)
print("Processing file...")
max_retries = 900
wait_time = 1
for i in range(max_retries):
initial_concentrations_raster = api_client.threedimodels_initial_concentrations_read(id=initial_concentrations_raster.id, threedimodel_pk=threedimodel.id)
if initial_concentrations_raster.file.state not in ["created", "uploaded", "processed"]:
raise Exception(
f"Something went wrong while processing. "
f"State: {initial_concentrations_raster.file.state}. "
f"State description: {initial_concentrations_raster.file.state_description}")
if initial_concentrations_raster.file.state == "valid":
break
time.sleep(wait_time)
print("Finished processing initial concentrations raster")
# Add it to the simulation
# Note that you have to define the substances first, see "Substances"
two_d_substance_concentration = api_client.simulations_initial2d_substance_concentrations_create(
simulation.id, {
"substance": substance.id,
"aggregation_method": "mean",
"initial_concentration": initial_concentrations_raster.id}
)
print(f"Created initial 2D substance concentration with id {two_d_substance_concentration.id} ")
Initial 2D substance concentrations (using previously uploaded 2D concentrations)💡
# variation on: Initial 2D substance concentrations (uploading a new set of initial substance concentrations to the 3Di model)
all_initial_concentrations = api_client.threedimodels_initial_concentrations_list(threedimodel_pk=threedimodel.id).results
for initial_concentrations_2d in all_initial_concentrations:
if initial_concentrations_1d.dimension == "two_d" and hasattr(initial_concentrations_2d.file, "filename"):
print(f"{initial_concentrations_2d.file.filename}: {initial_concentrations_2d.id} ({initial_concentrations_2d.state})")
initial_concentrations_2d = api_client.threedimodels_initial_concentrations_read(
id=<choose an id from the list printed above>,
threedimodel_pk=threedimodel.id
)
Initial groundwater substance concentrations💡
# Note that you have to define the substances first, see "Substances"
# Same as initial 2D substance concentrations, except that you need to use
#
# simulations_initial_groundwater_substance_concentrations_create()
#
# instead of
#
# simulations_initial2d_substance_concentrations_create()
Initial saved state💡
- Find the saved state you want to use
- Remove any existing saved states from the simulation
- Add saved state to the simulation
# variation on: Running simulation - minimum example
# List available saved states
saved_states=api_client.threedimodels_saved_states_list(threedimodel_pk=threedimodel.id)
print(saved_states)
# Find a specific saved state, e.g. searching by name (assuming that the name is unique)
search_for="<(part of) the name of your saved state>"
saved_state = None
for s in saved_states.results:
if search_for.upper() in s.name.upper():
saved_state = s
break
if not saved_state:
raise Exception(f"Saved state that icontains {search_for} not found")
# Remove any existing saved state from the simulation
simulation_saved_states = api_client.simulations_initial_saved_state_list(
simulation_pk=simulation.id
)
if simulation_saved_states.count > 0:
api_client.simulations_initial_saved_state_delete(
id=simulation_saved_states.results[0].id
simulation_pk=simulation.id
)
# Add saved state to the simulation as initial saved state
api_client.simulations_initial_saved_state_create(
simulation_pk=simulation.id,
data={"saved_state": saved_state.id}
)
Initial wind drag coefficient💡
# variation on: Running simulation - minimum example
api_client.simulations_initial_wind_drag_coefficient_create(
simulation_pk=simulation.id,
data={"value": 0.003}
)
Post-processing simulation results💡
To store and process 3Di simulation results in the Lizard Scenario Archive
# variation on: Running simulation - minimum example
# Basic processing
api_client.simulations_results_post_processing_lizarc_crd_basieatesimulation.ids
data={
scenario_name: "<give me a name>",
process_basic_results: False,
}cn
# Arrival times
api_client.simulations_results_post_processing_lizard_arrival_create(simulation.id, data={})
# Damage
api_client.simulations_results_post_processing_lizard_damage_create(
simulation.id,
data={
cost_type: "min",
flood_month: "sep",
inundation_period: 12.0,
repair_time_infrastructure: 120,
repair_time_buildings: 240
}
)g
)
Settings - aggregation💡
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: add the aggregation settings that are needed by the water balance tool
## Optional: remove all exisiting aggrgation settings first
current_settings = api_client.simulations_settings_aggregation_list(simulation.id, limit=99).results
for entry in current_settings:
entry_id = entry.url.split("/")[-2]
print(entry_id)
api_client.simulations_settings_aggregation_delete(id=entry_id, simulation_pk=simulation.id)
## Post the new settings
interval = 300
aggregation_settings = [
{
"flow_variable": "pump_discharge",
"method": "cum",
"interval": interval,
},
{
"flow_variable": "lateral_discharge",
"method": "cum",
"interval": interval,
},
{
"flow_variable": "simple_infiltration",
"method": "cum",
"interval": interval,
},
{
"flow_variable": "rain",
"method": "cum",
"interval": interval,
},
{
"flow_variable": "leakage",
"method": "cum",
"interval": interval,
},
{
"flow_variable": "interception",
"method": "current",
"interval": interval,
},
{
"flow_variable": "discharge",
"method": "cum",
"interval": interval,
},
{
"flow_variable": "discharge",
"method": "cum_negative",
"interval": interval,
},
{
"flow_variable": "discharge",
"method": "cum_positive",
"interval": interval,
},
{
"flow_variable": "volume",
"method": "current",
"interval": interval,
},
{
"flow_variable": "surface_source_sink_discharge",
"method": "cum_negative",
"interval": interval,
},
{
"flow_variable": "surface_source_sink_discharge",
"method": "cum_positive",
"interval": interval,
}
]
for aggregation_setting in aggregation_settings:
api_client.simulations_settings_aggregation_create(simulation.id, data = aggregation_setting)
Settings - customized result areas💡
Note that to use customized result areas, you will also need to define the customized results paramaters in the output settings
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: create customized result area
api_client.simulations_settings_customized_result_areas_create(
simulation.id,
data={
"name": "<How you want to name your customized result area>",
"geometry": "Polygon ((4.77423354 52.63953687, 4.77330784 52.6371667, 4.77181345 52.6373858, 4.77108553 52.63611832, 4.76915369 52.62956295, 4.77038341 52.62921291, 4.77146061 52.62834362, 4.77227968 52.62688974, 4.77443521 52.62505391, 4.77687182 52.62597372, 4.77993675 52.62767436, 4.78076013 52.62975233, 4.78138395 52.63085721, 4.77960075 52.63272746, 4.78089502 52.63542349, 4.78341307 52.63822293, 4.78345255 52.63922761, 4.77839281 52.63920181, 4.77423354 52.63953687))"
}
)
# Example: same as above, but now the polygon(s) are read from a shapefile
# assumes: the shapefile has attribute "name"
from osgeo import ogr
from osgeo import osr
polygons_fn = "customized_result_areas.shp"
datasource = ogr.Open(polygons_fn)
layer = datasource.GetLayer()
# Create a CoordinateTransformation to WGS84, as required by the API
source_srs = layer.GetSpatialRef()
target_srs = osr.SpatialReference()
target_srs.ImportFromEPSG(4326)
transform = osr.CoordinateTransformation(source_srs, target_srs)
for feature in layer:
geom = feature.GetGeometryRef()
geom.Transform(transform)
wkt = geom.ExportToWkt()
api_client.simulations_settings_customized_result_areas_create(
simulation.id,
data={
"name": feature['name'],
"geometry": wkt
}
)
Settings - numerical💡
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: change the thin water layer definition
api_client.simulations_settings_numerical_partial_update(simulation.id, {"limiter_slope_thin_water_layer": 0.5})
Settings - output settings💡
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: change the output time step for hydrodynamic calculation results
api_client.simulations_settings_output_settings_partial_update(simulation.id, {"hydro_output_time_step": 300})
# Example: define output settings for the customized results file to write only water levels, every second
# Note that to use a customized results file, you also need to define one or more customized result areas
api_client.simulations_settings_output_settings_partial_update(
simulation_pk=simulation.id,
data={
"customized_hydro_output_time_step": 1,
"customized_hydro_output_start_time": 0,
"customized_hydro_output_precision": 1, # single precision
"customized_hydro_output_variables": [
"s1"
]
}
)
Settings - physical💡
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: change the 1D advection scheme
api_client.simulations_settings_physical_partial_update(simulation.id, {"use_advection_1d": 3})
Settings - time step💡
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: change the simulation time step
api_client.simulations_settings_time_step_partial_update(simulation.id, {"time_step": 1.0})
Settings - water quality💡
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: change the simulation time step for the water quality module
api_client.simulations_settings_water_quality_partial_update(simulation.id, {"time_step": 1.0})
Settings - water quality customized result areas💡
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: create customized result area for water quality results
api_client.simulations_settings_water_quality_customized_result_areas_create(
simulation.id,
data={
"name": "<How you want to name your customized result area>",
"geometry": "Polygon ((4.77423354 52.63953687, 4.77330784 52.6371667, 4.77181345 52.6373858, 4.77108553 52.63611832, 4.76915369 52.62956295, 4.77038341 52.62921291, 4.77146061 52.62834362, 4.77227968 52.62688974, 4.77443521 52.62505391, 4.77687182 52.62597372, 4.77993675 52.62767436, 4.78076013 52.62975233, 4.78138395 52.63085721, 4.77960075 52.63272746, 4.78089502 52.63542349, 4.78341307 52.63822293, 4.78345255 52.63922761, 4.77839281 52.63920181, 4.77423354 52.63953687))"
}
)
# Example: same as above, but now the polygon(s) are read from a shapefile
# assumes: the shapefile has attribute "name"
from osgeo import ogr
from osgeo import osr
polygons_fn = "customized_result_areas.shp"
datasource = ogr.Open(polygons_fn)
layer = datasource.GetLayer()
# Create a CoordinateTransformation to WGS84, as required by the API
source_srs = layer.GetSpatialRef()
target_srs = osr.SpatialReference()
target_srs.ImportFromEPSG(4326)
transform = osr.CoordinateTransformation(source_srs, target_srs)
for feature in layer:
geom = feature.GetGeometryRef()
geom.Transform(transform)
wkt = geom.ExportToWkt()
api_client.simulations_settings_water_quality_customized_result_areas_create(
simulation.id,
data={
"name": feature['name'],
"geometry": wkt
}
)
Settings - water quality output settings💡
# variation on: Running simulation - minimum example
# Simulation settings can only be changed before the simulation is started/queued
# Example: change the output time step for water quality results
api_client.simulations_settings_water_quality_output_settings_partial_update(simulation.id, {"output_time_step": 300})
# Example: define output settings for the water quality customized results file to write only water levels, every second
# Note that to use a water quality customized results file, you also need to define one or more water quality customized result areas
api_client.simulations_settings_water_quality_output_settings_partial_update(
simulation_pk=simulation.id,
data={
"customized_output_time_step": 1,
"customized_output_start_time": 0,
"customized_output_precision": 2, # double precision
}
)
Start, pause, shut down simulations💡
Start a simulation💡
# Required: Authentication, Running simulation - minimum example
# It is usually best to queue the simulation, so that it will be started as soon as a session is available
api_client.simulations_actions_create(my_simulation.id, data={"name": "queue"})
# To start the simulation after it has been paused or initialized:
api_client.simulations_actions_create(my_simulation.id, data={"name": "start"})
# Use the duration parameter to automaticallly pauze the simulation after a specified time
api_client.simulations_actions_create(
simulation.id,
data={
"name": "start",
"duration": 2500
}
)
# It is also possible to initialize the simulation without starting it yet
{"name": "initialize"}
Pause a simulation💡
# Required: Authentication, Running simulation - minimum example
api_client.simulations_actions_create(my_simulation.id, data={"name": "pause"})
End a simulation💡
# Required: Authentication, Running simulation - minimum example
api_client.simulations_actions_create(my_simulation.id, data={"name": "shutdown"})
Information about simulations💡
Settings used in a simulation💡
# Required: Authentication, Running simulation - minimum example
api_client.simulations_settings_overview(simulation_pk=simulation.id)
Status and status history💡
# Required: Authentication, Running simulation - minimum example
# Current status
api_client.simulations_status_list(simulation_pk=simulation.id)
# Status history
api_client.simulations_status_history_list(simulation_pk=simulation.id)
Initials, forcings and events💡
# Required: Authentication, Running simulation - minimum example
api_client.simulations_events(id=simulation.id)
Downloading results💡
Downloading the computional grid (gridadmin.h5)💡
3Di Results Analysis and threedigrid require the "gridadmin.h5" file to analyse the results. This is a model specific file so it's under the threedimodels endpoint. We'll also download this file:
import urllib3
from threedi_api_client.files import download_file
DOWNLOAD_TIMEOUT = urllib3.Timeout(connect=60, read=600)
download_folder = Path("<path\to\target\directory>")
download_folder.mkdir(exist_ok=True)
download_url = api_client.threedimodels_gridadmin_download(simulation.threedimodel_id)
file_path = download_folder / "gridadmin.h5"
download_file(download_url.get_url, file_path, timeout=DOWNLOAD_TIMEOUT)
print(f"Finished downloading gridadmin.h5")
Downloading simulation results💡
# required: Running a simulation - minimum example
import urllib3
from threedi_api_client.files import download_file
DOWNLOAD_TIMEOUT = urllib3.Timeout(connect=60, read=600)
download_folder = Path("<path\to\target\directory>")
download_folder.mkdir(exist_ok=True)
result_files = api_client.simulations_results_files_list(simulation.id)
for file in result_files.results:
download_url = api_client.simulations_results_files_download(
id=file.id, simulation_pk=simulation.id
)
file_path = download_folder / file.filename
download_file(
download_url.get_url,
file_path,
timeout=DOWNLOAD_TIMEOUT
)
print(f"Finished downloading {file.filename}")