afonso/core-extra
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

gui: updated core.gui to not use deprecated type hinting

Browse source
This commit is contained in:
Blake Harnden 2023-04-13 15:53:16 -07:00
parent 69f05a6712
commit e7351b594d
40 changed files with 268 additions and 257 deletions
Download patch file Download diff file Expand all files Collapse all files

59
daemon/core/gui/coreclient.py
View file

@ -6,9 +6,10 @@ import json
import logging
import os
import tkinter as tk
from collections.abc import Iterable
from pathlib import Path
from tkinter import messagebox
from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Set, Tuple
from typing import TYPE_CHECKING, Optional
import grpc
@ -55,7 +56,7 @@ GUI_SOURCE = "gui"
CPU_USAGE_DELAY = 3
def to_dict(config: Dict[str, ConfigOption]) -> Dict[str, str]:
def to_dict(config: dict[str, ConfigOption]) -> dict[str, str]:
return {x: y.value for x, y in config.items()}
@ -74,26 +75,26 @@ class CoreClient:
self.show_throughputs: tk.BooleanVar = tk.BooleanVar(value=False)
# global service settings
self.services: Dict[str, Set[str]] = {}
self.config_services_groups: Dict[str, Set[str]] = {}
self.config_services: Dict[str, ConfigService] = {}
self.services: dict[str, set[str]] = {}
self.config_services_groups: dict[str, set[str]] = {}
self.config_services: dict[str, ConfigService] = {}
# loaded configuration data
self.emane_models: List[str] = []
self.servers: Dict[str, CoreServer] = {}
self.custom_nodes: Dict[str, NodeDraw] = {}
self.custom_observers: Dict[str, Observer] = {}
self.emane_models: list[str] = []
self.servers: dict[str, CoreServer] = {}
self.custom_nodes: dict[str, NodeDraw] = {}
self.custom_observers: dict[str, Observer] = {}
self.read_config()
# helpers
self.iface_to_edge: Dict[Tuple[int, ...], CanvasEdge] = {}
self.iface_to_edge: dict[tuple[int, ...], CanvasEdge] = {}
self.ifaces_manager: InterfaceManager = InterfaceManager(self.app)
self.observer: Optional[str] = None
# session data
self.mobility_players: Dict[int, MobilityPlayer] = {}
self.canvas_nodes: Dict[int, CanvasNode] = {}
self.links: Dict[str, CanvasEdge] = {}
self.mobility_players: dict[int, MobilityPlayer] = {}
self.canvas_nodes: dict[int, CanvasNode] = {}
self.links: dict[str, CanvasEdge] = {}
self.handling_throughputs: Optional[grpc.Future] = None
self.handling_cpu_usage: Optional[grpc.Future] = None
self.handling_events: Optional[grpc.Future] = None
@ -372,7 +373,7 @@ class CoreClient:
# existing session
sessions = self.client.get_sessions()
if session_id:
session_ids = set(x.id for x in sessions)
session_ids = {x.id for x in sessions}
if session_id not in session_ids:
self.app.show_error(
"Join Session Error",
@ -401,7 +402,7 @@ class CoreClient:
except grpc.RpcError as e:
self.app.show_grpc_exception("Edit Node Error", e)
def get_links(self, definition: bool = False) -> List[Link]:
def get_links(self, definition: bool = False) -> list[Link]:
if not definition:
self.ifaces_manager.set_macs([x.link for x in self.links.values()])
links = []
@ -419,7 +420,7 @@ class CoreClient:
links.append(edge.asymmetric_link)
return links
def start_session(self, definition: bool = False) -> Tuple[bool, List[str]]:
def start_session(self, definition: bool = False) -> tuple[bool, list[str]]:
self.session.links = self.get_links(definition)
self.session.metadata = self.get_metadata()
self.session.servers.clear()
@ -461,7 +462,7 @@ class CoreClient:
self.mobility_players[node.id] = mobility_player
mobility_player.show()
def get_metadata(self) -> Dict[str, str]:
def get_metadata(self) -> dict[str, str]:
# create canvas data
canvas_config = self.app.manager.get_metadata()
canvas_config = json.dumps(canvas_config)
@ -652,7 +653,7 @@ class CoreClient:
self.session.nodes[node.id] = node
return node
def deleted_canvas_nodes(self, canvas_nodes: List[CanvasNode]) -> None:
def deleted_canvas_nodes(self, canvas_nodes: list[CanvasNode]) -> None:
"""
remove the nodes selected by the user and anything related to that node
such as link, configurations, interfaces
@ -680,7 +681,7 @@ class CoreClient:
dst_iface_id = edge.link.iface2.id
self.iface_to_edge[(dst_node.id, dst_iface_id)] = edge
def get_wlan_configs(self) -> List[Tuple[int, Dict[str, str]]]:
def get_wlan_configs(self) -> list[tuple[int, dict[str, str]]]:
configs = []
for node in self.session.nodes.values():
if node.type != NodeType.WIRELESS_LAN:
@ -691,7 +692,7 @@ class CoreClient:
configs.append((node.id, config))
return configs
def get_mobility_configs(self) -> List[Tuple[int, Dict[str, str]]]:
def get_mobility_configs(self) -> list[tuple[int, dict[str, str]]]:
configs = []
for node in self.session.nodes.values():
if not nutils.is_mobility(node):
@ -702,7 +703,7 @@ class CoreClient:
configs.append((node.id, config))
return configs
def get_emane_model_configs(self) -> List[EmaneModelConfig]:
def get_emane_model_configs(self) -> list[EmaneModelConfig]:
configs = []
for node in self.session.nodes.values():
for key, config in node.emane_model_configs.items():
@ -716,7 +717,7 @@ class CoreClient:
configs.append(config)
return configs
def get_service_configs(self) -> List[ServiceConfig]:
def get_service_configs(self) -> list[ServiceConfig]:
configs = []
for node in self.session.nodes.values():
if not nutils.is_container(node):
@ -736,7 +737,7 @@ class CoreClient:
configs.append(config)
return configs
def get_service_file_configs(self) -> List[ServiceFileConfig]:
def get_service_file_configs(self) -> list[ServiceFileConfig]:
configs = []
for node in self.session.nodes.values():
if not nutils.is_container(node):
@ -749,12 +750,12 @@ class CoreClient:
configs.append(config)
return configs
def get_config_service_rendered(self, node_id: int, name: str) -> Dict[str, str]:
def get_config_service_rendered(self, node_id: int, name: str) -> dict[str, str]:
return self.client.get_config_service_rendered(self.session.id, node_id, name)
def get_config_service_configs_proto(
self,
) -> List[configservices_pb2.ConfigServiceConfig]:
) -> list[configservices_pb2.ConfigServiceConfig]:
config_service_protos = []
for node in self.session.nodes.values():
if not nutils.is_container(node):
@ -776,7 +777,7 @@ class CoreClient:
_, output = self.client.node_command(self.session.id, node_id, self.observer)
return output
def get_wlan_config(self, node_id: int) -> Dict[str, ConfigOption]:
def get_wlan_config(self, node_id: int) -> dict[str, ConfigOption]:
config = self.client.get_wlan_config(self.session.id, node_id)
logger.debug(
"get wlan configuration from node %s, result configuration: %s",
@ -785,10 +786,10 @@ class CoreClient:
)
return config
def get_wireless_config(self, node_id: int) -> Dict[str, ConfigOption]:
def get_wireless_config(self, node_id: int) -> dict[str, ConfigOption]:
return self.client.get_wireless_config(self.session.id, node_id)
def get_mobility_config(self, node_id: int) -> Dict[str, ConfigOption]:
def get_mobility_config(self, node_id: int) -> dict[str, ConfigOption]:
config = self.client.get_mobility_config(self.session.id, node_id)
logger.debug(
"get mobility config from node %s, result configuration: %s",
@ -799,7 +800,7 @@ class CoreClient:
def get_emane_model_config(
self, node_id: int, model: str, iface_id: int = None
) -> Dict[str, ConfigOption]:
) -> dict[str, ConfigOption]:
if iface_id is None:
iface_id = -1
config = self.client.get_emane_model_config(