# 竖井贯通关系 ## 前置条件 1. 竖井包含的有业务空间 ## 处理逻辑 1. 两两竖井对比, 前提条件: 对比的两竖井不能包含有相同的业务空间, 否则不能对比. (竖井包含元空间的数据是前端做划分) 2. 满足1之后, 如果其中一个竖井包含的业务空间与另一个竖井包含的业务空间在垂直方向上有重叠, 则认为这两个竖井有贯通关系 ## 函数
源码 ``` -- 竖井贯通关系 CREATE OR REPLACE FUNCTION "public"."rel_sh2sh"("project_id" varchar) RETURNS "pg_catalog"."bool" AS $BODY$ from shapely.geometry import Polygon import json # 获取Polygon对象 def get_polygon(single_poly): poly_len = len(single_poly) poly = [] for i in range(poly_len): pair = single_poly[i] poly.append((pair["X"], pair["Y"])) return Polygon(poly) # 在polygon1包含polygon2的时候, 检测是否polygon1内的空洞也包含polygon2 def is_include(polygon1, poly2): length1 = len(polygon1) for i in range(1, length1): poly1 = get_polygon(polygon1[i]) try: if poly1.overlaps(poly2): return True except Exception as e: plpy.warning('fff + {}'.format(e)) return False if poly1.equals(poly2) or poly1.contains(poly2): return False return True def is_sub_outline_overlap(polygon1, polygon2): try: poly1 = get_polygon(polygon1[0]) poly2 = get_polygon(polygon2[0]) if poly1.overlaps(poly2) or poly1.equals(poly2): return True if poly1.contains(poly2): return is_include(polygon1, poly2) if poly2.contains(poly1): return is_include(polygon2, poly1) except Exception as e: plpy.warning('dddd + {}'.format(e)) return False return False # 是否垂直方向上面积有重叠 def is_sp_vertically_overlap(polygon1, polygon2): length1 = len(polygon1) length2 = len(polygon2) if length1 == 0 or length2 == 0: return False for i in range(length1): for j in range(length2): if is_sub_outline_overlap(polygon1[i], polygon2[j]): return True return False def is_vertically_overlap(sp_set_i, sp_set_j, sp_outline_dict): for sp_i in sp_set_i: for sp_j in sp_set_j: outline_i = sp_outline_dict[sp_i] outline_j = sp_outline_dict[sp_j] if is_sp_vertically_overlap(outline_i, outline_j): return True return False # 计算竖井的贯通关系 def calc_sh2sh(rel_dict, sp_outline_dict): result = [] key_list = list(rel_dict.keys()) for i in range(len(key_list)): for j in range(i + 1, len(key_list)): sh_i = key_list[i] sh_j = key_list[j] sp_i = rel_dict[sh_i] sp_j = rel_dict[sh_j] intersection = sp_i.intersection(sp_j) if len(intersection) > 0: continue is_overlap = is_vertically_overlap(sp_i, sp_j, sp_outline_dict) if is_overlap: result.append((sh_i, sh_j)) return result try: # 将下面对数据库的操作作为一个事务, 出异常则自动rollback with plpy.subtransaction(): delete_plan = plpy.prepare("delete from r_sh_through_sh where project_id = $1 and sign = 2", ["text"]) delete_plan.execute([project_id]) sh_plan = plpy.prepare("select sh_id, sp_id, sp.outline from relationship.r_sh2sp rel inner join zone_space_base sp on sp.id = rel.sp_id where rel.project_id = $1", ["text"]) rel = sh_plan.execute([project_id]) rel_dict = dict() sp_outline_dict = dict() for row in rel: shaft_id = row['sh_id'] sp_id = row['sp_id'] sp_outline = row['outline'] if shaft_id not in rel_dict: rel_dict[shaft_id] = set() rel_dict[shaft_id].add(sp_id) if sp_id not in sp_outline_dict: sp_outline_dict[sp_id] = json.loads(sp_outline) sh2sh_arr = calc_sh2sh(rel_dict, sp_outline_dict) for (shaft1, shaft2) in sh2sh_arr: delete_duplicate_plan = plpy.prepare("delete from r_sh_through_sh where (shaft_id = $1 and shaft_other_id = $2) or (shaft_id = $2 and shaft_other_id = $1)", ["text", "text"]) delete_duplicate_plan.execute([shaft1, shaft2]) plan = plpy.prepare("insert into r_sh_through_sh(shaft_id, shaft_other_id, project_id, sign) values($1, $2, $3, 2)", ["text", "text", "text"]) plan.execute([shaft1, shaft2, project_id]) except Exception as e: plpy.warning(e) return False else: return True $BODY$ LANGUAGE plpython3u VOLATILE COST 100 select public.rel_sh2sh('Pj1102290002') ```
## 入参 1. 项目id ## 例子 select public.rel_sh2sh('Pj1102290002');