12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394 |
- '''calc flow direction.
- step1. division block.
- step2. appoint source.
- setp3. calc flow direction.
- '''
- from systemrelation import systemdatautils
- import networkx as nx
- from systemrelation import revit_const
- def calc(project_id,block_id):
- '''
- calc block flow direction.
- :param project_id:
- :param block_id:
- :return:
- '''
- block_datas=systemdatautils.get_connected_block_data(project_id,block_id)
- source_datas=systemdatautils.get_connected_block_source_data(project_id,block_id)
- # get roots
- roots=list(i.source_id for i in source_datas)
- g = nx.Graph()
- for item in block_datas:
- item.direction=None
- g.add_edge(item.id1,item.id2);
- G = g.to_directed()
- # get leaves
- leaves=list(v for v,d in G.out_degree() if d==1)
- G_new=nx.DiGraph()
- try:
- for i in range(len(roots)):
- root=roots[i]
- for leaf in leaves:
- paths = nx.all_simple_paths(G, root, leaf)
- for path in paths:
- # remove contain source path
- if is_path_contain_root(path, roots, root):
- continue
- # set path direction
- for j in range(len(path) - 1):
- G_new.add_edge(path[j], path[j + 1],weight=str(j))
- # add source direct to source path.
- for j in range(i + 1, len(roots)):
- next_root=roots[j]
- paths = list(nx.all_simple_paths(G, root, next_root))
- for path in paths:
- if len(path) == 2:
- G_new.add_edge(root, next_root,weight="0")
- G_new.add_edge(next_root, root,weight="0")
- except Exception as ex:
- print(ex)
- for edge in G_new.edges:
- edge_data=G_new.get_edge_data(edge[0],edge[1])
- weight=edge_data["weight"]
- set_edge_direction(block_datas,edge,weight)
- return block_datas
- def is_path_contain_root(path, roots, cur_root):
- '''
- judge whether path contain root node except cur_root.
- :param path:
- :param roots:
- :param cur_root:
- :return:
- '''
- result = False
- for root in roots:
- if root == cur_root:
- continue
- result = (root in path)
- if result:
- break
- return result
- def set_edge_direction(block_datas,edge,weight):
- '''
- set connected_block direction.
- connector1 to connector2.sign:1
- connector2 to connector1.sign:-1
- both way.sign:0
- un calc.sign:None
- :param block_datas:
- :param edge:
- :return:
- '''
- for block_data in block_datas:
- # connector1 to connector2.sign:1
- if block_data.id1 == edge[0] and block_data.id2 == edge[1]:
- j=1
- block_data.direction=(block_data.direction+j if block_data.direction else j)
- block_data.block_id=str(weight)
- # connector2 to connector1.sing:-1
- if block_data.id1 == edge[1] and block_data.id2 == edge[0]:
- j = -1
- block_data.direction=(block_data.direction+j if block_data.direction else j)
- block_data.block_id = str(weight)
|