River Flood Early Warning with Open-Meteo and a Local Ollama Model

id: river-flood-early-warning-ollama
namespace: company.team
description: |
  River flood early warning with a local LLM. Every six hours the flow pulls
  the 7-day river discharge forecast for each monitored river point from the
  Open-Meteo Flood API, flags the points whose forecast peak crosses their
  alert level or rises sharply, has an Ollama model write a short
  plain-language risk bulletin, posts it to Slack when a point is at risk, and
  sends a separate Slack alert if the flow fails.

inputs:
  - id: stations
    type: JSON
    displayName: Monitored river points
    description: >-
      List of river points to watch. Each item needs a name, latitude, longitude
      and alert_m3s (the discharge in m³/s that your team treats as an alert
      level). The defaults are two major rivers in Tabasco, Mexico; the alert
      levels are example values, so replace them with the levels your local
      water authority uses.
    defaults: |
      [
        {"name": "Usumacinta at Boca del Cerro (Tabasco, MX)", "latitude": 17.52, "longitude": -91.48, "alert_m3s": 5000},
        {"name": "Grijalva near Villahermosa (Tabasco, MX)", "latitude": 18.09, "longitude": -93.03, "alert_m3s": 2500}
      ]

  - id: rise_alert_pct
    type: INT
    displayName: Sudden rise alert (%)
    description: Non-negative percentage. Also flag a point when its 7-day forecast
      peak is at least this many percent above today's modeled discharge, even
      if it stays under alert_m3s.
    defaults: 50

  - id: bulletin_language
    type: SELECT
    displayName: Bulletin language
    description: Language the LLM writes the risk bulletin in.
    values:
      - English
      - Spanish
      - Portuguese
      - French
    defaults: English

  - id: model
    type: STRING
    displayName: Ollama model
    description: Local Ollama model that writes the bulletin. It must already be
      pulled with `ollama pull <model>`.
    defaults: gemma3:4b

  - id: ollama_endpoint
    type: STRING
    displayName: Ollama endpoint
    description: Base URL where Ollama listens. Use
      http://host.docker.internal:11434 when Kestra runs in Docker and Ollama
      runs on the host.
    defaults: "http://localhost:11434"

tasks:
  - id: check_rivers
    type: io.kestra.plugin.core.flow.Loop
    description: Fetch and evaluate the forecast for each river point, one point at
      a time to stay well inside the API's rate limits.
    values: "{{ inputs.stations }}"
    concurrencyLimit: 1
    tasks:
      - id: fetch_forecast
        type: io.kestra.plugin.core.http.Request
        description: Download today's modeled discharge and the 7-day maximum ensemble
          discharge forecast (GloFAS) for the current point.
        uri: "https://flood-api.open-meteo.com/v1/flood?latitude={{
          fromJson(item.value).latitude }}&longitude={{
          fromJson(item.value).longitude
          }}&daily=river_discharge,river_discharge_max&forecast_days=7"
        method: GET
    outputs:
      - id: name
        type: STRING
        description: Human-readable name of the river point.
        value: "{{ fromJson(item.value).name }}"
      - id: today_m3s
        type: STRING
        description: Modeled discharge for today, in m³/s.
        value: "{{ outputs.fetch_forecast.body | jq('.daily.river_discharge[0]') | first
          }}"
      - id: peak_m3s
        type: STRING
        description: Highest maximum ensemble discharge forecast over the next 7 days,
          in m³/s.
        value: "{{ outputs.fetch_forecast.body | jq('.daily.river_discharge_max | max')
          | first }}"
      - id: peak_date
        type: STRING
        description: Date of the forecast peak.
        value: "{{ outputs.fetch_forecast.body | jq('.daily as $d |
          ($d.river_discharge_max | index($d.river_discharge_max | max)) as $i |
          $d.time[$i]') | first }}"
      - id: alert_m3s
        type: STRING
        description: Alert level configured for this point, in m³/s.
        value: "{{ fromJson(item.value).alert_m3s }}"
      - id: at_risk
        type: BOOL
        description: True when the forecast peak reaches the alert level or is at least
          rise_alert_pct percent above today's modeled discharge.
        value: "{{ outputs.fetch_forecast.body | jq('.daily as $d |
          ($d.river_discharge_max | max) as $peak | ($peak >= ' ~
          fromJson(item.value).alert_m3s ~ ') or ($d.river_discharge[0] > 0 and
          $peak >= $d.river_discharge[0] * (1 + ' ~ inputs.rise_alert_pct ~ ' /
          100))') | first }}"

  - id: write_bulletin
    type: io.kestra.plugin.ai.completion.ChatCompletion
    description: Turn the raw numbers into a short, calm, plain-language bulletin
      with a local Ollama model, so no data leaves your infrastructure.
    provider:
      type: io.kestra.plugin.ai.provider.Ollama
      modelName: "{{ inputs.model }}"
      endpoint: "{{ inputs.ollama_endpoint }}"
    messages:
      - type: SYSTEM
        content: |
          You write short river flood bulletins for an operations team.
          Use only the data provided; never invent water levels, places or forecasts.
          Write in {{ inputs.bulletin_language }}, in plain text without markdown headings, under 150 words.
          Start with one line saying whether any point needs attention. Then give one line per point with
          today's discharge, the 7-day peak and its date. For at-risk points, add one practical, non-alarmist
          next step (for example: monitor every few hours, check local civil protection notices).
          Remind readers that this is a model forecast and official warnings take priority.
      - type: USER
        content: |
          Forecast summary (discharge in m³/s, source: Open-Meteo Flood API, GloFAS model):
          {% for p in (outputs.check_rivers.outputs | jq('sort_by(.item.iteration) | map(.outputs)') | first) %}
          - {{ p.name }}: today {{ p.today_m3s }}, 7-day peak {{ p.peak_m3s }} on {{ p.peak_date }}, alert level {{ p.alert_m3s }}, at risk: {{ p.at_risk }}
          {% endfor %}

  - id: save_bulletin
    type: io.kestra.plugin.core.storage.Write
    description: Keep every bulletin, with the numbers behind it, as a markdown file
      in Kestra internal storage for audit and later comparison.
    extension: .md
    content: |
      # River flood bulletin, {{ execution.startDate | date("yyyy-MM-dd HH:mm") }}

      {{ outputs.write_bulletin.textOutput }}

      ## Data
      | Point | Today (m³/s) | 7-day peak (m³/s) | Peak date | Alert level | At risk |
      |---|---|---|---|---|---|
      {% for p in (outputs.check_rivers.outputs | jq('sort_by(.item.iteration) | map(.outputs)') | first) %}| {{ p.name }} | {{ p.today_m3s }} | {{ p.peak_m3s }} | {{ p.peak_date }} | {{ p.alert_m3s }} | {{ p.at_risk }} |
      {% endfor %}

      Source: [Open-Meteo Flood API](https://open-meteo.com/en/docs/flood-api) (GloFAS). Execution {{ execution.id }}.

  - id: alert_if_needed
    type: io.kestra.plugin.core.flow.If
    description: Only notify people when at least one point is at risk, so the
      channel stays quiet on normal days.
    condition: "{{ outputs.check_rivers.outputs | jq('map(select((.outputs.at_risk |
      tostring) == \"true\")) | length > 0') | first }}"
    then:
      - id: notify_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the generated bulletin to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :ocean: River flood early warning ({{ flow.id }})
          {{ outputs.write_bulletin.textOutput }}
          Full bulletin and data: execution {{ execution.id }}
    else:
      - id: all_clear
        type: io.kestra.plugin.core.log.Log
        description: Record that no point met the alert-level or sudden-rise criteria.
        message: "None of the {{ outputs.check_rivers.iterationCount }} river points met
          the alert-level or sudden-rise criteria for the next 7 days."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Tell the team when the warning itself fails (API down, Ollama
      unreachable), because a silent early-warning flow is worse than none.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":warning: River flood early warning FAILED in {{ flow.id }}
      (execution {{ execution.id }}). Check whether the Open-Meteo request or
      the Ollama call broke."

triggers:
  - id: every_six_hours
    type: io.kestra.plugin.core.trigger.Schedule
    description: Re-check the forecast four times a day; GloFAS updates daily, so
      more often adds load without new information.
    cron: "0 */6 * * *"
    disabled: true