Merged PR 152: Model Link and NIC

## Summary
- #1715 #1715 - Added Link class in physical_layer.py.- Also added NIC class in physical_layer.py for
#1672. Added attributes and public API functions.
- Added test_physical_layer.py with some basic tests ready to house the tests once logic has been implemented.
- Made use of the pydantic model_post_init function for proper ipv4 configuration checking.
- Added NetworkError to exceptions.py.
- Added timestep int as a param to the apply_timestep function in core.py. Also added a reset_component_for_episode function.
- Updated docs with details of Link and NIC.

## Test process
- Added basic unit tests for instantiation that look at proper config and error raising.
- Added an integration test for linking NICs to Link.

## Checklist
- [ ] This PR is linked to a **work item**
- [ ] I have performed **self-review** of the code
- [ ] I have written **tests** for any new functionality added with this PR
- [ ] I have updated the **documentation** if this PR changes or adds functionality
- [ ] I have run **pre-commit** checks for code style

Related work items: #1672, #1715

.flake8

@@ -9,8 +9,12 @@ extend-ignore =
     E712
     D401
     F811
+    ANN002
+    ANN003
     ANN101
     ANN102
 exclude =
     docs/source/*
     tests/*
+suppress-none-returning=True
+suppress-dummy-args=True

docs/index.rst

@@ -36,7 +36,6 @@ Head over to the :ref:`getting-started` page to install and setup PrimAITE!
 .. toctree::
    :maxdepth: 8
    :caption: Contents:
-   :hidden:
 
    source/getting_started
    source/about

docs/source/simulation.rst

@@ -2,9 +2,18 @@
 
     © Crown-owned copyright 2023, Defence Science and Technology Laboratory UK
 
-Simulation Strucutre
-====================
 
-The simulation is made up of many smaller components which are related to each other in a tree-like structure. At the top level, there is an object called the ``SimulationController`` _(doesn't exist yet)_, which has a physical network and a software controller for managing software and users.
+Simulation
+==========
 
-Each node of the simulation 'tree' has responsibility for creating, deleting, and updating its direct descendants.
+.. TODO:: Add spiel here about what the simulation is.
+
+
+Contents
+########
+
+.. toctree::
+   :maxdepth: 8
+
+   simulation_structure
+   simulation_components/network/physical_layer

docs/source/simulation_components/network/physical_layer.rst

@@ -0,0 +1,75 @@
+.. only:: comment
+
+    © Crown-owned copyright 2023, Defence Science and Technology Laboratory UK
+
+Physical Layer
+==============
+
+The physical layer components are models of a ``NIC`` (Network Interface Card) and a ``Link``. These components allow
+modelling of layer 1 (physical layer) in the OSI model.
+
+NIC
+###
+The ``NIC`` class is a realistic model of a Network Interface Card. The ``NIC`` acts as the interface between the
+``Node`` and the ``Link``.
+
+NICs have the following attributes:
+
+- **ip_address:** The IPv4 address assigned to the NIC.
+- **subnet_mask:** The subnet mask assigned to the NIC.
+- **gateway:** The default gateway IP address for forwarding network traffic to other networks.
+- **mac_address:** The MAC address of the NIC. Defaults to a randomly set MAC address.
+- **speed:** The speed of the NIC in Mbps (default is 100 Mbps).
+- **mtu:** The Maximum Transmission Unit (MTU) of the NIC in Bytes, representing the largest data packet size it can handle without fragmentation (default is 1500 B).
+- **wake_on_lan:** Indicates if the NIC supports Wake-on-LAN functionality.
+- **dns_servers:** List of IP addresses of DNS servers used for name resolution.
+- **connected_link:** The link to which the NIC is connected.
+- **enabled:** Indicates whether the NIC is enabled.
+
+**Basic Example**
+
+.. code-block:: python
+
+    nic1 = NIC(
+        ip_address="192.168.1.100",
+        subnet_mask="255.255.255.0",
+        gateway="192.168.1.1"
+    )
+
+Link
+####
+
+The ``Link`` class represents a physical link between two network endpoints.
+
+Links have the following attributes:
+
+- **endpoint_a:** The first NIC connected to the Link.
+- **endpoint_b:** The second NIC connected to the Link.
+- **bandwidth:** The bandwidth of the Link in Mbps (default is 100 Mbps).
+- **current_load:** The current load on the link in Mbps.
+
+**Basic Example**
+
+.. code-block:: python
+
+    nic1 = NIC(
+        ip_address="192.168.1.100",
+        subnet_mask="255.255.255.0",
+        gateway="192.168.1.1"
+    )
+    nic1 = NIC(
+        ip_address="192.168.1.101",
+        subnet_mask="255.255.255.0",
+        gateway="192.168.1.1"
+    )
+
+    link = Link(
+        endpoint_a=nic1,
+        endpoint_b=nic2,
+        bandwidth=1000
+    )
+
+Link, NIC, Node Interface
+#########################
+
+.. image:: ../../../_static/node_nic_link_component_diagram.png

docs/source/simulation_structure.rst

@@ -0,0 +1,13 @@
+.. only:: comment
+
+    © Crown-owned copyright 2023, Defence Science and Technology Laboratory UK
+
+
+Simulation Structure
+====================
+
+The simulation is made up of many smaller components which are related to each other in a tree-like structure. At the
+top level, there is an object called the ``SimulationController`` _(doesn't exist yet)_, which has a physical network
+and a software controller for managing software and users.
+
+Each node of the simulation 'tree' has responsibility for creating, deleting, and updating its direct descendants.

src/primaite/exceptions.py

@@ -9,3 +9,9 @@ class RLlibAgentError(PrimaiteError):
     """Raised when there is a generic error with a RLlib agent that is specific to PRimAITE."""
 
     pass
+
+
+class NetworkError(PrimaiteError):
+    """Raised when an error occurs at the network level."""
+
+    pass

src/primaite/simulator/core.py

@@ -37,7 +37,7 @@ class SimComponent(BaseModel):
         possible_actions = self._possible_actions()
         if action[0] in possible_actions:
             # take the first element off the action list and pass the remaining arguments to the corresponding action
-            # funciton
+            # function
             possible_actions[action.pop(0)](action)
         else:
             raise ValueError(f"{self.__class__.__name__} received invalid action {action}")
@@ -45,7 +45,7 @@ class SimComponent(BaseModel):
     def _possible_actions(self) -> Dict[str, Callable[[List[str]], None]]:
         return {}
 
-    def apply_timestep(self) -> None:
+    def apply_timestep(self, timestep: int) -> None:
         """
         Apply a timestep evolution to this component.
 
@@ -53,3 +53,11 @@ class SimComponent(BaseModel):
         sending data.
         """
         pass
+
+    def reset_component_for_episode(self):
+        """
+        Reset this component to its original state for a new episode.
+
+        Override this method with anything that needs to happen within the component for it to be reset.
+        """
+        pass

src/primaite/simulator/network/physical_layer.py

@@ -0,0 +1,273 @@
+from __future__ import annotations
+
+import re
+import secrets
+from ipaddress import IPv4Address, IPv4Network
+from typing import Any, Dict, List, Optional
+
+from primaite import getLogger
+from primaite.exceptions import NetworkError
+from primaite.simulator.core import SimComponent
+
+_LOGGER = getLogger(__name__)
+
+
+def generate_mac_address(oui: Optional[str] = None) -> str:
+    """
+    Generate a random MAC Address.
+
+    :Example:
+
+    >>> generate_mac_address()
+    'ef:7e:97:c8:a8:ce'
+
+    >>> generate_mac_address(oui='aa:bb:cc')
+    'aa:bb:cc:42:ba:41'
+
+    :param oui: The Organizationally Unique Identifier (OUI) portion of the MAC address. It should be a string with
+        the first 3 bytes (24 bits) in the format "XX:XX:XX".
+    :raises ValueError: If the 'oui' is not in the correct format (hexadecimal and 6 characters).
+    """
+    random_bytes = [secrets.randbits(8) for _ in range(6)]
+
+    if oui:
+        oui_pattern = re.compile(r"^([0-9A-Fa-f]{2}[:-]){2}[0-9A-Fa-f]{2}$")
+        if not oui_pattern.match(oui):
+            msg = f"Invalid oui. The oui should be in the format xx:xx:xx, where x is a hexadecimal digit, got '{oui}'"
+            raise ValueError(msg)
+        oui_bytes = [int(chunk, 16) for chunk in oui.split(":")]
+        mac = oui_bytes + random_bytes[len(oui_bytes) :]
+    else:
+        mac = random_bytes
+
+    return ":".join(f"{b:02x}" for b in mac)
+
+
+class NIC(SimComponent):
+    """
+    Models a Network Interface Card (NIC) in a computer or network device.
+
+    :param ip_address: The IPv4 address assigned to the NIC.
+    :param subnet_mask: The subnet mask assigned to the NIC.
+    :param gateway: The default gateway IP address for forwarding network traffic to other networks.
+    :param mac_address: The MAC address of the NIC. Defaults to a randomly set MAC address.
+    :param speed: The speed of the NIC in Mbps (default is 100 Mbps).
+    :param mtu: The Maximum Transmission Unit (MTU) of the NIC in Bytes, representing the largest data packet size it
+        can handle without fragmentation (default is 1500 B).
+    :param wake_on_lan: Indicates if the NIC supports Wake-on-LAN functionality.
+    :param dns_servers: List of IP addresses of DNS servers used for name resolution.
+    """
+
+    ip_address: IPv4Address
+    "The IP address assigned to the NIC for communication on an IP-based network."
+    subnet_mask: str
+    "The subnet mask assigned to the NIC."
+    gateway: IPv4Address
+    "The default gateway IP address for forwarding network traffic to other networks. Randomly generated upon creation."
+    mac_address: str = generate_mac_address()
+    "The MAC address of the NIC. Defaults to a randomly set MAC address."
+    speed: int = 100
+    "The speed of the NIC in Mbps. Default is 100 Mbps."
+    mtu: int = 1500
+    "The Maximum Transmission Unit (MTU) of the NIC in Bytes. Default is 1500 B"
+    wake_on_lan: bool = False
+    "Indicates if the NIC supports Wake-on-LAN functionality."
+    dns_servers: List[IPv4Address] = []
+    "List of IP addresses of DNS servers used for name resolution."
+    connected_link: Optional[Link] = None
+    "The Link to which the NIC is connected."
+    enabled: bool = False
+    "Indicates whether the NIC is enabled."
+
+    def __init__(self, **kwargs):
+        """
+        NIC constructor.
+
+        Performs some type conversion the calls ``super().__init__()``. Then performs some checking on the ip_address
+        and gateway just to check that it's all been configured correctly.
+
+        :raises ValueError: When the ip_address and gateway are the same. And when the ip_address/subnet mask are a
+            network address.
+        """
+        if not isinstance(kwargs["ip_address"], IPv4Address):
+            kwargs["ip_address"] = IPv4Address(kwargs["ip_address"])
+        if not isinstance(kwargs["gateway"], IPv4Address):
+            kwargs["gateway"] = IPv4Address(kwargs["gateway"])
+        super().__init__(**kwargs)
+
+        if self.ip_address == self.gateway:
+            msg = f"NIC ip address {self.ip_address} cannot be the same as the gateway {self.gateway}"
+            _LOGGER.error(msg)
+            raise ValueError(msg)
+        if self.ip_network.network_address == self.ip_address:
+            msg = (
+                f"Failed to set IP address {self.ip_address} and subnet mask {self.subnet_mask} as it is a "
+                f"network address {self.ip_network.network_address}"
+            )
+            _LOGGER.error(msg)
+            raise ValueError(msg)
+
+    @property
+    def ip_network(self) -> IPv4Network:
+        """
+        Return the IPv4Network of the NIC.
+
+        :return: The IPv4Network from the ip_address/subnet mask.
+        """
+        return IPv4Network(f"{self.ip_address}/{self.subnet_mask}", strict=False)
+
+    def connect_link(self, link: Link):
+        """
+        Connect the NIC to a link.
+
+        :param link: The link to which the NIC is connected.
+        :type link: :class:`~primaite.simulator.network.physical_layer.Link`
+        :raise NetworkError: When an attempt to connect a Link is made while the NIC has a connected Link.
+        """
+        if not self.connected_link:
+            if self.connected_link != link:
+                # TODO: Inform the Node that a link has been connected
+                self.connected_link = link
+            else:
+                _LOGGER.warning(f"Cannot connect link to NIC ({self.mac_address}) as it is already connected")
+        else:
+            msg = f"Cannot connect link to NIC ({self.mac_address}) as it already has a connection"
+            _LOGGER.error(msg)
+            raise NetworkError(msg)
+
+    def disconnect_link(self):
+        """Disconnect the NIC from the connected :class:`~primaite.simulator.network.physical_layer.Link`."""
+        if self.connected_link.endpoint_a == self:
+            self.connected_link.endpoint_a = None
+        if self.connected_link.endpoint_b == self:
+            self.connected_link.endpoint_b = None
+        self.connected_link = None
+
+    def add_dns_server(self, ip_address: IPv4Address):
+        """
+        Add a DNS server IP address.
+
+        :param ip_address: The IP address of the DNS server to be added.
+        :type ip_address: ipaddress.IPv4Address
+        """
+        pass
+
+    def remove_dns_server(self, ip_address: IPv4Address):
+        """
+        Remove a DNS server IP Address.
+
+        :param ip_address: The IP address of the DNS server to be removed.
+        :type ip_address: ipaddress.IPv4Address
+        """
+        pass
+
+    def send_frame(self, frame: Any):
+        """
+        Send a network frame from the NIC to the connected link.
+
+        :param frame: The network frame to be sent.
+        :type frame: :class:`~primaite.simulator.network.osi_layers.Frame`
+        """
+        pass
+
+    def receive_frame(self, frame: Any):
+        """
+        Receive a network frame from the connected link.
+
+        The Frame is passed to the Node.
+
+        :param frame: The network frame being received.
+        :type frame: :class:`~primaite.simulator.network.osi_layers.Frame`
+        """
+        pass
+
+    def describe_state(self) -> Dict:
+        """
+        Get the current state of the NIC as a dict.
+
+        :return: A dict containing the current state of the NIC.
+        """
+        pass
+
+    def apply_action(self, action: str):
+        """
+        Apply an action to the NIC.
+
+        :param action: The action to be applied.
+        :type action: str
+        """
+        pass
+
+
+class Link(SimComponent):
+    """
+    Represents a network link between two network interface cards (NICs).
+
+    :param endpoint_a: The first NIC connected to the Link.
+    :type endpoint_a: NIC
+    :param endpoint_b: The second NIC connected to the Link.
+    :type endpoint_b: NIC
+    :param bandwidth: The bandwidth of the Link in Mbps (default is 100 Mbps).
+    :type bandwidth: int
+    """
+
+    endpoint_a: NIC
+    "The first NIC connected to the Link."
+    endpoint_b: NIC
+    "The second NIC connected to the Link."
+    bandwidth: int = 100
+    "The bandwidth of the Link in Mbps (default is 100 Mbps)."
+    current_load: int = 0
+    "The current load on the link in Mbps."
+
+    def model_post_init(self, __context: Any) -> None:
+        """
+        Ensure that endpoint_a and endpoint_b are not the same :class:`~primaite.simulator.network.physical_layer.NIC`.
+
+        :raises ValueError: If endpoint_a and endpoint_b are the same NIC.
+        """
+        if self.endpoint_a == self.endpoint_b:
+            msg = "endpoint_a and endpoint_b cannot be the same NIC"
+            _LOGGER.error(msg)
+            raise ValueError(msg)
+        self.endpoint_a.connect_link(self)
+        self.endpoint_b.connect_link(self)
+
+    def send_frame(self, sender_nic: NIC, frame: Any):
+        """
+        Send a network frame from one NIC to another connected NIC.
+
+        :param sender_nic: The NIC sending the frame.
+        :type sender_nic: NIC
+        :param frame: The network frame to be sent.
+        :type frame: Frame
+        """
+        pass
+
+    def receive_frame(self, sender_nic: NIC, frame: Any):
+        """
+        Receive a network frame from a connected NIC.
+
+        :param sender_nic: The NIC sending the frame.
+        :type sender_nic: NIC
+        :param frame: The network frame being received.
+        :type frame: Frame
+        """
+        pass
+
+    def describe_state(self) -> Dict:
+        """
+        Get the current state of the Libk as a dict.
+
+        :return: A dict containing the current state of the Link.
+        """
+        pass
+
+    def apply_action(self, action: str):
+        """
+        Apply an action to the Link.
+
+        :param action: The action to be applied.
+        :type action: str
+        """
+        pass

tests/integration_tests/network/test_nic_link_connection.py

@@ -0,0 +1,14 @@
+import pytest
+
+from primaite.simulator.network.physical_layer import Link, NIC
+
+
+def test_link_fails_with_same_nic():
+    """Tests Link creation fails with endpoint_a and endpoint_b are the same NIC."""
+    with pytest.raises(ValueError):
+        nic_a = NIC(
+            ip_address="192.168.1.2",
+            subnet_mask="255.255.255.0",
+            gateway="192.168.0.1",
+        )
+        Link(endpoint_a=nic_a, endpoint_b=nic_a)

tests/unit_tests/_primaite/_simulator/_network/test_physical_layer.py

@@ -0,0 +1,71 @@
+import re
+from ipaddress import IPv4Address
+
+import pytest
+
+from primaite.simulator.network.physical_layer import generate_mac_address, NIC
+
+
+def test_mac_address_generation():
+    """Tests random mac address generation."""
+    mac_address = generate_mac_address()
+    assert re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$", mac_address)
+
+
+def test_mac_address_with_oui():
+    """Tests random mac address generation with oui."""
+    oui = "aa:bb:cc"
+    mac_address = generate_mac_address(oui=oui)
+    assert re.match("[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$", mac_address)
+    assert mac_address[:8] == oui
+
+
+def test_invalid_oui_mac_address():
+    """Tests random mac address generation fails with invalid oui."""
+    invalid_oui = "aa-bb-cc"
+    with pytest.raises(ValueError):
+        generate_mac_address(oui=invalid_oui)
+
+
+def test_nic_ip_address_type_conversion():
+    """Tests NIC IP and gateway address is converted to IPv4Address is originally a string."""
+    nic = NIC(
+        ip_address="192.168.1.2",
+        subnet_mask="255.255.255.0",
+        gateway="192.168.0.1",
+    )
+    assert isinstance(nic.ip_address, IPv4Address)
+    assert isinstance(nic.gateway, IPv4Address)
+
+
+def test_nic_deserialize():
+    """Tests NIC serialization and deserialization."""
+    nic = NIC(
+        ip_address="192.168.1.2",
+        subnet_mask="255.255.255.0",
+        gateway="192.168.0.1",
+    )
+
+    nic_json = nic.model_dump_json()
+    deserialized_nic = NIC.model_validate_json(nic_json)
+    assert nic == deserialized_nic
+
+
+def test_nic_ip_address_as_gateway_fails():
+    """Tests NIC creation fails if ip address is the same as the gateway."""
+    with pytest.raises(ValueError):
+        NIC(
+            ip_address="192.168.0.1",
+            subnet_mask="255.255.255.0",
+            gateway="192.168.0.1",
+        )
+
+
+def test_nic_ip_address_as_network_address_fails():
+    """Tests NIC creation fails if ip address and subnet mask are a network address."""
+    with pytest.raises(ValueError):
+        NIC(
+            ip_address="192.168.0.0",
+            subnet_mask="255.255.255.0",
+            gateway="192.168.0.1",
+        )