💾 Event modifications in the database

This notebook demonstrates how to manipulate event objects in a FloodAdapt database, including loading, creating, editing, copying, and deleting events.

It is assumed that at the start of this notebook, you have a database and the events saved from the other event notebooks.

import flood_adapt.objects.forcing as f

from pathlib import Path
from datetime import datetime

from flood_adapt import unit_system as us
from flood_adapt import FloodAdapt, Settings
from flood_adapt.objects import HistoricalEvent
from flood_adapt.objects.forcing.time_frame import TimeFrame

# Setup FloodAdapt
STATIC_DATA_DIR = Path("../../_data/examples/static-data").resolve()
settings = Settings(
    DATABASE_ROOT=Path("../../_data/examples").resolve(),
    DATABASE_NAME="charleston_test",
)
fa = FloodAdapt(database_path=settings.database_path)

Analyze the effect of different rainfall forcings

Lets say we want to analyze the effect of rainfall, how would you approach that:

  1. Retrieve an event from the database
  2. Create the rainfall forcings you want to analyze
  3. Update the event’s name and rainfall forcings
  4. Save the updated event to the database
  5. Create scenarios using these events and run them (see this example) TODO

Of course, this approach also works for analyzing the difference between any other forcings.

# 1. Create an event or load it from the database
# event = fa.get_event(...)
event_name="database_manipulation_event"
event = HistoricalEvent(
    name=event_name,
    description="Some event description",
    time=TimeFrame(
        start_time=datetime(2020, 1, 1),
        end_time=datetime(2020, 1, 2),
    ),
    forcings={
        f.ForcingType.WATERLEVEL: [f.WaterlevelGauged()]
    }
)
fa.save_event(event=event)

# 2. Create rainfall forcings
rainfall_constant = f.RainfallConstant(
    intensity=us.UnitfulIntensity(value=10, units=us.UnitTypesIntensity.mm_hr)
)

rainfall_synthetic = f.RainfallSynthetic(
    timeseries=f.GaussianTimeseries(
        duration=us.UnitfulTime(value=12, units=us.UnitTypesTime.hours),
        # Choose the middle of the time frame for peak time
        peak_time=us.UnitfulTime.from_timedelta(event.time.duration / 2), 
        peak_value=us.UnitfulIntensity(value=10, units=us.UnitTypesIntensity.mm_hr),
    )
)

rainfall_csv = f.RainfallCSV(path=STATIC_DATA_DIR / "rainfall.csv")

rainfall_forcings = [
    (rainfall_constant, "constant"),
    (rainfall_synthetic, "synthetic"),
    (rainfall_csv, "csv")
]

for forcing, fsource in rainfall_forcings:
    # 3. Update the event
    event.forcings[f.ForcingType.RAINFALL] = [forcing]
    event.name = f"{event_name}_rainfall_{fsource}"
    
    # 4. Save the event
    fa.save_event(event=event)

fa.get_events()["name"]

# 5. Now the events can be used as the components for scenario creation. Similar to how the rainfall forcings are used to create various events
# For more information on how to create scenarios, see the scenario creation example
['database_manipulation_event',
 'database_manipulation_event_rainfall_constant',
 'database_manipulation_event_rainfall_csv',
 'database_manipulation_event_rainfall_synthetic',
 'event_set',
 'test_set']

Delete, Save, Copy, Edit

FloodAdapt supports deleting, editing, and copying objects by name, as shown below.

It is possible to achieve the exact same goal in multiple ways:

  1. event = fa.get_event(event_name) -> update event -> fa.delete_event(event_name) -> fa.save_event(event)
  2. event = fa.get_event(event_name) -> update event -> fa.edit_event(event)
print("Initial:", fa.get_events()["name"])

# Copy all events
for name in fa.get_events()["name"]:
    new_name = f"{name}_copy"
    fa.copy_event(old_name=name, new_name=new_name, new_description=f"Copy of {name}")
print("Names after `Copy`:", fa.get_events()["name"])

# Delete all copied events
for name in fa.get_events()["name"]:
    if "copy" in name:
        fa.delete_event(name)
print("Names after `Delete`:", fa.get_events()["name"])

# Edit an existing event
unedited_event = fa.get_event(event_name)
print("Description before `Edit`:", unedited_event.description)

unedited_event.description = f"Updated description"
fa.save_event(event=unedited_event, overwrite=True)

edited_event = fa.get_event(event_name)
print("Description after `Edit`:", edited_event.description)
Initial: ['database_manipulation_event', 'database_manipulation_event_rainfall_constant', 'database_manipulation_event_rainfall_csv', 'database_manipulation_event_rainfall_synthetic', 'event_set', 'test_set']
Names after `Copy`: ['database_manipulation_event', 'database_manipulation_event_copy', 'database_manipulation_event_rainfall_constant', 'database_manipulation_event_rainfall_constant_copy', 'database_manipulation_event_rainfall_csv', 'database_manipulation_event_rainfall_csv_copy', 'database_manipulation_event_rainfall_synthetic', 'database_manipulation_event_rainfall_synthetic_copy', 'event_set', 'event_set_copy', 'test_set', 'test_set_copy']
Names after `Delete`: ['database_manipulation_event', 'database_manipulation_event_rainfall_constant', 'database_manipulation_event_rainfall_csv', 'database_manipulation_event_rainfall_synthetic', 'event_set', 'test_set']
Description before `Edit`: Some event description
Description after `Edit`: Updated description
Back to top