|
1
|
|
|
"""Utils module of kytos/pathfinder Kytos Network Application.""" |
|
2
|
|
|
# pylint: disable=unused-argument |
|
3
|
|
|
|
|
4
|
|
|
|
|
5
|
1 |
|
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
|
1 |
|
def filter_closure(value, items): |
|
14
|
1 |
|
if not isinstance(value, filter_type): |
|
15
|
1 |
|
raise TypeError(f"Expected type: {filter_type}") |
|
16
|
1 |
|
return filter(filter_func(value), items) |
|
17
|
|
|
|
|
18
|
1 |
|
return filter_closure |
|
19
|
|
|
|
|
20
|
|
|
|
|
21
|
1 |
|
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
|
1 |
|
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
|
1 |
|
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
|
1 |
|
def filter_le(metric): |
|
43
|
|
|
"""Lazy filter_le.""" |
|
44
|
1 |
|
return lambda x: (lambda nx_edge_tup: nx_edge_tup[2].get(metric, x) <= x) |
|
|
|
|
|
|
45
|
|
|
|
|
46
|
|
|
|
|
47
|
1 |
|
def filter_ge(metric): |
|
48
|
|
|
"""Lazy filter_ge.""" |
|
49
|
1 |
|
return lambda x: (lambda nx_edge_tup: nx_edge_tup[2].get(metric, x) >= x) |
|
|
|
|
|
|
50
|
|
|
|
|
51
|
|
|
|
|
52
|
1 |
|
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
|
|
|
|