-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathfeatures_test.py
174 lines (146 loc) · 6.97 KB
/
features_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
"""Tests the feature extraction"""
# Tests are verbose.
# pylint: disable=too-many-statements, too-many-locals
import math
import numpy as np
from pytest import approx
from infrastructure import InfrastructureNetwork
from overlay import OverlayNetwork
from embedding import PartialEmbedding, ENode
from features import features_by_name
# easiest to do everything in one function, although it isn't pretty
def test_features():
"""Tests feature extractions on some hand-verified examples"""
infra = InfrastructureNetwork()
nso1 = infra.add_source(
name="nso1", pos=(0, 0), transmit_power_dbm=30, capacity=1.5
)
nso2 = infra.add_source(
name="nso2", pos=(0, 1), transmit_power_dbm=30, capacity=2.8
)
nsi = infra.set_sink(
name="nsi", pos=(2, 0), transmit_power_dbm=30, capacity=np.infty
)
ninterm = infra.add_intermediate(
name="ninterm", pos=(0, 1), transmit_power_dbm=30, capacity=0
)
overlay = OverlayNetwork()
bso1 = overlay.add_source(name="bso1", datarate=5, requirement=1)
bso2 = overlay.add_source(name="bso2", datarate=8, requirement=2)
bin1 = overlay.add_intermediate(name="bin1", datarate=5, requirement=3)
bin2 = overlay.add_intermediate(name="bin2", datarate=5, requirement=0)
bin3 = overlay.add_intermediate(name="bin3", datarate=42, requirement=0.7)
bin4 = overlay.add_intermediate(name="bin4", datarate=0, requirement=0.2)
bsi = overlay.set_sink(name="bsi", datarate=5, requirement=4)
overlay.add_link(bso1, bin1)
overlay.add_link(bso1, bin2)
overlay.add_link(bin1, bsi)
overlay.add_link(bin2, bsi)
overlay.add_link(bso2, bsi)
overlay.add_link(bso2, bin3)
overlay.add_link(bin3, bin4)
overlay.add_link(bin4, bsi)
embedding = PartialEmbedding(
infra, overlay, source_mapping=[(bso1, nso1), (bso2, nso2)]
)
eso1 = ENode(bso1, nso1)
eso2 = ENode(bso2, nso2)
esi = ENode(bsi, nsi)
erelay = ENode(bso1, ninterm, bin1)
ein = ENode(bin1, nsi)
erelay_unchosen = ENode(bso2, ninterm, bsi)
assert embedding.take_action(eso1, erelay, 0)
assert embedding.take_action(erelay, ein, 1)
feature_dict = features_by_name()
def node_feature(name, node):
assert node in embedding.nodes()
return tuple(
feature_dict["node_" + name].process_node(embedding, node)
)
assert node_feature("pos", ein) == (2, 0) # pos of nsi
assert node_feature("relay", erelay) == (1.0,)
assert node_feature("relay", erelay_unchosen) == (1.0,)
assert node_feature("relay", eso1) == (0.0,)
assert node_feature("relay", ein) == (0.0,)
assert node_feature("sink", eso2) == (0.0,)
assert node_feature("sink", ein) == (0.0,)
assert node_feature("sink", esi) == (1.0,)
num_sinks = 0
for node in embedding.nodes():
if node_feature("sink", node)[0] == 1.0:
num_sinks += 1
assert num_sinks == 1
# this is always pretending the node isn't already chosen, so the
# requirement of the block in question is always exempt
assert node_feature("remaining_capacity", eso1)[0] == approx(1.5)
assert node_feature("remaining_capacity", eso2)[0] == approx(2.8)
assert node_feature("remaining_capacity", esi)[0] == np.infty
assert node_feature("remaining_capacity", erelay)[0] == approx(0)
assert node_feature("remaining_capacity", ein)[0] == np.infty
assert node_feature("remaining_capacity", erelay_unchosen)[0] == approx(0)
assert node_feature("weight", eso1)[0] == approx(1)
assert node_feature("weight", esi)[0] == approx(4)
assert node_feature("weight", erelay)[0] == approx(0)
# the block itself should not be counted if it is already embedded
# therefore, nso1 has a remaining capacity of 1.5 (instead of 0.5)
assert node_feature("compute_fraction", eso1)[0] == approx(1 / 1.5)
assert node_feature("compute_fraction", esi)[0] == 0 # /infty
assert node_feature("compute_fraction", erelay)[0] == approx(0)
# capacity inf can always embed everything
assert node_feature("options_lost", esi)[0] == 0
# has remaining capacity .5 after, can embed bin2 or bin4; before
# also bin4
assert node_feature("options_lost", eso1)[0] == 1
# has remaining capacity .8 after, can embed bin2 or bin3 (just as
# before)
assert node_feature("options_lost", eso2)[0] == 0
# remaining capacity of .1 after, can only embed bin2 (before also
# bin4)
assert node_feature("options_lost", ENode(bin3, nso2))[0] == 1
def edge_feature(name, u, v, t):
assert (u, v, t) in embedding.graph.edges(keys=True)
return tuple(
feature_dict["edge_" + name].process_edge(embedding, u, v, t)
)
assert edge_feature("timeslot", eso1, erelay, 0)[0] == 0
assert edge_feature("timeslot", eso2, esi, 2)[0] == 2
assert edge_feature("chosen", eso1, erelay, 0)[0] == 1
assert edge_feature("chosen", eso2, esi, 2)[0] == 0
assert edge_feature("capacity", eso2, esi, 2)[0] == approx(17.61, abs=0.1)
# loops are special-cased; pretend perfect match with requirement
assert edge_feature("capacity", ein, esi, 2)[0] == 3.0
assert edge_feature("additional_timeslot", eso1, erelay, 0)[0] == 0
assert edge_feature("additional_timeslot", eso2, esi, 2)[0] == 1
assert edge_feature("datarate_requirement", eso1, erelay, 0)[0] == approx(
5
)
assert edge_feature("datarate_requirement", erelay, ein, 1)[0] == approx(5)
# Capacity if nothing else is sending (which we're assuming, since actually
# only nso1 is sending and we're ignoring the current edge)
sinr = infra.sinr(eso1.node, erelay.node, senders=frozenset())
capacity = 1 * math.log(1 + 10 ** (sinr / 10), 2)
assert edge_feature("datarate_fraction", eso1, erelay, 0)[0] == approx(
5 / capacity
)
# in this case there actually is nothing currently sending
sinr = infra.sinr(eso2.node, esi.node, senders=frozenset())
capacity = 1 * math.log(1 + 10 ** (sinr / 10), 2)
assert edge_feature("datarate_fraction", eso2, esi, 2)[0] == approx(
8 / capacity
)
# this is an edge within a node, nothing is actually sent
assert edge_feature("datarate_fraction", ein, esi, 2)[0] == 0
# since this is the only chosen edge from that block, no broadcast
assert edge_feature("is_broadcast", eso1, erelay, 0)[0] == 0
# since eso1 -> erelay1 is already chosen, broadcast
assert edge_feature("is_broadcast", eso1, ENode(bin2, nsi), 0)[0] == 1
print(embedding.why_infeasible(eso2, ENode(bin3, nso2), 0))
# take a loop in nso2
assert embedding.take_action(eso2, ENode(bin3, nso2), 2)
# this doesn't count as broadcast, looping is not actually sending
assert edge_feature("is_broadcast", eso2, esi, 2)[0] == 0
# make sure broadcasting works even with relays
# on the same node as the already taken relay that sends bso1s data
# to ein1 in ts 1
erelay2 = ENode(bso1, ninterm, bin2)
assert edge_feature("is_broadcast", erelay2, ENode(bin2, nso2), 1)[0] == 1