B4L/Northern-Thailand-Ping-River-Monitor
8
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

chore: remove the Python service; Python is the training stack only

Browse Source 
Phase 4 of docs/RUST_REWRITE.md. Since 2026-10-18 ping-collector (rust/)
collects, forecasts, notifies (ntfy + Matrix) and serves the website/API;
nothing in production ran the Python service code any more.

Removed (~17k lines): the FastAPI app (web_api, schemas, health_check,
metrics, postgres_history, forecast_history, thaiwater), the Python RID
collector and CLI (water_scraper_v3, main, run.py), the notifier and Matrix
alerting (notify, alerting), the multi-backend adapters (SQLite, MySQL,
InfluxDB, VictoriaMetrics) with demo databases, models/validators/exceptions/
logging/rate limiter, ml/skill.py (served by Rust now), the migration tooling
(api_parity, collector_parity, e2e_notify, install_collector.sh,
gen_notify_cases.py, spikes/rust-forecast), Docker/compose/Grafana files,
sql/, the release workflow and the docs for all of that. Their tests went
with them.

Kept in Python: src/ml (training, evaluation harness, export; predict.py as
the reference the Rust forecaster is tested against), the HII and RID-dam
clients/stores for the archive backfills (their live collectors removed),
the backfill scripts, retrain.sh + timer, forecast_parity.py.

- PostgreSQL only: src/config.py is database_url() from POSTGRES_* / .env;
  the stores lost their sqlite/mysql branches (and the 2026-08 column
  migration). The database tests run against a throwaway schema from
  TEST_DATABASE_URL (tests/conftest.py pg_url, skipped without it), as the
  Rust tests do; CI gets a PostgreSQL service. 100 tests (66 without a DB).
- The notifier settings moved from src/notify.py into src/ml/export.py;
  flood_config.json regenerates byte-identical.
- tests/test_dashboard.py checks every path the dashboard fetches against
  ping-collector's router.
- Dependencies: fastapi, uvicorn, pydantic, influxdb, pymysql, psutil,
  schedule, sphinx, pyinstaller, mypy, pytest-asyncio dropped; python-dotenv
  1.1.1 explicit (it came in through uvicorn[standard]).
- Formatting contract extended to scripts/ and tests/ (reformatted once).
- scripts/install.sh is the one installer (fresh host or update, in place in
  a checkout without touching its ownership): binary, units, uv sync, .env
  defaults, removes the retired water-monitor.service and any old
  --alert-check cron job, dry run + forecast check, restart, readiness wait.
- Docs, CLAUDE.md, CONTRIBUTING.md, .env.example, Makefile, unit comments
  and Rust doc comments updated to the Rust service.
This commit is contained in:
grabowski committed 2026-10-18 17:15:07 +07:00
1 parent 330fdf7804
commit 7e642555b4
125 files changed
+1300 -19558
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/api_parity.py
-389
View File
@@ -1,389 +0,0 @@
#!/usr/bin/env python3
"""Compare the Rust website/API with the FastAPI app, route by route.
Phase 3 of docs/RUST_REWRITE.md: before the Rust service takes over port
8000, both answer the same requests against the same database and every
response is compared: status, content type, Cache-Control and the JSON body
(floats exactly; lists of database rows without ORDER BY as sets). Static
files and the dashboard byte for byte.
# the FastAPI app on :8000, `ping-collector serve --listen 127.0.0.1:8001`
uv run python scripts/api_parity.py \\
--python http://127.0.0.1:8000 --rust http://127.0.0.1:8001
Query parameters vary per run (limits, hours) so the FastAPI app's response
caches (up to 15 min) answer with fresh data, not an older copy. Data the
collector writes during the run can still make a route differ once: a
mismatch is retried after a pause before it counts.
Intended differences (the Rust side may answer differently):
/health a lagging RID report with current ThaiWater telemetry for
the same gauges is healthy (as the notifier, phase 2)
/scraping/status extra fields (collector version, forecast run)
/metrics process counters: compared by shape only
/sensors/thaiwater live upstream: compared, a difference only warns
/api/hii/rainfall/catchment correlation within 1e-9 (numpy uses BLAS)
/api/stats newer counts on the Rust side are accepted (the FastAPI app
caches this one for 5 minutes)
*.geojson content type application/geo+json (FastAPI: octet-stream)
Exit status 0 when every hard check passed.
"""
import argparse
import random
import sys
import time
from typing import Any, Callable, List, Optional, Tuple
import requests
SOFT = "soft" # a difference warns, does not fail
def get(base: str, path: str, method: str = "GET", **kw) -> requests.Response:
last = None
for _ in range(3):
try:
return requests.request(method, base + path, timeout=120, **kw)
except requests.RequestException as e: # connection reset etc.
last = e
time.sleep(1)
raise last
def first_diff(a: Any, b: Any, path: str = "$", rel: float = 0.0) -> Optional[str]:
"""Where two decoded JSON values differ (None if equal)."""
if isinstance(a, bool) or isinstance(b, bool):
return None if a is b else f"{path}: {a!r} != {b!r}"
if isinstance(a, (int, float)) and isinstance(b, (int, float)):
if a == b:
return None
if rel and abs(a - b) <= rel * max(abs(a), abs(b)):
return None
return f"{path}: {a!r} != {b!r}"
if type(a) is not type(b):
return f"{path}: {type(a).__name__} {a!r} != {type(b).__name__} {b!r}"
if isinstance(a, dict):
if list(a.keys()) != list(b.keys()):
if set(a.keys()) != set(b.keys()):
return f"{path}: keys {sorted(set(a) ^ set(b))} only on one side"
for k in a:
d = first_diff(a[k], b[k], f"{path}.{k}", rel)
if d:
return d
return None
if isinstance(a, list):
if len(a) != len(b):
return f"{path}: {len(a)} items != {len(b)} items"
for i, (x, y) in enumerate(zip(a, b)):
d = first_diff(x, y, f"{path}[{i}]", rel)
if d:
return d
return None
return None if a == b else f"{path}: {a!r} != {b!r}"
def media_type(r: requests.Response) -> str:
return r.headers.get("content-type", "").split(";")[0].strip()
class Case:
def __init__(
self,
path: str,
method: str = "GET",
compare: Optional[Callable] = None,
sort_key: Optional[str] = None,
soft: bool = False,
body: Any = None,
headers: Optional[dict] = None,
):
self.path, self.method, self.compare = path, method, compare
self.sort_key, self.soft, self.body = sort_key, soft, body
self.headers = headers or {}
def run(self, py: str, rs: str) -> Tuple[Optional[str], Optional[str]]:
"""(hard difference, warning)"""
kw = {"headers": self.headers}
if self.body is not None:
kw["json"] = self.body
a = get(py, self.path, self.method, **kw)
b = get(rs, self.path, self.method, **kw)
if a.status_code != b.status_code:
return self._fail(
f"status {a.status_code} != {b.status_code} "
f"({a.text[:120]!r} / {b.text[:120]!r})"
)
if media_type(a) != media_type(b):
warn = f"content-type {media_type(a)!r} != {media_type(b)!r}"
if media_type(a).startswith("application/json") or media_type(b).startswith(
"application/json"
):
return self._fail(warn)
return None, warn
if self.method == "GET" and a.status_code == 200:
ca, cb = a.headers.get("cache-control"), b.headers.get("cache-control")
if ca != cb:
return self._fail(f"Cache-Control {ca!r} != {cb!r}")
if self.compare is not None:
d = self.compare(a, b)
elif media_type(a) == "application/json":
ja, jb = a.json(), b.json()
if a.status_code == 422:
# pydantic's messages are approximated; type and location must agree
ja = [(e["type"], e["loc"]) for e in ja["detail"]]
jb = [(e["type"], e["loc"]) for e in jb["detail"]]
if any(t.startswith("date") for t, _ in ja):
ja, jb = [loc for _, loc in ja], [loc for _, loc in jb]
elif self.sort_key and isinstance(ja, list):
ja = sorted(ja, key=lambda r: str(r.get(self.sort_key)))
jb = sorted(jb, key=lambda r: str(r.get(self.sort_key)))
d = first_diff(ja, jb)
else:
d = (
None
if a.content == b.content
else f"bodies differ ({len(a.content)} / {len(b.content)} bytes)"
)
return self._fail(d) if d else (None, None)
def _fail(self, d: str) -> Tuple[Optional[str], Optional[str]]:
return (None, d) if self.soft else (d, None)
def compare_health(a, b) -> Optional[str]:
ja, jb = a.json(), b.json()
if set(ja["checks"]) != set(jb["checks"]):
return f"checks {sorted(ja['checks'])} != {sorted(jb['checks'])}"
for name, ca in ja["checks"].items():
cb = jb["checks"][name]
if ca["status"] == cb["status"]:
continue
# the intended change: RID lagging, ThaiWater current
if (
name == "database"
and ca["status"] == "degraded"
and cb["status"] == "healthy"
and "ThaiWater telemetry current" in cb["message"]
):
continue
if name in ("api", "memory"):
continue # probes of the RID API / own process, compared loosely
return f"{name}: {ca} != {cb}"
return None
def compare_metrics(a, b) -> Optional[str]:
ja, jb = a.json(), b.json()
return None if set(ja) == set(jb) else f"keys {sorted(ja)} != {sorted(jb)}"
def compare_status(a, b) -> Optional[str]:
ja, jb = a.json(), b.json()
for k, v in ja.items():
if jb.get(k) != v:
return f"{k}: {v!r} != {jb.get(k)!r}"
return None
def compare_skill(a, b) -> Optional[str]:
ja, jb = a.json(), b.json()
va, vb = ja.pop("verified_until"), jb.pop("verified_until")
gap = abs(
time.mktime(time.strptime(va[:19], "%Y-%m-%dT%H:%M:%S"))
- time.mktime(time.strptime(vb[:19], "%Y-%m-%dT%H:%M:%S"))
)
if gap > 300:
return f"verified_until {va} != {vb}"
return first_diff(ja, jb)
def compare_stats(a, b) -> Optional[str]:
"""Exact, except that the FastAPI app may answer from its 5-minute cache
while the collector has written newer rows since."""
ja, jb = a.json(), b.json()
if ja == jb:
return None
for k, v in ja.items():
w = jb.get(k)
if k.endswith("_measurements") or k == "days_spanned":
if not isinstance(w, int) or w < v:
return f"{k}: {v!r} > {w!r}"
elif k == "last_timestamp":
if not isinstance(w, str) or w < v:
return f"{k}: {v!r} > {w!r}"
elif k == "coverage_percent":
if not isinstance(w, (int, float)) or abs(w - v) > 0.2:
return f"{k}: {v!r} != {w!r}"
elif v != w:
return f"{k}: {v!r} != {w!r}"
return None
def compare_catchment(a, b) -> Optional[str]:
return first_diff(a.json(), b.json(), rel=1e-9)
def compare_openapi(a, b) -> Optional[str]:
ja, jb = a.json(), b.json()
if ja["info"]["title"] != jb["info"]["title"]:
return "title differs"
ops = lambda spec: { # noqa: E731
(m, p): op for p, item in spec["paths"].items() for m, op in item.items()
}
oa, ob = ops(ja), ops(jb)
if set(oa) != set(ob):
return f"operations only on one side: {sorted(set(oa) ^ set(ob))}"
for k, op in oa.items():
other = ob[k]
for field in ("operationId", "summary", "parameters", "requestBody"):
d = first_diff(op.get(field), other.get(field), f"{k}.{field}")
if d:
return d
if set(op["responses"]) != set(other["responses"]):
return f"{k}: responses {sorted(op['responses'])} != {sorted(other['responses'])}"
for name in ja["components"]["schemas"]:
if name not in jb["components"]["schemas"]:
return f"schema {name} missing"
return None
def compare_html_docs(a, b) -> Optional[str]:
return (
None
if "swagger-ui" in b.text and "/openapi.json" in b.text
else "no Swagger UI"
)
def cases(admin_key: Optional[str]) -> List[Case]:
r = random.Random(time.time())
hours = r.randint(150, 190) # fresh keys for the Python app's caches
limit = r.randint(40000, 49999)
feed_hours = r.randint(20, 30)
skill_station = r.choice(
["P.67", "P.103", "P.20", "P.75", "P.92", "P.21", "P.5", "P.82"]
)
day = time.strftime("%Y-%m-%d", time.localtime(time.time() - 86400 * 3))
day2 = time.strftime("%Y-%m-%d", time.localtime(time.time() - 86400))
out = [
Case("/"),
Case("/robots.txt"),
Case("/llms.txt"),
Case("/sitemap.xml"),
Case("/static/ping-river-network.geojson", soft=True),
Case("/static/custom-markers.json"),
Case("/static/flood-zones-p1.geojson", soft=True),
Case("/static/flood-2024-all.json"),
Case("/static/nope.json"),
Case("/openapi.json", compare=compare_openapi),
Case("/docs", compare=compare_html_docs),
Case("/health", compare=compare_health),
Case("/metrics", compare=compare_metrics),
Case("/stations"),
Case("/stations/8"),
Case("/stations/1"),
Case("/stations/999"),
Case("/stations/abc"),
Case(f"/measurements/latest?limit={r.randint(200, 499)}"),
Case(f"/measurements/latest?limit={r.randint(1, 15)}"),
Case(f"/measurements/history/P.1?hours={hours}&limit={limit}"),
Case(f"/measurements/history/P.67?hours={hours + 1}"),
Case(f"/measurements/history/P.1?start={day}&end={day2}&limit={limit - 1}"),
Case(f"/measurements/history/P.1?start={day}&limit={r.randint(5, 50)}"),
Case(f"/measurements/history/P.1?start={day2}&end={day}"),
Case(f"/measurements/history/NOPE?hours={hours}"),
Case("/measurements/history/P.1?hours=0"),
Case("/measurements/history/P.1?limit=100001"),
Case("/measurements/history/P.1?start=yesterday"),
Case(f"/measurements/station/P.1?hours={r.randint(20, 48)}"),
Case("/measurements/station/P.103?hours=6&limit=2"),
Case(f"/api/hii/rainfall/latest?hours={feed_hours}", sort_key="station_id"),
Case(
f"/api/hii/waterlevel/latest?hours={feed_hours + 1}", sort_key="station_id"
),
Case("/api/hii/waterlevel/latest?hours=500"),
Case(
f"/api/hii/rainfall/catchment?days={r.randint(15, 30)}",
compare=compare_catchment,
),
Case("/sensors/thaiwater", sort_key="id", soft=True),
Case("/forecast"),
Case(f"/api/forecast/history/P.1?hours={hours}&horizon=24"),
Case(f"/api/forecast/history/P.103?hours={feed_hours}"),
Case(f"/api/forecast/history/P.1?start={day}&end={day2}&horizon=6"),
Case("/api/forecast/history/P.1?horizon=99"),
Case("/api/notifications"),
# a station/horizon pair the dashboard does not ask for: not in the
# FastAPI app's 15-minute cache; the dashboard's own one may be
Case(
f"/api/forecast/skill?station_code={skill_station}"
f"&horizon={r.randint(1, 48)}",
compare=compare_skill,
),
Case(
"/api/forecast/skill?station_code=P.1&horizon=24",
compare=compare_skill,
soft=True,
),
Case("/api/forecast/skill?horizon=0"),
Case("/api/stats", compare=compare_stats),
Case("/scraping/status", compare=compare_status),
Case("/nope"),
Case("/forecast", method="DELETE"),
Case("/config"),
Case("/scrape/trigger", method="POST"),
Case("/stations", method="POST", body={"station_code": "X"}),
Case("/stations/8", method="DELETE"),
]
if admin_key:
key = {"X-API-Key": admin_key}
out += [
Case("/config", headers={"X-API-Key": "wrong"}),
Case("/stations", method="POST", body={"thai_name": 1}, headers=key),
Case("/stations/999", method="PUT", body={}, headers=key),
]
return out
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
ap.add_argument("--python", default="http://127.0.0.1:8000")
ap.add_argument("--rust", default="http://127.0.0.1:8001")
ap.add_argument(
"--admin-key", help="ADMIN_API_KEY, to compare the admin errors too"
)
ap.add_argument("--retries", type=int, default=2)
ap.add_argument("--pause", type=float, default=30.0, help="seconds before a retry")
args = ap.parse_args()
failed, warned, passed = [], [], 0
for case in cases(args.admin_key):
label = f"{case.method} {case.path}"
for attempt in range(args.retries + 1):
try:
hard, warn = case.run(args.python, args.rust)
except Exception as e: # noqa: BLE001 - report and go on
hard, warn = f"{type(e).__name__}: {e}", None
if not hard or attempt == args.retries:
break
print(f" retry {label}: {hard}", flush=True)
time.sleep(args.pause)
if hard:
failed.append((label, hard))
print(f"FAIL {label}: {hard}", flush=True)
elif warn:
warned.append((label, warn))
print(f"warn {label}: {warn}", flush=True)
else:
passed += 1
print(
f"{passed} identical, {len(warned)} warnings, {len(failed)} different "
f"({args.python} vs {args.rust})"
)
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(main())
scripts/backfill_rain_db.py
+9 -13
View File
@@ -2,7 +2,7 @@
"""Backfill the openmeteo_rain table with the full 2021+ catchment history.
Usage:
uv run scripts/backfill_rain_db.py # DB from Config/.env
uv run scripts/backfill_rain_db.py # DB from POSTGRES_* / .env
uv run scripts/backfill_rain_db.py --db-url postgresql://...
"""
@@ -13,7 +13,7 @@ import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from src.config import Config
from src.config import database_url
from src.ml.rain import backfill_db
@@ -25,21 +25,17 @@ def main(argv=None) -> int:
logging.basicConfig(
level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s"
)
if args.db_url:
# postgresql+psycopg2://... -> postgresql (driver suffix is not a dialect)
connection_string = args.db_url
db_type = args.db_url.split(":", 1)[0].split("+", 1)[0]
else:
cfg = Config.get_database_config()
if cfg["type"] not in ("sqlite", "postgresql", "mysql"):
print(f"requires a SQL database, got {cfg['type']}", file=sys.stderr)
return 1
connection_string, db_type = cfg["connection_string"], cfg["type"]
connection_string = args.db_url or database_url()
if not connection_string:
print(
"no database: set POSTGRES_CONNECTION_STRING or POSTGRES_*", file=sys.stderr
)
return 1
from sqlalchemy import create_engine
engine = create_engine(connection_string, pool_pre_ping=True)
saved = backfill_db(engine, db_type)
saved = backfill_db(engine)
print(f"backfilled {saved} hourly rows into openmeteo_rain")
return 0 if saved else 1
scripts/backfill_rid_reservoir.py
+10 -19
View File
@@ -3,7 +3,7 @@
Two paths, both idempotent and both skipping what is already stored, so a
rerun repairs holes left by transient failures and is safe alongside the
hourly live collector:
live collector (ping-collector, today + yesterday):
--dam-id (default: Mae Ngat) one dam, whole range, via api/dam — a handful
of requests for the entire 2009-today archive
@@ -26,13 +26,8 @@ import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from src.config import Config
from src.rid_reservoir import (
MAE_NGAT_DAM_ID,
RidReservoirStore,
backfill,
backfill_dam,
)
from src.config import database_url
from src.rid_reservoir import MAE_NGAT_DAM_ID, RidReservoirStore, backfill, backfill_dam
DEFAULT_START = datetime.date(2018, 8, 1) # start of the water_measurements grid
@@ -71,18 +66,14 @@ def main(argv=None) -> int:
logging.basicConfig(
level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s"
)
if args.db_url:
# postgresql+psycopg2://... -> postgresql (driver suffix is not a dialect)
connection_string = args.db_url
db_type = args.db_url.split(":", 1)[0].split("+", 1)[0]
else:
cfg = Config.get_database_config()
if cfg["type"] not in ("sqlite", "postgresql", "mysql"):
print(f"requires a SQL database, got {cfg['type']}", file=sys.stderr)
return 1
connection_string, db_type = cfg["connection_string"], cfg["type"]
connection_string = args.db_url or database_url()
if not connection_string:
print(
"no database: set POSTGRES_CONNECTION_STRING or POSTGRES_*", file=sys.stderr
)
return 1
store = RidReservoirStore(connection_string, db_type)
store = RidReservoirStore(connection_string)
if not store.connect():
print(
"database connection failed — check the connection string",
scripts/backtest_render.py
+188 -75
View File
@@ -14,7 +14,7 @@ non-zero exit if the model gives less than 12 h of warning before the first
3.70 m crossing of the 2024 event.
Usage:
python scripts/backtest_render.py # uses FLOOD_ML_DB_URL/Config
python scripts/backtest_render.py # uses FLOOD_ML_DB_URL / POSTGRES_*
python scripts/backtest_render.py --db-url postgresql://...
"""
@@ -60,7 +60,11 @@ def fit_backtest_model(df_long: pd.DataFrame, train_end: str, use_dam: bool = Fa
rain_series = rain_mod.catchment_mean(rain_mod.load_history())
dam_frame = dam_mod.load_history() if use_dam else None
X, Y, _meta = features.build_matrix(
df_long, STATION, (HORIZON,), stats_end=train_end, rain=rain_series,
df_long,
STATION,
(HORIZON,),
stats_end=train_end,
rain=rain_series,
dam=dam_frame,
)
train_mask = X.index <= pd.Timestamp(train_end)
@@ -92,6 +96,7 @@ def event_series(df_long, X, reg, clf, window_start: str, window_end: str):
# whichever is more alarmed. The classifier alone proved unreliable on
# out-of-distribution extremes (silent on the 2024 record flood).
import numpy as np
p_clf = clf.predict_proba(Xw)[:, 1]
p_sig = 1.0 / (1.0 + np.exp(-(forecasts["pred_max"] - STAGE1) / 0.15))
forecasts["p_flood"] = np.maximum(p_clf, p_sig)
@@ -109,55 +114,107 @@ def _style_axes(ax):
ax.set_axisbelow(True)
def render(observed, forecasts, flood_start, first_alert, out_path, *,
title, subtitle, detail=False, show_stage7=False, peak_note=None):
def render(
observed,
forecasts,
flood_start,
first_alert,
out_path,
*,
title,
subtitle,
detail=False,
show_stage7=False,
peak_note=None,
):
fig, (ax, axp) = plt.subplots(
2, 1, figsize=(12.6, 7.6), sharex=True,
2,
1,
figsize=(12.6, 7.6),
sharex=True,
gridspec_kw={"height_ratios": [2.2, 1], "hspace": 0.12},
)
fig.patch.set_facecolor("white")
marker = dict(marker="o", markersize=3) if detail else {}
ax.plot(observed.index, observed.values, color=BLUE, linewidth=2.2,
label="Observed level" + (" (hourly)" if detail else ""), **marker)
ax.plot(
observed.index,
observed.values,
color=BLUE,
linewidth=2.2,
label="Observed level" + (" (hourly)" if detail else ""),
**marker,
)
marker = dict(marker="s", markersize=3) if detail else {}
ax.plot(forecasts.index, forecasts["pred_max"], color=AMBER, linewidth=2,
linestyle="--", label="Predicted 24 h peak (issued at that hour)", **marker)
ax.plot(
forecasts.index,
forecasts["pred_max"],
color=AMBER,
linewidth=2,
linestyle="--",
label="Predicted 24 h peak (issued at that hour)",
**marker,
)
ax.axhline(STAGE1, color=RED, linewidth=1, alpha=0.65)
ax.annotate(f"{STAGE1:.2f} m · stage 1 · flooding begins", xy=(0.06, STAGE1),
xycoords=("axes fraction", "data"), xytext=(0, 5),
textcoords="offset points", color=RED, fontsize=10.5)
ax.annotate(
f"{STAGE1:.2f} m · stage 1 · flooding begins",
xy=(0.06, STAGE1),
xycoords=("axes fraction", "data"),
xytext=(0, 5),
textcoords="offset points",
color=RED,
fontsize=10.5,
)
if show_stage7:
ax.axhline(STAGE7, color=RED, linewidth=1, alpha=0.65)
ax.annotate(f"{STAGE7:.2f} m · stage 7 · widespread", xy=(0.06, STAGE7),
xycoords=("axes fraction", "data"), xytext=(0, 5),
textcoords="offset points", color=RED, fontsize=10.5)
ax.annotate(
f"{STAGE7:.2f} m · stage 7 · widespread",
xy=(0.06, STAGE7),
xycoords=("axes fraction", "data"),
xytext=(0, 5),
textcoords="offset points",
color=RED,
fontsize=10.5,
)
if peak_note:
peak_ts = observed.idxmax()
ax.annotate(peak_note, xy=(peak_ts, observed.max()),
xytext=(12, 10), textcoords="offset points",
color=BLUE, fontsize=11.5, fontweight="bold")
ax.annotate(
peak_note,
xy=(peak_ts, observed.max()),
xytext=(12, 10),
textcoords="offset points",
color=BLUE,
fontsize=11.5,
fontweight="bold",
)
ax.set_ylabel("P.1 water level (m)", color=INK, fontsize=11.5)
ax.legend(loc="upper left", frameon=False, fontsize=10.5)
_style_axes(ax)
axp.plot(forecasts.index, forecasts["p_flood"], color=AMBER, linewidth=1.8)
axp.fill_between(forecasts.index, 0, forecasts["p_flood"],
color=AMBER, alpha=0.28)
axp.fill_between(forecasts.index, 0, forecasts["p_flood"], color=AMBER, alpha=0.28)
axp.axhline(0.5, color=INK, linewidth=0.9, linestyle=":", alpha=0.6)
axp.set_ylim(-0.02, 1.1)
axp.set_ylabel(f"P(flooding within {HORIZON} h)", color=INK, fontsize=11.5)
_style_axes(axp)
if first_alert is not None:
lead_h = None if flood_start is None else \
int((flood_start - first_alert).total_seconds() // 3600)
lead_txt = "" if lead_h is None else (
f"\n({lead_h} h before flooding began)" if lead_h >= 0
else f"\n({-lead_h} h after flooding began)"
lead_h = (
None
if flood_start is None
else int((flood_start - first_alert).total_seconds() // 3600)
)
lead_txt = (
""
if lead_h is None
else (
f"\n({lead_h} h before flooding began)"
if lead_h >= 0
else f"\n({-lead_h} h after flooding began)"
)
)
if detail and flood_start is not None:
for a in (ax, axp):
@@ -166,34 +223,61 @@ def render(observed, forecasts, flood_start, first_alert, out_path, *,
# Anchor labels away from each other in chronological order so a
# late alert (alert AFTER crossing) cannot overprint the labels.
events = sorted(
[(first_alert, "model alert", AMBER), (flood_start, "flooding begins", BLUE)]
[
(first_alert, "model alert", AMBER),
(flood_start, "flooding begins", BLUE),
]
)
for (ts, label, color), (offset, align) in zip(events, ((-8, "right"), (8, "left"))):
ax.annotate(f"{label}\n{ts:%d %b %H:%M}",
xy=(ts, observed.min()), xytext=(offset, 18),
textcoords="offset points", ha=align,
color=color, fontsize=11, fontweight="bold")
for (ts, label, color), (offset, align) in zip(
events, ((-8, "right"), (8, "left"))
):
ax.annotate(
f"{label}\n{ts:%d %b %H:%M}",
xy=(ts, observed.min()),
xytext=(offset, 18),
textcoords="offset points",
ha=align,
color=color,
fontsize=11,
fontweight="bold",
)
mid_y = observed.min() + (observed.max() - observed.min()) * 0.28
ax.annotate("", xy=(flood_start, mid_y), xytext=(first_alert, mid_y),
arrowprops=dict(arrowstyle="<->", color=INK, lw=1.3))
ax.annotate(
"",
xy=(flood_start, mid_y),
xytext=(first_alert, mid_y),
arrowprops=dict(arrowstyle="<->", color=INK, lw=1.3),
)
arrow_label = (
f"{lead_h} h warning" if lead_h >= 0 else f"alert {-lead_h} h late"
)
ax.annotate(arrow_label,
xy=(first_alert + (flood_start - first_alert) / 2, mid_y),
xytext=(0, 8), textcoords="offset points", ha="center",
color=INK, fontsize=11.5, fontweight="bold")
ax.annotate(
arrow_label,
xy=(first_alert + (flood_start - first_alert) / 2, mid_y),
xytext=(0, 8),
textcoords="offset points",
ha="center",
color=INK,
fontsize=11.5,
fontweight="bold",
)
else:
axp.annotate(f"first alert · {first_alert:%d %b %H:%M}{lead_txt}",
xy=(first_alert, 0.62), xytext=(10, 0),
textcoords="offset points", color=RED, fontsize=10.5,
bbox=dict(facecolor="white", alpha=0.75, edgecolor="none"))
axp.annotate(
f"first alert · {first_alert:%d %b %H:%M}{lead_txt}",
xy=(first_alert, 0.62),
xytext=(10, 0),
textcoords="offset points",
color=RED,
fontsize=10.5,
bbox=dict(facecolor="white", alpha=0.75, edgecolor="none"),
)
locator = mdates.DayLocator(interval=1 if detail else 3)
axp.xaxis.set_major_locator(locator)
axp.xaxis.set_major_formatter(mdates.DateFormatter("%d %b"))
fig.suptitle(f"{title}\n{subtitle}", x=0.07, y=0.985, ha="left",
fontsize=15, color=INK)
fig.suptitle(
f"{title}\n{subtitle}", x=0.07, y=0.985, ha="left", fontsize=15, color=INK
)
fig.subplots_adjust(top=0.885, left=0.07, right=0.97, bottom=0.07)
fig.savefig(out_path, dpi=110)
plt.close(fig)
@@ -204,16 +288,20 @@ def main(argv=None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--db-url", default=None)
parser.add_argument("--out-dir", default=os.path.join("docs", "img"))
parser.add_argument("--dam", action="store_true",
help="ablation: include Mae Ngat reservoir features "
"(2026-08 result: costs 1-3 h of alert lead)")
parser.add_argument("--no-hii-fill", action="store_true",
help="ablation: load without the HII gap-fill merge")
parser.add_argument(
"--dam",
action="store_true",
help="ablation: include Mae Ngat reservoir features "
"(2026-08 result: costs 1-3 h of alert lead)",
)
parser.add_argument(
"--no-hii-fill",
action="store_true",
help="ablation: load without the HII gap-fill merge",
)
args = parser.parse_args(argv)
df = data.load_measurements(
db_url=args.db_url, hii_fill=not args.no_hii_fill
)
df = data.load_measurements(db_url=args.db_url, hii_fill=not args.no_hii_fill)
if df.empty:
print("no measurement data available", file=sys.stderr)
return 1
@@ -222,41 +310,66 @@ def main(argv=None) -> int:
# --- October 2024 record flood: trained only on data before 1 Sep 2024 ---
X, reg, clf = fit_backtest_model(df, "2024-08-31", use_dam=args.dam)
obs, fc, flood_start, first_alert = event_series(
df, X, reg, clf, "2024-09-10", "2024-10-14 23:00")
df, X, reg, clf, "2024-09-10", "2024-10-14 23:00"
)
peak = float(obs.max())
render(obs, fc, flood_start, first_alert,
os.path.join(args.out_dir, "backtest-2024-p1.png"),
title="October 2024 flood: what the model saw coming",
subtitle="P.1 Nawarat Bridge — model trained only on data before 1 Sep 2024",
show_stage7=True, peak_note=f"record peak {peak:.2f} m")
render(
obs,
fc,
flood_start,
first_alert,
os.path.join(args.out_dir, "backtest-2024-p1.png"),
title="October 2024 flood: what the model saw coming",
subtitle="P.1 Nawarat Bridge — model trained only on data before 1 Sep 2024",
show_stage7=True,
peak_note=f"record peak {peak:.2f} m",
)
obs_d, fc_d, flood_d, alert_d = event_series(
df, X, reg, clf, "2024-09-21 18:00", "2024-09-28 06:00")
df, X, reg, clf, "2024-09-21 18:00", "2024-09-28 06:00"
)
lead_h = None
if alert_d is not None and flood_d is not None:
lead_h = int((flood_d - alert_d).total_seconds() // 3600)
render(obs_d, fc_d, flood_d, alert_d,
os.path.join(args.out_dir, "backtest-2024-p1-detail.png"),
title="Detection in detail: 22–28 September 2024, hour by hour",
subtitle=(
f"the model alerts {lead_h} h before the river crosses the flooding line"
if lead_h is not None and lead_h > 0
else "model alert vs the river crossing the flooding line"
),
detail=True)
render(
obs_d,
fc_d,
flood_d,
alert_d,
os.path.join(args.out_dir, "backtest-2024-p1-detail.png"),
title="Detection in detail: 22–28 September 2024, hour by hour",
subtitle=(
f"the model alerts {lead_h} h before the river crosses the flooding line"
if lead_h is not None and lead_h > 0
else "model alert vs the river crossing the flooding line"
),
detail=True,
)
# --- September 2025 flood: the deployed configuration (trained <= 2024) ---
X25, reg25, clf25 = fit_backtest_model(df, "2024-12-31", use_dam=args.dam)
obs25, fc25, flood25, alert25 = event_series(
df, X25, reg25, clf25, "2025-09-22", "2025-10-02 12:00")
pred_at_alert = float(fc25.loc[alert25:, "pred_max"].iloc[:24].max()) if alert25 is not None else None
df, X25, reg25, clf25, "2025-09-22", "2025-10-02 12:00"
)
pred_at_alert = (
float(fc25.loc[alert25:, "pred_max"].iloc[:24].max())
if alert25 is not None
else None
)
note = f"peak {float(obs25.max()):.2f} m" + (
f" (predicted {pred_at_alert:.2f} m)" if pred_at_alert is not None else "")
render(obs25, fc25, flood25, alert25,
os.path.join(args.out_dir, "backtest-2025-p1.png"),
title="The September 2025 flood — as forecast by the deployed configuration",
subtitle="model trained only on data through 2024; this event was never seen in training",
detail=True, peak_note=note)
f" (predicted {pred_at_alert:.2f} m)" if pred_at_alert is not None else ""
)
render(
obs25,
fc25,
flood25,
alert25,
os.path.join(args.out_dir, "backtest-2025-p1.png"),
title="The September 2025 flood — as forecast by the deployed configuration",
subtitle="model trained only on data through 2024; this event was never seen in training",
detail=True,
peak_note=note,
)
print(f"2024: flooding began {flood_start}, first alert {first_alert}")
print(f"2025: flooding began {flood25}, first alert {alert25}")
scripts/collector_parity.py
-150
View File
@@ -1,150 +0,0 @@
"""Live parity check: Python collectors vs the Rust ping-collector.
Runs one collection of every source with the Python code into PY_DB and with
`ping-collector` into RS_DB (both should start empty), back to back against the
real upstream APIs, then diffs all nine collector tables row by row (ids and
created/updated timestamps ignored). Exit status 1 on any difference.
uv run python scripts/collector_parity.py \
--py-db postgresql://postgres:ping@localhost:55432/py \
--rs-db postgresql://postgres:ping@localhost:55432/rs \
--collector rust/target/debug/ping-collector
Upstream snapshots can move between the two runs (a new HII hour, a revised
RID value); rerun before treating a small difference as a bug.
"""
import argparse
import os
import subprocess
import sys
from pathlib import Path
REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO))
TABLES = {
"stations": (
"id",
"id, station_code, thai_name, english_name, latitude, longitude, geohash",
),
"water_measurements": (
"timestamp, station_id",
"timestamp, station_id, water_level, discharge, discharge_percent, status",
),
"hii_rain_stations": (
"id",
"id, oldcode, name_th, name_en, latitude, longitude, sub_basin_id, agency",
),
"hii_rainfall": (
"station_id, timestamp",
"station_id, timestamp, rain_1h, rain_24h",
),
"hii_wl_stations": (
"id",
"id, oldcode, rid_code, name_th, name_en, latitude, longitude, agency, "
"river_name, offset_msl, ground_level_msl, min_bank_msl, "
"critical_level_msl, critical_level_m, qmax, is_key_station",
),
"hii_waterlevel": (
"station_id, timestamp",
"station_id, timestamp, wl_msl, wl_m, discharge, flow_rate, "
"storage_percent, situation_level, diff_wl_bank",
),
"rid_dams": (
"dam_id",
"dam_id, region, name_th, latitude, longitude, capacity_max_mcm, "
"capacity_normal_mcm",
),
"rid_reservoir_daily": (
"dam_id, date",
"dam_id, date, storage_mcm, storage_pct, inflow_mcm, outflow_mcm, level_msl",
),
"openmeteo_rain": (
"timestamp",
"timestamp, chiang_dao, mae_taeng, mae_ngat, mae_rim, chiang_mai, "
"catchment_mean",
),
}
def run_python(db_url: str, gap_days: int):
os.environ.update(
DB_TYPE="postgresql", POSTGRES_CONNECTION_STRING=db_url, LOG_LEVEL="WARNING"
)
from sqlalchemy import create_engine
from src.config import Config
from src.hii_collector import HiiCollector
from src.ml import rain
from src.rid_reservoir import RidReservoirCollector
from src.water_scraper_v3 import EnhancedWaterMonitorScraper
cfg = Config.get_database_config()
scraper = EnhancedWaterMonitorScraper(cfg)
scraper.fill_data_gaps(gap_days)
scraper.run_scraping_cycle()
HiiCollector(cfg).run_cycle()
RidReservoirCollector(cfg).run_cycle()
rain.save_to_db(rain.fetch_forecast(), create_engine(db_url), "postgresql")
def run_rust(db_url: str, collector: str, gap_days: int):
env = {**os.environ, "POSTGRES_CONNECTION_STRING": db_url, "DB_TYPE": "postgresql"}
for args in (["fill-gaps", str(gap_days)], ["once"]):
subprocess.run([collector, *args], env=env, check=True)
def table_rows(db_url: str, table: str, key: str, cols: str) -> dict:
import psycopg2
with psycopg2.connect(db_url) as conn:
cur = conn.cursor()
cur.execute(f"SELECT {cols} FROM {table} ORDER BY {key}")
nkey = len(key.split(","))
return {row[:nkey]: row for row in cur.fetchall()}
def diff(py_db: str, rs_db: str) -> bool:
identical = True
for table, (key, cols) in TABLES.items():
a = table_rows(py_db, table, key, cols)
b = table_rows(rs_db, table, key, cols)
only_py, only_rs = sorted(set(a) - set(b)), sorted(set(b) - set(a))
differ = sorted(k for k in set(a) & set(b) if a[k] != b[k])
same = not (only_py or only_rs or differ)
identical &= same
print(
f"{table:<20} py {len(a):>6} rs {len(b):>6} only-py {len(only_py):>4} "
f"only-rs {len(only_rs):>4} differing {len(differ):>4} "
f"{'identical' if same else 'DIFFERENT'}"
)
for k in only_py[:3]:
print(" only py:", a[k])
for k in only_rs[:3]:
print(" only rs:", b[k])
for k in differ[:3]:
print(" py:", a[k], "\n rs:", b[k])
return identical
def main():
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--py-db", required=True)
parser.add_argument("--rs-db", required=True)
parser.add_argument(
"--collector", default=str(REPO / "rust/target/debug/ping-collector")
)
parser.add_argument("--gap-days", type=int, default=2)
parser.add_argument("--diff-only", action="store_true")
args = parser.parse_args()
if not args.diff_only:
run_python(args.py_db, args.gap_days)
run_rust(args.rs_db, args.collector, args.gap_days)
ok = diff(args.py_db, args.rs_db)
print("ALL IDENTICAL" if ok else "differences found")
sys.exit(0 if ok else 1)
if __name__ == "__main__":
main()
scripts/dev_proxy.py
+15 -4
View File
@@ -25,18 +25,29 @@ class Handler(http.server.BaseHTTPRequestHandler):
self._send(200, "application/json", Path(file).read_bytes())
return
if self.path.startswith("/static/"):
f = STATIC / self.path[len("/static/"):].split("?")[0]
f = STATIC / self.path[len("/static/") :].split("?")[0]
if f.is_file():
ctype = "application/json" if f.suffix in (".json", ".geojson") else "application/octet-stream"
ctype = (
"application/json"
if f.suffix in (".json", ".geojson")
else "application/octet-stream"
)
self._send(200, ctype, f.read_bytes())
return
try:
req = urllib.request.Request(
UPSTREAM + self.path,
headers={"User-Agent": "Mozilla/5.0 (dev_proxy; +https://buildfor.life)", "Accept": "application/json"},
headers={
"User-Agent": "Mozilla/5.0 (dev_proxy; +https://buildfor.life)",
"Accept": "application/json",
},
)
with urllib.request.urlopen(req, timeout=60) as r:
self._send(r.status, r.headers.get("Content-Type", "application/json"), r.read())
self._send(
r.status,
r.headers.get("Content-Type", "application/json"),
r.read(),
)
except urllib.error.HTTPError as e:
self._send(e.code, "application/json", e.read())
scripts/e2e_notify.py
-132
View File
@@ -1,132 +0,0 @@
"""Drive the production notify path in-process: startup init -> seeded readings
-> forecast cache -> _notify_transitions -> sqlite state -> real ntfy."""
import asyncio
import datetime
import json
import os
import sys
import requests
os.environ.update(
DB_TYPE="sqlite",
WATER_DB_PATH=os.path.join(os.environ["LOCALAPPDATA"], "Temp", "smoke3.db"),
NTFY_SERVER="http://127.0.0.1:2586",
NTFY_TOKEN=os.environ.get("NTFY_TOKEN", ""),
NTFY_TOPIC_PREFIX="ping",
)
for f in ("smoke3.db",):
p = os.path.join(os.environ["LOCALAPPDATA"], "Temp", f)
if os.path.exists(p):
os.remove(p)
from src import web_api # noqa: E402
from src.config import Config # noqa: E402
assert Config.NTFY_SERVER
async def main():
# what the lifespan does at startup, minus the scheduler
from src import notify as notify_mod
from src.forecast_history import ForecastHistoryStore
from src.water_scraper_v3 import EnhancedWaterMonitorScraper
db_config = Config.get_database_config()
web_api.app_state["scraper"] = EnhancedWaterMonitorScraper(db_config)
store = ForecastHistoryStore(db_config["connection_string"], db_config["type"])
store.connect()
web_api.app_state["forecast_store"] = store
state = notify_mod.NotificationState(store.engine, store.db_type)
pub = notify_mod.NtfyPublisher(
Config.NTFY_SERVER, prefix=Config.NTFY_TOPIC_PREFIX, token=Config.NTFY_TOKEN
)
web_api.app_state["notify"] = (pub, state)
scraper = web_api.app_state["scraper"]
now = datetime.datetime.now().replace(minute=0, second=0, microsecond=0)
def seed(level_p1, level_p103, ts):
rows = [
{
"station_code": "P.1",
"station_id": 1,
"timestamp": ts,
"water_level": level_p1,
"discharge": 400.0,
"station_name_en": "Nawarat Bridge",
"station_name_th": "สะพานนวรัฐ",
"discharge_percent": 30.0,
"status": "active",
},
{
"station_code": "P.103",
"station_id": 2,
"timestamp": ts,
"water_level": level_p103,
"discharge": 300.0,
"station_name_en": "Ring Road 3",
"station_name_th": "วงแหวน 3",
"discharge_percent": 20.0,
"status": "active",
},
]
scraper.db_adapter.save_measurements(rows)
def forecast(p):
with web_api.FORECAST_CACHE_LOCK:
web_api.FORECAST_CACHE["all"] = (
0,
[
{
"station_code": "P.1",
"horizon_hours": 24,
"p_warning": p,
"predicted_max_level": 3.9,
"source": "model",
}
],
)
def poll(topic):
out = []
for line in (
requests.get(f"{Config.NTFY_SERVER}/{topic}/json?poll=1", timeout=5)
.text.strip()
.splitlines()
):
m = json.loads(line)
if m.get("event") == "message":
out.append(m.get("title") or m.get("message", "")[:40])
return out
# cycle 1: quiet
seed(1.6, 3.2, now - datetime.timedelta(hours=2))
forecast(0.02)
await web_api._notify_transitions()
# cycle 2: P.1 crosses warning, model outlook on
seed(3.75, 3.3, now - datetime.timedelta(hours=1))
forecast(0.7)
await web_api._notify_transitions()
# cycle 3: same state -> silence
seed(3.80, 3.3, now)
forecast(0.65)
await web_api._notify_transitions()
print("ping-p1-warning:", poll("ping-p1-warning"))
print("ping-warning: ", poll("ping-warning"))
print("ping-p1-outlook:", poll("ping-p1-outlook"))
print("ping-p103-warning:", poll("ping-p103-warning"))
from sqlalchemy import text
with store.engine.connect() as c:
print(
"state table:",
c.execute(
text("SELECT key, state, value FROM notification_state ORDER BY key")
).fetchall(),
)
asyncio.run(main())
scripts/forecast_parity.py
+5 -3
View File
@@ -1,8 +1,10 @@
#!/usr/bin/env python3
"""Compare the Python forecast with ping-collector's on the live database.
"""Compare ping-collector's forecast with the Python reference on the live database.
Run before setting FORECAST_IN_COLLECTOR=true, and after any change to src/ml
or rust/forecast. On the server, as the service user from the app dir:
src/ml/predict.py is the reference implementation the Rust forecaster
(rust/forecast) was ported from. Run after any change to src/ml or
rust/forecast; scripts/install.sh runs it on every install. On the server, as
the service user from the app dir:
sudo -u water-monitor .venv/bin/python scripts/forecast_parity.py
scripts/install.sh
+173 -98
View File
@@ -1,127 +1,202 @@
#!/usr/bin/env bash
# Install or update the Ping River Monitor on a systemd host.
#
# Install the Thailand Water Level Monitor as a hardened systemd service.
# sudo bash scripts/install.sh
#
# Creates a dedicated system user, deploys the code to /opt, builds a uv-managed
# virtualenv, installs the systemd unit, and enables the service. Idempotent:
# safe to re-run to update an existing install.
#
# Usage (as root, from a checkout of the repo):
# sudo bash scripts/install.sh
#
# Override defaults via environment variables:
# APP_DIR=/opt/thailand-water-monitor SERVICE_USER=water-monitor sudo -E bash scripts/install.sh
# What runs:
# ping-collector.service the service (rust/collector): RID/HII/dam/rain
# collection, forecast, ntfy + Matrix
# notifications, website + API on WEB_LISTEN
# water-monitor-retrain.timer monthly model retrain (Python, scripts/retrain.sh)
#
# Re-run after every update (git pull + build). Steps, all idempotent:
# 1. system user/group (water-monitor)
# 2. code: run from a checkout other than APP_DIR, it is rsynced there (the
# trained models, .env, stations.json and bin/ are kept) and handed to
# the service user. Production runs in place (APP_DIR is the checkout):
# then nothing is copied and no ownership changes.
# 3. the binary: BIN=/path/to/ping-collector (e.g. the CI artifact
# "ping-collector-linux-x86_64") or rust/target/release/ping-collector,
# built beforehand as a normal user:
# cargo build --release --locked --manifest-path rust/Cargo.toml
# (Rust >= 1.88 from rustup, plus pkg-config and libssl-dev)
# 4. the Python training environment (.venv, `uv sync --frozen`): in place
# as the user running sudo (who pulled and built), otherwise as root;
# SKIP_UV=1 skips it
# 5. .env: created from .env.example on a fresh install (edit, then re-run);
# WEB_LISTEN and FORECAST_IN_COLLECTOR default to the production values
# when absent
# 6. systemd units; the retired Python web app (water-monitor.service) and
# the old `run.py --alert-check` Matrix cron job are removed
# 7. checks (SKIP_CHECKS=1 skips them): a dry run of every source against
# the database, and the forecast compared with the Python reference on
# the live data (scripts/forecast_parity.py)
# 8. (re)start, then wait for the website to answer
set -euo pipefail
APP_DIR="${APP_DIR:-/opt/thailand-water-monitor}"
SERVICE_USER="${SERVICE_USER:-water-monitor}"
SERVICE_USER="${SERVICE_USER:-water-monitor}" # as in the units
SERVICE_GROUP="${SERVICE_GROUP:-${SERVICE_USER}}"
SERVICE_NAME="water-monitor.service"
RETRAIN_NAME="water-monitor-retrain"
# Resolve the repo root (parent of this scripts/ directory).
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
ENV_FILE="${APP_DIR}/.env"
PYTHON="${APP_DIR}/.venv/bin/python"
REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
RETRAIN="water-monitor-retrain"
log() { printf '\033[1;32m==>\033[0m %s\n' "$*"; }
warn() { printf '\033[1;33m[warn]\033[0m %s\n' "$*"; }
warn() { printf '\033[1;33m[warn]\033[0m %s\n' "$*" >&2; }
die() { printf '\033[1;31m[error]\033[0m %s\n' "$*" >&2; exit 1; }
[ "$(id -u)" -eq 0 ] || die "This script must be run as root (use sudo)."
[ "$(id -u)" -eq 0 ] || die "run as root (sudo)"
# 1. Dedicated system user/group (no login, no home) --------------------------
if ! getent group "${SERVICE_GROUP}" >/dev/null; then
log "Creating group ${SERVICE_GROUP}"
groupadd --system "${SERVICE_GROUP}"
fi
if ! id "${SERVICE_USER}" >/dev/null 2>&1; then
log "Creating system user ${SERVICE_USER}"
set_default() { # key value: append when the key is not set at all
grep -q "^$1=" "$ENV_FILE" || printf '%s=%s\n' "$1" "$2" >>"$ENV_FILE"
}
# A command the way the services run it: same user, working directory and
# EnvironmentFile parsing (systemd's).
as_service() {
if systemd-run --version >/dev/null 2>&1; then
systemd-run --quiet --pipe --wait --collect \
-p User="$SERVICE_USER" -p Group="$SERVICE_GROUP" \
-p WorkingDirectory="$APP_DIR" -p EnvironmentFile="$ENV_FILE" "$@"
else
(cd "$APP_DIR" && runuser -u "$SERVICE_USER" -- "$@")
fi
}
wait_http() { # url seconds
local i
for i in $(seq "$2"); do
curl -fs -o /dev/null --max-time 5 "$1" 2>/dev/null && return 0
sleep 1
done
return 1
}
# 1. user ---------------------------------------------------------------------
getent group "$SERVICE_GROUP" >/dev/null || groupadd --system "$SERVICE_GROUP"
if ! id "$SERVICE_USER" >/dev/null 2>&1; then
log "Creating system user $SERVICE_USER"
useradd --system --no-create-home --shell /usr/sbin/nologin \
--gid "${SERVICE_GROUP}" "${SERVICE_USER}"
--gid "$SERVICE_GROUP" "$SERVICE_USER"
fi
# 2. Deploy code to APP_DIR ----------------------------------------------------
log "Deploying code to ${APP_DIR}"
mkdir -p "${APP_DIR}"
if command -v rsync >/dev/null 2>&1; then
# 2. code ---------------------------------------------------------------------
in_place=0
[ -d "$APP_DIR" ] && [ "$(cd "$APP_DIR" && pwd -P)" = "$(cd "$REPO_DIR" && pwd -P)" ] && in_place=1
if [ "$in_place" = 0 ]; then
command -v rsync >/dev/null || die "rsync is needed to copy $REPO_DIR to $APP_DIR"
log "Copying $REPO_DIR to $APP_DIR"
mkdir -p "$APP_DIR"
rsync -a --delete \
--exclude '.git' --exclude '.venv' --exclude 'venv' \
--exclude '__pycache__' --exclude '*.pyc' \
--exclude '*.db' --exclude '.env' --exclude 'stations.json' \
"${REPO_DIR}/" "${APP_DIR}/"
else
warn "rsync not found; falling back to cp (will not prune deleted files)"
cp -r "${REPO_DIR}/." "${APP_DIR}/"
--exclude '.git' --exclude '.venv' --exclude '__pycache__' --exclude '/rust/target' \
--exclude '/.env' --exclude '/stations.json' --exclude '/bin' \
--exclude '/collector-status.json' --exclude '/crontab-*.bak' \
--exclude '/models/flood_*' --exclude '/models/metrics.json' \
--exclude '/models/cache' --exclude '/models/.staging' --exclude '/models/.previous' \
"$REPO_DIR/" "$APP_DIR/"
fi
# 3. Build the uv-managed virtualenv ------------------------------------------
# Prefer an already-installed uv. For stricter supply-chain control install uv
# ahead of time via your distro / package manager; this script only fetches the
# upstream installer (piped to a root shell) when AUTO_INSTALL_UV=1 is set, and
# pins the version so the fetched script is reproducible.
UV_VERSION="${UV_VERSION:-0.5.11}"
if ! command -v uv >/dev/null 2>&1; then
if [ "${AUTO_INSTALL_UV:-0}" = "1" ]; then
warn "uv not found; installing pinned uv ${UV_VERSION} from astral.sh (runs as root)"
curl -LsSf "https://astral.sh/uv/${UV_VERSION}/install.sh" \
| env UV_INSTALL_DIR=/usr/local/bin sh
# 3. binary -------------------------------------------------------------------
BIN="${BIN:-${REPO_DIR}/rust/target/release/ping-collector}"
[ -x "$BIN" ] || die "no binary at $BIN: build it (see the header) or pass BIN=/path/to/ping-collector"
log "Installing $BIN -> ${APP_DIR}/bin/ping-collector ($("$BIN" --version))"
install -D -m 0755 -o root -g root "$BIN" "${APP_DIR}/bin/ping-collector"
# 4. Python training environment ------------------------------------------------
if [ "${SKIP_UV:-0}" != 1 ]; then
uv_user=root # copy mode: handed to the service user below
[ "$in_place" = 1 ] && uv_user="${SUDO_USER:-root}"
log "Syncing the Python training environment (${APP_DIR}/.venv) as $uv_user"
if [ "$uv_user" = root ]; then
(cd "$APP_DIR" && uv sync --python 3.11 --frozen) ||
warn "uv sync failed: the monthly retrain needs ${APP_DIR}/.venv (uv sync --python 3.11 --frozen)"
else
die "uv not found. Install it (e.g. your package manager, or 'pipx install uv'),
or re-run with AUTO_INSTALL_UV=1 to fetch the pinned upstream installer."
fi
fi
UV="$(command -v uv)"
log "Syncing uv-managed virtualenv at ${APP_DIR}/.venv"
cd "${APP_DIR}"
# ONE environment: uv sync owns .venv/ (from pyproject.toml + uv.lock, so the
# ML extras such as scikit-learn/joblib are present) and both systemd units
# run its interpreter directly. Never create a second env by another name --
# a stale 'venv/' once coexisted here and broke manual retrains with
# ModuleNotFoundError while the service itself ran fine.
"${UV}" sync --python 3.11 --frozen
if [ -d "${APP_DIR}/venv" ]; then
warn "Removing stale ${APP_DIR}/venv (superseded by .venv)"
rm -rf "${APP_DIR}/venv"
fi
# 4. Environment file ----------------------------------------------------------
if [ ! -f "${APP_DIR}/.env" ]; then
if [ -f "${REPO_DIR}/.env" ]; then
log "Copying .env from checkout"
cp "${REPO_DIR}/.env" "${APP_DIR}/.env"
else
warn "No .env found. Copy .env.example to ${APP_DIR}/.env and fill in"
warn "MATRIX_ACCESS_TOKEN / MATRIX_ROOM_ID and DB settings before starting."
runuser -u "$uv_user" -- bash -lc "cd '$APP_DIR' && uv sync --python 3.11 --frozen" ||
warn "uv sync as $uv_user failed: run 'uv sync --python 3.11 --frozen' in $APP_DIR as the owner of .venv"
fi
fi
# 5. Ownership and permissions -------------------------------------------------
# Service user needs write access for logs / stations.json.
log "Setting ownership to ${SERVICE_USER}:${SERVICE_GROUP}"
chown -R "${SERVICE_USER}:${SERVICE_GROUP}" "${APP_DIR}"
# Restrict traversal to root + the service user, and lock down the secrets file
# (contains the Matrix token and DB credentials).
chmod 0750 "${APP_DIR}"
if [ -f "${APP_DIR}/.env" ]; then
chmod 0600 "${APP_DIR}/.env"
# 5. .env ---------------------------------------------------------------------
fresh_env=0
if [ ! -f "$ENV_FILE" ]; then
cp "${APP_DIR}/.env.example" "$ENV_FILE"
chmod 0600 "$ENV_FILE"
fresh_env=1
fi
set_default WEB_LISTEN 0.0.0.0:8000
set_default FORECAST_IN_COLLECTOR true
if [ "$in_place" = 0 ]; then
# a copy belongs to the service (it writes models/, stations.json and the
# status file); an in-place checkout keeps its owner (git pull, builds)
chown -R "${SERVICE_USER}:${SERVICE_GROUP}" "$APP_DIR"
chown root:root "${APP_DIR}/bin/ping-collector"
chmod 0750 "$APP_DIR"
fi
# 6. Install and enable the systemd units -------------------------------------
# 6. units --------------------------------------------------------------------
log "Installing systemd units"
install -m 0644 "${SCRIPT_DIR}/${SERVICE_NAME}" "/etc/systemd/system/${SERVICE_NAME}"
install -m 0644 "${SCRIPT_DIR}/${RETRAIN_NAME}.service" "/etc/systemd/system/${RETRAIN_NAME}.service"
install -m 0644 "${SCRIPT_DIR}/${RETRAIN_NAME}.timer" "/etc/systemd/system/${RETRAIN_NAME}.timer"
install -m 0644 "${APP_DIR}/scripts/ping-collector.service" /etc/systemd/system/ping-collector.service
install -m 0644 "${APP_DIR}/scripts/${RETRAIN}.service" "/etc/systemd/system/${RETRAIN}.service"
install -m 0644 "${APP_DIR}/scripts/${RETRAIN}.timer" "/etc/systemd/system/${RETRAIN}.timer"
if [ -f /etc/systemd/system/water-monitor.service ]; then
log "Removing the retired Python web app (water-monitor.service)"
systemctl disable --now water-monitor.service 2>/dev/null || true
rm -f /etc/systemd/system/water-monitor.service
fi
systemctl daemon-reload
systemctl enable "${SERVICE_NAME}"
systemctl enable --now "${RETRAIN_NAME}.timer"
systemctl enable ping-collector.service
systemctl enable --now "${RETRAIN}.timer"
log "Done."
# The old Matrix cron job (`run.py --alert-check`, repeating an alert every
# 15 minutes while a level stayed high): ping-collector sends Matrix now.
for user in root "$SERVICE_USER"; do
crontab -l -u "$user" 2>/dev/null | grep -v '^\s*#' | grep -q -- '--alert-check' || continue
backup="${APP_DIR}/crontab-${user}.$(date +%Y%m%d%H%M%S).bak"
crontab -l -u "$user" >"$backup"
sed '/--alert-check/ s|^\([^#]\)|# retired: Matrix is sent by ping-collector # \1|' "$backup" |
crontab -u "$user" -
log "Commented out the old --alert-check cron job of $user (backup: $backup)"
done
if grep -rlqs -- '--alert-check' /etc/cron.d /etc/crontab 2>/dev/null; then
warn "an --alert-check job is still in $(grep -rls -- '--alert-check' /etc/cron.d /etc/crontab | tr '\n' ' '): remove it"
fi
if [ "$fresh_env" = 1 ]; then
warn "created $ENV_FILE from .env.example: set the database (POSTGRES_*), NTFY_* and MATRIX_*, then re-run"
exit 0
fi
# 7. checks -------------------------------------------------------------------
if [ "${SKIP_CHECKS:-0}" != 1 ]; then
log "Dry run of every source against the configured database (writes nothing)"
as_service "${APP_DIR}/bin/ping-collector" once --dry-run
if compgen -G "${APP_DIR}/models/flood_*.joblib" >/dev/null && [ -x "$PYTHON" ]; then
log "Exporting trained models for the forecaster (models/flood_*.json.gz)"
as_service "$PYTHON" -m src.ml.export
log "Forecast check: ping-collector against the Python reference, same data, moment and rain"
as_service "$PYTHON" scripts/forecast_parity.py ||
warn "the forecast check found differences (see above); the service runs regardless"
else
warn "no trained models or no ${PYTHON}: forecast check skipped (the forecast falls back to the heuristic)"
fi
fi
# 8. (re)start ----------------------------------------------------------------
systemctl restart ping-collector.service
listen="$(grep '^WEB_LISTEN=' "$ENV_FILE" | tail -1 | cut -d= -f2- | tr -d '"'"'"' ')"
if [ -n "$listen" ]; then
port="${listen##*:}"
# readiness: a route that answers 200 whatever the data (/health may be 503)
if wait_http "http://127.0.0.1:${port}/api/notifications" 60; then
log "Website and API answer on :${port} ($(curl -s --max-time 5 "http://127.0.0.1:${port}/scraping/status" | grep -o '"collector":"[^"]*"'))"
else
warn "no answer on port ${port} yet: journalctl -u ping-collector"
fi
fi
sleep 3
systemctl --no-pager --lines=0 status ping-collector.service || true
journalctl -u ping-collector -n 25 --no-pager || true
echo
echo "Next steps:"
echo " sudo systemctl start ${SERVICE_NAME}"
echo " systemctl status ${SERVICE_NAME}"
echo " sudo journalctl -u ${SERVICE_NAME} -f"
echo " systemctl list-timers ${RETRAIN_NAME}.timer # monthly flood-model retrain"
echo " sudo systemctl start ${RETRAIN_NAME}.service # retrain now"
log "Done. Logs: journalctl -u ping-collector -f; retrain: systemctl list-timers ${RETRAIN}.timer"
scripts/install_collector.sh
-286
View File
@@ -1,286 +0,0 @@
#!/usr/bin/env bash
# Install the Rust service (rust/collector, `ping-collector`) and move work
# from the Python app to it, or back.
#
# sudo bash scripts/install_collector.sh # install binary + unit; checks only
# sudo bash scripts/install_collector.sh --cutover # Rust collects (COLLECTION_ENABLED=false)
# sudo bash scripts/install_collector.sh --cutover-forecast # Rust also forecasts and notifies:
# # ntfy + Matrix (FORECAST_IN_COLLECTOR=true)
# sudo bash scripts/install_collector.sh --check-api # compare the Rust website/API with the
# # Python app's, change nothing
# sudo bash scripts/install_collector.sh --cutover-api # Rust serves the website and API on
# # port 8000 (WEB_LISTEN); the Python app stops
# sudo bash scripts/install_collector.sh --rollback-api # the Python app serves the website again
# sudo bash scripts/install_collector.sh --rollback-forecast # forecasts + notifications back to Python
# sudo bash scripts/install_collector.sh --rollback # Python does everything again
#
# Every install also exports the trained models for the Rust forecaster and
# compares its forecast with Python's on the live data
# (scripts/forecast_parity.py); --cutover-forecast refuses to switch when they
# differ. --cutover-api first runs `ping-collector serve` next to the Python
# app and compares every route (scripts/api_parity.py); it refuses to switch
# when they differ. FORCE=1 overrides either check.
#
# Once ping-collector sends the Matrix messages (FORECAST_IN_COLLECTOR=true
# and MATRIX_* set), the old `run.py --alert-check` cron job is commented out
# (backup next to the app); it repeated an alert on every run.
#
# The binary comes from BIN=/path/to/ping-collector (e.g. the "ping-collector-
# linux-x86_64" CI artifact) or from rust/target/release/ping-collector, built
# beforehand as a normal user:
# cargo build --release --locked --manifest-path rust/Cargo.toml
# (needs Rust >= 1.88 from rustup -- distro cargo is too old -- plus pkg-config
# and libssl-dev; see rust/README.md).
set -euo pipefail
APP_DIR="${APP_DIR:-/opt/thailand-water-monitor}"
SERVICE_USER="${SERVICE_USER:-water-monitor}" # as in scripts/ping-collector.service
ENV_FILE="${APP_DIR}/.env"
STATUS_FILE="${APP_DIR}/collector-status.json"
PYTHON="${APP_DIR}/.venv/bin/python"
REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
WEB_ADDR="${WEB_ADDR:-0.0.0.0:8000}" # where the Python app listens today
CHECK_PORT="${CHECK_PORT:-8001}" # the Rust side during --check-api
MODE="${1:-install}"
log() { printf '\033[1;32m==>\033[0m %s\n' "$*"; }
warn() { printf '\033[1;33m[warn]\033[0m %s\n' "$*" >&2; }
die() { printf '\033[1;31m[error]\033[0m %s\n' "$*" >&2; exit 1; }
[ "$(id -u)" -eq 0 ] || die "run as root (sudo)"
[ -f "$ENV_FILE" ] || die "$ENV_FILE not found"
set_env() {
local key="$1" value="$2"
if grep -q "^${key}=" "$ENV_FILE"; then
sed -i "s|^${key}=.*|${key}=${value}|" "$ENV_FILE"
else
printf '%s=%s\n' "$key" "$value" >>"$ENV_FILE"
fi
}
# Run a command the way the services run: same user, working directory and
# EnvironmentFile parsing (systemd's, not a .env library's).
as_service() {
if systemd-run --version >/dev/null 2>&1; then
systemd-run --quiet --pipe --wait --collect \
-p User="$SERVICE_USER" -p Group="$(id -gn "$SERVICE_USER")" \
-p WorkingDirectory="$APP_DIR" -p EnvironmentFile="$ENV_FILE" "$@"
else
(cd "$APP_DIR" && runuser -u "$SERVICE_USER" -- "$@")
fi
}
show() {
sleep "$1"
systemctl --no-pager --lines=0 status ping-collector.service || true
journalctl -u ping-collector -n 25 --no-pager || true
}
env_is() { grep -q "^$1=$2\s*$" "$ENV_FILE"; }
wait_http() { # url seconds
local i
for i in $(seq "$2"); do
curl -fs -o /dev/null --max-time 5 "$1" 2>/dev/null && return 0
sleep 1
done
return 1
}
# The Matrix messages come from ping-collector once it notifies; the old
# cron job (`run.py --alert-check`, every 15 min, repeating while a level
# stays high) is commented out, with a backup of the crontab.
retire_matrix_cron() {
env_is FORECAST_IN_COLLECTOR true || return 0
grep -q '^MATRIX_ACCESS_TOKEN=..*' "$ENV_FILE" || return 0
local user backup
for user in root "$SERVICE_USER"; do
crontab -l -u "$user" 2>/dev/null | grep -v '^\s*#' | grep -q -- '--alert-check' || continue
backup="${APP_DIR}/crontab-${user}.$(date +%Y%m%d%H%M%S).bak"
crontab -l -u "$user" >"$backup"
sed '/--alert-check/ s|^\([^#]\)|# retired: Matrix is sent by ping-collector # \1|' "$backup" |
crontab -u "$user" -
log "Matrix: commented out the old --alert-check cron job of $user (backup: $backup)"
done
if grep -rlqs -- '--alert-check' /etc/cron.d /etc/crontab 2>/dev/null; then
warn "an --alert-check job is still in $(grep -rls -- '--alert-check' /etc/cron.d /etc/crontab | tr '\n' ' '): remove it, ping-collector sends Matrix now"
fi
if systemctl list-timers --all --no-pager 2>/dev/null | grep -qi alert; then
warn "a systemd timer mentions 'alert' (systemctl list-timers --all): if it runs --alert-check, disable it"
fi
}
# ping-collector serve next to the Python app; every route compared.
api_check() {
local unit=ping-api-check ok=0
systemctl stop "$unit" 2>/dev/null || true
systemd-run --quiet --unit="$unit" --collect \
-p User="$SERVICE_USER" -p Group="$(id -gn "$SERVICE_USER")" \
-p WorkingDirectory="$APP_DIR" -p EnvironmentFile="$ENV_FILE" \
"${APP_DIR}/bin/ping-collector" serve --listen "127.0.0.1:${CHECK_PORT}" --no-analytics
if wait_http "http://127.0.0.1:${CHECK_PORT}/api/notifications" 30; then
as_service "$PYTHON" scripts/api_parity.py \
--python "http://127.0.0.1:${WEB_ADDR##*:}" --rust "http://127.0.0.1:${CHECK_PORT}" && ok=1
else
warn "ping-collector serve did not answer on port ${CHECK_PORT}:"
journalctl -u "$unit" -n 20 --no-pager || true
fi
systemctl stop "$unit" 2>/dev/null || true
[ "$ok" = 1 ]
}
rollback_api() {
if env_is WEB_LISTEN '..*'; then
log "The Python app serves the website and API again"
set_env WEB_LISTEN ""
systemctl restart ping-collector.service
fi
systemctl enable water-monitor.service
systemctl restart water-monitor.service
}
matrix_backups() {
ls "${APP_DIR}"/crontab-*.bak 2>/dev/null | sed 's/^/ /' || true
}
case "$MODE" in
--rollback)
log "Stopping ping-collector; the Python app collects, serves, forecasts and notifies again"
systemctl disable --now ping-collector.service 2>/dev/null || true
set_env WEB_LISTEN ""
set_env FORECAST_IN_COLLECTOR false
set_env COLLECTION_ENABLED true
systemctl enable water-monitor.service
systemctl restart water-monitor.service
if [ -n "$(matrix_backups)" ]; then
warn "Matrix alerts came from ping-collector; to restore the old cron job: crontab -u USER BACKUP"
matrix_backups
fi
log "Done. Check: journalctl -u water-monitor -n 50"
exit 0
;;
--rollback-api)
rollback_api
log "Done. Check: journalctl -u water-monitor -n 50; curl -s http://127.0.0.1:${WEB_ADDR##*:}/health"
exit 0
;;
--rollback-forecast)
# the Python app forecasts in its own process: it must serve again
rollback_api
log "Forecasts and notifications back to the Python app; ping-collector keeps collecting"
set_env FORECAST_IN_COLLECTOR false
systemctl restart ping-collector.service water-monitor.service
if [ -n "$(matrix_backups)" ]; then
warn "Matrix alerts stop with this (the Python app sends ntfy only); old cron job: crontab -u USER BACKUP"
matrix_backups
fi
log "Done. Check: journalctl -u water-monitor -n 50"
exit 0
;;
--check-api)
[ -x "${APP_DIR}/bin/ping-collector" ] || die "install first"
if env_is WEB_LISTEN '..*'; then
die "ping-collector already serves the website (WEB_LISTEN is set)"
fi
api_check && log "The Rust API answers as the Python app" || die "the API check found differences"
exit 0
;;
install | --cutover | --cutover-forecast | --cutover-api) ;;
*) die "unknown option: $MODE" ;;
esac
BIN="${BIN:-${REPO_DIR}/rust/target/release/ping-collector}"
[ -x "$BIN" ] || die "no binary at $BIN: build it (see header) or pass BIN=/path/to/ping-collector"
log "Installing $BIN -> ${APP_DIR}/bin/ping-collector ($("$BIN" --version))"
install -D -m 0755 -o root -g root "$BIN" "${APP_DIR}/bin/ping-collector"
install -m 0644 "${REPO_DIR}/scripts/ping-collector.service" /etc/systemd/system/ping-collector.service
systemctl daemon-reload
# Both services read and write /scraping/status through this file (inside
# the app dir: the units use PrivateTmp).
set_env SCRAPING_STATUS_FILE "$STATUS_FILE"
touch "$STATUS_FILE"
chown "${SERVICE_USER}:$(id -gn "$SERVICE_USER")" "$STATUS_FILE"
log "Dry run against the configured database (writes nothing)"
as_service "${APP_DIR}/bin/ping-collector" once --dry-run
forecast_ok=1
if compgen -G "${APP_DIR}/models/flood_*.joblib" >/dev/null; then
log "Exporting the trained models for the Rust forecaster (models/flood_*.json.gz)"
as_service "$PYTHON" -m src.ml.export
log "Forecast check: Python and ping-collector on the same data, moment and rain"
as_service "$PYTHON" scripts/forecast_parity.py || forecast_ok=0
else
warn "no trained models in ${APP_DIR}/models: forecast check skipped"
forecast_ok=0
fi
case "$MODE" in
--cutover)
log "Cutover: ping-collector collects; the Python app forecasts and notifies"
set_env COLLECTION_ENABLED false
systemctl enable ping-collector.service
systemctl restart ping-collector.service water-monitor.service
show 20
log "Next step: --cutover-forecast. Rollback: sudo bash scripts/install_collector.sh --rollback"
;;
--cutover-forecast)
grep -q '^COLLECTION_ENABLED=false' "$ENV_FILE" ||
die "ping-collector must collect first: run --cutover"
if [ "$forecast_ok" != 1 ] && [ "${FORCE:-0}" != 1 ]; then
die "the forecast check did not pass; not switching (FORCE=1 overrides)"
fi
log "Cutover: ping-collector also forecasts and notifies; the Python app serves its forecast"
set_env FORECAST_IN_COLLECTOR true
retire_matrix_cron
systemctl enable ping-collector.service
systemctl restart ping-collector.service water-monitor.service
show 30
log "Next step: --cutover-api. Rollback: sudo bash scripts/install_collector.sh --rollback-forecast"
;;
--cutover-api)
env_is FORECAST_IN_COLLECTOR true || die "ping-collector must forecast first: run --cutover-forecast"
if env_is WEB_LISTEN '..*'; then
log "ping-collector already serves the website; restarting it on the new binary"
retire_matrix_cron
systemctl restart ping-collector.service
show 10
exit 0
fi
api_ok=1
if [ "$(systemctl is-active water-monitor.service 2>/dev/null)" = active ]; then
log "API check: ping-collector serve on :${CHECK_PORT} next to the Python app, every route compared"
api_check || api_ok=0
else
warn "the Python app is not running: nothing to compare the API with"
api_ok=0
fi
if [ "$api_ok" != 1 ] && [ "${FORCE:-0}" != 1 ]; then
die "the API check did not pass; not switching (FORCE=1 overrides)"
fi
log "Cutover: ping-collector serves the website and API on ${WEB_ADDR}; the Python app stops"
set_env WEB_LISTEN "$WEB_ADDR"
retire_matrix_cron
systemctl disable --now water-monitor.service
systemctl restart ping-collector.service
if wait_http "http://127.0.0.1:${WEB_ADDR##*:}/health" 60; then
log "The website answers from ping-collector: $(curl -s --max-time 5 "http://127.0.0.1:${WEB_ADDR##*:}/scraping/status" | grep -o '"collector":"[^"]*"')"
else
warn "no answer on port ${WEB_ADDR##*:} yet; check journalctl -u ping-collector"
fi
show 5
log "Rollback if needed: sudo bash scripts/install_collector.sh --rollback-api"
;;
*)
retire_matrix_cron
if [ "$(systemctl is-active ping-collector.service 2>/dev/null)" = active ]; then
systemctl restart ping-collector.service
log "Installed; ping-collector restarted on the new binary"
else
log "Installed, not started. Switch over with: sudo bash scripts/install_collector.sh --cutover"
fi
;;
esac
scripts/install_ntfy.sh
+1 -1
View File
@@ -92,7 +92,7 @@ if [[ -f "$env_file" ]] && ! grep -q '^NTFY_SERVER=' "$env_file"; then
echo "NTFY_TOPIC_PREFIX=ping"
echo "NTFY_TOKEN=${token}"
} >> "$env_file"
echo "wrote NTFY_* to ${env_file}; restart water-monitor to enable"
echo "wrote NTFY_* to ${env_file}; restart ping-collector to enable"
else
echo "NTFY_TOKEN=${token}"
fi
scripts/load_test.py
+2 -4
View File
@@ -9,7 +9,7 @@ and errors per stage plus the slowest endpoints.
Run against a LOCAL instance for full stress (never full-stress production —
it hosts live flood monitoring):
python -m uvicorn src.web_api:app --port 8125 # separate shell
rust/target/release/ping-collector serve --listen 127.0.0.1:8125 --no-analytics # separate shell
python scripts/load_test.py http://localhost:8125
A gentle production baseline (low, fixed concurrency):
@@ -98,9 +98,7 @@ def run_stage(base, clients, seconds):
by_endpoint = {}
for path, elapsed in results:
by_endpoint.setdefault(path, []).append(elapsed)
slowest = sorted(
by_endpoint.items(), key=lambda kv: -statistics.median(kv[1])
)[:4]
slowest = sorted(by_endpoint.items(), key=lambda kv: -statistics.median(kv[1]))[:4]
for path, values in slowest:
print(
f" slow: {path:45} n={len(values):5} "
scripts/locustfile.py
+7 -4
View File
@@ -7,7 +7,8 @@ the page makes, polling like the auto-refresh does) and API consumers
Full stress against a LOCAL instance (never full-stress production — it hosts
live flood monitoring):
# separate shell: python -m uvicorn src.web_api:app --port 8125
# separate shell:
# rust/target/release/ping-collector serve --listen 127.0.0.1:8125 --no-analytics
.venv/Scripts/python.exe -m locust -f scripts/locustfile.py \
--host http://localhost:8125 --headless \
--users 200 --spawn-rate 20 --run-time 2m \
@@ -77,9 +78,11 @@ class ApiConsumer(FastHttpUser):
@task(3)
def hii_feeds(self):
self.client.get(random.choice(
["/api/hii/waterlevel/latest", "/api/hii/rainfall/latest"]
), headers=GZIP, name="/api/hii/[feed]/latest")
self.client.get(
random.choice(["/api/hii/waterlevel/latest", "/api/hii/rainfall/latest"]),
headers=GZIP,
name="/api/hii/[feed]/latest",
)
@task(2)
def forecast(self):
scripts/migrate_sqlite_to_postgres.py
-619
View File
@@ -1,619 +0,0 @@
#!/usr/bin/env python3
"""
SQLite to PostgreSQL Migration Tool
Migrates all data from SQLite database to PostgreSQL
"""
import os
import sys
import logging
import sqlite3
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass
# Add src to path for imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
@dataclass
class MigrationStats:
stations_migrated: int = 0
measurements_migrated: int = 0
errors: List[str] = None
start_time: Optional[datetime] = None
end_time: Optional[datetime] = None
def __post_init__(self):
if self.errors is None:
self.errors = []
class SQLiteToPostgresMigrator:
def __init__(self, sqlite_path: str, postgres_config: Dict[str, Any]):
self.sqlite_path = sqlite_path
self.postgres_config = postgres_config
self.sqlite_conn = None
self.postgres_adapter = None
self.stats = MigrationStats()
# Setup logging with UTF-8 encoding
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler('migration.log', encoding='utf-8')
]
)
self.logger = logging.getLogger(__name__)
def connect_databases(self) -> bool:
"""Connect to both SQLite and PostgreSQL databases"""
try:
# Connect to SQLite
if not os.path.exists(self.sqlite_path):
self.logger.error(f"SQLite database not found: {self.sqlite_path}")
return False
self.sqlite_conn = sqlite3.connect(self.sqlite_path)
self.sqlite_conn.row_factory = sqlite3.Row # For dict-like access
self.logger.info(f"Connected to SQLite database: {self.sqlite_path}")
# Connect to PostgreSQL
from database_adapters import create_database_adapter
self.postgres_adapter = create_database_adapter(
self.postgres_config['type'],
connection_string=self.postgres_config['connection_string']
)
if not self.postgres_adapter.connect():
self.logger.error("Failed to connect to PostgreSQL")
return False
self.logger.info("Connected to PostgreSQL database")
return True
except Exception as e:
self.logger.error(f"Database connection error: {e}")
return False
def analyze_sqlite_schema(self) -> Dict[str, List[str]]:
"""Analyze SQLite database structure"""
try:
cursor = self.sqlite_conn.cursor()
# Get all tables
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")
tables = [row[0] for row in cursor.fetchall()]
schema_info = {}
for table in tables:
cursor.execute(f"PRAGMA table_info({table})")
columns = [row[1] for row in cursor.fetchall()]
schema_info[table] = columns
# Get row count
cursor.execute(f"SELECT COUNT(*) FROM {table}")
count = cursor.fetchone()[0]
self.logger.info(f"Table '{table}': {len(columns)} columns, {count} rows")
return schema_info
except Exception as e:
self.logger.error(f"Schema analysis error: {e}")
return {}
def migrate_stations(self) -> bool:
"""Migrate station data"""
try:
cursor = self.sqlite_conn.cursor()
# Try different possible table names and structures
station_queries = [
# Modern structure
"""SELECT id, station_code, station_name_th as thai_name, station_name_en as english_name,
latitude, longitude, geohash, created_at, updated_at
FROM stations""",
# Alternative structure 1
"""SELECT id, station_code, thai_name, english_name,
latitude, longitude, geohash, created_at, updated_at
FROM stations""",
# Legacy structure
"""SELECT station_id as id, station_code, station_name as thai_name,
station_name as english_name, lat as latitude, lon as longitude,
NULL as geohash, datetime('now') as created_at, datetime('now') as updated_at
FROM water_stations""",
# Simple structure
"""SELECT rowid as id, station_code, name as thai_name, name as english_name,
NULL as latitude, NULL as longitude, NULL as geohash,
datetime('now') as created_at, datetime('now') as updated_at
FROM stations""",
]
stations_data = []
for query in station_queries:
try:
cursor.execute(query)
rows = cursor.fetchall()
if rows:
self.logger.info(f"Found {len(rows)} stations using query variant")
for row in rows:
station = {
'station_id': row[0],
'station_code': row[1] or f"STATION_{row[0]}",
'station_name_th': row[2] or f"Station {row[0]}",
'station_name_en': row[3] or f"Station {row[0]}",
'latitude': row[4],
'longitude': row[5],
'geohash': row[6],
'status': 'active'
}
stations_data.append(station)
break
except sqlite3.OperationalError as e:
if "no such table" in str(e).lower() or "no such column" in str(e).lower():
continue
else:
raise
if not stations_data:
self.logger.warning("No stations found in SQLite database")
return True
# Insert stations into PostgreSQL using raw SQL
# Since the adapter is designed for measurements, we'll use direct SQL
try:
from sqlalchemy import create_engine, text
engine = create_engine(self.postgres_config['connection_string'])
# Process stations individually to avoid transaction rollback issues
for station in stations_data:
try:
with engine.begin() as conn:
# Use PostgreSQL UPSERT syntax with correct column names
station_sql = """
INSERT INTO stations (id, station_code, thai_name, english_name, latitude, longitude, geohash)
VALUES (:station_id, :station_code, :thai_name, :english_name, :latitude, :longitude, :geohash)
ON CONFLICT (id) DO UPDATE SET
thai_name = EXCLUDED.thai_name,
english_name = EXCLUDED.english_name,
latitude = EXCLUDED.latitude,
longitude = EXCLUDED.longitude,
geohash = EXCLUDED.geohash,
updated_at = CURRENT_TIMESTAMP
"""
conn.execute(text(station_sql), {
'station_id': station['station_id'],
'station_code': station['station_code'],
'thai_name': station['station_name_th'],
'english_name': station['station_name_en'],
'latitude': station.get('latitude'),
'longitude': station.get('longitude'),
'geohash': station.get('geohash')
})
self.stats.stations_migrated += 1
except Exception as e:
error_msg = f"Error migrating station {station.get('station_code', 'unknown')}: {str(e)[:100]}..."
self.logger.warning(error_msg)
self.stats.errors.append(error_msg)
self.logger.info(f"Migrated {self.stats.stations_migrated} stations")
except Exception as e:
self.logger.error(f"Station migration failed: {e}")
return False
self.logger.info(f"Migrated {self.stats.stations_migrated} stations")
return True
except Exception as e:
self.logger.error(f"Station migration error: {e}")
return False
def migrate_measurements(self, batch_size: int = 5000) -> bool:
"""Migrate measurement data in batches"""
try:
cursor = self.sqlite_conn.cursor()
# Try different possible measurement table structures
measurement_queries = [
# Modern structure
"""SELECT w.timestamp, w.station_id, s.station_code, s.station_name_th, s.station_name_en,
w.water_level, w.discharge, w.discharge_percent, w.status
FROM water_measurements w
JOIN stations s ON w.station_id = s.id
ORDER BY w.timestamp""",
# Alternative with different join
"""SELECT w.timestamp, w.station_id, s.station_code, s.thai_name, s.english_name,
w.water_level, w.discharge, w.discharge_percent, 'active' as status
FROM water_measurements w
JOIN stations s ON w.station_id = s.id
ORDER BY w.timestamp""",
# Legacy structure
"""SELECT timestamp, station_id, station_code, station_name, station_name,
water_level, discharge, discharge_percent, 'active' as status
FROM measurements
ORDER BY timestamp""",
# Simple structure without joins
"""SELECT timestamp, station_id, 'UNKNOWN' as station_code, 'Unknown' as station_name_th, 'Unknown' as station_name_en,
water_level, discharge, discharge_percent, 'active' as status
FROM water_measurements
ORDER BY timestamp""",
]
measurements_processed = 0
for query in measurement_queries:
try:
# Get total count first
count_query = query.replace("SELECT", "SELECT COUNT(*) FROM (SELECT").replace("ORDER BY w.timestamp", "") + ")"
cursor.execute(count_query)
total_measurements = cursor.fetchone()[0]
if total_measurements == 0:
continue
self.logger.info(f"Found {total_measurements} measurements to migrate")
# Process in batches
offset = 0
while True:
batch_query = f"{query} LIMIT {batch_size} OFFSET {offset}"
cursor.execute(batch_query)
rows = cursor.fetchall()
if not rows:
break
# Convert to measurement format
measurements = []
for row in rows:
try:
# Parse timestamp
timestamp_str = row[0]
if isinstance(timestamp_str, str):
try:
timestamp = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00'))
except:
# Try other common formats
for fmt in ['%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M:%S.%f', '%Y-%m-%dT%H:%M:%S']:
try:
timestamp = datetime.strptime(timestamp_str, fmt)
break
except:
continue
else:
timestamp = datetime.now()
else:
timestamp = timestamp_str
measurement = {
'timestamp': timestamp,
'station_id': row[1] or 999,
'station_code': row[2] or 'UNKNOWN',
'station_name_th': row[3] or 'Unknown',
'station_name_en': row[4] or 'Unknown',
'water_level': float(row[5]) if row[5] is not None else None,
'discharge': float(row[6]) if row[6] is not None else None,
'discharge_percent': float(row[7]) if row[7] is not None else None,
'status': row[8] or 'active'
}
measurements.append(measurement)
except Exception as e:
error_msg = f"Error processing measurement row: {e}"
self.logger.warning(error_msg)
continue
# Save batch to PostgreSQL using fast bulk insert
if measurements:
try:
self._fast_bulk_insert(measurements)
measurements_processed += len(measurements)
self.stats.measurements_migrated += len(measurements)
self.logger.info(f"Migrated {measurements_processed}/{total_measurements} measurements")
except Exception as e:
error_msg = f"Error saving measurement batch: {e}"
self.logger.error(error_msg)
self.stats.errors.append(error_msg)
offset += batch_size
# If we processed measurements, we're done
if measurements_processed > 0:
break
except sqlite3.OperationalError as e:
if "no such table" in str(e).lower() or "no such column" in str(e).lower():
continue
else:
raise
if measurements_processed == 0:
self.logger.warning("No measurements found in SQLite database")
else:
self.logger.info(f"Successfully migrated {measurements_processed} measurements")
return True
except Exception as e:
self.logger.error(f"Measurement migration error: {e}")
return False
def _fast_bulk_insert(self, measurements: List[Dict]) -> bool:
"""Super fast bulk insert using PostgreSQL COPY or VALUES clause"""
try:
import psycopg2
from urllib.parse import urlparse
import io
# Parse connection string for direct psycopg2 connection
parsed = urlparse(self.postgres_config['connection_string'])
# Try super fast COPY method first
try:
conn = psycopg2.connect(
host=parsed.hostname,
port=parsed.port or 5432,
database=parsed.path[1:],
user=parsed.username,
password=parsed.password
)
with conn:
with conn.cursor() as cur:
# Prepare data for COPY
data_buffer = io.StringIO()
null_val = '\\N'
for m in measurements:
data_buffer.write(f"{m['timestamp']}\t{m['station_id']}\t{m['water_level'] or null_val}\t{m['discharge'] or null_val}\t{m['discharge_percent'] or null_val}\t{m['status']}\n")
data_buffer.seek(0)
# Use COPY for maximum speed
cur.copy_from(
data_buffer,
'water_measurements',
columns=('timestamp', 'station_id', 'water_level', 'discharge', 'discharge_percent', 'status'),
sep='\t'
)
conn.close()
return True
except Exception as copy_error:
# Fallback to SQLAlchemy bulk insert
self.logger.debug(f"COPY failed, using bulk VALUES: {copy_error}")
from sqlalchemy import create_engine, text
engine = create_engine(self.postgres_config['connection_string'])
with engine.begin() as conn:
# Use PostgreSQL's fast bulk insert with ON CONFLICT
values_list = []
for m in measurements:
timestamp = m['timestamp'].isoformat() if hasattr(m['timestamp'], 'isoformat') else str(m['timestamp'])
values_list.append(
f"('{timestamp}', {m['station_id']}, {m['water_level'] or 'NULL'}, "
f"{m['discharge'] or 'NULL'}, {m['discharge_percent'] or 'NULL'}, '{m['status']}')"
)
# Build bulk insert query with ON CONFLICT handling
bulk_sql = f"""
INSERT INTO water_measurements (timestamp, station_id, water_level, discharge, discharge_percent, status)
VALUES {','.join(values_list)}
ON CONFLICT (timestamp, station_id) DO UPDATE SET
water_level = EXCLUDED.water_level,
discharge = EXCLUDED.discharge,
discharge_percent = EXCLUDED.discharge_percent,
status = EXCLUDED.status
"""
conn.execute(text(bulk_sql))
return True
except Exception as e:
self.logger.warning(f"Fast bulk insert failed: {e}")
# Final fallback to original method
try:
success = self.postgres_adapter.save_measurements(measurements)
return success
except Exception as fallback_e:
self.logger.error(f"All insert methods failed: {fallback_e}")
return False
def verify_migration(self) -> bool:
"""Verify the migration by comparing counts"""
try:
# Get SQLite counts
cursor = self.sqlite_conn.cursor()
sqlite_stations = 0
sqlite_measurements = 0
# Try to get station count
for table in ['stations', 'water_stations']:
try:
cursor.execute(f"SELECT COUNT(*) FROM {table}")
sqlite_stations = cursor.fetchone()[0]
break
except:
continue
# Try to get measurement count
for table in ['water_measurements', 'measurements']:
try:
cursor.execute(f"SELECT COUNT(*) FROM {table}")
sqlite_measurements = cursor.fetchone()[0]
break
except:
continue
# Get PostgreSQL counts
postgres_measurements = self.postgres_adapter.get_latest_measurements(limit=999999)
postgres_count = len(postgres_measurements)
self.logger.info("Migration Verification:")
self.logger.info(f"SQLite stations: {sqlite_stations}")
self.logger.info(f"SQLite measurements: {sqlite_measurements}")
self.logger.info(f"PostgreSQL measurements retrieved: {postgres_count}")
self.logger.info(f"Migrated stations: {self.stats.stations_migrated}")
self.logger.info(f"Migrated measurements: {self.stats.measurements_migrated}")
return True
except Exception as e:
self.logger.error(f"Verification error: {e}")
return False
def run_migration(self, sqlite_path: str = None) -> bool:
"""Run the complete migration process"""
self.stats.start_time = datetime.now()
if sqlite_path:
self.sqlite_path = sqlite_path
self.logger.info("=" * 60)
self.logger.info("SQLite to PostgreSQL Migration Tool")
self.logger.info("=" * 60)
self.logger.info(f"SQLite database: {self.sqlite_path}")
self.logger.info(f"PostgreSQL: {self.postgres_config['type']}")
try:
# Step 1: Connect to databases
self.logger.info("Step 1: Connecting to databases...")
if not self.connect_databases():
return False
# Step 2: Analyze SQLite schema
self.logger.info("Step 2: Analyzing SQLite database structure...")
schema_info = self.analyze_sqlite_schema()
if not schema_info:
self.logger.error("Could not analyze SQLite database structure")
return False
# Step 3: Migrate stations
self.logger.info("Step 3: Migrating station data...")
if not self.migrate_stations():
self.logger.error("Station migration failed")
return False
# Step 4: Migrate measurements
self.logger.info("Step 4: Migrating measurement data...")
if not self.migrate_measurements():
self.logger.error("Measurement migration failed")
return False
# Step 5: Verify migration
self.logger.info("Step 5: Verifying migration...")
self.verify_migration()
self.stats.end_time = datetime.now()
duration = self.stats.end_time - self.stats.start_time
# Final report
self.logger.info("=" * 60)
self.logger.info("MIGRATION COMPLETED")
self.logger.info("=" * 60)
self.logger.info(f"Duration: {duration}")
self.logger.info(f"Stations migrated: {self.stats.stations_migrated}")
self.logger.info(f"Measurements migrated: {self.stats.measurements_migrated}")
if self.stats.errors:
self.logger.warning(f"Errors encountered: {len(self.stats.errors)}")
for error in self.stats.errors[:10]: # Show first 10 errors
self.logger.warning(f" - {error}")
if len(self.stats.errors) > 10:
self.logger.warning(f" ... and {len(self.stats.errors) - 10} more errors")
else:
self.logger.info("No errors encountered")
return True
except Exception as e:
self.logger.error(f"Migration failed: {e}")
return False
finally:
# Cleanup
if self.sqlite_conn:
self.sqlite_conn.close()
def main():
"""Main entry point"""
import argparse
parser = argparse.ArgumentParser(description="Migrate SQLite data to PostgreSQL")
parser.add_argument("sqlite_path", nargs="?", help="Path to SQLite database file")
parser.add_argument("--batch-size", type=int, default=5000, help="Batch size for processing measurements")
parser.add_argument("--fast", action="store_true", help="Use maximum speed mode (batch-size 10000)")
parser.add_argument("--dry-run", action="store_true", help="Analyze only, don't migrate")
args = parser.parse_args()
# Set fast mode
if args.fast:
args.batch_size = 10000
# Get SQLite path
sqlite_path = args.sqlite_path
if not sqlite_path:
# Try to find common SQLite database files
possible_paths = [
"water_levels.db",
"water_monitoring.db",
"database.db",
"../water_levels.db"
]
for path in possible_paths:
if os.path.exists(path):
sqlite_path = path
break
if not sqlite_path:
print("SQLite database file not found. Please specify the path:")
print(" python migrate_sqlite_to_postgres.py /path/to/database.db")
return False
# Get PostgreSQL configuration
try:
from config import Config
postgres_config = Config.get_database_config()
if postgres_config['type'] != 'postgresql':
print("Error: PostgreSQL not configured. Set DB_TYPE=postgresql in your .env file")
return False
except Exception as e:
print(f"Error loading PostgreSQL configuration: {e}")
return False
# Run migration
migrator = SQLiteToPostgresMigrator(sqlite_path, postgres_config)
if args.dry_run:
print("DRY RUN MODE - Analyzing SQLite database structure only")
if migrator.connect_databases():
schema_info = migrator.analyze_sqlite_schema()
print("\nSQLite database structure analysis complete.")
print("Run without --dry-run to perform the actual migration.")
return True
success = migrator.run_migration()
return success
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)
scripts/ping-collector.service
+3 -4
View File
@@ -14,10 +14,9 @@ Restart=always
RestartSec=30
TimeoutStopSec=20
# Same .env as water-monitor.service (POSTGRES_*, RID_*, GAP_FILL_*,
# SCRAPING_STATUS_FILE, FORECAST_IN_COLLECTOR, NTFY_*). The Python app must run
# with COLLECTION_ENABLED=false. Models are read from models/ (relative to
# WorkingDirectory).
# Every key with its default: .env.example (POSTGRES_*, WEB_LISTEN,
# FORECAST_IN_COLLECTOR, NTFY_*, MATRIX_*, ...). Models are read from models/
# (relative to WorkingDirectory); installed by scripts/install.sh.
EnvironmentFile=/opt/thailand-water-monitor/.env
# Security settings
scripts/retrain.sh
+7 -7
View File
@@ -5,15 +5,15 @@
#
# Why a script rather than ExecStart=train_flood_model.py:
# * train.py writes each station's bundle straight into models/ over ~12 min,
# and the API's hourly precompute reloads bundles by mtime. Training into
# a staging dir and mv-ing (atomic on one filesystem) means the API never
# sees a half-written joblib file or a mixed old/new set.
# and ping-collector reloads bundles by mtime on every forecast run.
# Training into a staging dir and mv-ing (atomic on one filesystem) means
# it never sees a half-written bundle or a mixed old/new set.
# * A run that produced gauge-only (v2) bundles, or trained too few stations,
# must NOT replace the deployed models. train.py already aborts on a
# missing rain series; this script re-checks the written metrics anyway.
# * No restart is needed: predict.py (Python) and ping-collector (Rust,
# FORECAST_IN_COLLECTOR=true) reload changed bundles on their next
# forecast run, so the new models are live within an hour.
# * No restart is needed: ping-collector reloads changed bundles
# (models/flood_*.json.gz, written by train.py next to each .joblib) on
# its next forecast run, so the new models are live within an hour.
#
# Exit codes: 0 ok, 2 training refused (see log), 3 verification failed.
set -euo pipefail
@@ -88,4 +88,4 @@ for f in "${STAGE_DIR}"/flood_*.joblib "${STAGE_DIR}"/flood_*.json.gz "${STAGE_D
mv -f "${f}" "${MODELS_DIR}/${name}"
done
rm -rf "${STAGE_DIR}"
log "deployed ${VERSION} (${TRAINED} stations); previous generation in models/.previous. The API picks it up on its next hourly precompute."
log "deployed ${VERSION} (${TRAINED} stations); previous generation in models/.previous. ping-collector loads it on its next forecast run."
scripts/setup_postgres.py
-175
View File
@@ -1,175 +0,0 @@
#!/usr/bin/env python3
"""
PostgreSQL setup script for Northern Thailand Ping River Monitor
This script helps you configure and test your PostgreSQL connection
"""
import os
import sys
import logging
from typing import Optional
from urllib.parse import urlparse
def setup_logging():
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
def test_postgres_connection(connection_string: str) -> bool:
"""Test connection to PostgreSQL database"""
try:
from sqlalchemy import create_engine, text
# Test connection
engine = create_engine(connection_string, pool_pre_ping=True)
with engine.connect() as conn:
result = conn.execute(text("SELECT version()"))
version = result.fetchone()[0]
logging.info(f"✅ Connected to PostgreSQL successfully!")
logging.info(f"Database version: {version}")
return True
except ImportError:
logging.error("❌ psycopg2-binary not installed. Run: uv add psycopg2-binary")
return False
except Exception as e:
logging.error(f"❌ Connection failed: {e}")
return False
def parse_connection_string(connection_string: str) -> dict:
"""Parse PostgreSQL connection string into components"""
try:
parsed = urlparse(connection_string)
return {
'host': parsed.hostname,
'port': parsed.port or 5432,
'database': parsed.path[1:] if parsed.path else None,
'username': parsed.username,
'password': parsed.password,
}
except Exception as e:
logging.error(f"Failed to parse connection string: {e}")
return {}
def create_database_if_not_exists(connection_string: str, database_name: str) -> bool:
"""Create database if it doesn't exist"""
try:
from sqlalchemy import create_engine, text
# Connect to default postgres database to create our database
parsed = urlparse(connection_string)
admin_connection = connection_string.replace(f"/{parsed.path[1:]}", "/postgres")
engine = create_engine(admin_connection, pool_pre_ping=True)
with engine.connect() as conn:
# Check if database exists
result = conn.execute(text(
"SELECT 1 FROM pg_database WHERE datname = :db_name"
), {"db_name": database_name})
if result.fetchone():
logging.info(f"✅ Database '{database_name}' already exists")
return True
else:
# Create database
conn.execute(text("COMMIT")) # End transaction
conn.execute(text(f'CREATE DATABASE "{database_name}"'))
logging.info(f"✅ Created database '{database_name}'")
return True
except Exception as e:
logging.error(f"❌ Failed to create database: {e}")
return False
def initialize_tables(connection_string: str) -> bool:
"""Initialize database tables"""
try:
# Import the database adapter to create tables
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
from database_adapters import SQLAdapter
adapter = SQLAdapter(connection_string=connection_string, db_type='postgresql')
if adapter.connect():
logging.info("✅ Database tables initialized successfully")
return True
else:
logging.error("❌ Failed to initialize tables")
return False
except Exception as e:
logging.error(f"❌ Failed to initialize tables: {e}")
return False
def interactive_setup():
"""Interactive setup wizard"""
print("🐘 PostgreSQL Setup Wizard for Ping River Monitor")
print("=" * 50)
# Get connection details
host = input("PostgreSQL host (e.g., 192.168.1.100): ").strip()
port = input("PostgreSQL port [5432]: ").strip() or "5432"
database = input("Database name [water_monitoring]: ").strip() or "water_monitoring"
username = input("Username: ").strip()
password = input("Password: ").strip()
# Optional SSL
use_ssl = input("Use SSL connection? (y/N): ").strip().lower() == 'y'
ssl_params = "?sslmode=require" if use_ssl else ""
connection_string = f"postgresql://{username}:{password}@{host}:{port}/{database}{ssl_params}"
print(f"\nGenerated connection string:")
print(f"POSTGRES_CONNECTION_STRING={connection_string}")
return connection_string
def main():
setup_logging()
print("🚀 Northern Thailand Ping River Monitor - PostgreSQL Setup")
print("=" * 60)
# Check if connection string is provided via environment
connection_string = os.getenv('POSTGRES_CONNECTION_STRING')
if not connection_string:
print("No POSTGRES_CONNECTION_STRING found in environment.")
print("Starting interactive setup...\n")
connection_string = interactive_setup()
# Suggest adding to .env file
print(f"\n💡 Add this to your .env file:")
print(f"DB_TYPE=postgresql")
print(f"POSTGRES_CONNECTION_STRING={connection_string}")
# Parse connection details
config = parse_connection_string(connection_string)
if not config.get('host'):
logging.error("Invalid connection string format")
return False
print(f"\n🔗 Connecting to PostgreSQL at {config['host']}:{config['port']}")
# Test connection
if not test_postgres_connection(connection_string):
return False
# Try to create database
database_name = config.get('database', 'water_monitoring')
if database_name:
create_database_if_not_exists(connection_string, database_name)
# Initialize tables
if not initialize_tables(connection_string):
return False
print("\n🎉 PostgreSQL setup completed successfully!")
print("\nNext steps:")
print("1. Update your .env file with the connection string")
print("2. Run: make run-test")
print("3. Run: make run-api")
return True
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)
scripts/setup_uv.bat
-48
View File
@@ -1,48 +0,0 @@
@echo off
REM Setup script for uv-based development environment on Windows
echo 🚀 Setting up Northern Thailand Ping River Monitor with uv...
REM Check if uv is installed
uv --version >nul 2>&1
if %errorlevel% neq 0 (
echo ❌ uv is not installed. Please install it first:
echo powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
exit /b 1
)
echo ✅ uv found
uv --version
REM Initialize uv project if not already initialized
if not exist "uv.lock" (
echo 🔧 Initializing uv project...
uv sync
) else (
echo 📦 Syncing dependencies with uv...
uv sync
)
REM Install pre-commit hooks
echo 🎣 Installing pre-commit hooks...
uv run pre-commit install
REM Create .env file if it doesn't exist
if not exist ".env" (
if exist ".env.example" (
echo 📝 Creating .env file from template...
copy .env.example .env
echo ⚠️ Please edit .env file with your configuration
)
)
echo ✅ Setup complete!
echo.
echo 📚 Quick start commands:
echo make install-dev # Install all dependencies
echo make run-test # Run a test cycle
echo make run-api # Start the web API
echo make test # Run tests
echo make lint # Check code quality
echo.
echo 🎉 Happy monitoring!
scripts/setup_uv.sh
-46
View File
@@ -1,46 +0,0 @@
#!/bin/bash
# Setup script for uv-based development environment
set -e
echo "🚀 Setting up Northern Thailand Ping River Monitor with uv..."
# Check if uv is installed
if ! command -v uv &> /dev/null; then
echo "❌ uv is not installed. Please install it first:"
echo " curl -LsSf https://astral.sh/uv/install.sh | sh"
exit 1
fi
echo "✅ uv found: $(uv --version)"
# Initialize uv project if not already initialized
if [ ! -f "uv.lock" ]; then
echo "🔧 Initializing uv project..."
uv sync
else
echo "📦 Syncing dependencies with uv..."
uv sync
fi
# Install pre-commit hooks
echo "🎣 Installing pre-commit hooks..."
uv run pre-commit install
# Create .env file if it doesn't exist
if [ ! -f ".env" ] && [ -f ".env.example" ]; then
echo "📝 Creating .env file from template..."
cp .env.example .env
echo "⚠️ Please edit .env file with your configuration"
fi
echo "✅ Setup complete!"
echo ""
echo "📚 Quick start commands:"
echo " make install-dev # Install all dependencies"
echo " make run-test # Run a test cycle"
echo " make run-api # Start the web API"
echo " make test # Run tests"
echo " make lint # Check code quality"
echo ""
echo "🎉 Happy monitoring!"
scripts/summarize_eval.py
+18 -5
View File
@@ -25,7 +25,8 @@ def summarize(paths):
for fold in station["folds"]:
for name, m in fold["variants"].items():
a = agg.setdefault(
name, {"mae": [], "mae_hi": [], "brier": [], "fa": 0, "events": []}
name,
{"mae": [], "mae_hi": [], "brier": [], "fa": 0, "events": []},
)
a["mae"].append(m["mae"])
if m.get("mae_above_2p5") is not None:
@@ -39,8 +40,18 @@ def summarize(paths):
if e["peak_pred_24h_before"] is None
else e["peak_pred_24h_before"] - e["peak_level"]
)
a["events"].append((fold["year"], e["crossing"][:10], e["lead_h"], e["peak_level"], err))
print(f"{'variant':22} {'MAE':>6} {'MAE_hi':>7} {'Brier':>7} {'FA':>3} events: year crossing lead_h peak(err24h)")
a["events"].append(
(
fold["year"],
e["crossing"][:10],
e["lead_h"],
e["peak_level"],
err,
)
)
print(
f"{'variant':22} {'MAE':>6} {'MAE_hi':>7} {'Brier':>7} {'FA':>3} events: year crossing lead_h peak(err24h)"
)
for name, a in agg.items():
ev = " ".join(
f"{y} {d} {'—' if l is None else format(l, '+.0f')}h {p:.2f}({'—' if err is None else format(err, '+.2f')})"
@@ -54,8 +65,10 @@ def summarize(paths):
f"{a['fa']:>3} {ev}"
)
if leads:
print(f"{'':22} lead: mean {statistics.mean(leads):+.1f} h, min {min(leads):+.0f} h, "
f"missed {sum(1 for *_, l, _, _ in a['events'] if l is None)}/{len(a['events'])}")
print(
f"{'':22} lead: mean {statistics.mean(leads):+.1f} h, min {min(leads):+.0f} h, "
f"missed {sum(1 for *_, l, _, _ in a['events'] if l is None)}/{len(a['events'])}"
)
if __name__ == "__main__":
scripts/water-monitor-retrain.service
+3 -3
View File
@@ -10,11 +10,11 @@ User=water-monitor
Group=water-monitor
WorkingDirectory=/opt/thailand-water-monitor
EnvironmentFile=/opt/thailand-water-monitor/.env
# Same interpreter as water-monitor.service -- the uv-managed .venv.
# The uv-managed .venv (Python training stack; scripts/install.sh syncs it).
# scripts/retrain.sh trains into models/.staging, refuses to promote anything
# that is not a rain-enabled (hgb-v3) set covering the expected stations, then
# renames the bundles into place. The API reloads them on its next hourly
# precompute; no restart, so a failed run leaves the old models serving.
# renames the bundles into place. ping-collector reloads them on its next
# forecast run; no restart, so a failed run leaves the old models serving.
ExecStart=/bin/bash /opt/thailand-water-monitor/scripts/retrain.sh
# HistGradientBoosting is CPU-bound; cap threads so training cannot starve
# the API (docs/FLOOD_FORECASTING.md section 6 measured 4 as the sweet spot).
scripts/water-monitor.service
-41
View File
@@ -1,41 +0,0 @@
[Unit]
Description=Thailand Water Level Monitor
Documentation=https://git.b4l.co.th/B4L/Northern-Thailand-Ping-River-Monitor
After=network.target
Wants=network-online.target
[Service]
Type=simple
User=water-monitor
Group=water-monitor
WorkingDirectory=/opt/thailand-water-monitor
# The uv-managed env (uv sync -> .venv). Same interpreter for water-monitor-retrain.service.
ExecStart=/opt/thailand-water-monitor/.venv/bin/python run.py --web-api
ExecReload=/bin/kill -HUP $MAINPID
Restart=always
RestartSec=60
TimeoutStopSec=30
# DB_TYPE / POSTGRES_CONNECTION_STRING / MATRIX_* come from the .env file.
EnvironmentFile=/opt/thailand-water-monitor/.env
Environment=PYTHONPATH=/opt/thailand-water-monitor
# Serving path is latency-bound; single-threaded BLAS is 2.6x faster per call
# (docs/FLOOD_FORECASTING.md section 6). Training sets its own value.
Environment=OMP_NUM_THREADS=1
Environment=PYTHONUNBUFFERED=1
# Security settings
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/opt/thailand-water-monitor
CapabilityBoundingSet=
# Logging
StandardOutput=journal
StandardError=journal
SyslogIdentifier=water-monitor
[Install]
WantedBy=multi-user.target