-
Notifications
You must be signed in to change notification settings - Fork 4
/
activity_simulator.py
152 lines (101 loc) · 3.82 KB
/
activity_simulator.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
import time
import numpy as np
class ActivitySimulator:
r"""Simulates clients_dict activity
The activity of each client follows a Markovian model with two states, 'active' and 'inactive'.
Attributes
----------
clients_ids: 1-D numpy.array (dtype=int)
n_clients: int
availabilities: 1-D numpy.array (dtype=float)
stabilities: 1-D numpy.array (dtype=float)
activation_probabilities: 2-D array of size (`n_clients`, 2)
activation probability at every state per client,
state: 1-D numpy.array(int)
value `1` corresponds to state active and `0` corresponds to inactive
__rng: numpy.random._generator.Generator
Methods
-------
_init_state
_build_activation_probabilities
step
get_active_clients
"""
def __init__(self, clients_ids, availability_types, availabilities, stability_types, stabilities, rng=None):
"""
Parameters
----------
clients_ids: 1-D numpy.array (dtype=int)
availability_types: : 1-D list (dtype=str)
availabilities: 1-D numpy.array (dtype=float)
stability_types: : 1-D list (dtype=str)
stabilities: 1-D numpy.array (dtype=float)
rng: numpy.random._generator.Generator
"""
self.clients_ids = clients_ids
self.n_clients = len(self.clients_ids)
self.availability_types = availability_types
self.availabilities = availabilities
self.stability_types = stability_types
self.stabilities = stabilities
self.__rng = (np.random.default_rng(int(time.time())) if (rng is None) else rng)
self.activation_probabilities = self._build_activation_probabilities()
self.state = self._init_state()
def _init_state(self):
state = self.__rng.binomial(n=1, p=self.availabilities, size=self.n_clients)
return state
def _build_activation_probabilities(self):
activation_probabilities = np.zeros((self.n_clients, 2), dtype=np.float32)
activation_probabilities[:, 1] = self.stabilities + (1 - self.stabilities) * self.availabilities
activation_probabilities[:, 0] = (1 - self.stabilities) * self.availabilities
return activation_probabilities
def step(self):
p = np.take_along_axis(self.activation_probabilities, self.state.reshape(-1, 1), axis=-1).reshape(-1)
self.state = \
self.__rng.binomial(
n=1,
p=p,
size=self.n_clients
)
def get_active_clients(self):
"""returns indices of active clients_dict (i.e., having state=1)
Returns
-------
* List[int]
"""
return self.clients_ids[self.state == 1].tolist()
class FileActivitySimulator(ActivitySimulator):
r"""Reads the activity of clients from a .json file
"""
def __init__(
self,
clients_ids,
availability_types,
availabilities,
stability_types,
stabilities,
active_ids_per_time,
rng=None
):
super(FileActivitySimulator, self).__init__(
clients_ids=clients_ids,
availability_types=availability_types,
availabilities=availabilities,
stability_types=stability_types,
stabilities=stabilities,
rng=rng
)
self.active_ids_per_time = active_ids_per_time
self.time_step = -1
def step(self):
self.time_step += 1
def get_active_clients(self):
"""returns indices of active clients_dict (i.e., having state=1)
Returns
-------
* List[int]
"""
try:
return self.active_ids_per_time[self.time_step]
except IndexError:
print("You are outside the simulated range for activity!!")