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

pygui: added type hinting to everything under base core.gui

Browse source
This commit is contained in:
Blake Harnden 2020-06-19 22:08:24 -07:00
parent adfce52632
commit 0356f3b19c
17 changed files with 473 additions and 430 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
daemon/core/api/grpc/client.py
View file

@ -436,7 +436,7 @@ class CoreGrpcClient:
session_id: int,
handler: Callable[[core_pb2.Event], None],
events: List[core_pb2.Event] = None,
) -> Any:
) -> grpc.Channel:
"""
Listen for session events.
@ -453,7 +453,7 @@ class CoreGrpcClient:
def throughputs(
self, session_id: int, handler: Callable[[core_pb2.ThroughputsEvent], None]
) -> Any:
) -> grpc.Channel:
"""
Listen for throughput events with information for interfaces and bridges.

32
daemon/core/gui/app.py
View file

@ -3,10 +3,12 @@ import math
import tkinter as tk
from tkinter import PhotoImage, font, ttk
from tkinter.ttk import Progressbar
from typing import Dict, Optional
import grpc
from core.gui import appconfig, themes
from core.gui.appconfig import GuiConfig
from core.gui.coreclient import CoreClient
from core.gui.dialogs.error import ErrorDialog
from core.gui.graph.graph import CanvasGraph
@ -16,8 +18,8 @@ from core.gui.nodeutils import NodeUtils
from core.gui.statusbar import StatusBar
from core.gui.toolbar import Toolbar
WIDTH = 1000
HEIGHT = 800
WIDTH: int = 1000
HEIGHT: int = 800
class Application(ttk.Frame):
@ -27,25 +29,25 @@ class Application(ttk.Frame):
NodeUtils.setup()
# widgets
self.menubar = None
self.toolbar = None
self.right_frame = None
self.canvas = None
self.statusbar = None
self.progress = None
self.menubar: Optional[Menubar] = None
self.toolbar: Optional[Toolbar] = None
self.right_frame: Optional[ttk.Frame] = None
self.canvas: Optional[CanvasGraph] = None
self.statusbar: Optional[StatusBar] = None
self.progress: Optional[Progressbar] = None
# fonts
self.fonts_size = None
self.icon_text_font = None
self.edge_font = None
self.fonts_size: Dict[str, int] = {}
self.icon_text_font: Optional[font.Font] = None
self.edge_font: Optional[font.Font] = None
# setup
self.guiconfig = appconfig.read()
self.app_scale = self.guiconfig.scale
self.guiconfig: GuiConfig = appconfig.read()
self.app_scale: float = self.guiconfig.scale
self.setup_scaling()
self.style = ttk.Style()
self.style: ttk.Style = ttk.Style()
self.setup_theme()
self.core = CoreClient(self, proxy)
self.core: CoreClient = CoreClient(self, proxy)
self.setup_app()
self.draw()
self.core.setup()

136
daemon/core/gui/appconfig.py
View file

@ -1,32 +1,32 @@
import os
import shutil
from pathlib import Path
from typing import List, Optional
from typing import Dict, List, Optional, Type
import yaml
from core.gui import themes
HOME_PATH = Path.home().joinpath(".coregui")
BACKGROUNDS_PATH = HOME_PATH.joinpath("backgrounds")
CUSTOM_EMANE_PATH = HOME_PATH.joinpath("custom_emane")
CUSTOM_SERVICE_PATH = HOME_PATH.joinpath("custom_services")
ICONS_PATH = HOME_PATH.joinpath("icons")
MOBILITY_PATH = HOME_PATH.joinpath("mobility")
XMLS_PATH = HOME_PATH.joinpath("xmls")
CONFIG_PATH = HOME_PATH.joinpath("config.yaml")
LOG_PATH = HOME_PATH.joinpath("gui.log")
SCRIPT_PATH = HOME_PATH.joinpath("scripts")
HOME_PATH: Path = Path.home().joinpath(".coregui")
BACKGROUNDS_PATH: Path = HOME_PATH.joinpath("backgrounds")
CUSTOM_EMANE_PATH: Path = HOME_PATH.joinpath("custom_emane")
CUSTOM_SERVICE_PATH: Path = HOME_PATH.joinpath("custom_services")
ICONS_PATH: Path = HOME_PATH.joinpath("icons")
MOBILITY_PATH: Path = HOME_PATH.joinpath("mobility")
XMLS_PATH: Path = HOME_PATH.joinpath("xmls")
CONFIG_PATH: Path = HOME_PATH.joinpath("config.yaml")
LOG_PATH: Path = HOME_PATH.joinpath("gui.log")
SCRIPT_PATH: Path = HOME_PATH.joinpath("scripts")
# local paths
DATA_PATH = Path(__file__).parent.joinpath("data")
LOCAL_ICONS_PATH = DATA_PATH.joinpath("icons").absolute()
LOCAL_BACKGROUND_PATH = DATA_PATH.joinpath("backgrounds").absolute()
LOCAL_XMLS_PATH = DATA_PATH.joinpath("xmls").absolute()
LOCAL_MOBILITY_PATH = DATA_PATH.joinpath("mobility").absolute()
DATA_PATH: Path = Path(__file__).parent.joinpath("data")
LOCAL_ICONS_PATH: Path = DATA_PATH.joinpath("icons").absolute()
LOCAL_BACKGROUND_PATH: Path = DATA_PATH.joinpath("backgrounds").absolute()
LOCAL_XMLS_PATH: Path = DATA_PATH.joinpath("xmls").absolute()
LOCAL_MOBILITY_PATH: Path = DATA_PATH.joinpath("mobility").absolute()
# configuration data
TERMINALS = {
TERMINALS: Dict[str, str] = {
"xterm": "xterm -e",
"aterm": "aterm -e",
"eterm": "eterm -e",
@ -36,45 +36,45 @@ TERMINALS = {
"xfce4-terminal": "xfce4-terminal -x",
"gnome-terminal": "gnome-terminal --window --",
}
EDITORS = ["$EDITOR", "vim", "emacs", "gedit", "nano", "vi"]
EDITORS: List[str] = ["$EDITOR", "vim", "emacs", "gedit", "nano", "vi"]
class IndentDumper(yaml.Dumper):
def increase_indent(self, flow=False, indentless=False):
return super().increase_indent(flow, False)
def increase_indent(self, flow: bool = False, indentless: bool = False) -> None:
super().increase_indent(flow, False)
class CustomNode(yaml.YAMLObject):
yaml_tag = "!CustomNode"
yaml_loader = yaml.SafeLoader
yaml_tag: str = "!CustomNode"
yaml_loader: Type[yaml.SafeLoader] = yaml.SafeLoader
def __init__(self, name: str, image: str, services: List[str]) -> None:
self.name = name
self.image = image
self.services = services
self.name: str = name
self.image: str = image
self.services: List[str] = services
class CoreServer(yaml.YAMLObject):
yaml_tag = "!CoreServer"
yaml_loader = yaml.SafeLoader
yaml_tag: str = "!CoreServer"
yaml_loader: Type[yaml.SafeLoader] = yaml.SafeLoader
def __init__(self, name: str, address: str) -> None:
self.name = name
self.address = address
self.name: str = name
self.address: str = address
class Observer(yaml.YAMLObject):
yaml_tag = "!Observer"
yaml_loader = yaml.SafeLoader
yaml_tag: str = "!Observer"
yaml_loader: Type[yaml.SafeLoader] = yaml.SafeLoader
def __init__(self, name: str, cmd: str) -> None:
self.name = name
self.cmd = cmd
self.name: str = name
self.cmd: str = cmd
class PreferencesConfig(yaml.YAMLObject):
yaml_tag = "!PreferencesConfig"
yaml_loader = yaml.SafeLoader
yaml_tag: str = "!PreferencesConfig"
yaml_loader: Type[yaml.SafeLoader] = yaml.SafeLoader
def __init__(
self,
@ -85,17 +85,17 @@ class PreferencesConfig(yaml.YAMLObject):
width: int = 1000,
height: int = 750,
) -> None:
self.theme = theme
self.editor = editor
self.terminal = terminal
self.gui3d = gui3d
self.width = width
self.height = height
self.theme: str = theme
self.editor: str = editor
self.terminal: str = terminal
self.gui3d: str = gui3d
self.width: int = width
self.height: int = height
class LocationConfig(yaml.YAMLObject):
yaml_tag = "!LocationConfig"
yaml_loader = yaml.SafeLoader
yaml_tag: str = "!LocationConfig"
yaml_loader: Type[yaml.SafeLoader] = yaml.SafeLoader
def __init__(
self,
@ -107,18 +107,18 @@ class LocationConfig(yaml.YAMLObject):
alt: float = 2.0,
scale: float = 150.0,
) -> None:
self.x = x
self.y = y
self.z = z
self.lat = lat
self.lon = lon
self.alt = alt
self.scale = scale
self.x: float = x
self.y: float = y
self.z: float = z
self.lat: float = lat
self.lon: float = lon
self.alt: float = alt
self.scale: float = scale
class IpConfigs(yaml.YAMLObject):
yaml_tag = "!IpConfigs"
yaml_loader = yaml.SafeLoader
yaml_tag: str = "!IpConfigs"
yaml_loader: Type[yaml.SafeLoader] = yaml.SafeLoader
def __init__(
self,
@ -129,21 +129,21 @@ class IpConfigs(yaml.YAMLObject):
) -> None:
if ip4s is None:
ip4s = ["10.0.0.0", "192.168.0.0", "172.16.0.0"]
self.ip4s = ip4s
self.ip4s: List[str] = ip4s
if ip6s is None:
ip6s = ["2001::", "2002::", "a::"]
self.ip6s = ip6s
self.ip6s: List[str] = ip6s
if ip4 is None:
ip4 = self.ip4s[0]
self.ip4 = ip4
self.ip4: str = ip4
if ip6 is None:
ip6 = self.ip6s[0]
self.ip6 = ip6
self.ip6: str = ip6
class GuiConfig(yaml.YAMLObject):
yaml_tag = "!GuiConfig"
yaml_loader = yaml.SafeLoader
yaml_tag: str = "!GuiConfig"
yaml_loader: Type[yaml.SafeLoader] = yaml.SafeLoader
def __init__(
self,
@ -159,30 +159,30 @@ class GuiConfig(yaml.YAMLObject):
) -> None:
if preferences is None:
preferences = PreferencesConfig()
self.preferences = preferences
self.preferences: PreferencesConfig = preferences
if location is None:
location = LocationConfig()
self.location = location
self.location: LocationConfig = location
if servers is None:
servers = []
self.servers = servers
self.servers: List[CoreServer] = servers
if nodes is None:
nodes = []
self.nodes = nodes
self.nodes: List[CustomNode] = nodes
if recentfiles is None:
recentfiles = []
self.recentfiles = recentfiles
self.recentfiles: List[str] = recentfiles
if observers is None:
observers = []
self.observers = observers
self.scale = scale
self.observers: List[Observer] = observers
self.scale: float = scale
if ips is None:
ips = IpConfigs()
self.ips = ips
self.mac = mac
self.ips: IpConfigs = ips
self.mac: str = mac
def copy_files(current_path, new_path) -> None:
def copy_files(current_path: Path, new_path: Path) -> None:
for current_file in current_path.glob("*"):
new_file = new_path.joinpath(current_file.name)
shutil.copy(current_file, new_file)

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

@ -4,18 +4,41 @@ Incorporate grpc into python tkinter GUI
import json
import logging
import os
import tkinter as tk
from pathlib import Path
from tkinter import messagebox
from typing import TYPE_CHECKING, Dict, Iterable, List, Optional
from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Set, Tuple
import grpc
from core.api.grpc import client, common_pb2, configservices_pb2, core_pb2
from core.api.grpc import client
from core.api.grpc.common_pb2 import ConfigOption
from core.api.grpc.configservices_pb2 import ConfigService, ConfigServiceConfig
from core.api.grpc.core_pb2 import (
Event,
ExceptionEvent,
Hook,
Interface,
Link,
LinkEvent,
LinkType,
MessageType,
Node,
NodeEvent,
NodeType,
Position,
SessionLocation,
SessionState,
StartSessionResponse,
StopSessionResponse,
ThroughputsEvent,
)
from core.api.grpc.emane_pb2 import EmaneModelConfig
from core.api.grpc.mobility_pb2 import MobilityConfig
from core.api.grpc.services_pb2 import NodeServiceData, ServiceConfig, ServiceFileConfig
from core.api.grpc.wlan_pb2 import WlanConfig
from core.gui import appconfig
from core.gui.appconfig import CoreServer
from core.gui.dialogs.emaneinstall import EmaneInstallDialog
from core.gui.dialogs.error import ErrorDialog
from core.gui.dialogs.mobilityplayer import MobilityPlayer
@ -34,47 +57,46 @@ GUI_SOURCE = "gui"
class CoreClient:
def __init__(self, app: "Application", proxy: bool):
def __init__(self, app: "Application", proxy: bool) -> None:
"""
Create a CoreGrpc instance
"""
self._client = client.CoreGrpcClient(proxy=proxy)
self.session_id = None
self.node_ids = []
self.app = app
self.master = app.master
self.services = {}
self.config_services_groups = {}
self.config_services = {}
self.default_services = {}
self.emane_models = []
self.observer = None
self.app: "Application" = app
self.master: tk.Tk = app.master
self._client: client.CoreGrpcClient = client.CoreGrpcClient(proxy=proxy)
self.session_id: Optional[int] = None
self.services: Dict[str, Set[str]] = {}
self.config_services_groups: Dict[str, Set[str]] = {}
self.config_services: Dict[str, ConfigService] = {}
self.default_services: Dict[NodeType, Set[str]] = {}
self.emane_models: List[str] = []
self.observer: Optional[str] = None
# loaded configuration data
self.servers = {}
self.custom_nodes = {}
self.custom_observers = {}
self.servers: Dict[str, CoreServer] = {}
self.custom_nodes: Dict[str, NodeDraw] = {}
self.custom_observers: Dict[str, str] = {}
self.read_config()
# helpers
self.iface_to_edge = {}
self.ifaces_manager = InterfaceManager(self.app)
self.iface_to_edge: Dict[Tuple[int, int], Tuple[int, int]] = {}
self.ifaces_manager: InterfaceManager = InterfaceManager(self.app)
# session data
self.state = None
self.canvas_nodes = {}
self.location = None
self.links = {}
self.hooks = {}
self.emane_config = None
self.mobility_players = {}
self.handling_throughputs = None
self.handling_events = None
self.xml_dir = None
self.xml_file = None
self.state: Optional[SessionState] = None
self.canvas_nodes: Dict[int, CanvasNode] = {}
self.location: Optional[SessionLocation] = None
self.links: Dict[Tuple[int, int], CanvasEdge] = {}
self.hooks: Dict[str, Hook] = {}
self.emane_config: Dict[str, ConfigOption] = {}
self.mobility_players: Dict[int, MobilityPlayer] = {}
self.handling_throughputs: Optional[grpc.Channel] = None
self.handling_events: Optional[grpc.Channel] = None
self.xml_dir: Optional[str] = None
self.xml_file: Optional[str] = None
@property
def client(self):
def client(self) -> client.CoreGrpcClient:
if self.session_id:
response = self._client.check_session(self.session_id)
if not response.result:
@ -89,7 +111,7 @@ class CoreClient:
self.enable_throughputs()
return self._client
def reset(self):
def reset(self) -> None:
# helpers
self.ifaces_manager.reset()
self.iface_to_edge.clear()
@ -104,14 +126,14 @@ class CoreClient:
self.cancel_throughputs()
self.cancel_events()
def close_mobility_players(self):
def close_mobility_players(self) -> None:
for mobility_player in self.mobility_players.values():
mobility_player.close()
def set_observer(self, value: str):
def set_observer(self, value: Optional[str]) -> None:
self.observer = value
def read_config(self):
def read_config(self) -> None:
# read distributed servers
for server in self.app.guiconfig.servers:
self.servers[server.name] = server
@ -125,7 +147,7 @@ class CoreClient:
for observer in self.app.guiconfig.observers:
self.custom_observers[observer.name] = observer
def handle_events(self, event: core_pb2.Event):
def handle_events(self, event: Event) -> None:
if event.session_id != self.session_id:
logging.warning(
"ignoring event session(%s) current(%s)",
@ -139,7 +161,7 @@ class CoreClient:
elif event.HasField("session_event"):
logging.info("session event: %s", event)
session_event = event.session_event
if session_event.event <= core_pb2.SessionState.SHUTDOWN:
if session_event.event <= SessionState.SHUTDOWN:
self.state = event.session_event.event
elif session_event.event in {7, 8, 9}:
node_id = session_event.node_id
@ -162,7 +184,7 @@ class CoreClient:
else:
logging.info("unhandled event: %s", event)
def handle_link_event(self, event: core_pb2.LinkEvent):
def handle_link_event(self, event: LinkEvent) -> None:
logging.debug("Link event: %s", event)
node1_id = event.link.node1_id
node2_id = event.link.node2_id
@ -171,16 +193,16 @@ class CoreClient:
return
canvas_node1 = self.canvas_nodes[node1_id]
canvas_node2 = self.canvas_nodes[node2_id]
if event.message_type == core_pb2.MessageType.ADD:
if event.message_type == MessageType.ADD:
self.app.canvas.add_wireless_edge(canvas_node1, canvas_node2, event.link)
elif event.message_type == core_pb2.MessageType.DELETE:
elif event.message_type == MessageType.DELETE:
self.app.canvas.delete_wireless_edge(canvas_node1, canvas_node2, event.link)
elif event.message_type == core_pb2.MessageType.NONE:
elif event.message_type == MessageType.NONE:
self.app.canvas.update_wireless_edge(canvas_node1, canvas_node2, event.link)
else:
logging.warning("unknown link event: %s", event)
def handle_node_event(self, event: core_pb2.NodeEvent):
def handle_node_event(self, event: NodeEvent) -> None:
logging.debug("node event: %s", event)
if event.source == GUI_SOURCE:
return
@ -190,22 +212,22 @@ class CoreClient:
canvas_node = self.canvas_nodes[node_id]
canvas_node.move(x, y)
def enable_throughputs(self):
def enable_throughputs(self) -> None:
self.handling_throughputs = self.client.throughputs(
self.session_id, self.handle_throughputs
)
def cancel_throughputs(self):
def cancel_throughputs(self) -> None:
if self.handling_throughputs:
self.handling_throughputs.cancel()
self.handling_throughputs = None
def cancel_events(self):
def cancel_events(self) -> None:
if self.handling_events:
self.handling_events.cancel()
self.handling_events = None
def handle_throughputs(self, event: core_pb2.ThroughputsEvent):
def handle_throughputs(self, event: ThroughputsEvent) -> None:
if event.session_id != self.session_id:
logging.warning(
"ignoring throughput event session(%s) current(%s)",
@ -216,11 +238,11 @@ class CoreClient:
logging.debug("handling throughputs event: %s", event)
self.app.after(0, self.app.canvas.set_throughputs, event)
def handle_exception_event(self, event: core_pb2.ExceptionEvent):
def handle_exception_event(self, event: ExceptionEvent) -> None:
logging.info("exception event: %s", event)
self.app.statusbar.core_alarms.append(event)
def join_session(self, session_id: int, query_location: bool = True):
def join_session(self, session_id: int, query_location: bool = True) -> None:
logging.info("join session(%s)", session_id)
# update session and title
self.session_id = session_id
@ -331,9 +353,9 @@ class CoreClient:
self.app.after(0, self.app.joined_session_update)
def is_runtime(self) -> bool:
return self.state == core_pb2.SessionState.RUNTIME
return self.state == SessionState.RUNTIME
def parse_metadata(self, config: Dict[str, str]):
def parse_metadata(self, config: Dict[str, str]) -> None:
# canvas setting
canvas_config = config.get("canvas")
logging.debug("canvas metadata: %s", canvas_config)
@ -386,7 +408,7 @@ class CoreClient:
except ValueError:
logging.exception("unknown shape: %s", shape_type)
def create_new_session(self):
def create_new_session(self) -> None:
"""
Create a new session
"""
@ -394,7 +416,7 @@ class CoreClient:
response = self.client.create_session()
logging.info("created session: %s", response)
location_config = self.app.guiconfig.location
self.location = core_pb2.SessionLocation(
self.location = SessionLocation(
x=location_config.x,
y=location_config.y,
z=location_config.z,
@ -407,7 +429,7 @@ class CoreClient:
except grpc.RpcError as e:
self.app.show_grpc_exception("New Session Error", e)
def delete_session(self, session_id: int = None):
def delete_session(self, session_id: int = None) -> None:
if session_id is None:
session_id = self.session_id
try:
@ -416,7 +438,7 @@ class CoreClient:
except grpc.RpcError as e:
self.app.show_grpc_exception("Delete Session Error", e)
def setup(self):
def setup(self) -> None:
"""
Query sessions, if there exist any, prompt whether to join one
"""
@ -451,7 +473,7 @@ class CoreClient:
dialog.show()
self.app.close()
def edit_node(self, core_node: core_pb2.Node):
def edit_node(self, core_node: Node) -> None:
try:
self.client.edit_node(
self.session_id, core_node.id, core_node.position, source=GUI_SOURCE
@ -459,12 +481,12 @@ class CoreClient:
except grpc.RpcError as e:
self.app.show_grpc_exception("Edit Node Error", e)
def start_session(self) -> core_pb2.StartSessionResponse:
def start_session(self) -> StartSessionResponse:
self.ifaces_manager.reset_mac()
nodes = [x.core_node for x in self.canvas_nodes.values()]
links = []
for edge in self.links.values():
link = core_pb2.Link()
link = Link()
link.CopyFrom(edge.link)
if link.HasField("iface1") and not link.iface1.mac:
link.iface1.mac = self.ifaces_manager.next_mac()
@ -485,7 +507,7 @@ class CoreClient:
emane_config = {x: self.emane_config[x].value for x in self.emane_config}
else:
emane_config = None
response = core_pb2.StartSessionResponse(result=False)
response = StartSessionResponse(result=False)
try:
response = self.client.start_session(
self.session_id,
@ -511,10 +533,10 @@ class CoreClient:
self.app.show_grpc_exception("Start Session Error", e)
return response
def stop_session(self, session_id: int = None) -> core_pb2.StartSessionResponse:
def stop_session(self, session_id: int = None) -> StopSessionResponse:
if not session_id:
session_id = self.session_id
response = core_pb2.StopSessionResponse(result=False)
response = StopSessionResponse(result=False)
try:
response = self.client.stop_session(session_id)
logging.info("stopped session(%s), result: %s", session_id, response)
@ -522,9 +544,9 @@ class CoreClient:
self.app.show_grpc_exception("Stop Session Error", e)
return response
def show_mobility_players(self):
def show_mobility_players(self) -> None:
for canvas_node in self.canvas_nodes.values():
if canvas_node.core_node.type != core_pb2.NodeType.WIRELESS_LAN:
if canvas_node.core_node.type != NodeType.WIRELESS_LAN:
continue
if canvas_node.mobility_config:
mobility_player = MobilityPlayer(
@ -534,7 +556,7 @@ class CoreClient:
self.mobility_players[node_id] = mobility_player
mobility_player.show()
def set_metadata(self):
def set_metadata(self) -> None:
# create canvas data
wallpaper = None
if self.app.canvas.wallpaper_file:
@ -558,7 +580,7 @@ class CoreClient:
response = self.client.set_session_metadata(self.session_id, metadata)
logging.info("set session metadata %s, result: %s", metadata, response)
def launch_terminal(self, node_id: int):
def launch_terminal(self, node_id: int) -> None:
try:
terminal = self.app.guiconfig.preferences.terminal
if not terminal:
@ -575,12 +597,12 @@ class CoreClient:
except grpc.RpcError as e:
self.app.show_grpc_exception("Node Terminal Error", e)
def save_xml(self, file_path: str):
def save_xml(self, file_path: str) -> None:
"""
Save core session as to an xml file
"""
try:
if self.state != core_pb2.SessionState.RUNTIME:
if self.state != SessionState.RUNTIME:
logging.debug("Send session data to the daemon")
self.send_data()
response = self.client.save_xml(self.session_id, file_path)
@ -588,7 +610,7 @@ class CoreClient:
except grpc.RpcError as e:
self.app.show_grpc_exception("Save XML Error", e)
def open_xml(self, file_path: str):
def open_xml(self, file_path: str) -> None:
"""
Open core xml
"""
@ -627,7 +649,8 @@ class CoreClient:
shutdown=shutdowns,
)
logging.info(
"Set %s service for node(%s), files: %s, Startup: %s, Validation: %s, Shutdown: %s, Result: %s",
"Set %s service for node(%s), files: %s, Startup: %s, "
"Validation: %s, Shutdown: %s, Result: %s",
service_name,
node_id,
files,
@ -656,7 +679,7 @@ class CoreClient:
def set_node_service_file(
self, node_id: int, service_name: str, file_name: str, data: str
):
) -> None:
response = self.client.set_node_service_file(
self.session_id, node_id, service_name, file_name, data
)
@ -669,18 +692,16 @@ class CoreClient:
response,
)
def create_nodes_and_links(self):
def create_nodes_and_links(self) -> None:
"""
create nodes and links that have not been created yet
"""
node_protos = [x.core_node for x in self.canvas_nodes.values()]
link_protos = [x.link for x in self.links.values()]
if self.state != core_pb2.SessionState.DEFINITION:
self.client.set_session_state(
self.session_id, core_pb2.SessionState.DEFINITION
)
if self.state != SessionState.DEFINITION:
self.client.set_session_state(self.session_id, SessionState.DEFINITION)
self.client.set_session_state(self.session_id, core_pb2.SessionState.DEFINITION)
self.client.set_session_state(self.session_id, SessionState.DEFINITION)
for node_proto in node_protos:
response = self.client.add_node(self.session_id, node_proto)
logging.debug("create node: %s", response)
@ -695,7 +716,7 @@ class CoreClient:
)
logging.debug("create link: %s", response)
def send_data(self):
def send_data(self) -> None:
"""
send to daemon all session info, but don't start the session
"""
@ -738,10 +759,9 @@ class CoreClient:
if self.emane_config:
config = {x: self.emane_config[x].value for x in self.emane_config}
self.client.set_emane_config(self.session_id, config)
self.set_metadata()
def close(self):
def close(self) -> None:
"""
Clean ups when done using grpc
"""
@ -760,31 +780,31 @@ class CoreClient:
return i
def create_node(
self, x: float, y: float, node_type: core_pb2.NodeType, model: str
) -> Optional[core_pb2.Node]:
self, x: float, y: float, node_type: NodeType, model: str
) -> Optional[Node]:
"""
Add node, with information filled in, to grpc manager
"""
node_id = self.next_node_id()
position = core_pb2.Position(x=x, y=y)
position = Position(x=x, y=y)
image = None
if NodeUtils.is_image_node(node_type):
image = "ubuntu:latest"
emane = None
if node_type == core_pb2.NodeType.EMANE:
if node_type == NodeType.EMANE:
if not self.emane_models:
dialog = EmaneInstallDialog(self.app)
dialog.show()
return
emane = self.emane_models[0]
name = f"EMANE{node_id}"
elif node_type == core_pb2.NodeType.WIRELESS_LAN:
elif node_type == NodeType.WIRELESS_LAN:
name = f"WLAN{node_id}"
elif node_type in [core_pb2.NodeType.RJ45, core_pb2.NodeType.TUNNEL]:
elif node_type in [NodeType.RJ45, NodeType.TUNNEL]:
name = "UNASSIGNED"
else:
name = f"n{node_id}"
node = core_pb2.Node(
node = Node(
id=node_id,
type=node_type,
name=name,
@ -810,7 +830,7 @@ class CoreClient:
)
return node
def deleted_graph_nodes(self, canvas_nodes: List[core_pb2.Node]):
def deleted_graph_nodes(self, canvas_nodes: List[Node]) -> None:
"""
remove the nodes selected by the user and anything related to that node
such as link, configurations, interfaces
@ -826,14 +846,14 @@ class CoreClient:
links.append(edge.link)
self.ifaces_manager.removed(links)
def create_iface(self, canvas_node: CanvasNode) -> core_pb2.Interface:
def create_iface(self, canvas_node: CanvasNode) -> Interface:
node = canvas_node.core_node
ip4, ip6 = self.ifaces_manager.get_ips(node)
ip4_mask = self.ifaces_manager.ip4_mask
ip6_mask = self.ifaces_manager.ip6_mask
iface_id = canvas_node.next_iface_id()
name = f"eth{iface_id}"
iface = core_pb2.Interface(
iface = Interface(
id=iface_id,
name=name,
ip4=ip4,
@ -852,7 +872,7 @@ class CoreClient:
def create_link(
self, edge: CanvasEdge, canvas_src_node: CanvasNode, canvas_dst_node: CanvasNode
):
) -> None:
"""
Create core link for a pair of canvas nodes, with token referencing
the canvas edge.
@ -873,8 +893,8 @@ class CoreClient:
dst_iface = self.create_iface(canvas_dst_node)
self.iface_to_edge[(dst_node.id, dst_iface.id)] = edge.token
link = core_pb2.Link(
type=core_pb2.LinkType.WIRED,
link = Link(
type=LinkType.WIRED,
node1_id=src_node.id,
node2_id=dst_node.id,
iface1=src_iface,
@ -896,7 +916,7 @@ class CoreClient:
def get_wlan_configs_proto(self) -> List[WlanConfig]:
configs = []
for canvas_node in self.canvas_nodes.values():
if canvas_node.core_node.type != core_pb2.NodeType.WIRELESS_LAN:
if canvas_node.core_node.type != NodeType.WIRELESS_LAN:
continue
if not canvas_node.wlan_config:
continue
@ -910,7 +930,7 @@ class CoreClient:
def get_mobility_configs_proto(self) -> List[MobilityConfig]:
configs = []
for canvas_node in self.canvas_nodes.values():
if canvas_node.core_node.type != core_pb2.NodeType.WIRELESS_LAN:
if canvas_node.core_node.type != NodeType.WIRELESS_LAN:
continue
if not canvas_node.mobility_config:
continue
@ -924,7 +944,7 @@ class CoreClient:
def get_emane_model_configs_proto(self) -> List[EmaneModelConfig]:
configs = []
for canvas_node in self.canvas_nodes.values():
if canvas_node.core_node.type != core_pb2.NodeType.EMANE:
if canvas_node.core_node.type != NodeType.EMANE:
continue
node_id = canvas_node.core_node.id
for key, config in canvas_node.emane_model_configs.items():
@ -975,9 +995,7 @@ class CoreClient:
configs.append(config_proto)
return configs
def get_config_service_configs_proto(
self
) -> List[configservices_pb2.ConfigServiceConfig]:
def get_config_service_configs_proto(self) -> List[ConfigServiceConfig]:
config_service_protos = []
for canvas_node in self.canvas_nodes.values():
if not NodeUtils.is_container_node(canvas_node.core_node.type):
@ -987,7 +1005,7 @@ class CoreClient:
node_id = canvas_node.core_node.id
for name, service_config in canvas_node.config_service_configs.items():
config = service_config.get("config", {})
config_proto = configservices_pb2.ConfigServiceConfig(
config_proto = ConfigServiceConfig(
node_id=node_id,
name=name,
templates=service_config["templates"],
@ -1000,7 +1018,7 @@ class CoreClient:
logging.info("running node(%s) cmd: %s", node_id, self.observer)
return self.client.node_command(self.session_id, node_id, self.observer).output
def get_wlan_config(self, node_id: int) -> Dict[str, common_pb2.ConfigOption]:
def get_wlan_config(self, node_id: int) -> Dict[str, ConfigOption]:
response = self.client.get_wlan_config(self.session_id, node_id)
config = response.config
logging.debug(
@ -1010,7 +1028,7 @@ class CoreClient:
)
return dict(config)
def get_mobility_config(self, node_id: int) -> Dict[str, common_pb2.ConfigOption]:
def get_mobility_config(self, node_id: int) -> Dict[str, ConfigOption]:
response = self.client.get_mobility_config(self.session_id, node_id)
config = response.config
logging.debug(
@ -1022,7 +1040,7 @@ class CoreClient:
def get_emane_model_config(
self, node_id: int, model: str, iface_id: int = None
) -> Dict[str, common_pb2.ConfigOption]:
) -> Dict[str, ConfigOption]:
if iface_id is None:
iface_id = -1
response = self.client.get_emane_model_config(
@ -1030,7 +1048,8 @@ class CoreClient:
)
config = response.config
logging.debug(
"get emane model config: node id: %s, EMANE model: %s, interface: %s, config: %s",
"get emane model config: node id: %s, EMANE model: %s, "
"interface: %s, config: %s",
node_id,
model,
iface_id,
@ -1038,7 +1057,7 @@ class CoreClient:
)
return dict(config)
def execute_script(self, script):
def execute_script(self, script) -> None:
response = self.client.execute_script(script)
logging.info("execute python script %s", response)
if response.session_id != -1:

4
daemon/core/gui/graph/graph.py
View file

@ -2,7 +2,7 @@ import logging
import tkinter as tk
from copy import deepcopy
from tkinter import BooleanVar
from typing import TYPE_CHECKING, Tuple
from typing import TYPE_CHECKING, Optional, Tuple
from PIL import Image, ImageTk
@ -864,7 +864,7 @@ class CanvasGraph(tk.Canvas):
for tag in tags.ORGANIZE_TAGS:
self.tag_raise(tag)
def set_wallpaper(self, filename: str):
def set_wallpaper(self, filename: Optional[str]):
logging.debug("setting wallpaper: %s", filename)
if filename:
img = Image.open(filename)

56
daemon/core/gui/images.py
View file

@ -1,46 +1,44 @@
from enum import Enum
from tkinter import messagebox
from typing import Dict, Optional, Tuple
from PIL import Image, ImageTk
from PIL import Image
from PIL.ImageTk import PhotoImage
from core.api.grpc import core_pb2
from core.api.grpc.core_pb2 import NodeType
from core.gui.appconfig import LOCAL_ICONS_PATH
class Images:
images = {}
images: Dict[str, str] = {}
@classmethod
def create(cls, file_path: str, width: int, height: int = None):
def create(cls, file_path: str, width: int, height: int = None) -> PhotoImage:
if height is None:
height = width
image = Image.open(file_path)
image = image.resize((width, height), Image.ANTIALIAS)
return ImageTk.PhotoImage(image)
return PhotoImage(image)
@classmethod
def load_all(cls):
def load_all(cls) -> None:
for image in LOCAL_ICONS_PATH.glob("*"):
cls.images[image.stem] = str(image)
@classmethod
def get(
cls, image_enum: Enum, width: int, height: int = None
) -> ImageTk.PhotoImage:
def get(cls, image_enum: Enum, width: int, height: int = None) -> PhotoImage:
file_path = cls.images[image_enum.value]
return cls.create(file_path, width, height)
@classmethod
def get_with_image_file(
cls, stem: str, width: int, height: int = None
) -> ImageTk.PhotoImage:
) -> PhotoImage:
file_path = cls.images[stem]
return cls.create(file_path, width, height)
@classmethod
def get_custom(
cls, name: str, width: int, height: int = None
) -> ImageTk.PhotoImage:
def get_custom(cls, name: str, width: int, height: int = None) -> PhotoImage:
try:
file_path = cls.images[name]
return cls.create(file_path, width, height)
@ -95,22 +93,22 @@ class ImageEnum(Enum):
class TypeToImage:
type_to_image = {
(core_pb2.NodeType.DEFAULT, "router"): ImageEnum.ROUTER,
(core_pb2.NodeType.DEFAULT, "PC"): ImageEnum.PC,
(core_pb2.NodeType.DEFAULT, "host"): ImageEnum.HOST,
(core_pb2.NodeType.DEFAULT, "mdr"): ImageEnum.MDR,
(core_pb2.NodeType.DEFAULT, "prouter"): ImageEnum.PROUTER,
(core_pb2.NodeType.HUB, ""): ImageEnum.HUB,
(core_pb2.NodeType.SWITCH, ""): ImageEnum.SWITCH,
(core_pb2.NodeType.WIRELESS_LAN, ""): ImageEnum.WLAN,
(core_pb2.NodeType.EMANE, ""): ImageEnum.EMANE,
(core_pb2.NodeType.RJ45, ""): ImageEnum.RJ45,
(core_pb2.NodeType.TUNNEL, ""): ImageEnum.TUNNEL,
(core_pb2.NodeType.DOCKER, ""): ImageEnum.DOCKER,
(core_pb2.NodeType.LXC, ""): ImageEnum.LXC,
type_to_image: Dict[Tuple[NodeType, str], ImageEnum] = {
(NodeType.DEFAULT, "router"): ImageEnum.ROUTER,
(NodeType.DEFAULT, "PC"): ImageEnum.PC,
(NodeType.DEFAULT, "host"): ImageEnum.HOST,
(NodeType.DEFAULT, "mdr"): ImageEnum.MDR,
(NodeType.DEFAULT, "prouter"): ImageEnum.PROUTER,
(NodeType.HUB, ""): ImageEnum.HUB,
(NodeType.SWITCH, ""): ImageEnum.SWITCH,
(NodeType.WIRELESS_LAN, ""): ImageEnum.WLAN,
(NodeType.EMANE, ""): ImageEnum.EMANE,
(NodeType.RJ45, ""): ImageEnum.RJ45,
(NodeType.TUNNEL, ""): ImageEnum.TUNNEL,
(NodeType.DOCKER, ""): ImageEnum.DOCKER,
(NodeType.LXC, ""): ImageEnum.LXC,
}
@classmethod
def get(cls, node_type, model):
return cls.type_to_image.get((node_type, model), None)
def get(cls, node_type, model) -> Optional[ImageEnum]:
return cls.type_to_image.get((node_type, model))

40
daemon/core/gui/interface.py
View file

@ -1,18 +1,18 @@
import logging
from typing import TYPE_CHECKING, Any, List, Optional, Set, Tuple
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple
import netaddr
from netaddr import EUI, IPNetwork
from core.api.grpc.core_pb2 import Interface, Link, Node
from core.gui.graph.node import CanvasNode
from core.gui.nodeutils import NodeUtils
if TYPE_CHECKING:
from core.gui.app import Application
from core.api.grpc import core_pb2
from core.gui.graph.node import CanvasNode
def get_index(iface: "core_pb2.Interface") -> Optional[int]:
def get_index(iface: Interface) -> Optional[int]:
if not iface.ip4:
return None
net = netaddr.IPNetwork(f"{iface.ip4}/{iface.ip4_mask}")
@ -44,18 +44,18 @@ class Subnets:
class InterfaceManager:
def __init__(self, app: "Application") -> None:
self.app = app
self.app: "Application" = app
ip4 = self.app.guiconfig.ips.ip4
ip6 = self.app.guiconfig.ips.ip6
self.ip4_mask = 24
self.ip6_mask = 64
self.ip4_subnets = IPNetwork(f"{ip4}/{self.ip4_mask}")
self.ip6_subnets = IPNetwork(f"{ip6}/{self.ip6_mask}")
self.ip4_mask: int = 24
self.ip6_mask: int = 64
self.ip4_subnets: IPNetwork = IPNetwork(f"{ip4}/{self.ip4_mask}")
self.ip6_subnets: IPNetwork = IPNetwork(f"{ip6}/{self.ip6_mask}")
mac = self.app.guiconfig.mac
self.mac = EUI(mac, dialect=netaddr.mac_unix_expanded)
self.current_mac = None
self.current_subnets = None
self.used_subnets = {}
self.mac: EUI = EUI(mac, dialect=netaddr.mac_unix_expanded)
self.current_mac: Optional[EUI] = None
self.current_subnets: Optional[Subnets] = None
self.used_subnets: Dict[Tuple[IPNetwork, IPNetwork], Subnets] = {}
def update_ips(self, ip4: str, ip6: str) -> None:
self.reset()
@ -84,7 +84,7 @@ class InterfaceManager:
self.current_subnets = None
self.used_subnets.clear()
def removed(self, links: List["core_pb2.Link"]) -> None:
def removed(self, links: List[Link]) -> None:
# get remaining subnets
remaining_subnets = set()
for edge in self.app.core.links.values():
@ -114,7 +114,7 @@ class InterfaceManager:
subnets.used_indexes.discard(index)
self.current_subnets = None
def joined(self, links: List["core_pb2.Link"]) -> None:
def joined(self, links: List[Link]) -> None:
ifaces = []
for link in links:
if link.HasField("iface1"):
@ -132,7 +132,7 @@ class InterfaceManager:
if subnets.key() not in self.used_subnets:
self.used_subnets[subnets.key()] = subnets
def next_index(self, node: "core_pb2.Node") -> int:
def next_index(self, node: Node) -> int:
if NodeUtils.is_router_node(node):
index = 1
else:
@ -144,13 +144,13 @@ class InterfaceManager:
index += 1
return index
def get_ips(self, node: "core_pb2.Node") -> [str, str]:
def get_ips(self, node: Node) -> [str, str]:
index = self.next_index(node)
ip4 = self.current_subnets.ip4[index]
ip6 = self.current_subnets.ip6[index]
return str(ip4), str(ip6)
def get_subnets(self, iface: "core_pb2.Interface") -> Subnets:
def get_subnets(self, iface: Interface) -> Subnets:
ip4_subnet = self.ip4_subnets
if iface.ip4:
ip4_subnet = IPNetwork(f"{iface.ip4}/{iface.ip4_mask}").cidr
@ -161,7 +161,7 @@ class InterfaceManager:
return self.used_subnets.get(subnets.key(), subnets)
def determine_subnets(
self, canvas_src_node: "CanvasNode", canvas_dst_node: "CanvasNode"
self, canvas_src_node: CanvasNode, canvas_dst_node: CanvasNode
) -> None:
src_node = canvas_src_node.core_node
dst_node = canvas_dst_node.core_node
@ -185,7 +185,7 @@ class InterfaceManager:
logging.info("ignoring subnet change for link between network nodes")
def find_subnets(
self, canvas_node: "CanvasNode", visited: Set[int] = None
self, canvas_node: CanvasNode, visited: Set[int] = None
) -> Optional[IPNetwork]:
logging.info("finding subnet for node: %s", canvas_node.core_node.name)
canvas = self.app.canvas

18
daemon/core/gui/menubar.py
View file

@ -4,9 +4,10 @@ import tkinter as tk
import webbrowser
from functools import partial
from tkinter import filedialog, messagebox
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional
from core.gui.appconfig import XMLS_PATH
from core.gui.coreclient import CoreClient
from core.gui.dialogs.about import AboutDialog
from core.gui.dialogs.canvassizeandscale import SizeAndScaleDialog
from core.gui.dialogs.canvaswallpaper import CanvasWallpaperDialog
@ -22,6 +23,7 @@ from core.gui.dialogs.servers import ServersDialog
from core.gui.dialogs.sessionoptions import SessionOptionsDialog
from core.gui.dialogs.sessions import SessionsDialog
from core.gui.dialogs.throughput import ThroughputDialog
from core.gui.graph.graph import CanvasGraph
from core.gui.nodeutils import ICON_SIZE
from core.gui.observers import ObserversMenu
from core.gui.task import ProgressTask
@ -29,7 +31,7 @@ from core.gui.task import ProgressTask
if TYPE_CHECKING:
from core.gui.app import Application
MAX_FILES = 3
MAX_FILES: int = 3
class Menubar(tk.Menu):
@ -42,12 +44,12 @@ class Menubar(tk.Menu):
Create a CoreMenubar instance
"""
super().__init__(app)
self.app = app
self.core = app.core
self.canvas = app.canvas
self.recent_menu = None
self.edit_menu = None
self.observers_menu = None
self.app: "Application" = app
self.core: CoreClient = app.core
self.canvas: CanvasGraph = app.canvas
self.recent_menu: Optional[tk.Menu] = None
self.edit_menu: Optional[tk.Menu] = None
self.observers_menu: Optional[tk.Menu] = None
self.draw()
def draw(self) -> None:

60
daemon/core/gui/nodeutils.py
View file

@ -1,38 +1,36 @@
import logging
from typing import TYPE_CHECKING, List, Optional, Set
from typing import List, Optional, Set
from PIL.ImageTk import PhotoImage
from core.api.grpc.core_pb2 import Node, NodeType
from core.gui.appconfig import CustomNode, GuiConfig
from core.gui.images import ImageEnum, Images, TypeToImage
if TYPE_CHECKING:
from core.api.grpc import core_pb2
from PIL import ImageTk
ICON_SIZE = 48
ANTENNA_SIZE = 32
ICON_SIZE: int = 48
ANTENNA_SIZE: int = 32
class NodeDraw:
def __init__(self):
def __init__(self) -> None:
self.custom: bool = False
self.image = None
self.image: Optional[str] = None
self.image_enum: Optional[ImageEnum] = None
self.image_file = None
self.node_type: core_pb2.NodeType = None
self.image_file: Optional[str] = None
self.node_type: NodeType = None
self.model: Optional[str] = None
self.services: Set[str] = set()
self.label = None
self.label: Optional[str] = None
@classmethod
def from_setup(
cls,
image_enum: ImageEnum,
node_type: "core_pb2.NodeType",
node_type: NodeType,
label: str,
model: str = None,
tooltip=None,
):
tooltip: str = None,
) -> "NodeDraw":
node_draw = NodeDraw()
node_draw.image_enum = image_enum
node_draw.image = Images.get(image_enum, ICON_SIZE)
@ -43,7 +41,7 @@ class NodeDraw:
return node_draw
@classmethod
def from_custom(cls, custom_node: CustomNode):
def from_custom(cls, custom_node: CustomNode) -> "NodeDraw":
node_draw = NodeDraw()
node_draw.custom = True
node_draw.image_file = custom_node.image
@ -57,17 +55,17 @@ class NodeDraw:
class NodeUtils:
NODES = []
NETWORK_NODES = []
NODES: List[NodeDraw] = []
NETWORK_NODES: List[NodeDraw] = []
NODE_ICONS = {}
CONTAINER_NODES = {NodeType.DEFAULT, NodeType.DOCKER, NodeType.LXC}
IMAGE_NODES = {NodeType.DOCKER, NodeType.LXC}
WIRELESS_NODES = {NodeType.WIRELESS_LAN, NodeType.EMANE}
RJ45_NODES = {NodeType.RJ45}
IGNORE_NODES = {NodeType.CONTROL_NET, NodeType.PEER_TO_PEER}
NODE_MODELS = {"router", "host", "PC", "mdr", "prouter"}
ROUTER_NODES = {"router", "mdr"}
ANTENNA_ICON = None
CONTAINER_NODES: Set[NodeType] = {NodeType.DEFAULT, NodeType.DOCKER, NodeType.LXC}
IMAGE_NODES: Set[NodeType] = {NodeType.DOCKER, NodeType.LXC}
WIRELESS_NODES: Set[NodeType] = {NodeType.WIRELESS_LAN, NodeType.EMANE}
RJ45_NODES: Set[NodeType] = {NodeType.RJ45}
IGNORE_NODES: Set[NodeType] = {NodeType.CONTROL_NET, NodeType.PEER_TO_PEER}
NODE_MODELS: Set[str] = {"router", "host", "PC", "mdr", "prouter"}
ROUTER_NODES: Set[str] = {"router", "mdr"}
ANTENNA_ICON: PhotoImage = None
@classmethod
def is_router_node(cls, node: Node) -> bool:
@ -99,8 +97,8 @@ class NodeUtils:
@classmethod
def node_icon(
cls, node_type: NodeType, model: str, gui_config: GuiConfig, scale=1.0
) -> "ImageTk.PhotoImage":
cls, node_type: NodeType, model: str, gui_config: GuiConfig, scale: float = 1.0
) -> PhotoImage:
image_enum = TypeToImage.get(node_type, model)
if image_enum:
@ -112,8 +110,8 @@ class NodeUtils:
@classmethod
def node_image(
cls, core_node: "core_pb2.Node", gui_config: GuiConfig, scale=1.0
) -> "ImageTk.PhotoImage":
cls, core_node: Node, gui_config: GuiConfig, scale: float = 1.0
) -> PhotoImage:
image = cls.node_icon(core_node.type, core_node.model, gui_config, scale)
if core_node.icon:
try:
@ -141,7 +139,7 @@ class NodeUtils:
return None
@classmethod
def setup(cls):
def setup(cls) -> None:
nodes = [
(ImageEnum.ROUTER, NodeType.DEFAULT, "Router", "router"),
(ImageEnum.HOST, NodeType.DEFAULT, "Host", "host"),

10
daemon/core/gui/observers.py
View file

@ -1,13 +1,13 @@
import tkinter as tk
from functools import partial
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Dict
from core.gui.dialogs.observers import ObserverDialog
if TYPE_CHECKING:
from core.gui.app import Application
OBSERVERS = {
OBSERVERS: Dict[str, str] = {
"List Processes": "ps",
"Show Interfaces": "ip address",
"IPV4 Routes": "ip -4 route",
@ -23,9 +23,9 @@ OBSERVERS = {
class ObserversMenu(tk.Menu):
def __init__(self, master: tk.BaseWidget, app: "Application") -> None:
super().__init__(master)
self.app = app
self.observer = tk.StringVar(value=tk.NONE)
self.custom_index = 0
self.app: "Application" = app
self.observer: tk.StringVar = tk.StringVar(value=tk.NONE)
self.custom_index: int = 0
self.draw()
def draw(self) -> None:

28
daemon/core/gui/statusbar.py
View file

@ -3,8 +3,9 @@ status bar
"""
import tkinter as tk
from tkinter import ttk
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, List, Optional
from core.api.grpc.core_pb2 import ExceptionEvent
from core.gui.dialogs.alerts import AlertsDialog
from core.gui.themes import Styles
@ -13,20 +14,19 @@ if TYPE_CHECKING:
class StatusBar(ttk.Frame):
def __init__(self, master: tk.Widget, app: "Application"):
def __init__(self, master: tk.Widget, app: "Application") -> None:
super().__init__(master)
self.app = app
self.status = None
self.statusvar = tk.StringVar()
self.zoom = None
self.cpu_usage = None
self.memory = None
self.alerts_button = None
self.running = False
self.core_alarms = []
self.app: "Application" = app
self.status: Optional[ttk.Label] = None
self.statusvar: tk.StringVar = tk.StringVar()
self.zoom: Optional[ttk.Label] = None
self.cpu_usage: Optional[ttk.Label] = None
self.alerts_button: Optional[ttk.Button] = None
self.running: bool = False
self.core_alarms: List[ExceptionEvent] = []
self.draw()
def draw(self):
def draw(self) -> None:
self.columnconfigure(0, weight=7)
self.columnconfigure(1, weight=1)
self.columnconfigure(2, weight=1)
@ -64,9 +64,9 @@ class StatusBar(ttk.Frame):
)
self.alerts_button.grid(row=0, column=3, sticky="ew")
def click_alerts(self):
def click_alerts(self) -> None:
dialog = AlertsDialog(self.app)
dialog.show()
def set_status(self, message: str):
def set_status(self, message: str) -> None:
self.statusvar.set(message)

20
daemon/core/gui/task.py
View file

@ -1,7 +1,7 @@
import logging
import threading
import time
from typing import TYPE_CHECKING, Any, Callable, Tuple
from typing import TYPE_CHECKING, Any, Callable, Optional, Tuple
if TYPE_CHECKING:
from core.gui.app import Application
@ -16,14 +16,14 @@ class ProgressTask:
callback: Callable = None,
args: Tuple[Any] = None,
):
self.app = app
self.title = title
self.task = task
self.callback = callback
self.args = args
if self.args is None:
self.args = ()
self.time = None
self.app: "Application" = app
self.title: str = title
self.task: Callable = task
self.callback: Callable = callback
if args is None:
args = ()
self.args: Tuple[Any] = args
self.time: Optional[float] = None
def start(self) -> None:
self.app.progress.grid(sticky="ew")
@ -49,7 +49,7 @@ class ProgressTask:
finally:
self.app.after(0, self.complete)
def complete(self):
def complete(self) -> None:
self.app.progress.stop()
self.app.progress.grid_forget()
total = time.perf_counter() - self.time

61
daemon/core/gui/themes.py
View file

@ -1,39 +1,40 @@
import tkinter as tk
from tkinter import font, ttk
from typing import Dict, Tuple
THEME_DARK = "black"
PADX = (0, 5)
PADY = (0, 5)
FRAME_PAD = 5
DIALOG_PAD = 5
THEME_DARK: str = "black"
PADX: Tuple[int, int] = (0, 5)
PADY: Tuple[int, int] = (0, 5)
FRAME_PAD: int = 5
DIALOG_PAD: int = 5
class Styles:
tooltip = "Tooltip.TLabel"
tooltip_frame = "Tooltip.TFrame"
service_checkbutton = "Service.TCheckbutton"
picker_button = "Picker.TButton"
green_alert = "GAlert.TButton"
red_alert = "RAlert.TButton"
yellow_alert = "YAlert.TButton"
tooltip: str = "Tooltip.TLabel"
tooltip_frame: str = "Tooltip.TFrame"
service_checkbutton: str = "Service.TCheckbutton"
picker_button: str = "Picker.TButton"
green_alert: str = "GAlert.TButton"
red_alert: str = "RAlert.TButton"
yellow_alert: str = "YAlert.TButton"
class Colors:
disabledfg = "DarkGrey"
frame = "#424242"
dark = "#222222"
darker = "#121212"
darkest = "black"
lighter = "#626262"
lightest = "#ffffff"
selectbg = "#4a6984"
selectfg = "#ffffff"
white = "white"
black = "black"
listboxbg = "#f2f1f0"
disabledfg: str = "DarkGrey"
frame: str = "#424242"
dark: str = "#222222"
darker: str = "#121212"
darkest: str = "black"
lighter: str = "#626262"
lightest: str = "#ffffff"
selectbg: str = "#4a6984"
selectfg: str = "#ffffff"
white: str = "white"
black: str = "black"
listboxbg: str = "#f2f1f0"
def load(style: ttk.Style):
def load(style: ttk.Style) -> None:
style.theme_create(
THEME_DARK,
"clam",
@ -139,13 +140,13 @@ def load(style: ttk.Style):
)
def theme_change_menu(event: tk.Event):
def theme_change_menu(event: tk.Event) -> None:
if not isinstance(event.widget, tk.Menu):
return
style_menu(event.widget)
def style_menu(widget: tk.Widget):
def style_menu(widget: tk.Widget) -> None:
style = ttk.Style()
bg = style.lookup(".", "background")
fg = style.lookup(".", "foreground")
@ -157,7 +158,7 @@ def style_menu(widget: tk.Widget):
)
def style_listbox(widget: tk.Widget):
def style_listbox(widget: tk.Widget) -> None:
style = ttk.Style()
bg = style.lookup(".", "background")
fg = style.lookup(".", "foreground")
@ -174,7 +175,7 @@ def style_listbox(widget: tk.Widget):
)
def theme_change(event: tk.Event):
def theme_change(event: tk.Event) -> None:
style = ttk.Style()
style.configure(Styles.picker_button, font="TkSmallCaptionFont")
style.configure(
@ -203,7 +204,7 @@ def theme_change(event: tk.Event):
)
def scale_fonts(fonts_size, scale):
def scale_fonts(fonts_size: Dict[str, int], scale: float) -> None:
for name in font.names():
f = font.nametofont(name)
if name in fonts_size:

68
daemon/core/gui/toolbar.py
View file

@ -3,7 +3,7 @@ import tkinter as tk
from enum import Enum
from functools import partial
from tkinter import ttk
from typing import TYPE_CHECKING, Callable
from typing import TYPE_CHECKING, Callable, List, Optional
from PIL.ImageTk import PhotoImage
@ -23,8 +23,8 @@ from core.gui.tooltip import Tooltip
if TYPE_CHECKING:
from core.gui.app import Application
TOOLBAR_SIZE = 32
PICKER_SIZE = 24
TOOLBAR_SIZE: int = 32
PICKER_SIZE: int = 24
class NodeTypeEnum(Enum):
@ -42,8 +42,8 @@ def enable_buttons(frame: ttk.Frame, enabled: bool) -> None:
class PickerFrame(ttk.Frame):
def __init__(self, app: "Application", button: ttk.Button) -> None:
super().__init__(app)
self.app = app
self.button = button
self.app: "Application" = app
self.button: ttk.Button = button
def create_node_button(self, node_draw: NodeDraw, func: Callable) -> None:
self.create_button(
@ -85,10 +85,10 @@ class PickerFrame(ttk.Frame):
class ButtonBar(ttk.Frame):
def __init__(self, master: tk.Widget, app: "Application"):
def __init__(self, master: tk.Widget, app: "Application") -> None:
super().__init__(master)
self.app = app
self.radio_buttons = []
self.app: "Application" = app
self.radio_buttons: List[ttk.Button] = []
def create_button(
self, image_enum: ImageEnum, func: Callable, tooltip: str, radio: bool = False
@ -109,14 +109,14 @@ class ButtonBar(ttk.Frame):
class MarkerFrame(ttk.Frame):
PAD = 3
PAD: int = 3
def __init__(self, master: tk.BaseWidget, app: "Application") -> None:
super().__init__(master, padding=self.PAD)
self.app = app
self.color = "#000000"
self.size = tk.DoubleVar()
self.color_frame = None
self.app: "Application" = app
self.color: str = "#000000"
self.size: tk.DoubleVar = tk.DoubleVar()
self.color_frame: Optional[tk.Frame] = None
self.draw()
def draw(self) -> None:
@ -144,7 +144,7 @@ class MarkerFrame(ttk.Frame):
self.color_frame.bind("<Button-1>", self.click_color)
Tooltip(self.color_frame, "Marker Color")
def click_clear(self):
def click_clear(self) -> None:
self.app.canvas.delete(tags.MARKER)
def click_color(self, _event: tk.Event) -> None:
@ -163,37 +163,37 @@ class Toolbar(ttk.Frame):
Create a CoreToolbar instance
"""
super().__init__(app)
self.app = app
self.app: "Application" = app
# design buttons
self.play_button = None
self.select_button = None
self.link_button = None
self.node_button = None
self.network_button = None
self.annotation_button = None
self.play_button: Optional[ttk.Button] = None
self.select_button: Optional[ttk.Button] = None
self.link_button: Optional[ttk.Button] = None
self.node_button: Optional[ttk.Button] = None
self.network_button: Optional[ttk.Button] = None
self.annotation_button: Optional[ttk.Button] = None
# runtime buttons
self.runtime_select_button = None
self.stop_button = None
self.runtime_marker_button = None
self.run_command_button = None
self.runtime_select_button: Optional[ttk.Button] = None
self.stop_button: Optional[ttk.Button] = None
self.runtime_marker_button: Optional[ttk.Button] = None
self.run_command_button: Optional[ttk.Button] = None
# frames
self.design_frame = None
self.runtime_frame = None
self.marker_frame = None
self.picker = None
self.design_frame: Optional[ButtonBar] = None
self.runtime_frame: Optional[ButtonBar] = None
self.marker_frame: Optional[MarkerFrame] = None
self.picker: Optional[PickerFrame] = None
# observers
self.observers_menu = None
self.observers_menu: Optional[ObserversMenu] = None
# these variables help keep track of what images being drawn so that scaling
# is possible since PhotoImage does not have resize method
self.current_node = NodeUtils.NODES[0]
self.current_network = NodeUtils.NETWORK_NODES[0]
self.current_annotation = ShapeType.MARKER
self.annotation_enum = ImageEnum.MARKER
self.current_node: NodeDraw = NodeUtils.NODES[0]
self.current_network: NodeDraw = NodeUtils.NETWORK_NODES[0]
self.current_annotation: ShapeType = ShapeType.MARKER
self.annotation_enum: ImageEnum = ImageEnum.MARKER
# draw components
self.draw()

18
daemon/core/gui/tooltip.py
View file

@ -1,5 +1,6 @@
import tkinter as tk
from tkinter import ttk
from typing import Optional
from core.gui.themes import Styles
@ -9,19 +10,19 @@ class Tooltip(object):
Create tool tip for a given widget
"""
def __init__(self, widget: tk.Widget, text: str = "widget info"):
self.widget = widget
self.text = text
def __init__(self, widget: tk.BaseWidget, text: str = "widget info") -> None:
self.widget: tk.BaseWidget = widget
self.text: str = text
self.widget.bind("<Enter>", self.on_enter)
self.widget.bind("<Leave>", self.on_leave)
self.waittime = 400
self.id = None
self.tw = None
self.waittime: int = 400
self.id: Optional[str] = None
self.tw: Optional[tk.Toplevel] = None
def on_enter(self, event: tk.Event = None):
def on_enter(self, event: tk.Event = None) -> None:
self.schedule()
def on_leave(self, event: tk.Event = None):
def on_leave(self, event: tk.Event = None) -> None:
self.unschedule()
self.close(event)
@ -39,7 +40,6 @@ class Tooltip(object):
x, y, cx, cy = self.widget.bbox("insert")
x += self.widget.winfo_rootx()
y += self.widget.winfo_rooty() + 32
self.tw = tk.Toplevel(self.widget)
self.tw.wm_overrideredirect(True)
self.tw.wm_geometry("+%d+%d" % (x, y))

21
daemon/core/gui/validation.py
View file

@ -4,16 +4,23 @@ input validation
import re
import tkinter as tk
from tkinter import ttk
from typing import Any, Optional, Pattern
SMALLEST_SCALE = 0.5
LARGEST_SCALE = 5.0
HEX_REGEX = re.compile("^([#]([0-9]|[a-f])+)$|^[#]$")
SMALLEST_SCALE: float = 0.5
LARGEST_SCALE: float = 5.0
HEX_REGEX: Pattern = re.compile("^([#]([0-9]|[a-f])+)$|^[#]$")
class ValidationEntry(ttk.Entry):
empty = None
empty: Optional[str] = None
def __init__(self, master=None, widget=None, empty_enabled=True, **kwargs) -> None:
def __init__(
self,
master: tk.BaseWidget = None,
widget: tk.BaseWidget = None,
empty_enabled: bool = True,
**kwargs: Any
) -> None:
super().__init__(master, widget, **kwargs)
cmd = self.register(self.is_valid)
self.configure(validate="key", validatecommand=(cmd, "%P"))
@ -30,7 +37,7 @@ class ValidationEntry(ttk.Entry):
class PositiveIntEntry(ValidationEntry):
empty = "0"
empty: str = "0"
def is_valid(self, s: str) -> bool:
if not s:
@ -92,7 +99,7 @@ class HexEntry(ValidationEntry):
class NodeNameEntry(ValidationEntry):
empty = "noname"
empty: str = "noname"
def is_valid(self, s: str) -> bool:
if len(s) < 0:

102
daemon/core/gui/widgets.py
View file

@ -1,53 +1,63 @@
import logging
import tkinter as tk
from functools import partial
from pathlib import PosixPath
from pathlib import Path
from tkinter import filedialog, font, ttk
from typing import TYPE_CHECKING, Dict
from typing import TYPE_CHECKING, Any, Callable, Dict, Set, Type
from core.api.grpc import common_pb2, core_pb2
from core.api.grpc import core_pb2
from core.api.grpc.common_pb2 import ConfigOption
from core.api.grpc.core_pb2 import ConfigOptionType
from core.gui import themes, validation
from core.gui.dialogs.dialog import Dialog
from core.gui.themes import FRAME_PAD, PADX, PADY
if TYPE_CHECKING:
from core.gui.app import Application
from core.gui.dialogs.dialog import Dialog
INT_TYPES = {
core_pb2.ConfigOptionType.UINT8,
core_pb2.ConfigOptionType.UINT16,
core_pb2.ConfigOptionType.UINT32,
core_pb2.ConfigOptionType.UINT64,
core_pb2.ConfigOptionType.INT8,
core_pb2.ConfigOptionType.INT16,
core_pb2.ConfigOptionType.INT32,
core_pb2.ConfigOptionType.INT64,
INT_TYPES: Set[ConfigOptionType] = {
ConfigOptionType.UINT8,
ConfigOptionType.UINT16,
ConfigOptionType.UINT32,
ConfigOptionType.UINT64,
ConfigOptionType.INT8,
ConfigOptionType.INT16,
ConfigOptionType.INT32,
ConfigOptionType.INT64,
}
def file_button_click(value: tk.StringVar, parent: tk.Widget):
def file_button_click(value: tk.StringVar, parent: tk.Widget) -> None:
file_path = filedialog.askopenfilename(title="Select File", parent=parent)
if file_path:
value.set(file_path)
class FrameScroll(ttk.Frame):
def __init__(self, master: tk.Widget, app: "Application", _cls=ttk.Frame, **kw):
def __init__(
self,
master: tk.Widget,
app: "Application",
_cls: Type[ttk.Frame] = ttk.Frame,
**kw: Any
) -> None:
super().__init__(master, **kw)
self.app = app
self.app: "Application" = app
self.rowconfigure(0, weight=1)
self.columnconfigure(0, weight=1)
bg = self.app.style.lookup(".", "background")
self.canvas = tk.Canvas(self, highlightthickness=0, background=bg)
self.canvas: tk.Canvas = tk.Canvas(self, highlightthickness=0, background=bg)
self.canvas.grid(row=0, sticky="nsew", padx=2, pady=2)
self.canvas.columnconfigure(0, weight=1)
self.canvas.rowconfigure(0, weight=1)
self.scrollbar = ttk.Scrollbar(
self.scrollbar: ttk.Scrollbar = ttk.Scrollbar(
self, orient="vertical", command=self.canvas.yview
)
self.scrollbar.grid(row=0, column=1, sticky="ns")
self.frame = _cls(self.canvas)
self.frame_id = self.canvas.create_window(0, 0, anchor="nw", window=self.frame)
self.frame: ttk.Frame = _cls(self.canvas)
self.frame_id: int = self.canvas.create_window(
0, 0, anchor="nw", window=self.frame
)
self.canvas.update_idletasks()
self.canvas.configure(
scrollregion=self.canvas.bbox("all"), yscrollcommand=self.scrollbar.set
@ -55,16 +65,16 @@ class FrameScroll(ttk.Frame):
self.frame.bind("<Configure>", self._configure_frame)
self.canvas.bind("<Configure>", self._configure_canvas)
def _configure_frame(self, event: tk.Event):
def _configure_frame(self, event: tk.Event) -> None:
req_width = self.frame.winfo_reqwidth()
if req_width != self.canvas.winfo_reqwidth():
self.canvas.configure(width=req_width)
self.canvas.configure(scrollregion=self.canvas.bbox("all"))
def _configure_canvas(self, event: tk.Event):
def _configure_canvas(self, event: tk.Event) -> None:
self.canvas.itemconfig(self.frame_id, width=event.width)
def clear(self):
def clear(self) -> None:
for widget in self.frame.winfo_children():
widget.destroy()
@ -74,15 +84,15 @@ class ConfigFrame(ttk.Notebook):
self,
master: tk.Widget,
app: "Application",
config: Dict[str, common_pb2.ConfigOption],
**kw
):
config: Dict[str, ConfigOption],
**kw: Any
) -> None:
super().__init__(master, **kw)
self.app = app
self.config = config
self.values = {}
self.app: "Application" = app
self.config: Dict[str, ConfigOption] = config
self.values: Dict[str, tk.StringVar] = {}
def draw_config(self):
def draw_config(self) -> None:
group_mapping = {}
for key in self.config:
option = self.config[key]
@ -142,7 +152,7 @@ class ConfigFrame(ttk.Notebook):
logging.error("unhandled config option type: %s", option.type)
self.values[option.name] = value
def parse_config(self):
def parse_config(self) -> Dict[str, str]:
for key in self.config:
option = self.config[key]
value = self.values[key]
@ -169,13 +179,13 @@ class ConfigFrame(ttk.Notebook):
class ListboxScroll(ttk.Frame):
def __init__(self, master: tk.Widget = None, **kw):
def __init__(self, master: tk.BaseWidget = None, **kw: Any) -> None:
super().__init__(master, **kw)
self.columnconfigure(0, weight=1)
self.rowconfigure(0, weight=1)
self.scrollbar = ttk.Scrollbar(self, orient=tk.VERTICAL)
self.scrollbar: ttk.Scrollbar = ttk.Scrollbar(self, orient=tk.VERTICAL)
self.scrollbar.grid(row=0, column=1, sticky="ns")
self.listbox = tk.Listbox(
self.listbox: tk.Listbox = tk.Listbox(
self,
selectmode=tk.BROWSE,
yscrollcommand=self.scrollbar.set,
@ -187,12 +197,18 @@ class ListboxScroll(ttk.Frame):
class CheckboxList(FrameScroll):
def __init__(self, master: ttk.Widget, app: "Application", clicked=None, **kw):
def __init__(
self,
master: ttk.Widget,
app: "Application",
clicked: Callable = None,
**kw: Any
) -> None:
super().__init__(master, app, **kw)
self.clicked = clicked
self.clicked: Callable = clicked
self.frame.columnconfigure(0, weight=1)
def add(self, name: str, checked: bool):
def add(self, name: str, checked: bool) -> None:
var = tk.BooleanVar(value=checked)
func = partial(self.clicked, name, var)
checkbox = ttk.Checkbutton(self.frame, text=name, variable=var, command=func)
@ -200,16 +216,16 @@ class CheckboxList(FrameScroll):
class CodeFont(font.Font):
def __init__(self):
def __init__(self) -> None:
super().__init__(font="TkFixedFont", color="green")
class CodeText(ttk.Frame):
def __init__(self, master: tk.Widget, **kwargs):
def __init__(self, master: tk.BaseWidget, **kwargs: Any) -> None:
super().__init__(master, **kwargs)
self.rowconfigure(0, weight=1)
self.columnconfigure(0, weight=1)
self.text = tk.Text(
self.text: tk.Text = tk.Text(
self,
bd=0,
bg="black",
@ -229,14 +245,14 @@ class CodeText(ttk.Frame):
class Spinbox(ttk.Entry):
def __init__(self, master: tk.Widget = None, **kwargs):
def __init__(self, master: tk.BaseWidget = None, **kwargs: Any) -> None:
super().__init__(master, "ttk::spinbox", **kwargs)
def set(self, value):
def set(self, value: str) -> None:
self.tk.call(self._w, "set", value)
def image_chooser(parent: "Dialog", path: PosixPath):
def image_chooser(parent: Dialog, path: Path) -> str:
return filedialog.askopenfilename(
parent=parent,
initialdir=str(path),