From e4aa474393c465adcf8a7e274099a62fe7f2bf22 Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 24 Sep 2026 17:26:36 +0200 Subject: [PATCH 01/14] feat: add script to fetch, publish and replay real alerts --- .gitignore | 1 + scripts/replay_alerts.py | 312 ++++++++++++++++++++++++++++++++++++ tests/requirements.txt | 1 + tests/test_replay_alerts.py | 75 +++++++++ 4 files changed, 389 insertions(+) create mode 100644 scripts/replay_alerts.py create mode 100644 tests/test_replay_alerts.py diff --git a/.gitignore b/.gitignore index 50c0312..698edc0 100644 --- a/.gitignore +++ b/.gitignore @@ -155,3 +155,4 @@ containers/notebooks/app/.Trash-0 data/alert_samples/ data/triangulated_sequences/ +data/replay_alerts/ diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py new file mode 100644 index 0000000..7dff167 --- /dev/null +++ b/scripts/replay_alerts.py @@ -0,0 +1,312 @@ +#!/usr/bin/env python3 +"""Fetch real alerts from the production alert API, share them through a GitHub +release, and replay them on the local stack. + + fetch ALERT_ID... download alerts to data/replay_alerts/alert_.zip + (needs the admin DISTANT_* credentials in .env) + publish ALERT_ID... upload the zips to the `replay-alerts` GitHub release (needs gh) + list list the alerts available in the release + replay ALERT_ID... replay alerts on the local API, one after the other + --mode live one frame per camera every --interval seconds, timestamped now + --mode demo post everything at once, timestamped today at the original + time of day (--date today) or as in prod (--date original) + +Examples: + python3 scripts/replay_alerts.py fetch 54095 + python3 scripts/replay_alerts.py publish 54095 + python3 scripts/replay_alerts.py replay 54095 --mode demo +""" + +import argparse +import ast +import json +import os +import subprocess +import sys +import time +import zipfile +from datetime import date, datetime, timedelta, timezone +from itertools import zip_longest +from pathlib import Path + +import requests +from dotenv import load_dotenv + +REPO_ROOT = Path(__file__).resolve().parent.parent +ALERTS_DIR = REPO_ROOT / "data" / "replay_alerts" +GITHUB_REPO = "pyronear/pyro-envdev" +RELEASE_TAG = "replay-alerts" +CAMERA_FIELDS = ("name", "angle_of_view", "elevation", "lat", "lon") + + +def login(api_url, username, password): + r = requests.post( + f"{api_url}/api/v1/login/creds", + data={"username": username, "password": password}, + timeout=30, + ) + r.raise_for_status() + return {"Authorization": f"Bearer {r.json()['access_token']}"} + + +def get(api_url, path, headers, **params): + r = requests.get(f"{api_url}{path}", headers=headers, params=params, timeout=60) + r.raise_for_status() + return r.json() + + +def zip_path(alert_id): + return ALERTS_DIR / f"alert_{alert_id}.zip" + + +# ------------------------------------------------------------------- +# fetch / publish / list +# ------------------------------------------------------------------- + + +def fetch(alert_ids): + api_url = os.environ["DISTANT_API_URL"].rstrip("/") + headers = login( + api_url, + os.environ["DISTANT_ALERT_API_LOGIN"], + os.environ["DISTANT_ALERT_API_PASSWORD"], + ) + ALERTS_DIR.mkdir(parents=True, exist_ok=True) + for alert_id in alert_ids: + alert = get(api_url, f"/api/v1/alerts/{alert_id}", headers) + sequences = get( + api_url, f"/api/v1/alerts/{alert_id}/sequences", headers, limit=100 + ) + cameras = {} + with zipfile.ZipFile(zip_path(alert_id), "w") as zf: + for seq in sequences: + cam_id = seq["camera_id"] + if cam_id not in cameras: + cam = get(api_url, f"/api/v1/cameras/{cam_id}", headers) + cameras[cam_id] = {k: cam[k] for k in CAMERA_FIELDS} + seq["detections"] = [] + offset = 0 + while True: + page = get( + api_url, + f"/api/v1/sequences/{seq['id']}/detections", + headers, + limit=100, + offset=offset, + desc=False, + ) + for det in page: + image = f"images/{det['bucket_key']}" + if image not in zf.namelist(): + img = requests.get(det["url"], timeout=60) + img.raise_for_status() + zf.writestr(image, img.content) + seq["detections"].append( + { + "recorded_at": det["recorded_at"], + "bbox": det["bbox"], + "others_bboxes": det["others_bboxes"], + "image": image, + } + ) + if len(page) < 100: + break + offset += 100 + alert["sequences"] = sequences + alert["cameras"] = cameras + zf.writestr("alert.json", json.dumps(alert, indent=1)) + n_dets = sum(len(s["detections"]) for s in sequences) + print( + f"alert {alert_id}: {len(sequences)} sequences, {n_dets} detections, " + f"{len(cameras)} cameras -> {zip_path(alert_id)}" + ) + + +def publish(alert_ids): + files = [str(zip_path(a)) for a in alert_ids] + missing = [f for f in files if not Path(f).is_file()] + if missing: + sys.exit(f"missing {missing}, run `fetch` first") + gh = ["gh", "release", "-R", GITHUB_REPO] + if subprocess.run([*gh, "view", RELEASE_TAG], capture_output=True).returncode: + notes = "Real alerts for scripts/replay_alerts.py" + subprocess.run( + [*gh, "create", RELEASE_TAG, "--title", "Replay alerts", "--notes", notes], + check=True, + ) + subprocess.run([*gh, "upload", RELEASE_TAG, *files, "--clobber"], check=True) + + +def list_alerts(): + r = requests.get( + f"https://api.github.com/repos/{GITHUB_REPO}/releases/tags/{RELEASE_TAG}", + timeout=30, + ) + r.raise_for_status() + for asset in r.json()["assets"]: + print(asset["name"].removeprefix("alert_").removesuffix(".zip")) + + +# ------------------------------------------------------------------- +# replay +# ------------------------------------------------------------------- + + +def parse_bboxes(s): + """Parse a stored bbox string, dropping degenerate boxes the API rejects.""" + boxes = ast.literal_eval(s) if s else [] + return [ + tuple(b) + for b in boxes + if round(b[0], 3) < round(b[2], 3) and round(b[1], 3) < round(b[3], 3) + ] + + +def fmt_bboxes(boxes): + # The API accepts `0`, `1` or `0.xxx` only, never `1.0` or `0.0`. + def fmt(x): + return f"{round(float(x), 3):.3f}".rstrip("0").rstrip(".") or "0" + + return "[" + ",".join("(" + ",".join(map(fmt, b)) + ")" for b in boxes) + "]" + + +def build_rounds(alert): + """Group detections into frames (one per uploaded image) and interleave cameras. + + A prod upload with several boxes yields one detection per box plus continuity + rows, all sharing the image: they are replayed as a single upload. Round k holds + the k-th frame of every (camera, azimuth) stream, like parallel real cameras. + """ + frames = {} + for seq in alert["sequences"]: + stream = (seq["camera_id"], seq["camera_azimuth"]) + for det in seq["detections"]: + frame = frames.setdefault( + det["image"], + {"stream": stream, "image": det["image"], "boxes": []}, + ) + frame["recorded_at"] = datetime.fromisoformat(det["recorded_at"]) + for box in parse_bboxes(det["bbox"]) + parse_bboxes(det["others_bboxes"]): + if box not in frame["boxes"]: + frame["boxes"].append(box) + streams = {} + for frame in sorted(frames.values(), key=lambda f: f["recorded_at"]): + streams.setdefault(frame["stream"], []).append(frame) + return [[f for f in r if f] for r in zip_longest(*streams.values())] + + +def day_offset(rounds, when): + """Shift to apply to recorded_at: whole days, so all frames move together.""" + if when == "original": + return timedelta(0) + first = min(f["recorded_at"] for r in rounds for f in r) + return timedelta(days=(date.today() - first.date()).days) + + +def load_alert(alert_id): + path = zip_path(alert_id) + if not path.is_file(): + url = f"https://github.com/{GITHUB_REPO}/releases/download/{RELEASE_TAG}/{path.name}" + print(f"downloading {url}") + r = requests.get(url, timeout=300) + r.raise_for_status() + ALERTS_DIR.mkdir(parents=True, exist_ok=True) + path.write_bytes(r.content) + zf = zipfile.ZipFile(path) + return zf, json.loads(zf.read("alert.json")) + + +def replay_alert(alert_id, args, admin): + zf, alert = load_alert(alert_id) + rounds = build_rounds(alert) + offset = day_offset(rounds, args.date) + last = max(f["recorded_at"] for r in rounds for f in r) + offset + if args.mode == "demo" and last > datetime.now(timezone.utc).replace(tzinfo=None): + print(f"warning: alert {alert_id} ends at {last} UTC, later than now") + + api = args.api_url + local_cams = {c["name"]: c["id"] for c in get(api, "/api/v1/cameras/", admin)} + tokens, poses = {}, {} + for prod_id, cam in alert["cameras"].items(): + name = cam["name"] + if name not in local_cams: + payload = {**cam, "organization_id": args.org_id, "is_trustable": True} + r = requests.post( + f"{api}/api/v1/cameras/", headers=admin, json=payload, timeout=30 + ) + r.raise_for_status() + local_cams[name] = r.json()["id"] + print(f"created camera {name} (id {local_cams[name]})") + r = requests.post( + f"{api}/api/v1/cameras/{local_cams[name]}/token", headers=admin, timeout=30 + ) + r.raise_for_status() + tokens[int(prod_id)] = {"Authorization": f"Bearer {r.json()['access_token']}"} + # A fresh pose per stream keeps this replay apart from earlier ones. + for prod_cam_id, azimuth in {f["stream"] for r in rounds for f in r}: + name = alert["cameras"][str(prod_cam_id)]["name"] + payload = {"camera_id": local_cams[name], "azimuth": azimuth} + r = requests.post( + f"{api}/api/v1/poses/", headers=admin, json=payload, timeout=30 + ) + r.raise_for_status() + poses[(prod_cam_id, azimuth)] = r.json()["id"] + + print(f"alert {alert_id}: replaying {len(rounds)} rounds ({args.mode} mode)") + for i, frames in enumerate(rounds): + for f in frames: + data = {"bboxes": fmt_bboxes(f["boxes"]), "pose_id": poses[f["stream"]]} + if args.mode == "demo": + data["recorded_at"] = (f["recorded_at"] + offset).isoformat() + r = requests.post( + f"{api}/api/v1/detections/", + headers=tokens[f["stream"][0]], + data=data, + files={"file": ("frame.jpg", zf.read(f["image"]), "image/jpeg")}, + timeout=60, + ) + if r.status_code not in (201, 204): + print(f" error on {f['image']}: {r.status_code} {r.text[:200]}") + print(f" round {i + 1}/{len(rounds)} sent ({len(frames)} frames)") + if args.mode == "live" and i + 1 < len(rounds): + time.sleep(args.interval) + + +def replay(args): + admin = login( + args.api_url, os.environ["SUPERADMIN_LOGIN"], os.environ["SUPERADMIN_PWD"] + ) + for alert_id in args.alert_ids: + replay_alert(alert_id, args, admin) + + +def main(): + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + sub = parser.add_subparsers(dest="cmd", required=True) + for cmd in ("fetch", "publish"): + sub.add_parser(cmd).add_argument("alert_ids", nargs="+", type=int) + sub.add_parser("list") + p = sub.add_parser("replay") + p.add_argument("alert_ids", nargs="+", type=int) + p.add_argument("--mode", choices=("live", "demo"), default="demo") + p.add_argument("--date", choices=("today", "original"), default="today") + p.add_argument("--interval", type=float, default=30, help="live mode, seconds") + p.add_argument("--api-url", default="http://localhost:5050") + p.add_argument("--org-id", type=int, default=2, help="org of created cameras") + args = parser.parse_args() + + load_dotenv(REPO_ROOT / ".env") + if args.cmd == "fetch": + fetch(args.alert_ids) + elif args.cmd == "publish": + publish(args.alert_ids) + elif args.cmd == "list": + list_alerts() + else: + replay(args) + + +if __name__ == "__main__": + main() diff --git a/tests/requirements.txt b/tests/requirements.txt index 4977cf9..e3ac0ea 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -2,3 +2,4 @@ psycopg2-binary==2.9.9 pytest==9.0.3 python-dotenv==1.2.2 boto3==1.34.90 +requests==2.32.3 diff --git a/tests/test_replay_alerts.py b/tests/test_replay_alerts.py new file mode 100644 index 0000000..d37fa9e --- /dev/null +++ b/tests/test_replay_alerts.py @@ -0,0 +1,75 @@ +import sys +from datetime import date, timedelta +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) + +from replay_alerts import ( # noqa: E402 + build_rounds, + day_offset, + fmt_bboxes, + parse_bboxes, +) + + +def det(ts, image, bbox, others=None): + return {"recorded_at": ts, "image": image, "bbox": bbox, "others_bboxes": others} + + +def test_bboxes_drop_degenerate_and_format(): + boxes = parse_bboxes("[(0,0,0,0,0),(0.1,0.2,1.0,0.9999,0.5)]") + assert fmt_bboxes(boxes) == "[(0.1,0.2,1,1,0.5)]" + assert fmt_bboxes(parse_bboxes("[]")) == "[]" + + +def test_build_rounds_merges_shared_images_and_interleaves_cameras(): + alert = { + "sequences": [ + { + "camera_id": 1, + "camera_azimuth": 10.0, + "detections": [ + det("2026-01-01T10:00:00", "a1", "[(0.1,0.1,0.2,0.2,0.5)]"), + det("2026-01-01T10:00:30", "a2", "[(0.1,0.1,0.2,0.2,0.6)]"), + det("2026-01-01T10:01:00", "a3", "[(0.1,0.1,0.2,0.2,0.7)]"), + ], + }, + { + # second smoke on the same upload as a2: one frame, both boxes + "camera_id": 1, + "camera_azimuth": 10.0, + "detections": [ + det( + "2026-01-01T10:00:30", + "a2", + "[(0.5,0.5,0.6,0.6,0.4)]", + "[(0.1,0.1,0.2,0.2,0.6)]", + ), + ], + }, + { + "camera_id": 2, + "camera_azimuth": 200.0, + "detections": [det("2026-01-01T10:00:10", "b1", "[]")], + }, + ] + } + rounds = build_rounds(alert) + assert [[f["image"] for f in r] for r in rounds] == [["a1", "b1"], ["a2"], ["a3"]] + assert len(rounds[1][0]["boxes"]) == 2 + assert rounds[0][1]["boxes"] == [] + + +def test_day_offset(): + alert = { + "sequences": [ + { + "camera_id": 1, + "camera_azimuth": 0.0, + "detections": [det("2026-01-01T23:59:00", "a", "[]")], + } + ] + } + rounds = build_rounds(alert) + assert day_offset(rounds, "original") == timedelta(0) + assert day_offset(rounds, "today") == date.today() - date(2026, 1, 1) From c4975c6e160683f35a16cb4f0bf49fb5a35c5679 Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 24 Sep 2026 17:26:36 +0200 Subject: [PATCH 02/14] feat: add make targets for real alert replays --- .env.test | 4 ++-- Makefile | 22 ++++++++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/.env.test b/.env.test index b087a71..7288631 100644 --- a/.env.test +++ b/.env.test @@ -40,7 +40,7 @@ TEMPORAL_MODEL_THRESHOLD=0.45 # auto-sets a dev token; set the same value here to require auth on other targets. TEMPORAL_API_TOKEN= -# only usefull to download some real sequences from real/distant Pyronear alert API -DISTANT_API_URL='' +# Prod admin creds, only needed by `make fetch-alerts` (scripts/replay_alerts.py) +DISTANT_API_URL=https://alertapi.pyronear.org/ DISTANT_ALERT_API_LOGIN='' DISTANT_ALERT_API_PASSWORD='' diff --git a/Makefile b/Makefile index 1d6526e..525c6ba 100644 --- a/Makefile +++ b/Makefile @@ -20,6 +20,10 @@ help: @echo " ps Show compose status" @echo " logs Follow logs" @echo " test Run pytest" + @echo " fetch-alerts Download prod alerts: make fetch-alerts ALERTS=\"54095 53800\"" + @echo " publish-alerts Upload fetched alerts to the replay-alerts GitHub release" + @echo " list-alerts List the alerts available in the release" + @echo " replay-alerts Replay alerts locally: make replay-alerts ALERTS=53800 MODE=live" # ------------------------------------------------------------------- # Init @@ -104,3 +108,21 @@ logs: test: pytest -s tests/* + +# ------------------------------------------------------------------- +# Real alert replays (see scripts/replay_alerts.py) +# ------------------------------------------------------------------- + +MODE ?= demo + +fetch-alerts: + python3 scripts/replay_alerts.py fetch $(ALERTS) + +publish-alerts: + python3 scripts/replay_alerts.py publish $(ALERTS) + +list-alerts: + python3 scripts/replay_alerts.py list + +replay-alerts: + python3 scripts/replay_alerts.py replay $(ALERTS) --mode $(MODE) From 37a9965c51bd2b7d0fe300453352077e2f13a09a Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 24 Sep 2026 17:26:36 +0200 Subject: [PATCH 03/14] chore: remove sample alert notebooks replaced by the replay script --- ...oad_Sequences_from_a_DistantAlertAPI.ipynb | 222 ----------- .../notebooks/app/send_real_alerts.ipynb | 375 ------------------ containers/notebooks/app/utils.py | 72 ---- 3 files changed, 669 deletions(-) delete mode 100644 containers/notebooks/app/Download_Sequences_from_a_DistantAlertAPI.ipynb delete mode 100644 containers/notebooks/app/send_real_alerts.ipynb diff --git a/containers/notebooks/app/Download_Sequences_from_a_DistantAlertAPI.ipynb b/containers/notebooks/app/Download_Sequences_from_a_DistantAlertAPI.ipynb deleted file mode 100644 index b38b3a1..0000000 --- a/containers/notebooks/app/Download_Sequences_from_a_DistantAlertAPI.ipynb +++ /dev/null @@ -1,222 +0,0 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "id": "a1c721ae-469b-48d7-a1df-3cd24623b3ee", - "metadata": {}, - "source": [ - "# Intro\n", - "\n", - "This notebook aims at downloading sequences data(images and bboxes) from a Distant/Real Pyronear Alert API isntance in order to recreate alerts/sequences with ease in local dev environnement\n", - "\n", - "You need to have accesses to a distant api and provide ```DISTANT_API_URL```, ```DISTANT_ALERT_API_LOGIN```, ```DISTANT_ALERT_API_PASSWORD```\n", - "\n", - "Then, \n", - "- either fill ```SEQUENCE_ID_LIST``` with the sequence ids you want in order to download related data (pick api admin credentialsi if you intend to download alert from various organisation)\n", - "- or let ```SEQUENCE_ID_LIST``` empty, pick a day to download by filling ```DAY_TO_DOWNLOAD``` (in this case, yyou need credentials from the organisation you want to download seq data)\n", - "\n", - "Also, you can fill ```TRIANGULATED_SEQUENCE_LIST``` with sequences id that triangulates\n", - "\n", - "\n", - "If you do not have anything from above, do not worry, you can use the notebook \"send_real_alerts\" to send some alerts to local env dev" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "649157c6-0ce3-4324-a7b6-15c911ddaf36", - "metadata": {}, - "outputs": [], - "source": [ - "BASE_DIRECTORY = \"../data/alert_samples\" # directory where to put sequences data\n", - "DAY_TO_DOWNLOAD = \"2025-08-26\" #YYYY-MM-DD\n", - "SEQUENCE_ID_LIST = [] #fullfill this with alerts id" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "4e511d97-245c-45cc-9216-a16d31b663aa", - "metadata": {}, - "outputs": [], - "source": [ - "import requests\n", - "from dotenv import load_dotenv\n", - "import os\n", - "import pandas as pd\n", - "from pyroclient import Client\n", - "\n", - "load_dotenv(\"../.env\")\n", - "DISTANT_API_URL = os.environ.get(\"DISTANT_API_URL\")\n", - "DISTANT_ALERT_API_LOGIN = os.environ.get(\"DISTANT_ALERT_API_LOGIN\")\n", - "DISTANT_ALERT_API_PASSWORD = os.environ.get(\"DISTANT_ALERT_API_PASSWORD\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "ac07ba29-68d0-42d2-bafe-e94d961ec6c2", - "metadata": {}, - "outputs": [], - "source": [ - "token = requests.post(\n", - " f\"{DISTANT_API_URL}api/v1/login/creds\",\n", - " data={\"username\": DISTANT_ALERT_API_LOGIN, \"password\": DISTANT_ALERT_API_PASSWORD},\n", - " timeout=5,\n", - ").json()['access_token']\n", - "\n", - "api_client = Client(token, DISTANT_API_URL)\n", - "\n", - "# get cameras -> in order to get cam name from cam id\n", - "cameras = api_client.fetch_cameras().json()\n", - "\n", - "# define base directory\n", - "base_dir = os.path.join(BASE_DIRECTORY)\n", - "os.makedirs(base_dir, exist_ok=True)\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "512ce6f1-76db-43b6-adb3-131942e8a95e", - "metadata": {}, - "outputs": [], - "source": [ - "response = api_client.fetch_sequences_from_date(DAY_TO_DOWNLOAD, limit=50)\n", - "api_sequences = pd.DataFrame(response.json())\n", - "api_sequences.head()" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "1fa5e0df-a090-4f96-ae9a-cd54f365d2c2", - "metadata": {}, - "outputs": [], - "source": [ - "# you can also reproduce all alerts from this day\n", - "if len(SEQUENCE_ID_LIST)==0:\n", - " SEQUENCE_ID_LIST = list(api_sequences[\"id\"])" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "d3c12b1d-54f1-4d8d-b591-dc04ecc35db2", - "metadata": {}, - "outputs": [], - "source": [ - "def dl_seqs_in_target_dir(sequence_id_list, target_dir, api_client):\n", - " \"\"\"\n", - " Download sequences from sequence_id_list to target_dir, using an instanciated api_client (of distant Pyronear alert API)\n", - "\n", - " \"\"\"\n", - " \n", - " for seq_id in sequence_id_list: \n", - " sequences = api_client.fetch_sequences_detections(sequence_id=seq_id, limit=10, desc=False).json()\n", - " \n", - " cam_name = [item['name'] for item in cameras if item['id'] == sequences[0]['camera_id']][0]\n", - " created_at_rounded = sequences[0][\"created_at\"].split('.')[0].replace(':', '-').replace('T', '_')\n", - " cam_id_distant_api = sequences[0][\"camera_id\"]\n", - " print(f\"== Download Alerts data for sequence ID {seq_id} - camera {cam_name} at {created_at_rounded}\")\n", - " \n", - " alert_dir = os.path.join(f\"{cam_id_distant_api}_{cam_name}_{created_at_rounded}\")\n", - " image_dir = os.path.join(target_dir,alert_dir, \"images\")\n", - " pred_dir = os.path.join(target_dir, alert_dir, \"labels_predictions\")\n", - " os.makedirs(image_dir, exist_ok=True)\n", - " os.makedirs(pred_dir, exist_ok=True)\n", - " \n", - " for seq in sequences:\n", - " \n", - " # bbox\n", - " #yolo_format_bbox = ' '.join(map(str,ast.literal_eval(seq[\"bboxes\"])[0]))\n", - " bboxes = seq[\"bboxes\"]\n", - " \n", - " bbox_file_name = seq[\"bucket_key\"][:-4] + \".txt\"\n", - " \n", - " with open(os.path.join(target_dir, alert_dir, \"labels_predictions\",bbox_file_name), 'w') as f:\n", - " f.write(bboxes)\n", - " \n", - " url = seq['url']\n", - " nom_fichier = seq[\"bucket_key\"]\n", - " # image\n", - " response = requests.get(url, stream=True)\n", - " if response.status_code == 200:\n", - " full_img_path = os.path.join(image_dir, nom_fichier)\n", - " with open(full_img_path, 'wb') as f:\n", - " for chunk in response.iter_content(1024):\n", - " f.write(chunk)\n", - " else:\n", - " print(f\"Error during download.\")\n", - " print(\"Download complete\")\n", - " " - ] - }, - { - "cell_type": "markdown", - "id": "9503b5d5-9930-4c9e-9fbe-1eb1e563fd44", - "metadata": {}, - "source": [ - "# Download sequences from SEQUENCE_ID_LIST list" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "b2d033e9-7ce2-4cf4-995e-01ea2c15e235", - "metadata": {}, - "outputs": [], - "source": [ - "single_alerts = os.path.join(BASE_DIRECTORY, \"single_sequences\")\n", - "os.makedirs(single_alerts, exist_ok=True)\n", - "\n", - "dl_seqs_in_target_dir(SEQUENCE_ID_LIST, single_alerts, api_client)" - ] - }, - { - "cell_type": "markdown", - "id": "e9d109db-324c-4268-bae6-27c5960def55", - "metadata": {}, - "source": [ - "## Download triangulated alerts\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "68a2cff8-f847-4e6c-8c74-37b7ecffc9af", - "metadata": {}, - "outputs": [], - "source": [ - "\n", - "TRIANGULATED_SEQUENCE_LIST = [10535, 10537, 10538, 15462, 15463]\n", - "TRIANGULATED_DIRECTORY= \"triangulated_sequences\"\n", - "triangulated_dir = os.path.join(BASE_DIRECTORY, TRIANGULATED_DIRECTORY)\n", - "os.makedirs(triangulated_dir, exist_ok=True)\n", - "\n", - "dl_seqs_in_target_dir(TRIANGULATED_SEQUENCE_LIST, triangulated_dir, api_client)" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.13.7" - } - }, - "nbformat": 4, - "nbformat_minor": 5 -} diff --git a/containers/notebooks/app/send_real_alerts.ipynb b/containers/notebooks/app/send_real_alerts.ipynb deleted file mode 100644 index 07f3bb5..0000000 --- a/containers/notebooks/app/send_real_alerts.ipynb +++ /dev/null @@ -1,375 +0,0 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "# Intro\n", - "\n", - "This notebook is designed to generate alerts so that you can work on the development environment.\n", - "\n", - "It works as it is (just run blocks from the setup section), then you can run any section.\n", - "\n", - "It is also possible to run the entire notebook at once: To do so, in the menu, click ‘Run’ then ‘Run All Cells’." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "import ast\n", - "import pandas as pd\n", - "from dotenv import load_dotenv\n", - "import os\n", - "from api import get_token, get_camera_token\n", - "from pyroclient import Client\n", - "import glob\n", - "from PIL import Image\n", - "import numpy as np\n", - "import io\n", - "import itertools\n", - "import requests\n", - "import random\n", - "import shutil\n", - "from utils import dl_seqs_from_url, read_pred_file, send_triangulated_alerts\n", - "import time" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "API_URL = \"http://api:5050\"\n", - "load_dotenv(\"../.env\")\n", - "SUPERADMIN_LOGIN = os.environ.get(\"SUPERADMIN_LOGIN\")\n", - "SUPERADMIN_PWD = os.environ.get(\"SUPERADMIN_PWD\")\n", - "\n", - "# Get access token\n", - "admin_access_token = get_token(API_URL, SUPERADMIN_LOGIN, SUPERADMIN_PWD)\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "users = pd.read_csv(\"../data/csv/API_DATA_DEV - users.csv\")\n", - "cameras = pd.read_csv(\"../data/csv/API_DATA_DEV - cameras.csv\")\n", - "users" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "cameras" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "#### Download alerts sequences" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "BASE_DIRECTORY = \"../data\"\n", - "\n", - "SAMPLE_PATH = \"alert_samples\"\n", - "url = \"https://github.com/pyronear/pyro-envdev/releases/download/v0.0.1/alert_samples.zip\"\n", - "\n", - "if not os.path.isdir(os.path.join(BASE_DIRECTORY, SAMPLE_PATH)):\n", - " print(\"Images not found, dowloading ...\")\n", - " #os.makedirs(SAMPLE_PATH, exist_ok=True)\n", - " dl_seqs_from_url(url, os.path.join(BASE_DIRECTORY, SAMPLE_PATH))\n", - "else:\n", - " print(f\"SAMPLE_PATH {SAMPLE_PATH} directory exists, assume alerts inside, no download\")\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Send some random alerts sequences from a list of cameras\n", - "Download a file containing relevant images and predictions, then send alerts to selected cameras (with a random azimuth and random imgs picked from the alerts folder downloaded below)." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "send_alert_from_cam_ids = [2, 4, 5, 14] # select cameras" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "SAMPLE_PATH = \"selection-true-positives\"\n", - "url = \"https://github.com/pyronear/pyro-envdev/releases/download/v0.0.1/selection-true-positives.zip\"\n", - "\n", - "if not os.path.isdir(SAMPLE_PATH):\n", - " print(\"Images not found, dowloading ...\")\n", - " \n", - " os.makedirs(SAMPLE_PATH, exist_ok=True)\n", - " dl_seqs_from_url(url, SAMPLE_PATH)\n", - "else:\n", - " print(f\"SAMPLE_PATH {SAMPLE_PATH} directory exists, assume alerts inside, no download\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "send_alert_from_cam_ids = [2, 4, 5, 14] # select cameras\n", - "\n", - "sequances_folders = glob.glob(f\"{SAMPLE_PATH}/*\")\n", - "\n", - "for camera_id in send_alert_from_cam_ids:\n", - " \n", - " camera_token = get_camera_token(API_URL, camera_id, admin_access_token)\n", - " camera_client = Client(camera_token, API_URL)\n", - "\n", - " sequances_folder = sequances_folders[random.randint(0,len(sequances_folders)-1)]\n", - " imgs = glob.glob(f\"{sequances_folder}/images/*\")\n", - " imgs.sort()\n", - " preds = glob.glob(f\"{sequances_folder}/labels_predictions/*\")\n", - " preds.sort()\n", - "\n", - " cam_poses = camera_client.get_current_poses().json()\n", - " # pick a random pose id\n", - " pose_id = random.choice(cam_poses)['id']\n", - "\n", - " print(f\"Sending alerts from camera {camera_id} at (pose_id={pose_id})\")\n", - " for img_file, pred_file in zip(imgs, preds):\n", - " \n", - " stream = io.BytesIO()\n", - " im = Image.open(img_file)\n", - " im.save(stream, format=\"JPEG\", quality=80)\n", - "\n", - " bboxes = read_pred_file(pred_file)\n", - " response = camera_client.create_detection(stream.getvalue(), bboxes, pose_id=pose_id)\n", - " # Force a KeyError if the request failed\n", - " \n", - " response.json()[\"id\"]" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Send custom alerts sequences\n", - "\n", - "The following code block helps you send some alerts/sequences from a specific diretory containing sequences data\n", - "\n", - "- Create a directory with the name you want, ex ```alerts_with_style```\n", - "- In directory ```../data```or in current dir, there are some alerts folders downloaded\n", - "- Copy each inteterting alert dir (dir that contains a dir img, and a dir label predictions) and paste them to your ```alerts_with_style``` directory\n", - "- then below, fill the ```SEQUENCES_DIR_PATH``` with the path to you directory\n", - "- fill the CAM_MAPPING mapping dict : the first number in alert folder are dict keys that will map with camera id in database : ex {2:78} -> all alert folder starting with 2 will create alerts for the camera of id 78\n", - " - you can find cameras id in the object ```cameras``` (created at the beginning of this notebook), or look in the DB table cameras\n", - " - If you put alert dir with prefix number that does not match, no alerts will be created\n", - " " - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "SEQUENCES_DIR_PATH = \"../data/alert_samples/single_sequences/*\"\n", - "sequences_directories = glob.glob(SEQUENCES_DIR_PATH)\n", - "CAM_MAPPING = {22: 13, 42: 7, 59: 5, 14: 11, 43: 8, 79: 14, 13: 10, 11: 15, 60: 2, 15: 12, 12: 9, 10: 16, 41: 2, 65: 8}\n", - "\n", - "for seq_path in sequences_directories:\n", - " \n", - " seq_prefix_num = int(seq_path.split(\"/\")[-1].split(\"_\")[0])\n", - "\n", - " if CAM_MAPPING.get(seq_prefix_num) is None:\n", - " print(f\" === WARNING the prefix {seq_prefix_num} is not in CAM_MAPPING dict, hence no alerts will be generated (or from random camera but no implemented yet)\")\n", - " else: \n", - " camera_id = CAM_MAPPING[seq_prefix_num]\n", - " camera_token = get_camera_token(API_URL, camera_id, admin_access_token)\n", - " camera_client = Client(camera_token, API_URL)\n", - " \n", - " imgs = glob.glob(f\"{seq_path}/images/*\")\n", - " imgs.sort()\n", - " preds = glob.glob(f\"{seq_path}/labels_predictions/*\")\n", - " preds.sort()\n", - "\n", - " cam_poses = camera_client.get_current_poses().json()\n", - " # pick a random pose id\n", - " pose_id = random.choice(cam_poses)['id']\n", - "\n", - " \n", - " print(f\"Sending alerts from camera {camera_id} at (pose_id={pose_id})\")\n", - " for img_file, pred_file in zip(imgs, preds):\n", - " \n", - " stream = io.BytesIO()\n", - " im = Image.open(img_file)\n", - " im.save(stream, format=\"JPEG\", quality=80)\n", - " \n", - " #bboxes = read_pred_file(pred_file)\n", - " with open(pred_file, 'r', encoding='utf-8') as file:\n", - " bboxes = ast.literal_eval(file.read())\n", - " #bboxes = np.loadtxt(pred_file, ndmin=2)\n", - " response = camera_client.create_detection(stream.getvalue(), bboxes, pose_id=pose_id)\n", - "\n", - " # Force a KeyError if the request failed\n", - " time.sleep(0.5)\n", - " response.json()[\"id\"]" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Send 2 sequences that triangulate (user test07)\n", - "Download a file containning 2 alerts from 2 cameras that triangulate, then it send alerts to the api. " - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "BASE_DIRECTORY = \"../data\"\n", - "SAMPLE_PATH = \"triangulated_sequences\"\n", - "url = \"https://github.com/pyronear/pyro-envdev/releases/download/v0.0.1/triangulated_sequences.zip\" \n", - "if not os.path.isdir(os.path.join(BASE_DIRECTORY, SAMPLE_PATH)):\n", - " print(\"Images not found, dowloading ...\")\n", - " dl_seqs_from_url(url, os.path.join(BASE_DIRECTORY, SAMPLE_PATH))\n", - "else:\n", - " print(f\"SAMPLE_PATH {SAMPLE_PATH} directory exists, assume alerts inside, no download\")\n", - "\n", - "\n", - "cam_triangulation = {\n", - " \"12\": {\n", - " \"azimuth\": 226,\n", - " \"path\": os.path.join(BASE_DIRECTORY, SAMPLE_PATH, \"brison-03\")\n", - " },\n", - " \"13\": {\n", - " \"azimuth\": 54,\n", - " \"path\": os.path.join(BASE_DIRECTORY, SAMPLE_PATH, \"serre-de-barre-01\")\n", - " }\n", - "}\n", - "\n", - "send_triangulated_alerts(cam_triangulation, API_URL, Client, admin_access_token)\n" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Send 2 other sequences that triangulates (user test07)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "cam_triangulation = {\n", - " \"11\": {\n", - " \"azimuth\": 226,\n", - " \"path\": \"../data/alert_samples/triangulated_sequences/14_brison-03_2025-09-29_13-58-36\"\n", - " },\n", - " \"13\": {\n", - " \"azimuth\": 173,\n", - " \"path\": \"../data/alert_samples/triangulated_sequences/22_serre-de-barre-01_2025-09-29_13-56-58\"\n", - " }\n", - "}\n", - "\n", - "\n", - "send_triangulated_alerts(cam_triangulation, API_URL, Client, admin_access_token)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Send 3 sequences that triangulates (user test77)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "cam_triangulation = {\n", - " \"4\": {\n", - " \"azimuth\": 190,\n", - " \"path\": \"../data/alert_samples/triangulated_sequences/41_croix-augas-02_2025-07-14_11-24-31\"\n", - " },\n", - " \"8\": {\n", - " \"azimuth\": 25,\n", - " \"path\": \"../data/alert_samples/triangulated_sequences/65_nemours-02_2025-07-14_11-24-10\"\n", - " },\n", - " \"5\": {\n", - " \"azimuth\": 280,\n", - " \"path\": \"../data/alert_samples/triangulated_sequences/66_moret-sur-loing-01_2025-07-14_11-22-05\"\n", - " }\n", - "}\n", - "\n", - "send_triangulated_alerts(cam_triangulation, API_URL, Client, admin_access_token)\n" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3 (ipykernel)", - "language": "python", - "name": "python3" - }, - "language_info": { - "codemirror_mode": { - "name": "ipython", - "version": 3 - }, - "file_extension": ".py", - "mimetype": "text/x-python", - "name": "python", - "nbconvert_exporter": "python", - "pygments_lexer": "ipython3", - "version": "3.11.14" - } - }, - "nbformat": 4, - "nbformat_minor": 4 -} diff --git a/containers/notebooks/app/utils.py b/containers/notebooks/app/utils.py index cb20243..f5eb64c 100644 --- a/containers/notebooks/app/utils.py +++ b/containers/notebooks/app/utils.py @@ -1,16 +1,8 @@ -import ast import os -import io -import glob import numpy as np import shutil -import itertools -import time -from PIL import Image import requests -from api import get_camera_token - def xywh2xyxy(x: np.ndarray): """ @@ -140,67 +132,3 @@ def dl_seqs_from_url(url, output_path): shutil.unpack_archive(zip_path, output_path, "zip") os.remove(zip_path) print("Extraction completed.") - - -def get_or_create_pose_id_for_azimuth( - camera_client, camera_id, azimuth, tol=0.1, patrol_id=None -): - resp = camera_client.get_current_poses() - resp.raise_for_status() - for pose in resp.json(): - if abs(pose["azimuth"] - azimuth) <= tol: - return pose["id"] - print(f"did not found a matching pose_id, gonna create one, azimuth was {azimuth}") - create_resp = camera_client.create_pose( - camera_id=camera_id, azimuth=azimuth, patrol_id=patrol_id - ) - create_resp.raise_for_status() - return create_resp.json()["id"] - - -def send_triangulated_alerts( - cam_triangulation, API_URL, Client, admin_access_token, sleep_seconds=1 -): - for cam_id, info in cam_triangulation.items(): - camera_token = get_camera_token(API_URL, cam_id, admin_access_token) - camera_client = Client(camera_token, API_URL) - info["client"] = camera_client - - seq_folder = info["path"] - - imgs = glob.glob(f"{seq_folder}/images/*") - imgs.sort() - preds = glob.glob(f"{seq_folder}/labels_predictions/*") - preds.sort() - - print(f"Cam {cam_id}: {len(imgs)} images, {len(preds)} preds") # debug - - info["seq_data_pair"] = list(zip(imgs, preds)) - - print("Send some entrelaced detections") - for files in itertools.zip_longest( - *(info["seq_data_pair"] for info in cam_triangulation.values()) - ): - for (cam_id, info), pair in zip(cam_triangulation.items(), files): - if pair is None: - continue # len of sequences might be different - img_file, pred_file = pair - client = info["client"] - azimuth = info["azimuth"] - # print(f"cam if {cam_id}") - pose_id = get_or_create_pose_id_for_azimuth(client, cam_id, azimuth) - # print(f"pose_id:{pose_id}") - stream = io.BytesIO() - im = Image.open(img_file) - im.save(stream, format="JPEG", quality=80) - - with open(pred_file, "r") as file: - bboxes = ast.literal_eval(file.read()) - - response = client.create_detection( - stream.getvalue(), bboxes, pose_id=pose_id - ) - time.sleep(sleep_seconds) - - response.json()["id"] # Force a KeyError if the request failed - print(f"detection sent for cam {cam_id}") From 73887c93e2d1c59d2f94f80afcbc1cd2be0b472b Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 24 Sep 2026 17:26:36 +0200 Subject: [PATCH 04/14] docs: document real alert replays --- README.md | 31 +++++++++++++++++++++++++------ 1 file changed, 25 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 70fab60..06e67ec 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,7 @@ make build make run ``` -* Send an alert by opening [http://0.0.0.0:8889/notebooks/notebooks/send_real_alerts.ipynb](http://0.0.0.0:8889/notebooks/notebooks/send_real_alerts.ipynb) +* Replay a real alert with `make replay-alerts ALERTS=` (see [Replay real alerts](#replay-real-alerts)) * Observe the alert on the frontend at [http://localhost:8080/](http://localhost:8080/) * Use credentials from `data/csv/API_DATA_DEV/users.csv` * Or check directly on the API at [http://0.0.0.0:5050/docs](http://0.0.0.0:5050/docs) with the same creds @@ -117,13 +117,32 @@ docker logs engine Create a directory `data/images` before starting the environment and put your images inside. -### Send custom alerts +### Replay real alerts -Use Jupyter notebooks (e.g., `notebooks/send_real_alerts.ipynb`). -When running notebooks **inside Docker**, set: +`scripts/replay_alerts.py` replays real production alerts (all their sequences and +images) on the local stack. Alerts are stored as `alert_.zip` in the +[`replay-alerts`](https://github.com/pyronear/pyro-envdev/releases/tag/replay-alerts) +GitHub release. Needs `requests` and `python-dotenv` on the host. + +```bash +make list-alerts # alerts available in the release +make replay-alerts ALERTS="53800 53941" # demo mode (default) +make replay-alerts ALERTS=53800 MODE=live # live mode +``` + +* **demo**: posts everything at once, dated today at the original time of day. + Add `--date original` (script only) to keep the original dates. +* **live**: sends one frame per camera every 30 s (`--interval`), dated now. +* Missing cameras are created by name in organization 2 (`--org-id`), so the + alerts show up for the `test77` user. + +Add new alerts (needs prod admin creds `DISTANT_*` in `.env`, and `gh` for publishing): + +```bash +make fetch-alerts ALERTS="54095" # writes data/replay_alerts/alert_54095.zip +make publish-alerts ALERTS="54095" # uploads it to the release +``` -```python -API_URL = "http://api:5050" ### Update the last image for a camera From 8c38ea81707f09be34078ba8761cb8bc7893b595 Mon Sep 17 00:00:00 2001 From: Mateo Date: Thu, 24 Sep 2026 17:29:01 +0200 Subject: [PATCH 05/14] fix: write fetched zips atomically and fail replay on upload errors --- scripts/replay_alerts.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index 7dff167..8ef1c03 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -78,7 +78,9 @@ def fetch(alert_ids): api_url, f"/api/v1/alerts/{alert_id}/sequences", headers, limit=100 ) cameras = {} - with zipfile.ZipFile(zip_path(alert_id), "w") as zf: + # Written to a temp file so a failed fetch never leaves a broken zip behind. + tmp = zip_path(alert_id).with_suffix(".tmp") + with zipfile.ZipFile(tmp, "w") as zf: for seq in sequences: cam_id = seq["camera_id"] if cam_id not in cameras: @@ -115,6 +117,7 @@ def fetch(alert_ids): alert["sequences"] = sequences alert["cameras"] = cameras zf.writestr("alert.json", json.dumps(alert, indent=1)) + tmp.replace(zip_path(alert_id)) n_dets = sum(len(s["detections"]) for s in sequences) print( f"alert {alert_id}: {len(sequences)} sequences, {n_dets} detections, " @@ -253,6 +256,7 @@ def replay_alert(alert_id, args, admin): poses[(prod_cam_id, azimuth)] = r.json()["id"] print(f"alert {alert_id}: replaying {len(rounds)} rounds ({args.mode} mode)") + failures = 0 for i, frames in enumerate(rounds): for f in frames: data = {"bboxes": fmt_bboxes(f["boxes"]), "pose_id": poses[f["stream"]]} @@ -266,18 +270,21 @@ def replay_alert(alert_id, args, admin): timeout=60, ) if r.status_code not in (201, 204): + failures += 1 print(f" error on {f['image']}: {r.status_code} {r.text[:200]}") print(f" round {i + 1}/{len(rounds)} sent ({len(frames)} frames)") if args.mode == "live" and i + 1 < len(rounds): time.sleep(args.interval) + return failures def replay(args): admin = login( args.api_url, os.environ["SUPERADMIN_LOGIN"], os.environ["SUPERADMIN_PWD"] ) - for alert_id in args.alert_ids: - replay_alert(alert_id, args, admin) + failures = sum(replay_alert(a, args, admin) for a in args.alert_ids) + if failures: + sys.exit(f"{failures} detection uploads failed") def main(): From 9fc33d37ed2a804b57cb73217401768933199c53 Mon Sep 17 00:00:00 2001 From: Mateo Date: Mon, 28 Sep 2026 14:46:25 +0200 Subject: [PATCH 06/14] chore: run replay script with uv and polish release handling - Declare script deps inline (PEP 723) and call it with `uv run` - `list` prints "no alerts published" instead of failing when the release is missing - Create the release with --latest=false so it never replaces v0.0.1 as latest - Document uv and the publication flow in the README --- Makefile | 8 ++++---- README.md | 6 +++++- scripts/replay_alerts.py | 18 +++++++++++++----- 3 files changed, 22 insertions(+), 10 deletions(-) diff --git a/Makefile b/Makefile index 525c6ba..79896c3 100644 --- a/Makefile +++ b/Makefile @@ -116,13 +116,13 @@ test: MODE ?= demo fetch-alerts: - python3 scripts/replay_alerts.py fetch $(ALERTS) + uv run scripts/replay_alerts.py fetch $(ALERTS) publish-alerts: - python3 scripts/replay_alerts.py publish $(ALERTS) + uv run scripts/replay_alerts.py publish $(ALERTS) list-alerts: - python3 scripts/replay_alerts.py list + uv run scripts/replay_alerts.py list replay-alerts: - python3 scripts/replay_alerts.py replay $(ALERTS) --mode $(MODE) + uv run scripts/replay_alerts.py replay $(ALERTS) --mode $(MODE) diff --git a/README.md b/README.md index 06e67ec..e9f2d19 100644 --- a/README.md +++ b/README.md @@ -122,7 +122,7 @@ Create a directory `data/images` before starting the environment and put your im `scripts/replay_alerts.py` replays real production alerts (all their sequences and images) on the local stack. Alerts are stored as `alert_.zip` in the [`replay-alerts`](https://github.com/pyronear/pyro-envdev/releases/tag/replay-alerts) -GitHub release. Needs `requests` and `python-dotenv` on the host. +GitHub release. Needs [uv](https://docs.astral.sh/uv/) (dependencies are installed on the fly). ```bash make list-alerts # alerts available in the release @@ -143,6 +143,10 @@ make fetch-alerts ALERTS="54095" # writes data/replay_alerts/alert_54095.zi make publish-alerts ALERTS="54095" # uploads it to the release ``` +A zip holds `alert.json` (alert, sequences, detections, cameras) and the detection +images. `publish` creates the release if needed and overwrites an existing zip. +`replay` uses the local zip when present, so publishing is only needed to share. + ### Update the last image for a camera diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index 8ef1c03..3f52f8a 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -1,4 +1,8 @@ -#!/usr/bin/env python3 +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# dependencies = ["requests", "python-dotenv"] +# /// """Fetch real alerts from the production alert API, share them through a GitHub release, and replay them on the local stack. @@ -12,9 +16,9 @@ time of day (--date today) or as in prod (--date original) Examples: - python3 scripts/replay_alerts.py fetch 54095 - python3 scripts/replay_alerts.py publish 54095 - python3 scripts/replay_alerts.py replay 54095 --mode demo + uv run scripts/replay_alerts.py fetch 54095 + uv run scripts/replay_alerts.py publish 54095 + uv run scripts/replay_alerts.py replay 54095 --mode demo """ import argparse @@ -134,7 +138,8 @@ def publish(alert_ids): if subprocess.run([*gh, "view", RELEASE_TAG], capture_output=True).returncode: notes = "Real alerts for scripts/replay_alerts.py" subprocess.run( - [*gh, "create", RELEASE_TAG, "--title", "Replay alerts", "--notes", notes], + [*gh, "create", RELEASE_TAG, "--title", "Replay alerts", "--notes", notes, + "--latest=false"], check=True, ) subprocess.run([*gh, "upload", RELEASE_TAG, *files, "--clobber"], check=True) @@ -145,6 +150,9 @@ def list_alerts(): f"https://api.github.com/repos/{GITHUB_REPO}/releases/tags/{RELEASE_TAG}", timeout=30, ) + if r.status_code == 404: + print("no alerts published") + return r.raise_for_status() for asset in r.json()["assets"]: print(asset["name"].removeprefix("alert_").removesuffix(".zip")) From b539ca55ca303926f3fc09477b1277031a72358d Mon Sep 17 00:00:00 2001 From: Mateo Date: Mon, 28 Sep 2026 14:56:28 +0200 Subject: [PATCH 07/14] feat: fetch and replay detection crops - fetch downloads each detection crop (signed via /detections/{id}/url, since sequence detection lists return crop_url=null) into crops/ in the zip - replay sends one crop per box, or none if any box lacks one, as the API requires --- README.md | 4 ++-- scripts/replay_alerts.py | 36 +++++++++++++++++++++++++++++------- 2 files changed, 31 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index e9f2d19..48fac86 100644 --- a/README.md +++ b/README.md @@ -143,8 +143,8 @@ make fetch-alerts ALERTS="54095" # writes data/replay_alerts/alert_54095.zi make publish-alerts ALERTS="54095" # uploads it to the release ``` -A zip holds `alert.json` (alert, sequences, detections, cameras) and the detection -images. `publish` creates the release if needed and overwrites an existing zip. +A zip holds `alert.json` (alert, sequences, detections, cameras), the detection +images and their crops. `publish` creates the release if needed and overwrites an existing zip. `replay` uses the local zip when present, so publishing is only needed to share. diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index 3f52f8a..bcd6d5d 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -59,6 +59,13 @@ def get(api_url, path, headers, **params): return r.json() +def download(zf, name, url): + if name not in zf.namelist(): + r = requests.get(url, timeout=60) + r.raise_for_status() + zf.writestr(name, r.content) + + def zip_path(alert_id): return ALERTS_DIR / f"alert_{alert_id}.zip" @@ -103,16 +110,22 @@ def fetch(alert_ids): ) for det in page: image = f"images/{det['bucket_key']}" - if image not in zf.namelist(): - img = requests.get(det["url"], timeout=60) - img.raise_for_status() - zf.writestr(image, img.content) + download(zf, image, det["url"]) + crop = None + if det["crop_bucket_key"]: + # Detection lists return crop_url=null, only /url signs it. + crop = f"crops/{det['crop_bucket_key']}" + urls = get( + api_url, f"/api/v1/detections/{det['id']}/url", headers + ) + download(zf, crop, urls["crop_url"]) seq["detections"].append( { "recorded_at": det["recorded_at"], "bbox": det["bbox"], "others_bboxes": det["others_bboxes"], "image": image, + "crop": crop, } ) if len(page) < 100: @@ -194,12 +207,16 @@ def build_rounds(alert): for det in seq["detections"]: frame = frames.setdefault( det["image"], - {"stream": stream, "image": det["image"], "boxes": []}, + {"stream": stream, "image": det["image"], "boxes": [], "crops": {}}, ) frame["recorded_at"] = datetime.fromisoformat(det["recorded_at"]) - for box in parse_bboxes(det["bbox"]) + parse_bboxes(det["others_bboxes"]): + own = parse_bboxes(det["bbox"]) + for box in own + parse_bboxes(det["others_bboxes"]): if box not in frame["boxes"]: frame["boxes"].append(box) + # A prod detection holds one box and the crop of that box. + if det.get("crop") and len(own) == 1: + frame["crops"][own[0]] = det["crop"] streams = {} for frame in sorted(frames.values(), key=lambda f: f["recorded_at"]): streams.setdefault(frame["stream"], []).append(frame) @@ -270,11 +287,16 @@ def replay_alert(alert_id, args, admin): data = {"bboxes": fmt_bboxes(f["boxes"]), "pose_id": poses[f["stream"]]} if args.mode == "demo": data["recorded_at"] = (f["recorded_at"] + offset).isoformat() + files = [("file", ("frame.jpg", zf.read(f["image"]), "image/jpeg"))] + # The API wants one crop per box, in the same order, or none at all. + crops = [f["crops"].get(b) for b in f["boxes"]] + if crops and all(crops): + files += [("crop", ("crop.jpg", zf.read(c), "image/jpeg")) for c in crops] r = requests.post( f"{api}/api/v1/detections/", headers=tokens[f["stream"][0]], data=data, - files={"file": ("frame.jpg", zf.read(f["image"]), "image/jpeg")}, + files=files, timeout=60, ) if r.status_code not in (201, 204): From 8082ec6183f9eb0468ac38fa91b95ffb5e9e7615 Mon Sep 17 00:00:00 2001 From: Mateo Date: Mon, 28 Sep 2026 18:32:47 +0200 Subject: [PATCH 08/14] fix: shift demo replays as a block so they triangulate Demo mode dated alerts today at the original time of day, which could be in the future. The API sets last_seen_at to the server time, so a future started_at puts sequences out of the triangulation time window. - Shift all frames by one offset, keeping the prod gaps between them - By default the latest sequence starts 1 hour ago, so every sequence starts in the past; --start / START sets the first frame time instead - Replace --date today|original --- Makefile | 2 +- README.md | 7 +++++-- scripts/replay_alerts.py | 38 ++++++++++++++++++++++++-------------- 3 files changed, 30 insertions(+), 17 deletions(-) diff --git a/Makefile b/Makefile index 79896c3..35057e8 100644 --- a/Makefile +++ b/Makefile @@ -125,4 +125,4 @@ list-alerts: uv run scripts/replay_alerts.py list replay-alerts: - uv run scripts/replay_alerts.py replay $(ALERTS) --mode $(MODE) + uv run scripts/replay_alerts.py replay $(ALERTS) --mode $(MODE) $(if $(START),--start $(START)) diff --git a/README.md b/README.md index 48fac86..781ece4 100644 --- a/README.md +++ b/README.md @@ -127,11 +127,14 @@ GitHub release. Needs [uv](https://docs.astral.sh/uv/) (dependencies are install ```bash make list-alerts # alerts available in the release make replay-alerts ALERTS="53800 53941" # demo mode (default) +make replay-alerts ALERTS=53800 START=2026-07-10T18:53 # demo, first frame at a given local time make replay-alerts ALERTS=53800 MODE=live # live mode ``` -* **demo**: posts everything at once, dated today at the original time of day. - Add `--date original` (script only) to keep the original dates. +* **demo**: posts everything at once, shifted as a block (gaps between frames kept) + so its latest sequence starts 1 hour ago, or its first frame is at `START` (local + time). A sequence starting in the future does not triangulate: the API dates + `last_seen_at` with the server time. * **live**: sends one frame per camera every 30 s (`--interval`), dated now. * Missing cameras are created by name in organization 2 (`--org-id`), so the alerts show up for the `test77` user. diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index bcd6d5d..385bf28 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -12,8 +12,9 @@ list list the alerts available in the release replay ALERT_ID... replay alerts on the local API, one after the other --mode live one frame per camera every --interval seconds, timestamped now - --mode demo post everything at once, timestamped today at the original - time of day (--date today) or as in prod (--date original) + --mode demo post everything at once, shifted as a block: the first frame at + --start (local time), or by default the latest sequence + starting 1 hour ago Examples: uv run scripts/replay_alerts.py fetch 54095 @@ -29,7 +30,7 @@ import sys import time import zipfile -from datetime import date, datetime, timedelta, timezone +from datetime import datetime, timedelta, timezone from itertools import zip_longest from pathlib import Path @@ -223,12 +224,9 @@ def build_rounds(alert): return [[f for f in r if f] for r in zip_longest(*streams.values())] -def day_offset(rounds, when): - """Shift to apply to recorded_at: whole days, so all frames move together.""" - if when == "original": - return timedelta(0) - first = min(f["recorded_at"] for r in rounds for f in r) - return timedelta(days=(date.today() - first.date()).days) +def utc_start(s): + """--start value: local time (naive) or with offset, returned as naive UTC.""" + return datetime.fromisoformat(s).astimezone(timezone.utc).replace(tzinfo=None) def load_alert(alert_id): @@ -247,10 +245,20 @@ def load_alert(alert_id): def replay_alert(alert_id, args, admin): zf, alert = load_alert(alert_id) rounds = build_rounds(alert) - offset = day_offset(rounds, args.date) - last = max(f["recorded_at"] for r in rounds for f in r) + offset - if args.mode == "demo" and last > datetime.now(timezone.utc).replace(tzinfo=None): - print(f"warning: alert {alert_id} ends at {last} UTC, later than now") + # One shift for all frames keeps the gaps between them as in prod. + frames = [f for r in rounds for f in r] + first = min(f["recorded_at"] for f in frames) + latest_start = max( + min(f["recorded_at"] for f in frames if f["stream"] == s) + for s in {f["stream"] for f in frames} + ) + now = datetime.now(timezone.utc).replace(tzinfo=None) + # Default: the latest sequence starts 1 hour ago, so every sequence starts in the past. + offset = args.start - first if args.start else now - timedelta(hours=1) - latest_start + if args.mode == "demo" and latest_start + offset > now: + # The API sets last_seen_at to server time: a future started_at is out of the + # triangulation time window. + print(f"warning: alert {alert_id} has a sequence starting in the future") api = args.api_url local_cams = {c["name"]: c["id"] for c in get(api, "/api/v1/cameras/", admin)} @@ -328,7 +336,9 @@ def main(): p = sub.add_parser("replay") p.add_argument("alert_ids", nargs="+", type=int) p.add_argument("--mode", choices=("live", "demo"), default="demo") - p.add_argument("--date", choices=("today", "original"), default="today") + p.add_argument( + "--start", type=utc_start, help="demo mode, first frame time, e.g. 2026-07-10T18:53" + ) p.add_argument("--interval", type=float, default=30, help="live mode, seconds") p.add_argument("--api-url", default="http://localhost:5050") p.add_argument("--org-id", type=int, default=2, help="org of created cameras") From 54a0a89012b50e1f23257c26b6d8bf9a0518ac6c Mon Sep 17 00:00:00 2001 From: Mateo Date: Mon, 28 Sep 2026 19:08:00 +0200 Subject: [PATCH 09/14] fix: replay only the alert's own boxes others_bboxes belong to sequences outside the alert, which prod may have rejected (temporal model) but the local stack validates by default, creating extra alerts. They also had no crop, so their frames were sent without crops. --- scripts/replay_alerts.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index 385bf28..91f640a 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -201,6 +201,9 @@ def build_rounds(alert): A prod upload with several boxes yields one detection per box plus continuity rows, all sharing the image: they are replayed as a single upload. Round k holds the k-th frame of every (camera, azimuth) stream, like parallel real cameras. + Only the alert's own boxes are replayed: `others_bboxes` belong to sequences + outside the alert, which prod may have rejected but the local stack would + validate (no temporal model) into extra alerts. """ frames = {} for seq in alert["sequences"]: @@ -212,7 +215,7 @@ def build_rounds(alert): ) frame["recorded_at"] = datetime.fromisoformat(det["recorded_at"]) own = parse_bboxes(det["bbox"]) - for box in own + parse_bboxes(det["others_bboxes"]): + for box in own: if box not in frame["boxes"]: frame["boxes"].append(box) # A prod detection holds one box and the crop of that box. From 77ca880e0a0da93e065c3db3e4c74324e1afa6a3 Mon Sep 17 00:00:00 2001 From: Mateo Date: Tue, 29 Sep 2026 08:46:39 +0200 Subject: [PATCH 10/14] feat: demo replays write the prod alert straight into the DB Replaying through the API recomputed validation, triangulation and merges, so results depended on validation order and replays sharing cameras mixed. Demo mode now copies the alert as is: - images and crops go to the org bucket under replay-specific keys (the API deletes crops per detection, so replays must not share objects), and are removed again if the replay fails - alert, sequences, links and detections are inserted in one transaction, with prod azimuths, cones and location, shifted as a block in time - sequences are left unlabeled so the alert shows as live - replays reject cameras spanning several organizations Live mode still goes through the API. README documents both modes and the published alerts. --- README.md | 31 +++-- scripts/replay_alerts.py | 245 ++++++++++++++++++++++++++++++++------- 2 files changed, 223 insertions(+), 53 deletions(-) diff --git a/README.md b/README.md index 781ece4..c2ae656 100644 --- a/README.md +++ b/README.md @@ -126,16 +126,31 @@ GitHub release. Needs [uv](https://docs.astral.sh/uv/) (dependencies are install ```bash make list-alerts # alerts available in the release -make replay-alerts ALERTS="53800 53941" # demo mode (default) -make replay-alerts ALERTS=53800 START=2026-07-10T18:53 # demo, first frame at a given local time -make replay-alerts ALERTS=53800 MODE=live # live mode +make replay-alerts ALERTS="49767 54194" # demo mode (default) +make replay-alerts ALERTS=49767 START=2026-09-28T18:53 # demo, first sequence at a local time +make replay-alerts ALERTS=49767 MODE=live # live mode ``` -* **demo**: posts everything at once, shifted as a block (gaps between frames kept) - so its latest sequence starts 1 hour ago, or its first frame is at `START` (local - time). A sequence starting in the future does not triangulate: the API dates - `last_seen_at` with the server time. -* **live**: sends one frame per camera every 30 s (`--interval`), dated now. +Log in to the frontend as `test77` / `test` to see them. + +Published alerts (`make list-alerts` for the up-to-date list): + +| Alert | Prod date | Cameras | Detections | +|---|---|---|---| +| 49767 | 2026-07-10 | croix-augas-01, nemours-01, nemours-02 | 267 | +| 54194 | 2026-09-24 | moret-sur-loing-01, croix-augas-02, nemours-01, nemours-02, videlles-01 | 488 | + +* **demo**: copies the prod alert as is: images and crops go to the organization + bucket, alert, sequences and detections are written straight into the DB, with prod + azimuths, cones and location. Times are shifted as a block so the latest sequence + starts 1 hour ago, or the first one at `START`. Nothing is recomputed, so replays + never mix and it takes seconds. Each replay gets its own copy of the images. + Sequences are left unlabeled so the alert shows as live. Tied to the pyro-api DB + schema. +* **live**: posts one frame per camera every 30 s (`--interval`) through the API, + dated now, so validation and triangulation run locally. Replays within 2 hours of + each other get triangulated together when their cones cross: use a fresh stack + (`make stop && make run`) per alert. * Missing cameras are created by name in organization 2 (`--org-id`), so the alerts show up for the `test77` user. diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index 91f640a..dd83a18 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -1,7 +1,7 @@ #!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.9" -# dependencies = ["requests", "python-dotenv"] +# dependencies = ["requests", "python-dotenv", "boto3", "psycopg[binary]"] # /// """Fetch real alerts from the production alert API, share them through a GitHub release, and replay them on the local stack. @@ -10,11 +10,12 @@ (needs the admin DISTANT_* credentials in .env) publish ALERT_ID... upload the zips to the `replay-alerts` GitHub release (needs gh) list list the alerts available in the release - replay ALERT_ID... replay alerts on the local API, one after the other - --mode live one frame per camera every --interval seconds, timestamped now - --mode demo post everything at once, shifted as a block: the first frame at - --start (local time), or by default the latest sequence - starting 1 hour ago + replay ALERT_ID... replay alerts on the local stack, one after the other + --mode demo copy the prod alert as is (images to the bucket, rows into the + DB), shifted as a block: the first sequence starts at --start + (local time), or by default the latest one starts 1 hour ago + --mode live post one frame per camera every --interval seconds through the + API, timestamped now: validation and triangulation run locally Examples: uv run scripts/replay_alerts.py fetch 54095 @@ -29,11 +30,14 @@ import subprocess import sys import time +import uuid import zipfile from datetime import datetime, timedelta, timezone from itertools import zip_longest from pathlib import Path +import boto3 +import psycopg import requests from dotenv import load_dotenv @@ -152,8 +156,16 @@ def publish(alert_ids): if subprocess.run([*gh, "view", RELEASE_TAG], capture_output=True).returncode: notes = "Real alerts for scripts/replay_alerts.py" subprocess.run( - [*gh, "create", RELEASE_TAG, "--title", "Replay alerts", "--notes", notes, - "--latest=false"], + [ + *gh, + "create", + RELEASE_TAG, + "--title", + "Replay alerts", + "--notes", + notes, + "--latest=false", + ], check=True, ) subprocess.run([*gh, "upload", RELEASE_TAG, *files, "--clobber"], check=True) @@ -245,27 +257,14 @@ def load_alert(alert_id): return zf, json.loads(zf.read("alert.json")) -def replay_alert(alert_id, args, admin): - zf, alert = load_alert(alert_id) - rounds = build_rounds(alert) - # One shift for all frames keeps the gaps between them as in prod. - frames = [f for r in rounds for f in r] - first = min(f["recorded_at"] for f in frames) - latest_start = max( - min(f["recorded_at"] for f in frames if f["stream"] == s) - for s in {f["stream"] for f in frames} - ) - now = datetime.now(timezone.utc).replace(tzinfo=None) - # Default: the latest sequence starts 1 hour ago, so every sequence starts in the past. - offset = args.start - first if args.start else now - timedelta(hours=1) - latest_start - if args.mode == "demo" and latest_start + offset > now: - # The API sets last_seen_at to server time: a future started_at is out of the - # triangulation time window. - print(f"warning: alert {alert_id} has a sequence starting in the future") +def setup_cameras(alert, streams, args, admin): + """Find or create the alert's cameras locally, then a fresh pose per stream. + Returns {prod camera id: local camera}, {stream: local pose id}. + """ api = args.api_url - local_cams = {c["name"]: c["id"] for c in get(api, "/api/v1/cameras/", admin)} - tokens, poses = {}, {} + local_cams = {c["name"]: c for c in get(api, "/api/v1/cameras/", admin)} + cams = {} for prod_id, cam in alert["cameras"].items(): name = cam["name"] if name not in local_cams: @@ -274,37 +273,56 @@ def replay_alert(alert_id, args, admin): f"{api}/api/v1/cameras/", headers=admin, json=payload, timeout=30 ) r.raise_for_status() - local_cams[name] = r.json()["id"] - print(f"created camera {name} (id {local_cams[name]})") - r = requests.post( - f"{api}/api/v1/cameras/{local_cams[name]}/token", headers=admin, timeout=30 - ) - r.raise_for_status() - tokens[int(prod_id)] = {"Authorization": f"Bearer {r.json()['access_token']}"} + local_cams[name] = r.json() + print(f"created camera {name} (id {local_cams[name]['id']})") + cams[int(prod_id)] = local_cams[name] + # Images are stored and served per organization bucket: one org per alert. + orgs = {c["name"]: c["organization_id"] for c in cams.values()} + if len(set(orgs.values())) > 1: + sys.exit(f"alert cameras span several organizations: {orgs}") # A fresh pose per stream keeps this replay apart from earlier ones. - for prod_cam_id, azimuth in {f["stream"] for r in rounds for f in r}: - name = alert["cameras"][str(prod_cam_id)]["name"] - payload = {"camera_id": local_cams[name], "azimuth": azimuth} + poses = {} + for prod_cam_id, azimuth in streams: + payload = {"camera_id": cams[prod_cam_id]["id"], "azimuth": azimuth} r = requests.post( f"{api}/api/v1/poses/", headers=admin, json=payload, timeout=30 ) r.raise_for_status() poses[(prod_cam_id, azimuth)] = r.json()["id"] + return cams, poses + + +def replay_live(alert_id, args, admin): + """Post frames through the API, timestamped now: exercises the whole pipeline.""" + zf, alert = load_alert(alert_id) + rounds = build_rounds(alert) + cams, poses = setup_cameras( + alert, {f["stream"] for r in rounds for f in r}, args, admin + ) + tokens = {} + for prod_id, cam in cams.items(): + r = requests.post( + f"{args.api_url}/api/v1/cameras/{cam['id']}/token", + headers=admin, + timeout=30, + ) + r.raise_for_status() + tokens[prod_id] = {"Authorization": f"Bearer {r.json()['access_token']}"} - print(f"alert {alert_id}: replaying {len(rounds)} rounds ({args.mode} mode)") + print(f"alert {alert_id}: replaying {len(rounds)} rounds (live mode)") failures = 0 for i, frames in enumerate(rounds): for f in frames: data = {"bboxes": fmt_bboxes(f["boxes"]), "pose_id": poses[f["stream"]]} - if args.mode == "demo": - data["recorded_at"] = (f["recorded_at"] + offset).isoformat() files = [("file", ("frame.jpg", zf.read(f["image"]), "image/jpeg"))] # The API wants one crop per box, in the same order, or none at all. crops = [f["crops"].get(b) for b in f["boxes"]] if crops and all(crops): - files += [("crop", ("crop.jpg", zf.read(c), "image/jpeg")) for c in crops] + files += [ + ("crop", ("crop.jpg", zf.read(c), "image/jpeg")) for c in crops + ] r = requests.post( - f"{api}/api/v1/detections/", + f"{args.api_url}/api/v1/detections/", headers=tokens[f["stream"][0]], data=data, files=files, @@ -314,16 +332,151 @@ def replay_alert(alert_id, args, admin): failures += 1 print(f" error on {f['image']}: {r.status_code} {r.text[:200]}") print(f" round {i + 1}/{len(rounds)} sent ({len(frames)} frames)") - if args.mode == "live" and i + 1 < len(rounds): + if i + 1 < len(rounds): time.sleep(args.interval) return failures +def replay_demo(alert_id, args, admin): + """Copy the prod alert as is: images to the org bucket, rows straight into the DB. + + No validation, triangulation nor merge runs: the alert keeps its prod sequences and + location, and never mixes with other replays. Tied to the pyro-api DB schema. + """ + zf, alert = load_alert(alert_id) + seqs = alert["sequences"] + started = [datetime.fromisoformat(s["started_at"]) for s in seqs] + # One shift for all rows keeps the prod timeline. Default: the latest sequence + # starts 1 hour ago. + now = datetime.now(timezone.utc).replace(tzinfo=None) + offset = ( + args.start - min(started) + if args.start + else now - timedelta(hours=1) - max(started) + ) + streams = {(s["camera_id"], s["camera_azimuth"]) for s in seqs} + cams, poses = setup_cameras(alert, streams, args, admin) + org_id = next(iter(cams.values()))["organization_id"] + + s3 = boto3.client( + "s3", + endpoint_url=os.environ.get("S3_PROXY_URL") or "http://localhost:9000", + aws_access_key_id=os.environ["S3_ACCESS_KEY"], + aws_secret_access_key=os.environ["S3_SECRET_KEY"], + region_name=os.environ.get("S3_REGION"), + ) + bucket = f"{os.environ['SERVER_NAME']}-alert-api-{org_id}" + # Keys unique to this replay: the API deletes a crop along with its detection, so + # replays must not share objects, and a failed replay can drop its own safely. + tag = uuid.uuid4().hex[:8] + keys = { + name: f"replay-{tag}-{Path(name).name}" + for s in seqs + for d in s["detections"] + for name in (d["image"], d.get("crop")) + if name + } + uploaded = [] + try: + for name, key in keys.items(): + s3.put_object(Bucket=bucket, Key=key, Body=zf.read(name)) + uploaded.append(key) + alert_id_local = insert_alert(alert, cams, poses, org_id, keys, offset) + except BaseException: + for key in uploaded: + s3.delete_object(Bucket=bucket, Key=key) + raise + n_dets = sum(len(s["detections"]) for s in seqs) + print( + f"alert {alert_id} -> local alert {alert_id_local}: {len(seqs)} sequences, " + f"{n_dets} detections, {len(keys)} images and crops, " + f"starting {min(started) + offset} UTC" + ) + return 0 + + +def insert_alert(alert, cams, poses, org_id, keys, offset): + """Write the alert, its sequences and detections in one transaction.""" + + def shift(ts): + return datetime.fromisoformat(ts) + offset + + seqs = alert["sequences"] + dsn = ( + f"postgresql://{os.environ['POSTGRES_USER']}:{os.environ['POSTGRES_PASSWORD']}" + f"@localhost:5432/{os.environ['POSTGRES_DB']}" + ) + with psycopg.connect(dsn) as conn: # one transaction, rolled back on error + alert_id_local = conn.execute( + "INSERT INTO alerts (organization_id, lat, lon, started_at, last_seen_at)" + " VALUES (%s, %s, %s, %s, %s) RETURNING id", + ( + org_id, + alert["lat"], + alert["lon"], + shift(alert["started_at"]), + shift(alert["last_seen_at"]), + ), + ).fetchone()[0] + for s in seqs: + cam_id = cams[s["camera_id"]]["id"] + pose_id = poses[(s["camera_id"], s["camera_azimuth"])] + seq_id = conn.execute( + # is_wildfire is left unset: the prod label came afterwards, and a + # labeled alert is not listed as live in the frontend. + "INSERT INTO sequences (camera_id, pose_id, camera_azimuth," + " sequence_azimuth, cone_angle, started_at, last_seen_at, max_conf," + " temporal_model_score, temporal_model_version, temporal_api_version," + " is_validated) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" + " RETURNING id", + ( + cam_id, + pose_id, + s["camera_azimuth"], + s["sequence_azimuth"], + s["cone_angle"], + shift(s["started_at"]), + shift(s["last_seen_at"]), + s.get("max_conf"), + s.get("temporal_model_score"), + s.get("temporal_model_version"), + s.get("temporal_api_version"), + s["is_validated"], + ), + ).fetchone()[0] + conn.execute( + "INSERT INTO alerts_sequences (alert_id, sequence_id) VALUES (%s, %s)", + (alert_id_local, seq_id), + ) + with conn.cursor() as cur: + cur.executemany( + "INSERT INTO detections (camera_id, pose_id, sequence_id, bucket_key," + " crop_bucket_key, bbox, others_bboxes, created_at, recorded_at)" + " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)", + [ + ( + cam_id, + pose_id, + seq_id, + keys[d["image"]], + keys.get(d.get("crop")), + d["bbox"], + d["others_bboxes"], + shift(d["recorded_at"]), + shift(d["recorded_at"]), + ) + for d in s["detections"] + ], + ) + return alert_id_local + + def replay(args): admin = login( args.api_url, os.environ["SUPERADMIN_LOGIN"], os.environ["SUPERADMIN_PWD"] ) - failures = sum(replay_alert(a, args, admin) for a in args.alert_ids) + run = replay_demo if args.mode == "demo" else replay_live + failures = sum(run(a, args, admin) for a in args.alert_ids) if failures: sys.exit(f"{failures} detection uploads failed") @@ -340,7 +493,9 @@ def main(): p.add_argument("alert_ids", nargs="+", type=int) p.add_argument("--mode", choices=("live", "demo"), default="demo") p.add_argument( - "--start", type=utc_start, help="demo mode, first frame time, e.g. 2026-07-10T18:53" + "--start", + type=utc_start, + help="demo mode, first frame time, e.g. 2026-07-10T18:53", ) p.add_argument("--interval", type=float, default=30, help="live mode, seconds") p.add_argument("--api-url", default="http://localhost:5050") From 73927ef5e0e9111531780a4d5f18e001c1166a5a Mon Sep 17 00:00:00 2001 From: Mateo Date: Tue, 29 Sep 2026 10:57:59 +0200 Subject: [PATCH 11/14] fix: make pre-commit and tests pass for the replay script - test the time shift through time_offset instead of the removed day_offset - wrap a line ruff flagged as too long - pin psycopg for the tests and bump requests past the version safety flags - point make help at published alerts --- Makefile | 4 ++-- scripts/replay_alerts.py | 23 +++++++++++++---------- tests/requirements.txt | 3 ++- tests/test_replay_alerts.py | 28 +++++++++++++--------------- 4 files changed, 30 insertions(+), 28 deletions(-) diff --git a/Makefile b/Makefile index 35057e8..d386bee 100644 --- a/Makefile +++ b/Makefile @@ -20,10 +20,10 @@ help: @echo " ps Show compose status" @echo " logs Follow logs" @echo " test Run pytest" - @echo " fetch-alerts Download prod alerts: make fetch-alerts ALERTS=\"54095 53800\"" + @echo " fetch-alerts Download prod alerts: make fetch-alerts ALERTS=54194" @echo " publish-alerts Upload fetched alerts to the replay-alerts GitHub release" @echo " list-alerts List the alerts available in the release" - @echo " replay-alerts Replay alerts locally: make replay-alerts ALERTS=53800 MODE=live" + @echo " replay-alerts Replay alerts locally: make replay-alerts ALERTS=\"49767 54194\" [MODE=live] [START=...]" # ------------------------------------------------------------------- # Init diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index dd83a18..59180fd 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -244,6 +244,15 @@ def utc_start(s): return datetime.fromisoformat(s).astimezone(timezone.utc).replace(tzinfo=None) +def time_offset(starts, start=None, now=None): + """One shift for all rows, keeping the prod timeline: the first sequence starts at + `start`, or by default the latest one starts 1 hour ago.""" + if start: + return start - min(starts) + now = now or datetime.now(timezone.utc).replace(tzinfo=None) + return now - timedelta(hours=1) - max(starts) + + def load_alert(alert_id): path = zip_path(alert_id) if not path.is_file(): @@ -346,14 +355,7 @@ def replay_demo(alert_id, args, admin): zf, alert = load_alert(alert_id) seqs = alert["sequences"] started = [datetime.fromisoformat(s["started_at"]) for s in seqs] - # One shift for all rows keeps the prod timeline. Default: the latest sequence - # starts 1 hour ago. - now = datetime.now(timezone.utc).replace(tzinfo=None) - offset = ( - args.start - min(started) - if args.start - else now - timedelta(hours=1) - max(started) - ) + offset = time_offset(started, args.start) streams = {(s["camera_id"], s["camera_azimuth"]) for s in seqs} cams, poses = setup_cameras(alert, streams, args, admin) org_id = next(iter(cams.values()))["organization_id"] @@ -450,8 +452,9 @@ def shift(ts): ) with conn.cursor() as cur: cur.executemany( - "INSERT INTO detections (camera_id, pose_id, sequence_id, bucket_key," - " crop_bucket_key, bbox, others_bboxes, created_at, recorded_at)" + "INSERT INTO detections (camera_id, pose_id, sequence_id," + " bucket_key, crop_bucket_key, bbox, others_bboxes, created_at," + " recorded_at)" " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)", [ ( diff --git a/tests/requirements.txt b/tests/requirements.txt index e3ac0ea..26ec701 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -2,4 +2,5 @@ psycopg2-binary==2.9.9 pytest==9.0.3 python-dotenv==1.2.2 boto3==1.34.90 -requests==2.32.3 +requests==2.34.2 +psycopg[binary]==3.2.10 diff --git a/tests/test_replay_alerts.py b/tests/test_replay_alerts.py index d37fa9e..581da0e 100644 --- a/tests/test_replay_alerts.py +++ b/tests/test_replay_alerts.py @@ -1,14 +1,14 @@ import sys -from datetime import date, timedelta +from datetime import datetime, timedelta from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) from replay_alerts import ( # noqa: E402 build_rounds, - day_offset, fmt_bboxes, parse_bboxes, + time_offset, ) @@ -60,16 +60,14 @@ def test_build_rounds_merges_shared_images_and_interleaves_cameras(): assert rounds[0][1]["boxes"] == [] -def test_day_offset(): - alert = { - "sequences": [ - { - "camera_id": 1, - "camera_azimuth": 0.0, - "detections": [det("2026-01-01T23:59:00", "a", "[]")], - } - ] - } - rounds = build_rounds(alert) - assert day_offset(rounds, "original") == timedelta(0) - assert day_offset(rounds, "today") == date.today() - date(2026, 1, 1) +def test_time_offset(): + starts = [datetime(2026, 1, 1, 10), datetime(2026, 1, 1, 11)] + # --start places the first sequence start + assert time_offset(starts, datetime(2026, 5, 1, 8)) == datetime( + 2026, 5, 1, 8 + ) - datetime(2026, 1, 1, 10) + # default: the latest sequence starts 1 hour before now + now = datetime(2026, 6, 1, 12) + offset = time_offset(starts, now=now) + assert starts[1] + offset == now - timedelta(hours=1) + assert starts[0] + offset == now - timedelta(hours=2) From 9aba0a7cf0b5aa7728bfb74bc18554e8eb3fa1a1 Mon Sep 17 00:00:00 2001 From: Mateo Date: Tue, 29 Sep 2026 11:00:27 +0200 Subject: [PATCH 12/14] feat: add a script to refresh the alert cameras' last image and ping update_cameras.py sets each camera of the published alerts (or the given ones) to its most recent image in those alerts, then sends a heartbeat, as a real camera would. It reuses the replay helpers, split out for it (find_cameras, camera_token, published_alerts). Exposed as make update-cameras. --- Makefile | 4 +++ README.md | 5 +++ scripts/replay_alerts.py | 53 +++++++++++++++++++--------- scripts/update_cameras.py | 73 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 118 insertions(+), 17 deletions(-) create mode 100755 scripts/update_cameras.py diff --git a/Makefile b/Makefile index d386bee..88bc5de 100644 --- a/Makefile +++ b/Makefile @@ -23,6 +23,7 @@ help: @echo " fetch-alerts Download prod alerts: make fetch-alerts ALERTS=54194" @echo " publish-alerts Upload fetched alerts to the replay-alerts GitHub release" @echo " list-alerts List the alerts available in the release" + @echo " update-cameras Set last image and ping of the alert cameras: make update-cameras [ALERTS=49767]" @echo " replay-alerts Replay alerts locally: make replay-alerts ALERTS=\"49767 54194\" [MODE=live] [START=...]" # ------------------------------------------------------------------- @@ -126,3 +127,6 @@ list-alerts: replay-alerts: uv run scripts/replay_alerts.py replay $(ALERTS) --mode $(MODE) $(if $(START),--start $(START)) + +update-cameras: + uv run scripts/update_cameras.py $(ALERTS) diff --git a/README.md b/README.md index c2ae656..6ae1ff0 100644 --- a/README.md +++ b/README.md @@ -154,6 +154,11 @@ Published alerts (`make list-alerts` for the up-to-date list): * Missing cameras are created by name in organization 2 (`--org-id`), so the alerts show up for the `test77` user. +To make the alert cameras look alive, `make update-cameras` sets each camera's last +image to its most recent image in the published alerts (or `ALERTS="..."`) and sends +a heartbeat, as a real camera would. The ping goes stale like a real one: run it again +before a demo. + Add new alerts (needs prod admin creds `DISTANT_*` in `.env`, and `gh` for publishing): ```bash diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index 59180fd..8c51e6b 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -171,17 +171,23 @@ def publish(alert_ids): subprocess.run([*gh, "upload", RELEASE_TAG, *files, "--clobber"], check=True) -def list_alerts(): +def published_alerts(): r = requests.get( f"https://api.github.com/repos/{GITHUB_REPO}/releases/tags/{RELEASE_TAG}", timeout=30, ) if r.status_code == 404: - print("no alerts published") - return + return [] r.raise_for_status() - for asset in r.json()["assets"]: - print(asset["name"].removeprefix("alert_").removesuffix(".zip")) + return [ + int(a["name"].removeprefix("alert_").removesuffix(".zip")) + for a in r.json()["assets"] + ] + + +def list_alerts(): + ids = published_alerts() + print("\n".join(map(str, ids)) if ids else "no alerts published") # ------------------------------------------------------------------- @@ -266,10 +272,10 @@ def load_alert(alert_id): return zf, json.loads(zf.read("alert.json")) -def setup_cameras(alert, streams, args, admin): - """Find or create the alert's cameras locally, then a fresh pose per stream. +def find_cameras(alert, args, admin): + """Find the alert's cameras locally by name, creating the missing ones. - Returns {prod camera id: local camera}, {stream: local pose id}. + Returns {prod camera id: local camera}. """ api = args.api_url local_cams = {c["name"]: c for c in get(api, "/api/v1/cameras/", admin)} @@ -285,6 +291,24 @@ def setup_cameras(alert, streams, args, admin): local_cams[name] = r.json() print(f"created camera {name} (id {local_cams[name]['id']})") cams[int(prod_id)] = local_cams[name] + return cams + + +def camera_token(api, camera_id, admin): + r = requests.post( + f"{api}/api/v1/cameras/{camera_id}/token", headers=admin, timeout=30 + ) + r.raise_for_status() + return {"Authorization": f"Bearer {r.json()['access_token']}"} + + +def setup_cameras(alert, streams, args, admin): + """Find or create the alert's cameras locally, then a fresh pose per stream. + + Returns {prod camera id: local camera}, {stream: local pose id}. + """ + api = args.api_url + cams = find_cameras(alert, args, admin) # Images are stored and served per organization bucket: one org per alert. orgs = {c["name"]: c["organization_id"] for c in cams.values()} if len(set(orgs.values())) > 1: @@ -308,15 +332,10 @@ def replay_live(alert_id, args, admin): cams, poses = setup_cameras( alert, {f["stream"] for r in rounds for f in r}, args, admin ) - tokens = {} - for prod_id, cam in cams.items(): - r = requests.post( - f"{args.api_url}/api/v1/cameras/{cam['id']}/token", - headers=admin, - timeout=30, - ) - r.raise_for_status() - tokens[prod_id] = {"Authorization": f"Bearer {r.json()['access_token']}"} + tokens = { + prod_id: camera_token(args.api_url, cam["id"], admin) + for prod_id, cam in cams.items() + } print(f"alert {alert_id}: replaying {len(rounds)} rounds (live mode)") failures = 0 diff --git a/scripts/update_cameras.py b/scripts/update_cameras.py new file mode 100755 index 0000000..6194671 --- /dev/null +++ b/scripts/update_cameras.py @@ -0,0 +1,73 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# dependencies = ["requests", "python-dotenv", "boto3", "psycopg[binary]"] +# /// +"""Refresh the local cameras seen in the replay alerts: last image and ping. + +For each camera of the given alerts (default: every published alert), upload its most +recent image from the alerts as the camera's last image, then send a heartbeat, as a +real camera would. Missing cameras are created like in replay_alerts.py. + +Examples: + uv run scripts/update_cameras.py + uv run scripts/update_cameras.py 49767 +""" + +import argparse +import os +from pathlib import Path + +import requests +from dotenv import load_dotenv + +from replay_alerts import ( + REPO_ROOT, + camera_token, + find_cameras, + load_alert, + login, + published_alerts, +) + + +def main(): + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("alert_ids", nargs="*", type=int) + parser.add_argument("--api-url", default="http://localhost:5050") + parser.add_argument("--org-id", type=int, default=2, help="org of created cameras") + args = parser.parse_args() + load_dotenv(REPO_ROOT / ".env") + api = args.api_url + admin = login(api, os.environ["SUPERADMIN_LOGIN"], os.environ["SUPERADMIN_PWD"]) + + # Latest image of each camera across the alerts: + # {name: (recorded_at, zip, image path, local camera id)} + latest = {} + for alert_id in args.alert_ids or published_alerts(): + zf, alert = load_alert(alert_id) + cams = find_cameras(alert, args, admin) + for s in alert["sequences"]: + cam = cams[s["camera_id"]] + for d in s["detections"]: + if d["recorded_at"] > latest.get(cam["name"], ("",))[0]: + latest[cam["name"]] = (d["recorded_at"], zf, d["image"], cam["id"]) + + for name, (recorded_at, zf, image, cam_id) in sorted(latest.items()): + token = camera_token(api, cam_id, admin) + r = requests.patch( + f"{api}/api/v1/cameras/image", + headers=token, + files={"file": (Path(image).name, zf.read(image), "image/jpeg")}, + timeout=60, + ) + r.raise_for_status() + r = requests.patch(f"{api}/api/v1/cameras/heartbeat", headers=token, timeout=30) + r.raise_for_status() + print(f"{name}: last image from {recorded_at}, pinged") + + +if __name__ == "__main__": + main() From 9c4a107ee2772811ed810c7e74dd2c08147fba57 Mon Sep 17 00:00:00 2001 From: Mateo Date: Tue, 29 Sep 2026 11:04:26 +0200 Subject: [PATCH 13/14] docs: explain how to test the replay scripts Add make test-replay to run the unit tests without the stack, and document the end-to-end check on the local stack. --- Makefile | 4 ++++ README.md | 13 +++++++++++++ 2 files changed, 17 insertions(+) diff --git a/Makefile b/Makefile index 88bc5de..89634f8 100644 --- a/Makefile +++ b/Makefile @@ -20,6 +20,7 @@ help: @echo " ps Show compose status" @echo " logs Follow logs" @echo " test Run pytest" + @echo " test-replay Unit tests of the replay scripts (no stack needed)" @echo " fetch-alerts Download prod alerts: make fetch-alerts ALERTS=54194" @echo " publish-alerts Upload fetched alerts to the replay-alerts GitHub release" @echo " list-alerts List the alerts available in the release" @@ -110,6 +111,9 @@ logs: test: pytest -s tests/* +test-replay: + uv run --with-requirements tests/requirements.txt pytest -q tests/test_replay_alerts.py + # ------------------------------------------------------------------- # Real alert replays (see scripts/replay_alerts.py) # ------------------------------------------------------------------- diff --git a/README.md b/README.md index 6ae1ff0..c1adc89 100644 --- a/README.md +++ b/README.md @@ -159,6 +159,19 @@ image to its most recent image in the published alerts (or `ALERTS="..."`) and s a heartbeat, as a real camera would. The ping goes stale like a real one: run it again before a demo. +To test the replay: + +```bash +make test-replay # unit tests (bboxes, frames, time shift), no stack needed +make run # then, end to end on the local stack: +make replay-alerts ALERTS="49767 54194" +make update-cameras +``` + +Log in at [http://localhost:8080](http://localhost:8080) as `test77` / `test`: both alerts +are listed as live, each at its prod location, with images and crops, and their +cameras have a recent image. `make stop && make run` starts again from an empty DB. + Add new alerts (needs prod admin creds `DISTANT_*` in `.env`, and `gh` for publishing): ```bash From c1aaddffae3d3836f5fa468c88ec9a3aefb74401 Mon Sep 17 00:00:00 2001 From: Mateo Date: Tue, 29 Sep 2026 11:16:11 +0200 Subject: [PATCH 14/14] fix: set up replay cameras and poses as in prod The seeded local poses match no prod pose, local camera positions are rounded with a fixed elevation, and every replay created a new pose. - fetch stores each camera's prod poses in the zip - replays and update-cameras set the prod position and elevation, reuse or create a local pose per prod pose and deactivate the others (seeded or left by older replays) - each sequence is attached to the local copy of its prod pose instead of a new pose per replay The published zips were fetched again with the poses. --- README.md | 15 +++--- scripts/replay_alerts.py | 111 +++++++++++++++++++++++++++++--------- scripts/update_cameras.py | 13 ++--- 3 files changed, 101 insertions(+), 38 deletions(-) diff --git a/README.md b/README.md index c1adc89..5b34258 100644 --- a/README.md +++ b/README.md @@ -151,12 +151,15 @@ Published alerts (`make list-alerts` for the up-to-date list): dated now, so validation and triangulation run locally. Replays within 2 hours of each other get triangulated together when their cones cross: use a fresh stack (`make stop && make run`) per alert. -* Missing cameras are created by name in organization 2 (`--org-id`), so the - alerts show up for the `test77` user. - -To make the alert cameras look alive, `make update-cameras` sets each camera's last -image to its most recent image in the published alerts (or `ALERTS="..."`) and sends -a heartbeat, as a real camera would. The ping goes stale like a real one: run it again +* Cameras are matched by name and set up as in prod (position, elevation, poses); + local poses that do not exist in prod (seeded or left by older replays) are + deactivated. Each sequence is attached to the local copy of its prod pose. Missing + cameras are created in organization 2 (`--org-id`), so the alerts show up for the + `test77` user. + +To make the alert cameras look alive, `make update-cameras` sets up each camera of the +published alerts (or `ALERTS="..."`) as in prod, sets its last image to its most +recent image in those alerts and sends a heartbeat, as a real camera would. The ping goes stale like a real one: run it again before a demo. To test the replay: diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py index 8c51e6b..7d1553a 100644 --- a/scripts/replay_alerts.py +++ b/scripts/replay_alerts.py @@ -46,6 +46,7 @@ GITHUB_REPO = "pyronear/pyro-envdev" RELEASE_TAG = "replay-alerts" CAMERA_FIELDS = ("name", "angle_of_view", "elevation", "lat", "lon") +POSE_FIELDS = ("id", "azimuth", "patrol_id") def login(api_url, username, password): @@ -102,6 +103,9 @@ def fetch(alert_ids): if cam_id not in cameras: cam = get(api_url, f"/api/v1/cameras/{cam_id}", headers) cameras[cam_id] = {k: cam[k] for k in CAMERA_FIELDS} + cameras[cam_id]["poses"] = [ + {k: p[k] for k in POSE_FIELDS} for p in cam["poses"] + ] seq["detections"] = [] offset = 0 while True: @@ -272,26 +276,74 @@ def load_alert(alert_id): return zf, json.loads(zf.read("alert.json")) -def find_cameras(alert, args, admin): - """Find the alert's cameras locally by name, creating the missing ones. +def sync_cameras(alert, args, admin): + """Make the alert's local cameras match prod: position, elevation and poses. - Returns {prod camera id: local camera}. + Missing cameras are created. Returns {prod camera id: local camera} and + {prod pose id: local pose id}. """ api = args.api_url local_cams = {c["name"]: c for c in get(api, "/api/v1/cameras/", admin)} - cams = {} + cams, pose_ids = {}, {} for prod_id, cam in alert["cameras"].items(): name = cam["name"] - if name not in local_cams: - payload = {**cam, "organization_id": args.org_id, "is_trustable": True} + local = local_cams.get(name) + if local is None: + payload = {k: cam[k] for k in CAMERA_FIELDS} + payload.update(organization_id=args.org_id, is_trustable=True) r = requests.post( f"{api}/api/v1/cameras/", headers=admin, json=payload, timeout=30 ) r.raise_for_status() - local_cams[name] = r.json() - print(f"created camera {name} (id {local_cams[name]['id']})") - cams[int(prod_id)] = local_cams[name] - return cams + local = {**r.json(), "poses": []} + print(f"created camera {name} (id {local['id']})") + elif any(local[k] != cam[k] for k in ("lat", "lon", "elevation")): + payload = {k: cam[k] for k in ("lat", "lon", "elevation")} + r = requests.patch( + f"{api}/api/v1/cameras/{local['id']}/location", + headers=admin, + json=payload, + timeout=30, + ) + r.raise_for_status() + print(f"updated {name} position to prod") + cams[int(prod_id)] = local + if "poses" in cam: # zips fetched before poses were stored have none + pose_ids.update(sync_poses(api, admin, local, cam["poses"])) + return cams, pose_ids + + +def sync_poses(api, admin, camera, prod_poses): + """Match the camera's active local poses to the prod ones: reuse the one at the + same azimuth, create the missing ones, deactivate the rest (seeded or left by + earlier replays; deactivated, not deleted, as old sequences reference them).""" + free = list(camera["poses"]) + ids = {} + for p in prod_poses: + pose = next((q for q in free if abs(q["azimuth"] - p["azimuth"]) < 0.05), None) + if pose is None: + payload = {"camera_id": camera["id"], **p} + del payload["id"] + r = requests.post( + f"{api}/api/v1/poses/", headers=admin, json=payload, timeout=30 + ) + r.raise_for_status() + pose = r.json() + else: + free.remove(pose) + if pose["patrol_id"] != p["patrol_id"]: + patch_pose(api, admin, pose["id"], patrol_id=p["patrol_id"]) + ids[p["id"]] = pose["id"] + for pose in free: + patch_pose(api, admin, pose["id"], active=False) + return ids + + +def patch_pose(api, admin, pose_id, **fields): + r = requests.patch( + f"{api}/api/v1/poses/{pose_id}", headers=admin, json=fields, timeout=30 + ) + r.raise_for_status() def camera_token(api, camera_id, admin): @@ -302,26 +354,36 @@ def camera_token(api, camera_id, admin): return {"Authorization": f"Bearer {r.json()['access_token']}"} -def setup_cameras(alert, streams, args, admin): - """Find or create the alert's cameras locally, then a fresh pose per stream. +def setup_cameras(alert, args, admin): + """Sync the alert's cameras with prod and pick the local pose of each stream. Returns {prod camera id: local camera}, {stream: local pose id}. """ api = args.api_url - cams = find_cameras(alert, args, admin) + cams, pose_ids = sync_cameras(alert, args, admin) # Images are stored and served per organization bucket: one org per alert. orgs = {c["name"]: c["organization_id"] for c in cams.values()} if len(set(orgs.values())) > 1: sys.exit(f"alert cameras span several organizations: {orgs}") - # A fresh pose per stream keeps this replay apart from earlier ones. + # Each sequence goes to the local copy of its prod pose. poses = {} - for prod_cam_id, azimuth in streams: - payload = {"camera_id": cams[prod_cam_id]["id"], "azimuth": azimuth} - r = requests.post( - f"{api}/api/v1/poses/", headers=admin, json=payload, timeout=30 - ) - r.raise_for_status() - poses[(prod_cam_id, azimuth)] = r.json()["id"] + for s in alert["sequences"]: + stream = (s["camera_id"], s["camera_azimuth"]) + if stream in poses: + continue + poses[stream] = pose_ids.get(s["pose_id"]) + if poses[stream] is None: + # Prod pose deleted since, or zip without poses: an inactive stand-in. + payload = { + "camera_id": cams[s["camera_id"]]["id"], + "azimuth": s["camera_azimuth"], + "active": False, + } + r = requests.post( + f"{api}/api/v1/poses/", headers=admin, json=payload, timeout=30 + ) + r.raise_for_status() + poses[stream] = r.json()["id"] return cams, poses @@ -329,9 +391,7 @@ def replay_live(alert_id, args, admin): """Post frames through the API, timestamped now: exercises the whole pipeline.""" zf, alert = load_alert(alert_id) rounds = build_rounds(alert) - cams, poses = setup_cameras( - alert, {f["stream"] for r in rounds for f in r}, args, admin - ) + cams, poses = setup_cameras(alert, args, admin) tokens = { prod_id: camera_token(args.api_url, cam["id"], admin) for prod_id, cam in cams.items() @@ -375,8 +435,7 @@ def replay_demo(alert_id, args, admin): seqs = alert["sequences"] started = [datetime.fromisoformat(s["started_at"]) for s in seqs] offset = time_offset(started, args.start) - streams = {(s["camera_id"], s["camera_azimuth"]) for s in seqs} - cams, poses = setup_cameras(alert, streams, args, admin) + cams, poses = setup_cameras(alert, args, admin) org_id = next(iter(cams.values()))["organization_id"] s3 = boto3.client( diff --git a/scripts/update_cameras.py b/scripts/update_cameras.py index 6194671..be7d3ae 100755 --- a/scripts/update_cameras.py +++ b/scripts/update_cameras.py @@ -3,11 +3,12 @@ # requires-python = ">=3.9" # dependencies = ["requests", "python-dotenv", "boto3", "psycopg[binary]"] # /// -"""Refresh the local cameras seen in the replay alerts: last image and ping. +"""Refresh the local cameras seen in the replay alerts: prod setup, last image, ping. -For each camera of the given alerts (default: every published alert), upload its most -recent image from the alerts as the camera's last image, then send a heartbeat, as a -real camera would. Missing cameras are created like in replay_alerts.py. +For each camera of the given alerts (default: every published alert), set its prod +position, elevation and poses, upload its most recent image from the alerts as the +camera's last image, then send a heartbeat, as a real camera would. Missing cameras +are created like in replay_alerts.py. Examples: uv run scripts/update_cameras.py @@ -24,10 +25,10 @@ from replay_alerts import ( REPO_ROOT, camera_token, - find_cameras, load_alert, login, published_alerts, + sync_cameras, ) @@ -48,7 +49,7 @@ def main(): latest = {} for alert_id in args.alert_ids or published_alerts(): zf, alert = load_alert(alert_id) - cams = find_cameras(alert, args, admin) + cams, _ = sync_cameras(alert, args, admin) for s in alert["sequences"]: cam = cams[s["camera_id"]] for d in s["detections"]: