import psycopg2 from flask import Flask, jsonify from shapely import wkb from collections import Counter app = Flask(__name__) # 定义 h_xtfx 值的映射关系(数值用于插值计算) h_xtfx_mapping = { "优先保护类": 1, "安全利用类": 2, "严格管控类": 3 } reverse_h_xtfx_mapping = {v: k for k, v in h_xtfx_mapping.items()} def connect_to_database(): """建立数据库连接""" try: connection = psycopg2.connect( user="postgres", password="root", host="localhost", port="5432", database="testdb1" ) return connection except Exception as error: print("数据库连接失败:", error) return None def fetch_unit_and_point_data(connection): """获取单元和点位数据""" if not connection: return [], [] cursor = connection.cursor() try: # 查询单元信息(GID 和 几何图形) unit_query = "SELECT gid, geom FROM unit_ceil" cursor.execute(unit_query) unit_rows = cursor.fetchall() # 查询点位信息(ID、几何图形、h_xtfx 值) point_query = "SELECT id, geom, h_xtfx FROM fifty_thousand_survey_data" cursor.execute(point_query) point_rows = cursor.fetchall() return unit_rows, point_rows except Exception as error: print("数据查询失败:", error) return [], [] finally: cursor.close() def idw_interpolation(points, target_point): """反距离加权插值函数""" total_weight = 0 weighted_sum = 0 power = 2 # 距离权重的幂 for point, value in points: distance = point.distance(target_point) if distance == 0: return value # 距离为0时直接返回该点值 weight = 1 / (distance ** power) total_weight += weight weighted_sum += weight * value return weighted_sum / total_weight if total_weight != 0 else None def check_h_xtfx_values(unit_rows, point_rows): """核心逻辑:判断单元的 h_xtfx 值""" unit_point_mapping = {} for unit_id, unit_geom_wkb in unit_rows: try: unit_geom = wkb.loads(unit_geom_wkb, hex=True) unit_points = [] for point_id, point_geom_wkb, h_xtfx in point_rows: point_geom = wkb.loads(point_geom_wkb, hex=True) if unit_geom.contains(point_geom): unit_points.append((point_geom, h_xtfx)) # 存储几何坐标和 h_xtfx 值 unit_point_mapping[unit_id] = unit_points except Exception as e: print(f"处理单元 {unit_id} 失败:", e) result = {} for unit_id, points in unit_point_mapping.items(): if not points: result[unit_id] = None continue # 无点位,直接设为 None has_strict_control = any(h_xtfx == "严格管控类" for _, h_xtfx in points) h_xtfx_list = [h_xtfx for _, h_xtfx in points] if not has_strict_control: # 无严格管控类:先判断比例是否 ≥80% counter = Counter(h_xtfx_list) most_common, count = counter.most_common(1)[0] if count / len(points) >= 0.8: result[unit_id] = most_common continue # 比例达标,直接输出 # 比例不达标:对优先保护类和安全利用类进行插值 valid_points = [(geom, h_xtfx_mapping[h_xtfx]) for geom, h_xtfx in points if h_xtfx in ["优先保护类", "安全利用类"]] if len(valid_points) < 2: # 有效点位不足,取最常见值(避免插值错误) result[unit_id] = most_common else: unit_center = unit_geom.centroid interpolated = idw_interpolation(valid_points, unit_center) result[unit_id] = ( "优先保护类" if interpolated <= 1.5 else "安全利用类" if interpolated <= 2.5 else "严格管控类" # 理论上不会出现,因无严格管控类点位 ) else: # 存在严格管控类:对所有点位(含严格管控类)进行插值 point_coords = [(geom, h_xtfx_mapping[h_xtfx]) for geom, h_xtfx in points] unit_center = unit_geom.centroid interpolated = idw_interpolation(point_coords, unit_center) result[unit_id] = ( "优先保护类" if interpolated <= 1.5 else "安全利用类" if interpolated <= 2.5 else "严格管控类" ) return result @app.route('/get_h_xtfx_result', methods=['GET']) def get_h_xtfx_result(): """接口:返回单元 h_xtfx 结果""" connection = connect_to_database() if not connection: return jsonify({"error": "数据库连接失败"}) unit_rows, point_rows = fetch_unit_and_point_data(connection) result = check_h_xtfx_values(unit_rows, point_rows) connection.close() return jsonify(result) if __name__ == '__main__': app.run(debug=True)