"""Graph Model. """ from enum import Enum from systemrelation import graph_model from systemrelation import systemdatautils from systemrelation import systemutils from systemrelation import revit_const class SystemVertex: """Graph Vertex. """ vertex_type = 0 def __init__(self,parent, connector,system_data): self.connector=connector self.system_data = system_data self.parent= parent def __str__(self): return str(systemutils.get_specify_type_parent(self,FloorGraph).model.floor_name)+':'+str(self.system_data.source_id) def __eq__(self, other): return self.system_data.source_id==other.system_data.source_id def __hash__(self): return hash(self.system_data.source_id) def is_open_vertical_pipe(self): ''' Connector to other floor :return: ''' if (len(self.connector.connect_id_json)==0): if( self.connector.belong_type in revit_const.MEPCURVE): return True return False class SystemEdge: """Graph Edge.""" # edge data.contain all edge element flow_direction = 0 def __init__(self,parent, system_data): self.system_data = system_data self.parent = parent def get_vertexs(self): return [self.start_vertex,self.end_vertex] # @property # def start_vertex(self): # return self.start_vertex # # @property # def end_vertex(self): # return self.end_vertex class SystemGraph: """System Graph """ # system edge array. system_graph_id = '' # system type Name system_name = "" def __init__(self,parent): self.system_edges = [] self.parent=parent def add_edge(self, c1, c2, elements): """ Create edge and add to system_edges list :param c1: :param c2: :param elements: :return: """ edge = SystemEdge(self,elements) edge.start_vertex = SystemVertex(edge,c1,elements[0]) edge.end_vertex = SystemVertex(edge,c2,elements[-1]) self.system_edges.append(edge) return edge def get_open_vertical_pipes(self): """ Get all open vertical pipe Vertex :return: """ open_pipes=[] for edge in self.system_edges: for vertex in edge.get_vertexs(): if vertex.is_open_vertical_pipe(): open_pipes.append(vertex) return open_pipes class FloorGraph: """FLoor Graph. May contain one or more SystemGraph. """ def __init__(self,model, system_name, domain): self.model=model; self.system_name = system_name self.domain = domain floor_connectors = systemdatautils.get_connectors_data(self.model.file_id, system_name, domain) floor_connectors = systemutils.sort_connectors(floor_connectors) floor_connectors = list(map(systemutils.add_attr_is_used, floor_connectors)) self.connector_data = (floor_connectors) self.element_data = systemutils.list_to_dict(systemdatautils.get_element_data(self.model.file_id)) self.system_graphs = [] def get_floor_graphs(self): """Get current floor graph. system_name,domain """ for connector in self.connector_data: system_graph = graph_model.SystemGraph(self) graph_next_connectors = [connector] # dept-first for connector in graph_next_connectors: self.get_path(connector, system_graph, graph_next_connectors) if len(system_graph.system_edges) != 0: self.system_graphs.append(system_graph) print("Next loop:",connector.source_id) return self.system_graphs def get_path(self, connector, system_graph, graph_next_connectors): """ Get Graph_Edge start with connector,end with triple ,cross,equipment. :param connector: :return:Graph_Edge """ system_edge = None if (connector.is_used): return system_edge try: #Add other side to be spreaded graph_next_connectors.extend(self.get_other_connectors(connector)) edge_elements = [self.get_element(connector.belong)] current_connector = connector while (current_connector is not None): current_connector.is_used = True connecteds = self.get_connecteds(current_connector.connect_id_json) if len(connecteds) > 1: raise Exception("connected to many.connector_id:", current_connector.connect_id_json) #Open Connector if len(connecteds) == 0: if current_connector != connector: system_edge = system_graph.add_edge(connector, current_connector, edge_elements) break next_connector = connecteds[0] if next_connector is not None and next_connector.domain == connector.domain and next_connector.mep_system_type == connector.mep_system_type: next_element = self.get_element(next_connector.belong) edge_elements.append(next_element) other_connectors = self.get_other_connectors(next_connector) if systemutils.is_break_condition(next_element) or len(other_connectors) != 1: print(next_connector.source_id) next_connector.is_used = True if(len(other_connectors) != 1): graph_next_connectors.extend(other_connectors) system_edge = system_graph.add_edge(connector, next_connector, edge_elements) break if len(other_connectors)==1: next_connector.is_used = True current_connector = other_connectors[0] continue current_connector = None except Exception as e: print(e) return system_edge def get_other_connectors(self, use_connector): """ Get the connector owner other connectors. belong,domian,mep_system_type same. is_used==False. :param use_connector: :return: """ connectors = [connector for connector in self.connector_data if connector != use_connector and connector.belong == use_connector.belong and connector.domain == use_connector.domain and connector.mep_system_type == use_connector.mep_system_type] connectors = [connector for connector in connectors if bool(1 - connector.is_used)] return connectors def get_connecteds(self, connetor_id): """ Get Ref Connector. may have two.like system logic connector :param connetor_id: :return: """ connectors = list(self.get_connector(id) for id in connetor_id) return connectors def get_connector(self, connector_id): """ Get Connector by Id. :param connector_id: :return: """ cons = list(c for c in self.connector_data if c.id == connector_id) return cons[0] def get_element(self, element_id): """ Get Element by Id. :param eleemnt_id: :return: """ return self.element_data[element_id] class GroupedSystemGraph: ''' Systemgraph be connected via vertical pipe May contain more system.Different floor; ''' def __init__(self, systemgraph): self.systemgraphs=[systemgraph] self.connectedges=[] def add_system(self,systemgraph): ''' Add system into groupsystem :param systemgraph: :return: ''' self.systemgraphs.append(systemgraph) def add_edge(self,vertex1,vertex2): ''' Create virtual edge that connect two floor. Add edge to edge list. :param vertex1: :param vertex2: :return: ''' edge=SystemEdge(None,None) edge.start_vertex=vertex1 edge.end_vertex=vertex2 self.connectedges.append(edge) class ProjectGraph: ''' ProjectGraph contain more than one GroupedSystemGraph. ''' def __init__(self): self.groupedsystemgraphs=[] def remove_groupedsystemgraphs(self, systemgraph): ''' Check system has existed in grouped system list. if exist remove it . :param systemgraph: :return: ''' try: tt = next(groupedsystemgraph for groupedsystemgraph in self.groupedsystemgraphs if systemgraph in groupedsystemgraph.systemgraphs) if tt: self.groupedsystemgraphs.remove(tt) except StopIteration: pass class SystemFlowDirection(Enum): """Edge Flow Direction. """ flow_in = 1 flow_out = -1 flow_undefine = 0 class SystemVertexType(Enum): """Vertex Type. """ real_point = 0 imaginary_point = 1 hollow_dot = 2