id: nws-severe-weather-alert-router
namespace: company.team
description: |
Watch a list of sites (offices, warehouses, stores, cell towers, event venues) for National
Weather Service alerts and route each one once: new Extreme and Severe alerts, their material
updates (upgraded, extended, area changed) and their cancellations go to Slack, everything else
is logged. Alerts are tracked as threads (the VTEC event number, or the CAP `references` chain
when there is no VTEC), so the NWS re-issuing an unchanged warning every few hours does not
page anyone again, and an alert covering several of your sites is one message listing them all.
inputs:
- id: sites
type: JSON
description: JSON array of `{name, lat, lon}` objects. Points must be inside the
NWS area (US states, territories and coastal waters); others come back as
OUT_OF_BOUNDS. Names must be unique.
defaults: |
[
{"name": "Frederick MD distribution center", "lat": 39.4143, "lon": -77.4105},
{"name": "York PA plant", "lat": 39.9626, "lon": -76.7277},
{"name": "Hatteras NC field office", "lat": 35.2524, "lon": -75.5277},
{"name": "Virginia Beach VA store", "lat": 36.8529, "lon": -75.978},
{"name": "Juneau AK depot", "lat": 58.3019, "lon": -134.4197},
{"name": "Denver CO office", "lat": 39.7392, "lon": -104.9903},
{"name": "Chicago IL office", "lat": 41.8781, "lon": -87.6298}
]
- id: route_severities
type: ARRAY
itemType: STRING
defaults:
- Extreme
- Severe
description: CAP severities that go to Slack (`Extreme`, `Severe`, `Moderate`,
`Minor`, `Unknown`). An update or cancellation is routed when the alert's
current or previous severity is in the list.
- id: contact
type: STRING
defaults: "ops@example.com"
description: Contact put in the User-Agent. api.weather.gov rejects requests
without a User-Agent and asks for a contact so it can reach you instead of
blocking you.
- id: state_key
type: STRING
defaults: nws_alert_router_state
description: KV key holding the alert threads already routed. Use a different
key per site list.
concurrency:
limit: 1
behavior: CANCEL
triggers:
- id: every_10_minutes
type: io.kestra.plugin.core.trigger.Schedule
cron: "*/10 * * * *"
description: The NWS updates the active alert feed continuously; 10 minutes
keeps latency low and the request count at 6 per site per hour.
tasks:
- id: previous_state
type: io.kestra.plugin.core.kv.Get
key: "{{ inputs.state_key }}"
errorOnMissing: false
description: Alert threads routed by earlier runs, as a JSON document. Missing
on the first run, so every active alert is NEW once (baseline).
- id: fetch_sites
type: io.kestra.plugin.core.flow.Loop
description: One request per site to `/alerts/active?point=lat,lon`, which
returns every active alert whose zone or polygon contains the point. Two
requests at a time to stay polite.
values: "{{ inputs.sites }}"
concurrencyLimit: 2
tasks:
- id: fetch
type: io.kestra.plugin.core.http.Request
description: |
`status=actual` drops test and exercise messages. A 400 (point outside the NWS area) is
accepted and reported per site; 5xx and timeouts are retried, then tolerated so one bad
site does not hide the alerts of the others.
uri: "https://api.weather.gov/alerts/active?status=actual&point={{
fromJson(item.value).lat }},{{ fromJson(item.value).lon }}"
headers:
User-Agent: "(kestra-nws-severe-weather-alert-router, {{ inputs.contact }})"
Accept: application/geo+json
options:
allowedResponseCodes: [ 200, 400, 404 ]
timeout:
readIdleTimeout: PT30S
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
allowFailure: true
outputs:
- id: site
type: JSON
value: "{{ fromJson(item.value) }}"
- id: code
type: INT
value: "{{ outputs.fetch.code ?? 0 }}"
- id: body
type: STRING
value: "{{ outputs.fetch.body ?? '' }}"
- id: responses
type: io.kestra.plugin.core.storage.Write
description: Collects the per-site responses into one JSON file for the routing step.
extension: .json
content: "{{ outputs.fetch_sites.outputs | toJson }}"
- id: route
type: io.kestra.plugin.scripts.python.Script
description: |
Groups the alerts into threads, compares them with the stored state and decides what to
route. Standard library only. Prints the summary, the notifications and the new state as
Kestra outputs and writes a per-site CSV.
taskRunner:
type: io.kestra.plugin.core.runner.Process
inputFiles:
responses.json: "{{ outputs.responses.uri }}"
state.json: "{{ outputs.previous_state.value ?? '{}' }}"
sites.json: "{{ inputs.sites | toJson }}"
outputFiles:
- site_alerts.csv
env:
ROUTE_SEVERITIES: "{{ inputs.route_severities | join(',') }}"
script: |
import csv, json, os, re
from datetime import datetime, timezone
RANK = {"Extreme": 4, "Severe": 3, "Moderate": 2, "Minor": 1, "Unknown": 0}
CLOSING = {"CAN", "EXP", "UPG"}
VTEC_RE = re.compile(r"/([OEX])\.(NEW|CON|EXT|EXA|EXB|UPG|CAN|EXP|COR|ROU)\.(\w{4})\.(\w{2})\.(\w)\.(\d{4})\.")
route_sev = {s.strip() for s in os.environ["ROUTE_SEVERITIES"].split(",") if s.strip()}
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
sites = json.load(open("sites.json"))
responses = json.load(open("responses.json"))
state = json.load(open("state.json")) or {}
threads = state.get("threads", {})
is_baseline = not state
id_to_thread = {i: k for k, t in threads.items() for i in t.get("ids", [])}
def vtecs(p):
out = []
for v in (p.get("parameters") or {}).get("VTEC", []):
m = VTEC_RE.search(v)
if m:
out.append({"action": m.group(2), "key": ".".join(m.group(3, 4, 5, 6))})
return out
# 1. Per-site fetch status and alerts, merged by alert id (one alert can cover several sites)
site_status, alerts, rows = {}, {}, []
for r in (x["outputs"] for x in responses):
site = r["site"] if isinstance(r["site"], dict) else json.loads(r["site"])
name, code = site["name"], int(r.get("code") or 0)
body = json.loads(r["body"]) if r.get("body") else {}
if code == 200:
site_status[name] = "OK"
elif code in (400, 404):
site_status[name] = "OUT_OF_BOUNDS"
print(f"{name} ({site['lat']},{site['lon']}): {body.get('detail', code)}")
else:
site_status[name] = "FETCH_FAILED"
for f in body.get("features", []) if code == 200 else []:
p = f["properties"]
a = alerts.setdefault(p["id"], {"p": p, "sites": set()})
a["sites"].add(name)
if code != 200 or not body.get("features"):
rows.append({"site": name, "lat": site["lat"], "lon": site["lon"], "fetch_status": site_status[name]})
# 2. Thread each alert: VTEC event (office.phenomenon.significance.ETN), else the references chain
current, closed_by = {}, {}
for aid, a in alerts.items():
p = a["p"]
vs = vtecs(p)
primary = next((v for v in vs if v["action"] not in CLOSING), vs[0] if vs else None)
for v in vs:
if v is not primary and v["action"] in CLOSING:
closed_by[v["key"]] = {"action": v["action"], "by": p["event"]}
if primary:
key, action = primary["key"], primary["action"]
else:
refs = [r["identifier"] for r in p.get("references", [])]
key = next((id_to_thread[i] for i in [aid] + refs if i in id_to_thread), aid)
action = "CAN" if p["messageType"] == "Cancel" else ("NEW" if not refs else "CON")
if p["messageType"] == "Cancel":
action = "CAN"
t = current.setdefault(key, {"alerts": [], "sites": set(), "actions": set()})
t["alerts"].append(p)
t["sites"] |= a["sites"]
t["actions"].add(action)
for s in a["sites"]:
rows.append({"site": s, "fetch_status": "OK", "thread": key, "alert_id": aid, "event": p["event"],
"severity": p["severity"], "urgency": p["urgency"], "certainty": p["certainty"],
"message_type": p["messageType"], "vtec_action": action, "onset": p.get("onset"),
"ends": p.get("ends") or p.get("expires"), "sender": p.get("senderName"), "headline": p.get("headline")})
def summarise(key, t):
latest = max(t["alerts"], key=lambda p: p["sent"])
sev = max((p["severity"] for p in t["alerts"]), key=lambda s: RANK.get(s, 0))
ends = max((p.get("ends") or p.get("expires") or "") for p in t["alerts"])
return {"event": latest["event"], "severity": sev, "urgency": latest["urgency"], "certainty": latest["certainty"],
"headline": latest.get("headline") or latest["event"], "area": latest.get("areaDesc", ""),
"sender": latest.get("senderName", ""), "onset": min((p.get("onset") or p["sent"]) for p in t["alerts"]),
"ends": ends, "sites": sorted(t["sites"]), "ids": sorted(p["id"] for p in t["alerts"]),
"instruction": (latest.get("instruction") or "").strip().replace("\n", " ")[:400],
"link": latest["@id"] if "@id" in latest else "https://api.weather.gov/alerts/" + latest["id"],
"closed": bool(t["actions"]) and t["actions"] <= {"CAN", "EXP"}}
# 3. Compare with the stored threads
notes, new_threads = [], {}
def note(change, key, cur, prev=None, changes=None):
prev_sev = (prev or {}).get("severity")
sev = cur["severity"]
notes.append({"change": change, "thread": key, "event": cur["event"], "severity": sev,
"previous_severity": prev_sev, "routed": sev in route_sev or prev_sev in route_sev,
"sites": ", ".join(cur["sites"]), "area": cur["area"], "headline": cur["headline"],
"sender": cur["sender"], "onset": cur["onset"], "ends": cur["ends"], "urgency": cur.get("urgency"),
"certainty": cur.get("certainty"), "instruction": cur.get("instruction", ""),
"changes": "; ".join(changes or []), "link": cur.get("link", "")})
for key, t in current.items():
cur, prev = summarise(key, t), threads.get(key)
cur["ids"] = sorted(set(cur["ids"]) | set((prev or {}).get("ids", [])))[-50:]
if prev is None:
if not cur["closed"]:
note("NEW", key, cur)
elif cur["closed"]:
if not prev.get("closed"):
note("CANCELLED" if "CAN" in t["actions"] else "EXPIRED", key, cur, prev)
else:
changes = []
if RANK.get(cur["severity"], 0) != RANK.get(prev["severity"], 0):
changes.append(f"severity {prev['severity']} -> {cur['severity']}")
if cur["event"] != prev["event"]:
changes.append(f"event {prev['event']} -> {cur['event']}")
if (cur["ends"] or "")[:16] != (prev.get("ends") or "")[:16]:
changes.append(f"ends {prev.get('ends') or 'n/a'} -> {cur['ends'] or 'n/a'}")
added = set(cur["sites"]) - set(prev["sites"])
removed = {s for s in set(prev["sites"]) - set(cur["sites"]) if site_status.get(s) == "OK"}
unknown = {s for s in set(prev["sites"]) - set(cur["sites"]) if site_status.get(s) in ("FETCH_FAILED", "OUT_OF_BOUNDS")}
if added:
changes.append("now also affects " + ", ".join(sorted(added)))
if removed:
changes.append("no longer affects " + ", ".join(sorted(removed)))
cur["sites"] = sorted(set(cur["sites"]) | unknown) # a failed fetch is not an all-clear
if changes:
note("UPDATED", key, cur, prev, changes)
cur["first_seen"] = (prev or {}).get("first_seen", now)
cur["last_seen"] = now
new_threads[key] = cur
# 4. Threads that dropped off the feed: ended, upgraded, or unknown because a site failed
for key, prev in threads.items():
if key in current:
continue
watched = [s for s in prev["sites"] if s in site_status]
if not watched:
continue # site removed from the list
if any(site_status[s] != "OK" for s in watched):
new_threads[key] = prev # cannot tell this run, keep it
continue
if prev.get("closed"):
continue # cancellation already routed, forget it
if key in closed_by:
note("UPGRADED" if closed_by[key]["action"] == "UPG" else "CANCELLED", key, prev, prev,
[f"replaced by {closed_by[key]['by']}"] if closed_by[key]["action"] == "UPG" else None)
else:
note("ENDED", key, prev, prev, ["no longer in the active feed"])
# 5. Message text, CSV, outputs
ICON = {"NEW": ":rotating_light:", "UPDATED": ":arrows_counterclockwise:", "UPGRADED": ":arrow_up:",
"CANCELLED": ":white_check_mark:", "EXPIRED": ":white_check_mark:", "ENDED": ":white_check_mark:"}
for n in notes:
lines = [f"{ICON[n['change']]} *{n['change']}: {n['event']}* ({n['severity']}, {n['urgency']}, {n['certainty']})",
f"*Sites:* {n['sites']}", n["headline"]]
if n["changes"]:
lines.append(f"*Changes:* {n['changes']}")
if n["change"] in ("NEW", "UPDATED"):
lines.append(f"*Area:* {n['area']}\n*From* {n['onset']} *until* {n['ends'] or 'further notice'}")
if n["instruction"]:
lines.append(f"> {n['instruction']}")
lines.append(f"{n['sender']} | <{n['link']}|alert record>")
n["text"] = "\n".join(lines)
notes.sort(key=lambda n: (-RANK.get(n["severity"], 0), n["change"], n["thread"]))
fields = ["site", "lat", "lon", "fetch_status", "thread", "alert_id", "event", "severity", "urgency", "certainty",
"message_type", "vtec_action", "onset", "ends", "sender", "headline"]
coords = {s["name"]: s for s in sites}
with open("site_alerts.csv", "w", newline="", encoding="utf-8") as fh:
w = csv.DictWriter(fh, fieldnames=fields)
w.writeheader()
for r in sorted(rows, key=lambda r: (r["site"], -RANK.get(r.get("severity"), -1), r.get("event") or "")):
r.setdefault("lat", coords[r["site"]]["lat"]); r.setdefault("lon", coords[r["site"]]["lon"])
w.writerow(r)
summary = {
"is_baseline": is_baseline, "sites": len(sites),
"sites_ok": sum(v == "OK" for v in site_status.values()),
"sites_out_of_bounds": sorted(k for k, v in site_status.items() if v == "OUT_OF_BOUNDS"),
"sites_failed": sorted(k for k, v in site_status.items() if v == "FETCH_FAILED"),
"sites_with_alerts": len({s for t in current.values() for s in t["sites"]}),
"active_alerts": len(alerts), "active_threads": len(current),
"routed_severe_threads": sum(1 for k, t in new_threads.items() if t["severity"] in route_sev and not t.get("closed")),
"notifications": len(notes), "routed": sum(n["routed"] for n in notes), "logged": sum(not n["routed"] for n in notes),
"by_change": {c: sum(n["change"] == c for n in notes) for c in sorted({n["change"] for n in notes})},
}
logs = [{"level": "WARN" if n["routed"] else "INFO",
"message": f"{'ROUTE' if n['routed'] else 'LOG'} {n['change']} {n['severity']} {n['event']} [{n['thread']}] for {n['sites']}: "
f"{n['headline']}{' (' + n['changes'] + ')' if n['changes'] else ''}"} for n in notes]
new_state = {"updated": now, "threads": new_threads}
print("::" + json.dumps({"logs": logs, "outputs": {"summary": summary, "notifications": notes, "state": json.dumps(new_state)}}) + "::")
- id: dispatch
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.route.vars.summary.notifications > 0 }}"
then:
- id: each_notification
type: io.kestra.plugin.core.flow.Loop
values: "{{ outputs.route.vars.notifications }}"
description: One iteration per notification, most severe first. Iteration
outputs record whether Slack accepted each routed message.
tasks:
- id: by_severity
type: io.kestra.plugin.core.flow.If
description: Alerts whose current or previous severity is in `route_severities`
go to Slack; the rest are only logged.
condition: "{{ fromJson(item.value).routed }}"
then:
- id: try_slack
type: io.kestra.plugin.core.flow.AllowFailure
description: |
Keeps the router running when `SLACK_WEBHOOK_URL` is not set or Slack is down; the alert is still in the
route log, the flow outputs and the CSV. `sent` only runs when Slack accepted the message, which is how
`delivery_report` tells sent from failed.
tasks:
- id: slack
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
retry:
type: constant
interval: PT10S
maxAttempts: 2
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: "{{ {'text': fromJson(item.value).text} | toJson }}"
- id: sent
type: io.kestra.plugin.core.debug.Return
format: SLACK_SENT
else:
- id: log_other
type: io.kestra.plugin.core.log.Log
message: "{{ fromJson(item.value).change }} {{ fromJson(item.value).severity }}
{{ fromJson(item.value).event }} for {{
fromJson(item.value).sites }}: {{
fromJson(item.value).headline }}"
outputs:
- id: delivery
type: STRING
value: "{{ fromJson(item.value).routed ? (outputs.sent.value ?? 'SLACK_FAILED')
: 'LOGGED' }} {{ fromJson(item.value).change }} {{
fromJson(item.value).thread }}"
- id: delivery_report
type: io.kestra.plugin.core.log.Log
description: Slack failures (no secret, webhook revoked) are reported here; the
alert threads are saved either way so a missing webhook does not
re-send the whole backlog later.
level: "{{ outputs.each_notification.outputs | jq('[.[].outputs.delivery |
select(startswith(\"SLACK_FAILED\"))] | length') | first > 0 ? 'WARN'
: 'INFO' }}"
message: "Dispatch: {{ outputs.each_notification.outputs |
jq('[.[].outputs.delivery | split(\" \")[0]] | group_by(.) |
map({(.[0]): length}) | add') | first | toJson }}"
else:
- id: nothing_new
type: io.kestra.plugin.core.log.Log
message: "No new, changed or cleared alerts: {{
outputs.route.vars.summary.active_alerts }} active alerts in {{
outputs.route.vars.summary.active_threads }} threads across {{
outputs.route.vars.summary.sites_with_alerts }} of {{
outputs.route.vars.summary.sites }} sites, all already routed."
- id: save_state
type: io.kestra.plugin.core.kv.Set
description: Saved after dispatch, so a run that dies mid-way repeats its
notifications next time instead of losing them.
key: "{{ inputs.state_key }}"
value: "{{ outputs.route.vars.state }}"
- id: warn_sites
type: io.kestra.plugin.core.log.Log
runIf: "{{ (outputs.route.vars.summary.sites_out_of_bounds | length) +
(outputs.route.vars.summary.sites_failed | length) > 0 }}"
level: WARN
message: "Sites without an answer this run.{{
outputs.route.vars.summary.sites_out_of_bounds is not empty ? ' Outside
the NWS area (fix lat/lon or remove): ' ~
(outputs.route.vars.summary.sites_out_of_bounds | join(', ')) ~ '.' : ''
}}{{ outputs.route.vars.summary.sites_failed is not empty ? ' Fetch failed
after retries (their alerts are kept as-is until the next successful
fetch): ' ~ (outputs.route.vars.summary.sites_failed | join(', ')) ~ '.' :
'' }}"
- id: summary
type: io.kestra.plugin.core.log.Log
message: |
{{ outputs.route.vars.summary.sites_ok }} of {{ outputs.route.vars.summary.sites }} sites checked{{ outputs.route.vars.summary.is_baseline ? ' (BASELINE run, every active alert is new)' : '' }}: {{ outputs.route.vars.summary.active_alerts }} active alerts in {{ outputs.route.vars.summary.active_threads }} threads affecting {{ outputs.route.vars.summary.sites_with_alerts }} sites, {{ outputs.route.vars.summary.routed_severe_threads }} of them {{ inputs.route_severities | join('/') }}.
Notifications this run: {{ outputs.route.vars.summary.notifications }} ({{ outputs.route.vars.summary.routed }} routed to Slack, {{ outputs.route.vars.summary.logged }} logged) {{ outputs.route.vars.summary.by_change | toJson }}.
Site report: {{ outputs.route.outputFiles['site_alerts.csv'] }}
outputs:
- id: summary
type: JSON
value: "{{ outputs.route.vars.summary }}"
- id: notifications
type: JSON
value: "{{ outputs.route.vars.notifications }}"
- id: site_alerts_csv
type: FILE
value: "{{ outputs.route.outputFiles['site_alerts.csv'] }}"
