oxbrbtx/PrimAITE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merged PR 447: Bandwidth load / capacity checks before transmission

Browse Source 
## Summary

This pull request introduces significant enhancements to the AirSpace class within our network simulation software, aimed at improving the realism, configurability, and accuracy of wireless network simulations. These changes include the addition of new enums and attributes, enhancements to the configuration schema, and improvements in bandwidth management and transmission logic.

**Additions**

-   **Enums and Attributes:**

-   **AirSpaceEnvironmentType Enum**: Defines various environmental settings that affect wireless signal propagation and interference.
-   **ChannelWidth Enum**: Specifies available channel width options for wireless interfaces.
-   **Channel Width Attribute**: Added to WirelessNetworkInterface for dynamic adjustments based on the operational environment.
-   **airspace_key Attribute**: A tuple identifying the frequency and channel width combination for bandwidth management.
-   **airspace_environment_type Attribute**: Sets the overall environmental context of the airspace, influencing all contained devices.

-   **Functional Enhancements:**

-   **SNR and Capacity Calculation Functions**: New functions estimate_snr and calculate_total_channel_capacity have been implemented to compute signal-to-noise ratios and channel capacities dynamically.
-   **show_bandwidth_load Function**: Provides a visual representation of the current bandwidth load across different channels.
-   **Dynamic Speed Setting**: The speed attribute of WirelessInterface is now adjusted dynamically based on frequency, channel width, and environment.

-   **Configuration and Testing:**

-   **Configuration Schema Update**: The simulation.network config file schema now supports setting the airspace_environment_type.

**Changes**

-   **Interface and Performance Adjustments:**

-   **NetworkInterface Speed Type**: Changed from int to float for more precise speed definitions.
-   **Transmission Feasibility Check**: Updated the _can_transmit function in Link to better handle current load and bandwidth capacities.
-   **WirelessRouter Configurations**: The configure_wireless_access_point function now takes channel_width as an additional parameter.
-   **Grouping Adjustments**: WirelessNetworkInterfaces are now categorized by both AirSpaceFrequency and ChannelWidth.

-   **Transmission Logic Overhaul:**

-   **Interface Adjustments**: Modifying an interface's settings now necessitates its temporary removal from the airspace, followed by a recalculation of its data rate and reintegration under new settings.
-   **Blocking Overloads**: Strengthened the logic in AirSpace to prevent transmissions that would surpass the available capacity.

**Fixes**

-   **Transmission Permission Logic**: Fixed the can_transmit_frame function to accurately enforce transmission limits based on current network load and available bandwidth.

**Conclusion**

These updates significantly enhance the fidelity and flexibility of our network simulation tool, enabling more accurate m...
This commit is contained in:
Christopher McCarthy
2024-07-09 10:47:56 +00:00
parent 1fdd05a0d1 4410e05e3e
commit b45f6bbd40
20 changed files with 1297 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/primaite/game/game.py
View File

@@ -15,6 +15,7 @@ from primaite.game.agent.scripted_agents.probabilistic_agent import Probabilisti
from primaite.game.agent.scripted_agents.random_agent import PeriodicAgent
from primaite.game.agent.scripted_agents.tap001 import TAP001
from primaite.game.science import graph_has_cycle, topological_sort
from primaite.simulator.network.airspace import AirspaceEnvironmentType
from primaite.simulator.network.hardware.base import NodeOperatingState
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.host_node import NIC
@@ -227,6 +228,11 @@ class PrimaiteGame:
simulation_config = cfg.get("simulation", {})
network_config = simulation_config.get("network", {})
airspace_cfg = network_config.get("airspace", {})
airspace_environment_type_str = airspace_cfg.get("airspace_environment_type", "urban")
airspace_environment_type: AirspaceEnvironmentType = AirspaceEnvironmentType(airspace_environment_type_str)
net.airspace.airspace_environment_type = airspace_environment_type
nodes_cfg = network_config.get("nodes", [])
links_cfg = network_config.get("links", [])

2
src/primaite/simulator/file_system/file_type.py
View File

@@ -185,5 +185,5 @@ file_type_sizes_bytes = {
FileType.ZIP: 1024000,
FileType.TAR: 1024000,
FileType.GZ: 819200,
FileType.DB: 15360000,
FileType.DB: 5_000_000,
}

732
src/primaite/simulator/network/airspace.py
View File

@@ -3,9 +3,11 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from enum import Enum
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Tuple
from prettytable import PrettyTable
import numpy as np
from prettytable import MARKDOWN, PrettyTable
from pydantic import BaseModel, computed_field, Field, model_validator
from primaite import getLogger
from primaite.simulator.network.hardware.base import Layer3Interface, NetworkInterface, WiredNetworkInterface
@@ -15,90 +17,29 @@ from primaite.simulator.system.core.packet_capture import PacketCapture
_LOGGER = getLogger(__name__)
__all__ = ["AirSpaceFrequency", "WirelessNetworkInterface", "IPWirelessNetworkInterface"]
def format_hertz(hertz: float, format_terahertz: bool = False, decimals: int = 3) -> str:
"""
Convert a frequency in Hertz to a formatted string using the most appropriate unit.
class AirSpace:
"""Represents a wireless airspace, managing wireless network interfaces and handling wireless transmission."""
Optionally includes formatting for Terahertz.
def __init__(self):
self._wireless_interfaces: Dict[str, WirelessNetworkInterface] = {}
self._wireless_interfaces_by_frequency: Dict[AirSpaceFrequency, List[WirelessNetworkInterface]] = {}
def show(self, frequency: Optional[AirSpaceFrequency] = None):
"""
Displays a summary of wireless interfaces in the airspace, optionally filtered by a specific frequency.
:param frequency: The frequency band to filter devices by. If None, devices for all frequencies are shown.
"""
table = PrettyTable()
table.field_names = ["Connected Node", "MAC Address", "IP Address", "Subnet Mask", "Frequency", "Status"]
# If a specific frequency is provided, filter by it; otherwise, use all frequencies.
frequencies_to_show = [frequency] if frequency else self._wireless_interfaces_by_frequency.keys()
for freq in frequencies_to_show:
interfaces = self._wireless_interfaces_by_frequency.get(freq, [])
for interface in interfaces:
status = "Enabled" if interface.enabled else "Disabled"
table.add_row(
[
interface._connected_node.hostname, # noqa
interface.mac_address,
interface.ip_address if hasattr(interface, "ip_address") else None,
interface.subnet_mask if hasattr(interface, "subnet_mask") else None,
str(freq),
status,
]
)
print(table)
def add_wireless_interface(self, wireless_interface: WirelessNetworkInterface):
"""
Adds a wireless network interface to the airspace if it's not already present.
:param wireless_interface: The wireless network interface to be added.
"""
if wireless_interface.mac_address not in self._wireless_interfaces:
self._wireless_interfaces[wireless_interface.mac_address] = wireless_interface
if wireless_interface.frequency not in self._wireless_interfaces_by_frequency:
self._wireless_interfaces_by_frequency[wireless_interface.frequency] = []
self._wireless_interfaces_by_frequency[wireless_interface.frequency].append(wireless_interface)
def remove_wireless_interface(self, wireless_interface: WirelessNetworkInterface):
"""
Removes a wireless network interface from the airspace if it's present.
:param wireless_interface: The wireless network interface to be removed.
"""
if wireless_interface.mac_address in self._wireless_interfaces:
self._wireless_interfaces.pop(wireless_interface.mac_address)
self._wireless_interfaces_by_frequency[wireless_interface.frequency].remove(wireless_interface)
def clear(self):
"""
Clears all wireless network interfaces and their frequency associations from the airspace.
After calling this method, the airspace will contain no wireless network interfaces, and transmissions cannot
occur until new interfaces are added again.
"""
self._wireless_interfaces.clear()
self._wireless_interfaces_by_frequency.clear()
def transmit(self, frame: Frame, sender_network_interface: WirelessNetworkInterface):
"""
Transmits a frame to all enabled wireless network interfaces on a specific frequency within the airspace.
This ensures that a wireless interface does not receive its own transmission.
:param frame: The frame to be transmitted.
:param sender_network_interface: The wireless network interface sending the frame. This interface will be
excluded from the list of receivers to prevent it from receiving its own transmission.
"""
for wireless_interface in self._wireless_interfaces_by_frequency.get(sender_network_interface.frequency, []):
if wireless_interface != sender_network_interface and wireless_interface.enabled:
wireless_interface.receive_frame(frame)
:param hertz: Frequency in Hertz.
:param format_terahertz: Whether to format frequency in Terahertz, default is False.
:param decimals: Number of decimal places to round to, default is 3.
:returns: Formatted string with the frequency in the most suitable unit.
"""
format_str = f"{{:.{decimals}f}}"
if format_terahertz and hertz >= 1e12: # Terahertz
return format_str.format(hertz / 1e12) + " THz"
elif hertz >= 1e9: # Gigahertz
return format_str.format(hertz / 1e9) + " GHz"
elif hertz >= 1e6: # Megahertz
return format_str.format(hertz / 1e6) + " MHz"
elif hertz >= 1e3: # Kilohertz
return format_str.format(hertz / 1e3) + " kHz"
else: # Hertz
return format_str.format(hertz) + " Hz"
class AirSpaceFrequency(Enum):
@@ -110,12 +51,478 @@ class AirSpaceFrequency(Enum):
"""WiFi 5 GHz. Known for its higher data transmission speeds and reduced interference from other devices."""
def __str__(self) -> str:
hertz_str = format_hertz(hertz=self.value)
if self == AirSpaceFrequency.WIFI_2_4:
return "WiFi 2.4 GHz"
elif self == AirSpaceFrequency.WIFI_5:
return "WiFi 5 GHz"
else:
return "Unknown Frequency"
return f"WiFi {hertz_str}"
if self == AirSpaceFrequency.WIFI_5:
return f"WiFi {hertz_str}"
return "Unknown Frequency"
class ChannelWidth(Enum):
"""
Enumeration representing the available channel widths in MHz for wireless communications.
This enum facilitates standardising and validating channel width configurations.
Attributes:
WIDTH_20_MHZ (int): Represents a channel width of 20 MHz, commonly used for basic
Wi-Fi connectivity with standard range and interference resistance.
WIDTH_40_MHZ (int): Represents a channel width of 40 MHz, offering higher data
throughput at the expense of potentially increased interference.
WIDTH_80_MHZ (int): Represents a channel width of 80 MHz, typically used in modern
Wi-Fi setups for high data rate applications but with higher susceptibility to interference.
WIDTH_160_MHZ (int): Represents a channel width of 160 MHz, used for ultra-high-speed
network applications, providing maximum data throughput with significant
requirements on the spectral environment to minimize interference.
"""
WIDTH_20_MHZ = 20
"""
Represents a channel width of 20 MHz, commonly used for basic Wi-Fi connectivity with standard range and
interference resistance
"""
WIDTH_40_MHZ = 40
"""
Represents a channel width of 40 MHz, offering higher data throughput at the expense of potentially increased
interference.
"""
WIDTH_80_MHZ = 80
"""
Represents a channel width of 80 MHz, typically used in modern Wi-Fi setups for high data rate applications but
with higher susceptibility to interference.
"""
WIDTH_160_MHZ = 160
"""
Represents a channel width of 160 MHz, used for ultra-high-speed network applications, providing maximum data
throughput with significant requirements on the spectral environment to minimize interference.
"""
def __str__(self) -> str:
"""
Returns a string representation of the channel width.
:return: String in the format of "<value> MHz" indicating the channel width.
"""
return f"{self.value} MHz"
AirSpaceKeyType = Tuple[AirSpaceFrequency, ChannelWidth]
class AirspaceEnvironmentType(Enum):
"""Enum representing different types of airspace environments which affect wireless communication signals."""
RURAL = "rural"
"""
A rural environment offers clear channel conditions due to low population density and minimal electronic device
presence.
"""
OUTDOOR = "outdoor"
"""
Outdoor environments like parks or fields have minimal electronic interference.
"""
SUBURBAN = "suburban"
"""
Suburban environments strike a balance with fewer electronic interferences than urban but more than rural.
"""
OFFICE = "office"
"""
Office environments have moderate interference from numerous electronic devices and overlapping networks.
"""
URBAN = "urban"
"""
Urban environments are characterized by tall buildings and a high density of electronic devices, leading to
significant interference.
"""
INDUSTRIAL = "industrial"
"""
Industrial areas face high interference from heavy machinery and numerous electronic devices.
"""
TRANSPORT = "transport"
"""
Environments such as subways and buses where metal structures and high mobility create complex interference
patterns.
"""
DENSE_URBAN = "dense_urban"
"""
Dense urban areas like city centers have the highest level of signal interference due to the very high density of
buildings and devices.
"""
JAMMING_ZONE = "jamming_zone"
"""
A jamming zone environment where signals are actively interfered with, typically through the use of signal jammers
or scrambling devices. This represents the environment with the highest level of interference.
"""
BLOCKED = "blocked"
"""
A jamming zone environment with total levels of interference. Airspace is completely blocked.
"""
@property
def snr_impact(self) -> int:
"""
Returns the SNR impact associated with the environment.
:return: SNR impact in dB.
"""
impacts = {
AirspaceEnvironmentType.RURAL: 0,
AirspaceEnvironmentType.OUTDOOR: 1,
AirspaceEnvironmentType.SUBURBAN: -5,
AirspaceEnvironmentType.OFFICE: -7,
AirspaceEnvironmentType.URBAN: -10,
AirspaceEnvironmentType.INDUSTRIAL: -15,
AirspaceEnvironmentType.TRANSPORT: -12,
AirspaceEnvironmentType.DENSE_URBAN: -20,
AirspaceEnvironmentType.JAMMING_ZONE: -40,
AirspaceEnvironmentType.BLOCKED: -100,
}
return impacts[self]
def __str__(self) -> str:
return f"{self.value.title()} Environment (SNR Impact: {self.snr_impact})"
def estimate_snr(
frequency: AirSpaceFrequency, environment_type: AirspaceEnvironmentType, channel_width: ChannelWidth
) -> float:
"""
Estimate the Signal-to-Noise Ratio (SNR) based on the communication frequency, environment, and channel width.
This function considers both the base SNR value dependent on the frequency and the impact of environmental
factors and channel width on the SNR.
The SNR is adjusted by reducing it for wider channels, reflecting the increased noise floor from a broader
frequency range.
:param frequency: The operating frequency as defined by AirSpaceFrequency enum, influencing the base SNR. Higher
frequencies like 5 GHz generally start with a higher base SNR due to less noise.
:param environment_type: The type of environment from AirspaceEnvironmentType enum, which adjusts the SNR based on
expected environmental noise and interference levels.
:param channel_width: The channel width from ChannelWidth enum, where wider channels (80 MHz and 160 MHz) decrease
the SNR slightly due to an increased noise floor.
:return: Estimated SNR in dB, calculated as the base SNR modified by environmental and channel width impacts.
"""
base_snr = 40 if frequency == AirSpaceFrequency.WIFI_5 else 30
snr_impact = environment_type.snr_impact
# Adjust SNR impact based on channel width
if channel_width == ChannelWidth.WIDTH_80_MHZ or channel_width == ChannelWidth.WIDTH_160_MHZ:
snr_impact -= 3 # Assume wider channels have slightly lower SNR due to increased noise floor
return base_snr + snr_impact
def calculate_total_channel_capacity(
channel_width: ChannelWidth, frequency: AirSpaceFrequency, environment_type: AirspaceEnvironmentType
) -> float:
"""
Calculate the total theoretical data rate for the channel using the Shannon-Hartley theorem.
This function determines the channel's capacity by considering the bandwidth (derived from channel width),
and the signal-to-noise ratio (SNR) adjusted by frequency and environmental conditions.
The Shannon-Hartley theorem states that channel capacity C (in bits per second) can be calculated as:
``C = B * log2(1 + SNR)`` where B is the bandwidth in Hertz and SNR is the signal-to-noise ratio.
:param channel_width: The width of the channel as defined by ChannelWidth enum, converted to Hz for calculation.
:param frequency: The operating frequency as defined by AirSpaceFrequency enum, influencing the base SNR and part
of the SNR estimation.
:param environment_type: The type of environment as defined by AirspaceEnvironmentType enum, used in SNR estimation.
:return: Theoretical total data rate in Mbps for the entire channel.
"""
bandwidth_hz = channel_width.value * 1_000_000 # Convert MHz to Hz
snr_db = estimate_snr(frequency, environment_type, channel_width)
snr_linear = 10 ** (snr_db / 10)
total_capacity_bps = bandwidth_hz * np.log2(1 + snr_linear)
total_capacity_mbps = total_capacity_bps / 1_000_000
return total_capacity_mbps
def calculate_individual_device_rate(
channel_width: ChannelWidth,
frequency: AirSpaceFrequency,
environment_type: AirspaceEnvironmentType,
device_count: int,
) -> float:
"""
Calculate the theoretical data rate available to each individual device on the channel.
This function first calculates the total channel capacity and then divides this capacity by the number
of active devices to estimate each device's share of the bandwidth. This reflects the practical limitation
that multiple devices must share the same channel resources.
:param channel_width: The channel width as defined by ChannelWidth enum, used in total capacity calculation.
:param frequency: The operating frequency as defined by AirSpaceFrequency enum, used in total capacity calculation.
:param environment_type: The environment type as defined by AirspaceEnvironmentType enum, impacting SNR and
capacity.
:param device_count: The number of devices sharing the channel. If zero, returns zero to avoid division by zero.
:return: Theoretical data rate in Mbps available per device, based on shared channel capacity.
"""
total_capacity_mbps = calculate_total_channel_capacity(channel_width, frequency, environment_type)
if device_count == 0:
return 0 # Avoid division by zero
individual_device_rate_mbps = total_capacity_mbps / device_count
return individual_device_rate_mbps
class AirSpace(BaseModel):
"""
Represents a wireless airspace, managing wireless network interfaces and handling wireless transmission.
This class provides functionalities to manage a collection of wireless network interfaces, each associated with
specific frequencies and channel widths. It includes methods to calculate and manage bandwidth loads, add and
remove wireless interfaces, and handle data transmission across these interfaces.
"""
airspace_environment_type_: AirspaceEnvironmentType = AirspaceEnvironmentType.URBAN
wireless_interfaces: Dict[str, WirelessNetworkInterface] = Field(default_factory=lambda: {})
wireless_interfaces_by_frequency_channel_width: Dict[AirSpaceKeyType, List[WirelessNetworkInterface]] = Field(
default_factory=lambda: {}
)
bandwidth_load: Dict[AirSpaceKeyType, float] = Field(default_factory=lambda: {})
frequency_channel_width_max_capacity_mbps: Dict[AirSpaceKeyType, float] = Field(default_factory=lambda: {})
def model_post_init(self, __context: Any) -> None:
"""
Initialize the airspace metadata after instantiation.
This method is called to set up initial configurations like the maximum capacity of each channel width and
frequency based on the current environment setting.
:param __context: Contextual data or settings, typically used for further initializations beyond
the basic constructor.
"""
self._set_frequency_channel_width_max_capacity_mbps()
def _set_frequency_channel_width_max_capacity_mbps(self):
"""
Private method to compute and set the maximum channel capacity in Mbps for each frequency and channel width.
Based on the airspace environment type, this method calculates the maximum possible data transmission
capacity for each combination of frequency and channel width available and stores these values.
These capacities are critical for managing and limiting bandwidth load during operations.
"""
print(
f"Rebuilding the frequency channel width maximum capacity dictionary based on "
f"airspace environment type {self.airspace_environment_type_}"
)
for frequency in AirSpaceFrequency:
for channel_width in ChannelWidth:
max_capacity = calculate_total_channel_capacity(
frequency=frequency, channel_width=channel_width, environment_type=self.airspace_environment_type
)
self.frequency_channel_width_max_capacity_mbps[frequency, channel_width] = max_capacity
@computed_field
@property
def airspace_environment_type(self) -> AirspaceEnvironmentType:
"""
Gets the current environment type of the airspace.
:return: The AirspaceEnvironmentType representing the current environment type.
"""
return self.airspace_environment_type_
@airspace_environment_type.setter
def airspace_environment_type(self, value: AirspaceEnvironmentType) -> None:
"""
Sets a new environment type for the airspace and updates related configurations.
Changing the environment type triggers a re-calculation of the maximum channel capacities and
adjustments to the current setup of wireless interfaces to ensure they are aligned with the
new environment settings.
:param value: The new environment type as an AirspaceEnvironmentType.
"""
if value != self.airspace_environment_type_:
print(f"Setting airspace_environment_type to {value}")
self.airspace_environment_type_ = value
self._set_frequency_channel_width_max_capacity_mbps()
wireless_interface_keys = list(self.wireless_interfaces.keys())
for wireless_interface_key in wireless_interface_keys:
wireless_interface = self.wireless_interfaces[wireless_interface_key]
self.remove_wireless_interface(wireless_interface)
self.add_wireless_interface(wireless_interface)
def show_bandwidth_load(self, markdown: bool = False):
"""
Prints a table of the current bandwidth load for each frequency and channel width combination on the airspace.
This method prints a tabulated view showing the utilisation of available bandwidth capacities for all configured
frequency and channel width pairings. The table includes the current capacity usage as a percentage of the
maximum capacity, alongside the absolute maximum capacity values in Mbps.
:param markdown: Flag indicating if output should be in markdown format.
"""
headers = ["Frequency", "Channel Width", "Current Capacity (%)", "Maximum Capacity (Mbit)"]
table = PrettyTable(headers)
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = "Airspace Frequency Channel Loads"
for key, load in self.bandwidth_load.items():
frequency, channel_width = key
maximum_capacity = self.frequency_channel_width_max_capacity_mbps[key]
load_percent = load / maximum_capacity
if load_percent > 1.0:
load_percent = 1.0
table.add_row(
[format_hertz(frequency.value), str(channel_width), f"{load_percent:.0%}", f"{maximum_capacity:.3f}"]
)
print(table)
def show_wireless_interfaces(self, markdown: bool = False):
"""
Prints a table of wireless interfaces in the airspace.
:param markdown: Flag indicating if output should be in markdown format.
"""
headers = [
"Connected Node",
"MAC Address",
"IP Address",
"Subnet Mask",
"Frequency",
"Channel Width",
"Speed (Mbps)",
"Status",
]
table = PrettyTable(headers)
if markdown:
table.set_style(MARKDOWN)
table.align = "l"
table.title = f"Devices on Air Space - {self.airspace_environment_type}"
for interface in self.wireless_interfaces.values():
status = "Enabled" if interface.enabled else "Disabled"
table.add_row(
[
interface._connected_node.hostname, # noqa
interface.mac_address,
interface.ip_address if hasattr(interface, "ip_address") else None,
interface.subnet_mask if hasattr(interface, "subnet_mask") else None,
format_hertz(interface.frequency.value),
str(interface.channel_width),
f"{interface.speed:.3f}",
status,
]
)
print(table.get_string(sortby="Frequency"))
def show(self, markdown: bool = False):
"""
Prints a summary of the current state of the airspace, including both wireless interfaces and bandwidth loads.
This method is a convenient wrapper that calls two separate methods to display detailed tables: one for
wireless interfaces and another for bandwidth load across all frequencies and channel widths managed within the
airspace. It provides a holistic view of the operational status and performance metrics of the airspace.
:param markdown: Flag indicating if output should be in markdown format.
"""
self.show_wireless_interfaces(markdown)
self.show_bandwidth_load(markdown)
def add_wireless_interface(self, wireless_interface: WirelessNetworkInterface):
"""
Adds a wireless network interface to the airspace if it's not already present.
:param wireless_interface: The wireless network interface to be added.
"""
if wireless_interface.mac_address not in self.wireless_interfaces:
self.wireless_interfaces[wireless_interface.mac_address] = wireless_interface
if wireless_interface.airspace_key not in self.wireless_interfaces_by_frequency_channel_width:
self.wireless_interfaces_by_frequency_channel_width[wireless_interface.airspace_key] = []
self.wireless_interfaces_by_frequency_channel_width[wireless_interface.airspace_key].append(
wireless_interface
)
speed = calculate_total_channel_capacity(
wireless_interface.channel_width, wireless_interface.frequency, self.airspace_environment_type
)
wireless_interface.set_speed(speed)
def remove_wireless_interface(self, wireless_interface: WirelessNetworkInterface):
"""
Removes a wireless network interface from the airspace if it's present.
:param wireless_interface: The wireless network interface to be removed.
"""
if wireless_interface.mac_address in self.wireless_interfaces:
self.wireless_interfaces.pop(wireless_interface.mac_address)
self.wireless_interfaces_by_frequency_channel_width[wireless_interface.airspace_key].remove(
wireless_interface
)
def clear(self):
"""
Clears all wireless network interfaces and their frequency associations from the airspace.
After calling this method, the airspace will contain no wireless network interfaces, and transmissions cannot
occur until new interfaces are added again.
"""
self.wireless_interfaces.clear()
self.wireless_interfaces_by_frequency_channel_width.clear()
def reset_bandwidth_load(self):
"""
Resets the bandwidth load tracking for all frequencies in the airspace.
This method clears the current load metrics for all operating frequencies, effectively setting the load to zero.
"""
self.bandwidth_load = {}
def can_transmit_frame(self, frame: Frame, sender_network_interface: WirelessNetworkInterface) -> bool:
"""
Determines if a frame can be transmitted by the sender network interface based on the current bandwidth load.
This method checks if adding the size of the frame to the current bandwidth load of the frequency used by the
sender network interface would exceed the maximum allowed bandwidth for that frequency. It returns True if the
frame can be transmitted without exceeding the limit, and False otherwise.
:param frame: The frame to be transmitted, used to check its size against the frequency's bandwidth limit.
:param sender_network_interface: The network interface attempting to transmit the frame, used to determine the
relevant frequency and its current bandwidth load.
:return: True if the frame can be transmitted within the bandwidth limit, False if it would exceed the limit.
"""
if sender_network_interface.airspace_key not in self.bandwidth_load:
self.bandwidth_load[sender_network_interface.airspace_key] = 0.0
return (
self.bandwidth_load[sender_network_interface.airspace_key] + frame.size_Mbits
<= self.frequency_channel_width_max_capacity_mbps[sender_network_interface.airspace_key]
)
def transmit(self, frame: Frame, sender_network_interface: WirelessNetworkInterface):
"""
Transmits a frame to all enabled wireless network interfaces on a specific frequency within the airspace.
This ensures that a wireless interface does not receive its own transmission.
:param frame: The frame to be transmitted.
:param sender_network_interface: The wireless network interface sending the frame. This interface will be
excluded from the list of receivers to prevent it from receiving its own transmission.
"""
self.bandwidth_load[sender_network_interface.airspace_key] += frame.size_Mbits
for wireless_interface in self.wireless_interfaces_by_frequency_channel_width.get(
sender_network_interface.airspace_key, []
):
if wireless_interface != sender_network_interface and wireless_interface.enabled:
wireless_interface.receive_frame(frame)
class WirelessNetworkInterface(NetworkInterface, ABC):
@@ -139,7 +546,135 @@ class WirelessNetworkInterface(NetworkInterface, ABC):
"""
airspace: AirSpace
frequency: AirSpaceFrequency = AirSpaceFrequency.WIFI_2_4
frequency_: AirSpaceFrequency = AirSpaceFrequency.WIFI_2_4
channel_width_: ChannelWidth = ChannelWidth.WIDTH_40_MHZ
@model_validator(mode="after") # noqa
def validate_channel_width_for_2_4_ghz(self) -> "WirelessNetworkInterface":
"""
Validate the wireless interface's channel width settings after model changes.
This method serves as a model validator to ensure that the channel width settings for the 2.4 GHz frequency
comply with accepted standards (either 20 MHz or 40 MHz). It's triggered after model instantiation.
Ensures that the channel width is appropriate for the current frequency setting, particularly checking
and adjusting the settings for the 2.4 GHz frequency band to not exceed 40 MHz. This is crucial for
avoiding interference and ensuring optimal performance in densely populated wireless environments.
"""
self._check_wifi_24_channel_width()
return self
def model_post_init(self, __context: Any) -> None:
"""Initialise the model after its creation, setting the speed based on the calculated channel capacity."""
speed = calculate_total_channel_capacity(
channel_width=self.channel_width,
frequency=self.frequency,
environment_type=self.airspace.airspace_environment_type,
)
self.set_speed(speed)
def _check_wifi_24_channel_width(self) -> None:
"""
Ensures that the channel width for 2.4 GHz frequency does not exceed 40 MHz.
This method checks the current frequency and channel width settings and adjusts the channel width
to 40 MHz if the frequency is set to 2.4 GHz and the channel width exceeds 40 MHz. This is done to
comply with typical Wi-Fi standards for 2.4 GHz frequencies, which commonly support up to 40 MHz.
Logs a SysLog warning if the channel width had to be adjusted, logging this change either to the connected
node's system log or the global logger, depending on whether the interface is connected to a node.
"""
if self.frequency_ == AirSpaceFrequency.WIFI_2_4 and self.channel_width_.value > 40:
self.channel_width_ = ChannelWidth.WIDTH_40_MHZ
msg = (
f"Channel width must be either 20 Mhz or 40 Mhz when using {AirSpaceFrequency.WIFI_2_4}. "
f"Overriding value to use {ChannelWidth.WIDTH_40_MHZ}."
)
if self._connected_node:
self._connected_node.sys_log.warning(f"Wireless Interface {self.port_num}: {msg}")
else:
_LOGGER.warning(msg)
@computed_field
@property
def frequency(self) -> AirSpaceFrequency:
"""
Get the current operating frequency of the wireless interface.
:return: The current frequency as an AirSpaceFrequency enum value.
"""
return self.frequency_
@frequency.setter
def frequency(self, value: AirSpaceFrequency) -> None:
"""
Set the operating frequency of the wireless interface and update the network configuration.
This setter updates the frequency of the wireless interface if the new value differs from the current setting.
It handles the update by first removing the interface from the current airspace management to avoid conflicts,
setting the new frequency, ensuring the channel width remains compliant, and then re-adding the interface
to the airspace with the new settings.
:param value: The new frequency to set, as an AirSpaceFrequency enum value.
"""
if value != self.frequency_:
self.airspace.remove_wireless_interface(self)
self.frequency_ = value
self._check_wifi_24_channel_width()
self.airspace.add_wireless_interface(self)
@computed_field
@property
def channel_width(self) -> ChannelWidth:
"""
Get the current channel width setting of the wireless interface.
:return: The current channel width as a ChannelWidth enum value.
"""
return self.channel_width_
@channel_width.setter
def channel_width(self, value: ChannelWidth) -> None:
"""
Set the channel width of the wireless interface and manage configuration compliance.
Updates the channel width of the wireless interface. If the new channel width is different from the existing
one, it first removes the interface from the airspace to prevent configuration conflicts, sets the new channel
width, checks and adjusts it if necessary (especially for 2.4 GHz frequency to comply with typical standards),
and then re-registers the interface in the airspace with updated settings.
:param value: The new channel width to set, as a ChannelWidth enum value.
"""
if value != self.channel_width_:
self.airspace.remove_wireless_interface(self)
self.channel_width_ = value
self._check_wifi_24_channel_width()
self.airspace.add_wireless_interface(self)
@property
def airspace_key(self) -> tuple:
"""
The airspace bandwidth/channel identifier for the wireless interface based on its frequency and channel width.
:return: A tuple containing the frequency and channel width, serving as a bandwidth/channel key.
"""
return self.frequency_, self.channel_width_
def set_speed(self, speed: float):
"""
Sets the network interface speed to the specified value and logs this action.
This method updates the speed attribute of the network interface to the given value, reflecting
the theoretical maximum data rate that the interface can support based on the current settings.
It logs the new speed to the system log of the connected node if available.
:param speed: The speed in Mbps to be set for the network interface.
"""
self.speed = speed
if self._connected_node:
self._connected_node.sys_log.info(
f"Wireless Interface {self.port_num}: Setting theoretical maximum data rate to {speed:.3f} Mbps."
)
def enable(self):
"""Attempt to enable the network interface."""
@@ -185,13 +720,18 @@ class WirelessNetworkInterface(NetworkInterface, ABC):
:param frame: The network frame to be sent.
:return: True if the frame is sent successfully, False if the network interface is disabled.
"""
if self.enabled:
frame.set_sent_timestamp()
self.pcap.capture_outbound(frame)
self.airspace.transmit(frame, self)
return True
# Cannot send Frame as the network interface is not enabled
return False
if not self.enabled:
return False
if not self.airspace.can_transmit_frame(frame, self):
# Drop frame for now. Queuing will happen here (probably) if it's done in the future.
self._connected_node.sys_log.info(f"{self}: Frame dropped as Link is at capacity")
return False
super().send_frame(frame)
frame.set_sent_timestamp()
self.pcap.capture_outbound(frame)
self.airspace.transmit(frame, self)
return True
def receive_frame(self, frame: Frame) -> bool:
"""

2
src/primaite/simulator/network/container.py
View File

@@ -96,6 +96,8 @@ class Network(SimComponent):
"""Apply pre-timestep logic."""
super().pre_timestep(timestep)
self.airspace.reset_bandwidth_load()
for node in self.nodes.values():
node.pre_timestep(timestep)

41
src/primaite/simulator/network/hardware/base.py
View File

@@ -87,7 +87,7 @@ class NetworkInterface(SimComponent, ABC):
mac_address: str = Field(default_factory=generate_mac_address)
"The MAC address of the interface."
speed: int = 100
speed: float = 100.0
"The speed of the interface in Mbps. Default is 100 Mbps."
mtu: int = 1500
@@ -499,14 +499,17 @@ class WiredNetworkInterface(NetworkInterface, ABC):
:param frame: The network frame to be sent.
:return: True if the frame is sent, False if the Network Interface is disabled or not connected to a link.
"""
if not self.enabled:
return False
if not self._connected_link.can_transmit_frame(frame):
# Drop frame for now. Queuing will happen here (probably) if it's done in the future.
self._connected_node.sys_log.info(f"{self}: Frame dropped as Link is at capacity")
return False
super().send_frame(frame)
if self.enabled:
frame.set_sent_timestamp()
self.pcap.capture_outbound(frame)
self._connected_link.transmit_frame(sender_nic=self, frame=frame)
return True
# Cannot send Frame as the NIC is not enabled
return False
frame.set_sent_timestamp()
self.pcap.capture_outbound(frame)
self._connected_link.transmit_frame(sender_nic=self, frame=frame)
return True
@abstractmethod
def receive_frame(self, frame: Frame) -> bool:
@@ -737,12 +740,21 @@ class Link(SimComponent):
"""
return self.endpoint_a.enabled and self.endpoint_b.enabled
def _can_transmit(self, frame: Frame) -> bool:
def can_transmit_frame(self, frame: Frame) -> bool:
"""
Determines whether a frame can be transmitted considering the current Link load and the Link's bandwidth.
This method assesses if the transmission of a given frame is possible without exceeding the Link's total
bandwidth capacity. It checks if the current load of the Link plus the size of the frame (expressed in Mbps)
would remain within the defined bandwidth limits. The transmission is only feasible if the Link is active
('up') and the total load including the new frame does not surpass the bandwidth limit.
:param frame: The frame intended for transmission, which contains its size in Mbps.
:return: True if the frame can be transmitted without exceeding the bandwidth limit, False otherwise.
"""
if self.is_up:
frame_size_Mbits = frame.size_Mbits # noqa - Leaving it as Mbits as this is how they're expressed
# return self.current_load + frame_size_Mbits <= self.bandwidth
# TODO: re add this check once packet size limiting and MTU checks are implemented
return True
return self.current_load + frame.size_Mbits <= self.bandwidth
return False
def transmit_frame(self, sender_nic: WiredNetworkInterface, frame: Frame) -> bool:
@@ -753,11 +765,6 @@ class Link(SimComponent):
:param frame: The network frame to be sent.
:return: True if the Frame can be sent, otherwise False.
"""
can_transmit = self._can_transmit(frame)
if not can_transmit:
_LOGGER.debug(f"Cannot transmit frame as {self} is at capacity")
return False
receiver = self.endpoint_a
if receiver == sender_nic:
receiver = self.endpoint_b

16
src/primaite/simulator/network/hardware/nodes/network/switch.py
View File

@@ -58,12 +58,16 @@ class SwitchPort(WiredNetworkInterface):
:param frame: The network frame to be sent.
:return: A boolean indicating whether the frame was successfully sent.
"""
if self.enabled:
self.pcap.capture_outbound(frame)
self._connected_link.transmit_frame(sender_nic=self, frame=frame)
return True
# Cannot send Frame as the SwitchPort is not enabled
return False
if not self.enabled:
return False
if not self._connected_link.can_transmit_frame(frame):
# Drop frame for now. Queuing will happen here (probably) if it's done in the future.
self._connected_node.sys_log.info(f"{self}: Frame dropped as Link is at capacity")
return False
self.pcap.capture_outbound(frame)
self._connected_link.transmit_frame(sender_nic=self, frame=frame)
return True
def receive_frame(self, frame: Frame) -> bool:
"""

26
src/primaite/simulator/network/hardware/nodes/network/wireless_router.py
View File

@@ -1,10 +1,10 @@
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
from ipaddress import IPv4Address
from typing import Any, Dict, Union
from typing import Any, Dict, Optional, Union
from pydantic import validate_call
from primaite.simulator.network.airspace import AirSpace, AirSpaceFrequency, IPWirelessNetworkInterface
from primaite.simulator.network.airspace import AirSpace, AirSpaceFrequency, ChannelWidth, IPWirelessNetworkInterface
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.hardware.nodes.network.router import ACLAction, Router, RouterInterface
from primaite.simulator.network.transmission.data_link_layer import Frame
@@ -153,7 +153,8 @@ class WirelessRouter(Router):
self,
ip_address: IPV4Address,
subnet_mask: IPV4Address,
frequency: AirSpaceFrequency = AirSpaceFrequency.WIFI_2_4,
frequency: Optional[AirSpaceFrequency] = AirSpaceFrequency.WIFI_2_4,
channel_width: Optional[ChannelWidth] = ChannelWidth.WIDTH_40_MHZ,
):
"""
Configures a wireless access point (WAP).
@@ -170,13 +171,23 @@ class WirelessRouter(Router):
enum. This determines the frequency band (e.g., 2.4 GHz or 5 GHz) the access point will use for wireless
communication. Default is AirSpaceFrequency.WIFI_2_4.
"""
if not frequency:
frequency = AirSpaceFrequency.WIFI_2_4
if not channel_width:
channel_width = ChannelWidth.WIDTH_40_MHZ
self.sys_log.info("Configuring wireless access point")
self.wireless_access_point.disable() # Temporarily disable the WAP for reconfiguration
network_interface = self.network_interface[1]
network_interface.ip_address = ip_address
network_interface.subnet_mask = subnet_mask
self.sys_log.info(f"Configured WAP {network_interface}")
self.wireless_access_point.frequency = frequency # Set operating frequency
self.wireless_access_point.channel_width = channel_width
self.wireless_access_point.enable() # Re-enable the WAP with new settings
self.sys_log.info(f"Configured WAP {network_interface}")
@property
def router_interface(self) -> RouterInterface:
@@ -258,7 +269,12 @@ class WirelessRouter(Router):
ip_address = cfg["wireless_access_point"]["ip_address"]
subnet_mask = cfg["wireless_access_point"]["subnet_mask"]
frequency = AirSpaceFrequency[cfg["wireless_access_point"]["frequency"]]
router.configure_wireless_access_point(ip_address=ip_address, subnet_mask=subnet_mask, frequency=frequency)
channel_width = cfg["wireless_access_point"].get("channel_width")
if channel_width:
channel_width = ChannelWidth(channel_width)
router.configure_wireless_access_point(
ip_address=ip_address, subnet_mask=subnet_mask, frequency=frequency, channel_width=channel_width
)
if "acl" in cfg:
for r_num, r_cfg in cfg["acl"].items():

5
src/primaite/simulator/network/transmission/data_link_layer.py
View File

@@ -133,10 +133,11 @@ class Frame(BaseModel):
def size(self) -> float: # noqa - Keep it as MBits as this is how they're expressed
"""The size of the Frame in Bytes."""
# get the payload size if it is a data packet
payload_size = 0.0
if isinstance(self.payload, DataPacket):
return self.payload.get_packet_size()
payload_size = self.payload.get_packet_size()
return float(len(self.model_dump_json().encode("utf-8")))
return float(len(self.model_dump_json().encode("utf-8"))) + payload_size
@property
def size_Mbits(self) -> float: # noqa - Keep it as MBits as this is how they're expressed