"""Геослой по рекомендациям анализа: водоёмы + ж/д в зоне поиска. Рекомендации профилей (scoring_service.BEHAVIORAL_PROFILES) указывают критичные типы локаций, но коэффициенты railway/shelter долгое время были мёртвыми (railway_distance_km никто не вычислял — вклад в score_zone был всегда 0). Этот эндпоинт отдаёт реальные объекты OSM вокруг ТНП случая, чтобы подсветить их на карте зон поиска (SearchMap): - водоёмы/болота: РАС water ×3.0, эпилепсия ×3.5, возраст 0–7 ×2.2–2.5, критическое предупреждение «проверить ВСЕ водоёмы и ж/д»; - ж/д пути: РАС railway ×2.5 — «немедленно проверить ВСЕ ж/д пути». Данные: OpenStreetMap, лицензия ODbL — attribution обязателен на карте. Кэш дисковый; circuit breaker переиспользуем от geo_service, чтобы анализ и геослои отключались синхронно при отсутствии сети. """ from __future__ import annotations import hashlib import json import logging import math 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 _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; """ # Ж/д: railway=rail общего пользования (без промышленных/сервисных веток). _RAILWAY_QUERY_TEMPLATE = """ [out:json][timeout:25]; ( way["railway"="rail"](around:{radius_m},{lat},{lon}); ); out geom; """ _CLASS_WARNING = 'Водоём в зоне поиска — приоритетная проверка (профиль: {profiles})' _RAIL_WARNING = 'Ж/д пути в зоне поиска — проверить немедленно (профиль: {profiles})' class GeoFeature(BaseModel): """Объект геослоя в GeoJSON-геометрии + атрибуты для попапа.""" layer: str # water | wetland | railway name: Optional[str] = None area_m2: Optional[float] = None # для полигонов воды length_m: Optional[float] = None # для линий ж/д distance_m: Optional[float] = None geom: dict[str, Any] # GeoJSON (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 railway_warning: Optional[str] = None features: list[GeoFeature] = [] 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 _way_length_mid(geometry: list[dict]) -> tuple[float, float, float]: """(длина линии м, средняя широта, средняя долгота) по геометрии way.""" length = sum( _haversine(a['lat'], a['lon'], b['lat'], b['lon']) for a, b in zip(geometry, geometry[1:]) ) lat = sum(p['lat'] for p in geometry) / len(geometry) lon = sum(p['lon'] for p in geometry) / len(geometry) return length, lat, lon 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 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 _water_geoms(elements: list[dict]) -> list[dict]: """Ответ Overpass → список feature-словарей воды/болот (layer, name, geom, area).""" out: list[dict] = [] for el in elements: tags = el.get('tags') or {} layer = '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 area, lat, lon = _polygon_centroid_area(el['geometry']) out.append({'layer': layer, 'name': name, 'geom': poly, 'area_m2': area, 'lat': lat, 'lon': lon}) elif el.get('type') == 'relation' and el.get('members'): # Мультиполигон из внешних колец way-членов. polygons: list[list[list[float]]] = [] areas: list[float] = [] lats: list[float] = [] lons: list[float] = [] 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 # Кладём Polygon целиком (coordinates = [ring]), не сам ring — # иначе у одиночного кольца coordinates получатся depth-2. polygons.append(poly['coordinates']) area, lat, lon = _polygon_centroid_area(mg) areas.append(area) lats.append(lat) lons.append(lon) if not polygons: continue if len(polygons) == 1: geom = {'type': 'Polygon', 'coordinates': polygons[0]} else: # GeoJSON MultiPolygon: coordinates = список polygon.coordinates. geom = { 'type': 'MultiPolygon', 'coordinates': polygons, } out.append({'layer': layer, 'name': name, 'geom': geom, 'area_m2': sum(areas), 'lat': sum(lats) / len(lats), 'lon': sum(lons) / len(lons)}) return out def _railway_geoms(elements: list[dict]) -> list[dict]: """Ответ Overpass → список feature-словарей ж/д (LineString, длина).""" out: list[dict] = [] for el in elements: if el.get('type') != 'way': continue tags = el.get('tags') or {} # Только магистральные ж/д общего пользования: без сервисных/промышленных веток. if tags.get('service') or tags.get('usage') in ('industrial', 'military', 'test'): continue geometry = el.get('geometry') or [] if len(geometry) < 2: continue length, lat, lon = _way_length_mid(geometry) # Короткие обрезки путей (<300 м, нарезка по стрелкам внутри станции) # шумны и не дают штабу полезной информации — пропускаем. if length < 300: continue line = {'type': 'LineString', 'coordinates': [[p['lon'], p['lat']] for p in geometry]} out.append({'layer': 'railway', 'name': tags.get('name'), 'geom': line, 'length_m': length, 'lat': lat, 'lon': lon}) return out def _cache_get(key: str) -> Optional[list[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: list[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('geo layer cache write failed', exc_info=True) def _detect_profiles(case_detail: dict[str, Any]) -> list[str]: """Профили из диагнозов + флагов (логика как analyze._derive_profiles + cant_swim).""" 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 def _sort_features(features: list[dict]) -> list[dict]: """Крупные водные объекты вперёд (по площади), затем ж/д (по длине).""" def key(f: dict) -> tuple[int, float]: if f['layer'] == 'railway': return (1, -(f.get('length_m') or 0)) return (0, -(f.get('area_m2') or 0)) return sorted(features, key=key) @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 = _CLASS_WARNING.format(profiles=', '.join(profiles)) if profiles else None railway_warning = _RAIL_WARNING.format(profiles=', '.join(profiles)) if profiles else None radius_m = 3000 # перекрывает дефолтную зону анализа (max_distance_km ≤ ~3) cache_key = hashlib.md5(f'{lat:.5f}:{lon:.5f}:{radius_m}:v2'.encode()).hexdigest() cached = _cache_get(cache_key) if cached is not None: features_payload = cached is_cached = True else: features_payload = [] if not _overpass_circuit_open(): for template, extractor in ( (_WATER_QUERY_TEMPLATE, _water_geoms), (_RAILWAY_QUERY_TEMPLATE, _railway_geoms), ): query = template.format(lat=lat, lon=lon, radius_m=radius_m) data = await query_overpass(query) elements = data if isinstance(data, list) else data.get('elements', []) for item in extractor(elements): distance = round(_haversine(lat, lon, item['lat'], item['lon'])) feature = { 'layer': item['layer'], 'name': item.get('name'), 'distance_m': distance, 'geom': item['geom'], } if item['layer'] == 'railway': feature['length_m'] = round(item['length_m']) else: feature['area_m2'] = round(item['area_m2']) features_payload.append(feature) if not elements: break # сеть недоступна — не тянем второй запрос впустую if features_payload: # пустой ответ не кэшируем (breaker/сеть) _cache_put(cache_key, features_payload) is_cached = False features_sorted = _sort_features(features_payload) return WaterResponse(case_id=str(case_id), tnp={'lat': lat, 'lon': lon}, radius_m=radius_m, count=len(features_sorted), profiles=profiles, warning=warning, railway_warning=railway_warning, features=[GeoFeature(**f) for f in features_sorted], cached=is_cached)