123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278 |
- """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
|