# 计算各个建筑内不同楼层的业务空间在垂直方向的重叠关系 (业务空间的垂直交通关系?) ## 前置条件 1. 楼层下有业务空间 2. 业务空间有轮廓 ## 处理逻辑 1. 因为算法复杂度问题, 该算法分为两个版本, 区别为参与计算的业务空间的范围 版本一 (参与计算的业务空间 : 项目下全部的业务空间都参与计算): 1). 获取项目下所有有楼层关系, 并且有轮廓的业务空间 2). 将业务空间按照所属楼层, 业务空间类型分组 3). 以一群人两两握手的方式将两两楼层内的相同类型的业务空间做轮廓是否重叠判断 4). 如果有重叠, 则认为两业务空间有垂直方向的重叠关系 5). 删除上次自动计算的关系, 并插入新计算的关系 版本二 (参与计算的业务空间 : 指定业务空间类型下的若干个业务空间 与某一个建筑的内其他楼层的相同类型的业务空间): 1). 获取到指定类型的所有业务空间, 按照楼层分类 2). 如果某一层下有指定的业务空间, 则该楼层计算与其他楼层的垂直关系时, 那么该楼层只有指定的业务空间才参与计算轮廓重叠 3). 则计算结果一定是只与指定的业务空间有关系 4). 删除与指定的业务空间相关的关系, 并插入新计算的关系 # 函数 ## 版本1
版本1 : 计算一个项目内, 所有业务空间类型的业务空间垂直交通关系 ``` CREATE OR REPLACE FUNCTION "public"."is_vertically_overlap"("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: 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: return False return False # 是否垂直方向上面积有重叠 def is_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 # building -> floor -> object_type -> [space_id] def compose_dict(zone_data): building_map = dict() for row in zone_data: building_id = row['building_id'] floor_id = row['floor_id'] object_type = row['object_type'] if building_id not in building_map: building_map[building_id] = dict() floor_map = building_map[building_id] if floor_id not in floor_map: floor_map[floor_id] = dict() type_map = floor_map[floor_id] if object_type not in type_map: type_map[object_type] = [] arr = type_map[object_type] arr.append(row) return building_map try: # 获取所有建筑, for循环获取每个建筑下所有的业务空间, 按楼层分类 zone_plan = plpy.prepare("SELECT rel.space_id, fl.building_id, rel.floor_id, rel.object_type, sp.outline FROM r_sp_in_fl rel LEFT JOIN public.floor fl on fl.id = rel.floor_id left join zone_space_base sp on rel.space_id = sp.id where rel.project_id = $1 and sp.outline is not null", ["text"]) zone_data = zone_plan.execute([project_id]) if len(zone_data) <2: return True row_map = compose_dict(zone_data) space_outline_json_map = dict() result_arr = [] # 每个楼层的每个业务空间分别和别的楼层的每个业务空间判断is_vertically_overlap # 将结果是true的两个业务空间保存起来 for building_id, floor_map in row_map.items(): checked_floor = set() for floor_id, type_map in floor_map.items(): checked_floor.add(floor_id) for object_type, row_arr in type_map.items(): # 要被对比的楼层 for other_floor_id in floor_map.keys(): if other_floor_id in checked_floor: continue other_type_map = floor_map.get(other_floor_id) if object_type not in other_type_map: continue other_row_arr = other_type_map.get(object_type) for row in row_arr: for other_row in other_row_arr: space_id = row['space_id'] other_space_id = other_row['space_id'] if space_id == other_space_id: continue if space_id not in space_outline_json_map: outline_json = json.loads(row['outline']) space_outline_json_map[space_id] = outline_json if other_space_id not in space_outline_json_map: other_outline_json = json.loads(other_row['outline']) space_outline_json_map[other_space_id] = other_outline_json outline = space_outline_json_map[space_id] other_outline = space_outline_json_map[other_space_id] if is_vertically_overlap(outline, other_outline): single_result = [] single_result.append(space_id) single_result.append(other_space_id) single_result.append(object_type) result_arr.append(single_result) if len(result_arr) == 0: return True # 删除旧业务空间的垂直交通关系(自动计算的), 添加新关系 # 将下面对数据库的操作作为一个事务, 出异常则自动rollback with plpy.subtransaction(): del_plan = plpy.prepare("delete from r_sp_vertical_sp where project_id = $1 and sign = 2", ["text"]) del_plan.execute([project_id]) for single_result in result_arr: del_manual_plan = plpy.prepare("delete from r_sp_vertical_sp where (space_id = $1 and space_other_id = $2) or (space_other_id = $1 and space_id = $2)", ["text", "text"]) del_manual_plan.execute([single_result[0], single_result[1]]) insert_plan = plpy.prepare("insert into r_sp_vertical_sp(space_id, space_other_id, project_id, sign, object_type) values($1, $2, $3, 2, $4)", ["text", "text", "text", "text"]) insert_plan.execute([single_result[0], single_result[1], project_id, single_result[2]]) return True except Exception as e: plpy.info(e) return False $BODY$ LANGUAGE plpython3u VOLATILE COST 100 ```
## 版本2
版本2 : 计算一个建筑内, 同一业务空间类型下的某几个业务空间的垂直交通关系 ``` CREATE OR REPLACE FUNCTION "public"."is_vertically_overlap"("project_id" varchar, "_building_id" varchar, "zone_type" varchar, "space_ids" _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: 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: return False return False # 是否垂直方向上面积有重叠 def is_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 # building_map 结构 building -> floor -> object_type -> [space_id] def compose_dict(zone_data): building_map = dict() for row in zone_data: building_id = row['building_id'] floor_id = row['floor_id'] object_type = row['object_type'] if building_id not in building_map: building_map[building_id] = dict() floor_map = building_map[building_id] if floor_id not in floor_map: floor_map[floor_id] = dict() type_map = floor_map[floor_id] if object_type not in type_map: type_map[object_type] = [] arr = type_map[object_type] arr.append(row) return building_map def build_sql_str(space_ids_set): sql_str = "" for space_id in space_ids_set: sql_str += '\'' + space_id + '\',' if sql_str.endswith(','): sql_str = sql_str[0:-1] return sql_str try: if len(space_ids) < 1: plpy.info("no input space_ids") return True # 获取建筑下所有的业务空间, 按楼层分类 zone_plan = plpy.prepare("SELECT rel.space_id, fl.building_id, rel.floor_id, rel.object_type, sp.outline FROM r_sp_in_fl rel LEFT JOIN public.floor fl on fl.id = rel.floor_id " + "left join zone_space_base sp on rel.space_id = sp.id where rel.project_id = $1 and sp.outline is not null and fl.building_id = $2 and sp.object_type = $3", ["text", "text", "text"]) zone_data = zone_plan.execute([project_id, _building_id, zone_type]) plpy.info('all space size : {0}'.format(len(zone_data))) if len(zone_data) <2: plpy.info("no enough space to calc") return True # 转换space_ids为set类型 space_ids_set = set(space_ids) row_map = compose_dict(zone_data) space_outline_json_map = dict() result_arr = [] # 每个楼层的每个业务空间分别和别的楼层的每个业务空间判断is_vertically_overlap # 将结果是true的两个业务空间保存起来 for building_id, floor_map in row_map.items(): checked_floor = set() for floor_id, type_map in floor_map.items(): checked_floor.add(floor_id) for object_type, row_arr in type_map.items(): # 要被对比的楼层 for other_floor_id in floor_map.keys(): if other_floor_id in checked_floor: continue other_type_map = floor_map.get(other_floor_id) if object_type not in other_type_map: continue other_row_arr = other_type_map.get(object_type) for row in row_arr: for other_row in other_row_arr: space_id = row['space_id'] other_space_id = other_row['space_id'] if space_id not in space_ids_set and other_space_id not in space_ids_set: continue if space_id == other_space_id: continue if space_id not in space_outline_json_map: outline_json = json.loads(row['outline']) space_outline_json_map[space_id] = outline_json if other_space_id not in space_outline_json_map: other_outline_json = json.loads(other_row['outline']) space_outline_json_map[other_space_id] = other_outline_json outline = space_outline_json_map[space_id] other_outline = space_outline_json_map[other_space_id] if is_vertically_overlap(outline, other_outline): single_result = [] single_result.append(space_id) single_result.append(other_space_id) single_result.append(object_type) result_arr.append(single_result) # if len(result_arr) == 0: # return True # 删除旧业务空间的垂直交通关系(自动计算的), 添加新关系 # 将下面对数据库的操作作为一个事务, 出异常则自动rollback with plpy.subtransaction(): sql_str = build_sql_str(space_ids_set) del_plan = plpy.prepare("delete from r_sp_vertical_sp where space_id in ({0}) or space_other_id in ({0})".format(sql_str), []) del_plan.execute([]) for single_result in result_arr: #del_manual_plan = plpy.prepare("delete from r_sp_vertical_sp where (space_id = $1 and space_other_id = $2) or (space_other_id = $1 and space_id = $2)", ["text", "text"]) #del_manual_plan.execute([single_result[0], single_result[1]]) insert_plan = plpy.prepare("insert into r_sp_vertical_sp(space_id, space_other_id, project_id, sign, object_type) values($1, $2, $3, 2, $4)", ["text", "text", "text", "text"]) #plpy.info(single_result) insert_plan.execute([single_result[0], single_result[1], project_id, single_result[2]]) return True except Exception as e: plpy.info(e) return False $BODY$ LANGUAGE plpython3u VOLATILE COST 100 ```
## 输入 1. 版本一 1). 项目id 2. 版本二 1). 项目id 2). 建筑id 3). 业务空间类型 4). 指定参与计算的业务空间id ## 返回结果 true 成功 false 失败