"""Водоёмы в зоне поиска — геослой по рекомендациям анализа. Рекомендации профилей (scoring_service.BEHAVIORAL_PROFILES) говорят «водоёмы — критичный тип локаций»: РАС ×3.0 + «перекрыть ВСЕ водоёмы и ж/д», эпилепсия ×3.5, дети 0–7 лет ×2.2–2.5. Этот эндпоинт отдаёт полигоны водоёмов (OSM/Overpass) вокруг ТНП случая, чтобы подсветить их на карте зон поиска (AnalysisResult → SearchMap), а не только называть их текстом. Данные: OpenStreetMap, лицензия ODbL — attribution обязателен на карте. Кэш дисковый как у geo_service; circuit breaker переиспользуем оттуда же, чтобы анализ и водоёмы отключались синхронно при отсутствии сети. """ from __future__ import annotations import hashlib import json import logging import math from pathlib import Path from typing import Any, Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from backend.database import db from backend.routers.auth import require_roles from services.geo_service import _overpass_circuit_open, query_overpass logger = logging.getLogger(__name__) router = APIRouter(prefix='/api/v1/water', tags=['water'], dependencies=[Depends(require_roles(['operator', 'field', 'admin']))]) # Кэш водоёмов — свой TTL (медленнее устаревает, чем дорожные данные): # гидрография OSM стабильна в масштабе дней. _CACHE_DIR = '/tmp/overpass_cache/water' _CACHE_TTL_HOURS = 72 # Overpass-теги водных объектов, релевантных для поиска пропавшего. # natural=water покрывает озера/водохранилища/пруды/реки (полигоны), # wetland — болота (застревание, сложная проходимость). _WATER_QUERY_TEMPLATE = """ [out:json][timeout:25]; ( way["natural"="water"](around:{radius_m},{lat},{lon}); relation["natural"="water"](around:{radius_m},{lat},{lon}); way["natural"="wetland"](around:{radius_m},{lat},{lon}); relation["natural"="wetland"](around:{radius_m},{lat},{lon}); ); out geom meta; """ # Порядок = приоритет отрисовки (последний рисуется сверху). _KIND_STYLE = [ # (ключ фильтра, значение тега, цвет, заливка) ('natural', 'wetland', '#4a6a8a', 0.25), ('natural', 'water', '#2f81f7', 0.35), ] _CLASS_WARNING = 'Водоём в зоне поиска — приоритетная проверка (профиль: {profiles})' class WaterFeature(BaseModel): """Полигон водоёма в GeoJSON-геометрии + атрибуты для попапа.""" kind: str # water | wetland name: Optional[str] = None area_m2: Optional[float] = None distance_m: Optional[float] = None geom: dict[str, Any] # GeoJSON Polygon/MultiPolygon (WGS84) class WaterResponse(BaseModel): case_id: Optional[str] = None tnp: dict[str, Optional[float]] radius_m: int count: int profiles: list[str] = [] warning: Optional[str] = None features: list[WaterFeature] = [] cached: bool = False attribution: str = '© OpenStreetMap contributors' def _haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """Расстояние между точками, метры (как в geo_service).""" r = 6371000.0 p1, p2 = math.radians(lat1), math.radians(lat2) dp = p2 - p1 dl = math.radians(lon2 - lon1) a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2 return 2 * r * math.asin(math.sqrt(a)) def _polygon_centroid_area(ring: list[dict]) -> tuple[float, float, float]: """(площадь м², средняя широта, средняя долгота) по внешнему кольцу. Площадь — формула сферического избытка; для наших масштабов (пруды, водохранилища в радиусе поиска) достаточно точна для порогов «крупный/мелкий». """ n = len(ring) if n < 3: return 0.0, 0.0, 0.0 lat = sum(p['lat'] for p in ring) / n lon = sum(p['lon'] for p in ring) / n r = 6371000.0 # spherical excess (approx, unit sphere × r²) area = 0.0 for i in range(n - 1): p1, p2 = ring[i], ring[i + 1] area += math.radians(p2['lon'] - p1['lon']) * ( 2 + math.sin(math.radians(p1['lat'])) + math.sin(math.radians(p2['lat'])) ) area = abs(area) * r * r / 2 return area, lat, lon def _way_to_polygon(element: dict) -> Optional[dict[str, Any]]: """Замкнутый way → GeoJSON Polygon; незамкнутый — None.""" geometry = element.get('geometry') or [] if len(geometry) < 4: return None first, last = geometry[0], geometry[-1] if abs(first['lat'] - last['lat']) > 1e-9 or abs(first['lon'] - last['lon']) > 1e-9: return None coords = [[p['lon'], p['lat']] for p in geometry] return {'type': 'Polygon', 'coordinates': [coords]} def _element_geoms(elements: list[dict]) -> list[tuple[str, Optional[str], dict, float, float, float]]: """Из ответа Overpass → (kind, name, geom, area_m2, lat, lon).""" out: list[tuple[str, Optional[str], dict, float, float, float]] = [] for el in elements: tags = el.get('tags') or {} kind = 'wetland' if tags.get('natural') == 'wetland' else 'water' name = tags.get('name') if el.get('type') == 'way': poly = _way_to_polygon(el) if poly is None: continue ring = el['geometry'] area, lat, lon = _polygon_centroid_area(ring) out.append((kind, name, poly, area, lat, lon)) elif el.get('type') == 'relation' and el.get('members'): # Мультиполигон из внешних колец way-членов; без top-level geometry. polygons: list[list[list[float]]] = [] areas = [] lats, lons = [], [] for member in el.get('members', []): if member.get('role') not in ('outer', ''): continue mg = member.get('geometry') or [] if len(mg) < 4: continue poly = _way_to_polygon({'geometry': mg}) if poly is None: continue polygons.append(poly['coordinates'][0]) area, lat, lon = _polygon_centroid_area(mg) areas.append(area) lats.append(lat) lons.append(lon) if not polygons: continue geom: dict[str, Any] = ( {'type': 'MultiPolygon', 'coordinates': [p for p in polygons]} if len(polygons) > 1 else {'type': 'Polygon', 'coordinates': polygons[0]} ) out.append((kind, name, geom, sum(areas), sum(lats) / len(lats), sum(lons) / len(lons))) return out def _cache_get(key: str) -> Optional[dict]: import os from datetime import datetime, timedelta path = os.path.join(_CACHE_DIR, f'{key}.json') if not os.path.exists(path): return None try: with open(path, 'r', encoding='utf-8') as f: cached = json.load(f) ts = datetime.fromisoformat(cached['timestamp']) if datetime.now() - ts > timedelta(hours=_CACHE_TTL_HOURS): os.unlink(path) return None return cached['data'] except Exception: return None def _cache_put(key: str, data: dict) -> None: import os from datetime import datetime try: os.makedirs(_CACHE_DIR, exist_ok=True) with open(os.path.join(_CACHE_DIR, f'{key}.json'), 'w', encoding='utf-8') as f: json.dump({'timestamp': datetime.now().isoformat(), 'data': data}, f, ensure_ascii=False) except Exception: logger.warning('water cache write failed', exc_info=True) def _detect_profiles(case_detail: dict[str, Any]) -> list[str]: """Профили из диагнозов (та же логика, что analyze._derive_profiles, по detail).""" profiles: list[str] = [] for diag in (case_detail.get('diagnosis_type') or []): d = str(diag).lower() if 'рас' in d or 'аутизм' in d: profiles.append('РАС') elif 'эпилепси' in d: profiles.append('эпилепсия') if case_detail.get('cant_swim'): profiles.append('не_умеет_плавать') return profiles @router.get('/{case_id}', response_model=WaterResponse) async def water_for_case(case_id: str) -> WaterResponse: """Полигоны водоёмов вокруг ТНП случая (для слоя на карте зон поиска).""" case = db.get_case(str(case_id)) if case is None: raise HTTPException(status_code=404, detail='Case not found') detail = case.to_detail() lat, lon = detail.get('tnp_lat'), detail.get('tnp_lon') if lat is None or lon is None: raise HTTPException(status_code=409, detail='Case has no TNP coordinates (tnp_lat/tnp_lon)') profiles = _detect_profiles(detail) warning = None if profiles: warning = _CLASS_WARNING.format(profiles=', '.join(profiles)) radius_m = 3000 # перекрывает дефолтную зону анализа (max_distance_km ≤ ~3) cache_key = hashlib.md5(f'{lat:.5f}:{lon:.5f}:{radius_m}'.encode()).hexdigest() cached = _cache_get(cache_key) if cached is not None: features_payload = cached is_cached = True else: if _overpass_circuit_open(): return WaterResponse(case_id=str(case_id), tnp={'lat': lat, 'lon': lon}, radius_m=radius_m, count=0, profiles=profiles, features=[], cached=False) query = _WATER_QUERY_TEMPLATE.format(lat=lat, lon=lon, radius_m=radius_m) data = await query_overpass(query) if isinstance(data, list): elements = data else: elements = data.get('elements', []) features_payload = [] for kind, name, geom, area, flat, flon in _element_geoms(elements): features_payload.append({ 'kind': kind, 'name': name, 'area_m2': round(area) if area else None, 'distance_m': round(_haversine(lat, lon, flat, flon)) if area else None, 'geom': geom, }) # Крупные объекты вперёд, безымянные малые — в хвост. features_payload.sort(key=lambda f: f['area_m2'] or 0, reverse=True) if elements: # не кэшируем пустой ответ от трипнутого breaker'а _cache_put(cache_key, features_payload) is_cached = False features = [WaterFeature(**f) for f in features_payload] return WaterResponse(case_id=str(case_id), tnp={'lat': lat, 'lon': lon}, radius_m=radius_m, count=len(features), profiles=profiles, warning=warning, features=features, cached=is_cached)