Skip to content

Reference

API reference for the functions exported by agenet.

av_age_fn(destination_times, generation_times, lambha)

Calculate the AAoI given the parameters.

Parameters:

Name Type Description Default
destination_times list[float]

A sorted list of destination times.

 required
generation_times list[float]

A list of the generation times.

 required
lambha float

The arrival rate of information.

 required

Returns:

Type Description
tuple[float, ndarray, ndarray]

A tuple containing the AAoI, the array of ages for each time step, and the corresponding time step array.
Source code in agenet/av_age.py 
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
def av_age_fn(
    destination_times: list[float],
    generation_times: list[float],
    lambha: float,
) -> tuple[float, np.ndarray, np.ndarray]:
    """Calculate the AAoI given the parameters.

    Args:
      destination_times: A sorted list of destination times.
      generation_times: A list of the  generation times.
      lambha: The arrival rate of information.

    Returns:
      A tuple containing the AAoI, the array of ages for each time step, and
        the corresponding time step array.
    """
    # Define the time step (p) as a constant (lambha)
    p = lambha * 0.01
    # Initialize the times array with the first destination time plus the time
    # step
    times = np.arange(0, destination_times[0] + p, p)
    # Loop through the rest of the destination times
    for i in range(1, len(destination_times)):
        # Generate an array of times between two consecutive destination times
        dummy = np.arange(destination_times[i - 1], destination_times[i] + p, p)
        # Concatenate the times array with the dummy array
        times = np.concatenate((times, dummy))
    # Initialize a counter (ii) and an offset
    ii = 0
    offset = 0.0
    # Initialize the age array as the times array
    age = times.copy()
    # Loop through the times array
    for i in range(len(times)):
        # If the current time is equal to a destination time
        if times[i] == destination_times[ii]:
            # Update the offset with the corresponding generation time
            offset = generation_times[ii]
            # Increment the counter
            ii = ii + 1
        # Update the current value in the age array as the difference between
        # the time and the offset
        age[i] = age[i] - offset
    # Calculate the average age as the area under the curve of the age versus
    # time divided by the maximum time
    average_age = trapz(age, times) / np.amax(times)
    # Return the average age, the age array, and the times array
    return average_age, age, times

blercal(snr, n, k)

Calculate the Block Error Rate (BLER) for the given SNR, n, k.

Parameters:

Name Type Description Default
snr float

Signal-to-Noise Ratio (SNR).

 required
n int

Number of bits in the block.

 required
k int

Number of bits in the message.

 required

Returns:

Type Description
float

Block Error Rate (BLER) for the given SNR, n, k.
Source code in agenet/bler.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def blercal(snr: float, n: int, k: int) -> float:
    """Calculate the Block Error Rate (BLER) for the given SNR, n, k.

    Args:
      snr: Signal-to-Noise Ratio (SNR).
      n: Number of bits in the block.
      k: Number of bits in the message.

    Returns:
      Block Error Rate (BLER) for the given SNR, n, k.
    """
    if snr < 0:
        raise ValueError("SNR must be non-negative")
    if n <= 0:
        raise ValueError("n must be greater than 0")
    if k <= 0:
        raise ValueError("k must be greater than 0")
    if k > n:
        raise ValueError("k must be less than or equal to n")
    c = math.log2(1 + snr)
    v = 0.5 * (1 - (1 / (1 + snr) ** 2)) * ((math.log2(math.exp(1))) ** 2)
    err = _qfunc(((n * c) - k) / math.sqrt(n * v))
    return err

blercal_th(snr, n, k)

Calculate the theoretical Block Error Rate (BLER) for the given SNR, n, k.

Parameters:

Name Type Description Default
snr float

Signal-to-Noise Ratio (SNR).

 required
n int

Number of bits in the block.

 required
k int

Number of bits in the message.

 required

Returns:

Type Description
float

Theoretical Block Error Rate (BLER) for the given SNR, n, k.
Source code in agenet/bler.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def blercal_th(snr: float, n: int, k: int) -> float:
    """Calculate the theoretical Block Error Rate (BLER) for the given SNR, n, k.

    Args:
      snr: Signal-to-Noise Ratio (SNR).
      n: Number of bits in the block.
      k: Number of bits in the message.

    Returns:
      Theoretical Block Error Rate (BLER) for the given SNR, n, k.
    """
    beta = 1 / (2 * math.pi * math.sqrt((2 ** (2 * k / n)) - 1))
    sim_phi = (2 ** (k / n)) - 1
    phi_bas = sim_phi - (1 / (2 * beta * math.sqrt(n)))
    delta = sim_phi + (1 / (2 * beta * math.sqrt(n)))
    err_th = 1 - (
        (beta * math.sqrt(n) * snr)
        * (math.exp(-1 * phi_bas * (1 / snr)) - math.exp(-1 * delta * (1 / snr)))
    )
    return err_th

generate_table(num_nodes_const, active_prob_const, n_const, k_const, P_const, d_const, N0_const, fr_const, numevnts, numruns, num_nodes_vals, active_prob_vals, n_vals, k_vals, P_vals, csv_location=None)

Print the simulated and theoretical values for each variable.

Parameters:

Name Type Description Default
num_nodes_const int

Constant value for the number of nodes.

 required
active_prob_const float

Constant value for the active probability.

 required
n_const int

Constant value for the block length.

 required
k_const int

Constant value for the update size.

 required
P_const float

Constant value for the power.

 required
d_const int

Constant value for the distance between nodes.

 required
N0_const float

Constant value for the noise power.

 required
fr_const float

Constant value for the frequency of the signal.

 required
numevnts int

Number of events.

 required
numruns int

Number of runs.

 required
num_nodes_vals list[int]

Values for the number of nodes.

 required
active_prob_vals list[float]

Values for the active probability.

 required
n_vals list[int]

Values for the block length.

 required
k_vals list[int]

Values for the update size.

 required
P_vals list[float]

Values for the power.

 required
csv_location Optional[str]

Location to save csv file

 None
Source code in agenet/printplot.py 
 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
def generate_table(
    num_nodes_const: int,
    active_prob_const: float,
    n_const: int,
    k_const: int,
    P_const: float,
    d_const: int,
    N0_const: float,
    fr_const: float,
    numevnts: int,
    numruns: int,
    num_nodes_vals: list[int],
    active_prob_vals: list[float],
    n_vals: list[int],
    k_vals: list[int],
    P_vals: list[float],
    csv_location: Optional[str] = None,
) -> None:
    """Print the simulated and theoretical values for each variable.

    Args:
      num_nodes_const: Constant value for the number of nodes.
      active_prob_const: Constant value for the active probability.
      n_const: Constant value for the block length.
      k_const: Constant value for the update size.
      P_const: Constant value for the power.
      d_const: Constant value for the distance between nodes.
      N0_const: Constant value for the noise power.
      fr_const: Constant value for the frequency of the signal.
      numevnts: Number of events.
      numruns: Number of runs.
      num_nodes_vals: Values for the number of nodes.
      active_prob_vals: Values for the active probability.
      n_vals: Values for the block length.
      k_vals: Values for the update size.
      P_vals: Values for the power.
      csv_location: Location to save csv file
      """
    for i, var_name, var_vals in zip(
        range(5),
        [
            "number of nodes",
            "active probability",
            "block length",
            "update size",
            "Power",
        ],
        [
            num_nodes_vals,
            active_prob_vals,
            n_vals,
            k_vals,
            P_vals,
        ],
    ):
        const_vals: list[Any] = [
            num_nodes_const,
            active_prob_const,
            n_const,
            k_const,
            P_const,
            d_const,
            N0_const,
            fr_const,
            numevnts,
            numruns,
        ]

        headers = [var_name, "Theoretical", "Simulated"]

        table_rows = []
        for val in cast(List[Union[int, float]], var_vals):
            theoretical, simulated = run_simulation(
                *(const_vals[:i] + [val] + const_vals[i + 1 :])
            )
            table_rows.append([val, theoretical, simulated])

        if csv_location is not None:
            # Open the CSV file in append mode
            with open(csv_location, mode='a', newline='') as csvfile:
                writer = csv.writer(csvfile)
                # Write a separator and headers for each set of variable simulations
                # Update headers to include the variable name dynamically
                updated_headers = [var_name, 'Theoretical', ' Simulated']
                writer.writerow(updated_headers)
            for val in cast(List[Union[int, float]], var_vals):  
                    theoretical, simulated = run_simulation(*(const_vals[:i] + [val] + const_vals[i + 1 :]))
                    writer.writerow([val, theoretical, simulated])
        else:
            # If no CSV location is provided, or to additionally print the result:
            print(tabulate(table_rows, headers=headers, tablefmt="grid"))
            print("\n")

plot(args, plots_folder=None)

Plot the simulated and theoretical values for each variable and save the plots.

Parameters:

Name Type Description Default
args Namespace

Parsed command-line arguments.

 required
plots_folder str | None

Folder to save plots

 None
Source code in agenet/printplot.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def plot(args: argparse.Namespace, plots_folder: str | None = None) -> None:
    """Plot the simulated and theoretical values for each variable and save the plots.

    Args:
        args: Parsed command-line arguments.
        plots_folder: Folder to save plots
    """
    # Extracting values from the args
    num_nodes_const = args.num_nodes_const
    active_prob_const = args.active_prob_const
    n_const = args.n_const
    k_const = args.k_const
    P_const = args.P_const
    d_const = args.d_const
    N0_const = args.N0_const
    fr_const = args.fr_const

    num_nodes_vals = args.num_nodes_vals
    active_prob_vals = args.active_prob_vals
    n_vals = args.n_vals
    k_vals = args.k_vals
    P_vals = args.P_vals

    # Call plot_generate to create and save plots
    plot_generate(
        num_nodes_const=num_nodes_const,
        active_prob_const=active_prob_const,
        n_const=n_const,
        k_const=k_const,
        P_const=P_const,
        d_const=d_const,
        N0_const=N0_const,
        fr_const=fr_const,
        numevnts=args.numevnts,
        numruns=args.numruns,
        num_nodes_vals=num_nodes_vals,
        active_prob_vals=active_prob_vals,
        n_vals=n_vals,
        k_vals=k_vals,
        P_vals=P_vals,
        plots_folder=plots_folder,
    )

plot_generate(num_nodes_const, active_prob_const, n_const, k_const, P_const, d_const, N0_const, fr_const, numevnts, numruns, num_nodes_vals, active_prob_vals, n_vals, k_vals, P_vals, plots_folder=None)

Plot the simulated and theoretical values for each variable.

Parameters:

Name Type Description Default
num_nodes_const int

Constant value for the number of nodes.

 required
active_prob_const float

Constant value for the active probability.

 required
n_const int

Constant value for the block length.

 required
k_const int

Constant value for the update size.

 required
P_const float

Constant value for the power.

 required
d_const int

Constant value for the distance between nodes.

 required
N0_const float

Constant value for the noise power.

 required
fr_const float

Constant value for the frequency of the signal.

 required
numevnts int

Number of events.

 required
numruns int

Number of runs.

 required
num_nodes_vals list[int]

Values for the number of nodes.

 required
active_prob_vals list[float]

Values for the active probability.

 required
n_vals list[int]

Values for the block length.

 required
k_vals list[int]

Values for the update size.

 required
P_vals list[float]

Values for the power.

 required
plots_folder str | None

Folder to save the plots.

 None
Source code in agenet/printplot.py 
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
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def plot_generate(
    num_nodes_const: int,
    active_prob_const: float,
    n_const: int,
    k_const: int,
    P_const: float,
    d_const: int,
    N0_const: float,
    fr_const: float,
    numevnts: int,
    numruns: int,
    num_nodes_vals: list[int],
    active_prob_vals: list[float],
    n_vals: list[int],
    k_vals: list[int],
    P_vals: list[float],
    plots_folder: str | None = None,
) -> None:
    """Plot the simulated and theoretical values for each variable.

    Args:
      num_nodes_const: Constant value for the number of nodes.
      active_prob_const: Constant value for the active probability.
      n_const: Constant value for the block length.
      k_const: Constant value for the update size.
      P_const: Constant value for the power.
      d_const: Constant value for the distance between nodes.
      N0_const: Constant value for the noise power.
      fr_const: Constant value for the frequency of the signal.
      numevnts: Number of events.
      numruns: Number of runs.
      num_nodes_vals: Values for the number of nodes.
      active_prob_vals: Values for the active probability.
      n_vals: Values for the block length.
      k_vals: Values for the update size.
      P_vals: Values for the power.
      plots_folder: Folder to save the plots.
    """
    for i, var_name, var_vals in zip(
        range(5),
        [
            "number of nodes",
            "active probability",
            "block length",
            "update size",
            "Power",
        ],
        [num_nodes_vals, active_prob_vals, n_vals, k_vals, P_vals],
    ):
        const_vals: list[Any] = [
            num_nodes_const,
            active_prob_const,
            n_const,
            k_const,
            P_const,
            d_const,
            N0_const,
            fr_const,
            numevnts,
            numruns,
        ]

        theoretical_vals = []
        simulated_vals = []

        # Gather data for each value of the variable
        for val in cast(List[Union[int, float]], var_vals):
            theoretical, simulated = run_simulation(
                *(const_vals[:i] + [val] + const_vals[i + 1 :])
            )
            theoretical_vals.append(theoretical)
            simulated_vals.append(simulated)

        # Create a new plot for each variable
        fig, ax = plt.subplots()
        ax.plot(var_vals, theoretical_vals, label="Theoretical", marker="o")
        ax.plot(var_vals, simulated_vals, label="Simulated", marker="x")
        ax.set_xlabel(var_name)
        ax.set_ylabel("Values")
        ax.set_title(f"Theoretical vs Simulated Values for Varying {var_name}")
        ax.legend()
        ax.grid(True)

        # Save each plot with a unique filename
        if plots_folder:
            if not os.path.exists(plots_folder):
                os.makedirs(plots_folder)
            fig.savefig(os.path.join(plots_folder, f"{var_name}_plot.png"))
            plt.close(fig)  # Close the plot after saving
        else:
            plt.show()

run_simulation(num_nodes, active_prob, n, k, P, d, N0, fr, numevnts, numruns)

Run the simulation numruns times and return the AAoI.

Parameters:

Name Type Description Default
num_nodes int

Number of nodes in the network.

 required
active_prob float

Probability that a node is active in a given time slot.

 required
n int

Number of bits in a block.

 required
k int

Number of bits in a message.

 required
P float

Power of the nodes.

 required
numevnts int

Number of events.

 required
numruns int

Number of times to run the simulation.

 required

Returns:

Type Description
tuple[float, float]

A tuple containing the theoretical AAoI and the simulation AAoI.
Source code in agenet/maincom.py 
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
def run_simulation(
    num_nodes: int,
    active_prob: float,
    n: int,
    k: int,
    P: float,
    d: int,
    N0: float,
    fr: float,
    numevnts: int,
    numruns: int,
) -> tuple[float, float]:
    """Run the simulation `numruns` times and return the AAoI.

    Args:
      num_nodes: Number of nodes in the network.
      active_prob: Probability that a node is active in a given time slot.
      n: Number of bits in a block.
      k: Number of bits in a message.
      P: Power of the nodes.
      numevnts: Number of events.
      numruns: Number of times to run the simulation.

    Returns:
      A tuple containing the theoretical AAoI and the simulation AAoI.
    """
    num_runs = numruns
    av_age_theoretical_run = 0.0
    av_age_simulation_run = 0.0
    for _ in range(num_runs):
        av_age_theoretical_i, av_age_simulation_i = simulation(
            num_nodes, active_prob, n, k, P, d, N0, fr, numevnts
        )
        av_age_theoretical_run += av_age_theoretical_i
        av_age_simulation_run += av_age_simulation_i
    av_age_theoretical_run /= num_runs
    av_age_simulation_run /= num_runs
    return av_age_theoretical_run, av_age_simulation_run

simulation(num_nodes, active_prob, n, k, P, d, N0, fr, numevents)

Simulates a communication system and calculates the AAoI.

Parameters:

Name Type Description Default
num_nodes int

Number of nodes in the system

 required
active_prob float

Probability that a node is active.

 required
n int

Number of bits in a block.

 required
k int

Number of bits in a message.

 required
P float

Power of the nodes.

 required
d int

Distance between nodes.

 required
N0 float

Noise power.

 required
fr float

Frequency of the signal.

 required
numevents int

Number of events to simulate.

 required

Returns:

Type Description
tuple[float, float]

Theoretical AAoI and simulation AAoI.
Source code in agenet/maincom.py 
 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
def simulation(
    num_nodes: int,
    active_prob: float,
    n: int,
    k: int,
    P: float,
    d: int,
    N0: float,
    fr: float,
    numevents: int,
) -> tuple[float, float]:
    """Simulates a communication system and calculates the AAoI.

    Args:
      num_nodes: Number of nodes in the system
      active_prob: Probability that a node is active.
      n: Number of bits in a block.
      k: Number of bits in a message.
      P: Power of the nodes.
      d: Distance between nodes.
      N0: Noise power.
      fr: Frequency of the signal.
      numevents: Number of events to simulate.

    Returns:
       Theoretical AAoI and simulation AAoI.
    """
    lambda1 = 1  # arrival for one transmission period
    num_events = numevents  # number of events
    inter_arrival_times = (1 / lambda1) * (np.ones(num_events))  # inter arrival times
    arrival_timestamps = np.cumsum(inter_arrival_times)  # arrival timestamps
    d1 = d  # disatance between source nodes and relay
    d2 = d  # distance between the relay and destination
    P1 = P  # power of the source nodes
    P2 = P  # power of the relay or access point
    n1 = n  # number of bits in the block for the source nodes
    n2 = n  # number of bits in the block for the relay or access point
    k1 = k  # number of bits in the message for the source nodes
    k2 = k  # number of bits in the message for the relay or access point
    snr1_th = snr_th(
        N0, d1, P1, fr
    )  # block error rate for the relay or access point at the destination
    snr2_th = snr_th(
        N0, d2, P2, fr
    )  # block error rate for the source nodes at the relay or access point
    er1_th = blercal_th(snr1_th, n1, k1)
    er2_th = blercal_th(snr2_th, n2, k2)
    inter_service_times = (1 / lambda1) * np.ones((num_events))  # inter service times
    server_timestamps_1 = np.zeros(
        num_events
    )  # Generating departure timestamps for the node 1
    departure_timestamps_s = np.zeros(num_events)
    su_p_th = active_prob * (1 - er1_th) * ((1 - active_prob) ** (num_nodes - 1))
    er_f_th = 1 - su_p_th
    er_p_th = er_f_th + (er2_th * (er_f_th - 1))
    for i in range(0, num_events):
        snr1 = snr(
            N0, d1, P1, fr
        )  # snr for the source nodes at the relay or access point
        snr2 = snr(N0, d2, P2, fr)
        er1 = blercal(
            snr1, n1, k1
        )  # block error rate for the source nodes at the relay or access point
        er2 = blercal(
            snr2, n2, k2
        )  # block error rate for the relay or access point at the destination
        su_p = active_prob * (1 - er1) * ((1 - active_prob) ** (num_nodes - 1))
        er_f = 1 - su_p
        er_p = er_f + (er2 * (er_f - 1))
        er_indi = int(random.random() > er_p)
        if er_indi == 0:
            departure_timestamps_s[i] = 0
            server_timestamps_1[i] = 0
        else:
            departure_timestamps_s[i] = arrival_timestamps[i] + inter_service_times[i]
            server_timestamps_1[i] = arrival_timestamps[i]

    dep = [x for x in departure_timestamps_s if x != 0]
    sermat = [x for x in server_timestamps_1 if x != 0]
    depcop = dep.copy()
    if server_timestamps_1[-1] == 0:
        if len(depcop) != 0:
            depcop.pop()
            maxt = max(arrival_timestamps[-1], dep[-1])
        else:
            maxt = arrival_timestamps[-1]
        v1 = depcop + [maxt]
    else:
        v1 = dep

    if departure_timestamps_s[0] == 0:
        if len(sermat) != 0:
            t1 = sermat
        else:
            t1 = [0]
    else:
        t1 = [0] + sermat

    system_time = 1 / lambda1  # system time (time which update in the system)
    av_age_simulation, _, _ = av_age_fn(v1, t1, system_time)
    av_age_theoretical = (1 / lambda1) * (0.5 + (1 / (1 - er_p_th)))

    return av_age_theoretical, av_age_simulation

snr(N0, d, P, fr)

Computes the SNR of the received signal.

Parameters:

Name Type Description Default
N0 float

The power spectral density of the noise.

 required
d float

The distance between the transmitter and receiver.

 required
P float

The power of the transmitted signal.

 required
fr float

The frequency of the signal.

 required

Returns:

Type Description
float

The SNR of the received signal in linear scale.
Source code in agenet/snratio.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def snr(N0: float, d: float, P: float, fr: float) -> float:
    """Computes the SNR of the received signal.

    Args:
      N0: The power spectral density of the noise.
      d: The distance between the transmitter and receiver.
      P: The power of the transmitted signal.
      fr: The frequency of the signal.

    Returns:
      The SNR of the received signal in linear scale.
    """
    alpha = _alpha(d, fr)
    snr: float = (alpha * P * np.random.exponential(1)) / N0
    return snr

snr_th(N0, d, P, fr)

Calculates the theoretical SNR of the received signal.

Parameters:

Name Type Description Default
N0 float

The power spectral density of the noise.

 required
d float

The distance between the transmitter and receiver.

 required
P float

The power of the transmitted signal.

 required
fr float

The frequency of the signal.

 required

Returns:

Type Description
float

The theoretical SNR of the received signal in linear scale.
Source code in agenet/snratio.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def snr_th(N0: float, d: float, P: float, fr: float) -> float:
    """Calculates the theoretical SNR of the received signal.

    Args:
      N0: The power spectral density of the noise.
      d: The distance between the transmitter and receiver.
      P: The power of the transmitted signal.
      fr: The frequency of the signal.

    Returns:
      The theoretical SNR of the received signal in linear scale.
    """
    alpha = _alpha(d, fr)
    snr_th: float = (alpha * P) / N0
    return snr_th