1
|
|
|
"""Utils module of kytos/pathfinder Kytos Network Application.""" |
2
|
|
|
# pylint: disable=unused-argument |
3
|
|
|
|
4
|
|
|
|
5
|
|
|
def lazy_filter(filter_type, filter_func): |
6
|
|
|
""" |
7
|
|
|
Lazy typed filter on top of the built-in function. |
8
|
|
|
|
9
|
|
|
It's meant to be used when the values to be filtered for |
10
|
|
|
are only defined later on dynamically at runtime. |
11
|
|
|
""" |
12
|
|
|
|
13
|
|
|
def filter_closure(value, items): |
14
|
|
|
if not isinstance(value, filter_type): |
15
|
|
|
raise TypeError(f"Expected type: {filter_type}") |
16
|
|
|
return filter(filter_func(value), items) |
17
|
|
|
|
18
|
|
|
return filter_closure |
19
|
|
|
|
20
|
|
|
|
21
|
|
|
def nx_edge_data_weight(edge_u, edge_v, edge_data): |
22
|
|
|
"""Return custom edge data value to be used as a callback by nx.""" |
23
|
|
|
if edge_data.get("hop"): |
24
|
|
|
return edge_data["hop"] |
25
|
|
|
return 1 |
26
|
|
|
|
27
|
|
|
|
28
|
|
|
def nx_edge_data_delay(edge_u, edge_v, edge_data): |
29
|
|
|
"""Return custom edge data value to be used as a callback by nx.""" |
30
|
|
|
if edge_data.get("delay"): |
31
|
|
|
return edge_data["delay"] |
32
|
|
|
return 1 |
33
|
|
|
|
34
|
|
|
|
35
|
|
|
def nx_edge_data_priority(edge_u, edge_v, edge_data): |
36
|
|
|
"""Return custom edge data value to be used as a callback by nx.""" |
37
|
|
|
if edge_data.get("priority"): |
38
|
|
|
return edge_data["priority"] |
39
|
|
|
return 1 |
40
|
|
|
|
41
|
|
|
|
42
|
|
|
def filter_le(metric): |
43
|
|
|
"""Lazy filter_le.""" |
44
|
|
|
return lambda x: (lambda nx_edge_tup: nx_edge_tup[2].get(metric, x) <= x) |
|
|
|
|
45
|
|
|
|
46
|
|
|
|
47
|
|
|
def filter_ge(metric): |
48
|
|
|
"""Lazy filter_ge.""" |
49
|
|
|
return lambda x: (lambda nx_edge_tup: nx_edge_tup[2].get(metric, x) >= x) |
|
|
|
|
50
|
|
|
|
51
|
|
|
|
52
|
|
|
def filter_in(metric): |
53
|
|
|
"""Lazy filter_in.""" |
54
|
|
|
return lambda x: (lambda nx_edge_tup: x in nx_edge_tup[2].get(metric, {x})) |
|
|
|
|
55
|
|
|
|