Recipes¶

async with httpx.AsyncClient(timeout=120, proxy="http://proxy:3128") as client:
    ems = RapidMapping(client, directory=Path("data/ems"))
    events = await ems.observed_events("EMSR762")

from copernicus_ems import fetch_layer, fetch_observed_events, fetch_products

events = await fetch_observed_events(client, "EMSR762", directory=Path("data/ems"))
products = await fetch_products(client, "EMSR762")
depth = await fetch_layer(client, products[0], "floodDepthA")  # None if absent

from datetime import date

autumn = await ems.activations(country="Italy", since=date(2024, 9, 1), until=date(2024, 11, 30))

from copernicus_ems import MASS_MOVEMENT_EVENT_TYPE

landslides = await ems.observed_events(
    "EMSR762", event_types=[MASS_MOVEMENT_EVENT_TYPE], notations=None
)
everything = await ems.observed_events("EMSR762", event_types=None, notations=None)

import shapely  # pip install shapely
from shapely.geometry import shape

mask = shape(event.product.aoi_extent)  # GeoJSON; or shapely.from_wkt(aoi_extent_wkt)
flooded = shapely.union_all([shape(f["geometry"]) for f in event.features])
observed_dry = mask.difference(flooded)

from copernicus_ems import EmsError

try:
    events = await ems.observed_events("EMSR762")
except EmsError as exc:
    print("Rapid Mapping said no:", exc)  # HTML instead of GeoJSON, …
except httpx.HTTPError as exc:
    print("network:", exc)  # after the retries of RapidMapping

import httpx

from copernicus_ems import detail_url, parse_detail, parse_layer, select_features

with httpx.Client(timeout=60, follow_redirects=True) as http:
    products = parse_detail(http.get(detail_url("EMSR762")).content)
    for product in products:
        if (layer := product.layers.get("observedEventA")) is not None:
            floods = select_features(parse_layer(http.get(layer.url).content))
            print(product.label, len(floods))