adjust python code where possible to use updated Optional and Union type hinting for python 3.10

Author: bharnden

daemon/core/api/grpc/client.py

@@ -8,7 +8,7 @@
 from contextlib import contextmanager
 from pathlib import Path
 from queue import Queue
-from typing import Any, Optional
+from typing import Any
 
 import grpc
 
@@ -57,7 +57,7 @@
 class MoveNodesStreamer:
     def __init__(self, session_id: int, source: str = None) -> None:
         self.session_id: int = session_id
-        self.source: Optional[str] = source
+        self.source: str | None = source
         self.queue: SetQueue = SetQueue()
 
     def send_position(self, node_id: int, x: float, y: float) -> None:
@@ -83,8 +83,8 @@ def send(self, request: wrappers.MoveNodesRequest) -> None:
     def stop(self) -> None:
         self.queue.put(None)
 
-    def next(self) -> Optional[core_pb2.MoveNodesRequest]:
-        request: Optional[wrappers.MoveNodesRequest] = self.queue.get()
+    def next(self) -> core_pb2.MoveNodesRequest | None:
+        request: wrappers.MoveNodesRequest | None = self.queue.get()
         if request:
             return request.to_proto()
         else:
@@ -98,11 +98,11 @@ class EmanePathlossesStreamer:
     def __init__(self) -> None:
         self.queue: Queue = Queue()
 
-    def send(self, request: Optional[wrappers.EmanePathlossesRequest]) -> None:
+    def send(self, request: wrappers.EmanePathlossesRequest | None) -> None:
         self.queue.put(request)
 
-    def next(self) -> Optional[emane_pb2.EmanePathlossesRequest]:
-        request: Optional[wrappers.EmanePathlossesRequest] = self.queue.get()
+    def next(self) -> emane_pb2.EmanePathlossesRequest | None:
+        request: wrappers.EmanePathlossesRequest | None = self.queue.get()
         if request:
             return request.to_proto()
         else:
@@ -116,11 +116,11 @@ class EmaneEventsStreamer:
     def __init__(self) -> None:
         self.queue: Queue = Queue()
 
-    def send(self, request: Optional[wrappers.EmaneEventsRequest]) -> None:
+    def send(self, request: wrappers.EmaneEventsRequest | None) -> None:
         self.queue.put(request)
 
-    def next(self) -> Optional[emane_pb2.EmaneEventsRequest]:
-        request: Optional[wrappers.EmaneEventsRequest] = self.queue.get()
+    def next(self) -> emane_pb2.EmaneEventsRequest | None:
+        request: wrappers.EmaneEventsRequest | None = self.queue.get()
         if request:
             return request.to_proto()
         else:
@@ -246,8 +246,8 @@ def __init__(self, address: str = "localhost:50051", proxy: bool = False) -> Non
         :param address: grpc server address to connect to
         """
         self.address: str = address
-        self.stub: Optional[core_pb2_grpc.CoreApiStub] = None
-        self.channel: Optional[grpc.Channel] = None
+        self.stub: core_pb2_grpc.CoreApiStub | None = None
+        self.channel: grpc.Channel | None = None
         self.proxy: bool = proxy
 
     def start_session(
@@ -938,7 +938,7 @@ def get_emane_event_channel(
         response = self.stub.GetEmaneEventChannel(request)
         return wrappers.EmaneEventChannel.from_proto(response)
 
-    def execute_script(self, script: str, args: str) -> Optional[int]:
+    def execute_script(self, script: str, args: str) -> int | None:
         """
         Executes a python script given context of the current CoreEmu object.
 

daemon/core/api/grpc/events.py

@@ -1,7 +1,6 @@
 import logging
 from collections.abc import Iterable
 from queue import Empty, Queue
-from typing import Optional
 
 from core.api.grpc import core_pb2, grpcutils
 from core.api.grpc.grpcutils import convert_link_data
@@ -113,7 +112,7 @@ def add_handlers(self) -> None:
         if core_pb2.EventType.SESSION in self.event_types:
             self.session.broadcast_manager.add_handler(EventData, self.queue.put)
 
-    def process(self) -> Optional[core_pb2.Event]:
+    def process(self) -> core_pb2.Event | None:
         """
         Process the next event in the queue.
 

daemon/core/api/grpc/grpcutils.py

@@ -1,7 +1,7 @@
 import logging
 import time
 from pathlib import Path
-from typing import Any, Optional, Union
+from typing import Any
 
 import grpc
 from google.protobuf.internal.containers import RepeatedCompositeFieldContainer
@@ -248,7 +248,7 @@ def convert_session_options(session: Session) -> dict[str, common_pb2.ConfigOpti
 
 def get_config_options(
     config: dict[str, str],
-    configurable_options: Union[ConfigurableOptions, type[ConfigurableOptions]],
+    configurable_options: ConfigurableOptions | type[ConfigurableOptions],
 ) -> dict[str, common_pb2.ConfigOption]:
     """
     Retrieve configuration options in a form that is used by the grpc server.
@@ -524,9 +524,9 @@ def convert_options_proto(options: core_pb2.LinkOptions) -> LinkOptions:
 
 def convert_link(
     node1: NodeBase,
-    iface1: Optional[CoreInterface],
+    iface1: CoreInterface | None,
     node2: NodeBase,
-    iface2: Optional[CoreInterface],
+    iface2: CoreInterface | None,
     options: LinkOptions,
     unidirectional: bool,
 ) -> core_pb2.Link:
@@ -709,7 +709,7 @@ def get_hooks(session: Session) -> list[core_pb2.Hook]:
 
 def get_mobility_node(
     session: Session, node_id: int, context: ServicerContext
-) -> Union[WlanNode, EmaneNet]:
+) -> WlanNode | EmaneNet:
     """
     Get mobility node.
 
@@ -819,5 +819,5 @@ def configure_node(
                 service.set_template(name, template)
 
 
-def get_optional(message: Message, name: str) -> Optional[Any]:
+def get_optional(message: Message, name: str) -> Any | None:
     return getattr(message, name) if message.HasField(name) else None

daemon/core/api/grpc/server.py

@@ -9,7 +9,6 @@
 from concurrent import futures
 from pathlib import Path
 from re import Pattern
-from typing import Optional
 
 import grpc
 from grpc import ServicerContext
@@ -107,7 +106,7 @@ def __init__(self, coreemu: CoreEmu) -> None:
         super().__init__()
         self.coreemu: CoreEmu = coreemu
         self.running: bool = True
-        self.server: Optional[grpc.Server] = None
+        self.server: grpc.Server | None = None
         # catch signals
         signal.signal(signal.SIGHUP, self._signal_handler)
         signal.signal(signal.SIGINT, self._signal_handler)

daemon/core/api/grpc/wrappers.py

@@ -1,7 +1,6 @@
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import Path
-from typing import Optional
 
 from google.protobuf.internal.containers import MessageMap
 
@@ -615,9 +614,9 @@ class Node:
     canvas: int = None
 
     # configurations
-    emane_model_configs: dict[
-        tuple[str, Optional[int]], dict[str, ConfigOption]
-    ] = field(default_factory=dict, repr=False)
+    emane_model_configs: dict[tuple[str, int | None], 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)
     mobility_config: dict[str, ConfigOption] = field(default_factory=dict, repr=False)

daemon/core/config.py

@@ -5,7 +5,7 @@
 import logging
 from collections import OrderedDict
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import TYPE_CHECKING, Any
 
 from core.emane.nodes import EmaneNet
 from core.emulator.enumerations import ConfigDataTypes
@@ -119,7 +119,7 @@ class ConfigurableOptions:
     Provides a base for defining configuration options within CORE.
     """
 
-    name: Optional[str] = None
+    name: str | None = None
     options: list[Configuration] = []
 
     @classmethod
@@ -252,7 +252,7 @@ def get_config(
 
     def get_configs(
         self, node_id: int = _default_node, config_type: str = _default_type
-    ) -> Optional[dict[str, str]]:
+    ) -> dict[str, str] | None:
         """
         Retrieve configurations for a node and configuration type.
 
@@ -342,7 +342,7 @@ def get_model_config(self, node_id: int, model_name: str) -> dict[str, str]:
 
     def set_model(
         self,
-        node: Union[WlanNode, EmaneNet],
+        node: WlanNode | EmaneNet,
         model_class: "WirelessModelType",
         config: dict[str, str] = None,
     ) -> None:
@@ -362,7 +362,7 @@ def set_model(
         node.setmodel(model_class, config)
 
     def get_models(
-        self, node: Union[WlanNode, EmaneNet]
+        self, node: WlanNode | EmaneNet
     ) -> list[tuple[type, dict[str, str]]]:
         """
         Return a list of model classes and values for a net if one has been

daemon/core/emane/emanemanager.py

@@ -5,7 +5,7 @@
 import logging
 import threading
 from enum import Enum
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from core import utils
 from core.emane.emanemodel import EmaneModel
@@ -72,7 +72,7 @@ def __init__(self, session: "Session") -> None:
             "emane_transform_port", 8200
         )
         self.doeventloop: bool = False
-        self.eventmonthread: Optional[threading.Thread] = None
+        self.eventmonthread: threading.Thread | None = None
 
         # model for global EMANE configuration options
         self.node_configs: dict[int, dict[str, dict[str, str]]] = {}
@@ -96,7 +96,7 @@ def next_nem_id(self, iface: CoreInterface) -> int:
 
     def get_config(
         self, key: int, model: str, default: bool = True
-    ) -> Optional[dict[str, str]]:
+    ) -> dict[str, str] | None:
         """
         Get the current or default configuration for an emane model.
 
@@ -326,10 +326,10 @@ def setup_control_channels(
         if eventgroup != otagroup:
             node.node_net_client.create_route(eventgroup, eventdev)
 
-    def get_iface(self, nem_id: int) -> Optional[CoreInterface]:
+    def get_iface(self, nem_id: int) -> CoreInterface | None:
         return self.nems_to_ifaces.get(nem_id)
 
-    def get_nem_id(self, iface: CoreInterface) -> Optional[int]:
+    def get_nem_id(self, iface: CoreInterface) -> int | None:
         return self.ifaces_to_nems.get(iface)
 
     def get_nem_port(self, iface: CoreInterface) -> int:
@@ -338,7 +338,7 @@ def get_nem_port(self, iface: CoreInterface) -> int:
 
     def get_nem_position(
         self, iface: CoreInterface
-    ) -> Optional[tuple[int, float, float, int]]:
+    ) -> tuple[int, float, float, int] | None:
         """
         Retrieves nem position for a given interface.
 
@@ -480,7 +480,7 @@ def check_node_models(self) -> None:
 
     def get_nem_link(
         self, nem1: int, nem2: int, flags: MessageFlags = MessageFlags.NONE
-    ) -> Optional[LinkData]:
+    ) -> LinkData | None:
         iface1 = self.get_iface(nem1)
         if not iface1:
             logger.error("invalid nem: %s", nem1)

daemon/core/emane/emanemodel.py

@@ -3,7 +3,6 @@
 """
 import logging
 from pathlib import Path
-from typing import Optional
 
 from core.config import ConfigBool, ConfigGroup, ConfigString, Configuration
 from core.emane import emanemanifest
@@ -37,13 +36,13 @@ class EmaneModel(WirelessModel):
     platform_config: list[Configuration] = []
 
     # default mac configuration settings
-    mac_library: Optional[str] = None
-    mac_xml: Optional[str] = None
+    mac_library: str | None = None
+    mac_xml: str | None = None
     mac_defaults: dict[str, str] = {}
     mac_config: list[Configuration] = []
 
     # default phy configuration settings, using the universal model
-    phy_library: Optional[str] = None
+    phy_library: str | None = None
     phy_xml: str = "emanephy.xml"
     phy_defaults: dict[str, str] = {
         "subid": "1",

daemon/core/emane/eventmanager.py

@@ -1,6 +1,6 @@
 import logging
 import threading
-from typing import Callable, Optional, Union
+from typing import Callable, Union
 
 from core.errors import CoreError
 
@@ -51,7 +51,7 @@ def __init__(
         self.port: int = port
         self.location_handler: Callable[[LocationEvent], None] = location_handler
         self.running: bool = False
-        self.thread: Optional[threading.Thread] = None
+        self.thread: threading.Thread | None = None
         logger.info("starting emane event service %s %s:%s", device, group, port)
         self.events: EventService = EventService(
             eventchannel=(group, port, device), otachannel=None
@@ -125,7 +125,7 @@ def create_service(
         else:
             self.nem_service[nem_id] = service
 
-    def get_service(self, nem_id: int) -> Optional[EmaneEventService]:
+    def get_service(self, nem_id: int) -> EmaneEventService | None:
         service = self.nem_service.get(nem_id)
         if not service:
             logger.error("failure to find event service for nem(%s)", nem_id)

daemon/core/emane/linkmonitor.py

@@ -2,7 +2,7 @@
 import sched
 import threading
 import time
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from lxml import etree
 
@@ -40,7 +40,7 @@ def __init__(self, losses: dict[float, float]) -> None:
         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
+        self.mac_id: str | None = None
 
     def get_loss(self, sinr: float) -> float:
         index = self._get_index(sinr)
@@ -59,7 +59,7 @@ def __init__(self, from_nem: int, to_nem: int, sinr: float) -> None:
         self.from_nem: int = from_nem
         self.to_nem: int = to_nem
         self.sinr: float = sinr
-        self.last_seen: Optional[float] = None
+        self.last_seen: float | None = None
         self.updated: bool = False
         self.touch()
 
@@ -182,10 +182,10 @@ def __init__(self, emane_manager: "EmaneManager") -> None:
         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
-        self.scheduler: Optional[sched.scheduler] = None
+        self.loss_threshold: int | None = None
+        self.link_interval: int | None = None
+        self.link_timeout: int | None = None
+        self.scheduler: sched.scheduler | None = None
         self.running: bool = False
 
     def start(self) -> None: