Connect with the API

"""FlowAi API example (Python 3.9+, requests): a gauge's latest reading, its observations and forecast as CSV,
and several gauges in one call through the PI-JSON feed.

    pip install requests
    export FLOWAI_API_KEY="fai_..."        # from Profile > Integrations; never put the key in the script
    python flowai_example.py
"""
import csv
import os
import time
from datetime import datetime, timedelta, timezone

import requests

API = os.environ.get("FLOWAI_API_URL", "https://api.flowai-forecast.com")
GAUGE = "02KF019"                       # a WSC station id
GAUGES = ["02KF019", "02HB029"]

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['FLOWAI_API_KEY']}"


def get(path, params=None):
    """GET a FlowAi URL. On the key's rate limit (HTTP 429), wait as long as Retry-After says, then try once more."""
    r = session.get(f"{API}{path}", params=params, timeout=30)
    if r.status_code == 429:
        time.sleep(int(r.headers.get("Retry-After", "60")))
        r = session.get(f"{API}{path}", params=params, timeout=30)
    r.raise_for_status()
    return r


# 1. The gauge and its latest reading
gauge = get(f"/api/v1/gauges/{GAUGE}").json()
latest = get(f"/api/v1/gauges/{GAUGE}/latest").json()
print(f"{gauge['name']} ({GAUGE}): {latest['flow_cms']} m3/s at {latest['flow_timestamp']}")

# 2. Daily observed flow and level, last 7 days, newest first, to CSV. Each row is the provider's daily value,
#    stamped with its calendar date (00:00). For real-time readings use /recent (the last 48 hours) or the feed (step 4).
observed = get(f"/api/v1/gauges/{GAUGE}/observed", {"days": 7}).json()["readings"]
with open(f"{GAUGE}_observed.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["date", "flow_cms", "level_m", "flag"])
    for row in observed:
        w.writerow([row["timestamp"][:10], row["flow_cms"], row["level_m"], row["flag"]])
print(f"{len(observed)} days observed -> {GAUGE}_observed.csv")

# 3. The latest forecast, one row per day, to CSV
forecast = get(f"/api/v1/gauges/{GAUGE}/forecast").json()
days = [lead for lead in forecast["forecasts"] if lead["segment"] == "forecast"]
with open(f"{GAUGE}_forecast.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["issue_time_utc", "valid_date", "lead_days", "flow_cms"])
    for lead in days:
        w.writerow([forecast["issue_time"], lead["valid_date"], lead["lead_days"], lead["flow_cms"]])
print(f"forecast issued {forecast['issue_time']}: {len(days)} days -> {GAUGE}_forecast.csv")

# 4. Several gauges and series in one call: the PI-JSON feed
now = datetime.now(timezone.utc)
params = [("locationIds", g) for g in GAUGES] + [("parameterIds", p) for p in ("Q.obs", "Q.fcst.day")]
params += [("startTime", (now - timedelta(days=2)).strftime("%Y-%m-%dT%H:%M:%SZ")),
           ("endTime", (now + timedelta(days=11)).strftime("%Y-%m-%dT%H:%M:%SZ")),
           ("documentFormat", "PI_JSON")]
feed = get("/rest/fewspiservice/v1/timeseries", params)
for series in feed.json()["timeSeries"]:
    h, events = series["header"], series["events"]
    last = f"{events[-1]['date']} {events[-1]['time']} = {events[-1]['value']}" if events else "no values"
    print(f"{h['locationId']} {h['parameterId']}: {len(events)} values, last {last}")
print(feed.headers["X-FlowAi-Disclaimer"])

# FlowAi API example (R 4.1+, httr2): a gauge's latest reading, its observations and forecast as CSV,
# and several gauges in one call through the PI-JSON feed.
#
#   install.packages(c("httr2", "jsonlite"))
#   Sys.setenv(FLOWAI_API_KEY = "fai_...")   # from Profile > Integrations; better in ~/.Renviron than in the script
#   source("flowai_example.R")

library(httr2)

api <- Sys.getenv("FLOWAI_API_URL", "https://api.flowai-forecast.com")
gauge_id <- "02KF019"                       # a WSC station id
gauge_ids <- c("02KF019", "02HB029")

# GET a FlowAi URL with the key; waits and retries when the key's rate limit (HTTP 429) is reached.
flowai_get <- function(path, query = list()) {
  request(api) |>
    req_url_path_append(path) |>
    req_url_query(!!!query, .multi = "explode") |>
    req_auth_bearer_token(Sys.getenv("FLOWAI_API_KEY")) |>
    req_retry(max_tries = 2, is_transient = \(resp) resp_status(resp) == 429) |>
    req_timeout(30) |>
    req_perform()
}

# 1. The gauge and its latest reading
gauge <- flowai_get(paste0("api/v1/gauges/", gauge_id)) |> resp_body_json()
latest <- flowai_get(paste0("api/v1/gauges/", gauge_id, "/latest")) |> resp_body_json()
cat(sprintf("%s (%s): %s m3/s at %s\n", gauge$name, gauge_id, latest$flow_cms, latest$flow_timestamp))

# 2. Daily observed flow and level, last 7 days, newest first, to CSV. Each row is the provider's daily value,
#    stamped with its calendar date (00:00). For real-time readings use /recent (the last 48 hours) or the feed (step 4).
#    A missing value (JSON null) becomes NA.
observed <- flowai_get(paste0("api/v1/gauges/", gauge_id, "/observed"), list(days = 7)) |>
  resp_body_json(simplifyVector = TRUE)
readings <- observed$readings[, c("timestamp", "flow_cms", "level_m", "flag")]
readings$timestamp <- substr(readings$timestamp, 1, 10)
names(readings)[1] <- "date"
write.csv(readings, paste0(gauge_id, "_observed.csv"), row.names = FALSE)
cat(nrow(readings), "days observed ->", paste0(gauge_id, "_observed.csv\n"))

# 3. The latest forecast, one row per day, to CSV
forecast <- flowai_get(paste0("api/v1/gauges/", gauge_id, "/forecast")) |> resp_body_json(simplifyVector = TRUE)
days <- subset(forecast$forecasts, segment == "forecast", select = c(valid_date, lead_days, flow_cms))
days <- cbind(issue_time_utc = forecast$issue_time, days)
write.csv(days, paste0(gauge_id, "_forecast.csv"), row.names = FALSE)
cat(sprintf("forecast issued %s: %d days -> %s_forecast.csv\n", forecast$issue_time, nrow(days), gauge_id))

# 4. Several gauges and series in one call: the PI-JSON feed
now <- Sys.time()
utc <- \(t) format(t, "%Y-%m-%dT%H:%M:%SZ", tz = "UTC")
feed <- flowai_get("rest/fewspiservice/v1/timeseries", list(
  locationIds = gauge_ids, parameterIds = c("Q.obs", "Q.fcst.day"),
  startTime = utc(now - 2 * 86400), endTime = utc(now + 11 * 86400), documentFormat = "PI_JSON"))
for (series in resp_body_json(feed)$timeSeries) {
  n <- length(series$events)
  last <- if (n > 0) with(series$events[[n]], paste(date, time, "=", value)) else "no values"
  cat(sprintf("%s %s: %d values, last %s\n", series$header$locationId, series$header$parameterId, n, last))
}
cat(resp_header(feed, "X-FlowAi-Disclaimer"), "\n")