Sadmin/vector
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

A8: remove dead backend code (old api/ copy, models_* duplicates, .bak files)

Browse Source
This commit is contained in:
root
2026-07-25 11:48:30 +00:00
parent 5490f3c5f2
commit 6ca59a04d0
13 changed files with 0 additions and 2027 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/geo_service.py.bak
-377
View File
@@ -1,377 +0,0 @@
"""
Geo service for building search zones and querying OpenStreetMap data via Overpass API.
"""
import math
import json
import hashlib
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Optional, Tuple
import httpx
from pydantic import BaseModel
class Zone(BaseModel):
"""Search zone with geographic features."""
direction: str # N, NE, E, SE, S, SW, W, NW
distance_km: float
forest_pct: float
road_density: float # km of roads per km²
water_distance_km: Optional[float]
settlement_distance_km: Optional[float]
# Cache configuration
CACHE_DIR = Path("/tmp/overpass_cache")
CACHE_TTL_HOURS = 24
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
# Direction mappings
DIRECTIONS = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]
DIRECTION_ANGLES = {
"N": 0,
"NE": 45,
"E": 90,
"SE": 135,
"S": 180,
"SW": 225,
"W": 270,
"NW": 315
}
# Search distances in meters
SEARCH_DISTANCES = [500, 1000, 2000, 5000]
def haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""
Calculate distance between two points on Earth using Haversine formula.
Args:
lat1, lon1: First point coordinates
lat2, lon2: Second point coordinates
Returns:
Distance in kilometers
"""
R = 6371 # Earth radius in km
lat1_rad = math.radians(lat1)
lat2_rad = math.radians(lat2)
dlat = math.radians(lat2 - lat1)
dlon = math.radians(lon2 - lon1)
a = (math.sin(dlat / 2) ** 2 +
math.cos(lat1_rad) * math.cos(lat2_rad) * math.sin(dlon / 2) ** 2)
c = 2 * math.asin(math.sqrt(a))
return R * c
def get_sector_bounds(lat: float, lon: float, direction: str, radius_m: int) -> Tuple[float, float, float, float]:
"""
Calculate bounding box for a sector.
Args:
lat, lon: Center point
direction: Sector direction (N, NE, E, etc.)
radius_m: Radius in meters
Returns:
(min_lat, min_lon, max_lat, max_lon)
"""
# Convert radius to degrees (approximate)
radius_deg = radius_m / 111320 # 1 degree ≈ 111.32 km at equator
angle = DIRECTION_ANGLES[direction]
angle_rad = math.radians(angle)
# Calculate sector boundaries (45° sectors)
angle_start = angle - 22.5
angle_end = angle + 22.5
# Simple bounding box (can be optimized for actual sector shape)
lat_offset = radius_deg * math.cos(angle_rad)
lon_offset = radius_deg * math.sin(angle_rad) / math.cos(math.radians(lat))
min_lat = min(lat, lat + lat_offset) - radius_deg * 0.5
max_lat = max(lat, lat + lat_offset) + radius_deg * 0.5
min_lon = min(lon, lon + lon_offset) - radius_deg * 0.5
max_lon = max(lon, lon + lon_offset) + radius_deg * 0.5
return (min_lat, min_lon, max_lat, max_lon)
def get_cache_key(query: str) -> str:
"""Generate cache key from query."""
return hashlib.md5(query.encode()).hexdigest()
def get_cached_result(cache_key: str) -> Optional[Dict]:
"""Get cached Overpass API result if not expired."""
CACHE_DIR.mkdir(exist_ok=True)
cache_file = CACHE_DIR / f"{cache_key}.json"
if not cache_file.exists():
return None
try:
with open(cache_file, 'r') as f:
cached = json.load(f)
cached_time = datetime.fromisoformat(cached['timestamp'])
if datetime.now() - cached_time > timedelta(hours=CACHE_TTL_HOURS):
cache_file.unlink()
return None
return cached['data']
except Exception:
return None
def save_to_cache(cache_key: str, data: Dict):
"""Save Overpass API result to cache."""
CACHE_DIR.mkdir(exist_ok=True)
cache_file = CACHE_DIR / f"{cache_key}.json"
try:
with open(cache_file, 'w') as f:
json.dump({
'timestamp': datetime.now().isoformat(),
'data': data
}, f)
except Exception:
pass
async def query_overpass(query: str) -> Dict:
"""
Query Overpass API with caching.
Args:
query: Overpass QL query
Returns:
API response as dict
"""
cache_key = get_cache_key(query)
# Check cache
cached = get_cached_result(cache_key)
if cached is not None:
return cached
# Query API
async with httpx.AsyncClient(timeout=30.0) as client:
try:
response = await client.post(
OVERPASS_URL,
data={'data': query},
headers={'Content-Type': 'application/x-www-form-urlencoded'}
)
response.raise_for_status()
data = response.json()
# Save to cache
save_to_cache(cache_key, data)
return data
except Exception as e:
# Return empty result on error
return {'elements': []}
def calculate_road_length(elements: List[Dict]) -> float:
"""
Calculate total road length from Overpass way elements.
Args:
elements: List of way elements from Overpass
Returns:
Total length in kilometers
"""
total_length = 0.0
for element in elements:
if element.get('type') != 'way':
continue
nodes = element.get('geometry', [])
if len(nodes) < 2:
continue
# Calculate length by summing distances between consecutive nodes
for i in range(len(nodes) - 1):
lat1, lon1 = nodes[i]['lat'], nodes[i]['lon']
lat2, lon2 = nodes[i + 1]['lat'], nodes[i + 1]['lon']
total_length += haversine(lat1, lon1, lat2, lon2)
return total_length
def find_nearest_distance(lat: float, lon: float, elements: List[Dict]) -> Optional[float]:
"""
Find distance to nearest element.
Args:
lat, lon: Reference point
elements: List of node elements from Overpass
Returns:
Distance in kilometers, or None if no elements
"""
if not elements:
return None
min_distance = float('inf')
for element in elements:
if element.get('type') != 'node':
continue
elem_lat = element.get('lat')
elem_lon = element.get('lon')
if elem_lat is None or elem_lon is None:
continue
distance = haversine(lat, lon, elem_lat, elem_lon)
min_distance = min(min_distance, distance)
return min_distance if min_distance != float('inf') else None
def calculate_forest_coverage(elements: List[Dict], radius_m: int) -> float:
"""
Estimate forest coverage percentage.
Args:
elements: List of way elements from Overpass
radius_m: Search radius in meters
Returns:
Forest coverage as percentage (0-100)
"""
if not elements:
return 0.0
# Approximate: count forest ways and estimate coverage
# This is a simplified calculation
forest_ways = len([e for e in elements if e.get('type') == 'way'])
# Rough heuristic: each forest way covers ~0.1 km²
# Total search area = π * r²
search_area_km2 = math.pi * (radius_m / 1000) ** 2
estimated_forest_km2 = forest_ways * 0.1
coverage_pct = min(100.0, (estimated_forest_km2 / search_area_km2) * 100)
return round(coverage_pct, 1)
async def get_zone_features(lat: float, lon: float, direction: str, radius_m: int) -> Dict:
"""
Get geographic features for a zone using Overpass API.
Args:
lat, lon: Center point
direction: Sector direction
radius_m: Search radius in meters
Returns:
Dict with roads_km, water_distance_km, settlement_distance_km, forest_pct
"""
# Query roads
roads_query = f"""
[out:json];
(
way[highway](around:{radius_m},{lat},{lon});
);
out geom;
"""
roads_data = await query_overpass(roads_query)
roads_km = calculate_road_length(roads_data.get('elements', []))
# Query water bodies
water_query = f"""
[out:json];
(
node[natural=water](around:{radius_m},{lat},{lon});
way[natural=water](around:{radius_m},{lat},{lon});
);
out center;
"""
water_data = await query_overpass(water_query)
water_distance = find_nearest_distance(lat, lon, water_data.get('elements', []))
# Query settlements
settlement_query = f"""
[out:json];
(
node[place~"village|town|city"](around:{radius_m},{lat},{lon});
);
out;
"""
settlement_data = await query_overpass(settlement_query)
settlement_distance = find_nearest_distance(lat, lon, settlement_data.get('elements', []))
# Query forests
forest_query = f"""
[out:json];
(
way[landuse=forest](around:{radius_m},{lat},{lon});
way[natural=wood](around:{radius_m},{lat},{lon});
);
out geom;
"""
forest_data = await query_overpass(forest_query)
forest_pct = calculate_forest_coverage(forest_data.get('elements', []), radius_m)
# Calculate road density (km of roads per km²)
search_area_km2 = math.pi * (radius_m / 1000) ** 2
road_density = roads_km / search_area_km2 if search_area_km2 > 0 else 0.0
return {
'roads_km': roads_km,
'road_density': round(road_density, 2),
'water_distance_km': water_distance,
'settlement_distance_km': settlement_distance,
'forest_pct': forest_pct
}
async def build_search_zones(lat: float, lon: float, case_data: dict) -> List[Zone]:
"""
Build search zones around a point.
Creates 8 directional sectors (N, NE, E, SE, S, SW, W, NW) at multiple distances
(500m, 1000m, 2000m, 5000m) and queries geographic features for each.
Args:
lat: Latitude of search origin
lon: Longitude of search origin
case_data: Case information (for future enhancements)
Returns:
List of Zone objects with geographic features
"""
zones = []
for distance_m in SEARCH_DISTANCES:
for direction in DIRECTIONS:
# Get features for this zone
features = await get_zone_features(lat, lon, direction, distance_m)
zone = Zone(
direction=direction,
distance_km=distance_m / 1000,
forest_pct=features['forest_pct'],
road_density=features['road_density'],
water_distance_km=features['water_distance_km'],
settlement_distance_km=features['settlement_distance_km']
)
zones.append(zone)
return zones
backend/services/scoring_service.py.bak
-403
View File
@@ -1,403 +0,0 @@
"""
Сервис оценки и ранжирования зон поиска на основе взвешенных факторов.
Реализация согласно §8 и §9 контекста ВЕКТОР.
"""
from typing import Dict, List, Optional
from copy import deepcopy
class WeightedScorer:
"""
Система взвешенной оценки зон поиска с учетом множественных факторов.
Базовые веса из §9 контекста.
"""
# Базовые веса факторов из §9 (сумма = 1.0)
BASE_WEIGHTS = {
'forest': 0.25,
'water': 0.20,
'roads': 0.18,
'settlement': 0.15,
'historical': 0.12,
'direction': 0.07,
'shelter': 0.03
}
# Возрастные модификаторы
AGE_MODIFIERS = {
'0-4': {
'forest': 0.6,
'water': 2.5,
'roads': 1.3,
'settlement': 1.8,
'shelter': 1.5,
'distance_mult': 0.3
},
'5-7': {
'forest': 0.8,
'water': 2.2,
'roads': 1.4,
'settlement': 1.6,
'shelter': 1.4,
'distance_mult': 0.5
},
'8-11': {
'forest': 1.1,
'water': 1.8,
'roads': 1.2,
'settlement': 1.3,
'shelter': 1.2,
'distance_mult': 0.8
},
'12-14': {
'forest': 1.3,
'water': 1.4,
'roads': 1.1,
'settlement': 1.0,
'shelter': 1.0,
'distance_mult': 1.2
},
'15-17': {
'forest': 1.4,
'water': 1.2,
'roads': 1.3,
'settlement': 0.9,
'shelter': 0.9,
'distance_mult': 1.5
}
}
# Сезонные модификаторы
SEASON_MODIFIERS = {
'зима': {
'forest': 0.8,
'water': 0.6,
'roads': 1.3,
'settlement': 1.5,
'shelter': 2.0,
'distance_mult': 0.7
},
'весна': {
'forest': 1.1,
'water': 1.8,
'roads': 1.0,
'settlement': 1.0,
'shelter': 1.2,
'distance_mult': 1.0
},
'лето': {
'forest': 1.2,
'water': 1.3,
'roads': 0.9,
'settlement': 0.8,
'shelter': 0.8,
'distance_mult': 1.3
},
'осень': {
'forest': 1.3,
'water': 1.1,
'roads': 1.0,
'settlement': 1.1,
'shelter': 1.1,
'distance_mult': 1.0
}
}
# Поведенческие профили — ТОЧНЫЕ коэффициенты из §8 контекста
BEHAVIORAL_PROFILES = {
'РАС': {
'water': 3.0,
'railway': 2.5,
'shelter': 2.0,
'settlement': 0.4,
'distance_mult': 2.0,
'critical_warning': 'НЕ использовать громкоговоритель с именем ребёнка! Немедленно перекрыть ВСЕ водоёмы и ж/д пути.'
},
'эпилепсия': {
'water': 3.5,
'shelter': 2.5,
'distance_mult': 0.6,
'critical_warning': 'Медицинский приоритет — возможна потеря сознания. Радиус поиска МЕНЬШЕ среднего.'
},
'СДВГ': {
'roads': 1.6,
'distance_mult': 1.4,
'note': 'Импульсивное движение, меняет направление. Откликается, но может не идти целенаправленно.'
},
'ЗПР': {
'settlement': 0.7,
'shelter': 1.5,
'distance_mult': 0.8,
'note': 'Не ориентируется в пространстве'
},
'велосипед': {
'distance_mult': 5.0,
'roads': 1.8,
'forest': 0.8,
'critical_warning': 'Немедленно расширить зону до 10-15 км! Приоритет: дороги и велодорожки. Запросить данные дорожных камер.'
},
'самокат': {
'distance_mult': 3.0,
'roads': 1.6,
'forest': 0.9
},
'намеренный_уход': {
'forest': 0.2,
'roads': 2.5,
'settlement': 3.0,
'note': 'Не прочёсывание леса, а розыск. Транспортные узлы, камеры, соцсети, друзья.'
}
}
def __init__(self):
"""Инициализация скорера с базовыми весами."""
self.weights = deepcopy(self.BASE_WEIGHTS)
self.distance_multiplier = 1.0
self.active_profiles = []
self.critical_warnings = []
def _get_age_group(self, age: int) -> str:
"""Определяет возрастную группу."""
if age <= 4:
return '0-4'
elif age <= 7:
return '5-7'
elif age <= 11:
return '8-11'
elif age <= 14:
return '12-14'
elif age <= 17:
return '15-17'
else:
return '18-64'
def apply_age_modifiers(self, age: int):
"""Применяет возрастные модификаторы к весам."""
age_group = self._get_age_group(age)
modifiers = self.AGE_MODIFIERS.get(age_group, {})
for factor, modifier in modifiers.items():
if factor == 'distance_mult':
self.distance_multiplier *= modifier
elif factor in self.weights:
self.weights[factor] *= modifier
def apply_season_modifiers(self, season: str):
"""Применяет сезонные модификаторы к весам."""
season_lower = season.lower() if season else 'лето'
modifiers = self.SEASON_MODIFIERS.get(season_lower, {})
for factor, modifier in modifiers.items():
if factor == 'distance_mult':
self.distance_multiplier *= modifier
elif factor in self.weights:
self.weights[factor] *= modifier
def apply_profile(self, profile_list: List[str]):
"""
Применяет поведенческие профили к весам.
Точные коэффициенты из §8 контекста.
Args:
profile_list: Список профилей (РАС, эпилепсия, СДВГ, велосипед и т.д.)
"""
if not profile_list:
return
for profile_name in profile_list:
profile = self.BEHAVIORAL_PROFILES.get(profile_name)
if not profile:
continue
self.active_profiles.append(profile_name)
# Сохранить критические предупреждения
if 'critical_warning' in profile:
self.critical_warnings.append({
'profile': profile_name,
'warning': profile['critical_warning']
})
for factor, modifier in profile.items():
if factor in ['critical_warning', 'note']:
continue
elif factor == 'distance_mult':
self.distance_multiplier *= modifier
elif factor in self.weights:
self.weights[factor] *= modifier
elif factor == 'railway':
# Ж/д пути — добавляем как отдельный фактор для РАС
if 'railway' not in self.weights:
self.weights['railway'] = 0.05
self.weights['railway'] *= modifier
def _normalize_weights(self):
"""Нормализует веса так, чтобы их сумма была 1.0."""
# Фильтруем None значения
valid_weights = {k: v for k, v in self.weights.items() if v is not None}
total = sum(valid_weights.values())
if total > 0:
for key in self.weights:
if self.weights[key] is not None:
self.weights[key] /= total
else:
self.weights[key] = 0.0
def score_zone(self, zone: Dict, case: Dict) -> float:
"""
Оценивает зону поиска на основе её характеристик и данных случая.
Args:
zone: Словарь с характеристиками зоны
case: Данные случая (age, season, profiles и т.д.)
Returns:
float: Оценка зоны (0-100)
"""
# Сбрасываем веса к базовым
self.weights = deepcopy(self.BASE_WEIGHTS)
self.distance_multiplier = 1.0
self.active_profiles = []
self.critical_warnings = []
# Применяем модификаторы
if 'age' in case and case['age']:
self.apply_age_modifiers(case['age'])
if 'season' in case and case['season']:
self.apply_season_modifiers(case['season'])
if 'profiles' in case and case['profiles']:
self.apply_profile(case['profiles'])
# Нормализуем веса
self._normalize_weights()
# Вычисляем оценку
score = 0.0
# Лес
forest_score = zone.get('forest_pct', 0.5)
score += self.weights['forest'] * forest_score
# Вода (чем ближе, тем важнее)
water_dist = zone.get('water_distance_km', 5.0)
water_score = max(0, 1.0 - (water_dist / 10.0)) if water_dist is not None else 0.5
score += self.weights['water'] * water_score
# Дороги
road_density = zone.get('road_density', 0.5)
road_score = min(1.0, road_density / 2.0) if road_density is not None else 0.5
score += self.weights['roads'] * road_score
# Населенные пункты
settlement_dist = zone.get('settlement_distance_km', 10.0)
settlement_score = max(0, 1.0 - (settlement_dist / 20.0)) if settlement_dist is not None else 0.5
score += self.weights['settlement'] * settlement_score
# Историческая частота
historical_score = zone.get('historical_freq', 0.5)
score += self.weights['historical'] * historical_score
# Совпадение направления
direction_score = zone.get('direction_match', 0.5)
score += self.weights['direction'] * direction_score
# Укрытия
shelter_score = zone.get('shelter_pct', 0.3)
score += self.weights['shelter'] * shelter_score
# Ж/д пути (для РАС)
if 'railway' in self.weights:
railway_dist = zone.get('railway_distance_km', 10.0)
railway_score = max(0, 1.0 - (railway_dist / 5.0))
score += self.weights['railway'] * railway_score
# Применяем множитель расстояния
zone_distance = zone.get('distance_km', 1.0)
expected_distance = 2.0 * self.distance_multiplier
distance_factor = 1.0 - abs(zone_distance - expected_distance) / (expected_distance * 2)
distance_factor = max(0.3, min(1.0, distance_factor))
score *= distance_factor
# Конвертируем в шкалу 0-100
return round(score * 100, 2)
def rank_zones(self, zones: List[Dict], case: Dict) -> List[Dict]:
"""
Ранжирует зоны по приоритету на основе оценок.
Args:
zones: Список зон с характеристиками
case: Данные случая
Returns:
List[Dict]: Отсортированный список зон с оценками и приоритетами
"""
scored_zones = []
for zone in zones:
zone_copy = deepcopy(zone)
zone_copy['score'] = self.score_zone(zone, case)
scored_zones.append(zone_copy)
scored_zones.sort(key=lambda x: x['score'], reverse=True)
for i, zone in enumerate(scored_zones):
zone['priority'] = i + 1
return scored_zones
def get_active_profiles_info(self) -> List[Dict]:
"""Возвращает информацию об активных профилях с предупреждениями."""
profiles_info = []
for profile_name in self.active_profiles:
profile = self.BEHAVIORAL_PROFILES.get(profile_name, {})
info = {
'name': profile_name,
'modifiers': {k: v for k, v in profile.items() if k not in ['critical_warning', 'note']},
}
if 'critical_warning' in profile:
info['critical_warning'] = profile['critical_warning']
if 'note' in profile:
info['note'] = profile['note']
profiles_info.append(info)
return profiles_info
def create_scorer_for_case(case: Dict) -> WeightedScorer:
"""Создает и настраивает скорер для конкретного случая."""
scorer = WeightedScorer()
if 'age' in case and case['age']:
scorer.apply_age_modifiers(case['age'])
if 'season' in case and case['season']:
scorer.apply_season_modifiers(case['season'])
if 'profiles' in case and case['profiles']:
scorer.apply_profile(case['profiles'])
scorer._normalize_weights()
return scorer
def get_weight_explanation(case: Dict) -> Dict:
"""Возвращает объяснение весов для данного случая."""
scorer = create_scorer_for_case(case)
return {
'weights': scorer.weights,
'distance_multiplier': scorer.distance_multiplier,
'age_group': scorer._get_age_group(case.get('age', 10)) if case.get('age') else None,
'season': case.get('season'),
'profiles': scorer.get_active_profiles_info(),
'critical_warnings': scorer.critical_warnings
}