Storm Surge Studio#

An interactive coastal atlas for exploring a synthetic, toy storm. Terrain, landmarks, tides, wind effects, flood extents, and exposure estimates are invented for this demonstration. Not a forecast or a tool for real-world decisions.

The app keeps three Bokeh figures alive for the whole session. Widgets drive parameters on a Viewer, a derived scenario parameter caches the expensive flood fill, watchers push new data into the existing ColumnDataSources and Spans, and the text readouts are bound to param.depends methods.

from collections import deque
from dataclasses import dataclass

import numpy as np
import panel as pn
import param
from bokeh.events import Tap
from bokeh.models import (
    ColorBar, ColumnDataSource, HoverTool, LabelSet, LinearAxis, LinearColorMapper,
    Range1d, Span, TapTool,
)
from bokeh.plotting import figure

pn.extension(sizing_mode="stretch_width", throttled=True)

A synthetic coast#

The palette, a 70 x 90 elevation grid with a wavy shoreline and two inland depressions, and six landmarks. Everything here is invented.

NAVY = "#0a1e2d"
PANEL = "#102b3b"
INK = "#e8eee8"
MUTED = "#a7bdc0"
TEAL = "#79d4cb"
TEAL_LIGHT = "#98e4d9"
CORAL = "#ff9279"
SAND = "#e2c797"
OCEAN = "#0c5864"
EDGE = "#42606a"
GRID = "#31505e"
HOVER = "#254d59"
TERRAIN_PALETTE = ["#193f57", "#346e7a", "#5a9290", "#8ab5a3", "#b8c6a8",
                   "#e1d4aa", "#d9ba88", "#c6a275", "#aa8965", "#8e785e", "#756c5c"]

NY, NX = 70, 90
KM_PER_CELL = 0.12
WIDTH_KM, HEIGHT_KM = NX * KM_PER_CELL, NY * KM_PER_CELL
HOURS = np.arange(25)
X = (np.arange(NX) + 0.5) * KM_PER_CELL
Y = (np.arange(NY) + 0.5) * KM_PER_CELL
yy, xx = np.mgrid[:NY, :NX]
COAST = 20 + 2.2 * np.sin(np.arange(NY) / 7) + 1.1 * np.sin(np.arange(NY) / 2.7)
LAND = xx >= COAST[:, None]
TERRAIN = (
    0.08 * (xx - COAST[:, None])
    + 0.22 * np.sin(xx / 5 + yy / 9)
    + 0.16 * np.sin(xx / 2.5 - yy / 6)
    - 1.05 * np.exp(-((xx - 44) / 10) ** 2 - ((yy - 34) / 11) ** 2)
    - 0.48 * np.exp(-((xx - 38) / 7) ** 2 - ((yy - 53) / 9) ** 2)
)
LEVEE_COL, LEVEE_HEIGHT = 35, 3.9
ASSETS = [("Marina", 26, 18), ("Harbor", 29, 48), ("Clinic", 38, 32),
          ("Substation", 44, 44), ("School", 53, 28), ("Pump house", 48, 56)]
WIND = {"Onshore": 1.0, "Alongshore": 0.72, "Offshore": 0.4}

Model and helpers#

The toy model is plain module-level functions: a tide plus a Gaussian storm pulse, a flood fill from the sea edge, and the conversions from a wet mask to an RGBA image or profile patches. simulate bundles a full 24-hour run into a Scenario, which only has to be recomputed when the storm or levee changes, not when scrubbing hours or moving the transect.

def stillwater(hour, intensity, wind):
    tide = 0.34 + 0.52 * np.sin(2 * np.pi * (hour - 2) / 12)
    pulse = intensity * WIND[wind] * np.exp(-0.5 * ((hour - 14) / 3.5) ** 2)
    return tide + pulse


def connected_water(elevation, level):
    # Only cells reachable from the open-water (western) edge can flood.
    passable = elevation <= level
    wet = np.zeros_like(passable)
    queue = deque((row, 0) for row in range(NY) if passable[row, 0])
    for row, col in queue:
        wet[row, col] = True
    while queue:
        row, col = queue.popleft()
        for nr, nc in ((row - 1, col), (row + 1, col), (row, col - 1), (row, col + 1)):
            if 0 <= nr < NY and 0 <= nc < NX and passable[nr, nc] and not wet[nr, nc]:
                wet[nr, nc] = True
                queue.append((nr, nc))
    return wet


@dataclass(frozen=True, eq=False)
class Scenario:
    elevation: np.ndarray
    levels: np.ndarray
    wet: np.ndarray
    exposure: np.ndarray
    inundated_pct: np.ndarray


def simulate(intensity, wind, barrier):
    elevation = TERRAIN.copy()
    if barrier:
        elevation[:, LEVEE_COL] = np.maximum(elevation[:, LEVEE_COL], LEVEE_HEIGHT)
    levels = np.array([stillwater(hour, intensity, wind) for hour in HOURS])
    wet = np.array([connected_water(elevation, level) for level in levels])
    exposure = np.array([sum(int(mask[row, col]) for _, col, row in ASSETS) for mask in wet])
    inundated_pct = 100 * (wet & LAND).sum(axis=(1, 2)) / LAND.sum()
    return Scenario(elevation, levels, wet, exposure, inundated_pct)


def rgba(hex_color, alpha=255):
    # image_rgba reads uint32 pixels in little-endian byte order: R is the low byte.
    r, g, b = bytes.fromhex(hex_color.lstrip("#"))
    return np.uint32((alpha << 24) | (b << 16) | (g << 8) | r)


def water_pixels(wet):
    pixels = np.zeros((NY, NX), dtype=np.uint32)
    pixels[~LAND] = rgba(OCEAN)
    pixels[wet & LAND] = rgba(CORAL, 0xd8)
    return pixels


def flood_patches(profile, row_wet, level):
    edges = np.flatnonzero(np.diff(np.r_[False, row_wet, False]))
    spans = list(zip(edges[::2], edges[1::2]))
    return {
        "xs": [np.r_[X[start:end], X[end - 1], X[start]] for start, end in spans],
        "ys": [np.r_[profile[start:end], level, level] for start, end in spans],
    }


def coastal_plot(height, **kwargs):
    plot = figure(height=height, sizing_mode="stretch_width", toolbar_location=None,
                  background_fill_color=PANEL, border_fill_color=PANEL,
                  outline_line_color=EDGE, min_border=44, **kwargs)
    for axis in (plot.xaxis, plot.yaxis):
        axis.axis_line_color = MUTED
        axis.major_label_text_color = MUTED
        axis.axis_label_text_color = INK
    for grid in (plot.xgrid, plot.ygrid):
        grid.grid_line_color = GRID
        grid.grid_line_alpha = 0.25
    return plot


def kicker(text, **sx):
    return pn.ui.Typography(text, sx={"color": TEAL, "fontSize": "0.74rem", "letterSpacing": "0.13em", **sx})


def heading(text, size):
    return pn.ui.Typography(text, sx={"fontFamily": "Georgia, serif", "fontSize": size, "color": INK})


def card(*objects):
    return pn.ui.Paper(*objects, sx={"p": 2, "backgroundColor": PANEL, "border": f"1px solid {GRID}"})

State, figures and bindings#

scenario is derived from intensity, wind and barrier with on_init=True, so it exists as soon as the object does. The figures and their sources are created before super().__init__ because that initial scenario already fires the watchers that fill them. After that, Bokeh data only ever changes in place: watchers update ColumnDataSource.data and Span.location, while the readouts are Typography panes bound directly to param.depends methods. __panel__ only arranges these persistent pieces into the page.

# pyright: reportAssignmentType=false
WIND_TOGGLE_SX = {
    "& .MuiToggleButton-root": {"color": INK, "borderColor": EDGE},
    "& .MuiToggleButton-root:hover": {"backgroundColor": HOVER},
    "& .MuiToggleButton-root.Mui-selected": {"color": NAVY, "backgroundColor": TEAL, "borderColor": TEAL},
    "& .MuiToggleButton-root.Mui-selected:hover": {"backgroundColor": TEAL_LIGHT},
    "& .MuiToggleButton-root.Mui-focusVisible": {"outline": f"2px solid {SAND}", "outlineOffset": "2px"},
}


class StormSurgeStudio(pn.viewable.Viewer):
    hour: int = param.Integer(default=14, bounds=(0, 24))
    intensity: float = param.Number(default=2.0, bounds=(0, 3), step=0.1)
    wind: str = param.Selector(default="Onshore", objects=list(WIND))
    barrier: bool = param.Boolean(default=False)
    transect: int = param.Integer(default=34, bounds=(0, NY - 1))
    selected: str = param.Selector(default="Clinic", objects=[item[0] for item in ASSETS])
    scenario: Scenario = param.ClassSelector(class_=Scenario, precedence=-1)
    reset: bool = param.Event()

    def __init__(self, **params):
        self._map = self._make_map()
        self._section = self._make_section()
        self._timeline = self._make_timeline()
        super().__init__(**params)
        self._hour = pn.ui.IntSlider.from_param(self.param.hour, label="Hour | synthetic 24 h", margin=(10, 18))
        self._intensity = pn.ui.FloatSlider.from_param(
            self.param.intensity, label="Onshore-equivalent storm pulse | m", margin=(10, 18)
        )
        self._wind = pn.ui.RadioButtonGroup.from_param(
            self.param.wind, label="Wind orientation", orientation="vertical", sx=WIND_TOGGLE_SX
        )
        self._barrier = pn.ui.Switch.from_param(self.param.barrier, label=f"Raise continuous {LEVEE_HEIGHT} m inland levee")
        self._transect = pn.ui.IntSlider.from_param(self.param.transect, label="Profile line | grid row", margin=(10, 18))
        self._landmark = pn.ui.Select.from_param(self.param.selected, label="Landmark to inspect")
        self._reset = pn.ui.Button.from_param(self.param.reset, label="Reset scenario", icon="restart_alt", variant="outlined")

    def _make_map(self):
        plot = coastal_plot(465, x_range=(0, WIDTH_KM), y_range=(0, HEIGHT_KM),
                            x_axis_label="Eastward distance (km)", y_axis_label="Alongshore distance (km)")
        mapper = LinearColorMapper(palette=TERRAIN_PALETTE, low=-2, high=5)
        plot.image(image=[TERRAIN], x=0, y=0, dw=WIDTH_KM, dh=HEIGHT_KM, color_mapper=mapper)
        self._water_source = ColumnDataSource(data={"image": [np.zeros((NY, NX), dtype=np.uint32)]})
        plot.image_rgba(image="image", x=0, y=0, dw=WIDTH_KM, dh=HEIGHT_KM, source=self._water_source)
        plot.line((COAST + 0.5) * KM_PER_CELL, Y, line_color=SAND, line_width=2, line_alpha=0.9)
        self._barrier_source = ColumnDataSource(data={"x": [], "y": []})
        plot.line("x", "y", source=self._barrier_source, line_color=CORAL, line_width=4, line_dash="dashed")
        self._transect_source = ColumnDataSource(data={"x": [0, WIDTH_KM], "y": [0, 0]})
        plot.line("x", "y", source=self._transect_source, line_color=INK, line_width=2, line_dash="dotted")
        self._asset_source = ColumnDataSource(
            data={"x": [], "y": [], "label": [], "status": [], "color": [], "elevation": []}
        )
        dots = plot.scatter("x", "y", source=self._asset_source, size=13,
                            fill_color="color", line_color=NAVY, line_width=2)
        plot.add_layout(LabelSet(x="x", y="y", text="label", source=self._asset_source,
                                 x_offset=9, y_offset=6, text_color=INK, text_font_size="10px",
                                 text_font_style="bold"))
        plot.add_layout(ColorBar(color_mapper=mapper, title="Elevation | m",
                                 title_text_color=INK, major_label_text_color=MUTED,
                                 border_line_color=None, background_fill_alpha=0), "right")
        plot.add_tools(HoverTool(renderers=[dots], tooltips=[("Landmark", "@label"),
                       ("Status", "@status"), ("Ground", "@elevation{0.00} m")]), TapTool())
        plot.on_event(Tap, self._on_map_tap)
        return plot

    def _make_section(self):
        plot = coastal_plot(370, x_range=(0, WIDTH_KM), y_range=(-2.6, 6),
                            x_axis_label="Eastward distance (km)", y_axis_label="Elevation (m)")
        self._fill_source = ColumnDataSource(data={"xs": [], "ys": []})
        plot.patches("xs", "ys", source=self._fill_source, fill_color=CORAL, fill_alpha=0.65, line_alpha=0)
        self._profile_source = ColumnDataSource(data={"x": X, "z": np.zeros(NX)})
        profile = plot.line("x", "z", source=self._profile_source, line_color=SAND, line_width=3)
        self._sea_line = Span(location=0, dimension="width", line_color=TEAL, line_width=2, line_dash="dashed")
        plot.add_layout(self._sea_line)
        plot.add_tools(HoverTool(renderers=[profile], tooltips=[("Distance", "@x{0.00} km"),
                                                               ("Ground", "@z{0.00} m")], mode="vline"))
        return plot

    def _make_timeline(self):
        plot = coastal_plot(245, x_range=(0, 24), y_range=(-0.6, 4),
                            x_axis_label="Hour of synthetic day", y_axis_label="Stillwater (m)")
        self._timeline_source = ColumnDataSource(
            data={"hour": HOURS, "level": np.zeros(len(HOURS)), "exposed": np.zeros(len(HOURS), dtype=int)}
        )
        plot.line("hour", "level", source=self._timeline_source, color=TEAL, line_width=3, legend_label="Stillwater | m")
        plot.extra_y_ranges = {"assets": Range1d(0, len(ASSETS) + 1)}
        plot.add_layout(LinearAxis(y_range_name="assets", axis_label="Exposed landmarks",
                                   axis_label_text_color=CORAL, major_label_text_color=CORAL,
                                   axis_line_color=CORAL), "right")
        plot.line("hour", "exposed", source=self._timeline_source, y_range_name="assets",
                  color=CORAL, line_width=2, line_dash="dashed", legend_label="Exposed landmarks | count")
        self._hour_marker = Span(location=0, dimension="height", line_color=SAND, line_width=2)
        plot.add_layout(self._hour_marker)
        plot.legend.location = "top_left"
        plot.legend.background_fill_color = PANEL
        plot.legend.label_text_color = INK
        plot.legend.border_line_alpha = 0
        plot.add_tools(HoverTool(tooltips=[("Hour", "@hour h"), ("Water", "@level{0.00} m"),
                                           ("Exposed", "@exposed landmarks")], mode="vline"))
        return plot

    def _on_map_tap(self, event):
        if event.x is None or event.y is None:
            return
        for label, col, row in ASSETS:
            if np.hypot(event.x - X[col], event.y - Y[row]) < 0.23:
                self.param.update(selected=label, transect=row)
                return
        self.transect = int(np.clip(event.y / KM_PER_CELL, 0, NY - 1))

    @param.depends("intensity", "wind", "barrier", watch=True, on_init=True)
    def _simulate(self):
        self.scenario = simulate(self.intensity, self.wind, self.barrier)

    @param.depends("scenario", watch=True)
    def _sync_timeline(self):
        scenario = self.scenario
        self._timeline_source.data = {"hour": HOURS, "level": scenario.levels, "exposed": scenario.exposure}
        self._barrier_source.data = {"x": [X[LEVEE_COL]] * NY if self.barrier else [],
                                     "y": Y if self.barrier else []}

    @param.depends("scenario", "hour", "selected", watch=True)
    def _sync_map(self):
        wet = self.scenario.wet[self.hour]
        self._water_source.data = {"image": [water_pixels(wet)]}
        self._asset_source.data = {
            "x": [X[col] for _, col, _ in ASSETS], "y": [Y[row] for _, _, row in ASSETS],
            "label": [label for label, _, _ in ASSETS],
            "status": ["Inundated | toy" if wet[row, col] else "Dry | toy" for _, col, row in ASSETS],
            "color": [SAND if label == self.selected else CORAL if wet[row, col] else TEAL
                      for label, col, row in ASSETS],
            "elevation": [self.scenario.elevation[row, col] for _, col, row in ASSETS],
        }
        self._hour_marker.location = self.hour

    @param.depends("scenario", "hour", "transect", watch=True)
    def _sync_section(self):
        level = self.scenario.levels[self.hour]
        profile = self.scenario.elevation[self.transect]
        self._transect_source.data = {"x": [0, WIDTH_KM], "y": [Y[self.transect]] * 2}
        self._profile_source.data = {"x": X, "z": profile}
        self._sea_line.location = level
        self._fill_source.data = flood_patches(profile, self.scenario.wet[self.hour, self.transect], level)

    @param.depends("scenario", "hour")
    def _readout(self):
        return f"H{self.hour:02d}  /  {self.scenario.levels[self.hour]:.2f} m stillwater"

    @param.depends("scenario", "hour")
    def _metrics(self):
        return (f"{self.scenario.exposure[self.hour]} / {len(ASSETS)} landmarks exposed  |  "
                f"{self.scenario.inundated_pct[self.hour]:.1f}% of land cells inundated")

    @param.depends("transect")
    def _section_title(self):
        return f"TRANSECT AT {Y[self.transect]:.2f} KM ALONGSHORE"

    @param.depends("scenario", "hour", "selected")
    def _asset_note(self):
        label, col, row = next(asset for asset in ASSETS if asset[0] == self.selected)
        status = "connected to sea water" if self.scenario.wet[self.hour, row, col] else "dry in this scenario"
        return (f"Selected landmark: {label} | ground {self.scenario.elevation[row, col]:.2f} m | {status}. "
                "The landmark need not lie on this profile. "
                "Dashed line is stillwater; coral fill shows only connected water.")

    @param.depends("reset", watch=True)
    def _on_reset(self):
        controls = ("hour", "intensity", "wind", "barrier", "transect", "selected")
        self.param.update({name: self.param[name].default for name in controls})

    def __panel__(self):
        sidebar = pn.ui.Column(
            kicker("FIELD NOTES / 01", letterSpacing="0.2em", fontSize="0.7rem"),
            heading("Shape the storm", "1.5rem"),
            pn.ui.Typography(
                "Move through one invented day. Click the map to cut a new transect; click a landmark to inspect it.",
                sx={"color": MUTED},
            ),
            self._hour, self._intensity, self._wind, self._barrier, self._transect, self._landmark, self._reset,
            pn.ui.Divider(),
            kicker("MODEL KEY", letterSpacing="0.2em", fontSize="0.7rem"),
            pn.ui.Typography(
                "Coral = connected flooded land. Deep teal = open water. A submerged inland bowl "
                "stays dry until water can reach it. The inland levee is seaward of the Clinic, "
                "but Marina and Harbor are in front of it. Wind scales the storm pulse: "
                "onshore 100%, alongshore 72%, offshore 40%.",
                sx={"color": MUTED, "fontSize": "0.85rem"},
            ),
            min_width=255, sx={"gap": "12px"},
        )
        map_panel = card(
            kicker("01 / INUNDATION ATLAS"),
            heading("The shape of the shoreline", "1.45rem"),
            pn.ui.Bokeh(self._map),
        )
        section_panel = card(
            kicker("02 / COASTAL PROFILE"),
            pn.ui.Typography(self._section_title, sx={"color": SAND, "fontSize": "0.86rem", "letterSpacing": "0.08em"}),
            pn.ui.Bokeh(self._section),
            pn.ui.Typography(self._asset_note, sx={"color": MUTED, "fontSize": "0.87rem"}),
        )
        timeline_panel = card(
            kicker("03 / THE 24-HOUR WINDOW"),
            heading("Water level & landmark exposure", "1.35rem"),
            pn.ui.Bokeh(self._timeline),
        )
        metrics = pn.ui.Column(
            kicker("LIVE READING / SYNTHETIC", fontSize="0.72rem", letterSpacing="0.17em"),
            pn.ui.Typography(self._readout, sx={"fontFamily": "Georgia, serif", "fontSize": "1.65rem", "color": TEAL}),
            pn.ui.Typography(self._metrics, sx={"fontSize": "1.06rem", "color": INK}),
            sx={"py": 2, "borderTop": f"1px solid {EDGE}", "borderBottom": f"1px solid {EDGE}"},
        )
        atlas = pn.ui.Grid(
            pn.ui.Grid(map_panel, size={"xs": 12, "lg": 8}),
            pn.ui.Grid(section_panel, size={"xs": 12, "lg": 4}),
            container=True, spacing=2,
        )
        main = pn.ui.Column(
            kicker("A COAST THAT CHANGES BY THE HOUR", letterSpacing="0.21em", fontSize="0.73rem"),
            heading("Read the rising water", {"xs": "2rem", "md": "2.8rem"}),
            pn.ui.Typography(
                "An invented estuary, one passing storm, and the difference between low ground "
                "and reachable water. Open the menu to shape the storm. All numbers are synthetic toy estimates, not a forecast.",
                sx={"color": MUTED, "maxWidth": "760px"},
            ),
            metrics, atlas, timeline_panel,
            pn.ui.Typography(
                "METHOD / 70 x 90 invented elevation cells | 120 m per cell | four-neighbor "
                "flood fill from the west sea boundary | sinusoidal toy tide + Gaussian storm "
                "pulse. Exposure means a landmark cell is connected to the sea below "
                "stillwater level; it does not represent damage or risk.",
                sx={"color": MUTED, "fontSize": "0.76rem", "py": 2},
            ),
            sx={"gap": "20px"}, margin=10,
        )
        # The coast is designed as a dark chart, so both theme modes share one palette.
        theme = {
            "palette": {"primary": {"main": TEAL, "contrastText": NAVY},
                        "text": {"primary": INK, "secondary": MUTED},
                        "background": {"default": NAVY, "paper": PANEL}},
            "typography": {"fontFamily": "Arial, sans-serif"},
        }
        return pn.ui.Page(
            title="Storm Surge Studio", sidebar=[sidebar],
            main=[pn.ui.Container(main, width_option="xl")],
            sidebar_open=False, main_width={"xs": "100%", "xl": 1600},
            theme_config={"light": theme, "dark": theme},
            sx={"& .header": {"backgroundColor": NAVY, "backgroundImage": "none", "boxShadow": "none"},
                "& .MuiIconButton-root": {"color": INK}},
        )

Preview and serve#

The preview displays the same studio object that panel serve storm_surge_studio.ipynb serves.

studio = StormSurgeStudio()
studio
if pn.state.served:
    studio.servable()