mangrovedigital/tide-engine-api

1"""NOAA CO-OPS source for the Florida Bay honest-hybrid anchors.
2
3Florida Bay's interior has NO observed NOAA water-level stations (probe
42026-07-01, out/noaa_fb_probe.json) — only SUBORDINATE tide-prediction stations,
5which serve high/low predictions only (no hourly product). So the displayed
6curve for these anchors is a cosine interpolation between NOAA's predicted
7extremes — the standard tide-clock shape — and is labeled as such.
8
9Additive module: uses the same disk cache + provenance layer as sources.py.
10Levels returned in METERS (engine is metric internally). Datum stays the
11station's own (MLLW) end-to-end — never mixed with NAVD88 (no depth calls).
12"""
13import math
14import urllib.parse
15from datetime import datetime, timezone, timedelta
16
17from . import sources
18
19DATAGETTER = "https://api.tidesandcurrents.noaa.gov/api/prod/datagetter"
20FT = 0.3048
21HOUR = 3600
22
23
24def _epoch(s):
25    """'2026-06-25 12:00' (GMT) -> utc epoch seconds."""
26    return datetime.fromisoformat(s).replace(tzinfo=timezone.utc).timestamp()
27
28
29def _yyyymmdd(epoch):
30    return datetime.fromtimestamp(epoch, tz=timezone.utc).strftime("%Y%m%d")
31
32
33def hilo_predictions(station, begin_yyyymmdd, end_yyyymmdd, datum="MLLW", ttl=None):
34    """NOAA predicted highs/lows for a (possibly subordinate) station.
35
36    Returns (extremes, provenance): extremes = [(epoch, level_m, 'H'|'L'), ...].
37    Day-aligned begin/end keep the URL — and therefore the cache key —
38    deterministic for a given date window.
39    """
40    q = {
41        "station": station, "product": "predictions", "datum": datum,
42        "interval": "hilo", "begin_date": begin_yyyymmdd, "end_date": end_yyyymmdd,
43        "units": "english", "time_zone": "gmt", "format": "json",
44    }
45    url = DATAGETTER + "?" + urllib.parse.urlencode(q)
46    data, path = sources._fetch_json(url, "noaa|" + url, ttl_seconds=ttl)
47    rows = data.get("predictions")
48    if not rows:
49        raise RuntimeError("NOAA returned no predictions for %s: %s"
50                           % (station, data.get("error", {}).get("message", "empty")))
51    extremes = [(_epoch(r["t"]), float(r["v"]) * FT, r["type"]) for r in rows]
52    prov = {
53        "source": "NOAA CO-OPS tide predictions (subordinate hi/lo)",
54        "station": station, "datum": datum, "orig_units": "ft",
55        "window": [begin_yyyymmdd, end_yyyymmdd],
56        "url": url, "cache": path,
57        "curve": "cosine interpolation between NOAA predicted extremes",
58    }
59    return extremes, prov
60
61
62def curve_from_hilo(extremes, grid):
63    """Cosine-interpolated level at each grid epoch from consecutive extremes.
64
65    v(t) = v0 + (v1-v0) * (1 - cos(pi * (t-t0)/(t1-t0))) / 2
66    Grid points outside the extremes' span -> None (explicit hole, no guess).
67    """
68    out = []
69    k = 0
70    n = len(extremes)
71    for t in grid:
72        while k + 1 < n and extremes[k + 1][0] <= t:
73            k += 1
74        if t < extremes[0][0] or k + 1 >= n:
75            out.append(None)
76            continue
77        t0, v0, _ = extremes[k]
78        t1, v1, _ = extremes[k + 1]
79        if t1 <= t0:
80            out.append(None)
81            continue
82        f = (1.0 - math.cos(math.pi * (t - t0) / (t1 - t0))) / 2.0
83        out.append(v0 + (v1 - v0) * f)
84    return out
85
86
87def predicted_curve(station, now_epoch, history_h=18, horizon_h=72, datum="MLLW"):
88    """Hourly predicted-tide curve around now for a subordinate station.
89
90    Returns (grid, values_m, extremes, prov). Fetch window is day-aligned with
91    a 1-day margin each side so the interpolation brackets the whole grid.
92    """
93    begin = _yyyymmdd(now_epoch - history_h * HOUR - 24 * HOUR)
94    end = _yyyymmdd(now_epoch + horizon_h * HOUR + 24 * HOUR)
95    extremes, prov = hilo_predictions(station, begin, end, datum=datum)
96    t0 = int((now_epoch - history_h * HOUR) // HOUR) * HOUR
97    t1 = int((now_epoch + horizon_h * HOUR) // HOUR) * HOUR
98    grid = list(range(t0, t1 + HOUR, HOUR))
99    vals = curve_from_hilo(extremes, grid)
100    return grid, vals, extremes, prov
101
102
103def observed_water_level(station, begin_yyyymmdd, end_yyyymmdd, datum="NAVD", ttl=None):
104    """Observed 6-min water level (e.g. Vaca Key 8723970). Metres, station datum."""
105    q = {
106        "station": station, "product": "water_level", "datum": datum,
107        "begin_date": begin_yyyymmdd, "end_date": end_yyyymmdd,
108        "units": "english", "time_zone": "gmt", "format": "json",
109    }
110    url = DATAGETTER + "?" + urllib.parse.urlencode(q)
111    data, path = sources._fetch_json(url, "noaa|" + url, ttl_seconds=ttl)
112    rows = data.get("data")
113    if not rows:
114        raise RuntimeError("NOAA returned no water_level for %s: %s"
115                           % (station, data.get("error", {}).get("message", "empty")))
116    t, v = [], []
117    for r in rows:
118        if r.get("v") in (None, ""):
119            continue
120        t.append(_epoch(r["t"]))
121        v.append(float(r["v"]) * FT)
122    prov = {"source": "NOAA CO-OPS observed water level", "station": station,
123            "datum": datum, "orig_units": "ft", "url": url, "cache": path}
124    return sources.Series("noaa_obs:%s" % station, t, v, "m (%s)" % datum, prov)
125
126
127if __name__ == "__main__":
128    # math selftest (synthetic, known answer): a pure cosine tide sampled at its
129    # own extremes must be reconstructed exactly at every hour.
130    import sys
131    T = 12.42 * HOUR
132    ext = []
133    base = 1_700_000_000 - (1_700_000_000 % HOUR)
134    for i in range(8):
135        t = base + i * T / 2
136        ext.append((t, 1.0 if i % 2 == 0 else -1.0, "H" if i % 2 == 0 else "L"))
137    grid = [base + k * HOUR for k in range(int(3 * T / HOUR))]
138    got = curve_from_hilo(ext, grid)
139    worst = 0.0
140    for t, g in zip(grid, got):
141        truth = math.cos(2 * math.pi * (t - base) / T)
142        worst = max(worst, abs(g - truth))
143    print("curve_from_hilo selftest: worst |err| = %.6f m (gate < 0.02)" % worst)
144    sys.exit(0 if worst < 0.02 else 1)
145