""" 离散化模块 负责将连续值转换为离散状态 """ import os import yaml from typing import Optional, List, Dict, Any, Tuple from app.config.paths import DBN_CONFIG_DIR, get_logger logger = get_logger("dbn") class Discretizer: """离散化工具类""" def __init__(self, config_dir: Optional[str] = None): """ 初始化离散化器 Args: config_dir: 配置文件目录，默认为 app/config/dbn """ if config_dir is None: config_dir = str(DBN_CONFIG_DIR) self.config_dir = config_dir self.config = self._load_config() def _load_config(self) -> Dict[str, Any]: """加载离散化配置""" config_path = os.path.join(self.config_dir, 'discretization.yaml') if not os.path.exists(config_path): logger.warning(f"离散化配置文件不存在: {config_path}") return {} with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) return config def discretize(self, factor_name: str, value: float, region_code: Optional[str] = None) -> str: """ 将连续值离散化 Args: factor_name: 因子名称 value: 连续值 region_code: 行政区划代码，用于区域覆盖 Returns: 离散状态标签 """ if factor_name not in self.config: logger.warning(f"因子 {factor_name} 没有离散化配置") return "unknown" factor_config = self.config[factor_name] # 检查是否是分类变量（有mapping字段） if 'mapping' in factor_config: return self._discretize_categorical(factor_config, value) # 检查是否有区域覆盖 if region_code and 'region_overrides' in factor_config: if region_code in factor_config['region_overrides']: override_config = factor_config['region_overrides'][region_code] return self._discretize_continuous(override_config, value) # 使用默认配置 if 'default' in factor_config: return self._discretize_continuous(factor_config['default'], value) elif 'bins' in factor_config: return self._discretize_continuous(factor_config, value) logger.warning(f"因子 {factor_name} 的配置格式不正确") return "unknown" def _discretize_categorical(self, config: Dict[str, Any], value: float) -> str: """ 离散化分类变量 Args: config: 配置 value: 值 Returns: 离散状态标签 """ mapping = config.get('mapping', {}) default = config.get('default', 'unknown') # 将值转换为整数 int_value = int(value) return mapping.get(int_value, default) def _discretize_continuous(self, config: Dict[str, Any], value: float) -> str: """ 离散化连续变量 Args: config: 配置 value: 值 Returns: 离散状态标签 """ bins = config.get('bins', []) labels = config.get('labels', []) if not bins or not labels: logger.warning("离散化配置缺少bins或labels") return "unknown" # 确保bins和labels长度匹配 if len(bins) != len(labels) + 1: logger.warning(f"bins长度({len(bins)})应该比labels长度({len(labels)})多1") return "unknown" # 进行分箱 for i in range(len(bins) - 1): if bins[i] <= value < bins[i + 1]: return labels[i] # 如果值超出范围，返回最后一个标签 if value >= bins[-1]: return labels[-1] return labels[0] def discretize_rain_intensity(self, rainfall_mm_h: float) -> str: """ 离散化降雨强度 Args: rainfall_mm_h: 降雨强度（mm/h） Returns: 离散状态标签 """ return self.discretize('rain_intensity', rainfall_mm_h) def discretize_duration(self, duration_hours: float) -> str: """ 离散化持续时间 Args: duration_hours: 持续时间（小时） Returns: 离散状态标签 """ return self.discretize('duration', duration_hours) def discretize_accum_rain(self, accum_rain_mm: float) -> str: """ 离散化累计降雨量 Args: accum_rain_mm: 累计降雨量（mm） Returns: 离散状态标签 """ return self.discretize('accum_rain', accum_rain_mm) # ---- 地震触发层离散化 ---- def discretize_magnitude(self, magnitude: float) -> str: """ 离散化地震震级 Args: magnitude: 震级（Richter） Returns: 离散状态标签 """ return self.discretize('magnitude', magnitude) def discretize_epicenter_distance(self, distance_km: float) -> str: """ 离散化震中距 Args: distance_km: 震中距（km） Returns: 离散状态标签 """ return self.discretize('epicenter_distance', distance_km) def discretize_seismic_intensity(self, intensity: float) -> str: """ 离散化地震烈度 Args: intensity: 地震烈度（中国烈度表数值） Returns: 离散状态标签 """ return self.discretize('seismic_intensity', intensity) def discretize_elevation(self, elevation_m: float, region_code: Optional[str] = None) -> str: """ 离散化高程 Args: elevation_m: 高程（米） region_code: 行政区划代码 Returns: 离散状态标签 """ return self.discretize('elevation', elevation_m, region_code) def discretize_slope(self, slope_deg: float) -> str: """ 离散化坡度 Args: slope_deg: 坡度（度） Returns: 离散状态标签 """ return self.discretize('slope', slope_deg) def discretize_aspect(self, aspect_deg: float) -> str: """ 离散化坡向 Args: aspect_deg: 坡向（度） Returns: 离散状态标签 """ return self.discretize('aspect', aspect_deg) def discretize_soil_type(self, soil_type_code: int) -> str: """ 离散化土壤类型 Args: soil_type_code: 土壤类型代码 Returns: 离散状态标签 """ return self.discretize('soil_type', soil_type_code) def discretize_lithology(self, lithology_code: int) -> str: """ 离散化岩性 Args: lithology_code: 岩性代码 Returns: 离散状态标签 """ return self.discretize('lithology', lithology_code) def discretize_landuse(self, landuse_code: int) -> str: """ 离散化土地利用类型 Args: landuse_code: 土地利用类型代码 Returns: 离散状态标签 """ return self.discretize('landuse', landuse_code) def discretize_terrain(self, terrain_code: int) -> str: """ 离散化地形分类 Args: terrain_code: 地形分类代码 Returns: 离散状态标签 """ return self.discretize('terrain', terrain_code) def discretize_impervious(self, impervious_ratio: float) -> str: """ 离散化不透水面 Args: impervious_ratio: 不透水面比例 Returns: 离散状态标签 """ return self.discretize('impervious', impervious_ratio) def discretize_ndvi(self, ndvi_value: float) -> str: """ 离散化植被指数 Args: ndvi_value: NDVI值 Returns: 离散状态标签 """ return self.discretize('ndvi', ndvi_value) def discretize_sand_content(self, sand_percent: float) -> str: """ 离散化土壤含沙量 Args: sand_percent: 含沙量百分比 Returns: 离散状态标签 """ return self.discretize('sand_content', sand_percent) def discretize_ph(self, ph_value: float) -> str: """ 离散化土壤PH值 Args: ph_value: PH值 Returns: 离散状态标签 """ return self.discretize('ph', ph_value) def discretize_soil_moisture(self, moisture_percent: float) -> str: """ 离散化土壤湿度 Args: moisture_percent: 湿度百分比 Returns: 离散状态标签 """ return self.discretize('soil_moisture', moisture_percent) def discretize_organic_carbon(self, carbon_percent: float) -> str: """ 离散化有机碳 Args: carbon_percent: 有机碳百分比 Returns: 离散状态标签 """ return self.discretize('organic_carbon', carbon_percent) def discretize_dist_to_river(self, distance_m: float) -> str: """ 离散化距离河道距离 Args: distance_m: 距离（米） Returns: 离散状态标签 """ return self.discretize('dist_to_river', distance_m) def discretize_dist_to_fault(self, distance_m: float) -> str: """ 离散化距离断裂带距离 Args: distance_m: 距离（米） Returns: 离散状态标签 """ return self.discretize('dist_to_fault', distance_m) def discretize_pipe_density(self, density: float, region_code: Optional[str] = None) -> str: """ 离散化供水管网密度 Args: density: 管网密度（m/m²） region_code: 行政区划代码 Returns: 离散状态标签 """ return self.discretize('pipe_density', density, region_code) def discretize_all_factors(self, factors: Dict[str, Any], region_code: Optional[str] = None) -> Dict[str, str]: """ 离散化所有因子 Args: factors: 因子字典，key为因子名称，value为连续值 region_code: 行政区划代码 Returns: 离散化后的因子字典 """ result = {} # 暴雨触发层 if 'rain_intensity' in factors: result['rain_intensity'] = self.discretize_rain_intensity(factors['rain_intensity']) if 'duration' in factors: result['duration'] = self.discretize_duration(factors['duration']) if 'accum_rain' in factors: result['accum_rain'] = self.discretize_accum_rain(factors['accum_rain']) # 地震触发层 if 'magnitude' in factors: result['magnitude'] = self.discretize_magnitude(factors['magnitude']) if 'epicenter_distance' in factors: result['epicenter_distance'] = self.discretize_epicenter_distance(factors['epicenter_distance']) if 'seismic_intensity' in factors: result['seismic_intensity'] = self.discretize_seismic_intensity(factors['seismic_intensity']) # 环境层 if 'elevation' in factors: result['elevation'] = self.discretize_elevation(factors['elevation'], region_code) if 'slope' in factors: result['slope'] = self.discretize_slope(factors['slope']) if 'aspect' in factors: result['aspect'] = self.discretize_aspect(factors['aspect']) if 'soil_type' in factors: result['soil_type'] = self.discretize_soil_type(factors['soil_type']) if 'lithology' in factors: result['lithology'] = self.discretize_lithology(factors['lithology']) if 'landuse' in factors: result['landuse'] = self.discretize_landuse(factors['landuse']) if 'terrain' in factors: result['terrain'] = self.discretize_terrain(factors['terrain']) if 'impervious' in factors: result['impervious'] = self.discretize_impervious(factors['impervious']) if 'ndvi' in factors: result['ndvi'] = self.discretize_ndvi(factors['ndvi']) if 'sand_content' in factors: result['sand_content'] = self.discretize_sand_content(factors['sand_content']) if 'ph' in factors: result['ph'] = self.discretize_ph(factors['ph']) if 'soil_moisture' in factors: result['soil_moisture'] = self.discretize_soil_moisture(factors['soil_moisture']) if 'organic_carbon' in factors: result['organic_carbon'] = self.discretize_organic_carbon(factors['organic_carbon']) if 'dist_to_river' in factors: result['dist_to_river'] = self.discretize_dist_to_river(factors['dist_to_river']) if 'dist_to_fault' in factors: result['dist_to_fault'] = self.discretize_dist_to_fault(factors['dist_to_fault']) if 'pipe_density' in factors: result['pipe_density'] = self.discretize_pipe_density(factors['pipe_density'], region_code) return result # 创建全局实例 discretizer = Discretizer()