Recipes¶

Keep a national copy fresh, every night¶

from ispra_idrogeo import LANDSLIDE_LAYERS, IdroGEO, mosaic, opendata

async with IdroGEO() as idrogeo:
    for layer in LANDSLIDE_LAYERS:
        got = await idrogeo.download(opendata(layer))
        print(layer, "updated" if got.changed else "unchanged")
    for name in ("pai_frane", "idraulica"):
        await idrogeo.fetch(mosaic(name))

The first run downloads about 3 GB. Put it in a cron job: the following runs cost a few kilobytes, plus whatever ISPRA really changed.

Only what changed since last time¶

The nightly download above re-reads a whole file when anything in it changed. To keep your own copy (a database, a GeoParquet file) up to date, ask the API for the records changed since your last run instead:

from datetime import UTC, datetime

from ispra_idrogeo import IdroGEO

since = datetime(2026, 11, 1, tzinfo=UTC)  # saved from your last run
async with IdroGEO() as idrogeo:
    changed = [f async for f in idrogeo.iter_landslides(modified_since=since, select=["*"])]
print(len(changed))  # 20 landslides changed in two days, with their geometry
next_since = max(datetime.fromisoformat(f["modified"]) for f in changed)

Save the largest modified you received and pass it next time. The filter is "at or after", so the last record comes back once more: write by id_frana (upsert) and it does no harm.
select=["*"] brings every field and the geometry: geom in EPSG:32632, geom_polygon / geom_linestring when there is one, point in lon/lat.
iter_events and sites take modified_since too.
A removed landslide does not show up as changed: it just stops being listed. Now and then, compare your ids with the full list, one call of about 15 seconds for 721,377 ids:

ids = {f["id_frana"] async for f in idrogeo.iter_landslides(select=["id_frana"], page_size=100_000)}

Watch for changes¶

idrogeo watch --once >> changes.jsonl                 # one round, e.g. every hour from cron
idrogeo watch --kind events --interval 900            # or keep running, every 15 minutes

Each line is a landslide, an event or a monitoring site that changed since the previous round, with "kind" saying which. The first run starts from now (or from --since); after that <dir>/watch.json remembers where each kind stands, so a stopped watch resumes without missing or repeating a change. In Python, ispra_idrogeo.watch.changes(idrogeo, kind, mark) does one round.

Landslides per region (the map above)¶

from ispra_idrogeo import IdroGEO

async with IdroGEO() as idrogeo:
    for region in await idrogeo.summary("iffi", "regioni"):
        counts = await idrogeo.summary("iffi", "regioni", region["cod_reg"])
        hazard = await idrogeo.summary("pir", "regioni", region["cod_reg"])
        density = counts["frane_totali"] / hazard["ar_kmq"] * 100
        print(f"{region['nome']:<22} {counts['frane_totali']:>8,}  {density:5.0f} per 100 km²")
# Piemonte                 37,786    149 per 100 km²
# Valle d'Aosta             5,812    178 per 100 km²
# Lombardia               141,970    595 per 100 km²
# …

summary("iffi", …) gives landslide counts (also by type, in frane_stats); summary("pir", …) gives hazard and risk indicators: areas, residents, families, buildings, businesses and cultural heritage in each landslide (*_fr_*) and flood (*_idr_*) hazard class. Levels are "italia", "macroregioni", "regioni", "province" and "comuni", by ISTAT code.

Every municipality, in one table¶

from datetime import timedelta

from ispra_idrogeo import IdroGEO

async with IdroGEO() as idrogeo:
    rows = await idrogeo.municipalities("pir", max_age=timedelta(days=30))
print(len(rows))  # 7899
worst = sorted(rows, key=lambda r: r["pop_fr_p3"] + r["pop_fr_p4"], reverse=True)[:3]
for r in worst:
    print(r["nome"], r["pop_fr_p3"] + r["pop_fr_p4"], f"{r['popfrp3p4p']:.1f} %")
# Napoli 42390 4.6 %     ← residents in landslide hazard P3 + P4, and their share
# Genova 30148 5.4 %
# Sanremo 14607 27.6 %

One call lists the municipalities, then one call each asks for its indicators, 8 at a time: about a minute and 17 MB for all of Italy. Each answer is saved as <directory>/pir_comuni/<ISTAT code>.json, so the next run reads it back from disk (under two seconds), an interrupted run resumes where it stopped, and max_age decides when an answer is old enough to ask again. The indicators come from the 2020-2021 hazard mosaics and the 2021 census, so a monthly refresh is plenty.

municipalities("iffi") does the same with landslide counts, and summaries(summary, "regioni" | "province" | "comuni") with any level.

Translate the codes¶

The data stores codes (tipo_movimento = 4); the API has 73 code lists that say what they mean, in Italian, English, French and German. decode adds a <field>_label next to every coded field it knows:

from ispra_idrogeo import IdroGEO, decode

async with IdroGEO() as idrogeo:
    labels = await idrogeo.code_labels(lang="en")  # one call per code list
    record = decode(await idrogeo.landslide("0017680500"), labels)
print(record["tipo_movimento_label"], "·", record["stato_attivita_1liv_label"])
# rapid debris flow · active/reactivated/suspended

It works on monitoring sites too (strumentazione = [15, 22, 25] becomes ['inclinometer', 'topographic monitoring', 'Satellite SAR interferometry']) and on the rows of a shapefile, where the field is called tipo_movim. CODE_DICTIONARIES lists which code list decodes which field, and labels("cause", "it") gives one code list as {code: label}. A code that is not in its list becomes None rather than a guess.

Walk through every landslide or event¶

from datetime import date

from ispra_idrogeo import IdroGEO, event_filter, landslide_filter

async with IdroGEO() as idrogeo:
    active_sinkholes = landslide_filter(movement=6, activity=100)
    print(len([f async for f in idrogeo.iter_landslides(active_sinkholes)]))  # 217

    tuscany_2023 = event_filter(region=9, start=date(2023, 1, 1), end=date(2024, 1, 1))
    async for event in idrogeo.iter_events(tuscany_2023):  # 8 events
        print(event["uid"], event["computed_date"], event["comune_nome"])

iter_landslides and iter_events ask for one page after another (page_size, 1,000 by default) and stop at the last one. They sort by a unique key (id_frana, uid) so pages do not overlap. An empty filter matches everything; search_landslides and search_events return a single page, if you prefer to page yourself.

For the landslides of an area, iter_landslides_in checks each record's own region, province or municipality, because the search has no reliable area filter (see the traps). It reads the whole inventory with a few fields, about 20 seconds for Italy, and gives the exact counts (Campania 55,254, Sarno 53):

async with IdroGEO() as idrogeo:
    sarno = [f async for f in idrogeo.iter_landslides_in(municipality=65135)]
    debris_flows = idrogeo.iter_landslides_in(region=15, filters=landslide_filter(movement=5))

landslide_filter, event_filter and site_filter build the filters, because each search reads them its own way: landslides want {"tipo_movimento": 6} but {"stato_attivita_1liv": "eq.100"}, events and sites want operators everywhere, and a range of dates is an "and" of two conditions. Codes are ISTAT codes for areas and the code lists' codes for the rest; a list of codes means "any of these". You can still pass a plain dict if you know the syntax.

One landslide, in full¶

async with IdroGEO() as idrogeo:
    record = await idrogeo.landslide("0017680500")
print(len(record), record["stato"])  # 142 validato_ispra
# activations, causes, state and velocity, damage, interventions, precursors…

The full record of a validated landslide. Many fields are often empty, so check before you rely on them. An unknown id raises IdrogeoError (IdroGEO answers 401 to it, not 404).

API rows on a map¶

frame turns API rows into a GeoDataFrame in the right coordinate system:

from ispra_idrogeo.geo import frame

async with IdroGEO() as idrogeo:
    sites = frame(await idrogeo.sites())  # "point": lon/lat, EPSG:4326
    cams = frame(await idrogeo.webcams(), geometry="geom")  # EPSG:3857, as IdroGEO sends it
    changed = frame(
        [f async for f in idrogeo.iter_landslides(modified_since=since, select=["*"])],
        geometry="geom_polygon",  # or "geom" (point, EPSG:32632), "geom_linestring"
    )

Who is watching a landslide: monitoring sites and webcams¶

async with IdroGEO() as idrogeo:
    active = await idrogeo.sites({"stato_monitoraggio": "eq.1"})
    print(len(active))  # 558 of 1,063 sites are active
    site = active[0]
    print(site["uid"], site["point"], site["id_frane_correlati"])
    # SIABR001 [14.03, 42.24] ['0680013800']   ← lon/lat, and its IFFI landslides
    for station in await idrogeo.stations(site["uid"]):
        print(station["uid"], station["strumentazione"])

    cams = await idrogeo.webcams()  # 14, all in Lombardy

A site is a place where an agency (often an ARPA) monitors a landslide. It has stations, and each station has instruments (inclinometers, piezometers, rain gauges…, decoded by the strumento_monitoraggio dictionary). id_frane_correlati links 880 sites to their IFFI landslides, so you can tell which landslides someone is already watching. Look a site up by its uid ("SIPIE001"), not by its id.

Is this place at risk? Hazard at a point¶

async with IdroGEO() as idrogeo:
    here = await idrogeo.hazard_at(40.829, 14.617)  # the hills above Sarno
print(here.landslide, here.flood, here.municipality_name, here.slope_deg)
# P4 None Sarno 12.3

async with IdroGEO() as idrogeo:
    venice = await idrogeo.hazard_at(45.438, 12.330)
print(venice.flood, venice.flood_scenarios)  # P3 ('P3', 'P2', 'P1')

One request, about a second, to IdroGEO's public map service: no 1.2 GB mosaic to download. You get the PAI landslide class (P4 very high, P3 high, P2 medium, P1 moderate, AA attention area), the flood scenarios that reach the point (P3 frequent floods, P2 every 100–200 years, P1 rare and extreme), the municipality (ISTAT code, ready for summary("pir", "comuni", code)) and the slope in degrees. lithology=True adds the rock type, but takes about 16 seconds more.

Landslides near a point¶

async with IdroGEO() as idrogeo:
    near = await idrogeo.landslides_near(40.838, 14.630, radius_m=500)
for f in near[:3]:
    print(f["id_frana"], f["nome_tipo"], f["nome_com"], f["distance_m"], f["shapes"])
# 0640689600 Colamento rapido Quindici 397.7 ['polygon', 'point']   ← a rapid debris flow

Every IFFI landslide within radius_m (up to 5 km), nearest first, with its type and area, distance_m (0 when the point is on it) and the shapes it is mapped as. The distance is computed from the real outline, not from a centre point. Also: idrogeo hazard 40.829 14.617 and idrogeo near 40.838 14.630.

Offline, many places at once, your own SQL: a local warehouse¶

hazard_at asks the map service one point at a time. For thousands of points, for work offline, or for questions of your own, pip install "ispra-idrogeo[warehouse]" keeps the layers in one DuckDB file:

idrogeo sync frane_piff pai_frane idraulica   # once; then only what changed
idrogeo sql "SELECT nome_reg, count(*) AS n FROM frane_piff GROUP BY 1 ORDER BY n DESC"
idrogeo sql                                    # the tables and their columns

from ispra_idrogeo.warehouse import Warehouse

async with IdroGEO() as idrogeo:
    with Warehouse() as warehouse:  # ~/.cache/ispra-idrogeo/idrogeo.duckdb
        await warehouse.sync(idrogeo, ["frane_piff", "pai_frane", "idraulica"])
        warehouse.hazard_at(40.8205, 14.6297)  # Sarno: {'landslide': 'P4', 'flood': None, …}
        warehouse.landslides_in(municipality=65135)  # Sarno: 53 landslides, by type
        warehouse.query(
            "SELECT p.class, count(*) AS landslides FROM frane_piff f "
            "JOIN pai_frane p ON ST_Intersects(p.geom, f.geom) "
            "WHERE f.regione = ? GROUP BY 1 ORDER BY 1",
            [14],
        )  # Molise, in 9 s: P4 2,904 · P3 8,242 · P2 569 · P1 2,377 · AA 659
        # (IFFI points by the class of the PAI polygons they fall in; a point in
        # two overlapping polygons counts twice)

sync downloads each layer only if it changed (as fetch does) and loads it with an R-tree index, in one transaction: a failed load leaves the old table in place. Every table is in EPSG:32632 metres, the flood mosaic reprojected from EPSG:3035; pai_frane and idraulica carry a class column (P4 … P1, AA; P3 … P1). Points go in as latitude and longitude, as in hazard_at; geometries come out of query as WKT.

With the three national layers the file weighs 2.6 GB. Loading took 9 s for the 721,377 landslide points, 55 s for the 928,542 landslide hazard polygons and 9 minutes for the 85,460 flood polygons, whose reprojection is the slow part; a point query then takes 10–40 ms. Checked on 4 November 2026: the landslides counted per municipality match the API (Sarno 53, Casamicciola Terme 213, Trivento 921), the points agree with hazard_at (Sarno P4, Venice flood P3), and the flood areas add up to 16,283, 30,271 and 42,460 km² for P3, P2 and P1, against the API's 16,224, 30,196 and 42,376 km².

Warehouse(read_only=True) opens the file for readers (the MCP server, a notebook, idrogeo sql) and switches off DuckDB's access to files, URLs and other databases, so SQL written by someone else, or by an agent, reads the tables and nothing more. DuckDB downloads its spatial extension the first time.

Geology, aerial photos, bibliography¶

async with IdroGEO() as idrogeo:
    units = await idrogeo.lookup("formazioni", "argille", limit=2)
    # [{'sigla': 'AAC', 'nome_unita': 'ARGILLE E ARENARIE GLAUCONITICHE DI CATENANUOVA', …}, …]
    flights = await idrogeo.lookup("voli", "1954")  # [{'id': '0001', 'nome': 'Volo GAI', …}]

The reference lists behind the landslide records, searched by text: geological units (formazioni), aerial survey flights (voli), bibliography (bibliografie), institutions (istituzioni), professional bodies (ordini) and funding sources (finanziamenti). Ten answers by default; limit asks for more. Also idrogeo lookup formazioni argille.

For AI agents: the MCP server¶

pip install "ispra-idrogeo[mcp]"

idrogeo-mcp is an MCP server, so Claude Desktop, an IDE or your own agent can answer questions like "how many people live in high landslide hazard areas in Sarno, and is anyone monitoring the landslides there?" or "is this address in a flood zone?". For Claude Desktop, add it to claude_desktop_config.json:

{
  "mcpServers": {
    "idrogeo": { "command": "idrogeo-mcp" }
  }
}

As an always-on service, for an agent that talks HTTP (for example mcp-geo-server's chat and map), serve it over streamable HTTP at /mcp:

idrogeo-mcp --transport streamable-http --host 0.0.0.0 --port 8767

Every option also reads an environment variable, so a compose file needs no command line, and GET /health answers {"status": "ok"}:

services:
  idrogeo-mcp:
    image: ghcr.io/astral-sh/uv:python3.12-bookworm-slim
    command: uvx --from "ispra-idrogeo[mcp]>=0.6" idrogeo-mcp
    environment:
      IDROGEO_MCP_TRANSPORT: streamable-http
      IDROGEO_MCP_HOST: 0.0.0.0
      IDROGEO_MCP_PORT: "8767"
      IDROGEO_DATA_DIR: /data/idrogeo  # for download_dataset
    volumes:
      - idrogeo-data:/data/idrogeo
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8767/health')"]

On 127.0.0.1 the SDK refuses requests with a foreign Host header (DNS rebinding protection); on 0.0.0.0 it does not, unless you list the names clients use: IDROGEO_MCP_ALLOWED_HOSTS=idrogeo-mcp:*,localhost:*.

Tool	What the agent gets
`hazard_at`	Landslide and flood hazard, municipality and slope of a point
`landslides_near`	The landslides around a point, nearest first
`find_area`	The ISTAT code of a region, province or municipality from its name
`area_risk`	Hazard and exposure: residents, buildings, heritage per hazard class
`area_landslides`	IFFI counts by type of movement, events, monitoring
`landslide`	One landslide in full, codes decoded
`search_landslides` / `search_events`	Landslides by type and activity; events by area and date
`monitoring_sites` / `webcams`	Who is watching, with which instruments
`rank_areas`	The top (or bottom) regions, provinces or municipalities by any indicator, optionally inside a region or province
`compare_areas`	Up to 20 areas side by side, so the agent does not do arithmetic in its head
`code_list`	Any code list, in four languages
`download_dataset`	A layer downloaded (only if changed) to the local folder
`warehouse_schema` / `warehouse_sql`	The tables of the local warehouse, and one read-only SQL query on them

Names are matched without accents or apostrophes (forli finds Forlì), and a name that matches nothing comes back with "did you mean" suggestions.

Resource	What it holds
`idrogeo://indicators`	What `popfrp3p4p`, `pop_idr_p2` and the other indicator keys mean
`idrogeo://datasets`	The layers, mosaics and area names `download_dataset` accepts
`idrogeo://codes/{name}`	One code list, in English

Two prompts, area_profile and recent_events, give a client ready-made questions ("a risk profile of Sarno", "landslide events in Liguria since October") that use the tools the right way and credit the source.

Every tool but download_dataset only reads. Searches stop at limit rows (100 by default, 1,000 at most), so an agent never receives 700,000 landslides at once, and IdroGEO's refusals come back as readable errors. rank_areas keeps the summaries it reads on disk for 30 days: the first national ranking of municipalities asks IdroGEO 7,899 times (a few minutes), a region's ranking a few hundred.

Bring your own HTTP client¶

import httpx

from ispra_idrogeo import IdroGEO

http = httpx.AsyncClient(timeout=httpx.Timeout(30, read=600), proxy="http://proxy:3128")
async with IdroGEO(http, retries=0) as idrogeo:  # your client, your retry policy
    ...
await http.aclose()  # the client is yours: IdroGEO does not close it

Handle errors¶

import httpx

from ispra_idrogeo import IdrogeoError

try:
    folder = await idrogeo.fetch(opendata("frane_piff", "puglia"))
except IdrogeoError as exc:
    # IdroGEO answered, but not with what was asked: 404, API refusal, not JSON.
    print("IdroGEO said no:", exc)
except httpx.HTTPError as exc:
    # Network problems still failing after the retries.
    print("network:", exc)

An unknown layer, area or mosaic name raises ValueError before any request.

Example notebooks¶

examples/ has three notebooks run on the live service, with their outputs: a town and its landslides (Sarno, from exposure figures to a map of its debris flows), Italy in numbers (who lives in hazard areas, municipality by municipality) and what changed (following the inventory day by day).