build.main.Main._map_endpoints_from_link_ids() - Code Metrics - Inspection of "[Fix] safe guards to discard graph topology older..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#35)

by Vinicius

created 2022-11-23 18:41 UTC

build.main.Main._map_endpoints_from_link_ids() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	11
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	3

Importance

Changes

Metric	Value
cc	3
eloc	10
nop	2
dl	0
loc	11
ccs	10
cts	10
cp	1
crap	3
rs	9.9
c	0
b	0
f	0

"""Main module of kytos/pathfinder Kytos Network Application."""

from threading import Lock
from typing import Generator

from flask import jsonify, request
from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.kytos.pathfinder.graph import KytosGraph
# pylint: disable=import-error
from werkzeug.exceptions import BadRequest


class Main(KytosNApp):
    """
    Main class of kytos/pathfinder NApp.

    This class is the entry point for this napp.
    """

    def setup(self):
        """Create a graph to handle the nodes and edges."""
        self.graph = KytosGraph()
        self._topology = None
        self._lock = Lock()
        self._topology_updated_at = None
        self._links_updated_at = {}

    def execute(self):
        """Do nothing."""

    def shutdown(self):
        """Shutdown the napp."""

    def _filter_paths_le_cost(self, paths, max_cost):
        """Filter by paths where the cost is le <= max_cost."""
        if not max_cost:
            return paths
        return [path for path in paths if path["cost"] <= max_cost]

    def _map_endpoints_from_link_ids(self, link_ids: list[str]) -> dict:
        """Map endpoints from link ids."""
        endpoints = {}
        for link_id in link_ids:
            try:
                link = self._topology.links[link_id]
                endpoint_a, endpoint_b = link.endpoint_a, link.endpoint_b
                endpoints[(endpoint_a.id, endpoint_b.id)] = link
            except KeyError:
                pass
        return endpoints

    def _find_all_link_ids(
        self, paths: list[dict], link_ids: list[str]
    ) -> Generator[int, None, None]:
        """Find indexes of the paths that contain all link ids."""
        endpoints_links = self._map_endpoints_from_link_ids(link_ids)
        if not endpoints_links:
            return None
        endpoint_keys = set(endpoints_links.keys())
        for idx, path in enumerate(paths):
            head, tail, found_endpoints = path["hops"][:-1], path["hops"][1:], set()
            for endpoint_a, endpoint_b in zip(head, tail):
                if (endpoint_a, endpoint_b) in endpoints_links:
                    found_endpoints.add((endpoint_a, endpoint_b))
                if (endpoint_b, endpoint_a) in endpoints_links:
                    found_endpoints.add((endpoint_b, endpoint_a))
            if found_endpoints == endpoint_keys:
                yield idx
        return None

    def _find_any_link_ids(
        self, paths: list[dict], link_ids: list[str]
    ) -> Generator[int, None, None]:
        """Find indexes of the paths that contain any of the link ids."""
        endpoints_links = self._map_endpoints_from_link_ids(link_ids)
        if not endpoints_links:
            return None
        for idx, path in enumerate(paths):
            head, tail, found = path["hops"][:-1], path["hops"][1:], False
            for endpoint_a, endpoint_b in zip(head, tail):
                if any(
                    (
                        (endpoint_a, endpoint_b) in endpoints_links,
                        (endpoint_b, endpoint_a) in endpoints_links,
                    )
                ):
                    found = True
                    break
            if found:
                yield idx
        return None

    def _filter_paths_undesired_links(
        self, paths: list[dict], undesired: list[str]
    ) -> list[dict]:
        """Filter by undesired_links, it performs a logical OR."""
        if not undesired:
            return paths
        excluded_indexes = set(self._find_any_link_ids(paths, undesired))
        return [path for idx, path in enumerate(paths) if idx not in excluded_indexes]

    def _filter_paths_desired_links(
        self, paths: list[dict], desired: list[str]
    ) -> list[dict]:
        """Filter by desired_links, it performs a logical AND."""
        if not desired:
            return paths
        included_indexes = set(self._find_all_link_ids(paths, desired))
        return [path for idx, path in enumerate(paths) if idx in included_indexes]

    def _validate_payload(self, data):
        """Validate shortest_path v2/ POST endpoint."""
        if data.get("desired_links"):
            if not isinstance(data["desired_links"], list):
                raise BadRequest(
                    f"TypeError: desired_links is supposed to be a list."
                    f" type: {type(data['desired_links'])}"
                )

        if data.get("undesired_links"):
            if not isinstance(data["undesired_links"], list):
                raise BadRequest(
                    f"TypeError: undesired_links is supposed to be a list."
                    f" type: {type(data['undesired_links'])}"
                )

        parameter = data.get("parameter")
        spf_attr = data.get("spf_attribute")
        if not spf_attr:
            spf_attr = parameter or "hop"
        data["spf_attribute"] = spf_attr

        if spf_attr not in self.graph.spf_edge_data_cbs:
            raise BadRequest(
                "Invalid 'spf_attribute'. Valid values: "
                f"{', '.join(self.graph.spf_edge_data_cbs.keys())}"
            )

        try:
            data["spf_max_paths"] = max(int(data.get("spf_max_paths", 2)), 1)
        except (TypeError, ValueError):
            raise BadRequest(
                f"spf_max_paths {data.get('spf_max_pahts')} must be an int"
            )

        spf_max_path_cost = data.get("spf_max_path_cost")
        if spf_max_path_cost:
            try:
                spf_max_path_cost = max(int(spf_max_path_cost), 1)
                data["spf_max_path_cost"] = spf_max_path_cost
            except (TypeError, ValueError):
                raise BadRequest(
                    f"spf_max_path_cost {data.get('spf_max_path_cost')} must"
                    " be an int"
                )

        data["mandatory_metrics"] = data.get("mandatory_metrics", {})
        data["flexible_metrics"] = data.get("flexible_metrics", {})

        try:
            minimum_hits = data.get("minimum_flexible_hits")
            if minimum_hits:
                minimum_hits = min(
                    len(data["flexible_metrics"]), max(0, int(minimum_hits))
                )
            data["minimum_flexible_hits"] = minimum_hits
        except (TypeError, ValueError):
            raise BadRequest(
                f"minimum_hits {data.get('minimum_flexible_hits')} must be an int"
            )

        return data

    @rest("v2/", methods=["POST"])
    def shortest_path(self):
        """Calculate the best path between the source and destination."""
        data = request.get_json()
        data = self._validate_payload(data)

        desired = data.get("desired_links")
        undesired = data.get("undesired_links")

        spf_attr = data.get("spf_attribute")
        spf_max_paths = data.get("spf_max_paths")
        spf_max_path_cost = data.get("spf_max_path_cost")
        mandatory_metrics = data.get("mandatory_metrics")
        flexible_metrics = data.get("flexible_metrics")
        minimum_hits = data.get("minimum_flexible_hits")
        log.debug(f"POST v2/ payload data: {data}")

        try:
            with self._lock:
                if any([mandatory_metrics, flexible_metrics]):
                    paths = self.graph.constrained_k_shortest_paths(
                        data["source"],
                        data["destination"],
                        weight=self.graph.spf_edge_data_cbs[spf_attr],
                        k=spf_max_paths,
                        minimum_hits=minimum_hits,
                        mandatory_metrics=mandatory_metrics,
                        flexible_metrics=flexible_metrics,
                    )
                else:
                    paths = self.graph.k_shortest_paths(
                        data["source"],
                        data["destination"],
                        weight=self.graph.spf_edge_data_cbs[spf_attr],
                        k=spf_max_paths,
                    )

                paths = self.graph.path_cost_builder(
                    paths,
                    weight=spf_attr,
                )
            log.debug(f"Found paths: {paths}")
        except TypeError as err:
            raise BadRequest(str(err))

        paths = self._filter_paths_le_cost(paths, max_cost=spf_max_path_cost)
        paths = self._filter_paths_undesired_links(paths, undesired)
        paths = self._filter_paths_desired_links(paths, desired)
        log.debug(f"Filtered paths: {paths}")
        return jsonify({"paths": paths})

    @listen_to(
        "kytos.topology.updated",
        "kytos/topology.current",
        "kytos/topology.topology_loaded",
    )
    def on_topology_updated(self, event):
        """Update the graph when the network topology is updated."""
        self.update_topology(event)

    def update_topology(self, event):
        """Update the graph when the network topology is updated."""
        if "topology" not in event.content:
            return
        topology = event.content["topology"]
        with self._lock:
            if (
                self._topology_updated_at
                and self._topology_updated_at > event.timestamp
            ):
                return
            self._topology = topology
            self._topology_updated_at = event.timestamp
            self.graph.update_topology(topology)
        switches = list(topology.switches.keys())
        links = list(topology.links.keys())
        log.debug(f"Topology graph updated with switches: {switches}, links: {links}.")

    def update_links_metadata_changed(self, event) -> None:
        """Update the graph when links' metadata are added or removed."""
        link = event.content["link"]
        try:
            with self._lock:
                if (
                    link.id in self._links_updated_at
                    and self._links_updated_at[link.id] > event.timestamp
                ):
                    return
                self.graph.update_link_metadata(link)
                self._links_updated_at[link.id] = event.timestamp
            metadata = event.content["metadata"]
            log.debug(f"Topology graph updated link id: {link.id} metadata: {metadata}")
        except KeyError as exc:
            log.warning(
                f"Unexpected KeyError {str(exc)} on event {event}."
                " pathfinder will reconciliate the topology"
            )
            self.controller.buffers.app.put(KytosEvent(name="kytos/topology.get"))

    @listen_to("kytos/topology.links.metadata.(added|removed)")
    def on_links_metadata_changed(self, event):
        """Update the graph when links' metadata are added or removed."""
        self.update_links_metadata_changed(event)


1		"""Main module of kytos/pathfinder Kytos Network Application."""
2
3	1	from threading import Lock
4	1	from typing import Generator
5
6	1	from flask import jsonify, request
7	1	from kytos.core import KytosEvent, KytosNApp, log, rest
8	1	from kytos.core.helpers import listen_to
9	1	from napps.kytos.pathfinder.graph import KytosGraph
10		# pylint: disable=import-error
11	1	from werkzeug.exceptions import BadRequest
12
13
14	1	class Main(KytosNApp):
15		"""
16		Main class of kytos/pathfinder NApp.
17
18		This class is the entry point for this napp.
19		"""
20
21	1	def setup(self):
22		"""Create a graph to handle the nodes and edges."""
23	1	self.graph = KytosGraph()
24	1	self._topology = None
25	1	self._lock = Lock()
26	1	self._topology_updated_at = None
27	1	self._links_updated_at = {}
28
29	1	def execute(self):
30		"""Do nothing."""
31
32	1	def shutdown(self):
33		"""Shutdown the napp."""
34
35	1	def _filter_paths_le_cost(self, paths, max_cost):
36		"""Filter by paths where the cost is le <= max_cost."""
37	1	if not max_cost:
38	1	return paths
39	1	return [path for path in paths if path["cost"] <= max_cost]
40
41	1	def _map_endpoints_from_link_ids(self, link_ids: list[str]) -> dict:
42		"""Map endpoints from link ids."""
43	1	endpoints = {}
44	1	for link_id in link_ids:
45	1	try:
46	1	link = self._topology.links[link_id]
47	1	endpoint_a, endpoint_b = link.endpoint_a, link.endpoint_b
48	1	endpoints[(endpoint_a.id, endpoint_b.id)] = link
49	1	except KeyError:
50	1	pass
51	1	return endpoints
52
53	1	def _find_all_link_ids(
54		self, paths: list[dict], link_ids: list[str]
55		) -> Generator[int, None, None]:
56		"""Find indexes of the paths that contain all link ids."""
57	1	endpoints_links = self._map_endpoints_from_link_ids(link_ids)
58	1	if not endpoints_links:
59	1	return None
60	1	endpoint_keys = set(endpoints_links.keys())
61	1	for idx, path in enumerate(paths):
62	1	head, tail, found_endpoints = path["hops"][:-1], path["hops"][1:], set()
63	1	for endpoint_a, endpoint_b in zip(head, tail):
64	1	if (endpoint_a, endpoint_b) in endpoints_links:
65	1	found_endpoints.add((endpoint_a, endpoint_b))
66	1	if (endpoint_b, endpoint_a) in endpoints_links:
67		found_endpoints.add((endpoint_b, endpoint_a))
68	1	if found_endpoints == endpoint_keys:
69	1	yield idx
70	1	return None
71
72	1	def _find_any_link_ids(
73		self, paths: list[dict], link_ids: list[str]
74		) -> Generator[int, None, None]:
75		"""Find indexes of the paths that contain any of the link ids."""
76	1	endpoints_links = self._map_endpoints_from_link_ids(link_ids)
77	1	if not endpoints_links:
78	1	return None
79	1	for idx, path in enumerate(paths):
80	1	head, tail, found = path["hops"][:-1], path["hops"][1:], False
81	1	for endpoint_a, endpoint_b in zip(head, tail):
82	1	if any(
83		(
84		(endpoint_a, endpoint_b) in endpoints_links,
85		(endpoint_b, endpoint_a) in endpoints_links,
86		)
87		):
88	1	found = True
89	1	break
90	1	if found:
91	1	yield idx
92	1	return None
93
94	1	def _filter_paths_undesired_links(
95		self, paths: list[dict], undesired: list[str]
96		) -> list[dict]:
97		"""Filter by undesired_links, it performs a logical OR."""
98	1	if not undesired:
99	1	return paths
100	1	excluded_indexes = set(self._find_any_link_ids(paths, undesired))
101	1	return [path for idx, path in enumerate(paths) if idx not in excluded_indexes]
102
103	1	def _filter_paths_desired_links(
104		self, paths: list[dict], desired: list[str]
105		) -> list[dict]:
106		"""Filter by desired_links, it performs a logical AND."""
107	1	if not desired:
108	1	return paths
109	1	included_indexes = set(self._find_all_link_ids(paths, desired))
110	1	return [path for idx, path in enumerate(paths) if idx in included_indexes]
111
112	1	def _validate_payload(self, data):
113		"""Validate shortest_path v2/ POST endpoint."""
114	1	if data.get("desired_links"):
115	1	if not isinstance(data["desired_links"], list):
116		raise BadRequest(
117		f"TypeError: desired_links is supposed to be a list."
118		f" type: {type(data['desired_links'])}"
119		)
120
121	1	if data.get("undesired_links"):
122		if not isinstance(data["undesired_links"], list):
123		raise BadRequest(
124		f"TypeError: undesired_links is supposed to be a list."
125		f" type: {type(data['undesired_links'])}"
126		)
127
128	1	parameter = data.get("parameter")
129	1	spf_attr = data.get("spf_attribute")
130	1	if not spf_attr:
131	1	spf_attr = parameter or "hop"
132	1	data["spf_attribute"] = spf_attr
133
134	1	if spf_attr not in self.graph.spf_edge_data_cbs:
135		raise BadRequest(
136		"Invalid 'spf_attribute'. Valid values: "
137		f"{', '.join(self.graph.spf_edge_data_cbs.keys())}"
138		)
139
140	1	try:
141	1	data["spf_max_paths"] = max(int(data.get("spf_max_paths", 2)), 1)
142		except (TypeError, ValueError):
143		raise BadRequest(
144		f"spf_max_paths {data.get('spf_max_pahts')} must be an int"
145		)
146
147	1	spf_max_path_cost = data.get("spf_max_path_cost")
148	1	if spf_max_path_cost:
149		try:
150		spf_max_path_cost = max(int(spf_max_path_cost), 1)
151		data["spf_max_path_cost"] = spf_max_path_cost
152		except (TypeError, ValueError):
153		raise BadRequest(
154		f"spf_max_path_cost {data.get('spf_max_path_cost')} must"
155		" be an int"
156		)
157
158	1	data["mandatory_metrics"] = data.get("mandatory_metrics", {})
159	1	data["flexible_metrics"] = data.get("flexible_metrics", {})
160
161	1	try:
162	1	minimum_hits = data.get("minimum_flexible_hits")
163	1	if minimum_hits:
164	1	minimum_hits = min(
165		len(data["flexible_metrics"]), max(0, int(minimum_hits))
166		)
167	1	data["minimum_flexible_hits"] = minimum_hits
168		except (TypeError, ValueError):
169		raise BadRequest(
170		f"minimum_hits {data.get('minimum_flexible_hits')} must be an int"
171		)
172
173	1	return data
174
175	1	@rest("v2/", methods=["POST"])
176	1	def shortest_path(self):
177		"""Calculate the best path between the source and destination."""
178	1	data = request.get_json()
179	1	data = self._validate_payload(data)
180
181	1	desired = data.get("desired_links")
182	1	undesired = data.get("undesired_links")
183
184	1	spf_attr = data.get("spf_attribute")
185	1	spf_max_paths = data.get("spf_max_paths")
186	1	spf_max_path_cost = data.get("spf_max_path_cost")
187	1	mandatory_metrics = data.get("mandatory_metrics")
188	1	flexible_metrics = data.get("flexible_metrics")
189	1	minimum_hits = data.get("minimum_flexible_hits")
190	1	log.debug(f"POST v2/ payload data: {data}")
191
192	1	try:
193	1	with self._lock:
194	1	if any([mandatory_metrics, flexible_metrics]):
195	1	paths = self.graph.constrained_k_shortest_paths(
196		data["source"],
197		data["destination"],
198		weight=self.graph.spf_edge_data_cbs[spf_attr],
199		k=spf_max_paths,
200		minimum_hits=minimum_hits,
201		mandatory_metrics=mandatory_metrics,
202		flexible_metrics=flexible_metrics,
203		)
204		else:
205	1	paths = self.graph.k_shortest_paths(
206		data["source"],
207		data["destination"],
208		weight=self.graph.spf_edge_data_cbs[spf_attr],
209		k=spf_max_paths,
210		)
211
212	1	paths = self.graph.path_cost_builder(
213		paths,
214		weight=spf_attr,
215		)
216	1	log.debug(f"Found paths: {paths}")
217	1	except TypeError as err:
218	1	raise BadRequest(str(err))
219
220	1	paths = self._filter_paths_le_cost(paths, max_cost=spf_max_path_cost)
221	1	paths = self._filter_paths_undesired_links(paths, undesired)
222	1	paths = self._filter_paths_desired_links(paths, desired)
223	1	log.debug(f"Filtered paths: {paths}")
224	1	return jsonify({"paths": paths})
225
226	1	@listen_to(
227		"kytos.topology.updated",
228		"kytos/topology.current",
229		"kytos/topology.topology_loaded",
230		)
231	1	def on_topology_updated(self, event):
232		"""Update the graph when the network topology is updated."""
233		self.update_topology(event)
234
235	1	def update_topology(self, event):
236		"""Update the graph when the network topology is updated."""
237	1	if "topology" not in event.content:
238	1	return
239	1	topology = event.content["topology"]
240	1	with self._lock:
241	1	if (
242		self._topology_updated_at
243		and self._topology_updated_at > event.timestamp
244		):
245	1	return
246	1	self._topology = topology
247	1	self._topology_updated_at = event.timestamp
248	1	self.graph.update_topology(topology)
249	1	switches = list(topology.switches.keys())
250	1	links = list(topology.links.keys())
251	1	log.debug(f"Topology graph updated with switches: {switches}, links: {links}.")
252
253	1	def update_links_metadata_changed(self, event) -> None:
254		"""Update the graph when links' metadata are added or removed."""
255	1	link = event.content["link"]
256	1	try:
257	1	with self._lock:
258	1	if (
259		link.id in self._links_updated_at
260		and self._links_updated_at[link.id] > event.timestamp
261		):
262	1	return
263	1	self.graph.update_link_metadata(link)
264	1	self._links_updated_at[link.id] = event.timestamp
265	1	metadata = event.content["metadata"]
266	1	log.debug(f"Topology graph updated link id: {link.id} metadata: {metadata}")
267	1	except KeyError as exc:
268	1	log.warning(
269		f"Unexpected KeyError {str(exc)} on event {event}."
270		" pathfinder will reconciliate the topology"
271		)
272	1	self.controller.buffers.app.put(KytosEvent(name="kytos/topology.get"))
273
274	1	@listen_to("kytos/topology.links.metadata.(added\|removed)")
275	1	def on_links_metadata_changed(self, event):
276		"""Update the graph when links' metadata are added or removed."""
277		self.update_links_metadata_changed(event)
278

kytos-ng / pathfinder

Pull Request — master (#35)

build.main.Main._map_endpoints_from_link_ids() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like