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) |
|
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
![]() |
|||
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) |
|
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
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})) |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
|
|||
55 |