10114613213

Committed 26 Jul 2024 04:36PM UTC coverage: 85.131% (+0.6%) from 84.494%

Build # 10114613213

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #197 from atlanticwave-sdx/fix/issue_196

Changes to make PCE minimally compatible with topology spec 2.0.0

Run Details

406 of 502 branches covered (80.88%)

Branch coverage included in aggregate %.

50 of 50 new or added lines in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

1094 of 1260 relevant lines covered (86.83%)

3.46 hits per line

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

90.75

/src/sdx_pce/topology/manager.py

import copy
import datetime
import logging
from typing import Mapping

import networkx as nx
from sdx_datamodel.models.link import Link
from sdx_datamodel.models.topology import (
    TOPOLOGY_INITIAL_VERSION,
    SDX_TOPOLOGY_ID_prefix,
)
from sdx_datamodel.parsing.topologyhandler import TopologyHandler

from .grenmlconverter import GrenmlConverter


class TopologyManager:
    """
    Manager for topology operations.

    Operations are:

        - Merge multiple topologies.

        - Convert to grenml (XML).
    """

    def __init__(self):
        # The merged "super" topology of topologies of different
        # domains, with inter-domain links between them computed.
        self._topology = None

        # Mapping from topology ID to topology.
        self._topology_map = {}

        # Mapping from port ID to link.
        self._port_map = {}

        # Mapping from port ID to node.
        self._port_node_map = {}

        # Number of interdomain links we computed.
        self._num_interdomain_link = 0

        self._logger = logging.getLogger(__name__)

    def get_handler(self):
        return self.topology_handler

    def topology_id(self, id):
        self._topology._id(id)

    def set_topology(self, topology):
        self._topology = topology

    def get_topology(self):
        return self._topology

    def get_topology_map(self) -> dict:
        return self._topology_map

    def get_port_map(self) -> Mapping[str, dict]:
        """
        Return a mapping between port IDs and links.
        """
        return self._port_map

    def get_port_node_map(self) -> Mapping[str, dict]:
        """
        Return a mapping between port IDs and nodes.
        """
        return self._port_node_map

    def clear_topology(self):
        self._topology = None
        self._topology_map = {}
        self._port_map = {}

    def add_topology(self, data):
        topology = TopologyHandler().import_topology_data(data)
        self._topology_map[topology.id] = topology

        if self._topology is None:
            self._topology = copy.deepcopy(topology)
            interdomain_ports = []

            # Generate a new topology id
            self.generate_id()

            # Addding to the port list
            # links = topology.links
            # for link in links:
            #    for port in link.ports:
            #        self._port_map[port["id"]] = link
        else:
            # check the inter-domain links first.
            interdomain_ports = self.inter_domain_check(topology)
            self._num_interdomain_link += len(interdomain_ports)
            if self._num_interdomain_link == 0:
                self._logger.debug(
                    f"Warning: no interdomain links detected in {topology.id}!"
                )

            # Nodes
            nodes = topology.nodes
            self._topology.add_nodes(nodes)

            # links
            links = topology.links
            self._topology.add_links(links)

            # version
            self.update_version(False)

        # Addding to the port list
        links = topology.links
        for link in links:
            for port in link.ports:
                port_id = port if isinstance(port, str) else port["id"]
                self._port_map[port_id] = link

        # Addding to the port node
        nodes = topology.nodes
        for node in nodes:
            for port in node.ports:
                self._port_node_map[port.id] = node

        # inter-domain links
        self.add_inter_domain_links(topology, interdomain_ports)

        self.update_timestamp()

    def get_domain_name(self, node_id):
        """
        Find the topology ID associated with the given node ID.

        A topology ID is expected to be of the format
        "urn:sdx:topology:amlight.net", and from this, we
        can find the domain name associated with the topology.

        TODO: This function name may be a misnomer?
        """
        domain_id = None
        # print(f"len of topology_list: {len(self._topology_map)}")
        for topology_id, topology in self._topology_map.items():
            if topology.has_node_by_id(node_id):
                domain_id = topology_id
                break

        return domain_id

    def generate_id(self):
        self._topology.id = SDX_TOPOLOGY_ID_prefix
        self._topology.version = TOPOLOGY_INITIAL_VERSION
        return id

    def remove_topology(self, topology_id):
        self._topology_map.pop(topology_id, None)
        self.update_version(False)
        self.update_timestamp()

    def is_link_interdomain(self, link, topology):
        """
        Check if a link is an interdomain link.
        """
        for port in link.ports:
            port_id = port if isinstance(port, str) else port["id"]
            if port_id not in self._port_map:
                return True
        return False

    def is_interdomain_port(self, port_id, topology_id):
        """
        Check if a Port ID is interdomain
        """
        # Sanity checks
        if (
            not isinstance(port_id, str)
            or not port_id.startswith("urn:sdx:port:")
            or not isinstance(topology_id, str)
            or not topology_id.startswith("urn:sdx:topology:")
        ):
            return False
        return port_id.split(":")[3] != topology_id.split(":")[3]

    def update_topology(self, data):
        # likely adding new inter-domain links
        update_handler = TopologyHandler()
        topology = update_handler.import_topology_data(data)
        self._topology_map[topology.id] = topology

        # Nodes.
        nodes = topology.nodes
        for node in nodes:
            self._topology.remove_node(node.id)

        # Links.
        links = topology.links
        for link in links:
            if not self.is_link_interdomain(link, topology):
                # print(link.id+";......."+str(link.nni))
                self._topology.remove_link(link.id)
                for port in link.ports:
                    port_id = port if isinstance(port, str) else port["id"]
                    self._port_map.pop(port_id)

        # Check the inter-domain links first.
        interdomain_ports = self.inter_domain_check(topology)
        if len(interdomain_ports) == 0:
            self._logger.warning("Warning: no interdomain links detected!")

        # Nodes.
        nodes = topology.nodes
        self._topology.add_nodes(nodes)

        # Links.
        links = topology.links
        self._topology.add_links(links)

        # inter-domain links
        self.add_inter_domain_links(topology, interdomain_ports)

        self.update_version(True)
        self.update_timestamp()

    def update_version(self, sub: bool):
        try:
            [ver, sub_ver] = self._topology.version.split(".")
        except ValueError:
            ver = self._topology.version
            sub_ver = "0"

        self._topology.version = self.new_version(ver, sub_ver, sub)

        return self._topology.version

    def new_version(self, ver, sub_ver, sub: bool):
        if not sub:
            ver = str(int(ver) + 1)
            sub_ver = "0"
        else:
            sub_ver = str(int(sub_ver) + 1)

        return ver + "." + sub_ver

    def update_timestamp(self):
        ct = datetime.datetime.now().isoformat()
        self._topology.timestamp = ct

        return ct

    def inter_domain_check(self, topology):
        interdomain_port_dict = {}
        interdomain_ports = []
        interdomain_port_ids = []
        links = topology.links
        link_dict = {}
        for link in links:
            link_dict[link.id] = link
            for port in link.ports:
                port_id = port if isinstance(port, str) else port["id"]
                interdomain_port_dict[port_id] = link

        # match any ports in the existing topology
        for port_id in interdomain_port_dict:
            # print("interdomain_port:")
            # print(port_id)
            for existing_port, existing_link in self._port_map.items():
                # print(existing_port)
                if port_id == existing_port:
                    # print("Interdomain port:" + port_id)
                    # remove redundant link between two domains
                    self._topology.remove_link(existing_link.id)
                    interdomain_port_ids.append(port_id)
            self._port_map[port_id] = interdomain_port_dict[port_id]

        # count for inter-domain links according to topo spec 2.0.x
        for node in topology.nodes:
            for port in node.ports:
                # interdomain ports based on previous methodology
                if port.id in interdomain_port_ids:
                    interdomain_ports.append(port)
                # interdomain ports based on new methodology (spec 2.0)
                if self.is_interdomain_port(port.nni, topology.id):
                    interdomain_ports.append(port)

        return interdomain_ports

    def create_inter_domain_link(self, port1, port2):
        """Creates an interdomain link from two ports."""
        if port2.id < port1.id:
            port1, port2 = port2, port1

        port1_id = port1.id.replace("urn:sdx:port:", "", 1)
        port2_id = port2.id.replace("urn:sdx:port:", "", 1)
        link_id = f"urn:sdx:link:interdomain:{port1_id}:{port2_id}"

        return Link(
            id=link_id,
            name=f"{port1.name}--{port2.name}",
            ports=[port1.id, port2.id],
            bandwidth=100,
            residual_bandwidth=100,
            latency=0,
            packet_loss=0,
            availability=100,
        )

    def add_inter_domain_links(self, topology, interdomain_ports):
        """Add inter-domain links (whenever possible)."""
        for port in interdomain_ports:
            other_node = self._port_node_map.get(port.nni)
            other_ports = other_node.ports if other_node else []
            for other_port in other_ports:
                if other_port.id == port.nni and other_port.nni == port.id:
                    break
            else:
                self._logger.warning(
                    "Interdomin link not added now - didnt find other port:"
                    f" port={port.id} other_port={port.nni}"
                )
                continue
            link = self.create_inter_domain_link(port, other_port)
            self._topology.add_links([link])

    # adjacent matrix of the graph, in jason?
    def generate_graph(self):
        graph = nx.Graph()

        if self._topology is None:
            self._logger.warning("We do not have a topology yet")
            return None

        links = self._topology.links
        for link in links:
            inter_domain_link = False
            ports = link.ports
            end_nodes = []
            for port in ports:
                port_id = port if isinstance(port, str) else port["id"]
                node = self._topology.get_node_by_port(port_id)
                if node is None:
                    self._logger.warning(
                        f"This port (id: {port_id}) does not belong to "
                        f"any node in the topology, likely a Non-SDX port!"
                    )
                    inter_domain_link = True
                    break
                else:
                    end_nodes.append(node)
                    # print("graph node:"+node.id)
            if not inter_domain_link:
                graph.add_edge(end_nodes[0].id, end_nodes[1].id)
                edge = graph.edges[end_nodes[0].id, end_nodes[1].id]
                edge["id"] = link.id
                edge["latency"] = link.latency
                edge["bandwidth"] = link.bandwidth
                edge["residual_bandwidth"] = link.residual_bandwidth
                edge["weight"] = 1000.0 * (1.0 / link.residual_bandwidth)
                edge["packet_loss"] = link.packet_loss
                edge["availability"] = link.availability

        return graph

    def generate_grenml(self):
        self.converter = GrenmlConverter(self._topology)

        return self.converter.read_topology()

    def add_domain_service(self):
        pass

    # may need to read from a configuration file.
    def update_private_properties(self):
        pass

    # on performance properties for now
    def update_link_property(self, link_id, property, value):
        # 1. update the individual topology
        for id, topology in self._topology_map.items():
            links = topology.links
            for link in links:
                self._logger.info(f"link.id={link.id}; id={id}")
                if link.id == link_id:
                    setattr(link, property, value)
                    self._logger.info("updated the link.")
                    # 1.2 need to change the sub_ver of the topology?

        # 2. check on the inter-domain link?
        # 3. update the interdomain topology
        links = self._topology.links
        for link in links:
            if link.id == link_id:
                setattr(link, property, value)
                self._logger.info("updated the link.")
                # 2.2 need to change the sub_ver of the topology?

        self.update_version(True)
        self.update_timestamp()
        # 4. Signal update the (networkx) graph

        # 5. signal Reoptimization of TE?

    def update_element_property_json(self, data, element, element_id, property, value):
        elements = data[element]
        for element in elements:
            if element["id"] == element_id:
                element[property] = value

        try:
            [ver, sub_ver] = data["version"].split(".")
        except ValueError:
            ver = "0"
            sub_ver = "0"

        data["version"] = self.new_version(ver, sub_ver, True)
        data["timestamp"] = datetime.datetime.now().isoformat()

    def get_port_by_id(self, port_id: str):
        """
        Given port id, returns a Port.
        """
        for node in self.get_topology().nodes:
            for port in node.ports:
                if port.id == port_id:
                    return port.to_dict()
        return None

    def are_two_ports_same_domain(self, port1_id: str, port2_id: str):
        """
        Check if two ports are in the same domain.
        """
        node1 = self.get_topology().get_node_by_port(port1_id)
        node2 = self.get_topology().get_node_by_port(port2_id)
        if node1 is None or node2 is None:
            return False

        domain1 = self.get_domain_name(node1.id)
        domain2 = self.get_domain_name(node2.id)
        return domain1 == domain2

    def update_node_property(self):
        pass

    def update_port_property(self):
        pass

1	import copy	4✔
2	import datetime	4✔
3	import logging	4✔
4	from typing import Mapping	4✔
5
6	import networkx as nx	4✔
7	from sdx_datamodel.models.link import Link	4✔
8	from sdx_datamodel.models.topology import (	4✔
9	TOPOLOGY_INITIAL_VERSION,
10	SDX_TOPOLOGY_ID_prefix,
11	)
12	from sdx_datamodel.parsing.topologyhandler import TopologyHandler	4✔
13
14	from .grenmlconverter import GrenmlConverter	4✔
15
16
17	class TopologyManager:	4✔
18	"""	4✔
19	Manager for topology operations.
20
21	Operations are:
22
23	- Merge multiple topologies.
24
25	- Convert to grenml (XML).
26	"""
27
28	def __init__(self):	4✔
29	# The merged "super" topology of topologies of different
30	# domains, with inter-domain links between them computed.
31	self._topology = None	4✔
32
33	# Mapping from topology ID to topology.
34	self._topology_map = {}	4✔
35
36	# Mapping from port ID to link.
37	self._port_map = {}	4✔
38
39	# Mapping from port ID to node.
40	self._port_node_map = {}	4✔
41
42	# Number of interdomain links we computed.
43	self._num_interdomain_link = 0	4✔
44
45	self._logger = logging.getLogger(__name__)	4✔
46
47	def get_handler(self):	4✔
48	return self.topology_handler	×
49
50	def topology_id(self, id):	4✔
51	self._topology._id(id)	×
52
53	def set_topology(self, topology):	4✔
54	self._topology = topology	4✔
55
56	def get_topology(self):	4✔
57	return self._topology	4✔
58
59	def get_topology_map(self) -> dict:	4✔
60	return self._topology_map	4✔
61
62	def get_port_map(self) -> Mapping[str, dict]:	4✔
63	"""
64	Return a mapping between port IDs and links.
65	"""
66	return self._port_map	4✔
67
68	def get_port_node_map(self) -> Mapping[str, dict]:	4✔
69	"""
70	Return a mapping between port IDs and nodes.
71	"""
72	return self._port_node_map	4✔
73
74	def clear_topology(self):	4✔
75	self._topology = None	×
76	self._topology_map = {}	×
77	self._port_map = {}	×
78
79	def add_topology(self, data):	4✔
80	topology = TopologyHandler().import_topology_data(data)	4✔
81	self._topology_map[topology.id] = topology	4✔
82
83	if self._topology is None:	4✔
84	self._topology = copy.deepcopy(topology)	4✔
85	interdomain_ports = []	4✔
86
87	# Generate a new topology id
88	self.generate_id()	4✔
89
90	# Addding to the port list
91	# links = topology.links
92	# for link in links:
93	# for port in link.ports:
94	# self._port_map[port["id"]] = link
95	else:
96	# check the inter-domain links first.
97	interdomain_ports = self.inter_domain_check(topology)	4✔
98	self._num_interdomain_link += len(interdomain_ports)	4✔
99	if self._num_interdomain_link == 0:	4✔
100	self._logger.debug(	4✔
101	f"Warning: no interdomain links detected in {topology.id}!"
102	)
103
104	# Nodes
105	nodes = topology.nodes	4✔
106	self._topology.add_nodes(nodes)	4✔
107
108	# links
109	links = topology.links	4✔
110	self._topology.add_links(links)	4✔
111
112	# version
113	self.update_version(False)	4✔
114
115	# Addding to the port list
116	links = topology.links	4✔
117	for link in links:	4✔
118	for port in link.ports:	4✔
119	port_id = port if isinstance(port, str) else port["id"]	4✔
120	self._port_map[port_id] = link	4✔
121
122	# Addding to the port node
123	nodes = topology.nodes	4✔
124	for node in nodes:	4✔
125	for port in node.ports:	4✔
126	self._port_node_map[port.id] = node	4✔
127
128	# inter-domain links
129	self.add_inter_domain_links(topology, interdomain_ports)	4✔
130
131	self.update_timestamp()	4✔
132
133	def get_domain_name(self, node_id):	4✔
134	"""
135	Find the topology ID associated with the given node ID.
136
137	A topology ID is expected to be of the format
138	"urn:sdx:topology:amlight.net", and from this, we
139	can find the domain name associated with the topology.
140
141	TODO: This function name may be a misnomer?
142	"""
143	domain_id = None	4✔
144	# print(f"len of topology_list: {len(self._topology_map)}")
145	for topology_id, topology in self._topology_map.items():	4✔
146	if topology.has_node_by_id(node_id):	4✔
147	domain_id = topology_id	4✔
148	break	4✔
149
150	return domain_id	4!
151
152	def generate_id(self):	4✔
153	self._topology.id = SDX_TOPOLOGY_ID_prefix	4✔
154	self._topology.version = TOPOLOGY_INITIAL_VERSION	4✔
155	return id	4✔
156
157	def remove_topology(self, topology_id):	4✔
158	self._topology_map.pop(topology_id, None)	×
159	self.update_version(False)	×
160	self.update_timestamp()	×
161
162	def is_link_interdomain(self, link, topology):	4✔
163	"""
164	Check if a link is an interdomain link.
165	"""
166	for port in link.ports:	4✔
167	port_id = port if isinstance(port, str) else port["id"]	4✔
168	if port_id not in self._port_map:	4✔
UNCOV 169	return True	×
170	return False	4✔
171
172	def is_interdomain_port(self, port_id, topology_id):	4✔
173	"""
174	Check if a Port ID is interdomain
175	"""
176	# Sanity checks
177	if (	4✔
178	not isinstance(port_id, str)
179	or not port_id.startswith("urn:sdx:port:")
180	or not isinstance(topology_id, str)
181	or not topology_id.startswith("urn:sdx:topology:")
182	):
183	return False	4✔
184	return port_id.split(":")[3] != topology_id.split(":")[3]	4✔
185
186	def update_topology(self, data):	4✔
187	# likely adding new inter-domain links
188	update_handler = TopologyHandler()	4✔
189	topology = update_handler.import_topology_data(data)	4✔
190	self._topology_map[topology.id] = topology	4✔
191
192	# Nodes.
193	nodes = topology.nodes	4✔
194	for node in nodes:	4✔
195	self._topology.remove_node(node.id)	4✔
196
197	# Links.
198	links = topology.links	4✔
199	for link in links:	4!
200	if not self.is_link_interdomain(link, topology):	4✔
201	# print(link.id+";......."+str(link.nni))
202	self._topology.remove_link(link.id)	4✔
203	for port in link.ports:	4✔
204	port_id = port if isinstance(port, str) else port["id"]	4✔
205	self._port_map.pop(port_id)	4✔
206
207	# Check the inter-domain links first.
208	interdomain_ports = self.inter_domain_check(topology)	4✔
209	if len(interdomain_ports) == 0:	4✔
210	self._logger.warning("Warning: no interdomain links detected!")	4✔
211
212	# Nodes.
213	nodes = topology.nodes	4!
214	self._topology.add_nodes(nodes)	4✔
215
216	# Links.
217	links = topology.links	4✔
218	self._topology.add_links(links)	4✔
219
220	# inter-domain links
221	self.add_inter_domain_links(topology, interdomain_ports)	4✔
222
223	self.update_version(True)	4✔
224	self.update_timestamp()	4✔
225
226	def update_version(self, sub: bool):	4✔
227	try:	4✔
228	[ver, sub_ver] = self._topology.version.split(".")	4✔
229	except ValueError:	×
230	ver = self._topology.version	×
231	sub_ver = "0"	×
232
233	self._topology.version = self.new_version(ver, sub_ver, sub)	4✔
234
235	return self._topology.version	4✔
236
237	def new_version(self, ver, sub_ver, sub: bool):	4✔
238	if not sub:	4✔
239	ver = str(int(ver) + 1)	4✔
240	sub_ver = "0"	4✔
241	else:
242	sub_ver = str(int(sub_ver) + 1)	4✔
243
244	return ver + "." + sub_ver	4✔
245
246	def update_timestamp(self):	4✔
247	ct = datetime.datetime.now().isoformat()	4✔
248	self._topology.timestamp = ct	4✔
249
250	return ct	4✔
251
252	def inter_domain_check(self, topology):	4✔
253	interdomain_port_dict = {}	4✔
254	interdomain_ports = []	4✔
255	interdomain_port_ids = []	4✔
256	links = topology.links	4✔
257	link_dict = {}	4✔
258	for link in links:	4✔
259	link_dict[link.id] = link	4✔
260	for port in link.ports:	4✔
261	port_id = port if isinstance(port, str) else port["id"]	4✔
262	interdomain_port_dict[port_id] = link	4✔
263
264	# match any ports in the existing topology
265	for port_id in interdomain_port_dict:	4✔
266	# print("interdomain_port:")
267	# print(port_id)
268	for existing_port, existing_link in self._port_map.items():	4✔
269	# print(existing_port)
270	if port_id == existing_port:	4✔
271	# print("Interdomain port:" + port_id)
272	# remove redundant link between two domains
273	self._topology.remove_link(existing_link.id)	4✔
274	interdomain_port_ids.append(port_id)	4✔
275	self._port_map[port_id] = interdomain_port_dict[port_id]	4✔
276
277	# count for inter-domain links according to topo spec 2.0.x
278	for node in topology.nodes:	4✔
279	for port in node.ports:	4✔
280	# interdomain ports based on previous methodology
281	if port.id in interdomain_port_ids:	4✔
282	interdomain_ports.append(port)	4✔
283	# interdomain ports based on new methodology (spec 2.0)
284	if self.is_interdomain_port(port.nni, topology.id):	4✔
285	interdomain_ports.append(port)	4✔
286
287	return interdomain_ports	4✔
288
289	def create_inter_domain_link(self, port1, port2):	4✔
290	"""Creates an interdomain link from two ports."""
291	if port2.id < port1.id:	4✔
292	port1, port2 = port2, port1	4✔
293
294	port1_id = port1.id.replace("urn:sdx:port:", "", 1)	4✔
295	port2_id = port2.id.replace("urn:sdx:port:", "", 1)	4✔
296	link_id = f"urn:sdx:link:interdomain:{port1_id}:{port2_id}"	4✔
297
298	return Link(	4✔
299	id=link_id,
300	name=f"{port1.name}--{port2.name}",
301	ports=[port1.id, port2.id],
302	bandwidth=100,
303	residual_bandwidth=100,
304	latency=0,
305	packet_loss=0,
306	availability=100,
307	)
308
309	def add_inter_domain_links(self, topology, interdomain_ports):	4✔
310	"""Add inter-domain links (whenever possible)."""
311	for port in interdomain_ports:	4✔
312	other_node = self._port_node_map.get(port.nni)	4✔
313	other_ports = other_node.ports if other_node else []	4✔
314	for other_port in other_ports:	4✔
315	if other_port.id == port.nni and other_port.nni == port.id:	4✔
316	break	4✔
317	else:
318	self._logger.warning(	4✔
319	"Interdomin link not added now - didnt find other port:"
320	f" port={port.id} other_port={port.nni}"
321	)
322	continue	4✔
323	link = self.create_inter_domain_link(port, other_port)	4✔
324	self._topology.add_links([link])	4✔
325
326	# adjacent matrix of the graph, in jason?
327	def generate_graph(self):	4✔
328	graph = nx.Graph()	4✔
329
330	if self._topology is None:	4✔
331	self._logger.warning("We do not have a topology yet")	×
332	return None	×
333
334	links = self._topology.links	4✔
335	for link in links:	4✔
336	inter_domain_link = False	4✔
337	ports = link.ports	4✔
338	end_nodes = []	4✔
339	for port in ports:	4✔
340	port_id = port if isinstance(port, str) else port["id"]	4✔
341	node = self._topology.get_node_by_port(port_id)	4✔
342	if node is None:	4✔
343	self._logger.warning(	4✔
344	f"This port (id: {port_id}) does not belong to "
345	f"any node in the topology, likely a Non-SDX port!"
346	)
347	inter_domain_link = True	4✔
348	break	4✔
349	else:
350	end_nodes.append(node)	4✔
351	# print("graph node:"+node.id)
352	if not inter_domain_link:	4✔
353	graph.add_edge(end_nodes[0].id, end_nodes[1].id)	4✔
354	edge = graph.edges[end_nodes[0].id, end_nodes[1].id]	4✔
355	edge["id"] = link.id	4✔
356	edge["latency"] = link.latency	4✔
357	edge["bandwidth"] = link.bandwidth	4✔
358	edge["residual_bandwidth"] = link.residual_bandwidth	4✔
359	edge["weight"] = 1000.0 * (1.0 / link.residual_bandwidth)	4✔
360	edge["packet_loss"] = link.packet_loss	4✔
361	edge["availability"] = link.availability	4✔
362
363	return graph	4✔
364
365	def generate_grenml(self):	4✔
366	self.converter = GrenmlConverter(self._topology)	×
367
368	return self.converter.read_topology()	×
369
370	def add_domain_service(self):	4✔
371	pass	×
372
373	# may need to read from a configuration file.
374	def update_private_properties(self):	4✔
375	pass	×
376
377	# on performance properties for now
378	def update_link_property(self, link_id, property, value):	4✔
379	# 1. update the individual topology
380	for id, topology in self._topology_map.items():	4✔
381	links = topology.links	4✔
382	for link in links:	4✔
383	self._logger.info(f"link.id={link.id}; id={id}")	4✔
384	if link.id == link_id:	4✔
385	setattr(link, property, value)	4✔
386	self._logger.info("updated the link.")	4✔
387	# 1.2 need to change the sub_ver of the topology?
388
389	# 2. check on the inter-domain link?
390	# 3. update the interdomain topology
391	links = self._topology.links	4✔
392	for link in links:	4✔
393	if link.id == link_id:	4✔
394	setattr(link, property, value)	4✔
395	self._logger.info("updated the link.")	4✔
396	# 2.2 need to change the sub_ver of the topology?
397
398	self.update_version(True)	4✔
399	self.update_timestamp()	4✔
400	# 4. Signal update the (networkx) graph
401
402	# 5. signal Reoptimization of TE?
403
404	def update_element_property_json(self, data, element, element_id, property, value):	4✔
405	elements = data[element]	4✔
406	for element in elements:	4✔
407	if element["id"] == element_id:	4✔
408	element[property] = value	4✔
409
410	try:	4✔
411	[ver, sub_ver] = data["version"].split(".")	4✔
412	except ValueError:	×
413	ver = "0"	×
414	sub_ver = "0"	×
415
416	data["version"] = self.new_version(ver, sub_ver, True)	4✔
417	data["timestamp"] = datetime.datetime.now().isoformat()	4✔
418
419	def get_port_by_id(self, port_id: str):	4✔
420	"""
421	Given port id, returns a Port.
422	"""
423	for node in self.get_topology().nodes:	4✔
424	for port in node.ports:	4✔
425	if port.id == port_id:	4✔
426	return port.to_dict()	4✔
427	return None	×
428
429	def are_two_ports_same_domain(self, port1_id: str, port2_id: str):	4✔
430	"""
431	Check if two ports are in the same domain.
432	"""
433	node1 = self.get_topology().get_node_by_port(port1_id)	4✔
434	node2 = self.get_topology().get_node_by_port(port2_id)	4✔
435	if node1 is None or node2 is None:	4✔
436	return False	×
437
438	domain1 = self.get_domain_name(node1.id)	4✔
439	domain2 = self.get_domain_name(node2.id)	4✔
440	return domain1 == domain2	4✔
441
442	def update_node_property(self):	4✔
443	pass	×
444
445	def update_port_property(self):	4✔
446	pass	×

atlanticwave-sdx / pce / 10114613213

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