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');