基本逻辑

point_unit_grouping_api.py

@@ -0,0 +1,148 @@
+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)