diff --git a/src/app/api/api_v1/endpoints/alerts.py b/src/app/api/api_v1/endpoints/alerts.py index 2c6c0a60..2600e92e 100644 --- a/src/app/api/api_v1/endpoints/alerts.py +++ b/src/app/api/api_v1/endpoints/alerts.py @@ -4,15 +4,17 @@ # See LICENSE or go to for full license details. +import asyncio import csv import io -from datetime import date, datetime, time, timedelta -from typing import Any, AsyncIterator, Dict, Iterable, Iterator, List, Tuple, Union, cast +from datetime import date, datetime, time, timedelta, timezone +from typing import Any, AsyncIterator, Dict, Iterable, Iterator, List, Tuple, TypeVar, Union, cast +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from fastapi import APIRouter, Depends, HTTPException, Path, Query, Security, status from fastapi.responses import StreamingResponse from sqlalchemy import asc, desc, tuple_ -from sqlalchemy.sql import ColumnElement +from sqlalchemy.sql import ColumnElement, Select from sqlmodel import delete, func, select from sqlmodel.ext.asyncio.session import AsyncSession from sqlmodel.sql.expression import SelectOfScalar @@ -22,7 +24,7 @@ from app.crud import AlertCRUD, CameraCRUD, SequenceCRUD from app.db import get_session, session_factory from app.models import Alert, AlertSequence, AnnotationType, Camera, Sequence, UserRole -from app.schemas.alerts import AlertCount, AlertCreate, AlertReadWithSequences +from app.schemas.alerts import AlertCount, AlertCreate, AlertDayCount, AlertReadWithSequences from app.schemas.login import TokenPayload from app.schemas.sequences import SequenceRead from app.services.alerts import refresh_alert_state @@ -125,13 +127,23 @@ async def _build_alerts_from_date_stmt( alerts_stmt: Any = select(Alert).where(func.date(Alert.started_at) == from_date) if not is_admin: alerts_stmt = alerts_stmt.where(Alert.organization_id == organization_id) - if seq_filter is not None: - seq_match: Any = select(AlertSequence.alert_id).join( - Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id) - ) - seq_match = seq_match.where(seq_filter) - alerts_stmt = alerts_stmt.where(cast(Any, Alert.id).in_(seq_match)) - return alerts_stmt, seq_filter + return _filter_alerts_by_sequences(alerts_stmt, seq_filter), seq_filter + + +_SelectT = TypeVar("_SelectT", bound=Select) + + +def _filter_alerts_by_sequences(alerts_stmt: _SelectT, seq_filter: Union[ColumnElement[bool], None]) -> _SelectT: + """Keep the alerts with at least one sequence passing ``seq_filter`` (no-op when None).""" + if seq_filter is None: + return alerts_stmt + seq_match: Any = select(AlertSequence.alert_id).join(Sequence, cast(Any, Sequence.id == AlertSequence.sequence_id)) + return alerts_stmt.where(cast(Any, Alert.id).in_(seq_match.where(seq_filter))) + + +def _utc_start_of_day(day: date, zone: ZoneInfo) -> datetime: + """Start of ``day`` in ``zone``, as a naive UTC datetime like the stored timestamps.""" + return datetime.combine(day, time.min, tzinfo=zone).astimezone(timezone.utc).replace(tzinfo=None) def _serialize_sequence(sequence: Sequence, detections_count: int = 0) -> SequenceRead: @@ -458,6 +470,91 @@ async def count_alerts_from_date( return AlertCount(count=int((await session.exec(count_stmt)).one())) +# A month calendar shows at most 6 weeks: 62 days cover it, and bound the per-day risk lookups. +MAX_COUNT_PER_DAY_PERIOD = 62 + + +@router.get( + "/all/fromdate/count-per-day", + status_code=status.HTTP_200_OK, + summary="Count the alerts per day over a period", +) +async def count_alerts_per_day( + from_date: date = Query(..., description="First day of the period (inclusive), in `tz`"), + to_date: date = Query(..., description="Last day of the period (inclusive), in `tz`"), + tz: str = Query("UTC", description="IANA timezone used to split the days, e.g. Europe/Paris"), + risk_score: Union[FwiClass, None] = Query( + None, + description="Override FWI class applied to every sequence; bypasses risk-api lookup. Ignored for admins.", + ), + session: AsyncSession = Depends(get_session), + token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN, UserRole.AGENT, UserRole.USER]), +) -> List[AlertDayCount]: + if to_date < from_date: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="to_date must be on or after from_date", + ) + if (to_date - from_date).days + 1 > MAX_COUNT_PER_DAY_PERIOD: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"The period cannot exceed {MAX_COUNT_PER_DAY_PERIOD} days", + ) + try: + zone = ZoneInfo(tz) + except (ZoneInfoNotFoundError, ValueError): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Unknown timezone: {tz}" + ) from None + telemetry_client.capture( + token_payload.sub, + event="alerts-count-per-day", + properties={"from_date": from_date.isoformat(), "to_date": to_date.isoformat(), "tz": tz}, + ) + + # Days after today cannot have alerts yet: skip them, and their risk lookups + last_day = min(to_date, datetime.now(zone).date()) + days = [from_date + timedelta(days=offset) for offset in range((last_day - from_date).days + 1)] + if not days: + return [] + + # Same alerts as /all/fromdate/count, each day with its own risk scores: days sharing the + # same scores are counted by a single query. + days_by_scores: Dict[Tuple[Tuple[int, Union[str, None]], ...], List[date]] = {} + if token_payload.is_admin: + days_by_scores[()] = days + elif risk_score is not None: + class_per_camera = await _resolve_fwi_class_per_camera( + session, token_payload.organization_id, override_class=risk_score + ) + days_by_scores[tuple(sorted(class_per_camera.items()))] = days + else: + scores_per_day = await asyncio.gather( + *(risk_service.get_scores_for_date(day, organization_id=token_payload.organization_id) for day in days) + ) + for day, scores in zip(days, scores_per_day, strict=True): + days_by_scores.setdefault(tuple(sorted(scores.items())), []).append(day) + + counts: Dict[date, int] = {} + for scores_key, scores_days in days_by_scores.items(): + # The database returns the alerts of a plain UTC range of started_at (index-friendly), and + # they are counted per local day here: the timezone is only handled by zoneinfo. + stmt: Any = select(Alert.started_at).where( + Alert.started_at >= _utc_start_of_day(scores_days[0], zone), + Alert.started_at < _utc_start_of_day(scores_days[-1] + timedelta(days=1), zone), + ) + if not token_payload.is_admin: + stmt = stmt.where(Alert.organization_id == token_payload.organization_id) + stmt = _filter_alerts_by_sequences(stmt, max_conf_filter_clause(dict(scores_key))) + for started_at in (await session.exec(stmt)).all(): + # started_at is a naive UTC datetime: its day in the requested timezone + day = started_at.replace(tzinfo=timezone.utc).astimezone(zone).date() + # The range may include days counted by another query (other risk scores) + if day in scores_days: + counts[day] = counts.get(day, 0) + 1 + return [AlertDayCount(date=day, count=count) for day, count in sorted(counts.items())] + + @router.post( "/{alert_id}/sequences/{sequence_id}/unmatch", status_code=status.HTTP_200_OK, diff --git a/src/app/main.py b/src/app/main.py index b4bb7437..e950ed72 100644 --- a/src/app/main.py +++ b/src/app/main.py @@ -88,6 +88,7 @@ async def lifespan(_: FastAPI): with contextlib.suppress(asyncio.CancelledError): await task await temporal_service.aclose() + await risk_service.aclose() app = FastAPI( diff --git a/src/app/schemas/alerts.py b/src/app/schemas/alerts.py index 2be5c08e..36971e94 100644 --- a/src/app/schemas/alerts.py +++ b/src/app/schemas/alerts.py @@ -3,6 +3,7 @@ # This program is licensed under the Apache License 2.0. # See LICENSE or go to for full license details. +import datetime as dt from datetime import datetime from typing import List, Optional @@ -10,7 +11,15 @@ from app.schemas.sequences import SequenceRead -__all__ = ["AlertBase", "AlertCount", "AlertCreate", "AlertRead", "AlertReadWithSequences", "AlertUpdate"] +__all__ = [ + "AlertBase", + "AlertCount", + "AlertCreate", + "AlertDayCount", + "AlertRead", + "AlertReadWithSequences", + "AlertUpdate", +] class AlertBase(BaseModel): @@ -41,3 +50,9 @@ class AlertReadWithSequences(AlertRead): class AlertCount(BaseModel): count: int + + +class AlertDayCount(BaseModel): + # Module-qualified type: a field named `date` would shadow `datetime.date` + date: dt.date = Field(..., description="Day, in the requested timezone") + count: int = Field(..., ge=1, description="Number of alerts started that day") diff --git a/src/app/services/risk.py b/src/app/services/risk.py index deca342f..d3813578 100644 --- a/src/app/services/risk.py +++ b/src/app/services/risk.py @@ -3,6 +3,7 @@ # This program is licensed under the Apache License 2.0. # See LICENSE or go to for full license details. +import asyncio import logging from datetime import date from typing import Literal, Union, cast @@ -62,8 +63,19 @@ class RiskService: Fail-open: if the API is unreachable or a camera is unknown, no filter is applied. """ + # Bounds concurrent calls, e.g. one score lookup per day of a calendar month + MAX_CONCURRENT_REQUESTS: int = 8 + def __init__(self) -> None: self._scores: dict[int, str] = {} + self._client: Union[httpx.AsyncClient, None] = None + self._semaphore = asyncio.Semaphore(self.MAX_CONCURRENT_REQUESTS) + + async def aclose(self) -> None: + """Close the HTTP client on shutdown.""" + if self._client is not None: + await self._client.aclose() + self._client = None @property def is_configured(self) -> bool: @@ -89,14 +101,16 @@ async def _fetch(self, path: str, params: Union[dict, None] = None) -> object: return None host = risk_api_url.rstrip("/") try: - async with httpx.AsyncClient(timeout=5.0) as client: - response = await client.get( + async with self._semaphore: + if self._client is None: + self._client = httpx.AsyncClient(timeout=5.0) + response = await self._client.get( f"{host}/{path.lstrip('/')}", params=params, auth=(risk_api_login, risk_api_pwd), ) - response.raise_for_status() - return response.json() + response.raise_for_status() + return response.json() except (httpx.HTTPError, ValueError) as exc: logger.warning("Risk API call %s failed (%s)", path, exc) return None diff --git a/src/tests/endpoints/test_alerts.py b/src/tests/endpoints/test_alerts.py index 47c65434..59b352e8 100644 --- a/src/tests/endpoints/test_alerts.py +++ b/src/tests/endpoints/test_alerts.py @@ -6,8 +6,8 @@ import csv import io from datetime import datetime, timedelta -from typing import Any, Dict, List, Tuple, cast -from unittest.mock import AsyncMock +from typing import Any, Dict, List, Tuple, Union, cast +from unittest.mock import AsyncMock, patch import pytest # type: ignore from httpx import AsyncClient @@ -324,6 +324,134 @@ async def test_alerts_count_unauthenticated(async_client: AsyncClient): assert resp.status_code == 401, resp.text +async def _create_alert_started_at( + session: AsyncSession, org_id: int, started_at: datetime, camera_id: Union[int, None] = None, max_conf: float = 0.9 +) -> Alert: + alert = Alert(organization_id=org_id, started_at=started_at, last_seen_at=started_at + timedelta(minutes=5)) + session.add(alert) + await session.commit() + await session.refresh(alert) + if camera_id is not None: + seq = Sequence( + camera_id=camera_id, + pose_id=None, + camera_azimuth=180.0, + sequence_azimuth=175.0, + cone_angle=5.0, + is_wildfire=None, + started_at=started_at, + last_seen_at=started_at + timedelta(minutes=5), + max_conf=max_conf, + ) + session.add(seq) + await session.commit() + await session.refresh(seq) + session.add(AlertSequence(alert_id=alert.id, sequence_id=seq.id)) + await session.commit() + return alert + + +@pytest.mark.asyncio +async def test_alerts_count_per_day(async_client: AsyncClient, camera_session: AsyncSession): + admin_auth = pytest.get_token( + pytest.user_table[0]["id"], pytest.user_table[0]["role"].split(), pytest.user_table[0]["organization_id"] + ) + agent_auth = pytest.get_token( + pytest.user_table[1]["id"], pytest.user_table[1]["role"].split(), pytest.user_table[1]["organization_id"] + ) + day = datetime(2025, 7, 14) + await _create_alert_started_at(camera_session, 1, day + timedelta(hours=10)) + await _create_alert_started_at(camera_session, 1, day + timedelta(hours=23, minutes=30)) + await _create_alert_started_at(camera_session, 1, day + timedelta(days=2, hours=12)) + await _create_alert_started_at(camera_session, 2, day + timedelta(hours=10)) + period = "from_date=2025-07-14&to_date=2025-07-16" + + # Agent: its organization only, days without alerts are omitted + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=agent_auth) + assert resp.status_code == 200, resp.text + assert resp.json() == [{"date": "2025-07-14", "count": 2}, {"date": "2025-07-16", "count": 1}] + # Consistent with the single-date count + resp = await async_client.get("/alerts/all/fromdate/count?from_date=2025-07-14", headers=agent_auth) + assert resp.json() == {"count": 2} + + # Admin: every organization + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=admin_auth) + assert resp.json() == [{"date": "2025-07-14", "count": 3}, {"date": "2025-07-16", "count": 1}] + + # 23:30 UTC is already the next day in Paris (UTC+2 in July) + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}&tz=Europe/Paris", headers=agent_auth) + assert resp.json() == [ + {"date": "2025-07-14", "count": 1}, + {"date": "2025-07-15", "count": 1}, + {"date": "2025-07-16", "count": 1}, + ] + + resp = await async_client.get( + "/alerts/all/fromdate/count-per-day?from_date=2025-07-15&to_date=2025-07-15", headers=agent_auth + ) + assert resp.json() == [] + + +@pytest.mark.parametrize( + ("query", "status_code"), + [ + ("from_date=2025-07-16&to_date=2025-07-14", 422), + ("from_date=2025-07-01&to_date=2025-09-01", 422), # 63 days + ("from_date=2025-07-01&to_date=2025-08-31", 200), # 62 days + ("from_date=0001-01-01&to_date=9999-12-31", 422), # rejected before listing the days + ("from_date=2025-07-01&to_date=2025-07-31&tz=Mars/Olympus_Mons", 422), + ("from_date=2025-07-01", 422), + ], +) +@pytest.mark.asyncio +async def test_alerts_count_per_day_validation( + async_client: AsyncClient, camera_session: AsyncSession, query: str, status_code: int +): + agent_auth = pytest.get_token( + pytest.user_table[1]["id"], pytest.user_table[1]["role"].split(), pytest.user_table[1]["organization_id"] + ) + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{query}", headers=agent_auth) + assert resp.status_code == status_code, resp.text + + +@pytest.mark.asyncio +async def test_alerts_count_per_day_unauthenticated(async_client: AsyncClient): + resp = await async_client.get("/alerts/all/fromdate/count-per-day?from_date=2025-07-14&to_date=2025-07-16") + assert resp.status_code == 401, resp.text + + +@pytest.mark.asyncio +async def test_alerts_count_per_day_risk_scores_per_day(async_client: AsyncClient, camera_session: AsyncSession): + agent_auth = pytest.get_token( + pytest.user_table[1]["id"], pytest.user_table[1]["role"].split(), pytest.user_table[1]["organization_id"] + ) + camera_id = pytest.camera_table[0]["id"] + day = datetime(2025, 7, 14) + # Two low-confidence alerts, but only the first day is low risk (threshold above 0.2) + await _create_alert_started_at(camera_session, 1, day + timedelta(hours=10), camera_id, max_conf=0.2) + await _create_alert_started_at(camera_session, 1, day + timedelta(days=1, hours=10), camera_id, max_conf=0.2) + period = "from_date=2025-07-14&to_date=2025-07-15" + + def scores_for_date(target_date, organization_id=None): + return {camera_id: "low"} if target_date == day.date() else {} + + scores_mock = AsyncMock(side_effect=scores_for_date) + with patch("app.api.api_v1.endpoints.alerts.risk_service.get_scores_for_date", new=scores_mock): + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=agent_auth) + assert resp.status_code == 200, resp.text + assert resp.json() == [{"date": "2025-07-15", "count": 1}] + assert scores_mock.await_count == 2 + assert {call.kwargs["organization_id"] for call in scores_mock.await_args_list} == {1} + + # A risk score override applies to every day and skips the risk API + scores_mock.reset_mock() + resp = await async_client.get( + f"/alerts/all/fromdate/count-per-day?{period}&risk_score=moderate", headers=agent_auth + ) + assert resp.json() == [{"date": "2025-07-14", "count": 1}, {"date": "2025-07-15", "count": 1}] + scores_mock.assert_not_awaited() + + @pytest.mark.asyncio async def test_triangulation_creates_single_alert( async_client: AsyncClient, detection_session: AsyncSession, mock_img: bytes @@ -1027,3 +1155,34 @@ async def test_alert_changes_roll_back_on_late_failure( assert set(links) == set(sequence_ids) detections = (await observer.exec(select(Detection).where(Detection.sequence_id == sid))).all() assert len(detections) == counts[0] + + +@pytest.mark.asyncio +async def test_alerts_count_per_day_skips_future_days(async_client: AsyncClient, camera_session: AsyncSession): + agent_auth = pytest.get_token( + pytest.user_table[1]["id"], pytest.user_table[1]["role"].split(), pytest.user_table[1]["organization_id"] + ) + today = utcnow().date() + # Midnight today: always in the past and always today, whenever the test runs + await _create_alert_started_at(camera_session, 1, datetime.combine(today, datetime.min.time())) + period = f"from_date={(today - timedelta(days=2)).isoformat()}&to_date={(today + timedelta(days=10)).isoformat()}" + + scores_mock = AsyncMock(return_value={}) + with patch("app.api.api_v1.endpoints.alerts.risk_service.get_scores_for_date", new=scores_mock): + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{period}", headers=agent_auth) + assert resp.status_code == 200, resp.text + assert resp.json() == [{"date": today.isoformat(), "count": 1}] + # Risk scores are only fetched up to today + assert sorted(call.args[0] for call in scores_mock.await_args_list) == [ + today - timedelta(days=2), + today - timedelta(days=1), + today, + ] + + # A period entirely in the future is empty, without any lookup + scores_mock.reset_mock() + future = f"from_date={(today + timedelta(days=1)).isoformat()}&to_date={(today + timedelta(days=5)).isoformat()}" + with patch("app.api.api_v1.endpoints.alerts.risk_service.get_scores_for_date", new=scores_mock): + resp = await async_client.get(f"/alerts/all/fromdate/count-per-day?{future}", headers=agent_auth) + assert resp.json() == [] + scores_mock.assert_not_awaited() diff --git a/src/tests/services/test_risk.py b/src/tests/services/test_risk.py index 5f5085e2..22b3d639 100644 --- a/src/tests/services/test_risk.py +++ b/src/tests/services/test_risk.py @@ -3,6 +3,7 @@ # This program is licensed under the Apache License 2.0. # See LICENSE or go to for full license details. +import asyncio from datetime import date from unittest.mock import AsyncMock, MagicMock, patch @@ -14,7 +15,7 @@ def _fake_httpx_client(*, json_data=None, raise_exc=None, side_effect_per_call=None): - """Return a context manager replacing ``app.services.risk.httpx.AsyncClient``. + """Return a factory replacing ``app.services.risk.httpx.AsyncClient``, and the client it builds. Either pass ``json_data`` (single response), ``raise_exc`` (raised on get()), or ``side_effect_per_call`` (list of (json_or_exc) values returned in order). @@ -40,11 +41,9 @@ def _fake_httpx_client(*, json_data=None, raise_exc=None, side_effect_per_call=N else: inner.get = AsyncMock(return_value=response) - cm = MagicMock() - cm.__aenter__ = AsyncMock(return_value=inner) - cm.__aexit__ = AsyncMock(return_value=None) + inner.aclose = AsyncMock() - factory = MagicMock(return_value=cm) + factory = MagicMock(return_value=inner) return factory, inner @@ -176,3 +175,39 @@ async def test_get_scores_for_date_no_org_param_when_none(configured_risk): await service.get_scores_for_date(date(2026, 5, 5)) _, kwargs = inner.get.await_args assert kwargs["params"] is None + + +@pytest.mark.asyncio +async def test_risk_client_is_shared_and_closed(configured_risk): + service = RiskService() + factory, inner = _fake_httpx_client(json_data=[]) + with patch("app.services.risk.httpx.AsyncClient", factory): + await service.get_scores_for_date(date(2026, 5, 5)) + await service.get_scores_for_date(date(2026, 5, 6)) + await service.aclose() + factory.assert_called_once() + assert inner.get.await_count == 2 + inner.aclose.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_risk_concurrent_requests_are_bounded(configured_risk): + service = RiskService() + factory, inner = _fake_httpx_client(json_data=[]) + in_flight, peak = 0, 0 + + async def slow_get(*args, **kwargs): + nonlocal in_flight, peak + in_flight += 1 + peak = max(peak, in_flight) + await asyncio.sleep(0.01) + in_flight -= 1 + response = MagicMock() + response.json = MagicMock(return_value=[]) + return response + + inner.get = AsyncMock(side_effect=slow_get) + with patch("app.services.risk.httpx.AsyncClient", factory): + await asyncio.gather(*(service.get_scores_for_date(date(2026, 5, day)) for day in range(1, 31))) + assert inner.get.await_count == 30 + assert peak == RiskService.MAX_CONCURRENT_REQUESTS diff --git a/src/tests/test_main.py b/src/tests/test_main.py index 99074569..aa2a59f7 100644 --- a/src/tests/test_main.py +++ b/src/tests/test_main.py @@ -115,7 +115,7 @@ async def test_risk_refresh_loop_swallows_refresh_errors_and_continues(): @pytest.mark.asyncio async def test_lifespan_refreshes_and_cancels_background_tasks_when_risk_api_configured(): - fake_service = SimpleNamespace(is_configured=True, refresh=AsyncMock()) + fake_service = SimpleNamespace(is_configured=True, refresh=AsyncMock(), aclose=AsyncMock()) fake_tasks = [_CancelledTask(), _CancelledTask()] def fake_create_task(coro): @@ -133,11 +133,12 @@ def fake_create_task(coro): assert all(task.cancel_called is False for task in fake_tasks) assert all(task.cancel_called is True for task in fake_tasks) + fake_service.aclose.assert_awaited_once() @pytest.mark.asyncio async def test_lifespan_skips_risk_refresh_but_starts_validation_worker_when_risk_api_not_configured(): - fake_service = SimpleNamespace(is_configured=False, refresh=AsyncMock()) + fake_service = SimpleNamespace(is_configured=False, refresh=AsyncMock(), aclose=AsyncMock()) fake_task = _CancelledTask() def fake_create_task(coro): @@ -157,6 +158,7 @@ def fake_create_task(coro): assert fake_task.cancel_called is True close_client.assert_awaited_once() + fake_service.aclose.assert_awaited_once() @pytest.mark.asyncio