4503580382

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #98 from atlanticwave-sdx/97.refactor-connection-breakdown

Run Details

209 of 300 branches covered (69.67%)

Branch coverage included in aggregate %.

98 of 116 new or added lines in 5 files covered. (84.48%)

2 existing lines in 1 file now uncovered.

774 of 964 relevant lines covered (80.29%)

2.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.17

/src/sdx/pce/topology/temanager.py

import networkx as nx
from networkx.algorithms import approximation as approx

from sdx.datamodel.parsing.connectionhandler import ConnectionHandler
from sdx.pce.models import (
    ConnectionPath,
    ConnectionRequest,
    ConnectionSolution,
    TrafficMatrix,
)
from sdx.pce.topology.manager import TopologyManager


class TEManager:

    """
    TE Manager for connection - topology operations.

    Functions of this class are:

        - generate inputs to the PCE solver

        - converter the solver output.
    """

    def __init__(self, topology_data, connection_data):
        super().__init__()

        self.topology_manager = TopologyManager()
        self.connection_handler = ConnectionHandler()

        self.topology_manager.topology = (
            self.topology_manager.get_handler().import_topology_data(topology_data)
        )
        self.connection = self.connection_handler.import_connection_data(
            connection_data
        )

        self.graph = self.generate_graph_te()

    def add_topology(self, topology_data: dict):
        """
        Add a new topology to TEManager.

        :param topology_data: a dictionary that represents a topology.
        """
        self.topology_manager.add_topology(topology_data)

    def generate_connection_te(self) -> TrafficMatrix:
        """
        Generate a Traffic Matrix from the connection request we have.
        """
        ingress_port = self.connection.ingress_port
        ingress_node = self.topology_manager.topology.get_node_by_port(ingress_port.id)
        egress_port = self.connection.egress_port
        egress_node = self.topology_manager.topology.get_node_by_port(egress_port.id)

        ingress_nodes = [
            x for x, y in self.graph.nodes(data=True) if y["id"] == ingress_node.id
        ]

        egress_nodes = [
            x for x, y in self.graph.nodes(data=True) if y["id"] == egress_node.id
        ]

        if len(ingress_nodes) <= 0:
            raise Exception("No ingress nodes found in the graph")

        if len(egress_nodes) <= 0:
            raise Exception("No egress nodes found in the graph")

        required_bandwidth = self.connection.bandwidth
        required_latency = self.connection.latency

        request = ConnectionRequest(
            source=ingress_nodes[0],
            destination=egress_nodes[0],
            required_bandwidth=required_bandwidth,
            required_latency=required_latency,
        )

        return TrafficMatrix(connection_requests=[request])

    def generate_graph_te(self) -> nx.Graph:
        """
        Return the topology graph that we have.
        """
        graph = self.topology_manager.generate_graph()
        graph = nx.convert_node_labels_to_integers(graph, label_attribute="id")

        # TODO: why is this needed?
        self.graph = graph
        # print(list(graph.nodes(data=True)))

        return graph

    def graph_node_connectivity(self, source=None, dest=None):
        """
        Check that a source and destination node have connectivity.
        """
        # TODO: is this method really needed?
        return approx.node_connectivity(self.graph, source, dest)

    def requests_connectivity(self, tm: TrafficMatrix) -> bool:
        """
        Check that connectivity is possible.
        """
        # TODO: consider using filter() and reduce(), maybe?
        # TODO: write some tests for this method.
        for request in tm.connection_requests:
            conn = self.graph_node_connectivity(request.source, request.destination)
            print(
                f"Request connectivity: source {request.source}, destination: {request.destination} = {conn}"
            )
            if conn is False:
                return False

        return True

    def generate_connection_breakdown(self, connection) -> dict:
        """
        A "router" method for backward compatibility.
        """
        if isinstance(connection, ConnectionSolution):
            return self._generate_connection_breakdown_tm(connection)
        return self._generate_connection_breakdown_old(connection)

    def _generate_connection_breakdown_tm(self, connection: ConnectionSolution) -> dict:
        """
        Take a connection and generate a breakdown.

        This is an alternative to generate_connection_breakdown()
        below which uses the newly defined types from sdx.pce.models.
        """
        if connection is None or connection.connection_map is None:
            print(f"Can't find a breakdown for {connection}")
            return None

        breakdown = {}
        paths = connection.connection_map  # p2p for now

        # i_port = None
        # e_port = None

        for domain, links in paths.items():
            print(f"domain: {domain}, links: {links}")

            current_link_set = []

            for count, link in enumerate(links):
                print(f"count: {count}, link: {link}")

                assert isinstance(link, ConnectionPath)

                src_node = self.graph.nodes.get(link.source)
                assert src_node is not None

                dst_node = self.graph.nodes.get(link.destination)
                assert dst_node is not None

                print(f"source node: {src_node}, destination node: {dst_node}")

                src_domain = self.topology_manager.get_domain_name(src_node.get("id"))
                dst_domain = self.topology_manager.get_domain_name(dst_node.get("id"))

                # TODO: what do we do when a domain can't be
                # determined? Can a domain be `None`?
                print(f"source domain: {src_domain}, destination domain: {dst_domain}")

                current_link_set.append(link)
                current_domain = src_domain
                if src_domain == dst_domain:
                    # current_domain = domain_1
                    if count == len(links) - 1:
                        breakdown[current_domain] = current_link_set.copy()
                else:
                    breakdown[current_domain] = current_link_set.copy()
                    current_domain = None
                    current_link_set = []

        print(f"[intermediate] breakdown: {breakdown}")

        # now starting with the ingress_port
        first = True
        i = 0
        domain_breakdown = {}

        for domain, links in breakdown.items():
            print(f"Creating domain_breakdown: domain: {domain}, links: {links}")
            segment = {}
            if first:
                first = False
                last_link = links[-1]
                n1 = self.graph.nodes[last_link.source]["id"]
                n2 = self.graph.nodes[last_link.destination]["id"]
                n1, p1, n2, p2 = self.topology_manager.topology.get_port_by_link(n1, n2)
                i_port = self.connection.ingress_port.to_dict()
                e_port = p1
                next_i = p2
            elif i == len(breakdown) - 1:
                i_port = next_i
                e_port = self.connection.egress_port.to_dict()
            else:
                last_link = links[-1]
                n1 = self.graph.nodes[last_link.source]["id"]
                n2 = self.graph.nodes[last_link.destination]["id"]
                n1, p1, n2, p2 = self.topology_manager.topology.get_port_by_link(n1, n2)
                i_port = next_i
                e_port = p1
                next_i = p2
            segment["ingress_port"] = i_port
            segment["egress_port"] = e_port
            domain_breakdown[domain] = segment.copy()
            i = i + 1

        print(f"generate_connection_breakdown(): domain_breakdown: {domain_breakdown}")
        return domain_breakdown

    def _generate_connection_breakdown_old(self, connection):
        """
        Take a connection and generate a breakdown.
        """
        assert connection is not None

        breakdown = {}
        paths = connection[0]  # p2p for now
        # cost = connection[1]
        i_port = None
        e_port = None

        print(f"Domain breakdown with graph: {self.graph}")
        print(f"Graph nodes: {self.graph.nodes}")
        print(f"Graph edges: {self.graph.edges}")

        print(f"Paths: {paths}")

        for i, j in paths.items():
            print(f"i: {i}, j: {j}")
            current_link_set = []
            for count, link in enumerate(j):
                print(f"count: {count}, link: {link}")
                assert len(link) == 2

                node_1 = self.graph.nodes.get(link[0])
                assert node_1 is not None

                node_2 = self.graph.nodes.get(link[1])
                assert node_2 is not None

                print(f"node_1: {node_1}, node_2: {node_2}")

                domain_1 = self.topology_manager.get_domain_name(node_1["id"])
                domain_2 = self.topology_manager.get_domain_name(node_2["id"])

                # # TODO: handle the cases where a domain was not found.
                # if domain_1 is None:
                #     domain_1 = f"domain_{i}"
                # if domain_2 is None:
                #     domain_2 = f"domain_{i}"

                print(f"domain_1: {domain_1}, domain_2: {domain_2}")

                current_link_set.append(link)
                current_domain = domain_1
                if domain_1 == domain_2:
                    # current_domain = domain_1
                    if count == len(j) - 1:
                        breakdown[current_domain] = current_link_set.copy()
                else:
                    breakdown[current_domain] = current_link_set.copy()
                    current_domain = None
                    current_link_set = []

        print(f"[intermediate] breakdown: {breakdown}")

        # now starting with the ingress_port
        first = True
        i = 0
        domain_breakdown = {}

        for domain, links in breakdown.items():
            print(f"Creating domain_breakdown: domain: {domain}, links: {links}")
            segment = {}
            if first:
                first = False
                last_link = links[-1]
                n1 = self.graph.nodes[last_link[0]]["id"]
                n2 = self.graph.nodes[last_link[1]]["id"]
                n1, p1, n2, p2 = self.topology_manager.topology.get_port_by_link(n1, n2)
                i_port = self.connection.ingress_port.to_dict()
                e_port = p1
                next_i = p2
            elif i == len(breakdown) - 1:
                i_port = next_i
                e_port = self.connection.egress_port.to_dict()
            else:
                last_link = links[-1]
                n1 = self.graph.nodes[last_link[0]]["id"]
                n2 = self.graph.nodes[last_link[1]]["id"]
                n1, p1, n2, p2 = self.topology_manager.topology.get_port_by_link(n1, n2)
                i_port = next_i
                e_port = p1
                next_i = p2
            segment["ingress_port"] = i_port
            segment["egress_port"] = e_port
            domain_breakdown[domain] = segment.copy()
            i = i + 1

        print(f"generate_connection_breakdown(): domain_breakdown: {domain_breakdown}")
        return domain_breakdown

1	import networkx as nx	3✔
2	from networkx.algorithms import approximation as approx	3✔
3
4	from sdx.datamodel.parsing.connectionhandler import ConnectionHandler	3✔
5	from sdx.pce.models import (	3✔
6	ConnectionPath,
7	ConnectionRequest,
8	ConnectionSolution,
9	TrafficMatrix,
10	)
11	from sdx.pce.topology.manager import TopologyManager	3✔
12
13
14	class TEManager:	3✔
15
16	"""
17	TE Manager for connection - topology operations.
18
19	Functions of this class are:
20
21	- generate inputs to the PCE solver
22
23	- converter the solver output.
24	"""
25
26	def __init__(self, topology_data, connection_data):	3✔
27	super().__init__()	3✔
28
29	self.topology_manager = TopologyManager()	3✔
30	self.connection_handler = ConnectionHandler()	3✔
31
32	self.topology_manager.topology = (	3✔
33	self.topology_manager.get_handler().import_topology_data(topology_data)
34	)
35	self.connection = self.connection_handler.import_connection_data(	3✔
36	connection_data
37	)
38
39	self.graph = self.generate_graph_te()	3✔
40
41	def add_topology(self, topology_data: dict):	3✔
42	"""
43	Add a new topology to TEManager.
44
45	:param topology_data: a dictionary that represents a topology.
46	"""
47	self.topology_manager.add_topology(topology_data)	3✔
48
49	def generate_connection_te(self) -> TrafficMatrix:	3✔
50	"""
51	Generate a Traffic Matrix from the connection request we have.
52	"""
53	ingress_port = self.connection.ingress_port	3✔
54	ingress_node = self.topology_manager.topology.get_node_by_port(ingress_port.id)	3✔
55	egress_port = self.connection.egress_port	3✔
56	egress_node = self.topology_manager.topology.get_node_by_port(egress_port.id)	3✔
57
58	ingress_nodes = [	3✔
59	x for x, y in self.graph.nodes(data=True) if y["id"] == ingress_node.id
60	]
61
62	egress_nodes = [	3✔
63	x for x, y in self.graph.nodes(data=True) if y["id"] == egress_node.id
64	]
65
66	if len(ingress_nodes) <= 0:	3!
NEW 67	raise Exception("No ingress nodes found in the graph")	×
68
69	if len(egress_nodes) <= 0:	3!
NEW 70	raise Exception("No egress nodes found in the graph")	×
71
72	required_bandwidth = self.connection.bandwidth	3✔
73	required_latency = self.connection.latency	3✔
74
75	request = ConnectionRequest(	3✔
76	source=ingress_nodes[0],
77	destination=egress_nodes[0],
78	required_bandwidth=required_bandwidth,
79	required_latency=required_latency,
80	)
81
82	return TrafficMatrix(connection_requests=[request])	3✔
83
84	def generate_graph_te(self) -> nx.Graph:	3✔
85	"""
86	Return the topology graph that we have.
87	"""
88	graph = self.topology_manager.generate_graph()	3✔
89	graph = nx.convert_node_labels_to_integers(graph, label_attribute="id")	3✔
90
91	# TODO: why is this needed?
92	self.graph = graph	3✔
93	# print(list(graph.nodes(data=True)))
94
95	return graph	3✔
96
97	def graph_node_connectivity(self, source=None, dest=None):	3✔
98	"""
99	Check that a source and destination node have connectivity.
100	"""
101	# TODO: is this method really needed?
NEW 102	return approx.node_connectivity(self.graph, source, dest)	×
103
104	def requests_connectivity(self, tm: TrafficMatrix) -> bool:	3✔
105	"""
106	Check that connectivity is possible.
107	"""
108	# TODO: consider using filter() and reduce(), maybe?
109	# TODO: write some tests for this method.
NEW 110	for request in tm.connection_requests:	×
NEW 111	conn = self.graph_node_connectivity(request.source, request.destination)	×
UNCOV 112	print(	×
113	f"Request connectivity: source {request.source}, destination: {request.destination} = {conn}"
114	)
NEW 115	if conn is False:	×
NEW 116	return False	×
117
UNCOV 118	return True	×
119
120	def generate_connection_breakdown(self, connection) -> dict:	3✔
121	"""
122	A "router" method for backward compatibility.
123	"""
124	if isinstance(connection, ConnectionSolution):	3✔
125	return self._generate_connection_breakdown_tm(connection)	3✔
126	return self._generate_connection_breakdown_old(connection)	3✔
127
128	def _generate_connection_breakdown_tm(self, connection: ConnectionSolution) -> dict:	3✔
129	"""
130	Take a connection and generate a breakdown.
131
132	This is an alternative to generate_connection_breakdown()
133	below which uses the newly defined types from sdx.pce.models.
134	"""
135	if connection is None or connection.connection_map is None:	3✔
136	print(f"Can't find a breakdown for {connection}")	3✔
137	return None	3✔
138
139	breakdown = {}	3✔
140	paths = connection.connection_map # p2p for now	3✔
141
142	# i_port = None
143	# e_port = None
144
145	for domain, links in paths.items():	3✔
146	print(f"domain: {domain}, links: {links}")	3✔
147
148	current_link_set = []	3✔
149
150	for count, link in enumerate(links):	3✔
151	print(f"count: {count}, link: {link}")	3✔
152
153	assert isinstance(link, ConnectionPath)	3✔
154
155	src_node = self.graph.nodes.get(link.source)	3✔
156	assert src_node is not None	3✔
157
158	dst_node = self.graph.nodes.get(link.destination)	3✔
159	assert dst_node is not None	3✔
160
161	print(f"source node: {src_node}, destination node: {dst_node}")	3✔
162
163	src_domain = self.topology_manager.get_domain_name(src_node.get("id"))	3✔
164	dst_domain = self.topology_manager.get_domain_name(dst_node.get("id"))	3✔
165
166	# TODO: what do we do when a domain can't be
167	# determined? Can a domain be `None`?
168	print(f"source domain: {src_domain}, destination domain: {dst_domain}")	3✔
169
170	current_link_set.append(link)	3✔
171	current_domain = src_domain	3✔
172	if src_domain == dst_domain:	3!
173	# current_domain = domain_1
174	if count == len(links) - 1:	3!
175	breakdown[current_domain] = current_link_set.copy()	3✔
176	else:
NEW 177	breakdown[current_domain] = current_link_set.copy()	×
NEW 178	current_domain = None	×
NEW 179	current_link_set = []	×
180
181	print(f"[intermediate] breakdown: {breakdown}")	3✔
182
183	# now starting with the ingress_port
184	first = True	3✔
185	i = 0	3✔
186	domain_breakdown = {}	3✔
187
188	for domain, links in breakdown.items():	3✔
189	print(f"Creating domain_breakdown: domain: {domain}, links: {links}")	3✔
190	segment = {}	3✔
191	if first:	3✔
192	first = False	3✔
193	last_link = links[-1]	3✔
194	n1 = self.graph.nodes[last_link.source]["id"]	3✔
195	n2 = self.graph.nodes[last_link.destination]["id"]	3✔
196	n1, p1, n2, p2 = self.topology_manager.topology.get_port_by_link(n1, n2)	3✔
197	i_port = self.connection.ingress_port.to_dict()	3✔
198	e_port = p1	3✔
199	next_i = p2	3✔
200	elif i == len(breakdown) - 1:	3!
201	i_port = next_i	3✔
202	e_port = self.connection.egress_port.to_dict()	3✔
203	else:
NEW 204	last_link = links[-1]	×
NEW 205	n1 = self.graph.nodes[last_link.source]["id"]	×
NEW 206	n2 = self.graph.nodes[last_link.destination]["id"]	×
NEW 207	n1, p1, n2, p2 = self.topology_manager.topology.get_port_by_link(n1, n2)	×
NEW 208	i_port = next_i	×
NEW 209	e_port = p1	×
NEW 210	next_i = p2	×
211	segment["ingress_port"] = i_port	3✔
212	segment["egress_port"] = e_port	3✔
213	domain_breakdown[domain] = segment.copy()	3✔
214	i = i + 1	3✔
215
216	print(f"generate_connection_breakdown(): domain_breakdown: {domain_breakdown}")	3✔
217	return domain_breakdown	3✔
218
219	def _generate_connection_breakdown_old(self, connection):	3✔
220	"""
221	Take a connection and generate a breakdown.
222	"""
223	assert connection is not None	3✔
224
225	breakdown = {}	3✔
226	paths = connection[0] # p2p for now	3✔
227	# cost = connection[1]
228	i_port = None	3✔
229	e_port = None	3✔
230
231	print(f"Domain breakdown with graph: {self.graph}")	3✔
232	print(f"Graph nodes: {self.graph.nodes}")	3✔
233	print(f"Graph edges: {self.graph.edges}")	3✔
234
235	print(f"Paths: {paths}")	3✔
236
237	for i, j in paths.items():	3✔
238	print(f"i: {i}, j: {j}")	3✔
239	current_link_set = []	3✔
240	for count, link in enumerate(j):	3✔
241	print(f"count: {count}, link: {link}")	3✔
242	assert len(link) == 2	3✔
243
244	node_1 = self.graph.nodes.get(link[0])	3✔
245	assert node_1 is not None	3✔
246
247	node_2 = self.graph.nodes.get(link[1])	3✔
248	assert node_2 is not None	3✔
249
250	print(f"node_1: {node_1}, node_2: {node_2}")	3✔
251
252	domain_1 = self.topology_manager.get_domain_name(node_1["id"])	3✔
253	domain_2 = self.topology_manager.get_domain_name(node_2["id"])	3✔
254
255	# # TODO: handle the cases where a domain was not found.
256	# if domain_1 is None:
257	# domain_1 = f"domain_{i}"
258	# if domain_2 is None:
259	# domain_2 = f"domain_{i}"
260
261	print(f"domain_1: {domain_1}, domain_2: {domain_2}")	3✔
262
263	current_link_set.append(link)	3✔
264	current_domain = domain_1	3✔
265	if domain_1 == domain_2:	3!
266	# current_domain = domain_1
267	if count == len(j) - 1:	3✔
268	breakdown[current_domain] = current_link_set.copy()	3✔
269	else:
270	breakdown[current_domain] = current_link_set.copy()	×
271	current_domain = None	×
272	current_link_set = []	×
273
274	print(f"[intermediate] breakdown: {breakdown}")	3✔
275
276	# now starting with the ingress_port
277	first = True	3✔
278	i = 0	3✔
279	domain_breakdown = {}	3✔
280
281	for domain, links in breakdown.items():	3✔
282	print(f"Creating domain_breakdown: domain: {domain}, links: {links}")	3✔
283	segment = {}	3✔
284	if first:	3!
285	first = False	3✔
286	last_link = links[-1]	3✔
287	n1 = self.graph.nodes[last_link[0]]["id"]	3✔
288	n2 = self.graph.nodes[last_link[1]]["id"]	3✔
289	n1, p1, n2, p2 = self.topology_manager.topology.get_port_by_link(n1, n2)	3✔
290	i_port = self.connection.ingress_port.to_dict()	3✔
291	e_port = p1	3✔
292	next_i = p2	3✔
293	elif i == len(breakdown) - 1:	×
294	i_port = next_i	×
295	e_port = self.connection.egress_port.to_dict()	×
296	else:
297	last_link = links[-1]	×
298	n1 = self.graph.nodes[last_link[0]]["id"]	×
299	n2 = self.graph.nodes[last_link[1]]["id"]	×
NEW 300	n1, p1, n2, p2 = self.topology_manager.topology.get_port_by_link(n1, n2)	×
301	i_port = next_i	×
302	e_port = p1	×
303	next_i = p2	×
304	segment["ingress_port"] = i_port	3✔
305	segment["egress_port"] = e_port	3✔
306	domain_breakdown[domain] = segment.copy()	3✔
307	i = i + 1	3✔
308
309	print(f"generate_connection_breakdown(): domain_breakdown: {domain_breakdown}")	3✔
310	return domain_breakdown	3✔

atlanticwave-sdx / pce / 4503580382

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous