# -*- coding: utf-8 -*- import json from collections import ChainMap from itertools import groupby from operator import itemgetter import numpy as np import vg from utils import BinaryRelationItem, BinaryRelationCollection np.seterr(divide='ignore', invalid='ignore') def calc_adjacent_relation(columns, segments, v_walls, walls): columns = list(map(load_location, columns)) segments = list(map(load_curve, segments)) v_walls = list(map(load_location, v_walls)) walls = list(map(load_location, walls)) columns_dic = list_to_dict(columns) v_walls_dic = list_to_dict(v_walls) walls_dic = list_to_dict(walls) unit_dic = ChainMap( columns_dic, walls_dic, v_walls_dic, ) grouped_segments = group_segments(segments, unit_dic) binary_space_list = [] space_relation = BinaryRelationCollection() for group in grouped_segments: for i in range(len(group)): for j in range(i, len(group)): if are_adjacent(group[i], group[j], unit_dic): space1 = group[i]['space_id'] space2 = group[j]['space_id'] binary_space_list.append((space1, space2)) if space1 != space2: binary_space_relations = BinaryRelationItem(space1, space2) space_relation.update(binary_space_relations) relations = [] result_key = ['from', 'to'] for relation in space_relation: temp = dict(zip(result_key, relation.split('@'))) relations.append(temp) return relations def group_segments(segments, units): grouped_by_reference = dict() for idx, item in groupby(segments, key=itemgetter('reference')): if idx: if idx in grouped_by_reference: grouped_by_reference[idx] += list(item) else: # print(item) grouped_by_reference[idx] = list(item) binary_list = [] reference_list = list(grouped_by_reference.keys()) for i in range(len(reference_list)): for j in range(i + 1, len(reference_list)): if are_clung(reference_list[i], reference_list[j], units): binary_list.append((reference_list[i], reference_list[j])) results = [] for reference in grouped_by_reference.keys(): merged_group = [] merged_group += grouped_by_reference[reference] for binary in [item for item in binary_list if reference in item]: binary_relation = BinaryRelationItem(binary[0], binary[1]) another = binary_relation.get_another(reference) merged_group += grouped_by_reference[another] results.append(merged_group) return results def list_to_dict(lis): ids = [idx.get('revit_id') for idx in lis] dic = dict(zip(ids, lis)) return dic def are_clung(unit1_id, unit2_id, units): if unit1_id == unit2_id: return False unit1, unit2 = units[unit1_id], units[unit2_id] if unit1.get('type') == unit2.get('type') == 'Wall': if unit1['location']['Type'] == unit2['location']['Type'] == 'Line': location1 = np.array([list(p.values()) for p in unit1['location']['Points']]) location2 = np.array([list(p.values()) for p in unit2['location']['Points']]) v1 = location1[1] - location1[0] v2 = location2[1] - location2[0] # Judge parallel if vg.almost_collinear(v1, v2, atol=1e-08): # Calculate the distance between line segments v3 = location2[1] - location1[0] angle = vg.angle(v1, v3, units='rad') distance = np.around(vg.magnitude(v3) * np.sin(angle), decimals=4) wall_width = (float(unit1['width']) + float(unit2['width'])) / 2.0 if distance <= wall_width: return True return False def are_adjacent(segment1, segment2, units): base = units[segment1['reference']] base_location = base['location'] if base_location['Type'] == 'Line': line1, line2 = segment1['curve'], segment2['curve'] if len(line1) == len(line2) == 2: l1_p1 = np.array(list(line1[0].values())) l1_p2 = np.array(list(line1[1].values())) l2_p1 = np.array(list(line2[0].values())) l2_p2 = np.array(list(line2[1].values())) base_line = base_location['Points'] base_vec = np.array(list(base_line[1].values())) - np.array(list(base_line[0].values())) base_vec = vg.normalize(base_vec) l1_p1_projection = vg.dot(l1_p1, base_vec) l1_p2_projection = vg.dot(l1_p2, base_vec) l2_p1_projection = vg.dot(l2_p1, base_vec) l2_p2_projection = vg.dot(l2_p2, base_vec) projection1_min = min(l1_p1_projection, l1_p2_projection) projection1_max = max(l1_p1_projection, l1_p2_projection) projection2_min = min(l2_p1_projection, l2_p2_projection) projection2_max = max(l2_p1_projection, l2_p2_projection) return projection1_max > projection2_min and projection2_max > projection1_min return False def load_location(x): x['location'] = json.loads(str(x['location']).replace('\'', '"')) return x def load_curve(x): x['curve'] = json.loads(str(x['curve']).replace('\'', '"')) return x