Reference

API reference for the functions exported by agenet.

aaoi_fn(receiving_times, generation_times)

Calculate the average age of information.

Parameters:

Name	Type	Description	Default
receiving_times	NDArray	List of receiving times.	required
generation_times	NDArray	List of generation times.	required

Returns:

Type	Description
tuple[float, NDArray, NDArray]	Average age of information, age, times.

Source code in agenet/aaoi.py

def aaoi_fn(
    receiving_times: NDArray, generation_times: NDArray
) -> tuple[float, NDArray, NDArray]:
    """Calculate the average age of information.

    Args:
      receiving_times: List of receiving times.
      generation_times: List of generation times.

    Returns:
      Average age of information, age, times.
    """
    # Generate times for the time axis
    times: NDArray = np.arange(0, receiving_times[0] + 0.0005, 0.0005)
    num_events = len(receiving_times)

    for i in range(1, num_events):
        dummy = np.arange(receiving_times[i - 1], receiving_times[i] + 0.0001, 0.0001)
        times = np.concatenate((times, dummy))

    # Compute the age for every time instant
    j = 0
    offset = 0
    age = times.copy()

    for i in range(len(times)):
        if times[i] == receiving_times[j]:
            offset = generation_times[j]
            j += 1
        age[i] -= offset

    # Calculate the integral of age over time
    area = trapezoid(age, times)

    # Calculate the average Age of Information
    aaoi = area / times[-1]

    return aaoi, age, times

block_error(snr, n, k)

Calculate the Block Error Rate for the given instantaneous SNR, n, k.

Parameters:

Name	Type	Description	Default
snr	float	Instantaneous signal-to-noise ratio.	required
n	int	Total number of bits.	required
k	int	Number of information bits.	required

Returns:

Type	Description
float	The Block Error Rate.

Source code in agenet/blkerr.py

def block_error(snr: float, n: int, k: int) -> float:
    """Calculate the Block Error Rate for the given instantaneous SNR, n, k.

    Args:
      snr: Instantaneous signal-to-noise ratio.
      n: Total number of bits.
      k: Number of information bits.

    Returns:
      The Block Error Rate.
    """
    c = math.log2(1 + snr)
    v = 0.5 * ((1 - (1 / ((1 + snr) ** 2))) * ((math.log2(math.exp(1))) ** 2))

    # Handle potential division by zero
    if v == 0:
        return 1.0  # Assume worst-case scenario

    err = _qfunc(((n * c) - k) / math.sqrt(n * v))
    return err

block_error_th(snr_avg, n, k)

Calculate the theoretical Block Error Rate for the given average SNR, n, k.

Parameters:

Name	Type	Description	Default
snr_avg	float	Average Signal-to-noise ratio.	required
n	int	Total number of bits.	required
k	int	Number of information bits.	required

Returns:

Type	Description
float	The theoretical Block Error Rate.

Source code in agenet/blkerr.py

def block_error_th(snr_avg: float, n: int, k: int) -> float:
    """Calculate the theoretical Block Error Rate for the given average SNR, n, k.

    Args:
      snr_avg: Average Signal-to-noise ratio.
      n: Total number of bits.
      k: Number of information bits.

    Returns:
      The theoretical Block Error Rate.
    """
    try:
        beta = 1 / (2 * math.pi * math.sqrt((2 ** (2 * k / n)) - 1))
    except ArithmeticError:
        return 1.0  # Assume worst-case scenario if we can't calculate beta

    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_avg)
        * (
            math.exp(-1 * phi_bas * (1 / snr_avg))
            - math.exp(-1 * delta * (1 / snr_avg))
        )
    )
    return err_th

ev_sim(num_runs, frequency, num_events, num_bits, info_bits, power, distance, N0, num_bits_2=None, info_bits_2=None, power_2=None, distance_2=None, N0_2=None, seed=None)

Run the simulation num_runs times and return the AAoI expected value.

Parameters:

Name	Type	Description	Default
num_runs	int	Number of times to run the simulation.	required
frequency	float	Signal frequency in Hertz.	required
num_events	int	Number of events to simulate.	required
num_bits	int	Number of bits in a block.	required
info_bits	int	Number of bits in a message.	required
power	float	Transmission power in Watts.	required
distance	float	Distance between nodes.	required
N0	float	Noise power in Watts.	required
num_bits_2	int | None	Number of bits in a block at relay or access point.	None
info_bits_2	int | None	Number of bits in a message at relay or access point.	None
power_2	float | None	Transmission power in Watts at relay or access point.	None
distance_2	float | None	Distance between relay or access point and the destination.	None
N0_2	float | None	Noise power in Watts at relay or access point.	None
seed	int | signedinteger | None	Seed for the random number generator (optional).	None

Returns:

Type	Description
tuple[float, float, float, float, float, float]	A tuple containing the expected value for the theoretical AAoI and the simulation AAoI.

Source code in agenet/simulation.py

def ev_sim(
    num_runs: int,
    frequency: float,
    num_events: int,
    num_bits: int,
    info_bits: int,
    power: float,
    distance: float,
    N0: float,
    num_bits_2: int | None = None,
    info_bits_2: int | None = None,
    power_2: float | None = None,
    distance_2: float | None = None,
    N0_2: float | None = None,
    seed: int | np.signedinteger | None = None,
) -> tuple[float, float, float, float, float, float]:
    """Run the simulation `num_runs` times and return the AAoI expected value.

    Args:
      num_runs: Number of times to run the simulation.
      frequency: Signal frequency in Hertz.
      num_events: Number of events to simulate.
      num_bits: Number of bits in a block.
      info_bits: Number of bits in a message.
      power: Transmission power in Watts.
      distance: Distance between nodes.
      N0: Noise power in Watts.
      num_bits_2: Number of bits in a block at relay or access point.
      info_bits_2: Number of bits in a message at relay or access point.
      power_2: Transmission power in Watts at relay or access point.
      distance_2: Distance between relay or access point and the destination.
      N0_2: Noise power in Watts at relay or access point.
      seed: Seed for the random number generator (optional).

    Returns:
      A tuple containing the expected value for the theoretical AAoI and the
        simulation AAoI.
    """
    # Parse params and get an object of validated simulation parameters
    params = _param_validate(
        frequency=frequency,
        num_events=num_events,
        num_bits=num_bits,
        info_bits=info_bits,
        power=power,
        distance=distance,
        N0=N0,
        num_bits_2=num_bits_2,
        info_bits_2=info_bits_2,
        power_2=power_2,
        distance_2=distance_2,
        N0_2=N0_2,
        seed=seed,
    )

    ev_aaoi_th_run = 0.0
    ev_aaoi_sim_run = 0.0

    for _ in range(num_runs):

        # Run the simulation
        av_aaoi_th_i, av_aaoi_sim_i = _sim(
            frequency=params.frequency,
            num_events=params.num_events,
            num_bits_1=params.num_bits_1,
            info_bits_1=params.info_bits_1,
            power_1=params.power_1,
            distance_1=params.distance_1,
            N0_1=params.N0_1,
            blkerr1_th=params.blkerr1_th,
            num_bits_2=params.num_bits_2,
            info_bits_2=params.info_bits_2,
            power_2=params.power_2,
            distance_2=params.distance_2,
            N0_2=params.N0_2,
            blkerr2_th=params.blkerr2_th,
            rng=params.rng,
        )

        # Return infinity for both if theoretical is infinity
        if np.isinf(av_aaoi_th_i):
            return (
                float("inf"),
                float("inf"),
                params.snr1_avg,
                params.snr2_avg,
                params.blkerr1_th,
                params.blkerr2_th,
            )

        # Sum the AAoI's
        ev_aaoi_th_run += av_aaoi_th_i
        ev_aaoi_sim_run += av_aaoi_sim_i

    # Divide the AAoI's by the number of runs to get the expected value (mean)
    ev_aaoi_th_run /= num_runs
    ev_aaoi_sim_run /= num_runs

    # Return results
    return (
        ev_aaoi_th_run,
        ev_aaoi_sim_run,
        params.snr1_avg,
        params.snr2_avg,
        params.blkerr1_th,
        params.blkerr2_th,
    )

multi_param_ev_sim(num_runs, frequency, num_events, num_bits, info_bits, power, distance, N0, num_bits_2=[None], info_bits_2=[None], power_2=[None], distance_2=[None], N0_2=[None], seed=None, counter=None, stop_event=None)

Run the simulation for multiple parameters and return the results.

Parameters:

Name	Type	Description	Default
num_runs	int	Number of times to run the simulation.	required
frequency	Sequence[float]	List of frequencies.	required
num_events	Sequence[int]	List of number of events.	required
num_bits	Sequence[int]	List of number of bits in a block.	required
info_bits	Sequence[int]	List of number of bits in a message.	required
power	Sequence[float]	List of powers.	required
distance	Sequence[float]	List of distances between nodes.	required
N0	Sequence[float]	List of noise powers.	required
num_bits_2	Sequence[int | None]	List of number of bits in a block for relay or access point (optional, if different than source).	[None]
info_bits_2	Sequence[int | None]	List of number of bits in a message for relay or access point (optional, if different than source).	[None]
power_2	Sequence[float | None]	List of powers for relay or access point (optional, if different than source).	[None]
distance_2	Sequence[float | None]	List of distances between nodes for relay or access point (optional, if different than source).	[None]
N0_2	Sequence[float | None]	List of noise powers for relay or access point (optional, if different than source).	[None]
seed	int | signedinteger | None	Seed for the random number generator (optional).	None
counter	Synchronized[int] | None	An optional multiprocessing.Value which will be incremented after each inner loop pass. Only relevant if this function is executed in a separate thread.	None
stop_event	Event | None	The simulation will stop if this optional event is set externally. Only relevant if this function is executed in a separate thread.	None

Returns:

Type	Description
tuple[DataFrame, dict[str, Sequence[NamedTuple]]]	A tuple containing a DataFrame with the results of the simulation and a log highlighting invalid parameters or parameter combinations.

Source code in agenet/simulation.py

def multi_param_ev_sim(
    num_runs: int,
    frequency: Sequence[float],
    num_events: Sequence[int],
    num_bits: Sequence[int],
    info_bits: Sequence[int],
    power: Sequence[float],
    distance: Sequence[float],
    N0: Sequence[float],
    num_bits_2: Sequence[int | None] = [None],
    info_bits_2: Sequence[int | None] = [None],
    power_2: Sequence[float | None] = [None],
    distance_2: Sequence[float | None] = [None],
    N0_2: Sequence[float | None] = [None],
    seed: int | np.signedinteger | None = None,
    counter: Synchronized[int] | None = None,
    stop_event: Event | None = None,
) -> tuple[pd.DataFrame, dict[str, Sequence[NamedTuple]]]:
    """Run the simulation for multiple parameters and return the results.

    Args:
      num_runs: Number of times to run the simulation.
      frequency: List of frequencies.
      num_events: List of number of events.
      num_bits: List of number of bits in a block.
      info_bits: List of number of bits in a message.
      power: List of powers.
      distance: List of distances between nodes.
      N0: List of noise powers.
      num_bits_2: List of number of bits in a block for relay or access point
        (optional, if different than source).
      info_bits_2: List of number of bits in a message for relay or access point
        (optional, if different than source).
      power_2: List of powers for relay or access point
        (optional, if different than source).
      distance_2: List of distances between nodes for relay or access point
        (optional, if different than source).
      N0_2: List of noise powers for relay or access point (optional, if
        different than source).
      seed: Seed for the random number generator (optional).
      counter: An optional `multiprocessing.Value` which will be incremented
        after each inner loop pass. Only relevant if this function is executed
        in a separate thread.
      stop_event: The simulation will stop if this optional event is set
        externally. Only relevant if this function is executed in a separate
        thread.

    Returns:
      A tuple containing a DataFrame with the results of the simulation and a
        log highlighting invalid parameters or parameter combinations.
    """
    rng = Generator(Philox(seed))

    results = []

    param_error_log: dict[str, Sequence[NamedTuple]] = {}

    # Define the named tuple
    ParamCombo = namedtuple(
        "ParamCombo",
        [
            "frequency",
            "num_events",
            "num_bits",
            "info_bits",
            "power",
            "distance",
            "N0",
            "num_bits_2",
            "info_bits_2",
            "power_2",
            "distance_2",
            "N0_2",
        ],
    )

    # Get all combinations and create a parameter combo for each combination
    combos = [
        ParamCombo(f, e, n1, k1, p1, d1, n01, n2, k2, p2, d2, n02)
        for f, e, n1, k1, p1, d1, n01, n2, k2, p2, d2, n02 in itertools.product(
            frequency,
            num_events,
            num_bits,
            info_bits,
            power,
            distance,
            N0,
            num_bits_2,
            info_bits_2,
            power_2,
            distance_2,
            N0_2,
        )
    ]

    # Obtain PRNG seeds for each combo
    seeds = rng.integers(np.iinfo(np.int64).max, size=len(combos), dtype=np.int64)

    # Perform `num_runs` simulations for each parameter combo and get the
    # expected value of the AAoI for each combination
    for combo, seed in zip(combos, seeds):

        try:
            (
                aaoi_th,
                aaoi_sim,
                snr1_avg,
                snr2_avg,
                blkerr1_th,
                blkerr2_th,
            ) = ev_sim(
                num_runs=num_runs,
                frequency=combo.frequency,
                num_events=combo.num_events,
                num_bits=combo.num_bits,
                info_bits=combo.info_bits,
                power=combo.power,
                distance=combo.distance,
                N0=combo.N0,
                num_bits_2=combo.num_bits_2,
                info_bits_2=combo.info_bits_2,
                power_2=combo.power_2,
                distance_2=combo.distance_2,
                N0_2=combo.N0_2,
                seed=seed,
            )

            results.append(
                {
                    "frequency": combo.frequency,
                    "num_events": combo.num_events,
                    "num_bits": combo.num_bits,
                    "info_bits": combo.info_bits,
                    "power": combo.power,
                    "distance": combo.distance,
                    "N0": combo.N0,
                    "num_bits_2": (
                        combo.num_bits if combo.num_bits_2 is None else combo.num_bits_2
                    ),
                    "info_bits_2": (
                        combo.info_bits
                        if combo.info_bits_2 is None
                        else combo.info_bits_2
                    ),
                    "power_2": combo.power if combo.power_2 is None else combo.power_2,
                    "distance_2": (
                        combo.distance if combo.distance_2 is None else combo.distance_2
                    ),
                    "N0_2": combo.N0 if combo.N0_2 is None else combo.N0_2,
                    "aaoi_theory": aaoi_th,
                    "aaoi_sim": aaoi_sim,
                    "snr1_avg": snr1_avg,
                    "snr2_avg": snr2_avg,
                    "blkerr1_th": blkerr1_th,
                    "blkerr2_th": blkerr2_th,
                }
            )

        except _SimParamError as spe:
            # In case of invalid parameters or parameter combinations, log the
            # error and proceed to the next combination
            err_msg = str(spe)
            if err_msg not in param_error_log:
                param_error_log[err_msg] = []
            cast(MutableSequence, param_error_log[err_msg]).append(combo)

        if counter is not None:
            counter.value += 1
        if stop_event is not None and stop_event.is_set():
            break

    return pd.DataFrame(results), param_error_log

sim(frequency, num_events, num_bits, info_bits, power, distance, N0, num_bits_2=None, info_bits_2=None, power_2=None, distance_2=None, N0_2=None, seed=None)

Simulates a communication system and calculates the AAoI.

Parameters:

Name	Type	Description	Default
frequency	float	Signal frequency in Hertz.	required
num_events	int	Number of events to simulate.	required
num_bits	int	Number of bits in a block.	required
info_bits	int	Number of bits in a message.	required
power	float	Transmission power in Watts.	required
distance	float	Distance between nodes.	required
N0	float	Noise power in Watts.	required
num_bits_2	int | None	Number of bits in a block at relay or access point.	None
info_bits_2	int | None	Number of bits in a message at relay or access point.	None
power_2	float | None	Transmission power in Watts at relay or access point.	None
distance_2	float | None	Distance between relay or access point and the destination.	None
N0_2	float | None	Noise power in Watts at relay or access point.	None
seed	int | signedinteger | None	Seed for the random number generator (optional).	None

Returns:

Type	Description
tuple[float, float, float, float, float, float]	A tuple containing: theoretical AAoI, simulation AAoI, theoretical SNR at source node, theoretical SNR at relay or access point, theoretical block error at source node, theoretical SNR at relay or access point.

Source code in agenet/simulation.py

def sim(
    frequency: float,
    num_events: int,
    num_bits: int,
    info_bits: int,
    power: float,
    distance: float,
    N0: float,
    num_bits_2: int | None = None,
    info_bits_2: int | None = None,
    power_2: float | None = None,
    distance_2: float | None = None,
    N0_2: float | None = None,
    seed: int | np.signedinteger | None = None,
) -> tuple[float, float, float, float, float, float]:
    """Simulates a communication system and calculates the AAoI.

    Args:
      frequency: Signal frequency in Hertz.
      num_events: Number of events to simulate.
      num_bits: Number of bits in a block.
      info_bits: Number of bits in a message.
      power: Transmission power in Watts.
      distance: Distance between nodes.
      N0: Noise power in Watts.
      num_bits_2: Number of bits in a block at relay or access point.
      info_bits_2: Number of bits in a message at relay or access point.
      power_2: Transmission power in Watts at relay or access point.
      distance_2: Distance between relay or access point and the destination.
      N0_2: Noise power in Watts at relay or access point.
      seed: Seed for the random number generator (optional).

    Returns:
       A tuple containing: theoretical AAoI, simulation AAoI, theoretical SNR at
         source node, theoretical SNR at relay or access point, theoretical block
         error at source node, theoretical SNR at relay or access point.
    """
    # Parse params and get an object of validated simulation parameters
    params = _param_validate(
        frequency=frequency,
        num_events=num_events,
        num_bits=num_bits,
        info_bits=info_bits,
        power=power,
        distance=distance,
        N0=N0,
        num_bits_2=num_bits_2,
        info_bits_2=info_bits_2,
        power_2=power_2,
        distance_2=distance_2,
        N0_2=N0_2,
        seed=seed,
    )

    # Call the low-level function to actually perform the simulation
    return (
        *_sim(
            frequency=params.frequency,
            num_events=params.num_events,
            num_bits_1=params.num_bits_1,
            info_bits_1=params.info_bits_1,
            power_1=params.power_1,
            distance_1=params.distance_1,
            N0_1=params.N0_1,
            blkerr1_th=params.blkerr1_th,
            num_bits_2=params.num_bits_2,
            info_bits_2=params.info_bits_2,
            power_2=params.power_2,
            distance_2=params.distance_2,
            N0_2=params.N0_2,
            blkerr2_th=params.blkerr2_th,
            rng=params.rng,
        ),
        params.snr1_avg,
        params.snr2_avg,
        params.blkerr1_th,
        params.blkerr2_th,
    )

snr(N0, d, P, fr, seed=None)

Computes the instantaneous SNR of the received signal.

Parameters:

Name	Type	Description	Default
N0	float	Noise power in Watts.	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
seed	int | Generator | None	Seed for the random number generator or a previously instantied random number generator (optional).	None

Returns:

Type	Description
float	The instantaneous SNR of the received signal in linear scale.

Source code in agenet/snratio.py

def snr(
    N0: float, d: float, P: float, fr: float, seed: int | Generator | None = None
) -> float:
    """Computes the instantaneous SNR of the received signal.

    Args:
      N0: Noise power in Watts.
      d: The distance between the transmitter and receiver.
      P: The power of the transmitted signal.
      fr: The frequency of the signal.
      seed: Seed for the random number generator or a previously instantied
        random number generator (optional).

    Returns:
      The instantaneous SNR of the received signal in linear scale.
    """
    if isinstance(seed, Generator):
        rng = cast(Generator, seed)
    else:
        rng = Generator(np.random.SFC64(seed))

    # Calculate large-scale gain using _alpha function
    alpha = _alpha(d, fr)

    # Complex channel coefficient (small-scale fading)
    chah = 1 / np.sqrt(2) * (rng.standard_normal() + 1j * rng.standard_normal())

    # Signal gain (combining large-scale and small-scale effects)
    s_gain = alpha * np.abs(chah**2)

    #  Instantaneous SNR
    snr: float = (P * s_gain) / N0
    return snr

snr_avg(N0, d, P, fr)

Computes the average SNR of the received signal.

Parameters:

Name	Type	Description	Default
N0	float	Noise power in Watts.	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 average SNR of the received signal in linear scale.

Source code in agenet/snratio.py

def snr_avg(N0: float, d: float, P: float, fr: float) -> float:
    """Computes the average SNR of the received signal.

    Args:
      N0: Noise power in Watts.
      d: The distance between the transmitter and receiver.
      P: The power of the transmitted signal.
      fr: The frequency of the signal.

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