pygui: adjustments to have canvas manager manage all edges, allow shadow nodes to be moved, and updates to account for old usages of a universal canvas
This commit is contained in:
Blake Harnden
parent
f171c6111a
commit
3e2ea42ebd
7 changed files with 605 additions and 411 deletions
|
|
@ -204,27 +204,26 @@ class CoreClient:
|
|||
canvas_node2 = self.canvas_nodes[node2_id]
|
||||
if event.link.type == LinkType.WIRELESS:
|
||||
if event.message_type == MessageType.ADD:
|
||||
self.app.canvas.add_wireless_edge(
|
||||
self.app.manager.add_wireless_edge(
|
||||
canvas_node1, canvas_node2, event.link
|
||||
)
|
||||
elif event.message_type == MessageType.DELETE:
|
||||
self.app.canvas.delete_wireless_edge(
|
||||
self.app.manager.delete_wireless_edge(
|
||||
canvas_node1, canvas_node2, event.link
|
||||
)
|
||||
elif event.message_type == MessageType.NONE:
|
||||
self.app.canvas.update_wireless_edge(
|
||||
self.app.manager.update_wireless_edge(
|
||||
canvas_node1, canvas_node2, event.link
|
||||
)
|
||||
else:
|
||||
logging.warning("unknown link event: %s", event)
|
||||
else:
|
||||
if event.message_type == MessageType.ADD:
|
||||
self.app.canvas.add_wired_edge(canvas_node1, canvas_node2, event.link)
|
||||
self.app.canvas.organize()
|
||||
self.app.manager.add_wired_edge(canvas_node1, canvas_node2, event.link)
|
||||
elif event.message_type == MessageType.DELETE:
|
||||
self.app.canvas.delete_wired_edge(event.link)
|
||||
self.app.manager.delete_wired_edge(event.link)
|
||||
elif event.message_type == MessageType.NONE:
|
||||
self.app.canvas.update_wired_edge(event.link)
|
||||
self.app.manager.update_wired_edge(event.link)
|
||||
else:
|
||||
logging.warning("unknown link event: %s", event)
|
||||
|
||||
|
|
@ -246,7 +245,7 @@ class CoreClient:
|
|||
elif event.message_type == MessageType.ADD:
|
||||
if node.id in self.session.nodes:
|
||||
logging.error("core node already exists: %s", node)
|
||||
self.app.canvas.add_core_node(node)
|
||||
self.app.manager.add_core_node(node)
|
||||
else:
|
||||
logging.warning("unknown node event: %s", event)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,18 +1,25 @@
|
|||
import functools
|
||||
import logging
|
||||
import math
|
||||
import tkinter as tk
|
||||
from typing import TYPE_CHECKING, Optional, Tuple
|
||||
|
||||
from PIL.ImageTk import PhotoImage
|
||||
|
||||
from core.api.grpc.wrappers import Interface, Link
|
||||
from core.gui import themes
|
||||
from core.gui.dialogs.linkconfig import LinkConfigurationDialog
|
||||
from core.gui.frames.link import EdgeInfoFrame, WirelessEdgeInfoFrame
|
||||
from core.gui.graph import tags
|
||||
from core.gui.nodeutils import NodeUtils
|
||||
from core.gui.images import ImageEnum
|
||||
from core.gui.nodeutils import ICON_SIZE, NodeUtils
|
||||
from core.gui.utils import bandwidth_text, delay_jitter_text
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from core.gui.app import Application
|
||||
from core.gui.graph.graph import CanvasGraph
|
||||
from core.gui.graph.manager import CanvasManager
|
||||
from core.gui.graph.node import CanvasNode
|
||||
|
||||
TEXT_DISTANCE: int = 60
|
||||
EDGE_WIDTH: int = 3
|
||||
|
|
@ -33,6 +40,19 @@ def create_edge_token(link: Link) -> str:
|
|||
return f"{link.node1_id}-{iface1_id}-{link.node2_id}-{iface2_id}"
|
||||
|
||||
|
||||
def node_label_positions(
|
||||
src_x: int, src_y: int, dst_x: int, dst_y: int
|
||||
) -> Tuple[Tuple[float, float], Tuple[float, float]]:
|
||||
v_x, v_y = dst_x - src_x, dst_y - src_y
|
||||
v_len = math.sqrt(v_x ** 2 + v_y ** 2)
|
||||
if v_len == 0:
|
||||
u_x, u_y = 0.0, 0.0
|
||||
else:
|
||||
u_x, u_y = v_x / v_len, v_y / v_len
|
||||
offset_x, offset_y = TEXT_DISTANCE * u_x, TEXT_DISTANCE * u_y
|
||||
return ((src_x + offset_x, src_y + offset_y), (dst_x - offset_x, dst_y - offset_y))
|
||||
|
||||
|
||||
def arc_edges(edges) -> None:
|
||||
if not edges:
|
||||
return
|
||||
|
|
@ -62,24 +82,91 @@ def arc_edges(edges) -> None:
|
|||
edge.redraw()
|
||||
|
||||
|
||||
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_icon(ImageEnum.ROUTER, ICON_SIZE)
|
||||
self.draw()
|
||||
|
||||
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
|
||||
|
||||
def position(self) -> Tuple[int, int]:
|
||||
return self.canvas.coords(self.id)
|
||||
|
||||
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.delete(self.id)
|
||||
self.canvas.delete(self.text_id)
|
||||
|
||||
|
||||
class Edge:
|
||||
tag: str = tags.EDGE
|
||||
|
||||
def __init__(self, canvas: "CanvasGraph", src: int, dst: int = None) -> None:
|
||||
self.canvas = canvas
|
||||
def __init__(
|
||||
self, app: "Application", src: "CanvasNode", dst: "CanvasNode" = None
|
||||
) -> None:
|
||||
self.app: "Application" = app
|
||||
self.manager: CanvasManager = app.manager
|
||||
self.id: Optional[int] = None
|
||||
self.src: int = src
|
||||
self.dst: int = dst
|
||||
self.id2: Optional[int] = None
|
||||
self.src: "CanvasNode" = src
|
||||
self.src_shadow: Optional[ShadowNode] = None
|
||||
self.dst: Optional["CanvasNode"] = dst
|
||||
self.dst_shadow: Optional[ShadowNode] = None
|
||||
self.arc: int = 0
|
||||
self.token: Optional[str] = None
|
||||
self.src_label: Optional[int] = None
|
||||
self.src_label2: Optional[int] = None
|
||||
self.middle_label: Optional[int] = None
|
||||
self.middle_label2: Optional[int] = None
|
||||
self.dst_label: Optional[int] = None
|
||||
self.dst_label2: Optional[int] = None
|
||||
self.color: str = EDGE_COLOR
|
||||
self.width: int = EDGE_WIDTH
|
||||
|
||||
def scaled_width(self) -> float:
|
||||
return self.width * self.canvas.app.app_scale
|
||||
return self.width * self.app.app_scale
|
||||
|
||||
def _get_arcpoint(
|
||||
self, src_pos: Tuple[float, float], dst_pos: Tuple[float, float]
|
||||
|
|
@ -110,52 +197,126 @@ class Edge:
|
|||
arc_y = (perp_m * arc_x) + b
|
||||
return arc_x, arc_y
|
||||
|
||||
def draw(
|
||||
self, src_pos: Tuple[float, float], dst_pos: Tuple[float, float], state: str
|
||||
) -> None:
|
||||
arc_pos = self._get_arcpoint(src_pos, dst_pos)
|
||||
self.id = self.canvas.create_line(
|
||||
*src_pos,
|
||||
*arc_pos,
|
||||
*dst_pos,
|
||||
smooth=True,
|
||||
tags=self.tag,
|
||||
width=self.scaled_width(),
|
||||
fill=self.color,
|
||||
state=state,
|
||||
)
|
||||
def arc_common_edges(self) -> None:
|
||||
common_edges = list(self.src.edges & self.dst.edges)
|
||||
common_edges += list(self.src.wireless_edges & self.dst.wireless_edges)
|
||||
arc_edges(common_edges)
|
||||
|
||||
def is_same_canvas(self) -> bool:
|
||||
# actively drawing same canvas link
|
||||
if not self.dst:
|
||||
return True
|
||||
return self.src.canvas == self.dst.canvas
|
||||
|
||||
def draw(self, state: str) -> None:
|
||||
src_pos = self.src.position()
|
||||
if self.is_same_canvas():
|
||||
dst_pos = src_pos
|
||||
if self.dst:
|
||||
dst_pos = self.dst.position()
|
||||
arc_pos = self._get_arcpoint(src_pos, dst_pos)
|
||||
self.id = self.src.canvas.create_line(
|
||||
*src_pos,
|
||||
*arc_pos,
|
||||
*dst_pos,
|
||||
smooth=True,
|
||||
tags=self.tag,
|
||||
width=self.scaled_width(),
|
||||
fill=self.color,
|
||||
state=state,
|
||||
)
|
||||
else:
|
||||
# draw shadow nodes and 2 lines
|
||||
dst_pos = self.dst.position()
|
||||
arc_pos = self._get_arcpoint(src_pos, dst_pos)
|
||||
self.src_shadow = ShadowNode(self.app, self.dst.canvas, self.src)
|
||||
self.dst_shadow = ShadowNode(self.app, self.src.canvas, self.dst)
|
||||
self.id = self.src.canvas.create_line(
|
||||
*src_pos,
|
||||
*arc_pos,
|
||||
*dst_pos,
|
||||
smooth=True,
|
||||
tags=self.tag,
|
||||
width=self.scaled_width(),
|
||||
fill=self.color,
|
||||
state=state,
|
||||
)
|
||||
self.id2 = self.dst.canvas.create_line(
|
||||
*src_pos,
|
||||
*arc_pos,
|
||||
*dst_pos,
|
||||
smooth=True,
|
||||
tags=self.tag,
|
||||
width=self.scaled_width(),
|
||||
fill=self.color,
|
||||
state=state,
|
||||
)
|
||||
|
||||
def redraw(self) -> None:
|
||||
self.canvas.itemconfig(self.id, width=self.scaled_width(), fill=self.color)
|
||||
src_x, src_y, _, _, _, _ = self.canvas.coords(self.id)
|
||||
src_pos = src_x, src_y
|
||||
self.move_src(src_pos)
|
||||
|
||||
def middle_label_pos(self) -> Tuple[float, float]:
|
||||
_, _, x, y, _, _ = self.canvas.coords(self.id)
|
||||
return x, y
|
||||
self.src.canvas.itemconfig(self.id, width=self.scaled_width(), fill=self.color)
|
||||
# src_x, src_y, _, _, _, _ = self.src.canvas.coords(self.id)
|
||||
# src_pos = src_x, src_y
|
||||
self.move_src()
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.itemconfig(
|
||||
self.id2, width=self.scaled_width(), fill=self.color
|
||||
)
|
||||
# src_x, src_y, _, _, _, _ = self.dst.canvas.coords(self.id2)
|
||||
# src_pos = src_x, src_y
|
||||
# self.move_src(src_pos)
|
||||
self.move_dst()
|
||||
|
||||
def middle_label_text(self, text: str) -> None:
|
||||
if self.middle_label is None:
|
||||
x, y = self.middle_label_pos()
|
||||
self.middle_label = self.canvas.create_text(
|
||||
_, _, x, y, _, _ = self.src.canvas.coords(self.id)
|
||||
self.middle_label = self.src.canvas.create_text(
|
||||
x,
|
||||
y,
|
||||
font=self.canvas.app.edge_font,
|
||||
font=self.app.edge_font,
|
||||
text=text,
|
||||
tags=tags.LINK_LABEL,
|
||||
justify=tk.CENTER,
|
||||
state=self.canvas.manager.show_link_labels.state(),
|
||||
state=self.manager.show_link_labels.state(),
|
||||
)
|
||||
if not self.is_same_canvas():
|
||||
_, _, x, y, _, _ = self.dst.canvas.coords(self.id2)
|
||||
self.middle_label2 = self.dst.canvas.create_text(
|
||||
x,
|
||||
y,
|
||||
font=self.app.edge_font,
|
||||
text=text,
|
||||
tags=tags.LINK_LABEL,
|
||||
justify=tk.CENTER,
|
||||
state=self.manager.show_link_labels.state(),
|
||||
)
|
||||
else:
|
||||
self.canvas.itemconfig(self.middle_label, text=text)
|
||||
self.src.canvas.itemconfig(self.middle_label, text=text)
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.itemconfig(self.middle_label2, text=text)
|
||||
|
||||
def clear_middle_label(self) -> None:
|
||||
self.canvas.delete(self.middle_label)
|
||||
self.src.canvas.delete(self.middle_label)
|
||||
self.middle_label = None
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.delete(self.middle_label2)
|
||||
self.middle_label2 = None
|
||||
|
||||
def node_label_positions(self) -> Tuple[Tuple[float, float], Tuple[float, float]]:
|
||||
src_x, src_y, _, _, dst_x, dst_y = self.canvas.coords(self.id)
|
||||
src_x, src_y, _, _, dst_x, dst_y = self.src.canvas.coords(self.id)
|
||||
v_x, v_y = dst_x - src_x, dst_y - src_y
|
||||
v_len = math.sqrt(v_x ** 2 + v_y ** 2)
|
||||
if v_len == 0:
|
||||
u_x, u_y = 0.0, 0.0
|
||||
else:
|
||||
u_x, u_y = v_x / v_len, v_y / v_len
|
||||
offset_x, offset_y = TEXT_DISTANCE * u_x, TEXT_DISTANCE * u_y
|
||||
return (
|
||||
(src_x + offset_x, src_y + offset_y),
|
||||
(dst_x - offset_x, dst_y - offset_y),
|
||||
)
|
||||
|
||||
def node_label_positions2(self) -> Tuple[Tuple[float, float], Tuple[float, float]]:
|
||||
src_x, src_y, _, _, dst_x, dst_y = self.dst.canvas.coords(self.id2)
|
||||
v_x, v_y = dst_x - src_x, dst_y - src_y
|
||||
v_len = math.sqrt(v_x ** 2 + v_y ** 2)
|
||||
if v_len == 0:
|
||||
|
|
@ -170,68 +331,148 @@ class Edge:
|
|||
|
||||
def src_label_text(self, text: str) -> None:
|
||||
if self.src_label is None:
|
||||
src_pos, _ = self.node_label_positions()
|
||||
self.src_label = self.canvas.create_text(
|
||||
src_x, src_y, _, _, dst_x, dst_y = self.src.canvas.coords(self.id)
|
||||
src_pos, _ = node_label_positions(src_x, src_y, dst_x, dst_y)
|
||||
self.src_label = self.src.canvas.create_text(
|
||||
*src_pos,
|
||||
text=text,
|
||||
justify=tk.CENTER,
|
||||
font=self.canvas.app.edge_font,
|
||||
font=self.app.edge_font,
|
||||
tags=tags.LINK_LABEL,
|
||||
state=self.canvas.manager.show_link_labels.state(),
|
||||
state=self.manager.show_link_labels.state(),
|
||||
)
|
||||
if not self.is_same_canvas():
|
||||
src_x, src_y, _, _, dst_x, dst_y = self.dst.canvas.coords(self.id2)
|
||||
src_pos, _ = node_label_positions(src_x, src_y, dst_x, dst_y)
|
||||
self.src_label2 = self.dst.canvas.create_text(
|
||||
*src_pos,
|
||||
text=text,
|
||||
justify=tk.CENTER,
|
||||
font=self.app.edge_font,
|
||||
tags=tags.LINK_LABEL,
|
||||
state=self.manager.show_link_labels.state(),
|
||||
)
|
||||
else:
|
||||
self.canvas.itemconfig(self.src_label, text=text)
|
||||
self.src.canvas.itemconfig(self.src_label, text=text)
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.itemconfig(self.src_label2, text=text)
|
||||
|
||||
def dst_label_text(self, text: str) -> None:
|
||||
if self.dst_label is None:
|
||||
_, dst_pos = self.node_label_positions()
|
||||
self.dst_label = self.canvas.create_text(
|
||||
src_x, src_y, _, _, dst_x, dst_y = self.src.canvas.coords(self.id)
|
||||
_, dst_pos = node_label_positions(src_x, src_y, dst_x, dst_y)
|
||||
self.dst_label = self.src.canvas.create_text(
|
||||
*dst_pos,
|
||||
text=text,
|
||||
justify=tk.CENTER,
|
||||
font=self.canvas.app.edge_font,
|
||||
font=self.app.edge_font,
|
||||
tags=tags.LINK_LABEL,
|
||||
state=self.canvas.manager.show_link_labels.state(),
|
||||
state=self.manager.show_link_labels.state(),
|
||||
)
|
||||
if not self.is_same_canvas():
|
||||
src_x, src_y, _, _, dst_x, dst_y = self.dst.canvas.coords(self.id2)
|
||||
_, dst_pos = node_label_positions(src_x, src_y, dst_x, dst_y)
|
||||
self.dst_label2 = self.dst.canvas.create_text(
|
||||
*dst_pos,
|
||||
text=text,
|
||||
justify=tk.CENTER,
|
||||
font=self.app.edge_font,
|
||||
tags=tags.LINK_LABEL,
|
||||
state=self.manager.show_link_labels.state(),
|
||||
)
|
||||
else:
|
||||
self.canvas.itemconfig(self.dst_label, text=text)
|
||||
self.src.canvas.itemconfig(self.dst_label, text=text)
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.itemconfig(self.dst_label2, text=text)
|
||||
|
||||
def move_node(self, node_id: int, pos: Tuple[float, float]) -> None:
|
||||
if self.src == node_id:
|
||||
self.move_src(pos)
|
||||
else:
|
||||
self.move_dst(pos)
|
||||
|
||||
def move_dst(self, dst_pos: Tuple[float, float]) -> None:
|
||||
src_x, src_y, _, _, _, _ = self.canvas.coords(self.id)
|
||||
def drawing(self, pos: Tuple[float, float]) -> None:
|
||||
src_x, src_y, _, _, _, _ = self.src.canvas.coords(self.id)
|
||||
src_pos = src_x, src_y
|
||||
self.moved(src_pos, dst_pos)
|
||||
self.moved(src_pos, pos)
|
||||
|
||||
def move_src(self, src_pos: Tuple[float, float]) -> None:
|
||||
_, _, _, _, dst_x, dst_y = self.canvas.coords(self.id)
|
||||
def move_node(self, node: "CanvasNode") -> None:
|
||||
if self.src == node:
|
||||
self.move_src()
|
||||
else:
|
||||
self.move_dst()
|
||||
|
||||
def move_shadow(self, node: "ShadowNode") -> None:
|
||||
if self.src_shadow == node:
|
||||
self.move_src_shadow()
|
||||
else:
|
||||
self.move_dst_shadow()
|
||||
|
||||
def move_src_shadow(self) -> None:
|
||||
_, _, _, _, dst_x, dst_y = self.dst.canvas.coords(self.id2)
|
||||
dst_pos = dst_x, dst_y
|
||||
self.moved(src_pos, dst_pos)
|
||||
self.moved2(self.src_shadow.position(), dst_pos)
|
||||
|
||||
def move_dst_shadow(self) -> None:
|
||||
src_x, src_y, _, _, _, _ = self.src.canvas.coords(self.id)
|
||||
src_pos = src_x, src_y
|
||||
self.moved(src_pos, self.dst_shadow.position())
|
||||
|
||||
def move_dst(self) -> None:
|
||||
src_x, src_y, _, _, _, _ = self.dst.canvas.coords(self.id)
|
||||
src_pos = src_x, src_y
|
||||
dst_pos = self.dst.position()
|
||||
if self.is_same_canvas():
|
||||
self.moved(src_pos, dst_pos)
|
||||
else:
|
||||
self.moved2(src_pos, dst_pos)
|
||||
|
||||
def move_src(self) -> None:
|
||||
_, _, _, _, dst_x, dst_y = self.src.canvas.coords(self.id)
|
||||
dst_pos = dst_x, dst_y
|
||||
self.moved(self.src.position(), dst_pos)
|
||||
|
||||
def moved(self, src_pos: Tuple[float, float], dst_pos: Tuple[float, float]) -> None:
|
||||
arc_pos = self._get_arcpoint(src_pos, dst_pos)
|
||||
self.canvas.coords(self.id, *src_pos, *arc_pos, *dst_pos)
|
||||
self.src.canvas.coords(self.id, *src_pos, *arc_pos, *dst_pos)
|
||||
if self.middle_label:
|
||||
self.canvas.coords(self.middle_label, *arc_pos)
|
||||
self.src.canvas.coords(self.middle_label, *arc_pos)
|
||||
src_pos, dst_pos = self.node_label_positions()
|
||||
if self.src_label:
|
||||
self.canvas.coords(self.src_label, *src_pos)
|
||||
self.src.canvas.coords(self.src_label, *src_pos)
|
||||
if self.dst_label:
|
||||
self.canvas.coords(self.dst_label, *dst_pos)
|
||||
self.src.canvas.coords(self.dst_label, *dst_pos)
|
||||
|
||||
def moved2(
|
||||
self, src_pos: Tuple[float, float], dst_pos: Tuple[float, float]
|
||||
) -> None:
|
||||
arc_pos = self._get_arcpoint(src_pos, dst_pos)
|
||||
self.dst.canvas.coords(self.id2, *src_pos, *arc_pos, *dst_pos)
|
||||
if self.middle_label2:
|
||||
self.dst.canvas.coords(self.middle_label2, *arc_pos)
|
||||
src_pos, dst_pos = self.node_label_positions2()
|
||||
if self.src_label2:
|
||||
self.dst.canvas.coords(self.src_label2, *src_pos)
|
||||
if self.dst_label2:
|
||||
self.dst.canvas.coords(self.dst_label2, *dst_pos)
|
||||
|
||||
def delete(self) -> None:
|
||||
logging.debug("deleting canvas edge, id: %s", self.id)
|
||||
self.canvas.delete(self.id)
|
||||
self.canvas.delete(self.src_label)
|
||||
self.canvas.delete(self.dst_label)
|
||||
self.src.canvas.delete(self.id)
|
||||
self.src.canvas.delete(self.src_label)
|
||||
self.src.canvas.delete(self.dst_label)
|
||||
if self.dst:
|
||||
self.dst.canvas.delete(self.id2)
|
||||
self.dst.canvas.delete(self.src_label2)
|
||||
self.dst.canvas.delete(self.dst_label2)
|
||||
if self.src_shadow:
|
||||
self.src_shadow.delete()
|
||||
self.src_shadow = None
|
||||
if self.dst_shadow:
|
||||
self.dst_shadow.delete()
|
||||
self.dst_shadow = None
|
||||
self.clear_middle_label()
|
||||
self.id = None
|
||||
self.id2 = None
|
||||
self.src_label = None
|
||||
self.src_label2 = None
|
||||
self.dst_label = None
|
||||
self.dst_label2 = None
|
||||
self.manager.edges.pop(self.token, None)
|
||||
|
||||
|
||||
class CanvasWirelessEdge(Edge):
|
||||
|
|
@ -239,35 +480,33 @@ class CanvasWirelessEdge(Edge):
|
|||
|
||||
def __init__(
|
||||
self,
|
||||
canvas: "CanvasGraph",
|
||||
src: int,
|
||||
dst: int,
|
||||
app: "Application",
|
||||
src: "CanvasNode",
|
||||
dst: "CanvasNode",
|
||||
network_id: int,
|
||||
token: str,
|
||||
src_pos: Tuple[float, float],
|
||||
dst_pos: Tuple[float, float],
|
||||
link: Link,
|
||||
) -> None:
|
||||
logging.debug("drawing wireless link from node %s to node %s", src, dst)
|
||||
super().__init__(canvas, src, dst)
|
||||
super().__init__(app, src, dst)
|
||||
self.network_id: int = network_id
|
||||
self.link: Link = link
|
||||
self.token: str = token
|
||||
self.width: float = WIRELESS_WIDTH
|
||||
color = link.color if link.color else WIRELESS_COLOR
|
||||
self.color: str = color
|
||||
self.draw(src_pos, dst_pos, self.canvas.manager.show_wireless.state())
|
||||
self.draw(self.manager.show_wireless.state())
|
||||
if link.label:
|
||||
self.middle_label_text(link.label)
|
||||
self.set_binding()
|
||||
|
||||
def set_binding(self) -> None:
|
||||
self.canvas.tag_bind(self.id, "<Button-1>", self.show_info)
|
||||
self.src.canvas.tag_bind(self.id, "<Button-1>", self.show_info)
|
||||
if self.id2 is not None:
|
||||
self.dst.canvas.tag_bind(self.id2, "<Button-1>", self.show_info)
|
||||
|
||||
def show_info(self, _event: tk.Event) -> None:
|
||||
self.canvas.app.display_info(
|
||||
WirelessEdgeInfoFrame, app=self.canvas.app, edge=self
|
||||
)
|
||||
self.app.display_info(WirelessEdgeInfoFrame, app=self.app, edge=self)
|
||||
|
||||
|
||||
class CanvasEdge(Edge):
|
||||
|
|
@ -276,47 +515,41 @@ class CanvasEdge(Edge):
|
|||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
canvas: "CanvasGraph",
|
||||
src: int,
|
||||
src_pos: Tuple[float, float],
|
||||
dst_pos: Tuple[float, float],
|
||||
self, app: "Application", src: "CanvasNode", dst: "CanvasNode" = None
|
||||
) -> None:
|
||||
"""
|
||||
Create an instance of canvas edge object
|
||||
"""
|
||||
super().__init__(canvas, src)
|
||||
super().__init__(app, src, dst)
|
||||
self.text_src: Optional[int] = None
|
||||
self.text_dst: Optional[int] = None
|
||||
self.link: Optional[Link] = None
|
||||
self.linked_wireless: bool = False
|
||||
self.asymmetric_link: Optional[Link] = None
|
||||
self.throughput: Optional[float] = None
|
||||
self.draw(src_pos, dst_pos, tk.NORMAL)
|
||||
self.draw(tk.NORMAL)
|
||||
self.set_binding()
|
||||
self.context: tk.Menu = tk.Menu(self.canvas)
|
||||
self.create_context()
|
||||
|
||||
def is_customized(self) -> bool:
|
||||
return self.width != EDGE_WIDTH or self.color != EDGE_COLOR
|
||||
|
||||
def create_context(self) -> None:
|
||||
themes.style_menu(self.context)
|
||||
self.context.add_command(label="Configure", command=self.click_configure)
|
||||
self.context.add_command(label="Delete", command=self.click_delete)
|
||||
|
||||
def set_binding(self) -> None:
|
||||
self.canvas.tag_bind(self.id, "<ButtonRelease-3>", self.show_context)
|
||||
self.canvas.tag_bind(self.id, "<Button-1>", self.show_info)
|
||||
show_context = functools.partial(self.show_info, self.src.canvas)
|
||||
self.src.canvas.tag_bind(self.id, "<ButtonRelease-3>", show_context)
|
||||
self.src.canvas.tag_bind(self.id, "<Button-1>", self.show_info)
|
||||
if self.dst and not self.is_same_canvas():
|
||||
show_context = functools.partial(self.show_info, self.dst.canvas)
|
||||
self.dst.canvas.tag_bind(self.id2, "<ButtonRelease-3>", show_context)
|
||||
self.dst.canvas.tag_bind(self.id2, "<Button-1>", self.show_info)
|
||||
|
||||
def iface_label(self, iface: Interface) -> str:
|
||||
label = ""
|
||||
if iface.name and self.canvas.manager.show_iface_names.get():
|
||||
if iface.name and self.manager.show_iface_names.get():
|
||||
label = f"{iface.name}"
|
||||
if iface.ip4 and self.canvas.manager.show_ip4s.get():
|
||||
if iface.ip4 and self.manager.show_ip4s.get():
|
||||
label = f"{label}\n" if label else ""
|
||||
label += f"{iface.ip4}/{iface.ip4_mask}"
|
||||
if iface.ip6 and self.canvas.manager.show_ip6s.get():
|
||||
if iface.ip6 and self.manager.show_ip6s.get():
|
||||
label = f"{label}\n" if label else ""
|
||||
label += f"{iface.ip6}/{iface.ip6_mask}"
|
||||
return label
|
||||
|
|
@ -346,77 +579,97 @@ class CanvasEdge(Edge):
|
|||
return
|
||||
if self.link.options.loss == EDGE_LOSS:
|
||||
state = tk.HIDDEN
|
||||
self.canvas.addtag_withtag(tags.LOSS_EDGES, self.id)
|
||||
self.src.canvas.addtag_withtag(tags.LOSS_EDGES, self.id)
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.addtag_withtag(tags.LOSS_EDGES, self.id2)
|
||||
else:
|
||||
state = tk.NORMAL
|
||||
self.canvas.dtag(self.id, tags.LOSS_EDGES)
|
||||
if self.canvas.manager.show_loss_links.state() == tk.HIDDEN:
|
||||
self.canvas.itemconfigure(self.id, state=state)
|
||||
self.src.canvas.dtag(self.id, tags.LOSS_EDGES)
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.dtag(self.id2, tags.LOSS_EDGES)
|
||||
if self.manager.show_loss_links.state() == tk.HIDDEN:
|
||||
self.src.canvas.itemconfigure(self.id, state=state)
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.itemconfigure(self.id2, state=state)
|
||||
|
||||
def set_throughput(self, throughput: float) -> None:
|
||||
throughput = 0.001 * throughput
|
||||
text = f"{throughput:.3f} kbps"
|
||||
self.middle_label_text(text)
|
||||
if throughput > self.canvas.manager.throughput_threshold:
|
||||
color = self.canvas.manager.throughput_color
|
||||
width = self.canvas.manager.throughput_width
|
||||
if throughput > self.manager.throughput_threshold:
|
||||
color = self.manager.throughput_color
|
||||
width = self.manager.throughput_width
|
||||
else:
|
||||
color = self.color
|
||||
width = self.scaled_width()
|
||||
self.canvas.itemconfig(self.id, fill=color, width=width)
|
||||
self.src.canvas.itemconfig(self.id, fill=color, width=width)
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.itemconfig(self.id2, fill=color, width=width)
|
||||
|
||||
def clear_throughput(self) -> None:
|
||||
self.clear_middle_label()
|
||||
if not self.linked_wireless:
|
||||
self.draw_link_options()
|
||||
|
||||
def complete(self, dst: int, linked_wireless: bool) -> None:
|
||||
def complete(self, dst: "CanvasNode", linked_wireless: bool) -> None:
|
||||
self.dst = dst
|
||||
self.linked_wireless = linked_wireless
|
||||
dst_pos = self.canvas.coords(self.dst)
|
||||
self.move_dst(dst_pos)
|
||||
self.move_dst()
|
||||
self.check_wireless()
|
||||
logging.debug("draw wired link from node %s to node %s", self.src, dst)
|
||||
|
||||
def check_wireless(self) -> None:
|
||||
if self.linked_wireless:
|
||||
self.canvas.itemconfig(self.id, state=tk.HIDDEN)
|
||||
self.canvas.dtag(self.id, tags.EDGE)
|
||||
if not self.linked_wireless:
|
||||
return
|
||||
self.src.canvas.itemconfig(self.id, state=tk.HIDDEN)
|
||||
self.src.canvas.dtag(self.id, tags.EDGE)
|
||||
self._check_antenna()
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.itemconfig(self.id2, state=tk.HIDDEN)
|
||||
self.dst.canvas.dtag(self.id2, tags.EDGE)
|
||||
self._check_antenna()
|
||||
|
||||
def _check_antenna(self) -> None:
|
||||
src_node = self.canvas.nodes[self.src]
|
||||
dst_node = self.canvas.nodes[self.dst]
|
||||
src_node_type = src_node.core_node.type
|
||||
dst_node_type = dst_node.core_node.type
|
||||
src_node_type = self.src.core_node.type
|
||||
dst_node_type = self.dst.core_node.type
|
||||
is_src_wireless = NodeUtils.is_wireless_node(src_node_type)
|
||||
is_dst_wireless = NodeUtils.is_wireless_node(dst_node_type)
|
||||
if is_src_wireless or is_dst_wireless:
|
||||
if is_src_wireless and not is_dst_wireless:
|
||||
dst_node.add_antenna()
|
||||
self.dst.add_antenna()
|
||||
elif not is_src_wireless and is_dst_wireless:
|
||||
src_node.add_antenna()
|
||||
self.src.add_antenna()
|
||||
else:
|
||||
src_node.add_antenna()
|
||||
self.src.add_antenna()
|
||||
|
||||
def reset(self) -> None:
|
||||
self.canvas.delete(self.middle_label)
|
||||
self.src.canvas.delete(self.middle_label)
|
||||
self.middle_label = None
|
||||
self.canvas.itemconfig(self.id, fill=self.color, width=self.scaled_width())
|
||||
self.src.canvas.itemconfig(self.id, fill=self.color, width=self.scaled_width())
|
||||
if not self.is_same_canvas():
|
||||
self.dst.canvas.delete(self.middle_label2)
|
||||
self.middle_label2 = None
|
||||
self.dst.canvas.itemconfig(
|
||||
self.id2, fill=self.color, width=self.scaled_width()
|
||||
)
|
||||
|
||||
def show_info(self, _event: tk.Event) -> None:
|
||||
self.canvas.app.display_info(EdgeInfoFrame, app=self.canvas.app, edge=self)
|
||||
self.app.display_info(EdgeInfoFrame, app=self.app, edge=self)
|
||||
|
||||
def show_context(self, event: tk.Event) -> None:
|
||||
state = tk.DISABLED if self.canvas.core.is_runtime() else tk.NORMAL
|
||||
self.context.entryconfigure(1, state=state)
|
||||
self.context.tk_popup(event.x_root, event.y_root)
|
||||
def show_context(self, canvas: "CanvasGraph", event: tk.Event) -> None:
|
||||
context: tk.Menu = tk.Menu(canvas)
|
||||
themes.style_menu(context)
|
||||
context.add_command(label="Configure", command=self.click_configure)
|
||||
context.add_command(label="Delete", command=self.click_delete)
|
||||
state = tk.DISABLED if self.app.core.is_runtime() else tk.NORMAL
|
||||
context.entryconfigure(1, state=state)
|
||||
context.tk_popup(event.x_root, event.y_root)
|
||||
|
||||
def click_delete(self) -> None:
|
||||
self.canvas.delete_edge(self)
|
||||
self.delete()
|
||||
|
||||
def click_configure(self) -> None:
|
||||
dialog = LinkConfigurationDialog(self.canvas.app, self)
|
||||
dialog = LinkConfigurationDialog(self.app, self)
|
||||
dialog.show()
|
||||
|
||||
def draw_link_options(self):
|
||||
|
|
@ -455,3 +708,21 @@ class CanvasEdge(Edge):
|
|||
lines.append(dup_line)
|
||||
label = "\n".join(lines)
|
||||
self.middle_label_text(label)
|
||||
|
||||
def delete(self) -> None:
|
||||
super().delete()
|
||||
self.src.edges.discard(self)
|
||||
if self.link.iface1:
|
||||
del self.src.ifaces[self.link.iface1.id]
|
||||
if self.dst:
|
||||
self.dst.edges.discard(self)
|
||||
if self.link.iface2:
|
||||
del self.dst.ifaces[self.link.iface2.id]
|
||||
src_wireless = NodeUtils.is_wireless_node(self.src.core_node.type)
|
||||
if src_wireless:
|
||||
self.dst.delete_antenna()
|
||||
dst_wireless = NodeUtils.is_wireless_node(self.dst.core_node.type)
|
||||
if dst_wireless:
|
||||
self.src.delete_antenna()
|
||||
self.app.core.deleted_canvas_edges([self])
|
||||
self.arc_common_edges()
|
||||
|
|
|
|||
|
|
@ -7,24 +7,16 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple
|
|||
from PIL import Image
|
||||
from PIL.ImageTk import PhotoImage
|
||||
|
||||
from core.api.grpc.wrappers import Interface, Link, LinkType, Node, Session
|
||||
from core.api.grpc.wrappers import Interface, Link
|
||||
from core.gui import appconfig
|
||||
from core.gui.dialogs.shapemod import ShapeDialog
|
||||
from core.gui.graph import tags
|
||||
from core.gui.graph.edges import (
|
||||
EDGE_WIDTH,
|
||||
CanvasEdge,
|
||||
CanvasWirelessEdge,
|
||||
Edge,
|
||||
arc_edges,
|
||||
create_edge_token,
|
||||
create_wireless_token,
|
||||
)
|
||||
from core.gui.graph.edges import EDGE_WIDTH, CanvasEdge, ShadowNode
|
||||
from core.gui.graph.enums import GraphMode, ScaleOption
|
||||
from core.gui.graph.node import CanvasNode
|
||||
from core.gui.graph.shape import Shape
|
||||
from core.gui.graph.shapeutils import ShapeType, is_draw_shape, is_marker
|
||||
from core.gui.images import ImageEnum, TypeToImage
|
||||
from core.gui.images import TypeToImage
|
||||
from core.gui.nodeutils import NodeUtils
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -59,9 +51,8 @@ class CanvasGraph(tk.Canvas):
|
|||
self.select_box: Optional[Shape] = None
|
||||
self.selected: Optional[int] = None
|
||||
self.nodes: Dict[int, CanvasNode] = {}
|
||||
self.edges: Dict[str, CanvasEdge] = {}
|
||||
self.shadow_nodes: Dict[int, ShadowNode] = {}
|
||||
self.shapes: Dict[int, Shape] = {}
|
||||
self.wireless_edges: Dict[str, CanvasWirelessEdge] = {}
|
||||
|
||||
# map wireless/EMANE node to the set of MDRs connected to that node
|
||||
self.wireless_network: Dict[int, Set[int]] = {}
|
||||
|
|
@ -107,21 +98,6 @@ class CanvasGraph(tk.Canvas):
|
|||
)
|
||||
self.configure(scrollregion=self.bbox(tk.ALL))
|
||||
|
||||
def reset_and_redraw(self, session: Session) -> None:
|
||||
# delete any existing drawn items
|
||||
for tag in tags.RESET_TAGS:
|
||||
self.delete(tag)
|
||||
|
||||
# set the private variables to default value
|
||||
self.selected = None
|
||||
self.nodes.clear()
|
||||
self.edges.clear()
|
||||
self.shapes.clear()
|
||||
self.wireless_edges.clear()
|
||||
self.wireless_network.clear()
|
||||
self.drawing_edge = None
|
||||
self.draw_session(session)
|
||||
|
||||
def setup_bindings(self) -> None:
|
||||
"""
|
||||
Bind any mouse events or hot keys to the matching action
|
||||
|
|
@ -173,123 +149,6 @@ class CanvasGraph(tk.Canvas):
|
|||
self.tag_lower(tags.GRIDLINE)
|
||||
self.tag_lower(self.rect)
|
||||
|
||||
def add_wired_edge(self, src: CanvasNode, dst: CanvasNode, link: Link) -> None:
|
||||
token = create_edge_token(link)
|
||||
if token in self.edges and link.options.unidirectional:
|
||||
edge = self.edges[token]
|
||||
edge.asymmetric_link = link
|
||||
elif token not in self.edges:
|
||||
node1 = src.core_node
|
||||
node2 = dst.core_node
|
||||
src_pos = (node1.position.x, node1.position.y)
|
||||
dst_pos = (node2.position.x, node2.position.y)
|
||||
edge = CanvasEdge(self, src.id, src_pos, dst_pos)
|
||||
self.complete_edge(src, dst, edge, link)
|
||||
|
||||
def delete_wired_edge(self, link: Link) -> None:
|
||||
token = create_edge_token(link)
|
||||
edge = self.edges.get(token)
|
||||
if edge:
|
||||
self.delete_edge(edge)
|
||||
|
||||
def update_wired_edge(self, link: Link) -> None:
|
||||
token = create_edge_token(link)
|
||||
edge = self.edges.get(token)
|
||||
if edge:
|
||||
edge.link.options = deepcopy(link.options)
|
||||
edge.draw_link_options()
|
||||
edge.check_options()
|
||||
|
||||
def add_wireless_edge(self, src: CanvasNode, dst: CanvasNode, link: Link) -> None:
|
||||
network_id = link.network_id if link.network_id else None
|
||||
token = create_wireless_token(src.id, dst.id, network_id)
|
||||
if token in self.wireless_edges:
|
||||
logging.warning("ignoring link that already exists: %s", link)
|
||||
return
|
||||
src_pos = self.coords(src.id)
|
||||
dst_pos = self.coords(dst.id)
|
||||
edge = CanvasWirelessEdge(
|
||||
self, src.id, dst.id, network_id, token, src_pos, dst_pos, link
|
||||
)
|
||||
self.wireless_edges[token] = edge
|
||||
src.wireless_edges.add(edge)
|
||||
dst.wireless_edges.add(edge)
|
||||
self.tag_raise(src.id)
|
||||
self.tag_raise(dst.id)
|
||||
self.arc_common_edges(edge)
|
||||
|
||||
def delete_wireless_edge(
|
||||
self, src: CanvasNode, dst: CanvasNode, link: Link
|
||||
) -> None:
|
||||
network_id = link.network_id if link.network_id else None
|
||||
token = create_wireless_token(src.id, dst.id, network_id)
|
||||
if token not in self.wireless_edges:
|
||||
return
|
||||
edge = self.wireless_edges.pop(token)
|
||||
edge.delete()
|
||||
src.wireless_edges.remove(edge)
|
||||
dst.wireless_edges.remove(edge)
|
||||
self.arc_common_edges(edge)
|
||||
|
||||
def update_wireless_edge(
|
||||
self, src: CanvasNode, dst: CanvasNode, link: Link
|
||||
) -> None:
|
||||
if not link.label:
|
||||
return
|
||||
network_id = link.network_id if link.network_id else None
|
||||
token = create_wireless_token(src.id, dst.id, network_id)
|
||||
if token not in self.wireless_edges:
|
||||
self.add_wireless_edge(src, dst, link)
|
||||
else:
|
||||
edge = self.wireless_edges[token]
|
||||
edge.middle_label_text(link.label)
|
||||
|
||||
def add_core_node(self, core_node: Node) -> None:
|
||||
logging.debug("adding node: %s", core_node)
|
||||
# if the gui can't find node's image, default to the "edit-node" image
|
||||
image = NodeUtils.node_image(core_node, self.app.guiconfig, self.app.app_scale)
|
||||
if not image:
|
||||
image = self.app.get_icon(ImageEnum.EDITNODE, ICON_SIZE)
|
||||
x = core_node.position.x
|
||||
y = core_node.position.y
|
||||
node = CanvasNode(self.app, self, x, y, core_node, image)
|
||||
self.nodes[node.id] = node
|
||||
self.core.set_canvas_node(core_node, node)
|
||||
|
||||
def draw_session(self, session: Session) -> None:
|
||||
"""
|
||||
Draw existing session.
|
||||
"""
|
||||
# draw existing nodes
|
||||
for core_node in session.nodes.values():
|
||||
logging.debug("drawing node: %s", core_node)
|
||||
# peer to peer node is not drawn on the GUI
|
||||
if NodeUtils.is_ignore_node(core_node.type):
|
||||
continue
|
||||
self.add_core_node(core_node)
|
||||
# draw existing links
|
||||
for link in session.links:
|
||||
logging.debug("drawing link: %s", link)
|
||||
canvas_node1 = self.core.get_canvas_node(link.node1_id)
|
||||
canvas_node2 = self.core.get_canvas_node(link.node2_id)
|
||||
if link.type == LinkType.WIRELESS:
|
||||
self.add_wireless_edge(canvas_node1, canvas_node2, link)
|
||||
else:
|
||||
self.add_wired_edge(canvas_node1, canvas_node2, link)
|
||||
|
||||
def stopped_session(self) -> None:
|
||||
# clear wireless edges
|
||||
for edge in self.wireless_edges.values():
|
||||
edge.delete()
|
||||
src_node = self.nodes[edge.src]
|
||||
src_node.wireless_edges.remove(edge)
|
||||
dst_node = self.nodes[edge.dst]
|
||||
dst_node.wireless_edges.remove(edge)
|
||||
self.wireless_edges.clear()
|
||||
|
||||
# clear throughputs
|
||||
self.clear_throughputs()
|
||||
|
||||
def canvas_xy(self, event: tk.Event) -> Tuple[float, float]:
|
||||
"""
|
||||
Convert window coordinate to canvas coordinate
|
||||
|
|
@ -308,14 +167,12 @@ class CanvasGraph(tk.Canvas):
|
|||
for _id in overlapping:
|
||||
if self.drawing_edge and self.drawing_edge.id == _id:
|
||||
continue
|
||||
|
||||
if _id in self.nodes:
|
||||
elif _id in self.nodes:
|
||||
selected = _id
|
||||
break
|
||||
|
||||
if _id in self.shapes:
|
||||
elif _id in self.shapes:
|
||||
selected = _id
|
||||
elif _id in self.shadow_nodes:
|
||||
selected = _id
|
||||
|
||||
return selected
|
||||
|
||||
def click_release(self, event: tk.Event) -> None:
|
||||
|
|
@ -367,24 +224,16 @@ class CanvasGraph(tk.Canvas):
|
|||
return
|
||||
# edge dst must be a node
|
||||
logging.debug("current selected: %s", self.selected)
|
||||
src_node = self.nodes.get(edge.src)
|
||||
dst_node = self.nodes.get(self.selected)
|
||||
if not dst_node or not src_node:
|
||||
if not dst_node:
|
||||
edge.delete()
|
||||
return
|
||||
# check if node can be linked
|
||||
if not src_node.is_linkable(dst_node):
|
||||
if not edge.src.is_linkable(dst_node):
|
||||
edge.delete()
|
||||
return
|
||||
# finalize edge creation
|
||||
self.complete_edge(src_node, dst_node, edge)
|
||||
|
||||
def arc_common_edges(self, edge: Edge) -> None:
|
||||
src_node = self.nodes[edge.src]
|
||||
dst_node = self.nodes[edge.dst]
|
||||
common_edges = list(src_node.edges & dst_node.edges)
|
||||
common_edges += list(src_node.wireless_edges & dst_node.wireless_edges)
|
||||
arc_edges(common_edges)
|
||||
self.manager.complete_edge(edge.src, dst_node, edge)
|
||||
|
||||
def select_object(self, object_id: int, choose_multiple: bool = False) -> None:
|
||||
"""
|
||||
|
|
@ -439,15 +288,13 @@ class CanvasGraph(tk.Canvas):
|
|||
if edge in edges:
|
||||
continue
|
||||
edges.add(edge)
|
||||
del self.edges[edge.token]
|
||||
edge.delete()
|
||||
# update node connected to edge being deleted
|
||||
other_id = edge.src
|
||||
other_node = edge.src
|
||||
other_iface = edge.link.iface1
|
||||
if edge.src == object_id:
|
||||
other_id = edge.dst
|
||||
other_node = edge.dst
|
||||
other_iface = edge.link.iface2
|
||||
other_node = self.nodes[other_id]
|
||||
other_node.edges.remove(edge)
|
||||
if other_iface:
|
||||
del other_node.ifaces[other_iface.id]
|
||||
|
|
@ -463,26 +310,6 @@ class CanvasGraph(tk.Canvas):
|
|||
self.core.deleted_canvas_nodes(nodes)
|
||||
self.core.deleted_canvas_edges(edges)
|
||||
|
||||
def delete_edge(self, edge: CanvasEdge) -> None:
|
||||
edge.delete()
|
||||
del self.edges[edge.token]
|
||||
src_node = self.nodes[edge.src]
|
||||
src_node.edges.discard(edge)
|
||||
if edge.link.iface1:
|
||||
del src_node.ifaces[edge.link.iface1.id]
|
||||
dst_node = self.nodes[edge.dst]
|
||||
dst_node.edges.discard(edge)
|
||||
if edge.link.iface2:
|
||||
del dst_node.ifaces[edge.link.iface2.id]
|
||||
src_wireless = NodeUtils.is_wireless_node(src_node.core_node.type)
|
||||
if src_wireless:
|
||||
dst_node.delete_antenna()
|
||||
dst_wireless = NodeUtils.is_wireless_node(dst_node.core_node.type)
|
||||
if dst_wireless:
|
||||
src_node.delete_antenna()
|
||||
self.core.deleted_canvas_edges([edge])
|
||||
self.arc_common_edges(edge)
|
||||
|
||||
def zoom(self, event: tk.Event, factor: float = None) -> None:
|
||||
if not factor:
|
||||
factor = ZOOM_IN if event.delta > 0 else ZOOM_OUT
|
||||
|
|
@ -516,8 +343,8 @@ class CanvasGraph(tk.Canvas):
|
|||
logging.debug("click press offset(%s, %s)", x_check, y_check)
|
||||
is_node = selected in self.nodes
|
||||
if self.manager.mode == GraphMode.EDGE and is_node:
|
||||
pos = self.coords(selected)
|
||||
self.drawing_edge = CanvasEdge(self, selected, pos, pos)
|
||||
node = self.nodes[selected]
|
||||
self.drawing_edge = CanvasEdge(self.app, node)
|
||||
self.organize()
|
||||
|
||||
if self.manager.mode == GraphMode.ANNOTATION:
|
||||
|
|
@ -556,6 +383,16 @@ class CanvasGraph(tk.Canvas):
|
|||
node.core_node.position.x,
|
||||
node.core_node.position.y,
|
||||
)
|
||||
elif selected in self.shadow_nodes:
|
||||
shadow_node = self.shadow_nodes[selected]
|
||||
self.select_object(shadow_node.id)
|
||||
self.selected = selected
|
||||
logging.debug(
|
||||
"selected shadow node(%s), coords: (%s, %s)",
|
||||
shadow_node.node.core_node.name,
|
||||
shadow_node.node.core_node.position.x,
|
||||
shadow_node.node.core_node.position.y,
|
||||
)
|
||||
else:
|
||||
if self.manager.mode == GraphMode.SELECT:
|
||||
shape = Shape(self.app, self, ShapeType.RECTANGLE, x, y)
|
||||
|
|
@ -597,7 +434,7 @@ class CanvasGraph(tk.Canvas):
|
|||
self.cursor = x, y
|
||||
|
||||
if self.manager.mode == GraphMode.EDGE and self.drawing_edge is not None:
|
||||
self.drawing_edge.move_dst(self.cursor)
|
||||
self.drawing_edge.drawing(self.cursor)
|
||||
if self.manager.mode == GraphMode.ANNOTATION:
|
||||
if is_draw_shape(self.manager.annotation_type) and self.shape_drawing:
|
||||
shape = self.shapes[self.selected]
|
||||
|
|
@ -625,10 +462,15 @@ class CanvasGraph(tk.Canvas):
|
|||
if self.manager.mode in MOVE_SHAPE_MODES and selected_id in self.shapes:
|
||||
shape = self.shapes[selected_id]
|
||||
shape.motion(x_offset, y_offset)
|
||||
|
||||
if self.manager.mode in MOVE_NODE_MODES and selected_id in self.nodes:
|
||||
elif self.manager.mode in MOVE_NODE_MODES and selected_id in self.nodes:
|
||||
node = self.nodes[selected_id]
|
||||
node.motion(x_offset, y_offset, update=self.core.is_runtime())
|
||||
elif (
|
||||
self.manager.mode in MOVE_NODE_MODES
|
||||
and selected_id in self.shadow_nodes
|
||||
):
|
||||
shadow_node = self.shadow_nodes[selected_id]
|
||||
shadow_node.motion(x_offset, y_offset)
|
||||
else:
|
||||
if self.select_box and self.manager.mode == GraphMode.SELECT:
|
||||
self.select_box.shape_motion(x, y)
|
||||
|
|
@ -823,38 +665,10 @@ class CanvasGraph(tk.Canvas):
|
|||
"""
|
||||
create an edge between source node and destination node
|
||||
"""
|
||||
pos = (src.core_node.position.x, src.core_node.position.y)
|
||||
edge = CanvasEdge(self, src.id, pos, pos)
|
||||
self.complete_edge(src, dst, edge)
|
||||
edge = CanvasEdge(self.app, src)
|
||||
self.manager.complete_edge(src, dst, edge)
|
||||
return edge
|
||||
|
||||
def complete_edge(
|
||||
self,
|
||||
src: CanvasNode,
|
||||
dst: CanvasNode,
|
||||
edge: CanvasEdge,
|
||||
link: Optional[Link] = None,
|
||||
) -> None:
|
||||
linked_wireless = self.is_linked_wireless(src.id, dst.id)
|
||||
edge.complete(dst.id, linked_wireless)
|
||||
if link is None:
|
||||
link = self.core.create_link(edge, src, dst)
|
||||
edge.link = link
|
||||
if link.iface1:
|
||||
iface1 = link.iface1
|
||||
src.ifaces[iface1.id] = iface1
|
||||
if link.iface2:
|
||||
iface2 = link.iface2
|
||||
dst.ifaces[iface2.id] = iface2
|
||||
src.edges.add(edge)
|
||||
dst.edges.add(edge)
|
||||
edge.token = create_edge_token(edge.link)
|
||||
self.arc_common_edges(edge)
|
||||
edge.draw_labels()
|
||||
edge.check_options()
|
||||
self.edges[edge.token] = edge
|
||||
self.core.save_edge(edge, src, dst)
|
||||
|
||||
def copy(self) -> None:
|
||||
if self.core.is_runtime():
|
||||
logging.debug("copy is disabled during runtime state")
|
||||
|
|
@ -974,30 +788,6 @@ class CanvasGraph(tk.Canvas):
|
|||
)
|
||||
self.tag_raise(tags.NODE)
|
||||
|
||||
def is_linked_wireless(self, src: int, dst: int) -> bool:
|
||||
src_node = self.nodes[src]
|
||||
dst_node = self.nodes[dst]
|
||||
src_node_type = src_node.core_node.type
|
||||
dst_node_type = dst_node.core_node.type
|
||||
is_src_wireless = NodeUtils.is_wireless_node(src_node_type)
|
||||
is_dst_wireless = NodeUtils.is_wireless_node(dst_node_type)
|
||||
|
||||
# update the wlan/EMANE network
|
||||
wlan_network = self.wireless_network
|
||||
if is_src_wireless and not is_dst_wireless:
|
||||
if src not in wlan_network:
|
||||
wlan_network[src] = set()
|
||||
wlan_network[src].add(dst)
|
||||
elif not is_src_wireless and is_dst_wireless:
|
||||
if dst not in wlan_network:
|
||||
wlan_network[dst] = set()
|
||||
wlan_network[dst].add(src)
|
||||
return is_src_wireless or is_dst_wireless
|
||||
|
||||
def clear_throughputs(self) -> None:
|
||||
for edge in self.edges.values():
|
||||
edge.clear_throughput()
|
||||
|
||||
def scale_graph(self) -> None:
|
||||
for nid, canvas_node in self.nodes.items():
|
||||
img = None
|
||||
|
|
|
|||
|
|
@ -1,14 +1,23 @@
|
|||
import logging
|
||||
import tkinter as tk
|
||||
from copy import deepcopy
|
||||
from tkinter import BooleanVar, messagebox, ttk
|
||||
from typing import TYPE_CHECKING, Any, Dict, Optional, Set, Tuple, ValuesView
|
||||
|
||||
from core.api.grpc.wrappers import LinkType, Session, ThroughputsEvent
|
||||
from core.api.grpc.wrappers import Link, LinkType, Node, Session, ThroughputsEvent
|
||||
from core.gui.graph import tags
|
||||
from core.gui.graph.edges import (
|
||||
CanvasEdge,
|
||||
CanvasWirelessEdge,
|
||||
create_edge_token,
|
||||
create_wireless_token,
|
||||
)
|
||||
from core.gui.graph.enums import GraphMode
|
||||
from core.gui.graph.graph import CanvasGraph
|
||||
from core.gui.graph.node import CanvasNode
|
||||
from core.gui.graph.shapeutils import ShapeType
|
||||
from core.gui.nodeutils import NodeDraw, NodeUtils
|
||||
from core.gui.images import ImageEnum
|
||||
from core.gui.nodeutils import ICON_SIZE, NodeDraw, NodeUtils
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from core.gui.app import Application
|
||||
|
|
@ -46,6 +55,8 @@ class CanvasManager:
|
|||
# canvas object storage
|
||||
# TODO: validate this
|
||||
self.wireless_network: Dict[int, Set[int]] = {}
|
||||
self.edges: Dict[str, CanvasEdge] = {}
|
||||
self.wireless_edges: Dict[str, CanvasWirelessEdge] = {}
|
||||
|
||||
# global canvas settings
|
||||
self.default_dimensions: Tuple[int, int] = (
|
||||
|
|
@ -174,27 +185,21 @@ class CanvasManager:
|
|||
def draw_session(self, session: Session) -> None:
|
||||
# create session nodes
|
||||
for core_node in session.nodes.values():
|
||||
# get tab id for node
|
||||
canvas_id = core_node.canvas if core_node.canvas > 0 else 1
|
||||
canvas = self.get(canvas_id)
|
||||
# add node, avoiding ignored nodes
|
||||
if NodeUtils.is_ignore_node(core_node.type):
|
||||
continue
|
||||
logging.debug("drawing node: %s", core_node)
|
||||
canvas.add_core_node(core_node)
|
||||
self.add_core_node(core_node)
|
||||
|
||||
# draw existing links
|
||||
for link in session.links:
|
||||
logging.debug("drawing link: %s", link)
|
||||
node1 = self.core.get_canvas_node(link.node1_id)
|
||||
node2 = self.core.get_canvas_node(link.node2_id)
|
||||
# TODO: handle edges for nodes on different canvases
|
||||
if node1.canvas == node2.canvas:
|
||||
canvas = node1.canvas
|
||||
if link.type == LinkType.WIRELESS:
|
||||
canvas.add_wireless_edge(node1, node2, link)
|
||||
self.add_wireless_edge(node1, node2, link)
|
||||
else:
|
||||
canvas.add_wired_edge(node1, node2, link)
|
||||
self.add_wired_edge(node1, node2, link)
|
||||
else:
|
||||
logging.error("cant handle nodes linked between canvases")
|
||||
|
||||
|
|
@ -238,6 +243,21 @@ class CanvasManager:
|
|||
canvas = self.get(canvas_id)
|
||||
canvas.parse_metadata(canvas_config)
|
||||
|
||||
def add_core_node(self, core_node: Node) -> None:
|
||||
logging.debug("adding node: %s", core_node)
|
||||
# get canvas tab for node
|
||||
canvas_id = core_node.canvas if core_node.canvas > 0 else 1
|
||||
canvas = self.get(canvas_id)
|
||||
# if the gui can't find node's image, default to the "edit-node" image
|
||||
image = NodeUtils.node_image(core_node, self.app.guiconfig, self.app.app_scale)
|
||||
if not image:
|
||||
image = self.app.get_icon(ImageEnum.EDITNODE, ICON_SIZE)
|
||||
x = core_node.position.x
|
||||
y = core_node.position.y
|
||||
node = CanvasNode(self.app, canvas, x, y, core_node, image)
|
||||
canvas.nodes[node.id] = node
|
||||
self.core.set_canvas_node(core_node, node)
|
||||
|
||||
def set_throughputs(self, throughputs_event: ThroughputsEvent):
|
||||
for iface_throughput in throughputs_event.iface_throughputs:
|
||||
node_id = iface_throughput.node_id
|
||||
|
|
@ -249,6 +269,123 @@ class CanvasManager:
|
|||
edge.set_throughput(throughput)
|
||||
|
||||
def clear_throughputs(self) -> None:
|
||||
for canvas in self.all():
|
||||
for edge in canvas.edges.values():
|
||||
edge.clear_throughput()
|
||||
for edge in self.edges.values():
|
||||
edge.clear_throughput()
|
||||
|
||||
def stopped_session(self) -> None:
|
||||
# clear wireless edges
|
||||
for edge in self.wireless_edges.values():
|
||||
edge.delete()
|
||||
edge.src.wireless_edges.remove(edge)
|
||||
edge.dst.wireless_edges.remove(edge)
|
||||
self.wireless_edges.clear()
|
||||
self.clear_throughputs()
|
||||
|
||||
def update_wired_edge(self, link: Link) -> None:
|
||||
token = create_edge_token(link)
|
||||
edge = self.edges.get(token)
|
||||
if edge:
|
||||
edge.link.options = deepcopy(link.options)
|
||||
edge.draw_link_options()
|
||||
edge.check_options()
|
||||
|
||||
def delete_wired_edge(self, link: Link) -> None:
|
||||
token = create_edge_token(link)
|
||||
edge = self.edges.get(token)
|
||||
if edge:
|
||||
edge.delete()
|
||||
|
||||
def add_wired_edge(self, src: CanvasNode, dst: CanvasNode, link: Link) -> None:
|
||||
token = create_edge_token(link)
|
||||
if token in self.edges and link.options.unidirectional:
|
||||
edge = self.edges[token]
|
||||
edge.asymmetric_link = link
|
||||
elif token not in self.edges:
|
||||
edge = CanvasEdge(self.app, src)
|
||||
self.complete_edge(src, dst, edge, link)
|
||||
|
||||
def add_wireless_edge(self, src: CanvasNode, dst: CanvasNode, link: Link) -> None:
|
||||
network_id = link.network_id if link.network_id else None
|
||||
token = create_wireless_token(src.id, dst.id, network_id)
|
||||
if token in self.wireless_edges:
|
||||
logging.warning("ignoring link that already exists: %s", link)
|
||||
return
|
||||
edge = CanvasWirelessEdge(self.app, src, dst, network_id, token, link)
|
||||
self.wireless_edges[token] = edge
|
||||
src.wireless_edges.add(edge)
|
||||
dst.wireless_edges.add(edge)
|
||||
src.canvas.tag_raise(src.id)
|
||||
dst.canvas.tag_raise(dst.id)
|
||||
edge.arc_common_edges()
|
||||
|
||||
def delete_wireless_edge(
|
||||
self, src: CanvasNode, dst: CanvasNode, link: Link
|
||||
) -> None:
|
||||
network_id = link.network_id if link.network_id else None
|
||||
token = create_wireless_token(src.id, dst.id, network_id)
|
||||
if token not in self.wireless_edges:
|
||||
return
|
||||
edge = self.wireless_edges.pop(token)
|
||||
edge.delete()
|
||||
src.wireless_edges.remove(edge)
|
||||
dst.wireless_edges.remove(edge)
|
||||
edge.arc_common_edges()
|
||||
|
||||
def update_wireless_edge(
|
||||
self, src: CanvasNode, dst: CanvasNode, link: Link
|
||||
) -> None:
|
||||
if not link.label:
|
||||
return
|
||||
network_id = link.network_id if link.network_id else None
|
||||
token = create_wireless_token(src.id, dst.id, network_id)
|
||||
if token not in self.wireless_edges:
|
||||
self.add_wireless_edge(src, dst, link)
|
||||
else:
|
||||
edge = self.wireless_edges[token]
|
||||
edge.middle_label_text(link.label)
|
||||
|
||||
# TODO: remove src parameter as edge already has value
|
||||
def complete_edge(
|
||||
self,
|
||||
src: CanvasNode,
|
||||
dst: CanvasNode,
|
||||
edge: CanvasEdge,
|
||||
link: Optional[Link] = None,
|
||||
) -> None:
|
||||
linked_wireless = self.is_linked_wireless(src, dst)
|
||||
edge.complete(dst, linked_wireless)
|
||||
if link is None:
|
||||
link = self.core.create_link(edge, src, dst)
|
||||
edge.link = link
|
||||
if link.iface1:
|
||||
iface1 = link.iface1
|
||||
src.ifaces[iface1.id] = iface1
|
||||
if link.iface2:
|
||||
iface2 = link.iface2
|
||||
dst.ifaces[iface2.id] = iface2
|
||||
src.edges.add(edge)
|
||||
dst.edges.add(edge)
|
||||
edge.token = create_edge_token(edge.link)
|
||||
edge.arc_common_edges()
|
||||
edge.draw_labels()
|
||||
edge.check_options()
|
||||
self.edges[edge.token] = edge
|
||||
self.core.save_edge(edge, src, dst)
|
||||
|
||||
def is_linked_wireless(self, src: CanvasNode, dst: CanvasNode) -> bool:
|
||||
src_node_type = src.core_node.type
|
||||
dst_node_type = dst.core_node.type
|
||||
is_src_wireless = NodeUtils.is_wireless_node(src_node_type)
|
||||
is_dst_wireless = NodeUtils.is_wireless_node(dst_node_type)
|
||||
|
||||
# update the wlan/EMANE network
|
||||
wlan_network = self.wireless_network
|
||||
if is_src_wireless and not is_dst_wireless:
|
||||
if src not in wlan_network:
|
||||
wlan_network[src.core_node.id] = set()
|
||||
wlan_network[src.core_node.id].add(dst.core_node.id)
|
||||
elif not is_src_wireless and is_dst_wireless:
|
||||
if dst not in wlan_network:
|
||||
wlan_network[dst.core_node.id] = set()
|
||||
wlan_network[dst.core_node.id].add(src.core_node.id)
|
||||
return is_src_wireless or is_dst_wireless
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import functools
|
|||
import logging
|
||||
import tkinter as tk
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Dict, List, Set
|
||||
from typing import TYPE_CHECKING, Dict, List, Set, Tuple
|
||||
|
||||
import grpc
|
||||
from PIL.ImageTk import PhotoImage
|
||||
|
|
@ -67,6 +67,9 @@ class CanvasNode:
|
|||
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:
|
||||
|
|
@ -87,7 +90,7 @@ class CanvasNode:
|
|||
self.delete_antennas()
|
||||
|
||||
def add_antenna(self) -> None:
|
||||
x, y = self.canvas.coords(self.id)
|
||||
x, y = self.position()
|
||||
offset = len(self.antennas) * 8 * self.app.app_scale
|
||||
img = self.app.get_icon(ImageEnum.ANTENNA, ANTENNA_SIZE)
|
||||
antenna_id = self.canvas.create_image(
|
||||
|
|
@ -145,15 +148,14 @@ class CanvasNode:
|
|||
|
||||
def move(self, x: float, y: float) -> None:
|
||||
x, y = self.canvas.get_scaled_coords(x, y)
|
||||
current_x, current_y = self.canvas.coords(self.id)
|
||||
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.canvas.coords(self.id)
|
||||
original_position = self.position()
|
||||
self.canvas.move(self.id, x_offset, y_offset)
|
||||
pos = self.canvas.coords(self.id)
|
||||
|
||||
# check new position
|
||||
bbox = self.canvas.bbox(self.id)
|
||||
|
|
@ -171,11 +173,12 @@ class CanvasNode:
|
|||
|
||||
# move edges
|
||||
for edge in self.edges:
|
||||
edge.move_node(self.id, pos)
|
||||
edge.move_node(self)
|
||||
for edge in self.wireless_edges:
|
||||
edge.move_node(self.id, pos)
|
||||
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
|
||||
|
|
@ -297,14 +300,12 @@ class CanvasNode:
|
|||
self.canvas_delete()
|
||||
|
||||
def click_unlink(self, edge: CanvasEdge) -> None:
|
||||
self.canvas.delete_edge(edge)
|
||||
edge.delete()
|
||||
self.app.default_info()
|
||||
|
||||
def click_link(self, node: "CanvasNode") -> None:
|
||||
pos = self.canvas.coords(self.id)
|
||||
edge = CanvasEdge(self.canvas, self.id, pos, pos)
|
||||
self.canvas.complete_edge(self, node, edge)
|
||||
self.canvas.organize()
|
||||
edge = CanvasEdge(self.app, self, node)
|
||||
self.app.manager.complete_edge(self, node, edge)
|
||||
|
||||
def canvas_delete(self) -> None:
|
||||
self.canvas.clear_selection()
|
||||
|
|
|
|||
|
|
@ -188,19 +188,16 @@ class InterfaceManager:
|
|||
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
|
||||
subnets = None
|
||||
if not visited:
|
||||
visited = set()
|
||||
visited.add(canvas_node.core_node.id)
|
||||
for edge in canvas_node.edges:
|
||||
src_node = canvas.nodes[edge.src]
|
||||
dst_node = canvas.nodes[edge.dst]
|
||||
iface = edge.link.iface1
|
||||
check_node = src_node
|
||||
if src_node == canvas_node:
|
||||
check_node = edge.src
|
||||
if edge.src == canvas_node:
|
||||
iface = edge.link.iface2
|
||||
check_node = dst_node
|
||||
check_node = edge.dst
|
||||
if check_node.core_node.id in visited:
|
||||
continue
|
||||
visited.add(check_node.core_node.id)
|
||||
|
|
|
|||
|
|
@ -407,8 +407,7 @@ class Toolbar(ttk.Frame):
|
|||
|
||||
def stop_callback(self, result: bool) -> None:
|
||||
self.set_design()
|
||||
for canvas in self.app.manager.all():
|
||||
canvas.stopped_session()
|
||||
self.app.manager.stopped_session()
|
||||
|
||||
def update_annotation(
|
||||
self, shape_type: ShapeType, image_enum: ImageEnum, image: PhotoImage
|
||||
|
|
|
|||
Loading…
Reference in a new issue