daemon: update deprecated typing for core.api

This commit is contained in:
Blake Harnden 2023-04-13 11:22:37 -07:00
parent 6ff2abf0b8
commit e770bcd47c
5 changed files with 113 additions and 109 deletions

View file

@ -4,10 +4,11 @@ gRpc client for interfacing with CORE.
import logging import logging
import threading import threading
from collections.abc import Callable, Generator, Iterable
from contextlib import contextmanager from contextlib import contextmanager
from pathlib import Path from pathlib import Path
from queue import Queue from queue import Queue
from typing import Any, Callable, Dict, Generator, Iterable, List, Optional, Tuple from typing import Any, Optional
import grpc import grpc
@ -235,7 +236,7 @@ class CoreGrpcClient:
def start_session( def start_session(
self, session: wrappers.Session, definition: bool = False self, session: wrappers.Session, definition: bool = False
) -> Tuple[bool, List[str]]: ) -> tuple[bool, list[str]]:
""" """
Start a session. Start a session.
@ -285,7 +286,7 @@ class CoreGrpcClient:
response = self.stub.DeleteSession(request) response = self.stub.DeleteSession(request)
return response.result return response.result
def get_sessions(self) -> List[wrappers.SessionSummary]: def get_sessions(self) -> list[wrappers.SessionSummary]:
""" """
Retrieves all currently known sessions. Retrieves all currently known sessions.
@ -354,7 +355,7 @@ class CoreGrpcClient:
self, self,
session_id: int, session_id: int,
handler: Callable[[wrappers.Event], None], handler: Callable[[wrappers.Event], None],
events: List[wrappers.EventType] = None, events: list[wrappers.EventType] = None,
) -> grpc.Future: ) -> grpc.Future:
""" """
Listen for session events. Listen for session events.
@ -428,7 +429,7 @@ class CoreGrpcClient:
def get_node( def get_node(
self, session_id: int, node_id: int self, session_id: int, node_id: int
) -> Tuple[wrappers.Node, List[wrappers.Interface], List[wrappers.Link]]: ) -> tuple[wrappers.Node, list[wrappers.Interface], list[wrappers.Link]]:
""" """
Get node details. Get node details.
@ -536,7 +537,7 @@ class CoreGrpcClient:
command: str, command: str,
wait: bool = True, wait: bool = True,
shell: bool = False, shell: bool = False,
) -> Tuple[int, str]: ) -> tuple[int, str]:
""" """
Send command to a node and get the output. Send command to a node and get the output.
@ -575,7 +576,7 @@ class CoreGrpcClient:
def add_link( def add_link(
self, session_id: int, link: wrappers.Link, source: str = None self, session_id: int, link: wrappers.Link, source: str = None
) -> Tuple[bool, wrappers.Interface, wrappers.Interface]: ) -> tuple[bool, wrappers.Interface, wrappers.Interface]:
""" """
Add a link between nodes. Add a link between nodes.
@ -646,7 +647,7 @@ class CoreGrpcClient:
def get_mobility_config( def get_mobility_config(
self, session_id: int, node_id: int self, session_id: int, node_id: int
) -> Dict[str, wrappers.ConfigOption]: ) -> dict[str, wrappers.ConfigOption]:
""" """
Get mobility configuration for a node. Get mobility configuration for a node.
@ -660,7 +661,7 @@ class CoreGrpcClient:
return wrappers.ConfigOption.from_dict(response.config) return wrappers.ConfigOption.from_dict(response.config)
def set_mobility_config( def set_mobility_config(
self, session_id: int, node_id: int, config: Dict[str, str] self, session_id: int, node_id: int, config: dict[str, str]
) -> bool: ) -> bool:
""" """
Set mobility configuration for a node. Set mobility configuration for a node.
@ -706,7 +707,7 @@ class CoreGrpcClient:
response = self.stub.GetConfig(request) response = self.stub.GetConfig(request)
return wrappers.CoreConfig.from_proto(response) return wrappers.CoreConfig.from_proto(response)
def get_service_defaults(self, session_id: int) -> List[wrappers.ServiceDefault]: def get_service_defaults(self, session_id: int) -> list[wrappers.ServiceDefault]:
""" """
Get default services for different default node models. Get default services for different default node models.
@ -723,7 +724,7 @@ class CoreGrpcClient:
return defaults return defaults
def set_service_defaults( def set_service_defaults(
self, session_id: int, service_defaults: Dict[str, List[str]] self, session_id: int, service_defaults: dict[str, list[str]]
) -> bool: ) -> bool:
""" """
Set default services for node models. Set default services for node models.
@ -829,7 +830,7 @@ class CoreGrpcClient:
def get_wlan_config( def get_wlan_config(
self, session_id: int, node_id: int self, session_id: int, node_id: int
) -> Dict[str, wrappers.ConfigOption]: ) -> dict[str, wrappers.ConfigOption]:
""" """
Get wlan configuration for a node. Get wlan configuration for a node.
@ -843,7 +844,7 @@ class CoreGrpcClient:
return wrappers.ConfigOption.from_dict(response.config) return wrappers.ConfigOption.from_dict(response.config)
def set_wlan_config( def set_wlan_config(
self, session_id: int, node_id: int, config: Dict[str, str] self, session_id: int, node_id: int, config: dict[str, str]
) -> bool: ) -> bool:
""" """
Set wlan configuration for a node. Set wlan configuration for a node.
@ -861,7 +862,7 @@ class CoreGrpcClient:
def get_emane_model_config( def get_emane_model_config(
self, session_id: int, node_id: int, model: str, iface_id: int = -1 self, session_id: int, node_id: int, model: str, iface_id: int = -1
) -> Dict[str, wrappers.ConfigOption]: ) -> dict[str, wrappers.ConfigOption]:
""" """
Get emane model configuration for a node or a node's interface. Get emane model configuration for a node or a node's interface.
@ -909,7 +910,7 @@ class CoreGrpcClient:
with open(file_path, "w") as xml_file: with open(file_path, "w") as xml_file:
xml_file.write(response.data) xml_file.write(response.data)
def open_xml(self, file_path: Path, start: bool = False) -> Tuple[bool, int]: def open_xml(self, file_path: Path, start: bool = False) -> tuple[bool, int]:
""" """
Load a local scenario XML file to open as a new session. Load a local scenario XML file to open as a new session.
@ -940,7 +941,7 @@ class CoreGrpcClient:
response = self.stub.EmaneLink(request) response = self.stub.EmaneLink(request)
return response.result return response.result
def get_ifaces(self) -> List[str]: def get_ifaces(self) -> list[str]:
""" """
Retrieves a list of interfaces available on the host machine that are not Retrieves a list of interfaces available on the host machine that are not
a part of a CORE session. a part of a CORE session.
@ -964,7 +965,7 @@ class CoreGrpcClient:
def get_node_config_service( def get_node_config_service(
self, session_id: int, node_id: int, name: str self, session_id: int, node_id: int, name: str
) -> Dict[str, str]: ) -> dict[str, str]:
""" """
Retrieves information for a specific config service on a node. Retrieves information for a specific config service on a node.
@ -982,7 +983,7 @@ class CoreGrpcClient:
def get_config_service_rendered( def get_config_service_rendered(
self, session_id: int, node_id: int, name: str self, session_id: int, node_id: int, name: str
) -> Dict[str, str]: ) -> dict[str, str]:
""" """
Retrieve the rendered config service files for a node. Retrieve the rendered config service files for a node.
@ -1129,7 +1130,7 @@ class CoreGrpcClient:
def get_wireless_config( def get_wireless_config(
self, session_id: int, node_id: int self, session_id: int, node_id: int
) -> Dict[str, wrappers.ConfigOption]: ) -> dict[str, wrappers.ConfigOption]:
request = GetWirelessConfigRequest(session_id=session_id, node_id=node_id) request = GetWirelessConfigRequest(session_id=session_id, node_id=node_id)
response = self.stub.GetWirelessConfig(request) response = self.stub.GetWirelessConfig(request)
return wrappers.ConfigOption.from_dict(response.config) return wrappers.ConfigOption.from_dict(response.config)
@ -1156,7 +1157,7 @@ class CoreGrpcClient:
self.channel = None self.channel = None
@contextmanager @contextmanager
def context_connect(self) -> Generator: def context_connect(self) -> Generator[None, None, None]:
""" """
Makes a context manager based connection to the server, will close after Makes a context manager based connection to the server, will close after
context ends. context ends.

View file

@ -1,6 +1,7 @@
import logging import logging
from collections.abc import Iterable
from queue import Empty, Queue from queue import Empty, Queue
from typing import Iterable, Optional from typing import Optional
from core.api.grpc import core_pb2, grpcutils from core.api.grpc import core_pb2, grpcutils
from core.api.grpc.grpcutils import convert_link_data from core.api.grpc.grpcutils import convert_link_data

View file

@ -1,7 +1,7 @@
import logging import logging
import time import time
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Type, Union from typing import Any, Optional, Union
import grpc import grpc
from grpc import ServicerContext from grpc import ServicerContext
@ -63,8 +63,8 @@ class CpuUsage:
def add_node_data( def add_node_data(
_class: Type[NodeBase], node_proto: core_pb2.Node _class: type[NodeBase], node_proto: core_pb2.Node
) -> Tuple[Position, NodeOptions]: ) -> tuple[Position, NodeOptions]:
""" """
Convert node protobuf message to data for creating a node. Convert node protobuf message to data for creating a node.
@ -118,7 +118,7 @@ def link_iface(iface_proto: core_pb2.Interface) -> InterfaceData:
def add_link_data( def add_link_data(
link_proto: core_pb2.Link, link_proto: core_pb2.Link,
) -> Tuple[InterfaceData, InterfaceData, LinkOptions]: ) -> tuple[InterfaceData, InterfaceData, LinkOptions]:
""" """
Convert link proto to link interfaces and options data. Convert link proto to link interfaces and options data.
@ -145,8 +145,8 @@ def add_link_data(
def create_nodes( def create_nodes(
session: Session, node_protos: List[core_pb2.Node] session: Session, node_protos: list[core_pb2.Node]
) -> Tuple[List[NodeBase], List[Exception]]: ) -> tuple[list[NodeBase], list[Exception]]:
""" """
Create nodes using a thread pool and wait for completion. Create nodes using a thread pool and wait for completion.
@ -176,8 +176,8 @@ def create_nodes(
def create_links( def create_links(
session: Session, link_protos: List[core_pb2.Link] session: Session, link_protos: list[core_pb2.Link]
) -> Tuple[List[NodeBase], List[Exception]]: ) -> tuple[list[NodeBase], list[Exception]]:
""" """
Create links using a thread pool and wait for completion. Create links using a thread pool and wait for completion.
@ -200,8 +200,8 @@ def create_links(
def edit_links( def edit_links(
session: Session, link_protos: List[core_pb2.Link] session: Session, link_protos: list[core_pb2.Link]
) -> Tuple[List[None], List[Exception]]: ) -> tuple[list[None], list[Exception]]:
""" """
Edit links using a thread pool and wait for completion. Edit links using a thread pool and wait for completion.
@ -235,7 +235,7 @@ def convert_value(value: Any) -> str:
return value return value
def convert_session_options(session: Session) -> Dict[str, common_pb2.ConfigOption]: def convert_session_options(session: Session) -> dict[str, common_pb2.ConfigOption]:
config_options = {} config_options = {}
for option in session.options.options: for option in session.options.options:
value = session.options.get(option.id) value = session.options.get(option.id)
@ -252,9 +252,9 @@ def convert_session_options(session: Session) -> Dict[str, common_pb2.ConfigOpti
def get_config_options( def get_config_options(
config: Dict[str, str], config: dict[str, str],
configurable_options: Union[ConfigurableOptions, Type[ConfigurableOptions]], configurable_options: Union[ConfigurableOptions, type[ConfigurableOptions]],
) -> Dict[str, common_pb2.ConfigOption]: ) -> dict[str, common_pb2.ConfigOption]:
""" """
Retrieve configuration options in a form that is used by the grpc server. Retrieve configuration options in a form that is used by the grpc server.
@ -283,7 +283,7 @@ def get_config_options(
def get_node_proto( def get_node_proto(
session: Session, node: NodeBase, emane_configs: List[NodeEmaneConfig] session: Session, node: NodeBase, emane_configs: list[NodeEmaneConfig]
) -> core_pb2.Node: ) -> core_pb2.Node:
""" """
Convert CORE node to protobuf representation. Convert CORE node to protobuf representation.
@ -390,7 +390,7 @@ def get_node_proto(
) )
def get_links(session: Session, node: NodeBase) -> List[core_pb2.Link]: def get_links(session: Session, node: NodeBase) -> list[core_pb2.Link]:
""" """
Retrieve a list of links for grpc to use. Retrieve a list of links for grpc to use.
@ -435,7 +435,7 @@ def convert_iface(iface: CoreInterface) -> core_pb2.Interface:
) )
def convert_core_link(core_link: CoreLink) -> List[core_pb2.Link]: def convert_core_link(core_link: CoreLink) -> list[core_pb2.Link]:
""" """
Convert core link to protobuf data. Convert core link to protobuf data.
@ -581,7 +581,7 @@ def convert_link(
) )
def parse_proc_net_dev(lines: List[str]) -> Dict[str, Any]: def parse_proc_net_dev(lines: list[str]) -> dict[str, dict[str, float]]:
""" """
Parse lines of output from /proc/net/dev. Parse lines of output from /proc/net/dev.
@ -599,7 +599,7 @@ def parse_proc_net_dev(lines: List[str]) -> Dict[str, Any]:
return stats return stats
def get_net_stats() -> Dict[str, Dict]: def get_net_stats() -> dict[str, dict[str, float]]:
""" """
Retrieve status about the current interfaces in the system Retrieve status about the current interfaces in the system
@ -728,7 +728,7 @@ def get_nem_id(
return nem_id return nem_id
def get_emane_model_configs_dict(session: Session) -> Dict[int, List[NodeEmaneConfig]]: def get_emane_model_configs_dict(session: Session) -> dict[int, list[NodeEmaneConfig]]:
""" """
Get emane model configuration protobuf data. Get emane model configuration protobuf data.
@ -751,7 +751,7 @@ def get_emane_model_configs_dict(session: Session) -> Dict[int, List[NodeEmaneCo
return configs return configs
def get_hooks(session: Session) -> List[core_pb2.Hook]: def get_hooks(session: Session) -> list[core_pb2.Hook]:
""" """
Retrieve hook protobuf data for a session. Retrieve hook protobuf data for a session.
@ -767,7 +767,7 @@ def get_hooks(session: Session) -> List[core_pb2.Hook]:
return hooks return hooks
def get_default_services(session: Session) -> List[ServiceDefaults]: def get_default_services(session: Session) -> list[ServiceDefaults]:
""" """
Retrieve the default service sets for a given session. Retrieve the default service sets for a given session.

View file

@ -5,9 +5,11 @@ import signal
import sys import sys
import tempfile import tempfile
import time import time
from collections.abc import Iterable
from concurrent import futures from concurrent import futures
from pathlib import Path from pathlib import Path
from typing import Iterable, Optional, Pattern, Type from re import Pattern
from typing import Optional
import grpc import grpc
from grpc import ServicerContext from grpc import ServicerContext
@ -105,7 +107,7 @@ from core.services.coreservices import ServiceManager
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_ONE_DAY_IN_SECONDS: int = 60 * 60 * 24 _ONE_DAY_IN_SECONDS: int = 60 * 60 * 24
_INTERFACE_REGEX: Pattern = re.compile(r"beth(?P<node>[0-9a-fA-F]+)") _INTERFACE_REGEX: Pattern[str] = re.compile(r"beth(?P<node>[0-9a-fA-F]+)")
_MAX_WORKERS = 1000 _MAX_WORKERS = 1000
@ -171,7 +173,7 @@ class CoreGrpcServer(core_pb2_grpc.CoreApiServicer):
return session return session
def get_node( def get_node(
self, session: Session, node_id: int, context: ServicerContext, _class: Type[NT] self, session: Session, node_id: int, context: ServicerContext, _class: type[NT]
) -> NT: ) -> NT:
""" """
Retrieve node given session and node id Retrieve node given session and node id
@ -210,7 +212,7 @@ class CoreGrpcServer(core_pb2_grpc.CoreApiServicer):
def validate_service( def validate_service(
self, name: str, context: ServicerContext self, name: str, context: ServicerContext
) -> Type[ConfigService]: ) -> type[ConfigService]:
""" """
Validates a configuration service is a valid known service. Validates a configuration service is a valid known service.

View file

@ -1,7 +1,7 @@
from dataclasses import dataclass, field from dataclasses import dataclass, field
from enum import Enum from enum import Enum
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple from typing import Any, Optional
from core.api.grpc import ( from core.api.grpc import (
common_pb2, common_pb2,
@ -114,13 +114,13 @@ class EventType:
class ConfigService: class ConfigService:
group: str group: str
name: str name: str
executables: List[str] executables: list[str]
dependencies: List[str] dependencies: list[str]
directories: List[str] directories: list[str]
files: List[str] files: list[str]
startup: List[str] startup: list[str]
validate: List[str] validate: list[str]
shutdown: List[str] shutdown: list[str]
validation_mode: ConfigServiceValidationMode validation_mode: ConfigServiceValidationMode
validation_timer: int validation_timer: int
validation_period: float validation_period: float
@ -147,8 +147,8 @@ class ConfigService:
class ConfigServiceConfig: class ConfigServiceConfig:
node_id: int node_id: int
name: str name: str
templates: Dict[str, str] templates: dict[str, str]
config: Dict[str, str] config: dict[str, str]
@classmethod @classmethod
def from_proto( def from_proto(
@ -164,15 +164,15 @@ class ConfigServiceConfig:
@dataclass @dataclass
class ConfigServiceData: class ConfigServiceData:
templates: Dict[str, str] = field(default_factory=dict) templates: dict[str, str] = field(default_factory=dict)
config: Dict[str, str] = field(default_factory=dict) config: dict[str, str] = field(default_factory=dict)
@dataclass @dataclass
class ConfigServiceDefaults: class ConfigServiceDefaults:
templates: Dict[str, str] templates: dict[str, str]
config: Dict[str, "ConfigOption"] config: dict[str, "ConfigOption"]
modes: Dict[str, Dict[str, str]] modes: dict[str, dict[str, str]]
@classmethod @classmethod
def from_proto( def from_proto(
@ -211,7 +211,7 @@ class Service:
@dataclass @dataclass
class ServiceDefault: class ServiceDefault:
model: str model: str
services: List[str] services: list[str]
@classmethod @classmethod
def from_proto(cls, proto: services_pb2.ServiceDefaults) -> "ServiceDefault": def from_proto(cls, proto: services_pb2.ServiceDefaults) -> "ServiceDefault":
@ -220,15 +220,15 @@ class ServiceDefault:
@dataclass @dataclass
class NodeServiceData: class NodeServiceData:
executables: List[str] = field(default_factory=list) executables: list[str] = field(default_factory=list)
dependencies: List[str] = field(default_factory=list) dependencies: list[str] = field(default_factory=list)
dirs: List[str] = field(default_factory=list) dirs: list[str] = field(default_factory=list)
configs: List[str] = field(default_factory=list) configs: list[str] = field(default_factory=list)
startup: List[str] = field(default_factory=list) startup: list[str] = field(default_factory=list)
validate: List[str] = field(default_factory=list) validate: list[str] = field(default_factory=list)
validation_mode: ServiceValidationMode = ServiceValidationMode.NON_BLOCKING validation_mode: ServiceValidationMode = ServiceValidationMode.NON_BLOCKING
validation_timer: int = 5 validation_timer: int = 5
shutdown: List[str] = field(default_factory=list) shutdown: list[str] = field(default_factory=list)
meta: str = None meta: str = None
@classmethod @classmethod
@ -266,7 +266,7 @@ class NodeServiceConfig:
node_id: int node_id: int
service: str service: str
data: NodeServiceData data: NodeServiceData
files: Dict[str, str] = field(default_factory=dict) files: dict[str, str] = field(default_factory=dict)
@classmethod @classmethod
def from_proto(cls, proto: services_pb2.NodeServiceConfig) -> "NodeServiceConfig": def from_proto(cls, proto: services_pb2.NodeServiceConfig) -> "NodeServiceConfig":
@ -282,11 +282,11 @@ class NodeServiceConfig:
class ServiceConfig: class ServiceConfig:
node_id: int node_id: int
service: str service: str
files: List[str] = None files: list[str] = None
directories: List[str] = None directories: list[str] = None
startup: List[str] = None startup: list[str] = None
validate: List[str] = None validate: list[str] = None
shutdown: List[str] = None shutdown: list[str] = None
def to_proto(self) -> services_pb2.ServiceConfig: def to_proto(self) -> services_pb2.ServiceConfig:
return services_pb2.ServiceConfig( return services_pb2.ServiceConfig(
@ -339,8 +339,8 @@ class InterfaceThroughput:
@dataclass @dataclass
class ThroughputsEvent: class ThroughputsEvent:
session_id: int session_id: int
bridge_throughputs: List[BridgeThroughput] bridge_throughputs: list[BridgeThroughput]
iface_throughputs: List[InterfaceThroughput] iface_throughputs: list[InterfaceThroughput]
@classmethod @classmethod
def from_proto(cls, proto: core_pb2.ThroughputsEvent) -> "ThroughputsEvent": def from_proto(cls, proto: core_pb2.ThroughputsEvent) -> "ThroughputsEvent":
@ -428,19 +428,19 @@ class ConfigOption:
label: str = None label: str = None
type: ConfigOptionType = None type: ConfigOptionType = None
group: str = None group: str = None
select: List[str] = None select: list[str] = None
@classmethod @classmethod
def from_dict( def from_dict(
cls, config: Dict[str, common_pb2.ConfigOption] cls, config: dict[str, common_pb2.ConfigOption]
) -> Dict[str, "ConfigOption"]: ) -> dict[str, "ConfigOption"]:
d = {} d = {}
for key, value in config.items(): for key, value in config.items():
d[key] = ConfigOption.from_proto(value) d[key] = ConfigOption.from_proto(value)
return d return d
@classmethod @classmethod
def to_dict(cls, config: Dict[str, "ConfigOption"]) -> Dict[str, str]: def to_dict(cls, config: dict[str, "ConfigOption"]) -> dict[str, str]:
return {k: v.value for k, v in config.items()} return {k: v.value for k, v in config.items()}
@classmethod @classmethod
@ -671,7 +671,7 @@ class EmaneModelConfig:
node_id: int node_id: int
model: str model: str
iface_id: int = -1 iface_id: int = -1
config: Dict[str, ConfigOption] = None config: dict[str, ConfigOption] = None
@classmethod @classmethod
def from_proto(cls, proto: emane_pb2.GetEmaneModelConfig) -> "EmaneModelConfig": def from_proto(cls, proto: emane_pb2.GetEmaneModelConfig) -> "EmaneModelConfig":
@ -725,8 +725,8 @@ class Node:
type: NodeType = NodeType.DEFAULT type: NodeType = NodeType.DEFAULT
model: str = None model: str = None
position: Position = Position(x=0, y=0) position: Position = Position(x=0, y=0)
services: Set[str] = field(default_factory=set) services: set[str] = field(default_factory=set)
config_services: Set[str] = field(default_factory=set) config_services: set[str] = field(default_factory=set)
emane: str = None emane: str = None
icon: str = None icon: str = None
image: str = None image: str = None
@ -737,19 +737,19 @@ class Node:
canvas: int = None canvas: int = None
# configurations # configurations
emane_model_configs: Dict[ emane_model_configs: dict[
Tuple[str, Optional[int]], Dict[str, ConfigOption] tuple[str, Optional[int]], dict[str, ConfigOption]
] = field(default_factory=dict, repr=False) ] = field(default_factory=dict, repr=False)
wlan_config: Dict[str, ConfigOption] = field(default_factory=dict, repr=False) wlan_config: dict[str, ConfigOption] = field(default_factory=dict, repr=False)
wireless_config: Dict[str, ConfigOption] = field(default_factory=dict, repr=False) wireless_config: dict[str, ConfigOption] = field(default_factory=dict, repr=False)
mobility_config: Dict[str, ConfigOption] = field(default_factory=dict, repr=False) mobility_config: dict[str, ConfigOption] = field(default_factory=dict, repr=False)
service_configs: Dict[str, NodeServiceData] = field( service_configs: dict[str, NodeServiceData] = field(
default_factory=dict, repr=False default_factory=dict, repr=False
) )
service_file_configs: Dict[str, Dict[str, str]] = field( service_file_configs: dict[str, dict[str, str]] = field(
default_factory=dict, repr=False default_factory=dict, repr=False
) )
config_service_configs: Dict[str, ConfigServiceData] = field( config_service_configs: dict[str, ConfigServiceData] = field(
default_factory=dict, repr=False default_factory=dict, repr=False
) )
@ -849,18 +849,18 @@ class Node:
wireless_config={k: v.to_proto() for k, v in self.wireless_config.items()}, wireless_config={k: v.to_proto() for k, v in self.wireless_config.items()},
) )
def set_wlan(self, config: Dict[str, str]) -> None: def set_wlan(self, config: dict[str, str]) -> None:
for key, value in config.items(): for key, value in config.items():
option = ConfigOption(name=key, value=value) option = ConfigOption(name=key, value=value)
self.wlan_config[key] = option self.wlan_config[key] = option
def set_mobility(self, config: Dict[str, str]) -> None: def set_mobility(self, config: dict[str, str]) -> None:
for key, value in config.items(): for key, value in config.items():
option = ConfigOption(name=key, value=value) option = ConfigOption(name=key, value=value)
self.mobility_config[key] = option self.mobility_config[key] = option
def set_emane_model( def set_emane_model(
self, model: str, config: Dict[str, str], iface_id: int = None self, model: str, config: dict[str, str], iface_id: int = None
) -> None: ) -> None:
key = (model, iface_id) key = (model, iface_id)
config_options = self.emane_model_configs.setdefault(key, {}) config_options = self.emane_model_configs.setdefault(key, {})
@ -873,23 +873,23 @@ class Node:
class Session: class Session:
id: int = None id: int = None
state: SessionState = SessionState.DEFINITION state: SessionState = SessionState.DEFINITION
nodes: Dict[int, Node] = field(default_factory=dict) nodes: dict[int, Node] = field(default_factory=dict)
links: List[Link] = field(default_factory=list) links: list[Link] = field(default_factory=list)
dir: str = None dir: str = None
user: str = None user: str = None
default_services: Dict[str, Set[str]] = field(default_factory=dict) default_services: dict[str, set[str]] = field(default_factory=dict)
location: SessionLocation = SessionLocation( location: SessionLocation = SessionLocation(
x=0.0, y=0.0, z=0.0, lat=47.57917, lon=-122.13232, alt=2.0, scale=150.0 x=0.0, y=0.0, z=0.0, lat=47.57917, lon=-122.13232, alt=2.0, scale=150.0
) )
hooks: Dict[str, Hook] = field(default_factory=dict) hooks: dict[str, Hook] = field(default_factory=dict)
metadata: Dict[str, str] = field(default_factory=dict) metadata: dict[str, str] = field(default_factory=dict)
file: Path = None file: Path = None
options: Dict[str, ConfigOption] = field(default_factory=dict) options: dict[str, ConfigOption] = field(default_factory=dict)
servers: List[Server] = field(default_factory=list) servers: list[Server] = field(default_factory=list)
@classmethod @classmethod
def from_proto(cls, proto: core_pb2.Session) -> "Session": def from_proto(cls, proto: core_pb2.Session) -> "Session":
nodes: Dict[int, Node] = {x.id: Node.from_proto(x) for x in proto.nodes} nodes: dict[int, Node] = {x.id: Node.from_proto(x) for x in proto.nodes}
links = [Link.from_proto(x) for x in proto.links] links = [Link.from_proto(x) for x in proto.links]
default_services = {x.model: set(x.services) for x in proto.default_services} default_services = {x.model: set(x.services) for x in proto.default_services}
hooks = {x.file: Hook.from_proto(x) for x in proto.hooks} hooks = {x.file: Hook.from_proto(x) for x in proto.hooks}
@ -987,7 +987,7 @@ class Session:
self.links.append(link) self.links.append(link)
return link return link
def set_options(self, config: Dict[str, str]) -> None: def set_options(self, config: dict[str, str]) -> None:
for key, value in config.items(): for key, value in config.items():
option = ConfigOption(name=key, value=value) option = ConfigOption(name=key, value=value)
self.options[key] = option self.options[key] = option
@ -995,9 +995,9 @@ class Session:
@dataclass @dataclass
class CoreConfig: class CoreConfig:
services: List[Service] = field(default_factory=list) services: list[Service] = field(default_factory=list)
config_services: List[ConfigService] = field(default_factory=list) config_services: list[ConfigService] = field(default_factory=list)
emane_models: List[str] = field(default_factory=list) emane_models: list[str] = field(default_factory=list)
@classmethod @classmethod
def from_proto(cls, proto: core_pb2.GetConfigResponse) -> "CoreConfig": def from_proto(cls, proto: core_pb2.GetConfigResponse) -> "CoreConfig":
@ -1088,7 +1088,7 @@ class ConfigEvent:
node_id: int node_id: int
object: str object: str
type: int type: int
data_types: List[int] data_types: list[int]
data_values: str data_values: str
captions: str captions: str
bitmap: str bitmap: str