""" Сервис оценки и ранжирования зон поиска на основе взвешенных факторов. Реализация согласно §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, max_distance_km: float = 2.0) -> 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 = max_distance_km * 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 # Психотип: множитель полосы дистанции (§6 контекста) — порт # фронтендовой getPsychotypeModifiers. Таблицу формирует бэкенд # (psychotype_service.get_psychotype_modifiers) и прокидывает через # case['psychotype_modifiers']. Без психотипа — множитель 1.0. mods = case.get('psychotype_modifiers') if mods: zdist = zone.get('distance_km', 1.0) if zdist < 0.5: band = mods.get('zone_0_500', 1.0) elif zdist < 1.5: band = mods.get('zone_500_1500', 1.0) elif zdist < 2.5: band = mods.get('zone_1500_2500', 1.0) else: band = mods.get('zone_2500plus', 1.0) score *= band # Конвертируем в шкалу 0-100 return round(score * 100, 2) def rank_zones(self, zones: List[Dict], case: Dict, max_distance_km: float = 2.0) -> 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, max_distance_km=max_distance_km) 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 }