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/.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/Makefile b/Makefile index 1d6526e..89634f8 100644 --- a/Makefile +++ b/Makefile @@ -20,6 +20,12 @@ 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" + @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=...]" # ------------------------------------------------------------------- # Init @@ -104,3 +110,27 @@ 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) +# ------------------------------------------------------------------- + +MODE ?= demo + +fetch-alerts: + uv run scripts/replay_alerts.py fetch $(ALERTS) + +publish-alerts: + uv run scripts/replay_alerts.py publish $(ALERTS) + +list-alerts: + uv run scripts/replay_alerts.py list + +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 70fab60..5b34258 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,75 @@ 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 [uv](https://docs.astral.sh/uv/) (dependencies are installed on the fly). + +```bash +make list-alerts # alerts available in the release +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 +``` + +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. +* 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: + +```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 +make fetch-alerts ALERTS="54095" # writes data/replay_alerts/alert_54095.zip +make publish-alerts ALERTS="54095" # uploads it to the release +``` + +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. -```python -API_URL = "http://api:5050" ### Update the last image for a camera 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}") diff --git a/scripts/replay_alerts.py b/scripts/replay_alerts.py new file mode 100644 index 0000000..7d1553a --- /dev/null +++ b/scripts/replay_alerts.py @@ -0,0 +1,598 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# 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. + + 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 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 + uv run scripts/replay_alerts.py publish 54095 + uv run scripts/replay_alerts.py replay 54095 --mode demo +""" + +import argparse +import ast +import json +import os +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 + +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") +POSE_FIELDS = ("id", "azimuth", "patrol_id") + + +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 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" + + +# ------------------------------------------------------------------- +# 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 = {} + # 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: + 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: + 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']}" + 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: + break + offset += 100 + 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, " + 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, + "--latest=false", + ], + check=True, + ) + subprocess.run([*gh, "upload", RELEASE_TAG, *files, "--clobber"], check=True) + + +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: + return [] + r.raise_for_status() + 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") + + +# ------------------------------------------------------------------- +# 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. + 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"]: + stream = (seq["camera_id"], seq["camera_azimuth"]) + for det in seq["detections"]: + frame = frames.setdefault( + det["image"], + {"stream": stream, "image": det["image"], "boxes": [], "crops": {}}, + ) + frame["recorded_at"] = datetime.fromisoformat(det["recorded_at"]) + own = parse_bboxes(det["bbox"]) + 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. + 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) + return [[f for f in r if f] for r in zip_longest(*streams.values())] + + +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 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(): + 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 sync_cameras(alert, args, admin): + """Make the alert's local cameras match prod: position, elevation and poses. + + 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, pose_ids = {}, {} + for prod_id, cam in alert["cameras"].items(): + name = cam["name"] + 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 = {**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): + 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, 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, 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}") + # Each sequence goes to the local copy of its prod pose. + poses = {} + 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 + + +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, args, admin) + 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 + for i, frames in enumerate(rounds): + for f in frames: + data = {"bboxes": fmt_bboxes(f["boxes"]), "pose_id": poses[f["stream"]]} + 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"{args.api_url}/api/v1/detections/", + headers=tokens[f["stream"][0]], + data=data, + files=files, + 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 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] + offset = time_offset(started, args.start) + cams, poses = setup_cameras(alert, 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"] + ) + 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") + + +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( + "--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") + 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/scripts/update_cameras.py b/scripts/update_cameras.py new file mode 100755 index 0000000..be7d3ae --- /dev/null +++ b/scripts/update_cameras.py @@ -0,0 +1,74 @@ +#!/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: prod setup, last image, ping. + +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 + 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, + load_alert, + login, + published_alerts, + sync_cameras, +) + + +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, _ = sync_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() diff --git a/tests/requirements.txt b/tests/requirements.txt index 4977cf9..26ec701 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -2,3 +2,5 @@ psycopg2-binary==2.9.9 pytest==9.0.3 python-dotenv==1.2.2 boto3==1.34.90 +requests==2.34.2 +psycopg[binary]==3.2.10 diff --git a/tests/test_replay_alerts.py b/tests/test_replay_alerts.py new file mode 100644 index 0000000..581da0e --- /dev/null +++ b/tests/test_replay_alerts.py @@ -0,0 +1,73 @@ +import sys +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, + fmt_bboxes, + parse_bboxes, + time_offset, +) + + +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_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)