ULP-ActivitySupport – blob

1 #!/usr/bin/env python3\r

2 \r

3 import copy\r

4 import json\r

5 import math\r

6 import random\r

7 \r

8 from data import Position\r

9 \r

10 \r

11 class RoutingEngine:\r

12     def __init__(self, context, network):\r

13         self.context = context\r

14         self.network = network\r

15         for edge_id in self.network['edges']:\r

16             nodeA = self.network['edges'][edge_id]['nodes'][0]\r

17             nodeB = self.network['edges'][edge_id]['nodes'][1]\r

18             distance = self.get_node_distance(nodeA, nodeB)\r

19             self.network['edges'][edge_id]['length'] = distance\r

20         self.hops = self.calculate_hops(self.network)\r

21 \r

22     def calculate_hops(self, network):\r

23         result = {}\r

24         for source_node in network['nodes']:\r

25             result[source_node] = {}\r

26             for edge_id in network['edges']:\r

27                 edge = network['edges'][edge_id]\r

28                 target_node = None\r

29                 if edge['nodes'][0] == source_node:\r

30                     target_node = edge['nodes'][1]\r

31                 if edge['nodes'][1] == source_node:\r

32                     target_node = edge['nodes'][0]\r

33                 if target_node is not None:\r

34                     result[source_node][target_node] = edge['length']\r

35             if len(result[source_node].keys()) > 2:\r

36                 self.network['nodes'][source_node]['is_crossroad'] = True\r

37             else:\r

38                 self.network['nodes'][source_node]['is_crossroad'] = False\r

39         return result\r

40 \r

41     def shuffle_mirs(self, density):\r

42         crossing_nodes = [node for node in self.network['nodes'] if self.network['nodes'][node]['is_crossroad']]\r

43         targets = random.sample(crossing_nodes, round(density * len(crossing_nodes)))\r

44         for node in self.network['nodes']:\r

45             self.network['nodes'][node]['has_mir'] = node in targets\r

46 \r

47     def is_confusing(self, node):\r

48         return self.network['nodes'][node]['is_crossroad'] and not self.network['nodes'][node]['has_mir']\r

49 \r

50     def get_network(self):\r

51         return self.network\r

52 \r

53     def get_connected_nodes(self, node):\r

54         return list(self.hops[node].keys())\r

55 \r

56     def get_node_distance(self, nodeA, nodeB):\r

57         coordsA = self.network['nodes'][nodeA]\r

58         coordsB = self.network['nodes'][nodeB]\r

59         return math.sqrt((float(coordsB['x']) - float(coordsA['x'])) ** 2 + (float(coordsB['y']) - float(coordsA['y'])) ** 2)\r

60 \r

61     def get_route_distance(self, nodeA, nodeB):\r

62         route = self.route_between_nodes(nodeA, nodeB)\r

63         distance = 0\r

64         while len(route) >= 2:\r

65             distance += self.get_node_distance(route[0], route[1])\r

66             del route[0]\r

67         return distance\r

68 \r

69     def get_angle(self, x1, y1, x2, y2):\r

70         dotProduct = x1 * x2 + y1 * y2\r

71         lengthProduct = math.sqrt(x1 * x1 + y1 * y1) * math.sqrt(x2 * x2 + y2 * y2)\r

72         cos = dotProduct / lengthProduct\r

73         return math.acos(cos)\r

74 \r

75     def get_next_straight_node(self, previous, current, candidates):\r

76         coordsPrev = self.network['nodes'][previous]\r

77         coordsCur = self.network['nodes'][current]\r

78         direction = [float(coordsCur['x']) - float(coordsPrev['x']), float(coordsCur['y']) - float(coordsPrev['y'])]\r

79         chosen = None\r

80         bestAngle = None\r

81         for candidate in candidates:\r

82             coordsCand = self.network['nodes'][candidate]\r

83             candDir = [float(coordsCand['x']) - float(coordsCur['x']), float(coordsCand['y']) - float(coordsCur['y'])]\r

84             angle = self.get_angle(candDir[0], candDir[1], direction[0], direction[1])\r

85             if bestAngle is None or angle < bestAngle:\r

86                 chosen = candidate\r

87                 bestAngle = angle\r

88         return chosen\r

89 \r

90     def route_between_nodes(self, nodeA, nodeB):\r

91         unvisited = {node: float('inf') for node in self.hops}\r

92         visited = {}\r

93         previous = {}\r

94         current = nodeA\r

95         currentDistance = 0\r

96         unvisited[current] = currentDistance\r

97 \r

98         while True:\r

99             for neighbour, distance in self.hops[current].items():\r

100                 if neighbour not in unvisited:\r

101                     continue\r

102                 newDistance = currentDistance + distance\r

103                 if unvisited[neighbour] is None or unvisited[neighbour] > newDistance:\r

104                     unvisited[neighbour] = newDistance\r

105                     previous[neighbour] = current\r

106             visited[current] = currentDistance\r

107             del unvisited[current]\r

108             if current == nodeB or not unvisited:\r

109                 break\r

110             candidates = [node for node in unvisited.items() if node[1]]\r

111             current, currentDistance = sorted(candidates, key = lambda x: x[1])[0]\r

112 \r

113         result = []\r

114         node = nodeB\r

115         while node != nodeA:\r

116             result.insert(0, node)\r

117             node = previous[node]\r

118         result.insert(0, node)\r

119         return result\r

120 \r