"""Graph Model. """ from enum import Enum from systemrelation import graph_model from systemrelation import systemdatautils from systemrelation import systemutils class SystemEdge: """Graph Edge.""" # edge data.contain all edge element flow_direction = 0 def __init__(self, system_data): self.system_data = system_data # @property # def start_vertex(self): # return self.start_vertex # # @property # def end_vertex(self): # return self.end_vertex class SystemVertex: """Graph Vertex. """ vertex_type = 0 def __init__(self, system_data): self.system_data = system_data def __str__(self): return str(self.system_data.source_id) class SystemGraph: """System Graph """ # system edge array. system_graph_id = '' # system type Name system_name = "" def __init__(self): self.system_edges = [] def add_edge(self, c1, c2, elements): edge = SystemEdge(elements) edge.start_vertex = SystemVertex(elements[0]) edge.end_vertex = SystemVertex(elements[-1]) self.system_edges.append(edge) return edge class FloorGraph: """FLoor Graph. May contain one or more SystemGraph. """ def __init__(self, floor_id, system_name, domain): self.floor_id = floor_id self.system_name = system_name self.domain = domain floor_connectors = systemdatautils.get_connectors_data(floor_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(floor_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() 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 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