import functools
import logging
import tkinter as tk
from pathlib import Path
from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple
import grpc
from PIL.ImageTk import PhotoImage
from core.api.grpc.services_pb2 import ServiceAction
from core.api.grpc.wrappers import Interface, Node, NodeType
from core.gui import images
from core.gui import nodeutils as nutils
from core.gui import themes
from core.gui.dialogs.emaneconfig import EmaneConfigDialog
from core.gui.dialogs.mobilityconfig import MobilityConfigDialog
from core.gui.dialogs.nodeconfig import NodeConfigDialog
from core.gui.dialogs.nodeconfigservice import NodeConfigServiceDialog
from core.gui.dialogs.nodeservice import NodeServiceDialog
from core.gui.dialogs.wlanconfig import WlanConfigDialog
from core.gui.frames.node import NodeInfoFrame
from core.gui.graph import tags
from core.gui.graph.edges import CanvasEdge, CanvasWirelessEdge
from core.gui.graph.tooltip import CanvasTooltip
from core.gui.images import ImageEnum
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from core.gui.app import Application
from core.gui.graph.graph import CanvasGraph
NODE_TEXT_OFFSET: int = 5
class CanvasNode:
def __init__(
self,
app: "Application",
canvas: "CanvasGraph",
x: float,
y: float,
core_node: Node,
image: PhotoImage,
):
self.app: "Application" = app
self.canvas: "CanvasGraph" = canvas
self.image: PhotoImage = image
self.core_node: Node = core_node
self.id: int = self.canvas.create_image(
x, y, anchor=tk.CENTER, image=self.image, tags=tags.NODE
)
label_y = self._get_label_y()
label = self.get_label()
self.text_id: int = self.canvas.create_text(
x,
label_y,
text=label,
tags=tags.NODE_LABEL,
font=self.app.icon_text_font,
fill="#0000CD",
state=self.app.manager.show_node_labels.state(),
)
self.tooltip: CanvasTooltip = CanvasTooltip(self.canvas)
self.edges: Set[CanvasEdge] = set()
self.ifaces: Dict[int, Interface] = {}
self.wireless_edges: Set[CanvasWirelessEdge] = set()
self.antennas: List[int] = []
self.antenna_images: Dict[int, PhotoImage] = {}
self.hidden: bool = False
self.setup_bindings()
self.context: tk.Menu = tk.Menu(self.canvas)
themes.style_menu(self.context)
def position(self) -> Tuple[int, int]:
return self.canvas.coords(self.id)
def next_iface_id(self) -> int:
i = 0
while i in self.ifaces:
i += 1
return i
def setup_bindings(self) -> None:
self.canvas.tag_bind(self.id, "<Double-Button-1>", self.double_click)
self.canvas.tag_bind(self.id, "<Enter>", self.on_enter)
self.canvas.tag_bind(self.id, "<Leave>", self.on_leave)
self.canvas.tag_bind(self.id, "<ButtonRelease-3>", self.show_context)
self.canvas.tag_bind(self.id, "<Button-1>", self.show_info)
def delete(self) -> None:
logger.debug("Delete canvas node for %s", self.core_node)
self.canvas.delete(self.id)
self.canvas.delete(self.text_id)
self.delete_antennas()
def add_antenna(self) -> None:
x, y = self.position()
offset = len(self.antennas) * 8 * self.app.app_scale
img = self.app.get_enum_icon(ImageEnum.ANTENNA, width=images.ANTENNA_SIZE)
antenna_id = self.canvas.create_image(
x - 16 + offset,
y - int(23 * self.app.app_scale),
anchor=tk.CENTER,
image=img,
tags=tags.ANTENNA,
)
self.antennas.append(antenna_id)
self.antenna_images[antenna_id] = img
def delete_antenna(self) -> None:
"""
delete one antenna
"""
logger.debug("Delete an antenna on %s", self.core_node.name)
if self.antennas:
antenna_id = self.antennas.pop()
self.canvas.delete(antenna_id)
self.antenna_images.pop(antenna_id, None)
def delete_antennas(self) -> None:
"""
delete all antennas
"""
logger.debug("Remove all antennas for %s", self.core_node.name)
for antenna_id in self.antennas:
self.canvas.delete(antenna_id)
self.antennas.clear()
self.antenna_images.clear()
def get_label(self) -> str:
label = self.core_node.name
if self.core_node.server:
label = f"{self.core_node.name}({self.core_node.server})"
return label
def redraw(self) -> None:
self.canvas.itemconfig(self.id, image=self.image)
label = self.get_label()
self.canvas.itemconfig(self.text_id, text=label)
for edge in self.edges:
edge.redraw()
def _get_label_y(self) -> int:
image_box = self.canvas.bbox(self.id)
return image_box[3] + NODE_TEXT_OFFSET
def scale_text(self) -> None:
text_bound = self.canvas.bbox(self.text_id)
prev_y = (text_bound[3] + text_bound[1]) / 2
new_y = self._get_label_y()
self.canvas.move(self.text_id, 0, new_y - prev_y)
def move(self, x: float, y: float) -> None:
x, y = self.canvas.get_scaled_coords(x, y)
current_x, current_y = self.position()
x_offset = x - current_x
y_offset = y - current_y
self.motion(x_offset, y_offset, update=False)
def motion(self, x_offset: float, y_offset: float, update: bool = True) -> None:
original_position = self.position()
self.canvas.move(self.id, x_offset, y_offset)
# check new position
bbox = self.canvas.bbox(self.id)
if not self.canvas.valid_position(*bbox):
self.canvas.coords(self.id, original_position)
return
# move test and selection box
self.canvas.move(self.text_id, x_offset, y_offset)
self.canvas.move_selection(self.id, x_offset, y_offset)
# move antennae
for antenna_id in self.antennas:
self.canvas.move(antenna_id, x_offset, y_offset)
# move edges
for edge in self.edges:
edge.move_node(self)
for edge in self.wireless_edges:
edge.move_node(self)
# set actual coords for node and update core is running
pos = self.position()
real_x, real_y = self.canvas.get_actual_coords(*pos)
self.core_node.position.x = real_x
self.core_node.position.y = real_y
if self.app.core.is_runtime() and update:
self.app.core.edit_node(self.core_node)
def on_enter(self, event: tk.Event) -> None:
is_runtime = self.app.core.is_runtime()
has_observer = self.app.core.observer is not None
is_container = nutils.is_container(self.core_node)
if is_runtime and has_observer and is_container:
self.tooltip.text.set("waiting...")
self.tooltip.on_enter(event)
try:
output = self.app.core.run(self.core_node.id)
self.tooltip.text.set(output)
except grpc.RpcError as e:
self.app.show_grpc_exception("Observer Error", e)
def on_leave(self, event: tk.Event) -> None:
self.tooltip.on_leave(event)
def double_click(self, event: tk.Event) -> None:
if self.app.core.is_runtime():
if nutils.is_container(self.core_node):
self.canvas.core.launch_terminal(self.core_node.id)
else:
self.show_config()
def show_info(self, _event: tk.Event) -> None:
self.app.display_info(NodeInfoFrame, app=self.app, canvas_node=self)
def show_context(self, event: tk.Event) -> None:
# clear existing menu
self.context.delete(0, tk.END)
is_wlan = self.core_node.type == NodeType.WIRELESS_LAN
is_emane = self.core_node.type == NodeType.EMANE
is_mobility = is_wlan or is_emane
if self.app.core.is_runtime():
self.context.add_command(label="Configure", command=self.show_config)
if is_emane:
self.context.add_command(
label="EMANE Config", command=self.show_emane_config
)
if is_wlan:
self.context.add_command(
label="WLAN Config", command=self.show_wlan_config
)
if is_mobility and self.core_node.id in self.app.core.mobility_players:
self.context.add_command(
label="Mobility Player", command=self.show_mobility_player
)
if nutils.is_container(self.core_node):
services_menu = tk.Menu(self.context)
for service in sorted(self.core_node.services):
service_menu = tk.Menu(services_menu)
themes.style_menu(service_menu)
start_func = functools.partial(self.start_service, service)
service_menu.add_command(label="Start", command=start_func)
stop_func = functools.partial(self.stop_service, service)
service_menu.add_command(label="Stop", command=stop_func)
restart_func = functools.partial(self.restart_service, service)
service_menu.add_command(label="Restart", command=restart_func)
validate_func = functools.partial(self.validate_service, service)
service_menu.add_command(label="Validate", command=validate_func)
services_menu.add_cascade(label=service, menu=service_menu)
themes.style_menu(services_menu)
self.context.add_cascade(label="Services", menu=services_menu)
else:
self.context.add_command(label="Configure", command=self.show_config)
if nutils.is_container(self.core_node):
self.context.add_command(label="Services", command=self.show_services)
self.context.add_command(
label="Config Services", command=self.show_config_services
)
if is_emane:
self.context.add_command(
label="EMANE Config", command=self.show_emane_config
)
if is_wlan:
self.context.add_command(
label="WLAN Config", command=self.show_wlan_config
)
if is_mobility:
self.context.add_command(
label="Mobility Config", command=self.show_mobility_config
)
if nutils.is_wireless(self.core_node):
self.context.add_command(
label="Link To Selected", command=self.wireless_link_selected
)
link_menu = tk.Menu(self.context)
for canvas in self.app.manager.all():
canvas_menu = tk.Menu(link_menu)
themes.style_menu(canvas_menu)
for node in canvas.nodes.values():
if not self.is_linkable(node):
continue
func_link = functools.partial(self.click_link, node)
canvas_menu.add_command(
label=node.core_node.name, command=func_link
)
link_menu.add_cascade(label=f"Canvas {canvas.id}", menu=canvas_menu)
themes.style_menu(link_menu)
self.context.add_cascade(label="Link", menu=link_menu)
unlink_menu = tk.Menu(self.context)
for edge in self.edges:
other_node = edge.other_node(self)
other_iface = edge.other_iface(self)
label = other_node.core_node.name
if other_iface:
label = f"{label}:{other_iface.name}"
func_unlink = functools.partial(self.click_unlink, edge)
unlink_menu.add_command(label=label, command=func_unlink)
themes.style_menu(unlink_menu)
self.context.add_cascade(label="Unlink", menu=unlink_menu)
edit_menu = tk.Menu(self.context)
themes.style_menu(edit_menu)
edit_menu.add_command(label="Cut", command=self.click_cut)
edit_menu.add_command(label="Copy", command=self.canvas_copy)
edit_menu.add_command(label="Delete", command=self.canvas_delete)
edit_menu.add_command(label="Hide", command=self.click_hide)
self.context.add_cascade(label="Edit", menu=edit_menu)
self.context.tk_popup(event.x_root, event.y_root)
def click_cut(self) -> None:
self.canvas_copy()
self.canvas_delete()
def click_hide(self) -> None:
self.canvas.clear_selection()
self.hide()
def click_unlink(self, edge: CanvasEdge) -> None:
edge.delete()
self.app.default_info()
def click_link(self, node: "CanvasNode") -> None:
edge = CanvasEdge(self.app, self, node)
edge.complete(node)
def canvas_delete(self) -> None:
self.canvas.clear_selection()
self.canvas.select_object(self.id)
self.canvas.delete_selected_objects()
def canvas_copy(self) -> None:
self.canvas.clear_selection()
self.canvas.select_object(self.id)
self.canvas.copy()
def show_config(self) -> None:
dialog = NodeConfigDialog(self.app, self)
dialog.show()
def show_wlan_config(self) -> None:
dialog = WlanConfigDialog(self.app, self)
if not dialog.has_error:
dialog.show()
def show_mobility_config(self) -> None:
dialog = MobilityConfigDialog(self.app, self.core_node)
if not dialog.has_error:
dialog.show()
def show_mobility_player(self) -> None:
mobility_player = self.app.core.mobility_players[self.core_node.id]
mobility_player.show()
def show_emane_config(self) -> None:
dialog = EmaneConfigDialog(self.app, self.core_node)
dialog.show()
def show_services(self) -> None:
dialog = NodeServiceDialog(self.app, self.core_node)
dialog.show()
def show_config_services(self) -> None:
dialog = NodeConfigServiceDialog(self.app, self.core_node)
dialog.show()
def has_emane_link(self, iface_id: int) -> Node:
result = None
for edge in self.edges:
other_node = edge.other_node(self)
iface = edge.iface(self)
edge_iface_id = iface.id if iface else None
if edge_iface_id != iface_id:
continue
if other_node.core_node.type == NodeType.EMANE:
result = other_node.core_node
break
return result
def wireless_link_selected(self) -> None:
nodes = [x for x in self.canvas.selection if x in self.canvas.nodes]
for node_id in nodes:
canvas_node = self.canvas.nodes[node_id]
self.canvas.create_edge(self, canvas_node)
self.canvas.clear_selection()
def scale_antennas(self) -> None:
for i in range(len(self.antennas)):
antenna_id = self.antennas[i]
image = self.app.get_enum_icon(ImageEnum.ANTENNA, width=images.ANTENNA_SIZE)
self.canvas.itemconfig(antenna_id, image=image)
self.antenna_images[antenna_id] = image
node_x, node_y = self.canvas.coords(self.id)
x, y = self.canvas.coords(antenna_id)
dx = node_x - 16 + (i * 8 * self.app.app_scale) - x
dy = node_y - int(23 * self.app.app_scale) - y
self.canvas.move(antenna_id, dx, dy)
def update_icon(self, icon_path: str) -> None:
if not Path(icon_path).exists():
logger.error(f"node icon does not exist: {icon_path}")
return
self.core_node.icon = icon_path
self.image = images.from_file(icon_path, width=images.NODE_SIZE)
self.canvas.itemconfig(self.id, image=self.image)
def is_linkable(self, node: "CanvasNode") -> bool:
# cannot link to self
if self == node:
return False
# rj45 nodes can only support one link
if nutils.is_rj45(self.core_node) and self.edges:
return False
if nutils.is_rj45(node.core_node) and node.edges:
return False
# only 1 link between bridge based nodes
is_src_bridge = nutils.is_bridge(self.core_node)
is_dst_bridge = nutils.is_bridge(node.core_node)
common_links = self.edges & node.edges
if all([is_src_bridge, is_dst_bridge, common_links]):
return False
# valid link
return True
def hide(self) -> None:
self.hidden = True
self.canvas.itemconfig(self.id, state=tk.HIDDEN)
self.canvas.itemconfig(self.text_id, state=tk.HIDDEN)
for antenna in self.antennas:
self.canvas.itemconfig(antenna, state=tk.HIDDEN)
for edge in self.edges:
if not edge.hidden:
edge.hide()
for edge in self.wireless_edges:
if not edge.hidden:
edge.hide()
def show(self) -> None:
self.hidden = False
self.canvas.itemconfig(self.id, state=tk.NORMAL)
state = self.app.manager.show_node_labels.state()
self.set_label(state)
for antenna in self.antennas:
self.canvas.itemconfig(antenna, state=tk.NORMAL)
for edge in self.edges:
other_node = edge.other_node(self)
if edge.hidden and not other_node.hidden:
edge.show()
for edge in self.wireless_edges:
other_node = edge.other_node(self)
if edge.hidden and not other_node.hidden:
edge.show()
def set_label(self, state: str) -> None:
self.canvas.itemconfig(self.text_id, state=state)
def _service_action(self, service: str, action: ServiceAction) -> None:
session_id = self.app.core.session.id
try:
result = self.app.core.client.service_action(
session_id, self.core_node.id, service, action
)
if not result:
self.app.show_error("Service Action Error", "Action Failed!")
except grpc.RpcError as e:
self.app.show_grpc_exception("Service Error", e)
def start_service(self, service: str) -> None:
self._service_action(service, ServiceAction.START)
def stop_service(self, service: str) -> None:
self._service_action(service, ServiceAction.STOP)
def restart_service(self, service: str) -> None:
self._service_action(service, ServiceAction.RESTART)
def validate_service(self, service: str) -> None:
self._service_action(service, ServiceAction.VALIDATE)
def is_wireless(self) -> bool:
return nutils.is_wireless(self.core_node)
class ShadowNode:
def __init__(
self, app: "Application", canvas: "CanvasGraph", node: "CanvasNode"
) -> None:
self.app: "Application" = app
self.canvas: "CanvasGraph" = canvas
self.node: "CanvasNode" = node
self.id: Optional[int] = None
self.text_id: Optional[int] = None
self.image: PhotoImage = self.app.get_enum_icon(
ImageEnum.SHADOW, width=images.NODE_SIZE
)
self.draw()
self.setup_bindings()
def setup_bindings(self) -> None:
self.canvas.tag_bind(self.id, "<Double-Button-1>", self.node.double_click)
self.canvas.tag_bind(self.id, "<Enter>", self.node.on_enter)
self.canvas.tag_bind(self.id, "<Leave>", self.node.on_leave)
self.canvas.tag_bind(self.id, "<ButtonRelease-3>", self.node.show_context)
self.canvas.tag_bind(self.id, "<Button-1>", self.node.show_info)
def draw(self) -> None:
x, y = self.node.position()
self.id: int = self.canvas.create_image(
x, y, anchor=tk.CENTER, image=self.image, tags=tags.NODE
)
self.text_id = self.canvas.create_text(
x,
y + 20,
text=f"{self.node.get_label()} [{self.node.canvas.id}]",
tags=tags.NODE_LABEL,
font=self.app.icon_text_font,
fill="#0000CD",
state=self.app.manager.show_node_labels.state(),
justify=tk.CENTER,
)
self.canvas.shadow_nodes[self.id] = self
self.canvas.shadow_core_nodes[self.node.core_node.id] = self
def position(self) -> Tuple[int, int]:
return self.canvas.coords(self.id)
def should_delete(self) -> bool:
for edge in self.node.edges:
other_node = edge.other_node(self.node)
if not other_node.is_wireless() and other_node.canvas == self.canvas:
return False
return True
def motion(self, x_offset, y_offset) -> None:
original_position = self.position()
self.canvas.move(self.id, x_offset, y_offset)
# check new position
bbox = self.canvas.bbox(self.id)
if not self.canvas.valid_position(*bbox):
self.canvas.coords(self.id, original_position)
return
# move text and selection box
self.canvas.move(self.text_id, x_offset, y_offset)
self.canvas.move_selection(self.id, x_offset, y_offset)
# move edges
for edge in self.node.edges:
edge.move_shadow(self)
for edge in self.node.wireless_edges:
edge.move_shadow(self)
def delete(self):
self.canvas.shadow_nodes.pop(self.id, None)
self.canvas.shadow_core_nodes.pop(self.node.core_node.id, None)
self.canvas.delete(self.id)
self.canvas.delete(self.text_id)
def hide(self) -> None:
self.canvas.itemconfig(self.id, state=tk.HIDDEN)
self.canvas.itemconfig(self.text_id, state=tk.HIDDEN)
def show(self) -> None:
self.canvas.itemconfig(self.id, state=tk.NORMAL)
self.canvas.itemconfig(self.text_id, state=tk.NORMAL)