diff --git a/daemon/core/emane/emanemanager.py b/daemon/core/emane/emanemanager.py index 294ae528..c02570c9 100644 --- a/daemon/core/emane/emanemanager.py +++ b/daemon/core/emane/emanemanager.py @@ -6,7 +6,7 @@ import logging import os import threading from enum import Enum -from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type, Union +from typing import TYPE_CHECKING, Optional, Union from core import utils from core.emane.emanemodel import EmaneModel @@ -126,9 +126,9 @@ class EmaneManager: """ super().__init__() self.session: "Session" = session - self.nems_to_ifaces: Dict[int, CoreInterface] = {} - self.ifaces_to_nems: Dict[CoreInterface, int] = {} - self._emane_nets: Dict[int, EmaneNet] = {} + self.nems_to_ifaces: dict[int, CoreInterface] = {} + self.ifaces_to_nems: dict[CoreInterface, int] = {} + self._emane_nets: dict[int, EmaneNet] = {} self._emane_node_lock: threading.Lock = threading.Lock() # port numbers are allocated from these counters self.platformport: int = self.session.options.get_int( @@ -141,14 +141,14 @@ class EmaneManager: self.eventmonthread: Optional[threading.Thread] = None # model for global EMANE configuration options - self.node_configs: Dict[int, Dict[str, Dict[str, str]]] = {} - self.node_models: Dict[int, str] = {} + self.node_configs: dict[int, dict[str, dict[str, str]]] = {} + self.node_models: dict[int, str] = {} # link monitor self.link_monitor: EmaneLinkMonitor = EmaneLinkMonitor(self) # emane event monitoring - self.services: Dict[str, EmaneEventService] = {} - self.nem_service: Dict[int, EmaneEventService] = {} + self.services: dict[str, EmaneEventService] = {} + self.nem_service: dict[int, EmaneEventService] = {} def next_nem_id(self, iface: CoreInterface) -> int: nem_id = self.session.options.get_int("nem_id_start") @@ -161,7 +161,7 @@ class EmaneManager: def get_config( self, key: int, model: str, default: bool = True - ) -> Optional[Dict[str, str]]: + ) -> Optional[dict[str, str]]: """ Get the current or default configuration for an emane model. @@ -181,7 +181,7 @@ class EmaneManager: config = model_class.default_values() return config - def set_config(self, key: int, model: str, config: Dict[str, str] = None) -> None: + def set_config(self, key: int, model: str, config: dict[str, str] = None) -> None: """ Sets and update the provided configuration against the default model or currently set emane model configuration. @@ -199,7 +199,7 @@ class EmaneManager: model_configs = self.node_configs.setdefault(key, {}) model_configs[model] = model_config - def get_model(self, model_name: str) -> Type[EmaneModel]: + def get_model(self, model_name: str) -> type[EmaneModel]: """ Convenience method for getting globally loaded emane models. @@ -211,7 +211,7 @@ class EmaneManager: def get_iface_config( self, emane_net: EmaneNet, iface: CoreInterface - ) -> Dict[str, str]: + ) -> dict[str, str]: """ Retrieve configuration for a given interface, first checking for interface specific config, node specific config, network specific config, and finally @@ -260,7 +260,7 @@ class EmaneManager: ) self._emane_nets[emane_net.id] = emane_net - def getnodes(self) -> Set[CoreNode]: + def getnodes(self) -> set[CoreNode]: """ Return a set of CoreNodes that are linked to an EMANE network, e.g. containers having one or more radio interfaces. @@ -335,7 +335,7 @@ class EmaneManager: self.start_daemon(iface) self.install_iface(iface, config) - def get_ifaces(self) -> List[Tuple[EmaneNet, TunTap]]: + def get_ifaces(self) -> list[tuple[EmaneNet, TunTap]]: ifaces = [] for emane_net in self._emane_nets.values(): if not emane_net.wireless_model: @@ -354,7 +354,7 @@ class EmaneManager: return sorted(ifaces, key=lambda x: (x[1].node.id, x[1].id)) def setup_control_channels( - self, nem_id: int, iface: CoreInterface, config: Dict[str, str] + self, nem_id: int, iface: CoreInterface, config: dict[str, str] ) -> None: node = iface.node # setup ota device @@ -419,7 +419,7 @@ class EmaneManager: def get_nem_position( self, iface: CoreInterface - ) -> Optional[Tuple[int, float, float, int]]: + ) -> Optional[tuple[int, float, float, int]]: """ Retrieves nem position for a given interface. @@ -453,7 +453,7 @@ class EmaneManager: event.append(nemid, latitude=lat, longitude=lon, altitude=alt) self.publish_event(nemid, event, send_all=True) - def set_nem_positions(self, moved_ifaces: List[CoreInterface]) -> None: + def set_nem_positions(self, moved_ifaces: list[CoreInterface]) -> None: """ Several NEMs have moved, from e.g. a WaypointMobilityModel calculation. Generate an EMANE Location Event having several @@ -480,7 +480,7 @@ class EmaneManager: try: with path.open("a") as f: f.write(f"{iface.node.name} {iface.name} {nem_id}\n") - except IOError: + except OSError: logger.exception("error writing to emane nem file") def links_enabled(self) -> bool: @@ -624,7 +624,7 @@ class EmaneManager: args = f"{emanecmd} -f {log_file} {platform_xml}" node.host_cmd(args, cwd=self.session.directory) - def install_iface(self, iface: TunTap, config: Dict[str, str]) -> None: + def install_iface(self, iface: TunTap, config: dict[str, str]) -> None: external = config.get("external", "0") if external == "0": iface.set_ips() diff --git a/daemon/core/emane/emanemanifest.py b/daemon/core/emane/emanemanifest.py index 0fb5bc17..ea2b05fd 100644 --- a/daemon/core/emane/emanemanifest.py +++ b/daemon/core/emane/emanemanifest.py @@ -1,6 +1,5 @@ import logging from pathlib import Path -from typing import Dict, List from core.config import Configuration from core.emulator.enumerations import ConfigDataTypes @@ -33,7 +32,7 @@ def _type_value(config_type: str) -> ConfigDataTypes: return ConfigDataTypes[config_type] -def _get_possible(config_type: str, config_regex: str) -> List[str]: +def _get_possible(config_type: str, config_regex: str) -> list[str]: """ Retrieve possible config value options based on emane regexes. @@ -51,7 +50,7 @@ def _get_possible(config_type: str, config_regex: str) -> List[str]: return [] -def _get_default(config_type_name: str, config_value: List[str]) -> str: +def _get_default(config_type_name: str, config_value: list[str]) -> str: """ Convert default configuration values to one used by core. @@ -74,7 +73,7 @@ def _get_default(config_type_name: str, config_value: List[str]) -> str: return config_default -def parse(manifest_path: Path, defaults: Dict[str, str]) -> List[Configuration]: +def parse(manifest_path: Path, defaults: dict[str, str]) -> list[Configuration]: """ Parses a valid emane manifest file and converts the provided configuration values into ones used by core. diff --git a/daemon/core/emane/emanemodel.py b/daemon/core/emane/emanemodel.py index cc5b0f4d..4e31d632 100644 --- a/daemon/core/emane/emanemodel.py +++ b/daemon/core/emane/emanemodel.py @@ -3,7 +3,7 @@ Defines Emane Models used within CORE. """ import logging from pathlib import Path -from typing import Dict, List, Optional, Set +from typing import Optional from core.config import ConfigBool, ConfigGroup, ConfigString, Configuration from core.emane import emanemanifest @@ -28,38 +28,38 @@ class EmaneModel(WirelessModel): # default platform configuration settings platform_controlport: str = "controlportendpoint" platform_xml: str = "nemmanager.xml" - platform_defaults: Dict[str, str] = { + platform_defaults: dict[str, str] = { "eventservicedevice": DEFAULT_DEV, "eventservicegroup": "224.1.2.8:45703", "otamanagerdevice": DEFAULT_DEV, "otamanagergroup": "224.1.2.8:45702", } - platform_config: List[Configuration] = [] + platform_config: list[Configuration] = [] # default mac configuration settings mac_library: Optional[str] = None mac_xml: Optional[str] = None - mac_defaults: Dict[str, str] = {} - mac_config: List[Configuration] = [] + mac_defaults: dict[str, str] = {} + mac_config: list[Configuration] = [] # default phy configuration settings, using the universal model phy_library: Optional[str] = None phy_xml: str = "emanephy.xml" - phy_defaults: Dict[str, str] = { + phy_defaults: dict[str, str] = { "subid": "1", "propagationmodel": "2ray", "noisemode": "none", } - phy_config: List[Configuration] = [] + phy_config: list[Configuration] = [] # support for external configurations - external_config: List[Configuration] = [ + external_config: list[Configuration] = [ ConfigBool(id="external", default="0"), ConfigString(id="platformendpoint", default="127.0.0.1:40001"), ConfigString(id="transportendpoint", default="127.0.0.1:50002"), ] - config_ignore: Set[str] = set() + config_ignore: set[str] = set() @classmethod def load(cls, emane_prefix: Path) -> None: @@ -94,7 +94,7 @@ class EmaneModel(WirelessModel): cls.platform_config.pop(controlport_index) @classmethod - def configurations(cls) -> List[Configuration]: + def configurations(cls) -> list[Configuration]: """ Returns the combination all all configurations (mac, phy, and external). @@ -105,7 +105,7 @@ class EmaneModel(WirelessModel): ) @classmethod - def config_groups(cls) -> List[ConfigGroup]: + def config_groups(cls) -> list[ConfigGroup]: """ Returns the defined configuration groups. @@ -122,7 +122,7 @@ class EmaneModel(WirelessModel): ConfigGroup("External Parameters", phy_len + 1, config_len), ] - def build_xml_files(self, config: Dict[str, str], iface: CoreInterface) -> None: + def build_xml_files(self, config: dict[str, str], iface: CoreInterface) -> None: """ Builds xml files for this emane model. Creates a nem.xml file that points to both mac.xml and phy.xml definitions. @@ -146,7 +146,7 @@ class EmaneModel(WirelessModel): """ logger.debug("emane model(%s) has no post setup tasks", self.name) - def update(self, moved_ifaces: List[CoreInterface]) -> None: + def update(self, moved_ifaces: list[CoreInterface]) -> None: """ Invoked from MobilityModel when nodes are moved; this causes emane location events to be generated for the nodes in the moved diff --git a/daemon/core/emane/linkmonitor.py b/daemon/core/emane/linkmonitor.py index 5ed6d49d..1997e9f8 100644 --- a/daemon/core/emane/linkmonitor.py +++ b/daemon/core/emane/linkmonitor.py @@ -2,7 +2,7 @@ import logging import sched import threading import time -from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple +from typing import TYPE_CHECKING, Optional from lxml import etree @@ -34,10 +34,10 @@ NEM_SELF: int = 65535 class LossTable: - def __init__(self, losses: Dict[float, float]) -> None: - self.losses: Dict[float, float] = losses - self.sinrs: List[float] = sorted(self.losses.keys()) - self.loss_lookup: Dict[int, float] = {} + def __init__(self, losses: dict[float, float]) -> None: + self.losses: dict[float, float] = losses + self.sinrs: list[float] = sorted(self.losses.keys()) + self.loss_lookup: dict[int, float] = {} for index, value in enumerate(self.sinrs): self.loss_lookup[index] = self.losses[value] self.mac_id: Optional[str] = None @@ -84,7 +84,7 @@ class EmaneClient: self.client: shell.ControlPortClient = shell.ControlPortClient( self.address, port ) - self.nems: Dict[int, LossTable] = {} + self.nems: dict[int, LossTable] = {} self.setup() def setup(self) -> None: @@ -110,7 +110,7 @@ class EmaneClient: self.nems[nem_id] = loss_table def check_links( - self, links: Dict[Tuple[int, int], EmaneLink], loss_threshold: int + self, links: dict[tuple[int, int], EmaneLink], loss_threshold: int ) -> None: for from_nem, loss_table in self.nems.items(): tables = self.client.getStatisticTable(loss_table.mac_id, (SINR_TABLE,)) @@ -138,11 +138,11 @@ class EmaneClient: link = EmaneLink(from_nem, to_nem, sinr) links[link_key] = link - def handle_tdma(self, config: Dict[str, Tuple]): + def handle_tdma(self, config: dict[str, tuple]): pcr = config["pcrcurveuri"][0][0] logger.debug("tdma pcr: %s", pcr) - def handle_80211(self, config: Dict[str, Tuple]) -> LossTable: + def handle_80211(self, config: dict[str, tuple]) -> LossTable: unicastrate = config["unicastrate"][0][0] pcr = config["pcrcurveuri"][0][0] logger.debug("80211 pcr: %s", pcr) @@ -159,7 +159,7 @@ class EmaneClient: losses[sinr] = por return LossTable(losses) - def handle_rfpipe(self, config: Dict[str, Tuple]) -> LossTable: + def handle_rfpipe(self, config: dict[str, tuple]) -> LossTable: pcr = config["pcrcurveuri"][0][0] logger.debug("rfpipe pcr: %s", pcr) tree = etree.parse(pcr) @@ -179,9 +179,9 @@ class EmaneClient: class EmaneLinkMonitor: def __init__(self, emane_manager: "EmaneManager") -> None: self.emane_manager: "EmaneManager" = emane_manager - self.clients: List[EmaneClient] = [] - self.links: Dict[Tuple[int, int], EmaneLink] = {} - self.complete_links: Set[Tuple[int, int]] = set() + self.clients: list[EmaneClient] = [] + self.links: dict[tuple[int, int], EmaneLink] = {} + self.complete_links: set[tuple[int, int]] = set() self.loss_threshold: Optional[int] = None self.link_interval: Optional[int] = None self.link_timeout: Optional[int] = None @@ -210,7 +210,7 @@ class EmaneLinkMonitor: if client.nems: self.clients.append(client) - def get_addresses(self) -> List[Tuple[str, int]]: + def get_addresses(self) -> list[tuple[str, int]]: addresses = [] nodes = self.emane_manager.getnodes() for node in nodes: @@ -273,25 +273,25 @@ class EmaneLinkMonitor: if self.running: self.scheduler.enter(self.link_interval, 0, self.check_links) - def get_complete_id(self, link_id: Tuple[int, int]) -> Tuple[int, int]: + def get_complete_id(self, link_id: tuple[int, int]) -> tuple[int, int]: value1, value2 = link_id if value1 < value2: return value1, value2 else: return value2, value1 - def is_complete_link(self, link_id: Tuple[int, int]) -> bool: + def is_complete_link(self, link_id: tuple[int, int]) -> bool: reverse_id = link_id[1], link_id[0] return link_id in self.links and reverse_id in self.links - def get_link_label(self, link_id: Tuple[int, int]) -> str: + def get_link_label(self, link_id: tuple[int, int]) -> str: source_id = tuple(sorted(link_id)) source_link = self.links[source_id] dest_id = link_id[::-1] dest_link = self.links[dest_id] return f"{source_link.sinr:.1f} / {dest_link.sinr:.1f}" - def send_link(self, message_type: MessageFlags, link_id: Tuple[int, int]) -> None: + def send_link(self, message_type: MessageFlags, link_id: tuple[int, int]) -> None: nem1, nem2 = link_id link = self.emane_manager.get_nem_link(nem1, nem2, message_type) if link: diff --git a/daemon/core/emane/modelmanager.py b/daemon/core/emane/modelmanager.py index 989802c4..92dd5b8e 100644 --- a/daemon/core/emane/modelmanager.py +++ b/daemon/core/emane/modelmanager.py @@ -1,7 +1,6 @@ import logging import pkgutil from pathlib import Path -from typing import Dict, List, Type from core import utils from core.emane import models as emane_models @@ -12,10 +11,10 @@ logger = logging.getLogger(__name__) class EmaneModelManager: - models: Dict[str, Type[EmaneModel]] = {} + models: dict[str, type[EmaneModel]] = {} @classmethod - def load_locals(cls, emane_prefix: Path) -> List[str]: + def load_locals(cls, emane_prefix: Path) -> list[str]: """ Load local core emane models and make them available. @@ -38,7 +37,7 @@ class EmaneModelManager: return errors @classmethod - def load(cls, path: Path, emane_prefix: Path) -> List[str]: + def load(cls, path: Path, emane_prefix: Path) -> list[str]: """ Search and load custom emane models and make them available. @@ -63,7 +62,7 @@ class EmaneModelManager: return errors @classmethod - def get(cls, name: str) -> Type[EmaneModel]: + def get(cls, name: str) -> type[EmaneModel]: model = cls.models.get(name) if model is None: raise CoreError(f"emame model does not exist {name}") diff --git a/daemon/core/emane/models/bypass.py b/daemon/core/emane/models/bypass.py index 25841114..e8f2ed39 100644 --- a/daemon/core/emane/models/bypass.py +++ b/daemon/core/emane/models/bypass.py @@ -2,7 +2,6 @@ EMANE Bypass model for CORE """ from pathlib import Path -from typing import List, Set from core.config import ConfigBool, Configuration from core.emane import emanemodel @@ -12,11 +11,11 @@ class EmaneBypassModel(emanemodel.EmaneModel): name: str = "emane_bypass" # values to ignore, when writing xml files - config_ignore: Set[str] = {"none"} + config_ignore: set[str] = {"none"} # mac definitions mac_library: str = "bypassmaclayer" - mac_config: List[Configuration] = [ + mac_config: list[Configuration] = [ ConfigBool( id="none", default="0", @@ -26,7 +25,7 @@ class EmaneBypassModel(emanemodel.EmaneModel): # phy definitions phy_library: str = "bypassphylayer" - phy_config: List[Configuration] = [] + phy_config: list[Configuration] = [] @classmethod def load(cls, emane_prefix: Path) -> None: diff --git a/daemon/core/emane/models/commeffect.py b/daemon/core/emane/models/commeffect.py index c3f0b07b..aa093a93 100644 --- a/daemon/core/emane/models/commeffect.py +++ b/daemon/core/emane/models/commeffect.py @@ -4,7 +4,6 @@ commeffect.py: EMANE CommEffect model for CORE import logging from pathlib import Path -from typing import Dict, List from lxml import etree @@ -42,12 +41,12 @@ class EmaneCommEffectModel(emanemodel.EmaneModel): name: str = "emane_commeffect" shim_library: str = "commeffectshim" shim_xml: str = "commeffectshim.xml" - shim_defaults: Dict[str, str] = {} - config_shim: List[Configuration] = [] + shim_defaults: dict[str, str] = {} + config_shim: list[Configuration] = [] # comm effect does not need the default phy and external configurations - phy_config: List[Configuration] = [] - external_config: List[Configuration] = [] + phy_config: list[Configuration] = [] + external_config: list[Configuration] = [] @classmethod def load(cls, emane_prefix: Path) -> None: @@ -56,11 +55,11 @@ class EmaneCommEffectModel(emanemodel.EmaneModel): cls.config_shim = emanemanifest.parse(shim_xml_path, cls.shim_defaults) @classmethod - def configurations(cls) -> List[Configuration]: + def configurations(cls) -> list[Configuration]: return cls.platform_config + cls.config_shim @classmethod - def config_groups(cls) -> List[ConfigGroup]: + def config_groups(cls) -> list[ConfigGroup]: platform_len = len(cls.platform_config) return [ ConfigGroup("Platform Parameters", 1, platform_len), @@ -71,7 +70,7 @@ class EmaneCommEffectModel(emanemodel.EmaneModel): ), ] - def build_xml_files(self, config: Dict[str, str], iface: CoreInterface) -> None: + def build_xml_files(self, config: dict[str, str], iface: CoreInterface) -> None: """ Build the necessary nem and commeffect XMLs in the given path. If an individual NEM has a nonstandard config, we need to build diff --git a/daemon/core/emane/models/tdma.py b/daemon/core/emane/models/tdma.py index c6ac631b..100e960d 100644 --- a/daemon/core/emane/models/tdma.py +++ b/daemon/core/emane/models/tdma.py @@ -4,7 +4,6 @@ tdma.py: EMANE TDMA model bindings for CORE import logging from pathlib import Path -from typing import Set from core import constants, utils from core.config import ConfigString @@ -28,7 +27,7 @@ class EmaneTdmaModel(emanemodel.EmaneModel): default_schedule: Path = ( constants.CORE_DATA_DIR / "examples" / "tdma" / "schedule.xml" ) - config_ignore: Set[str] = {schedule_name} + config_ignore: set[str] = {schedule_name} @classmethod def load(cls, emane_prefix: Path) -> None: diff --git a/daemon/core/emane/nodes.py b/daemon/core/emane/nodes.py index 7264fe88..ecf684d7 100644 --- a/daemon/core/emane/nodes.py +++ b/daemon/core/emane/nodes.py @@ -6,7 +6,7 @@ share the same MAC+PHY model. import logging import time from dataclasses import dataclass -from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Type, Union +from typing import TYPE_CHECKING, Callable, Optional, Union from core.emulator.data import InterfaceData, LinkData, LinkOptions from core.emulator.distributed import DistributedServer @@ -196,7 +196,7 @@ class EmaneNet(CoreNetworkBase): def unlink(self, iface1: CoreInterface, iface2: CoreInterface) -> None: pass - def updatemodel(self, config: Dict[str, str]) -> None: + def updatemodel(self, config: dict[str, str]) -> None: """ Update configuration for the current model. @@ -212,8 +212,8 @@ class EmaneNet(CoreNetworkBase): def setmodel( self, - model: Union[Type["EmaneModel"], Type["WayPointMobility"]], - config: Dict[str, str], + model: Union[type["EmaneModel"], type["WayPointMobility"]], + config: dict[str, str], ) -> None: """ set the EmaneModel associated with this node @@ -225,7 +225,7 @@ class EmaneNet(CoreNetworkBase): self.mobility = model(session=self.session, _id=self.id) self.mobility.update_config(config) - def links(self, flags: MessageFlags = MessageFlags.NONE) -> List[LinkData]: + def links(self, flags: MessageFlags = MessageFlags.NONE) -> list[LinkData]: links = [] emane_manager = self.session.emane # gather current emane links