root
163 lines
5.4 KiB
Python
163 lines
5.4 KiB
Python
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
from pathlib import Path
|
|
import sys
|
|
import types
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from backend import database as backend_database
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
@dataclass
|
|
class CaseRecord:
|
|
id: str
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
age: int | None = None
|
|
gender: str | None = None
|
|
health_flags: list[str] = field(default_factory=list)
|
|
elapsed_hours: int | None = None
|
|
last_known_place: str | None = None
|
|
direction: str | None = None
|
|
terrain_primary: str | None = None
|
|
weather: str | None = None
|
|
resources: list[str] = field(default_factory=list)
|
|
gps_lat: float | None = None
|
|
gps_lon: float | None = None
|
|
notes: str | None = None
|
|
status: str = 'new'
|
|
found_alive: bool | None = None
|
|
found_distance_km: float | None = None
|
|
found_direction: str | None = None
|
|
found_location_type: str | None = None
|
|
found_lat: float | None = None
|
|
found_lon: float | None = None
|
|
who_found: str | None = None
|
|
raw_text: str | None = None
|
|
analysis_log: str | None = None
|
|
result: dict[str, Any] | None = None
|
|
|
|
def to_summary(self) -> dict[str, Any]:
|
|
return {
|
|
'id': self.id,
|
|
'created_at': self.created_at.isoformat(),
|
|
'updated_at': self.updated_at.isoformat(),
|
|
'age': self.age,
|
|
'gender': self.gender,
|
|
'status': self.status,
|
|
'found_alive': self.found_alive,
|
|
'found_distance_km': self.found_distance_km,
|
|
'found_direction': self.found_direction,
|
|
'found_location_type': self.found_location_type,
|
|
'who_found': self.who_found,
|
|
'last_known_place': self.last_known_place,
|
|
'direction': self.direction,
|
|
'terrain_primary': self.terrain_primary,
|
|
'weather': self.weather,
|
|
}
|
|
|
|
def to_detail(self) -> dict[str, Any]:
|
|
data = self.to_summary()
|
|
data.update({
|
|
'health_flags': list(self.health_flags),
|
|
'elapsed_hours': self.elapsed_hours,
|
|
'resources': list(self.resources),
|
|
'gps_lat': self.gps_lat,
|
|
'gps_lon': self.gps_lon,
|
|
'notes': self.notes,
|
|
'found_lat': self.found_lat,
|
|
'found_lon': self.found_lon,
|
|
'raw_text': self.raw_text,
|
|
'analysis_log': self.analysis_log,
|
|
'result': self.result,
|
|
})
|
|
return data
|
|
|
|
|
|
class TestDB:
|
|
def __init__(self) -> None:
|
|
self._cases: dict[str, CaseRecord] = {}
|
|
|
|
def create_case(self, **data: Any) -> CaseRecord:
|
|
now = utcnow()
|
|
data.pop('mobile', None)
|
|
data.setdefault('status', 'new')
|
|
case = CaseRecord(id=str(uuid4()), created_at=now, updated_at=now, **data)
|
|
self._cases[case.id] = case
|
|
return case
|
|
|
|
def list_cases(self) -> list[CaseRecord]:
|
|
return sorted(self._cases.values(), key=lambda c: c.created_at, reverse=True)
|
|
|
|
def get_case(self, case_id: str) -> CaseRecord | None:
|
|
return self._cases.get(case_id)
|
|
|
|
def update_case(self, case_id: str, **patch: Any) -> CaseRecord | None:
|
|
case = self._cases.get(case_id)
|
|
if not case:
|
|
return None
|
|
for key, value in patch.items():
|
|
if hasattr(case, key) and value is not None:
|
|
setattr(case, key, value)
|
|
case.updated_at = utcnow()
|
|
return case
|
|
|
|
def delete_case(self, case_id: str) -> bool:
|
|
return self._cases.pop(case_id, None) is not None
|
|
|
|
def stats(self) -> dict[str, Any]:
|
|
cases = list(self._cases.values())
|
|
total = len(cases)
|
|
found_alive = sum(1 for c in cases if c.found_alive is True)
|
|
found_deceased = sum(1 for c in cases if c.found_alive is False)
|
|
unknown = total - found_alive - found_deceased
|
|
distances = [c.found_distance_km for c in cases if c.found_distance_km is not None]
|
|
median_distance = None
|
|
if distances:
|
|
distances = sorted(distances)
|
|
mid = len(distances) // 2
|
|
if len(distances) % 2:
|
|
median_distance = distances[mid]
|
|
else:
|
|
median_distance = (distances[mid - 1] + distances[mid]) / 2
|
|
return {
|
|
'total_cases': total,
|
|
'found_alive_count': found_alive,
|
|
'found_deceased_count': found_deceased,
|
|
'unknown_outcome_count': unknown,
|
|
'median_found_distance_km': median_distance,
|
|
'recent_activity_count': sum(1 for c in cases if (utcnow() - c.updated_at).days < 7),
|
|
}
|
|
|
|
def heatmap(self) -> list[dict[str, Any]]:
|
|
heat = []
|
|
for case in self._cases.values():
|
|
if case.gps_lat is not None and case.gps_lon is not None:
|
|
heat.append({'lat': case.gps_lat, 'lon': case.gps_lon, 'weight': 1})
|
|
return heat
|
|
|
|
|
|
db = TestDB()
|
|
backend_database.db = db
|
|
|
|
services_pkg = types.ModuleType('backend.services')
|
|
services_pkg.__path__ = []
|
|
stats_mod = types.ModuleType('backend.services.stats_service')
|
|
stats_mod.summary = db.stats
|
|
stats_mod.heatmap = db.heatmap
|
|
services_pkg.stats_service = stats_mod
|
|
sys.modules['backend.services'] = services_pkg
|
|
sys.modules['backend.services.stats_service'] = stats_mod
|