From c7fd3720028f61c2cba3ecd44520064d42301397 Mon Sep 17 00:00:00 2001 From: F-G Fernandez <26927750+frgfm@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:18:58 +0200 Subject: [PATCH] perf(triangulation): keep API responsive during overlap work --- src/app/api/api_v1/endpoints/detections.py | 9 +++--- src/app/services/alerts.py | 4 +-- src/app/services/overlap.py | 8 +++++ src/tests/endpoints/test_detections.py | 10 +++--- src/tests/services/test_overlap.py | 36 +++++++++++++++++++++- 5 files changed, 56 insertions(+), 11 deletions(-) diff --git a/src/app/api/api_v1/endpoints/detections.py b/src/app/api/api_v1/endpoints/detections.py index 37bcee36..52d4ab5e 100644 --- a/src/app/api/api_v1/endpoints/detections.py +++ b/src/app/api/api_v1/endpoints/detections.py @@ -50,7 +50,7 @@ from app.schemas.login import TokenPayload from app.schemas.sequences import SequenceUpdate from app.services.cones import resolve_cone -from app.services.overlap import compute_overlap, haversine_km +from app.services.overlap import compute_overlap_async, haversine_km from app.services.sequence_confidence import max_conf_from_bboxes from app.services.storage import s3_service, upload_file from app.services.telemetry import telemetry_client @@ -233,13 +233,14 @@ def _build_overlap_records( return records -def _resolve_groups_and_locations( +async def _resolve_groups_and_locations( records: List[Dict[str, Any]], sequence_id: int, ) -> Optional[Tuple[List[Tuple[int, ...]], Dict[Tuple[int, ...], Optional[Tuple[float, float]]]]]: if not records: return None - row = next((row for row in compute_overlap(records) if row["id"] == int(sequence_id)), None) + result = await compute_overlap_async(records) + row = next((row for row in result if row["id"] == int(sequence_id)), None) if row is None: return None groups = [tuple(g) for g in row["event_groups"]] @@ -471,7 +472,7 @@ async def _attach_sequence_to_alert( # Build records for overlap computation records = _build_overlap_records(recent_sequences, camera_by_id) - resolved = _resolve_groups_and_locations(records, int(sequence_.id)) + resolved = await _resolve_groups_and_locations(records, int(sequence_.id)) if resolved is None: return None groups, group_locations = resolved diff --git a/src/app/services/alerts.py b/src/app/services/alerts.py index 952b71ac..3ef69933 100644 --- a/src/app/services/alerts.py +++ b/src/app/services/alerts.py @@ -12,7 +12,7 @@ from app.crud import AlertCRUD from app.models import AlertSequence, Camera, Sequence from app.schemas.alerts import AlertUpdate -from app.services.overlap import compute_overlap +from app.services.overlap import compute_overlap_async __all__ = ["refresh_alert_state"] @@ -51,7 +51,7 @@ async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Aler "started_at": seq.started_at, "last_seen_at": seq.last_seen_at, }) - result = compute_overlap(records) + result = await compute_overlap_async(records) loc = next((loc for row in result for loc in row["event_smoke_locations"] if loc is not None), None) await alerts.update( diff --git a/src/app/services/overlap.py b/src/app/services/overlap.py index 810e2bb1..284eb369 100644 --- a/src/app/services/overlap.py +++ b/src/app/services/overlap.py @@ -15,6 +15,7 @@ import networkx as nx import numpy as np +from anyio import CapacityLimiter, to_thread from pyproj import Geod, Transformer from shapely import transform as shapely_transform from shapely.geometry import Polygon @@ -28,6 +29,8 @@ _TO_MERCATOR = Transformer.from_crs("EPSG:4326", "EPSG:3857", always_xy=True) _TO_LATLON = Transformer.from_crs("EPSG:3857", "EPSG:4326", always_xy=True) _PairCentroids = Dict[Tuple[int, int], Optional[Tuple[float, float]]] +# Keep geometry jobs serial per process without blocking unrelated API requests. +_TRIANGULATION_LIMITER = CapacityLimiter(1) def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: @@ -415,6 +418,11 @@ def _attach_groups_to_records( row["event_smoke_locations"] = seq_to_smokes.get(row["id"], []) +async def compute_overlap_async(api_sequences: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Run overlap computation in a bounded worker thread.""" + return await to_thread.run_sync(compute_overlap, api_sequences, limiter=_TRIANGULATION_LIMITER) + + def compute_overlap( api_sequences: List[Dict[str, Any]], r_km: float = 35.0, diff --git a/src/tests/endpoints/test_detections.py b/src/tests/endpoints/test_detections.py index fb3a107c..ac0936ae 100644 --- a/src/tests/endpoints/test_detections.py +++ b/src/tests/endpoints/test_detections.py @@ -604,11 +604,13 @@ async def test_build_overlap_records_skips_missing_cone_data(detection_session: assert records == [] -def test_resolve_groups_and_locations_empty_records_returns_none(): - assert _resolve_groups_and_locations([], 1) is None +@pytest.mark.asyncio +async def test_resolve_groups_and_locations_empty_records_returns_none(): + assert await _resolve_groups_and_locations([], 1) is None -def test_resolve_groups_and_locations_no_match_returns_none(): +@pytest.mark.asyncio +async def test_resolve_groups_and_locations_no_match_returns_none(): now = utcnow() records = [ { @@ -622,7 +624,7 @@ def test_resolve_groups_and_locations_no_match_returns_none(): "last_seen_at": now, } ] - assert _resolve_groups_and_locations(records, 999) is None + assert await _resolve_groups_and_locations(records, 999) is None @pytest.mark.parametrize( diff --git a/src/tests/services/test_overlap.py b/src/tests/services/test_overlap.py index 27e7732e..9cc355ef 100644 --- a/src/tests/services/test_overlap.py +++ b/src/tests/services/test_overlap.py @@ -3,13 +3,16 @@ # This program is licensed under the Apache License 2.0. # See LICENSE or go to for full license details. +import asyncio from datetime import datetime, timedelta +from threading import Event, get_ident import pytest from shapely.geometry import box from app.core.time import utcnow -from app.services.overlap import _find_overlapping_pairs, compute_overlap +from app.services import overlap +from app.services.overlap import _find_overlapping_pairs, compute_overlap, compute_overlap_async @pytest.mark.parametrize("relaxation", [0, 30]) @@ -173,3 +176,34 @@ def test_compute_overlap_copies_records_and_handles_empty_input(label) -> None: assert result[0]["event_groups"] == [(1,)] assert result[0]["event_smoke_locations"] == ([None] if label is None else []) assert compute_overlap([]) == [] + + +@pytest.mark.asyncio +async def test_compute_overlap_async_keeps_loop_responsive_and_serializes(monkeypatch) -> None: + now = utcnow() + records = [_make_sequence(1, 48.3792, 2.8208, 276.5, 3.0, now, now)] + expected = compute_overlap(records) + entered, release = Event(), Event() + loop_thread = get_ident() + calls = [] + + def blocking_compute(data): + assert get_ident() != loop_thread + calls.append(data) + entered.set() + assert release.wait(5) + return compute_overlap(data) + + monkeypatch.setattr(overlap, "compute_overlap", blocking_compute) + tasks = [asyncio.create_task(compute_overlap_async(records))] + try: + assert await asyncio.to_thread(entered.wait, 1) + assert not tasks[0].done() + tasks.append(asyncio.create_task(compute_overlap_async(records))) + await asyncio.sleep(0.05) + assert calls == [records] + finally: + release.set() + results = await asyncio.gather(*tasks) + assert results == [expected, expected] + assert calls == [records, records]