该项目是基于GIS空间分析和遗传算法的地下管线智能选线系统。包含以下核心功能模块：

1)数据模型模块，用于加载和管理现状管线、地形等数据；

2)空间分析模块，计算路径冲突、地形成本等指标；

3)遗传算法优化模块，通过DEAP框架实现路径优化；

4)可视化模块，生成规划结果的多维分析图表。

        系统采用多目标优化方法，综合考虑路径长度、地形复杂度、管线冲突等因素，能够自动生成3种最优路径方案，并提供详细的技术参数和风险评估。

一、系统安装

基础依赖

pip install geopandas shapely networkx deap numpy pandas matplotlib

可选增强包

pip install rasterio fiona pyproj folium plotly

数据库支持（可选）

pip install psycopg2-binary sqlalchemy

二、完整代码

"""
地下管线智能选线系统
基于GIS空间分析和遗传算法的管线路径优化系统
版本: 1.0
"""

import numpy as np
import pandas as pd
import geopandas as gpd
from shapely.geometry import Point, LineString, Polygon
from shapely.ops import nearest_points
import networkx as nx
from deap import creator, base, tools, algorithms
import random
import math
import matplotlib.pyplot as plt
import matplotlib.patches as patches
from typing import List, Tuple, Dict, Optional
import warnings
warnings.filterwarnings('ignore')

# =====================================================================
# 1. 数据模型定义
# =====================================================================

class PipelineData:
    """管线数据类"""
    def __init__(self):
        self.existing_pipes = None  # 现状管线数据
        self.terrain_data = None    # 地形数据
        self.obstacles = None       # 障碍物数据
        self.cost_matrix = None     # 成本矩阵
        
    def load_existing_pipes(self, data_path: str = None):
        """加载现状管线数据"""
        if data_path:
            self.existing_pipes = gpd.read_file(data_path)
        else:
            # 模拟现状管线数据
            pipes_data = []
            for i in range(20):
                start_x = random.uniform(0, 1000)
                start_y = random.uniform(0, 1000)
                end_x = start_x + random.uniform(-200, 200)
                end_y = start_y + random.uniform(-200, 200)
                
                pipe = {
                    'id': f'PIPE_{i:03d}',
                    'geometry': LineString([(start_x, start_y), (end_x, end_y)]),
                    'pipe_type': random.choice(['供水', '燃气', '电力', '通讯', '污水']),
                    'diameter': random.choice([200, 300, 400, 500, 600]),
                    'depth': random.uniform(1.0, 3.0),
                    'material': random.choice(['PE', '钢管', 'PVC', '铸铁']),
                    'status': random.choice(['正常', '老化', '待维修'])
                }
                pipes_data.append(pipe)
                
            self.existing_pipes = gpd.GeoDataFrame(pipes_data)
            
    def load_terrain_data(self):
        """加载地形数据（模拟）"""
        # 创建地形网格
        x = np.linspace(0, 1000, 100)
        y = np.linspace(0, 1000, 100)
        X, Y = np.meshgrid(x, y)
        
        # 模拟地形高程（山丘地形）
        Z = (np.sin(X/100) * np.cos(Y/100) * 20 + 
             np.sin(X/200) * 15 + 
             np.random.normal(0, 2, X.shape))
             
        self.terrain_data = {
            'x': X, 'y': Y, 'elevation': Z,
            'soil_type': np.random.choice(['粘土', '砂土', '岩石'], X.shape, 
                                        p=[0.4, 0.4, 0.2])
        }
        
    def create_cost_matrix(self, grid_size: int = 100):
        """创建成本矩阵"""
        self.cost_matrix = np.ones((grid_size, grid_size))
        
        # 基于土质设置基础成本
        for i in range(grid_size):
            for j in range(grid_size):
                soil = self.terrain_data['soil_type'][i, j]
                if soil == '岩石':
                    self.cost_matrix[i, j] = 3.0
                elif soil == '粘土':
                    self.cost_matrix[i, j] = 1.5
                else:  # 砂土
                    self.cost_matrix[i, j] = 1.0


# =====================================================================
# 2. 空间分析模块
# =====================================================================

class SpatialAnalyzer:
    """GIS空间分析类"""
    
    def __init__(self, pipeline_data: PipelineData):
        self.data = pipeline_data
        
    def calculate_conflicts(self, route: LineString, buffer_distance: float = 5.0) -> int:
        """计算路径与现状管线的冲突数量"""
        route_buffer = route.buffer(buffer_distance)
        conflicts = 0
        
        for _, pipe in self.data.existing_pipes.iterrows():
            if route_buffer.intersects(pipe.geometry):
                # 计算交叉深度冲突
                depth_diff = abs(1.5 - pipe['depth'])  # 假设新管线深度1.5m
                if depth_diff < 0.5:  # 深度差小于50cm认为有冲突
                    conflicts += 1
                    
        return conflicts
        
    def calculate_terrain_cost(self, route: LineString) -> float:
        """计算地形成本"""
        coords = list(route.coords)
        total_cost = 0.0
        
        for i in range(len(coords) - 1):
            x1, y1 = coords[i]
            x2, y2 = coords[i + 1]
            
            # 计算段长
            segment_length = math.sqrt((x2-x1)**2 + (y2-y1)**2)
            
            # 获取地形成本（简化处理）
            grid_x = int(x1 / 10)  # 转换为网格坐标
            grid_y = int(y1 / 10)
            
            if 0 <= grid_x < 100 and 0 <= grid_y < 100:
                terrain_cost = self.data.cost_matrix[grid_y, grid_x]
                total_cost += segment_length * terrain_cost
                
        return total_cost
        
    def calculate_slope_penalty(self, route: LineString) -> float:
        """计算坡度惩罚"""
        coords = list(route.coords)
        total_penalty = 0.0
        
        for i in range(len(coords) - 1):
            x1, y1 = coords[i]
            x2, y2 = coords[i + 1]
            
            # 获取高程
            grid_x1, grid_y1 = int(x1/10), int(y1/10)
            grid_x2, grid_y2 = int(x2/10), int(y2/10)
            
            if (0 <= grid_x1 < 100 and 0 <= grid_y1 < 100 and 
                0 <= grid_x2 < 100 and 0 <= grid_y2 < 100):
                
                elev1 = self.data.terrain_data['elevation'][grid_y1, grid_x1]
                elev2 = self.data.terrain_data['elevation'][grid_y2, grid_x2]
                
                distance = math.sqrt((x2-x1)**2 + (y2-y1)**2)
                if distance > 0:
                    slope = abs(elev2 - elev1) / distance
                    if slope > 0.1:  # 坡度超过10%增加惩罚
                        total_penalty += slope * 100
                        
        return total_penalty


# =====================================================================
# 3. 遗传算法模块
# =====================================================================

class GeneticRouteOptimizer:
    """基于遗传算法的路径优化器"""
    
    def __init__(self, start_point: Tuple[float, float], 
                 end_point: Tuple[float, float],
                 spatial_analyzer: SpatialAnalyzer,
                 grid_size: int = 50):
        self.start_point = start_point
        self.end_point = end_point
        self.analyzer = spatial_analyzer
        self.grid_size = grid_size
        
        # 创建搜索网格
        self.create_search_grid()
        
        # 设置DEAP
        self.setup_deap()
        
    def create_search_grid(self):
        """创建路径搜索网格"""
        x_min, x_max = 0, 1000
        y_min, y_max = 0, 1000
        
        x_coords = np.linspace(x_min, x_max, self.grid_size)
        y_coords = np.linspace(y_min, y_max, self.grid_size)
        
        self.grid_points = []
        for x in x_coords:
            for y in y_coords:
                self.grid_points.append((x, y))
                
    def setup_deap(self):
        """设置DEAP遗传算法框架"""
        # 定义适应度（最小化问题）
        creator.create("FitnessMin", base.Fitness, weights=(-1.0,))
        creator.create("Individual", list, fitness=creator.FitnessMin)
        
        self.toolbox = base.Toolbox()
        
        # 注册遗传算子
        self.toolbox.register("indices", self.generate_random_route)
        self.toolbox.register("individual", tools.initIterate, 
                            creator.Individual, self.toolbox.indices)
        self.toolbox.register("population", tools.initRepeat, 
                            list, self.toolbox.individual)
        
        self.toolbox.register("evaluate", self.evaluate_route)
        self.toolbox.register("mate", self.crossover)
        self.toolbox.register("mutate", self.mutate)
        self.toolbox.register("select", tools.selTournament, tournsize=3)
        
    def generate_random_route(self) -> List[int]:
        """生成随机路径（网格点索引序列）"""
        # 选择3-8个中间节点
        num_waypoints = random.randint(3, 8)
        
        # 找到起点和终点在网格中的索引
        start_idx = self.find_nearest_grid_point(self.start_point)
        end_idx = self.find_nearest_grid_point(self.end_point)
        
        # 生成中间路径点
        route = [start_idx]
        
        for _ in range(num_waypoints):
            # 在起终点之间的区域随机选择点
            candidate_points = []
            for i, point in enumerate(self.grid_points):
                if self.is_point_in_corridor(point, self.start_point, self.end_point):
                    candidate_points.append(i)
                    
            if candidate_points:
                route.append(random.choice(candidate_points))
                
        route.append(end_idx)
        return route
        
    def find_nearest_grid_point(self, point: Tuple[float, float]) -> int:
        """找到最近的网格点索引"""
        min_dist = float('inf')
        nearest_idx = 0
        
        for i, grid_point in enumerate(self.grid_points):
            dist = math.sqrt((point[0] - grid_point[0])**2 + 
                           (point[1] - grid_point[1])**2)
            if dist < min_dist:
                min_dist = dist
                nearest_idx = i
                
        return nearest_idx
        
    def is_point_in_corridor(self, point: Tuple[float, float],
                           start: Tuple[float, float], 
                           end: Tuple[float, float]) -> bool:
        """判断点是否在起终点连线的走廊范围内"""
        # 计算点到起终点连线的距离
        line = LineString([start, end])
        point_geom = Point(point)
        
        distance = point_geom.distance(line)
        return distance < 200  # 200米走廊宽度
        
    def evaluate_route(self, individual: List[int]) -> Tuple[float]:
        """评估路径适应度"""
        if len(individual) < 2:
            return (float('inf'),)
            
        # 构建路径几何
        route_points = [self.grid_points[i] for i in individual]
        
        try:
            route_line = LineString(route_points)
        except:
            return (float('inf'),)
            
        # 计算多目标成本
        length_cost = route_line.length
        terrain_cost = self.analyzer.calculate_terrain_cost(route_line)
        conflict_cost = self.analyzer.calculate_conflicts(route_line) * 1000
        slope_penalty = self.analyzer.calculate_slope_penalty(route_line)
        
        # 综合适应度
        total_cost = (length_cost * 0.3 + 
                     terrain_cost * 0.3 + 
                     conflict_cost * 0.3 + 
                     slope_penalty * 0.1)
                     
        return (total_cost,)
        
    def crossover(self, ind1: List[int], ind2: List[int]) -> Tuple[List[int], List[int]]:
        """路径交叉算子"""
        if len(ind1) < 3 or len(ind2) < 3:
            return ind1, ind2
            
        # 保持起终点不变，交叉中间部分
        size1, size2 = len(ind1) - 2, len(ind2) - 2
        
        if size1 > 0 and size2 > 0:
            cx_point1 = random.randint(1, min(size1, size2))
            cx_point2 = random.randint(cx_point1, min(size1, size2))
            
            # 交叉中间段
            ind1[1:cx_point2+1], ind2[1:cx_point2+1] = \
                ind2[1:cx_point2+1], ind1[1:cx_point2+1]
                
        return ind1, ind2
        
    def mutate(self, individual: List[int]) -> Tuple[List[int]]:
        """路径变异算子"""
        if len(individual) < 3:
            return (individual,)
            
        # 随机选择一个中间点进行变异
        if len(individual) > 2:
            mut_idx = random.randint(1, len(individual) - 2)
            
            # 在附近区域选择新的网格点
            current_point = self.grid_points[individual[mut_idx]]
            candidates = []
            
            for i, point in enumerate(self.grid_points):
                dist = math.sqrt((point[0] - current_point[0])**2 + 
                               (point[1] - current_point[1])**2)
                if dist < 100 and i != individual[mut_idx]:  # 100米范围内
                    candidates.append(i)
                    
            if candidates:
                individual[mut_idx] = random.choice(candidates)
                
        return (individual,)
        
    def optimize_routes(self, population_size: int = 100, 
                       generations: int = 500) -> List[Dict]:
        """运行遗传算法优化，返回3条最优路径"""
        
        # 创建初始种群
        population = self.toolbox.population(n=population_size)
        
        # 评估初始种群
        fitnesses = list(map(self.toolbox.evaluate, population))
        for ind, fit in zip(population, fitnesses):
            ind.fitness.values = fit
            
        # 进化统计
        stats = tools.Statistics(lambda ind: ind.fitness.values)
        stats.register("avg", np.mean)
        stats.register("min", np.min)
        stats.register("max", np.max)
        
        # 运行遗传算法
        population, logbook = algorithms.eaSimple(
            population, self.toolbox, cxpb=0.8, mutpb=0.2,
            ngen=generations, stats=stats, verbose=True
        )
        
        # 获取最优解
        population.sort(key=lambda x: x.fitness.values[0])
        
        # 返回前3个最优解
        best_routes = []
        for i in range(min(3, len(population))):
            individual = population[i]
            route_points = [self.grid_points[idx] for idx in individual]
            route_line = LineString(route_points)
            
            route_info = {
                'route_id': f'路径方案{i+1}',
                'geometry': route_line,
                'waypoints': route_points,
                'total_cost': individual.fitness.values[0],
                'length': route_line.length,
                'conflicts': self.analyzer.calculate_conflicts(route_line),
                'terrain_cost': self.analyzer.calculate_terrain_cost(route_line)
            }
            best_routes.append(route_info)
            
        return best_routes


# =====================================================================
# 4. 主系统类
# =====================================================================

class PipelineRoutingSystem:
    """地下管线智能选线系统主类"""
    
    def __init__(self):
        self.pipeline_data = PipelineData()
        self.spatial_analyzer = None
        self.optimizer = None
        self.results = None
        
    def initialize_system(self):
        """初始化系统"""
        print("正在初始化系统...")
        
        # 加载数据
        self.pipeline_data.load_existing_pipes()
        self.pipeline_data.load_terrain_data()
        self.pipeline_data.create_cost_matrix()
        
        # 初始化空间分析器
        self.spatial_analyzer = SpatialAnalyzer(self.pipeline_data)
        
        print("系统初始化完成！")
        print(f"已加载现状管线: {len(self.pipeline_data.existing_pipes)} 条")
        
    def plan_route(self, start_point: Tuple[float, float], 
                   end_point: Tuple[float, float]) -> List[Dict]:
        """规划路径"""
        print(f"\n开始规划路径: {start_point} -> {end_point}")
        
        # 创建优化器
        self.optimizer = GeneticRouteOptimizer(
            start_point, end_point, self.spatial_analyzer
        )
        
        # 运行优化
        print("运行遗传算法优化...")
        self.results = self.optimizer.optimize_routes(
            population_size=50, generations=100
        )
        
        return self.results
        
    def generate_report(self) -> Dict:
        """生成分析报告"""
        if not self.results:
            return {"error": "尚未执行路径规划"}
            
        report = {
            "规划时间": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"),
            "方案数量": len(self.results),
            "详细方案": []
        }
        
        for i, route in enumerate(self.results):
            route_report = {
                "方案编号": i + 1,
                "路径长度": f"{route['length']:.2f} 米",
                "总成本": f"{route['total_cost']:.2f}",
                "冲突数量": route['conflicts'],
                "地形成本": f"{route['terrain_cost']:.2f}",
                "节点数量": len(route['waypoints']),
                "推荐等级": "优先推荐" if i == 0 else "备选方案"
            }
            report["详细方案"].append(route_report)
            
        return report
        
    def visualize_results(self, figsize: Tuple[int, int] = (15, 10)):
        """可视化结果"""
        if not self.results:
            print("没有结果可视化")
            return
            
        fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=figsize)
        fig.suptitle('地下管线智能选线系统分析结果', fontsize=16, fontweight='bold')
        
        # 1. 现状管线分布
        ax1.set_title('现状管线分布')
        for _, pipe in self.pipeline_data.existing_pipes.iterrows():
            x, y = pipe.geometry.xy
            ax1.plot(x, y, linewidth=2, alpha=0.7, 
                    label=pipe['pipe_type'] if pipe.name < 5 else "")
        ax1.set_xlabel('X坐标 (m)')
        ax1.set_ylabel('Y坐标 (m)')
        ax1.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
        ax1.grid(True, alpha=0.3)
        
        # 2. 地形高程
        terrain = self.pipeline_data.terrain_data
        im = ax2.contourf(terrain['x'], terrain['y'], terrain['elevation'], 
                         levels=20, cmap='terrain', alpha=0.8)
        ax2.set_title('地形高程分布')
        ax2.set_xlabel('X坐标 (m)')
        ax2.set_ylabel('Y坐标 (m)')
        plt.colorbar(im, ax=ax2, label='高程 (m)')
        
        # 3. 优化路径方案
        ax3.set_title('智能选线结果对比')
        colors = ['red', 'blue', 'green']
        
        for i, route in enumerate(self.results):
            x, y = route['geometry'].xy
            ax3.plot(x, y, color=colors[i], linewidth=3, 
                    label=f"{route['route_id']} (成本:{route['total_cost']:.0f})")
            
            # 标记起终点
            ax3.scatter(x[0], y[0], color=colors[i], s=100, marker='o', 
                       edgecolor='black', linewidth=2)
            ax3.scatter(x[-1], y[-1], color=colors[i], s=100, marker='s', 
                       edgecolor='black', linewidth=2)
        
        ax3.set_xlabel('X坐标 (m)')
        ax3.set_ylabel('Y坐标 (m)')
        ax3.legend()
        ax3.grid(True, alpha=0.3)
        
        # 4. 方案对比柱状图
        route_names = [route['route_id'] for route in self.results]
        costs = [route['total_cost'] for route in self.results]
        conflicts = [route['conflicts'] for route in self.results]
        
        x_pos = np.arange(len(route_names))
        width = 0.35
        
        ax4_twin = ax4.twinx()
        
        bars1 = ax4.bar(x_pos - width/2, costs, width, label='总成本', 
                       color='skyblue', alpha=0.8)
        bars2 = ax4_twin.bar(x_pos + width/2, conflicts, width, label='冲突数量', 
                           color='salmon', alpha=0.8)
        
        ax4.set_title('方案成本与冲突对比')
        ax4.set_xlabel('路径方案')
        ax4.set_ylabel('总成本')
        ax4_twin.set_ylabel('冲突数量')
        ax4.set_xticks(x_pos)
        ax4.set_xticklabels(route_names)
        
        # 添加数值标签
        for i, (bar1, bar2) in enumerate(zip(bars1, bars2)):
            ax4.text(bar1.get_x() + bar1.get_width()/2, bar1.get_height() + 50,
                    f'{costs[i]:.0f}', ha='center', va='bottom', fontsize=10)
            ax4_twin.text(bar2.get_x() + bar2.get_width()/2, bar2.get_height() + 0.1,
                         f'{conflicts[i]}', ha='center', va='bottom', fontsize=10)
        
        ax4.legend(loc='upper left')
        ax4_twin.legend(loc='upper right')
        
        plt.tight_layout()
        plt.show()


# =====================================================================
# 5. 使用示例和测试
# =====================================================================

def main():
    """主函数 - 系统使用示例"""
    print("=" * 60)
    print("地下管线智能选线系统 v1.0")
    print("=" * 60)
    
    # 创建系统实例
    routing_system = PipelineRoutingSystem()
    
    # 初始化系统
    routing_system.initialize_system()
    
    # 设置起终点（示例：从西南到东北的管线规划）
    start_point = (100, 150)  # 起点坐标
    end_point = (800, 750)    # 终点坐标
    
    # 执行路径规划
    results = routing_system.plan_route(start_point, end_point)
    
    # 生成报告
    report = routing_system.generate_report()
    
    # 输出结果
    print("\n" + "="*50)
    print("路径规划完成！")
    print("="*50)
    
    print(f"\n规划时间: {report['规划时间']}")
    print(f"生成方案数量: {report['方案数量']}")
    
    print("\n详细方案信息:")
    print("-" * 80)
    print(f"{'方案':<10} {'长度(m)':<12} {'总成本':<12} {'冲突数':<8} {'推荐等级':<12}")
    print("-" * 80)
    
    for plan in report['详细方案']:
        print(f"方案{plan['方案编号']:<9} {plan['路径长度']:<12} "
              f"{plan['总成本']:<12} {plan['冲突数量']:<8} {plan['推荐等级']:<12}")
    
    print("-" * 80)
    
    # 输出技术参数
    print(f"\n技术参数:")
    print(f"- 现状管线数量: {len(routing_system.pipeline_data.existing_pipes)} 条")
    print(f"- 搜索网格规模: 50×50")
    print(f"- 遗传算法参数: 种群50, 迭代100代")
    print(f"- 缓冲区距离: 5.0米")
    
    # 可视化结果
    print(f"\n正在生成可视化图表...")
    routing_system.visualize_results()
    
    print(f"\n系统运行完成！")
    return routing_system

# 高级功能扩展示例
class AdvancedFeatures:
    """高级功能扩展"""
    
    @staticmethod
    def export_to_gis(results: List[Dict], output_path: str):
        """导出结果到GIS格式"""
        gdf_results = gpd.GeoDataFrame(results)
        gdf_results.to_file(output_path, driver='ESRI Shapefile')
        print(f"结果已导出到: {output_path}")
        
    @staticmethod
    def cost_sensitivity_analysis(routing_system, cost_weights: List[Tuple]):
        """成本权重敏感性分析"""
        print("执行成本敏感性分析...")
        sensitivity_results = []
        
        for weights in cost_weights:
            # 重新配置权重并运行优化
            # 这里可以修改适应度函数的权重配置
            print(f"测试权重组合: {weights}")
            
        return sensitivity_results

# =====================================================================
# 6. 系统配置和优化工具
# =====================================================================

class SystemConfig:
    """系统配置类"""
    
    # 遗传算法参数
    GA_PARAMS = {
        'population_size': 100,     # 种群大小
        'generations': 500,         # 迭代次数
        'crossover_prob': 0.8,      # 交叉概率
        'mutation_prob': 0.2,       # 变异概率
        'tournament_size': 3        # 锦标赛选择大小
    }
    
    # 成本权重配置
    COST_WEIGHTS = {
        'length': 0.3,              # 长度成本权重
        'terrain': 0.3,             # 地形成本权重
        'conflict': 0.3,            # 冲突成本权重
        'slope': 0.1                # 坡度惩罚权重
    }
    
    # 空间分析参数
    SPATIAL_PARAMS = {
        'buffer_distance': 5.0,     # 缓冲区距离(米)
        'grid_size': 50,            # 搜索网格大小
        'corridor_width': 200,      # 搜索走廊宽度(米)
        'min_depth_diff': 0.5       # 最小深度差(米)
    }

class PerformanceOptimizer:
    """性能优化工具"""
    
    @staticmethod
    def optimize_for_large_dataset(routing_system, max_pipes: int = 1000):
        """大数据集性能优化"""
        if len(routing_system.pipeline_data.existing_pipes) > max_pipes:
            print(f"检测到大规模数据集，启用性能优化...")
            
            # 空间索引优化
            routing_system.pipeline_data.existing_pipes = \
                routing_system.pipeline_data.existing_pipes.iloc[:max_pipes]
            
            # 网格粗化
            routing_system.optimizer.grid_size = 30
            
            print(f"性能优化完成")
    
    @staticmethod
    def parallel_optimization(routing_system, num_processes: int = 4):
        """并行化优化（多进程）"""
        import multiprocessing as mp
        
        print(f"启用{num_processes}进程并行优化...")
        
        # 这里可以实现多进程版本的遗传算法
        # 将种群分割到不同进程中并行计算
        pass

class ValidationTools:
    """验证工具"""
    
    @staticmethod
    def validate_route_feasibility(route_geometry, existing_pipes, 
                                 min_clearance: float = 1.0):
        """验证路径可行性"""
        issues = []
        
        # 检查与现状管线的间距
        for _, pipe in existing_pipes.iterrows():
            distance = route_geometry.distance(pipe.geometry)
            if distance < min_clearance:
                issues.append(f"与管线{pipe['id']}间距不足: {distance:.2f}m")
        
        # 检查路径合理性
        coords = list(route_geometry.coords)
        for i in range(len(coords)-1):
            segment_length = Point(coords[i]).distance(Point(coords[i+1]))
            if segment_length > 500:  # 单段长度过长
                issues.append(f"段{i+1}长度过长: {segment_length:.2f}m")
        
        return issues
    
    @staticmethod
    def generate_compliance_report(results: List[Dict]):
        """生成合规性报告"""
        compliance_report = {
            "检查时间": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"),
            "合规项目": [],
            "风险提示": []
        }
        
        for route in results:
            route_id = route['route_id']
            
            # 检查技术指标
            if route['conflicts'] == 0:
                compliance_report["合规项目"].append(f"{route_id}: 无管线冲突")
            else:
                compliance_report["风险提示"].append(
                    f"{route_id}: 存在{route['conflicts']}处潜在冲突"
                )
            
            if route['length'] < 2000:
                compliance_report["合规项目"].append(f"{route_id}: 路径长度合理")
            else:
                compliance_report["风险提示"].append(f"{route_id}: 路径较长，需评估经济性")
        
        return compliance_report

# =====================================================================
# 7. 实际项目部署指南
# =====================================================================

class DeploymentGuide:
    """部署指南"""
    
    @staticmethod
    def create_project_structure():
        """创建项目目录结构"""
        import os
        
        directories = [
            "data/input",           # 输入数据
            "data/output",          # 输出结果
            "config",               # 配置文件
            "logs",                 # 日志文件
            "reports",              # 分析报告
            "temp"                  # 临时文件
        ]
        
        for directory in directories:
            os.makedirs(directory, exist_ok=True)
        
        print("项目目录结构创建完成")
    
    @staticmethod
    def generate_config_template():
        """生成配置文件模板"""
        config_template = """
# 地下管线智能选线系统配置文件
# ================================

[DATABASE]
# 数据库连接配置
host = localhost
port = 5432
database = pipeline_gis
user = gis_user
password = your_password

[PATHS]
# 数据路径配置
input_data_path = ./data/input/
output_data_path = ./data/output/
temp_path = ./temp/

[ALGORITHM]
# 算法参数配置
population_size = 100
generations = 500
crossover_rate = 0.8
mutation_rate = 0.2

[CONSTRAINTS]
# 工程约束配置
min_pipe_clearance = 1.0
max_slope = 0.15
min_cover_depth = 0.8
max_cover_depth = 4.0

[COST_MODEL]
# 成本模型配置
excavation_cost_per_m3 = 45.0
pipe_cost_per_m = 120.0
labor_cost_per_day = 800.0
"""
        
        with open("config/system_config.ini", "w", encoding="utf-8") as f:
            f.write(config_template)
        
        print("配置文件模板已生成: config/system_config.ini")

# =====================================================================
# 8. 扩展功能模块
# =====================================================================

class AdvancedAnalytics:
    """高级分析功能"""
    
    @staticmethod
    def route_risk_assessment(route_info: Dict, risk_factors: Dict) -> Dict:
        """路径风险评估"""
        risk_score = 0
        risk_details = []
        
        # 施工风险评估
        if route_info['conflicts'] > 0:
            risk_score += route_info['conflicts'] * 10
            risk_details.append(f"管线冲突风险: {route_info['conflicts']}处")
        
        # 地质风险评估
        terrain_risk = route_info['terrain_cost'] / route_info['length']
        if terrain_risk > 2.0:
            risk_score += 20
            risk_details.append("地质条件复杂")
        
        # 经济风险评估
        cost_per_meter = route_info['total_cost'] / route_info['length']
        if cost_per_meter > 200:
            risk_score += 15
            risk_details.append("施工成本较高")
        
        risk_level = "低风险"
        if risk_score > 30:
            risk_level = "高风险"
        elif risk_score > 15:
            risk_level = "中等风险"
        
        return {
            "risk_score": risk_score,
            "risk_level": risk_level,
            "risk_factors": risk_details
        }
    
    @staticmethod
    def generate_construction_phases(route_geometry, phase_length: float = 200):
        """生成施工分段建议"""
        coords = list(route_geometry.coords)
        phases = []
        
        current_phase = [coords[0]]
        current_length = 0
        
        for i in range(1, len(coords)):
            segment_length = Point(coords[i-1]).distance(Point(coords[i]))
            
            if current_length + segment_length > phase_length and len(current_phase) > 1:
                # 完成当前阶段
                phases.append({
                    "phase_id": len(phases) + 1,
                    "geometry": LineString(current_phase),
                    "length": current_length,
                    "start_point": current_phase[0],
                    "end_point": current_phase[-1]
                })
                
                # 开始新阶段
                current_phase = [coords[i-1]]
                current_length = 0
            
            current_phase.append(coords[i])
            current_length += segment_length
        
        # 添加最后一个阶段
        if len(current_phase) > 1:
            phases.append({
                "phase_id": len(phases) + 1,
                "geometry": LineString(current_phase),
                "length": current_length,
                "start_point": current_phase[0],
                "end_point": current_phase[-1]
            })
        
        return phases

# =====================================================================
# 9. 完整使用示例
# =====================================================================

def advanced_example():
    """高级使用示例"""
    print("=" * 60)
    print("地下管线智能选线系统 - 高级功能演示")
    print("=" * 60)
    
    # 1. 创建项目结构
    DeploymentGuide.create_project_structure()
    DeploymentGuide.generate_config_template()
    
    # 2. 初始化系统
    routing_system = PipelineRoutingSystem()
    routing_system.initialize_system()
    
    # 3. 性能优化（针对大数据集）
    PerformanceOptimizer.optimize_for_large_dataset(routing_system)
    
    # 4. 执行路径规划
    start_point = (200, 200)
    end_point = (700, 600)
    
    results = routing_system.plan_route(start_point, end_point)
    
    # 5. 高级分析
    print("\n执行高级风险评估...")
    for route in results:
        risk_analysis = AdvancedAnalytics.route_risk_assessment(
            route, {"soil_condition": "normal", "traffic": "medium"}
        )
        
        print(f"\n{route['route_id']} 风险评估:")
        print(f"  风险等级: {risk_analysis['risk_level']}")
        print(f"  风险评分: {risk_analysis['risk_score']}")
        for factor in risk_analysis['risk_factors']:
            print(f"  - {factor}")
    
    # 6. 施工分段建议
    print(f"\n生成施工分段建议...")
    best_route = results[0]
    phases = AdvancedAnalytics.generate_construction_phases(
        best_route['geometry'], phase_length=150
    )
    
    print(f"推荐分{len(phases)}个施工阶段:")
    for phase in phases:
        print(f"  阶段{phase['phase_id']}: {phase['length']:.1f}米")
    
    # 7. 合规性检查
    validation_issues = ValidationTools.validate_route_feasibility(
        best_route['geometry'], 
        routing_system.pipeline_data.existing_pipes
    )
    
    if validation_issues:
        print(f"\n发现{len(validation_issues)}个潜在问题:")
        for issue in validation_issues:
            print(f"  - {issue}")
    else:
        print(f"\n✅ 路径验证通过，符合技术规范")
    
    # 8. 生成合规性报告
    compliance = ValidationTools.generate_compliance_report(results)
    print(f"\n合规性报告:")
    print(f"  合规项目: {len(compliance['合规项目'])}项")
    print(f"  风险提示: {len(compliance['风险提示'])}项")
    
    # 9. 导出结果（如果安装了相关库）
    try:
        AdvancedFeatures.export_to_gis(results, "data/output/routing_results.shp")
        print(f"\n✅ 结果已导出到GIS格式")
    except Exception as e:
        print(f"\n⚠️  GIS导出失败: {e}")
    
    return routing_system, results

if __name__ == "__main__":
    # 设置matplotlib中文字体（如果需要）
    plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
    plt.rcParams['axes.unicode_minus'] = False
    
    # 选择运行模式
    print("请选择运行模式:")
    print("1. 基础演示")
    print("2. 高级功能演示")
    
    choice = input("请输入选择 (1/2): ").strip()
    
    if choice == "2":
        system, results = advanced_example()
    else:
        system = main()
    
    print(f"\n🎉 系统演示完成！")
    print(f"💡 提示: 你可以根据实际需求修改参数和算法配置")