build.graph.KytosGraph.get_metadata_from_link() - Code Metrics - Inspection of "Find Best Path Using Constraints" - kytos/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#58)

unknown

created 2020-07-17 00:06 UTC

build.graph.KytosGraph.get_metadata_from_link() A

↳ Parent: build.graph

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

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

"""Module Graph of kytos/pathfinder Kytos Network Application."""

from itertools import combinations

from kytos.core import log

try:
    import networkx as nx
    from networkx.exception import NodeNotFound, NetworkXNoPath
except ImportError:
    PACKAGE = 'networkx>=2.2'
    log.error(f"Package {PACKAGE} not found. Please 'pip install {PACKAGE}'")


class Filter:
    """Class responsible for removing items with disqualifying values."""

    def __init__(self, filter_type, filter_function):
        self._filter_type = filter_type
        self._filter_function = filter_function

    def run(self, value, items):
        """Filter out items. Filter chosen is picked at runtime."""
        if isinstance(value, self._filter_type):
            return filter(self._filter_function(value), items)

        raise TypeError(f"Expected type: {self._filter_type}")


class KytosGraph:
    """Class responsible for the graph generation."""

    def __init__(self):
        self.graph = nx.Graph()
        self._filter_functions = {}

        def filter_leq(metric):  # Lower values are better
            return lambda x: (lambda y: y[2].get(metric, x) <= x)


        def filter_geq(metric):  # Higher values are better
            return lambda x: (lambda y: y[2].get(metric, x) >= x)


        def filter_eeq(metric):  # Equivalence
            return lambda x: (lambda y: y[2].get(metric, x) == x)


        self._filter_functions["ownership"] = Filter(
            str, filter_eeq("ownership"))
        self._filter_functions["bandwidth"] = Filter(
            (int, float), filter_geq("bandwidth"))
        self._filter_functions["priority"] = Filter(
            (int, float), filter_geq("priority"))
        self._filter_functions["reliability"] = Filter(
            (int, float), filter_geq("reliability"))
        self._filter_functions["utilization"] = Filter(
            (int, float), filter_leq("utilization"))
        self._filter_functions["delay"] = Filter(
            (int, float), filter_leq("delay"))
        self._path_function = nx.all_shortest_paths

    def clear(self):
        """Remove all nodes and links registered."""
        self.graph.clear()

    def update_topology(self, topology):
        """Update all nodes and links inside the graph."""
        self.graph.clear()
        self.update_nodes(topology.switches)
        self.update_links(topology.links)

    def update_nodes(self, nodes):
        """Update all nodes inside the graph."""
        for node in nodes.values():
            try:
                self.graph.add_node(node.id)

                for interface in node.interfaces.values():
                    self.graph.add_node(interface.id)
                    self.graph.add_edge(node.id, interface.id)

            except AttributeError:
                pass

    def update_links(self, links):
        """Update all links inside the graph."""
        keys = []
        for link in links.values():
            if link.is_active():
                self.graph.add_edge(link.endpoint_a.id, link.endpoint_b.id)
                for key, value in link.metadata.items():
                    keys.append(key)
                    endpoint_a = link.endpoint_a.id
                    endpoint_b = link.endpoint_b.id
                    self.graph[endpoint_a][endpoint_b][key] = value

        self._set_default_metadata(keys)

    def _set_default_metadata(self, keys):
        """Set metadata to all links.

        Set the value to zero for inexistent metadata in a link to make those
        irrelevant in pathfinding.
        """
        for key in keys:
            for endpoint_a, endpoint_b in self.graph.edges:
                if key not in self.graph[endpoint_a][endpoint_b]:
                    self.graph[endpoint_a][endpoint_b][key] = 0

    def get_metadata_from_link(self, endpoint_a, endpoint_b):
        """Return the metadata of a link."""
        return self.graph.edges[endpoint_a, endpoint_b]

    @staticmethod
    def _remove_switch_hops(circuit):
        """Remove switch hops from a circuit hops list."""
        for hop in circuit['hops']:
            if len(hop.split(':')) == 8:
                circuit['hops'].remove(hop)

    def shortest_paths(self, source, destination, parameter=None):
        """Calculate the shortest paths and return them."""
        try:
            paths = list(nx.shortest_simple_paths(self.graph,
                                                  source, destination,
                                                  parameter))
        except (NodeNotFound, NetworkXNoPath):
            return []
        return paths

    def constrained_flexible_paths(self, source, destination,
                                   maximum_misses=None, **metrics):
        """Calculate the constrained shortest paths with flexibility."""
        base = metrics.get("base", {})
        flexible = metrics.get("flexible", {})
        default_edge_list = list(self._filter_edges(
            self.graph.edges(data=True), **base))
        length = len(flexible)
        if maximum_misses is None:
            maximum_misses = length
        maximum_misses = min(length, max(0, maximum_misses))
        results = []
        paths = []
        i = 0
        while (paths == [] and i in range(0, maximum_misses+1)):
            for combo in combinations(flexible.items(), length-i):
                additional = dict(combo)
                paths = self._constrained_shortest_paths(
                    source, destination, ((u, v) for u, v, d in

                                          self._filter_edges(default_edge_list,
                                                             **additional)))
                if paths != []:
                    results.append(
                        {"paths": paths, "metrics": {**base, **additional}})
            i = i + 1
        return results

    def _constrained_shortest_paths(self, source, destination, edges):
        paths = []
        try:
            paths = list(self._path_function(self.graph.edge_subgraph(edges),
                                             source, destination))
        except NetworkXNoPath:
            pass
        except NodeNotFound:
            if source == destination:
                if source in self.graph.nodes:
                    paths = [[source]]
        return paths

    def _filter_edges(self, edges, **metrics):
        for metric, value in metrics.items():
            filter_ = self._filter_functions.get(metric, None)
            if filter_ is not None:
                try:
                    edges = filter_.run(value, edges)
                except TypeError as err:
                    raise TypeError(f"Error in {metric} value: {err}")
        return edges


1		"""Module Graph of kytos/pathfinder Kytos Network Application."""
2
3	1	from itertools import combinations
4
5	1	from kytos.core import log
6
7	1	try:
8	1	import networkx as nx
9	1	from networkx.exception import NodeNotFound, NetworkXNoPath
10		except ImportError:
11		PACKAGE = 'networkx>=2.2'
12		log.error(f"Package {PACKAGE} not found. Please 'pip install {PACKAGE}'")
13
14
15	1	class Filter:
16		"""Class responsible for removing items with disqualifying values."""
17
18	1	def __init__(self, filter_type, filter_function):
19	1	self._filter_type = filter_type
20	1	self._filter_function = filter_function
21
22	1	def run(self, value, items):
23		"""Filter out items. Filter chosen is picked at runtime."""
24	1	if isinstance(value, self._filter_type):
25	1	return filter(self._filter_function(value), items)
26
27	1	raise TypeError(f"Expected type: {self._filter_type}")
28
29
30	1	class KytosGraph:
31		"""Class responsible for the graph generation."""
32
33	1	def __init__(self):
34	1	self.graph = nx.Graph()
35	1	self._filter_functions = {}
36
37	1	def filter_leq(metric): # Lower values are better
38	1	return lambda x: (lambda y: y[2].get(metric, x) <= x)
		0 ignored issues – show Comprehensibility Best Practice introduced 2020-05-28 14:34 UTC by Report Bug Copy Issue Report The variable `x` does not seem to be defined. Loading history...
39
40	1	def filter_geq(metric): # Higher values are better
41	1	return lambda x: (lambda y: y[2].get(metric, x) >= x)
		0 ignored issues – show Comprehensibility Best Practice introduced 2020-05-28 14:34 UTC by Report Bug Copy Issue Report The variable `x` does not seem to be defined. Loading history...
42
43	1	def filter_eeq(metric): # Equivalence
44	1	return lambda x: (lambda y: y[2].get(metric, x) == x)
		0 ignored issues – show Comprehensibility Best Practice introduced 2020-05-28 14:34 UTC by Report Bug Copy Issue Report The variable `x` does not seem to be defined. Loading history...
45
46	1	self._filter_functions["ownership"] = Filter(
47		str, filter_eeq("ownership"))
48	1	self._filter_functions["bandwidth"] = Filter(
49		(int, float), filter_geq("bandwidth"))
50	1	self._filter_functions["priority"] = Filter(
51		(int, float), filter_geq("priority"))
52	1	self._filter_functions["reliability"] = Filter(
53		(int, float), filter_geq("reliability"))
54	1	self._filter_functions["utilization"] = Filter(
55		(int, float), filter_leq("utilization"))
56	1	self._filter_functions["delay"] = Filter(
57		(int, float), filter_leq("delay"))
58	1	self._path_function = nx.all_shortest_paths
59
60	1	def clear(self):
61		"""Remove all nodes and links registered."""
62	1	self.graph.clear()
63
64	1	def update_topology(self, topology):
65		"""Update all nodes and links inside the graph."""
66	1	self.graph.clear()
67	1	self.update_nodes(topology.switches)
68	1	self.update_links(topology.links)
69
70	1	def update_nodes(self, nodes):
71		"""Update all nodes inside the graph."""
72	1	for node in nodes.values():
73	1	try:
74	1	self.graph.add_node(node.id)
75
76	1	for interface in node.interfaces.values():
77	1	self.graph.add_node(interface.id)
78	1	self.graph.add_edge(node.id, interface.id)
79
80		except AttributeError:
81		pass
82
83	1	def update_links(self, links):
84		"""Update all links inside the graph."""
85	1	keys = []
86	1	for link in links.values():
87	1	if link.is_active():
88	1	self.graph.add_edge(link.endpoint_a.id, link.endpoint_b.id)
89	1	for key, value in link.metadata.items():
90	1	keys.append(key)
91	1	endpoint_a = link.endpoint_a.id
92	1	endpoint_b = link.endpoint_b.id
93	1	self.graph[endpoint_a][endpoint_b][key] = value
94
95	1	self._set_default_metadata(keys)
96
97	1	def _set_default_metadata(self, keys):
98		"""Set metadata to all links.
99
100		Set the value to zero for inexistent metadata in a link to make those
101		irrelevant in pathfinding.
102		"""
103	1	for key in keys:
104	1	for endpoint_a, endpoint_b in self.graph.edges:
105	1	if key not in self.graph[endpoint_a][endpoint_b]:
106	1	self.graph[endpoint_a][endpoint_b][key] = 0
107
108	1	def get_metadata_from_link(self, endpoint_a, endpoint_b):
109		"""Return the metadata of a link."""
110	1	return self.graph.edges[endpoint_a, endpoint_b]
111
112	1	@staticmethod
113		def _remove_switch_hops(circuit):
114		"""Remove switch hops from a circuit hops list."""
115	1	for hop in circuit['hops']:
116	1	if len(hop.split(':')) == 8:
117	1	circuit['hops'].remove(hop)
118
119	1	def shortest_paths(self, source, destination, parameter=None):
120		"""Calculate the shortest paths and return them."""
121	1	try:
122	1	paths = list(nx.shortest_simple_paths(self.graph,
123		source, destination,
124		parameter))
125		except (NodeNotFound, NetworkXNoPath):
126		return []
127	1	return paths
128
129	1	def constrained_flexible_paths(self, source, destination,
130		maximum_misses=None, **metrics):
131		"""Calculate the constrained shortest paths with flexibility."""
132	1	base = metrics.get("base", {})
133	1	flexible = metrics.get("flexible", {})
134	1	default_edge_list = list(self._filter_edges(
135		self.graph.edges(data=True), **base))
136	1	length = len(flexible)
137	1	if maximum_misses is None:
138	1	maximum_misses = length
139	1	maximum_misses = min(length, max(0, maximum_misses))
140	1	results = []
141	1	paths = []
142	1	i = 0
143	1	while (paths == [] and i in range(0, maximum_misses+1)):
144	1	for combo in combinations(flexible.items(), length-i):
145	1	additional = dict(combo)
146	1	paths = self._constrained_shortest_paths(
147		source, destination, ((u, v) for u, v, d in
		0 ignored issues – show introduced 2020-06-18 21:53 UTC by Report Bug Copy Issue Report The variable `u` does not seem to be defined for all execution paths. Loading history... introduced 2020-06-18 21:53 UTC by Report Bug Copy Issue Report The variable `v` does not seem to be defined for all execution paths. Loading history...
148		self._filter_edges(default_edge_list,
149		**additional)))
150	1	if paths != []:
151	1	results.append(
152		{"paths": paths, "metrics": {base, additional}})
153	1	i = i + 1
154	1	return results
155
156	1	def _constrained_shortest_paths(self, source, destination, edges):
157	1	paths = []
158	1	try:
159	1	paths = list(self._path_function(self.graph.edge_subgraph(edges),
160		source, destination))
161	1	except NetworkXNoPath:
162	1	pass
163	1	except NodeNotFound:
164	1	if source == destination:
165	1	if source in self.graph.nodes:
166	1	paths = [[source]]
167	1	return paths
168
169	1	def _filter_edges(self, edges, **metrics):
170	1	for metric, value in metrics.items():
171	1	filter_ = self._filter_functions.get(metric, None)
172	1	if filter_ is not None:
173	1	try:
174	1	edges = filter_.run(value, edges)
175	1	except TypeError as err:
176	1	raise TypeError(f"Error in {metric} value: {err}")
177		return edges
178

kytos / pathfinder

Pull Request — master (#58)

build.graph.KytosGraph.get_metadata_from_link() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like