daemon: updated core.scripts and core.xml to avoid using deprecated type hinting
This commit is contained in:
parent
7ea950f8ec
commit
f9505b3173
5 changed files with 26 additions and 27 deletions
|
@ -8,7 +8,7 @@ from argparse import (
|
|||
)
|
||||
from functools import wraps
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional, Tuple
|
||||
from typing import Any, Optional
|
||||
|
||||
import grpc
|
||||
import netaddr
|
||||
|
@ -30,7 +30,7 @@ from core.api.grpc.wrappers import (
|
|||
NODE_TYPES = [x.name for x in NodeType if x != NodeType.PEER_TO_PEER]
|
||||
|
||||
|
||||
def protobuf_to_json(message: Any) -> Dict[str, Any]:
|
||||
def protobuf_to_json(message: Any) -> dict[str, Any]:
|
||||
return MessageToDict(
|
||||
message, including_default_value_fields=True, preserving_proto_field_name=True
|
||||
)
|
||||
|
@ -82,7 +82,7 @@ def ip6_type(value: str) -> IPNetwork:
|
|||
raise ArgumentTypeError(f"invalid ip6 address: {value}")
|
||||
|
||||
|
||||
def position_type(value: str) -> Tuple[float, float]:
|
||||
def position_type(value: str) -> tuple[float, float]:
|
||||
error = "invalid position, must be in the format: float,float"
|
||||
try:
|
||||
values = [float(x) for x in value.split(",")]
|
||||
|
@ -94,7 +94,7 @@ def position_type(value: str) -> Tuple[float, float]:
|
|||
return x, y
|
||||
|
||||
|
||||
def geo_type(value: str) -> Tuple[float, float, float]:
|
||||
def geo_type(value: str) -> tuple[float, float, float]:
|
||||
error = "invalid geo, must be in the format: float,float,float"
|
||||
try:
|
||||
values = [float(x) for x in value.split(",")]
|
||||
|
|
|
@ -9,7 +9,6 @@ from argparse import ArgumentDefaultsHelpFormatter
|
|||
from functools import cmp_to_key
|
||||
from queue import Queue
|
||||
from threading import Thread
|
||||
from typing import Dict, Tuple
|
||||
|
||||
import grpc
|
||||
|
||||
|
@ -31,7 +30,7 @@ class RouteEnum(enum.Enum):
|
|||
|
||||
|
||||
class SdtClient:
|
||||
def __init__(self, address: Tuple[str, int]) -> None:
|
||||
def __init__(self, address: tuple[str, int]) -> None:
|
||||
self.sock = socket.create_connection(address)
|
||||
self.links = []
|
||||
self.send(f'layer "{ROUTE_LAYER}"')
|
||||
|
@ -85,7 +84,7 @@ class RouterMonitor:
|
|||
self.sdt = SdtClient((sdt_host, sdt_port))
|
||||
self.nodes = self.get_nodes()
|
||||
|
||||
def get_nodes(self) -> Dict[int, str]:
|
||||
def get_nodes(self) -> dict[int, str]:
|
||||
with self.core.context_connect():
|
||||
if self.session is None:
|
||||
self.session = self.get_session()
|
||||
|
@ -146,7 +145,7 @@ class RouterMonitor:
|
|||
self.manage_routes()
|
||||
self.route_time = time.monotonic()
|
||||
|
||||
def route_sort(self, x: Tuple[str, int], y: Tuple[str, int]) -> int:
|
||||
def route_sort(self, x: tuple[str, int], y: tuple[str, int]) -> int:
|
||||
x_node = x[0]
|
||||
y_node = y[0]
|
||||
if x_node == self.src_id:
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
import logging
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, Dict, Generic, Optional, Type, TypeVar
|
||||
from typing import TYPE_CHECKING, Any, Generic, Optional, TypeVar
|
||||
|
||||
from lxml import etree
|
||||
|
||||
|
@ -26,7 +26,7 @@ if TYPE_CHECKING:
|
|||
from core.emane.emanemodel import EmaneModel
|
||||
from core.emulator.session import Session
|
||||
|
||||
EmaneModelType = Type[EmaneModel]
|
||||
EmaneModelType = type[EmaneModel]
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
|
@ -86,7 +86,7 @@ def create_iface_data(iface_element: etree.Element) -> InterfaceData:
|
|||
def create_emane_model_config(
|
||||
node_id: int,
|
||||
model: "EmaneModelType",
|
||||
config: Dict[str, str],
|
||||
config: dict[str, str],
|
||||
iface_id: Optional[int],
|
||||
) -> etree.Element:
|
||||
emane_element = etree.Element("emane_configuration")
|
||||
|
@ -148,8 +148,8 @@ class NodeElement:
|
|||
|
||||
|
||||
class ServiceElement:
|
||||
def __init__(self, service: Type[CoreService]) -> None:
|
||||
self.service: Type[CoreService] = service
|
||||
def __init__(self, service: type[CoreService]) -> None:
|
||||
self.service: type[CoreService] = service
|
||||
self.element: etree.Element = etree.Element("service")
|
||||
add_attribute(self.element, "name", service.name)
|
||||
self.add_directories()
|
||||
|
@ -266,7 +266,7 @@ class NetworkElement(NodeElement):
|
|||
node_type = self.session.get_node_type(type(self.node))
|
||||
add_attribute(self.element, "type", node_type.name)
|
||||
|
||||
def add_wireless_config(self, config: Dict[str, Configuration]) -> None:
|
||||
def add_wireless_config(self, config: dict[str, Configuration]) -> None:
|
||||
wireless_element = etree.SubElement(self.element, "wireless")
|
||||
for config_item in config.values():
|
||||
add_configuration(wireless_element, config_item.id, config_item.default)
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
import os
|
||||
import socket
|
||||
from typing import TYPE_CHECKING, List, Tuple
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import netaddr
|
||||
from lxml import etree
|
||||
|
@ -78,7 +78,7 @@ def get_address_type(address: str) -> str:
|
|||
return address_type
|
||||
|
||||
|
||||
def get_ipv4_addresses(hostname: str) -> List[Tuple[str, str]]:
|
||||
def get_ipv4_addresses(hostname: str) -> list[tuple[str, str]]:
|
||||
if hostname == "localhost":
|
||||
addresses = []
|
||||
args = f"{IP} -o -f inet address show"
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
import logging
|
||||
from pathlib import Path
|
||||
from tempfile import NamedTemporaryFile
|
||||
from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple
|
||||
from typing import TYPE_CHECKING, Optional
|
||||
|
||||
from lxml import etree
|
||||
|
||||
|
@ -22,7 +22,7 @@ if TYPE_CHECKING:
|
|||
_MAC_PREFIX = "02:02"
|
||||
|
||||
|
||||
def is_external(config: Dict[str, str]) -> bool:
|
||||
def is_external(config: dict[str, str]) -> bool:
|
||||
"""
|
||||
Checks if the configuration is for an external transport.
|
||||
|
||||
|
@ -32,7 +32,7 @@ def is_external(config: Dict[str, str]) -> bool:
|
|||
return config.get("external") == "1"
|
||||
|
||||
|
||||
def _value_to_params(value: str) -> Optional[Tuple[str]]:
|
||||
def _value_to_params(value: str) -> Optional[tuple[str]]:
|
||||
"""
|
||||
Helper to convert a parameter to a parameter tuple.
|
||||
|
||||
|
@ -113,9 +113,9 @@ def add_param(xml_element: etree.Element, name: str, value: str) -> None:
|
|||
|
||||
def add_configurations(
|
||||
xml_element: etree.Element,
|
||||
configurations: List[Configuration],
|
||||
config: Dict[str, str],
|
||||
config_ignore: Set,
|
||||
configurations: list[Configuration],
|
||||
config: dict[str, str],
|
||||
config_ignore: set[str],
|
||||
) -> None:
|
||||
"""
|
||||
Add emane model configurations to xml element.
|
||||
|
@ -148,7 +148,7 @@ def build_platform_xml(
|
|||
nem_port: int,
|
||||
emane_net: EmaneNet,
|
||||
iface: CoreInterface,
|
||||
config: Dict[str, str],
|
||||
config: dict[str, str],
|
||||
) -> None:
|
||||
"""
|
||||
Create platform xml for a nem/interface.
|
||||
|
@ -209,7 +209,7 @@ def build_platform_xml(
|
|||
create_node_file(iface.node, platform_element, doc_name, file_name)
|
||||
|
||||
|
||||
def create_transport_xml(iface: CoreInterface, config: Dict[str, str]) -> None:
|
||||
def create_transport_xml(iface: CoreInterface, config: dict[str, str]) -> None:
|
||||
"""
|
||||
Build transport xml file for node and transport type.
|
||||
|
||||
|
@ -240,7 +240,7 @@ def create_transport_xml(iface: CoreInterface, config: Dict[str, str]) -> None:
|
|||
|
||||
|
||||
def create_phy_xml(
|
||||
emane_model: "EmaneModel", iface: CoreInterface, config: Dict[str, str]
|
||||
emane_model: "EmaneModel", iface: CoreInterface, config: dict[str, str]
|
||||
) -> None:
|
||||
"""
|
||||
Create the phy xml document.
|
||||
|
@ -261,7 +261,7 @@ def create_phy_xml(
|
|||
|
||||
|
||||
def create_mac_xml(
|
||||
emane_model: "EmaneModel", iface: CoreInterface, config: Dict[str, str]
|
||||
emane_model: "EmaneModel", iface: CoreInterface, config: dict[str, str]
|
||||
) -> None:
|
||||
"""
|
||||
Create the mac xml document.
|
||||
|
@ -284,7 +284,7 @@ def create_mac_xml(
|
|||
|
||||
|
||||
def create_nem_xml(
|
||||
emane_model: "EmaneModel", iface: CoreInterface, config: Dict[str, str]
|
||||
emane_model: "EmaneModel", iface: CoreInterface, config: dict[str, str]
|
||||
) -> None:
|
||||
"""
|
||||
Create the nem xml document.
|
||||
|
|
Loading…
Reference in a new issue