From 69f05a6712f2df014e823e6e32f4dc80cb43cf3c Mon Sep 17 00:00:00 2001 From: Blake Harnden <32446120+bharnden@users.noreply.github.com> Date: Thu, 13 Apr 2023 15:48:02 -0700 Subject: [PATCH] daemon: updated top level core modules from using deprecated type hinting --- daemon/core/config.py | 36 +++++++++++------------ daemon/core/executables.py | 8 ++---- daemon/core/player.py | 4 +-- daemon/core/utils.py | 58 +++++++++++++++----------------------- 4 files changed, 46 insertions(+), 60 deletions(-) diff --git a/daemon/core/config.py b/daemon/core/config.py index ae40627e..7a6ffa49 100644 --- a/daemon/core/config.py +++ b/daemon/core/config.py @@ -5,7 +5,7 @@ Common support for configurable CORE objects. import logging from collections import OrderedDict from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Type, Union +from typing import TYPE_CHECKING, Any, Optional, Union from core.emane.nodes import EmaneNet from core.emulator.enumerations import ConfigDataTypes @@ -17,9 +17,9 @@ logger = logging.getLogger(__name__) if TYPE_CHECKING: from core.location.mobility import WirelessModel - WirelessModelType = Type[WirelessModel] + WirelessModelType = type[WirelessModel] -_BOOL_OPTIONS: Set[str] = {"0", "1"} +_BOOL_OPTIONS: set[str] = {"0", "1"} @dataclass @@ -43,7 +43,7 @@ class Configuration: type: ConfigDataTypes label: str = None default: str = "" - options: List[str] = field(default_factory=list) + options: list[str] = field(default_factory=list) group: str = "Configuration" def __post_init__(self) -> None: @@ -118,10 +118,10 @@ class ConfigurableOptions: """ name: Optional[str] = None - options: List[Configuration] = [] + options: list[Configuration] = [] @classmethod - def configurations(cls) -> List[Configuration]: + def configurations(cls) -> list[Configuration]: """ Provides the configurations for this class. @@ -130,7 +130,7 @@ class ConfigurableOptions: return cls.options @classmethod - def config_groups(cls) -> List[ConfigGroup]: + def config_groups(cls) -> list[ConfigGroup]: """ Defines how configurations are grouped. @@ -139,7 +139,7 @@ class ConfigurableOptions: return [ConfigGroup("Options", 1, len(cls.configurations()))] @classmethod - def default_values(cls) -> Dict[str, str]: + def default_values(cls) -> dict[str, str]: """ Provides an ordered mapping of configuration keys to default values. @@ -165,7 +165,7 @@ class ConfigurableManager: """ self.node_configurations = {} - def nodes(self) -> List[int]: + def nodes(self) -> list[int]: """ Retrieves the ids of all node configurations known by this manager. @@ -208,7 +208,7 @@ class ConfigurableManager: def set_configs( self, - config: Dict[str, str], + config: dict[str, str], node_id: int = _default_node, config_type: str = _default_type, ) -> None: @@ -250,7 +250,7 @@ class ConfigurableManager: def get_configs( self, node_id: int = _default_node, config_type: str = _default_type - ) -> Optional[Dict[str, str]]: + ) -> Optional[dict[str, str]]: """ Retrieve configurations for a node and configuration type. @@ -264,7 +264,7 @@ class ConfigurableManager: result = node_configs.get(config_type) return result - def get_all_configs(self, node_id: int = _default_node) -> Dict[str, Any]: + def get_all_configs(self, node_id: int = _default_node) -> dict[str, Any]: """ Retrieve all current configuration types for a node. @@ -284,11 +284,11 @@ class ModelManager(ConfigurableManager): Creates a ModelManager object. """ super().__init__() - self.models: Dict[str, Any] = {} - self.node_models: Dict[int, str] = {} + self.models: dict[str, Any] = {} + self.node_models: dict[int, str] = {} def set_model_config( - self, node_id: int, model_name: str, config: Dict[str, str] = None + self, node_id: int, model_name: str, config: dict[str, str] = None ) -> None: """ Set configuration data for a model. @@ -317,7 +317,7 @@ class ModelManager(ConfigurableManager): # set configuration self.set_configs(model_config, node_id=node_id, config_type=model_name) - def get_model_config(self, node_id: int, model_name: str) -> Dict[str, str]: + def get_model_config(self, node_id: int, model_name: str) -> dict[str, str]: """ Retrieve configuration data for a model. @@ -342,7 +342,7 @@ class ModelManager(ConfigurableManager): self, node: Union[WlanNode, EmaneNet], model_class: "WirelessModelType", - config: Dict[str, str] = None, + config: dict[str, str] = None, ) -> None: """ Set model and model configuration for node. @@ -361,7 +361,7 @@ class ModelManager(ConfigurableManager): def get_models( self, node: Union[WlanNode, EmaneNet] - ) -> List[Tuple[Type, Dict[str, str]]]: + ) -> list[tuple[type, dict[str, str]]]: """ Return a list of model classes and values for a net if one has been configured. This is invoked when exporting a session to XML. diff --git a/daemon/core/executables.py b/daemon/core/executables.py index 95c97378..f04d88de 100644 --- a/daemon/core/executables.py +++ b/daemon/core/executables.py @@ -1,5 +1,3 @@ -from typing import List - BASH: str = "bash" ETHTOOL: str = "ethtool" IP: str = "ip" @@ -13,7 +11,7 @@ UMOUNT: str = "umount" VCMD: str = "vcmd" VNODED: str = "vnoded" -COMMON_REQUIREMENTS: List[str] = [ +COMMON_REQUIREMENTS: list[str] = [ BASH, ETHTOOL, IP, @@ -26,10 +24,10 @@ COMMON_REQUIREMENTS: List[str] = [ VCMD, VNODED, ] -OVS_REQUIREMENTS: List[str] = [OVS_VSCTL] +OVS_REQUIREMENTS: list[str] = [OVS_VSCTL] -def get_requirements(use_ovs: bool) -> List[str]: +def get_requirements(use_ovs: bool) -> list[str]: """ Retrieve executable requirements needed to run CORE. diff --git a/daemon/core/player.py b/daemon/core/player.py index 6ba0d602..d06e7b97 100644 --- a/daemon/core/player.py +++ b/daemon/core/player.py @@ -5,7 +5,7 @@ import logging import sched from pathlib import Path from threading import Thread -from typing import IO, Callable, Dict, Optional +from typing import IO, Callable, Optional import grpc @@ -230,7 +230,7 @@ class CorePlayer: self.node_streamer: Optional[MoveNodesStreamer] = None self.node_streamer_thread: Optional[Thread] = None self.scheduler: sched.scheduler = sched.scheduler() - self.handlers: Dict[PlayerEvents, Callable] = { + self.handlers: dict[PlayerEvents, Callable] = { PlayerEvents.XY: self.handle_xy, PlayerEvents.GEO: self.handle_geo, PlayerEvents.CMD: self.handle_cmd, diff --git a/daemon/core/utils.py b/daemon/core/utils.py index 244590f8..5947c006 100644 --- a/daemon/core/utils.py +++ b/daemon/core/utils.py @@ -17,23 +17,11 @@ import shutil import sys import threading from collections import OrderedDict +from collections.abc import Iterable from pathlib import Path from queue import Queue from subprocess import PIPE, STDOUT, Popen -from typing import ( - TYPE_CHECKING, - Any, - Callable, - Dict, - Generic, - Iterable, - List, - Optional, - Tuple, - Type, - TypeVar, - Union, -) +from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union import netaddr @@ -70,7 +58,7 @@ def execute_script(coreemu: "CoreEmu", file_path: Path, args: str) -> None: def execute_file( - path: Path, exec_globals: Dict[str, str] = None, exec_locals: Dict[str, str] = None + path: Path, exec_globals: dict[str, str] = None, exec_locals: dict[str, str] = None ) -> None: """ Provides a way to execute a file. @@ -131,7 +119,7 @@ def _valid_module(path: Path) -> bool: return True -def _is_class(module: Any, member: Type, clazz: Type) -> bool: +def _is_class(module: Any, member: type, clazz: type) -> bool: """ Validates if a module member is a class and an instance of a CoreService. @@ -175,7 +163,7 @@ def which(command: str, required: bool) -> str: return found_path -def make_tuple_fromstr(s: str, value_type: Callable[[str], T]) -> Tuple[T]: +def make_tuple_fromstr(s: str, value_type: Callable[[str], T]) -> tuple[T]: """ Create a tuple from a string. @@ -193,7 +181,7 @@ def make_tuple_fromstr(s: str, value_type: Callable[[str], T]) -> Tuple[T]: return tuple(value_type(i) for i in values) -def mute_detach(args: str, **kwargs: Dict[str, Any]) -> int: +def mute_detach(args: str, **kwargs: dict[str, Any]) -> int: """ Run a muted detached process by forking it. @@ -210,7 +198,7 @@ def mute_detach(args: str, **kwargs: Dict[str, Any]) -> int: def cmd( args: str, - env: Dict[str, str] = None, + env: dict[str, str] = None, cwd: Path = None, wait: bool = True, shell: bool = False, @@ -249,7 +237,7 @@ def cmd( raise CoreCommandError(1, input_args, "", e.strerror) -def run_cmds(args: List[str], wait: bool = True, shell: bool = False) -> List[str]: +def run_cmds(args: list[str], wait: bool = True, shell: bool = False) -> list[str]: """ Execute a series of commands on the host and returns a list of the combined stderr stdout output. @@ -294,7 +282,7 @@ def file_demunge(pathname: str, header: str) -> None: :param header: header text to target for removal :return: nothing """ - with open(pathname, "r") as read_file: + with open(pathname) as read_file: lines = read_file.readlines() start = None @@ -348,7 +336,7 @@ def sysctl_devname(devname: str) -> Optional[str]: return devname.replace(".", "/") -def load_config(file_path: Path, d: Dict[str, str]) -> None: +def load_config(file_path: Path, d: dict[str, str]) -> None: """ Read key=value pairs from a file, into a dict. Skip comments; strip newline characters and spacing. @@ -369,7 +357,7 @@ def load_config(file_path: Path, d: Dict[str, str]) -> None: logger.exception("error reading file to dict: %s", file_path) -def load_module(import_statement: str, clazz: Generic[T]) -> List[T]: +def load_module(import_statement: str, clazz: Generic[T]) -> list[T]: classes = [] try: module = importlib.import_module(import_statement) @@ -384,7 +372,7 @@ def load_module(import_statement: str, clazz: Generic[T]) -> List[T]: return classes -def load_classes(path: Path, clazz: Generic[T]) -> List[T]: +def load_classes(path: Path, clazz: Generic[T]) -> list[T]: """ Dynamically load classes for use within CORE. @@ -426,12 +414,12 @@ def load_logging_config(config_path: Path) -> None: def run_cmds_threaded( - nodes: List["CoreNode"], - cmds: List[str], + nodes: list["CoreNode"], + cmds: list[str], wait: bool = True, shell: bool = False, workers: int = None, -) -> Tuple[Dict[int, List[str]], List[Exception]]: +) -> tuple[dict[int, list[str]], list[Exception]]: """ Run a set of commands in order across a provided set of nodes. Each node will run the commands within the context of a threadpool. @@ -446,8 +434,8 @@ def run_cmds_threaded( """ def _node_cmds( - _target: "CoreNode", _cmds: List[str], _wait: bool, _shell: bool - ) -> List[str]: + _target: "CoreNode", _cmds: list[str], _wait: bool, _shell: bool + ) -> list[str]: outputs = [] for _cmd in _cmds: output = _target.cmd(_cmd, wait=_wait, shell=_shell) @@ -475,12 +463,12 @@ def run_cmds_threaded( def run_cmds_mp( - nodes: List["CoreNode"], - cmds: List[str], + nodes: list["CoreNode"], + cmds: list[str], wait: bool = True, shell: bool = False, workers: int = None, -) -> Tuple[Dict[int, List[str]], List[Exception]]: +) -> tuple[dict[int, list[str]], list[Exception]]: """ Run a set of commands in order across a provided set of nodes. Each node will run the commands within the context of a process pool. This will not work @@ -521,8 +509,8 @@ def run_cmds_mp( def threadpool( - funcs: List[Tuple[Callable, Iterable[Any], Dict[Any, Any]]], workers: int = 10 -) -> Tuple[List[Any], List[Exception]]: + funcs: list[tuple[Callable, Iterable[Any], dict[Any, Any]]], workers: int = 10 +) -> tuple[list[Any], list[Exception]]: """ Run provided functions, arguments, and keywords within a threadpool collecting results and exceptions. @@ -575,7 +563,7 @@ def iface_config_id(node_id: int, iface_id: int = None) -> int: return node_id -def parse_iface_config_id(config_id: int) -> Tuple[int, Optional[int]]: +def parse_iface_config_id(config_id: int) -> tuple[int, Optional[int]]: """ Parses configuration id, that may be potentially derived from an interface for a node.