build.graph.KytosGraph.__init__() - Code Metrics - Inspection of "Logic Changes" - kytos/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#58)

unknown

created 2020-06-20 02:30 UTC

build.graph.KytosGraph.init() B

↳ Parent: build.graph

Complexity

Conditions

Size

Total Lines	26
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	16
CRAP Score	7.0671

Importance

Changes

Metric	Value
cc	7
eloc	22
nop	1
dl	0
loc	26
rs	7.952
c	0
b	0
f	0
ccs	16
cts	18
cp	0.8889
crap	7.0671

"""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_fun = 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_fun(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_fun_dict = {}

        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_fun_dict["ownership"] = Filter(
            str, filter_eeq("ownership"))
        self._filter_fun_dict["bandwidth"] = Filter(
            (int, float), filter_geq("bandwidth"))
        self._filter_fun_dict["priority"] = Filter(
            (int, float), filter_geq("priority"))
        self._filter_fun_dict["reliability"] = Filter(
            (int, float), filter_geq("reliability"))
        self._filter_fun_dict["utilization"] = Filter(
            (int, float), filter_leq("utilization"))
        self._filter_fun_dict["delay"] = Filter(
            (int, float), filter_leq("delay"))
        self._path_fun = nx.all_shortest_paths

    def set_path_fun(self, path_fun):
        """Set the shortest path function."""
        self._path_fun = path_fun

    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

    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(self._path_fun(self.graph,
                                        source, destination, parameter))
        except (NodeNotFound, NetworkXNoPath):
            return []
        return paths

    def constrained_flexible_paths(self, source, destination, complete_metrics,
                                   flexible=None):
        """Calculate the constrained shortest paths with flexibility."""
        metrics, flexible_metrics = KytosGraph.unpack_metrics(complete_metrics)
        # Grab all edges with meta data and remove the
        # ones that do not meet metric requirements.
        default_edge_list = list(self._filter_edges(
            self.graph.edges(data=True), **metrics))
        # Determine the size of the power subset of flexible metrics.
        length = len(flexible_metrics)
        if flexible is None:
            flexible = length
        flexible = min(length, max(0, flexible))
        results = []
        result = []
        i = 0
        # Traverse through each combination in the power subset
        # and use the combination to find edges that partially
        # meet flexible metric requirements.
        while (result == [] and i in range(0, flexible+1)):
            for combo in combinations(flexible_metrics.items(), length-i):
                temp_dict = dict(combo)
                result = self._constrained_shortest_paths(
                    source, destination, ((u, v) for u, v, d in

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

    @staticmethod
    def pack_metrics(inflexible=None, flexible=None):
        """Store flexible and inflexible metrics in a dictionary."""
        if inflexible is None:
            inflexible = {}
        if flexible is None:
            flexible = {}
        return dict(metrics=inflexible, flexible_metrics=flexible)

    @staticmethod
    def unpack_metrics(complete_metrics):
        """Return inflexible and flexible metrics."""
        return (complete_metrics["metrics"],
                complete_metrics["flexible_metrics"])

    def _constrained_shortest_paths(self, source, destination, edges):
        paths = []
        try:
            paths = list(self._path_fun(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():
            fil = self._filter_fun_dict.get(metric, None)
            if fil is not None:
                try:
                    edges = fil.run(value, edges)
                except TypeError as err:
                    raise TypeError(f"Error in {metric} filter: {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	1
7	1	try:
8		import networkx as nx
9		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	1
14
15		class Filter:
16	1	"""Class responsible for removing items with disqualifying values."""
17	1
18		def __init__(self, filter_type, filter_function):
19	1	self._filter_type = filter_type
20		self._filter_fun = filter_function
21	1
22		def run(self, value, items):
23	1	"""Filter out items. Filter chosen is picked at runtime."""
24		if isinstance(value, self._filter_type):
25	1	return filter(self._filter_fun(value), items)
26	1
27	1	raise TypeError(f"Expected type: {self._filter_type}")
28
29	1
30		class KytosGraph:
31	1	"""Class responsible for the graph generation."""
32	1
33	1	def __init__(self):
34		self.graph = nx.Graph()
35	1	self._filter_fun_dict = {}
36	1
37	1	def filter_leq(metric): # Lower values are better
38		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		def filter_geq(metric): # Higher values are better
41		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	1
43		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	1
46	1	self._filter_fun_dict["ownership"] = Filter(
47	1	str, filter_eeq("ownership"))
48	1	self._filter_fun_dict["bandwidth"] = Filter(
49	1	(int, float), filter_geq("bandwidth"))
50	1	self._filter_fun_dict["priority"] = Filter(
51	1	(int, float), filter_geq("priority"))
52	1	self._filter_fun_dict["reliability"] = Filter(
53		(int, float), filter_geq("reliability"))
54	1	self._filter_fun_dict["utilization"] = Filter(
55		(int, float), filter_leq("utilization"))
56	1	self._filter_fun_dict["delay"] = Filter(
57		(int, float), filter_leq("delay"))
58		self._path_fun = nx.all_shortest_paths
59
60		def set_path_fun(self, path_fun):
61		"""Set the shortest path function."""
62	1	self._path_fun = path_fun
63	1
64	1	def clear(self):
65	1	"""Remove all nodes and links registered."""
66		self.graph.clear()
67	1
68		def update_topology(self, topology):
69		"""Update all nodes and links inside the graph."""
70	1	self.graph.clear()
71	1	self.update_nodes(topology.switches)
72	1	self.update_links(topology.links)
73
74	1	def update_nodes(self, nodes):
75		"""Update all nodes inside the graph."""
76	1	for node in nodes.values():
77	1	try:
78		self.graph.add_node(node.id)
79
80		for interface in node.interfaces.values():
81		self.graph.add_node(interface.id)
82	1	self.graph.add_edge(node.id, interface.id)
83
84		except AttributeError:
85		pass
86
87		def update_links(self, links):
88		"""Update all links inside the graph."""
89		keys = []
90		for link in links.values():
91		if link.is_active():
92		self.graph.add_edge(link.endpoint_a.id, link.endpoint_b.id)
93		for key, value in link.metadata.items():
94		keys.append(key)
95		endpoint_a = link.endpoint_a.id
96		endpoint_b = link.endpoint_b.id
97		self.graph[endpoint_a][endpoint_b][key] = value
98
99		def get_metadata_from_link(self, endpoint_a, endpoint_b):
100		"""Return the metadata of a link."""
101		return self.graph.edges[endpoint_a, endpoint_b]
102
103		@staticmethod
104		def _remove_switch_hops(circuit):
105		"""Remove switch hops from a circuit hops list."""
106		for hop in circuit['hops']:
107		if len(hop.split(':')) == 8:
108		circuit['hops'].remove(hop)
109
110		def shortest_paths(self, source, destination, parameter=None):
111		"""Calculate the shortest paths and return them."""
112		try:
113		paths = list(self._path_fun(self.graph,
114		source, destination, parameter))
115		except (NodeNotFound, NetworkXNoPath):
116		return []
117		return paths
118
119		def constrained_flexible_paths(self, source, destination, complete_metrics,
120		flexible=None):
121		"""Calculate the constrained shortest paths with flexibility."""
122		metrics, flexible_metrics = KytosGraph.unpack_metrics(complete_metrics)
123		# Grab all edges with meta data and remove the
124		# ones that do not meet metric requirements.
125		default_edge_list = list(self._filter_edges(
126		self.graph.edges(data=True), **metrics))
127		# Determine the size of the power subset of flexible metrics.
128		length = len(flexible_metrics)
129		if flexible is None:
130		flexible = length
131		flexible = min(length, max(0, flexible))
132		results = []
133		result = []
134		i = 0
135		# Traverse through each combination in the power subset
136		# and use the combination to find edges that partially
137		# meet flexible metric requirements.
138		while (result == [] and i in range(0, flexible+1)):
139		for combo in combinations(flexible_metrics.items(), length-i):
140		temp_dict = dict(combo)
141		result = self._constrained_shortest_paths(
142		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 `v` 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 `u` does not seem to be defined for all execution paths. Loading history...
143		self._filter_edges(default_edge_list,
144		**temp_dict)))
145		if result != []:
146		results.append(
147		{"paths": result, "metrics": {metrics, temp_dict}})
148		i = i + 1
149		return results
150
151		@staticmethod
152		def pack_metrics(inflexible=None, flexible=None):
153		"""Store flexible and inflexible metrics in a dictionary."""
154		if inflexible is None:
155		inflexible = {}
156		if flexible is None:
157		flexible = {}
158		return dict(metrics=inflexible, flexible_metrics=flexible)
159
160		@staticmethod
161		def unpack_metrics(complete_metrics):
162		"""Return inflexible and flexible metrics."""
163		return (complete_metrics["metrics"],
164		complete_metrics["flexible_metrics"])
165
166		def _constrained_shortest_paths(self, source, destination, edges):
167		paths = []
168		try:
169		paths = list(self._path_fun(self.graph.edge_subgraph(edges),
170		source, destination))
171		except NetworkXNoPath:
172		pass
173		except NodeNotFound:
174		if source == destination:
175		if source in self.graph.nodes:
176		paths = [[source]]
177		return paths
178
179		def _filter_edges(self, edges, **metrics):
180		for metric, value in metrics.items():
181		fil = self._filter_fun_dict.get(metric, None)
182		if fil is not None:
183		try:
184		edges = fil.run(value, edges)
185		except TypeError as err:
186		raise TypeError(f"Error in {metric} filter: {err}")
187		return edges
188

kytos / pathfinder

Pull Request — master (#58)

build.graph.KytosGraph.__init__() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

build.graph.KytosGraph.init() B