Convert office lan adder to a class and make it extendable
This commit is contained in:
@@ -1539,6 +1539,25 @@ class Node(SimComponent):
|
||||
SYSTEM_SOFTWARE: ClassVar[Dict[str, Type[Software]]] = {}
|
||||
"Base system software that must be preinstalled."
|
||||
|
||||
_registry: ClassVar[Dict[str, Type["Node"]]] = {}
|
||||
"""Registry of application types. Automatically populated when subclasses are defined."""
|
||||
|
||||
def __init_subclass__(cls, identifier: str = "default", **kwargs: Any) -> None:
|
||||
"""
|
||||
Register a node type.
|
||||
|
||||
:param identifier: Uniquely specifies an node class by name. Used for finding items by config.
|
||||
:type identifier: str
|
||||
:raises ValueError: When attempting to register an node with a name that is already allocated.
|
||||
"""
|
||||
if identifier == "default":
|
||||
return
|
||||
identifier = identifier.lower()
|
||||
super().__init_subclass__(**kwargs)
|
||||
if identifier in cls._registry:
|
||||
raise ValueError(f"Tried to define new node {identifier}, but this name is already reserved.")
|
||||
cls._registry[identifier] = cls
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
"""
|
||||
Initialize the Node with various components and managers.
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from ipaddress import IPv4Address
|
||||
from typing import Any, ClassVar, Dict, Optional, Type
|
||||
from typing import Any, ClassVar, Dict, Optional
|
||||
|
||||
from primaite import getLogger
|
||||
from primaite.simulator.network.hardware.base import (
|
||||
@@ -325,30 +325,10 @@ class HostNode(Node):
|
||||
network_interface: Dict[int, NIC] = {}
|
||||
"The NICs on the node by port id."
|
||||
|
||||
_registry: ClassVar[Dict[str, Type["HostNode"]]] = {}
|
||||
"""Registry of application types. Automatically populated when subclasses are defined."""
|
||||
|
||||
def __init__(self, ip_address: IPV4Address, subnet_mask: IPV4Address, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self.connect_nic(NIC(ip_address=ip_address, subnet_mask=subnet_mask))
|
||||
|
||||
def __init_subclass__(cls, identifier: str = "default", **kwargs: Any) -> None:
|
||||
"""
|
||||
Register a hostnode type.
|
||||
|
||||
:param identifier: Uniquely specifies an hostnode class by name. Used for finding items by config.
|
||||
:type identifier: str
|
||||
:raises ValueError: When attempting to register an hostnode with a name that is already allocated.
|
||||
"""
|
||||
if identifier == "default":
|
||||
return
|
||||
# Enforce lowercase registry entries because it makes comparisons everywhere else much easier.
|
||||
identifier = identifier.lower()
|
||||
super().__init_subclass__(**kwargs)
|
||||
if identifier in cls._registry:
|
||||
raise ValueError(f"Tried to define new hostnode {identifier}, but this name is already reserved.")
|
||||
cls._registry[identifier] = cls
|
||||
|
||||
@property
|
||||
def nmap(self) -> Optional[NMAP]:
|
||||
"""
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
|
||||
from abc import abstractmethod
|
||||
from typing import Any, ClassVar, Dict, Optional, Type
|
||||
from typing import Optional
|
||||
|
||||
from primaite.simulator.network.hardware.base import NetworkInterface, Node
|
||||
from primaite.simulator.network.transmission.data_link_layer import Frame
|
||||
@@ -16,25 +16,6 @@ class NetworkNode(Node):
|
||||
provide functionality for receiving and processing frames received on their network interfaces.
|
||||
"""
|
||||
|
||||
_registry: ClassVar[Dict[str, Type["NetworkNode"]]] = {}
|
||||
"""Registry of application types. Automatically populated when subclasses are defined."""
|
||||
|
||||
def __init_subclass__(cls, identifier: str = "default", **kwargs: Any) -> None:
|
||||
"""
|
||||
Register a networknode type.
|
||||
|
||||
:param identifier: Uniquely specifies an networknode class by name. Used for finding items by config.
|
||||
:type identifier: str
|
||||
:raises ValueError: When attempting to register an networknode with a name that is already allocated.
|
||||
"""
|
||||
if identifier == "default":
|
||||
return
|
||||
identifier = identifier.lower()
|
||||
super().__init_subclass__(**kwargs)
|
||||
if identifier in cls._registry:
|
||||
raise ValueError(f"Tried to define new networknode {identifier}, but this name is already reserved.")
|
||||
cls._registry[identifier] = cls
|
||||
|
||||
@abstractmethod
|
||||
def receive_frame(self, frame: Frame, from_network_interface: NetworkInterface):
|
||||
"""
|
||||
|
||||
@@ -4,7 +4,7 @@ from typing import Any, Dict, Optional, Union
|
||||
|
||||
from pydantic import validate_call
|
||||
|
||||
from primaite.simulator.network.airspace import AirSpace, IPWirelessNetworkInterface
|
||||
from primaite.simulator.network.airspace import AirSpace, AirSpaceFrequency, FREQ_WIFI_2_4, 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,7 @@ class WirelessRouter(Router):
|
||||
self,
|
||||
ip_address: IPV4Address,
|
||||
subnet_mask: IPV4Address,
|
||||
frequency: Optional[str] = "WIFI_2_4",
|
||||
frequency: Optional[AirSpaceFrequency] = FREQ_WIFI_2_4,
|
||||
):
|
||||
"""
|
||||
Configures a wireless access point (WAP).
|
||||
@@ -171,7 +171,7 @@ class WirelessRouter(Router):
|
||||
communication. Default is "WIFI_2_4".
|
||||
"""
|
||||
if not frequency:
|
||||
frequency = "WIFI_2_4"
|
||||
frequency = FREQ_WIFI_2_4
|
||||
self.sys_log.info("Configuring wireless access point")
|
||||
|
||||
self.wireless_access_point.disable() # Temporarily disable the WAP for reconfiguration
|
||||
@@ -264,7 +264,7 @@ class WirelessRouter(Router):
|
||||
if "wireless_access_point" in cfg:
|
||||
ip_address = cfg["wireless_access_point"]["ip_address"]
|
||||
subnet_mask = cfg["wireless_access_point"]["subnet_mask"]
|
||||
frequency = cfg["wireless_access_point"]["frequency"]
|
||||
frequency = AirSpaceFrequency._registry[cfg["wireless_access_point"]["frequency"]]
|
||||
router.configure_wireless_access_point(ip_address=ip_address, subnet_mask=subnet_mask, frequency=frequency)
|
||||
|
||||
if "acl" in cfg:
|
||||
|
||||
Reference in New Issue
Block a user