pfzs9fyia
/
NetworkAnalog
1
0
Fork 0
Code Issues Pull Requests Projects Releases 1 Wiki Activity

修改程序UI，将整体UI大小放大

Browse Source
master
bettleChen 2 years ago
parent d26eb16b1d
commit 3eb07c0b57
8 changed files with 181 additions and 875 deletions
Show Stats Download Patch File Download Diff File

12
.idea/dataSources.xml
Unescape View File

@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="DataSourceManagerImpl" format="xml" multifile-model="true">
<data-source source="LOCAL" name="network.db" uuid="c2716c2d-ad8d-45c8-9b7d-8ee175d220a5">
<driver-ref>sqlite.xerial</driver-ref>
<synchronize>true</synchronize>
<jdbc-driver>org.sqlite.JDBC</jdbc-driver>
<jdbc-url>jdbc:sqlite:$PROJECT_DIR$/../网络仿真/NetworkAnalog/network.db</jdbc-url>
<working-dir>$ProjectFileDir$</working-dir>
</data-source>
</component>
</project>

227
NetworkAnalog/NetworkAnalog.py
Unescape View File

@ -7,12 +7,35 @@ from ttkbootstrap import ttk
from tkinter import messagebox
import re
from PIL import ImageTk, Image
import platform
from SimObjs import SimPacket, SimHost, AllSimConnect, SimRouter, SimSwitch, SimHub, SimBase
from dbUtil import search, execute_sql, delete_obj, truncate_db
def round_rectangle(cv, x1, y1, x2, y2, radius=25, **kwargs):
points = [x1 + radius, y1,
x1 + radius, y1,
x2 - radius, y1,
x2 - radius, y1,
x2, y1,
x2, y1 + radius,
x2, y1 + radius,
x2, y2 - radius,
x2, y2 - radius,
x2, y2,
x2 - radius, y2,
x2 - radius, y2,
x1 + radius, y2,
x1 + radius, y2,
x1, y2,
x1, y2 - radius,
x1, y2 - radius,
x1, y1 + radius,
x1, y1 + radius,
x1, y1]
return cv.create_polygon(points, **kwargs, smooth=True)
def validate_ip_address(ip_address):
"""
匹配ip地址格式是否规范
@ -62,16 +85,16 @@ class RouterConfigWindow(tk.Toplevel):
def create_interface_inputs(self):
label_text = ["接口1", "接口2", "接口3", "接口4"]
for i in range(4):
label = tk.Label(self, text=label_text[i])
label = tk.Label(self, text=label_text[i], font=("黑体", 16))
label.grid(row=i, column=0, padx=10, pady=5, sticky="w")
entry = tk.Entry(self, width=20)
entry = tk.Entry(self, width=20, font=("黑体", 16),)
entry.grid(row=i, column=1, padx=10, pady=5, sticky="w")
self.interface_entries.append(entry)
button = tk.Button(self, text="添加", command=lambda index=i: self.add_router_entry(index))
button = tk.Button(self, text="添加", font=("黑体", 16), command=lambda index=i: self.add_router_entry(index))
button.grid(row=i, column=2, padx=10, pady=5)
lab = LabelFrame(self, text="示例")
lab.grid(row=4, column=0, columnspan=3, sticky=W, padx=20)
Label(lab, text="10.1.2.0/24 或者 10.1.2.12" if self.router_obj.ObjType == 2 else "MAC11").pack()
Label(lab, text="10.1.2.0/24 或者 10.1.2.12" if self.router_obj.ObjType == 2 else "MAC11", font=("黑体", 16)).pack()
def create_router_table(self):
def on_right_click(event):
@ -88,7 +111,10 @@ class RouterConfigWindow(tk.Toplevel):
self.router_treeview.delete(item)
self.router_table_frame = tk.Frame(self)
self.router_table_frame.grid(row=5, column=0, columnspan=3, padx=10, pady=5)
self.router_treeview = ttk.Treeview(self.router_table_frame, columns=("Interface", "Route"), show="headings")
style = ttk.Style()
style.configure("Custom.Treeview.Heading", font=("宋体", 15))
style.configure("Custom.Treeview", rowheight=30, font=("宋体", 15))
self.router_treeview = ttk.Treeview(self.router_table_frame, style="Custom.Treeview", columns=("Interface", "Route"), show="headings")
self.router_treeview.heading("Interface", text="接口")
self.router_treeview.heading("Route", text="网段")
self.router_treeview.pack(side="left", fill="both")
@ -203,6 +229,8 @@ class NetWorkAnalog(Canvas):
Image.open(sys.path[0] + "/../datas/images/集线器.png").resize((60, 60)))
self.host_img = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/主机.png").resize((60, 60)))
self.width = int(self.cget("width"))
self.height = int(self.cget("height"))
self.chose = self.host_img
self.AllSimObjs = {}
self.conns = []
@ -282,6 +310,7 @@ class NetWorkAnalog(Canvas):
return
conn = AllSimConnect(self, self.line_start_obj, self.line_start_ifs, self.line_end_obj,
self.line_end_ifs)
conn.draw_line()
for conn_obj in self.line_start_obj.connections: # 判断两个连接对象是否已经连接过了
if conn_obj == conn:
flag = True
@ -309,15 +338,15 @@ class NetWorkAnalog(Canvas):
if self.drawLine:
self.delete("Line")
x, y = event.x, event.y
if not (150 < x < 650 and 40 < y < 490):
if x < 150:
x = 150
elif x > 650:
x = 650
if y < 40:
y = 40
elif y > 490:
y = 490
if not (0 < x < self.width * 0.8 and 0 < y < self.height * 0.85):
if x < 0:
x = 0
elif x > self.width * 0.8:
x = self.width * 0.8
if y < 0:
y = 0
elif y > self.height * 0.85:
y = self.height * 0.85
self.create_line(self.line_start_obj.ObjX, self.line_start_obj.ObjY, x + 8, y + 8, fill="#5b9bd5",
width=1,
tags="Line")
@ -345,7 +374,7 @@ class NetWorkAnalog(Canvas):
鼠标左键松开事件
:param event: 事件对象
"""
if 170 < event.x < 630 and 60 < event.y < 470: # 在方框内，无变化
if 20 < event.x < self.width * 0.8 + 20 and 20 < event.y < self.height * 0.85 - 20: # 在方框内，无变化
if name == "路由器":
tag = SimRouter(self, event.x, event.y)
elif name == "集线器":
@ -354,6 +383,7 @@ class NetWorkAnalog(Canvas):
tag = SimSwitch(self, event.x, event.y)
else:
tag = SimHost(self, event.x, event.y)
tag.create_img()
self.AllSimObjs[tag.ObjID] = tag
tag.save()
self.tag_bind_event()
@ -374,7 +404,7 @@ class NetWorkAnalog(Canvas):
else:
self.chose = self.host_img
self.delete("L")
if 170 < event.x < 630 and 60 < event.y < 470: # 在方框内，无变化
if 20 < event.x < self.width * 0.8 + 20 and 20 < event.y < self.height * 0.85 - 20: # 在方框内，无变化
pass
else: # 在方框外，显示禁止放置标识
self.create_oval(event.x - 10, event.y - 45, event.x + 10, event.y - 25, outline="red", width=2,
@ -410,6 +440,7 @@ class NetWorkAnalog(Canvas):
tag = SimSwitch(self, ObjX, ObjY, ObjID, ConfigCorrect, ObjLable)
else:
tag = SimHub(self, ObjX, ObjY, ObjID, ConfigCorrect, ObjLable)
tag.create_img()
self.AllSimObjs[tag.ObjID] = tag
sim_conn_sql = "select s.conn_id, ConfigCorrect, node_id, node_ifs from sim_conn s join conn_config c on s.conn_id=c.conn_id"
@ -423,6 +454,7 @@ class NetWorkAnalog(Canvas):
for key, value in conn_datas.items():
conn_obj = AllSimConnect(self, self.AllSimObjs[value[0][0]], value[0][1],
self.AllSimObjs[value[1][0]], value[1][1], key[1])
conn_obj.draw_line()
self.AllSimObjs[value[0][0]].connections[value[0][1] - 1] = conn_obj # 将连接对象传入组件对象
self.AllSimObjs[value[1][0]].connections[value[1][1] - 1] = conn_obj
self.conns.append(conn_obj)
@ -447,7 +479,9 @@ class NetWorkAnalog(Canvas):
if isinstance(conn, AllSimConnect):
conn.delete_line()
delete_sql = f"delete from sim_conn where conn_id in (select conn_id from conn_config where node_id='{self.chose_obj.ObjID}')"
delete_config_sql = f"delete from conn_config where node_id='{self.chose_obj.ObjID}'"
execute_sql(delete_sql)
execute_sql(delete_config_sql)
delete_obj(self.chose_obj.ObjID)
self.delete("rectangle")
self.reload_data()
@ -532,12 +566,12 @@ class NetWorkAnalog(Canvas):
return
child1 = tk.Toplevel()
child1.title(self.chose_obj.ObjLabel + "的标签信息")
child1.geometry('240x100+450+250')
child1.geometry('360x150+200+150')
child1.grab_set() # 设置组件焦点抓取。使焦点在释放之前永远保持在这个组件上,只能在这个组件上操作
tk.Label(child1, text='原标签:' + self.chose_obj.ObjLabel,
font=('黑体', 12)).grid(row=0, column=0, columnspan=2, sticky='w')
tk.Label(child1, text='新标签:', font=('黑体', 12)).grid(row=1, column=0, sticky='w') # ,sticky='w'靠左显示
new_name = tk.Entry(child1, font=('黑体', 12), textvariable=tk.StringVar())
font=('黑体', 16)).grid(row=0, column=0, columnspan=2, sticky='w')
tk.Label(child1, text='新标签:', font=('黑体', 16)).grid(row=1, column=0, sticky='w') # ,sticky='w'靠左显示
new_name = tk.Entry(child1, font=('黑体', 16), textvariable=tk.StringVar())
new_name.grid(row=1, column=1)
def update_tag():
@ -553,10 +587,10 @@ class NetWorkAnalog(Canvas):
child1.destroy()
messagebox.showinfo("提示", message="修改成功！")
tk.Button(child1, text='确定', font=('黑体', 10), height=1, command=update_tag).grid(row=2, column=0,
sticky='e')
tk.Button(child1, text='取消', font=('黑体', 10), height=1, command=child1.destroy).grid(row=2, column=1,
sticky='e')
tk.Button(child1, text='确定', font=('黑体', 16), height=1, command=update_tag).grid(row=2, column=0,
sticky='e', pady=10)
tk.Button(child1, text='取消', font=('黑体', 16), height=1, command=child1.destroy).grid(row=2, column=1,
sticky='e', pady=10)
def network_config(self):
# todo: 网络配置
@ -572,7 +606,7 @@ class NetWorkAnalog(Canvas):
return
child_r = tk.Toplevel()
child_r.title(self.chose_obj.ObjLabel + "的网络配置信息")
child_r.geometry('530x395+350+200')
child_r.geometry('713x522+350+200')
child_r.grab_set() # 设置组件焦点抓取。使焦点在释放之前永远保持在这个组件上,只能在这个组件上操作
ifs_frame = tk.Frame(child_r)
ifs_frame.grid(row=1, column=0, columnspan=4) # 装接口的框架
@ -585,20 +619,20 @@ class NetWorkAnalog(Canvas):
index = self.chose_obj.connections.index(conn)
num_label = tk.LabelFrame(ifs_frame, text=f'接口{index + 1}')
num_label.grid(row=num, column=0, padx=18, ipady=5)
tk.Label(num_label, text='MAC:', font=('黑体', 10)).grid(row=0, column=0)
mac_en = tk.Entry(num_label, font=('黑体', 12), textvariable=tk.StringVar(), state=DISABLED if self.chose_obj.ObjType not in [1, 2, 3] else NORMAL)
tk.Label(num_label, text='MAC:', font=('黑体', 16)).grid(row=0, column=0)
mac_en = tk.Entry(num_label, font=('黑体', 16), textvariable=tk.StringVar(), state=DISABLED if self.chose_obj.ObjType not in [1, 2, 3] else NORMAL)
mac_en.insert('0', str(self.chose_obj.interface[index].get("mac", "")))
mac_en.grid(row=0, column=1, padx=10)
tk.Label(num_label, text='IP:', font=('黑体', 10)).grid(row=1, column=0)
ip_en = tk.Entry(num_label, font=('黑体', 12), textvariable=tk.StringVar(), state=DISABLED if self.chose_obj.ObjType not in [1, 2] else NORMAL)
tk.Label(num_label, text='IP:', font=('黑体', 16)).grid(row=1, column=0)
ip_en = tk.Entry(num_label, font=('黑体', 16), textvariable=tk.StringVar(), state=DISABLED if self.chose_obj.ObjType not in [1, 2] else NORMAL)
ip_en.insert('0', str(self.chose_obj.interface[index].get("ip", "")))
ip_en.grid(row=1, column=1, padx=10)
tk.Label(num_label, text='端口:', font=('黑体', 10)).grid(row=0, column=2)
port_en = tk.Entry(num_label, font=('黑体', 12), textvariable=tk.StringVar(), state=DISABLED if self.chose_obj.ObjType != 1 else NORMAL)
tk.Label(num_label, text='端口:', font=('黑体', 16)).grid(row=0, column=2)
port_en = tk.Entry(num_label, font=('黑体', 16), textvariable=tk.StringVar(), state=DISABLED if self.chose_obj.ObjType != 1 else NORMAL)
port_en.insert('0', str(self.chose_obj.interface[index].get("conn_port", "")))
port_en.grid(row=0, column=3, padx=10)
tk.Label(num_label, text='应用层地址:', font=('黑体', 10)).grid(row=1, column=2)
addr_en = tk.Entry(num_label, font=('黑体', 12), textvariable=tk.StringVar(), state=DISABLED if self.chose_obj.ObjType != 1 else NORMAL)
tk.Label(num_label, text='应用层地址:', font=('黑体', 16)).grid(row=1, column=2)
addr_en = tk.Entry(num_label, font=('黑体', 16), textvariable=tk.StringVar(), state=DISABLED if self.chose_obj.ObjType != 1 else NORMAL)
addr_en.insert('0', str(self.chose_obj.interface[index].get("addr", "")))
addr_en.grid(row=1, column=3, padx=10)
num += 1
@ -606,23 +640,23 @@ class NetWorkAnalog(Canvas):
num_label = tk.LabelFrame(ifs_frame, text=f'示例')
num_label.grid(row=num, column=0, padx=18, ipady=5)
tk.Label(num_label, text='MAC:', font=('黑体', 10)).grid(row=0, column=0)
mac_en = tk.Entry(num_label, font=('黑体', 12), textvariable=tk.StringVar())
tk.Label(num_label, text='MAC:', font=('黑体', 16)).grid(row=0, column=0)
mac_en = tk.Entry(num_label, font=('黑体', 16), textvariable=tk.StringVar())
mac_en.insert('0', "MAC11")
mac_en.config(state=READONLY)
mac_en.grid(row=0, column=1, padx=10)
tk.Label(num_label, text='IP:', font=('黑体', 10)).grid(row=1, column=0)
ip_en = tk.Entry(num_label, font=('黑体', 12), textvariable=tk.StringVar())
tk.Label(num_label, text='IP:', font=('黑体', 16)).grid(row=1, column=0)
ip_en = tk.Entry(num_label, font=('黑体', 16), textvariable=tk.StringVar())
ip_en.insert(0, "10.1.1.10")
ip_en.config(state=READONLY)
ip_en.grid(row=1, column=1, padx=10)
tk.Label(num_label, text='端口:', font=('黑体', 10)).grid(row=0, column=2)
port_en = tk.Entry(num_label, font=('黑体', 12), textvariable=tk.StringVar())
tk.Label(num_label, text='端口:', font=('黑体', 16)).grid(row=0, column=2)
port_en = tk.Entry(num_label, font=('黑体', 16), textvariable=tk.StringVar())
port_en.insert('0', "80")
port_en.config(state=READONLY)
port_en.grid(row=0, column=3, padx=10)
tk.Label(num_label, text='应用层地址:', font=('黑体', 10)).grid(row=1, column=2)
addr_en = tk.Entry(num_label, font=('黑体', 12), textvariable=tk.StringVar())
tk.Label(num_label, text='应用层地址:', font=('黑体', 16)).grid(row=1, column=2)
addr_en = tk.Entry(num_label, font=('黑体', 16), textvariable=tk.StringVar())
addr_en.insert('0', "10.1.2.10:10810:Name3")
addr_en.config(state=READONLY)
addr_en.grid(row=1, column=3, padx=10)
@ -630,10 +664,10 @@ class NetWorkAnalog(Canvas):
self.chose_obj.config(datas)
child_r.destroy()
tk.Button(ifs_frame_YN, text='确定', font=('黑体', 10), height=1, command=commit).grid(row=0, column=0,
padx=20) # ,sticky="w"
tk.Button(ifs_frame_YN, text='取消', font=('黑体', 10), height=1,
command=child_r.destroy).grid(row=0, column=2, padx=20)
tk.Button(ifs_frame_YN, text='确定', font=('黑体', 16), height=1, command=commit).grid(row=0, column=0,
padx=20, pady=20) # ,sticky="w"
tk.Button(ifs_frame_YN, text='取消', font=('黑体', 16), height=1,
command=child_r.destroy).grid(row=0, column=2, padx=20, pady=20)
def router_table_config(self):
# todo: 路由表配置
@ -700,10 +734,10 @@ class NetWorkAnalog(Canvas):
messagebox.showerror("注意", message="请先选择要显示的对象！")
return
self.delete("netSet")
self.create_text(740, 120, text="(" + self.chose_obj.ObjLabel + ")", anchor="n", font=('微软雅黑', 10, 'bold'),
self.create_text(self.width * 0.82 + 120, 80, text="(" + self.chose_obj.ObjLabel + ")", anchor="n", font=('微软雅黑', 14, 'bold'),
fill="#7030a0", tags="netSet")
self.create_text(675, 145, text=self.chose_obj,
anchor="nw", font=('宋体', 11), tags="netSet")
self.create_text(self.width * 0.82 + 20, 80 + 25, text=self.chose_obj,
anchor="nw", font=('宋体', 14), tags="netSet")
def show_router_config(self):
# todo: 显示路由表交换表信息
@ -718,10 +752,10 @@ class NetWorkAnalog(Canvas):
messagebox.showerror("注意", message="请选择路由器/交换机对象！")
return
self.delete("routerSet")
self.create_text(905, 120, text="(" + self.chose_obj.ObjLabel + ")", anchor="n", font=('微软雅黑', 10, 'bold'),
self.create_text(self.width * 0.82 + 120, self.height * 0.85 / 2 + 50, text="(" + self.chose_obj.ObjLabel + ")", anchor="n", font=('微软雅黑', 14, 'bold'),
fill="#7030a0", tags="routerSet")
self.create_text(835, 145, text=self.chose_obj.get_table_config(),
anchor="nw", font=('宋体', 11), tags="routerSet")
self.create_text(self.width * 0.82 + 20, self.height * 0.85 / 2 + 50 + 25, text=self.chose_obj.get_table_config(),
anchor="nw", font=('宋体', 14), tags="routerSet")
def send_packet(self):
"""
@ -739,17 +773,17 @@ class NetWorkAnalog(Canvas):
return
child2 = tk.Toplevel()
child2.title("数据包配置")
child2.geometry('240x100+450+250')
child2.geometry('330x195+200+110')
child2.grab_set() # 设置组件焦点抓取。使焦点在释放之前永远保持在这个组件上,只能在这个组件上操作
tk.Label(child2, text='目的IP:', font=('黑体', 12)).grid(row=0, column=0, columnspan=2, sticky='w')
packet_ip = tk.Entry(child2, font=('黑体', 12), textvariable=tk.StringVar())
packet_ip.grid(row=0, column=1)
tk.Label(child2, text='目的MAC:', font=('黑体', 12)).grid(row=1, column=0, sticky='w') # ,sticky='w'靠左显示
packet_mac = tk.Entry(child2, font=('黑体', 12), textvariable=tk.StringVar())
packet_mac.grid(row=1, column=1)
tk.Label(child2, text='消息:', font=('黑体', 12)).grid(row=2, column=0, sticky='w') # ,sticky='w'靠左显示
packet_message = tk.Entry(child2, font=('黑体', 12), textvariable=tk.StringVar())
packet_message.grid(row=2, column=1)
tk.Label(child2, text='目的IP:', font=('黑体', 16)).grid(row=0, column=0, columnspan=2, sticky='w', pady=10)
packet_ip = tk.Entry(child2, font=('黑体', 16), textvariable=tk.StringVar())
packet_ip.grid(row=0, column=1, pady=10)
tk.Label(child2, text='目的MAC:', font=('黑体', 16)).grid(row=1, column=0, sticky='w', pady=10) # ,sticky='w'靠左显示
packet_mac = tk.Entry(child2, font=('黑体', 16), textvariable=tk.StringVar())
packet_mac.grid(row=1, column=1, pady=10)
tk.Label(child2, text='消息:', font=('黑体', 16)).grid(row=2, column=0, sticky='w', pady=10) # ,sticky='w'靠左显示
packet_message = tk.Entry(child2, font=('黑体', 16), textvariable=tk.StringVar())
packet_message.grid(row=2, column=1, pady=10)
def send():
"""
@ -773,9 +807,8 @@ class NetWorkAnalog(Canvas):
packet_message.get())
child2.destroy()
tk.Button(child2, text='确定', font=('黑体', 10), height=1, command=send).grid(row=3, column=0, sticky='e')
tk.Button(child2, text='取消', font=('黑体', 10), height=1, command=child2.destroy).grid(row=3, column=1,
sticky='e')
tk.Button(child2, text='确定', font=('黑体', 16), height=1, command=send).grid(row=3, column=0, sticky='e', pady=10)
tk.Button(child2, text='取消', font=('黑体', 16), height=1, command=child2.destroy).grid(row=3, column=1, sticky='e', pady=10)
def send_packet_list(self):
"""
@ -788,26 +821,26 @@ class NetWorkAnalog(Canvas):
hosts[tag.ObjLabel] = tag.ObjID
child2 = tk.Toplevel()
child2.title("批量数据包配置")
child2.geometry('400x420+450+200')
tk.Label(child2, text='目的IP:', font=('黑体', 12)).grid(row=0, column=0, columnspan=2, sticky='w', ipady=10)
packet_ip = tk.Entry(child2, font=('黑体', 12), textvariable=tk.StringVar())
child2.geometry('462x452+200+110')
tk.Label(child2, text='目的IP:', font=('黑体', 16)).grid(row=0, column=0, columnspan=2, sticky='w', ipady=10)
packet_ip = tk.Entry(child2, font=('黑体', 16), textvariable=tk.StringVar())
packet_ip.grid(row=0, column=1, pady=5)
tk.Label(child2, text='目的MAC:', font=('黑体', 12)).grid(row=1, column=0, sticky='w',
tk.Label(child2, text='目的MAC:', font=('黑体', 16)).grid(row=1, column=0, sticky='w',
pady=5) # ,sticky='w'靠左显示
packet_mac = tk.Entry(child2, font=('黑体', 12), textvariable=tk.StringVar())
packet_mac = tk.Entry(child2, font=('黑体', 16), textvariable=tk.StringVar())
packet_mac.grid(row=1, column=1, pady=5)
tk.Label(child2, text='消息:', font=('黑体', 12)).grid(row=2, column=0, sticky='w', pady=5) # ,sticky='w'靠左显示
packet_message = tk.Entry(child2, font=('黑体', 12), textvariable=tk.StringVar())
tk.Label(child2, text='消息:', font=('黑体', 16)).grid(row=2, column=0, sticky='w', pady=5) # ,sticky='w'靠左显示
packet_message = tk.Entry(child2, font=('黑体', 16), textvariable=tk.StringVar())
packet_message.grid(row=2, column=1, pady=5)
host = StringVar()
tk.Label(child2, text='发送主机:', font=('黑体', 12)).grid(row=3, column=0, sticky='w',
tk.Label(child2, text='发送主机:', font=('黑体', 16)).grid(row=3, column=0, sticky='w',
pady=5) # ,sticky='w'靠左显示
combobox = ttk.Combobox(
master=child2, # 父容器
height=5, # 高度,下拉显示的条目数量
width=12, # 宽度
state='readonly', # readonly(只可选)
font=('', 18), # 字体
font=('', 16), # 字体
textvariable=host, # 通过StringVar设置可改变的值
values=list(hosts.keys()), # 设置下拉框的选项
)
@ -856,10 +889,10 @@ class NetWorkAnalog(Canvas):
threading.Thread(target=data[0].send, args=(data[1],)).start()
child2.destroy()
tk.Button(child2, text="添加数据包", font=('黑体', 12), height=3, command=add).grid(row=0, column=2, rowspan=4)
tk.Button(child2, text="添加数据包", font=('黑体', 14), height=3, command=add).grid(row=0, column=2, rowspan=4, padx=10)
button_frame = Frame(child2)
tk.Button(button_frame, text="发送", font=('黑体', 12), height=1, command=send).pack(side=RIGHT, padx=20)
tk.Button(button_frame, text="取消", font=('黑体', 12), height=1, command=child2.destroy).pack(side=RIGHT)
tk.Button(button_frame, text="发送", font=('黑体', 16), height=1, command=send).pack(side=RIGHT, padx=20)
tk.Button(button_frame, text="取消", font=('黑体', 16), height=1, command=child2.destroy).pack(side=RIGHT)
button_frame.grid(row=6, column=0, columnspan=3)
def clear_canvas(self):
@ -886,20 +919,21 @@ class NetWorkAnalog(Canvas):
创建整体页面布局
:return:
"""
self.create_rectangle(150, 40, 650, 490, outline="#7f6000", width=3) # 矩形框，左上角坐标，右下角坐标
self.create_rectangle(660, 100, 815, 400, outline="#ffff00", width=3, fill="#f4b88e") # 矩形框，左上角坐标，右下角坐标
self.create_text(735, 105, text="网络配置信息", anchor="n", font=('微软雅黑', 11, 'bold'))
self.create_rectangle(825, 100, 1000, 400, outline="#ffff00", width=3, fill="#f4b88e") # 矩形框，左上角坐标，右下角坐标
self.create_text(915, 105, text="路由/交换表信息", anchor="n", font=('微软雅黑', 11, 'bold'))
self.create_rectangle(0, 0, self.width * 0.8, self.height * 0.85, outline="#7f6000", width=3) # 矩形框，左上角坐标，右下角坐标
round_rectangle(self, self.width * 0.82, 30, self.width * 0.98, self.height * 0.85 / 2 - 30, outline="#ffff00", width=3, fill="#f4b88e") # 矩形框，左上角坐标，右下角坐标
self.create_text(self.width * 0.82 + 120, 30 + 15, text="网络配置信息", anchor="n", font=('微软雅黑', 18, 'bold'))
round_rectangle(self, self.width * 0.82, self.height * 0.85 / 2, self.width * 0.98, self.height * 0.85 - 30, outline="#ffff00", width=2, fill="#f4b88e") # 矩形框，左上角坐标，右下角坐标
self.create_text(self.width * 0.82 + 120, self.height * 0.85 / 2 + 15, text="路由/交换表信息", anchor="n", font=('微软雅黑', 18, 'bold'))
# 显示左边的固定图片
self.create_text(80 - 55, 120 + 15, text="路由器", anchor="nw", font=('微软雅黑', 14, 'bold')) # 显示文字
router = self.create_image(80, 120, image=self.router_img, anchor="nw")
self.create_text(80 - 55, 190 + 15, text="交换机", anchor="nw", font=('微软雅黑', 14, 'bold')) # 显示文字
switch = self.create_image(80, 190, image=self.switch_img, anchor="nw")
self.create_text(80 - 55, 260 + 15, text="集线器", anchor="nw", font=('微软雅黑', 14, 'bold')) # 显示文字
hub = self.create_image(80, 260, image=self.hub_img, anchor="nw")
self.create_text(80 - 55, 330 + 15, text="主机", anchor="nw", font=('微软雅黑', 14, 'bold')) # 显示文字
host = self.create_image(80, 330, image=self.host_img, anchor="nw")
img_height, text_height, split_width = self.height - 120, self.height - 60, 120
self.create_text(split_width, text_height, text="路由器", anchor="nw", font=('微软雅黑', 18, 'bold')) # 显示文字
router = self.create_image(split_width, img_height, image=self.router_img, anchor="nw")
self.create_text(split_width * 2, text_height, text="交换机", anchor="nw", font=('微软雅黑', 18, 'bold')) # 显示文字
switch = self.create_image(split_width * 2, img_height, image=self.switch_img, anchor="nw")
self.create_text(split_width * 3, text_height, text="集线器", anchor="nw", font=('微软雅黑', 18, 'bold')) # 显示文字
hub = self.create_image(split_width * 3, img_height, image=self.hub_img, anchor="nw")
self.create_text(split_width * 4, text_height, text="主机", anchor="nw", font=('微软雅黑', 18, 'bold')) # 显示文字
host = self.create_image(split_width * 4, img_height, image=self.host_img, anchor="nw")
self.bind_event(router, "路由器")
self.bind_event(switch, "交换机")
self.bind_event(hub, "集线器")
@ -941,14 +975,15 @@ class NetWorkAnalog(Canvas):
if __name__ == '__main__':
root = Window()
root.title('网络拓扑图')
width = 1030
height = 530 # 窗口大小
screen_width = root.winfo_screenwidth() # winfo方法来获取当前电脑屏幕大小
screen_height = root.winfo_screenheight()
x = int((screen_width - width) / 2)
y = int((screen_height - height) / 2) - 40
size = '{}x{}+{}+{}'.format(width, height, x, y)
canvas = NetWorkAnalog(root, width=1030, heigh=height, bg="white")
root_attr = {
"width": screen_width * 0.83,
"height": screen_height * 0.85,
}
size = '%dx%d+%d+%d' % (root_attr['width'], root_attr['height'], (screen_width - root_attr['width']) / 2,
(screen_height - root_attr['height']) / 2 - 30)
canvas = NetWorkAnalog(root, width=root_attr['width'], heigh=root_attr['height'], bg="white")
canvas.place(x=0, y=0, anchor='nw')
root.geometry(size)
root.mainloop()

757
NetworkAnalog/NetworkAnalog_X1.py
Unescape View File

@ -1,757 +0,0 @@
import math
import sys
import threading
from time import sleep
from ttkbootstrap import *
from uuid import uuid4
import ipaddress
from PIL import ImageTk, Image
from dbUtil import search, execute_sql
class SimBase():
# todo: 组件父类
"""
图标类，所有组件的父类
"""
def __init__(self, canvas: Canvas, x, y, id, config=None, label=None):
self.ConfigCorrect = 0 if config is None else config # 是否进行配置
self.ObjType = None # 组件类型 1->主机，2->路由器，3->交换机，4->集线器
self.ObjLabel = ""
self.ObjID = id
self.ObjX = x
self.ObjY = y
self.canvas = canvas
self.img = None
self.img_tm = None
self.interface = [{}, {}, {}, {}]
self.connections = ["1", "2", "3", "4"]
self.set_default_config()
def bind_event(self):
self.canvas.tag_bind(self.ObjID, "<Button-1>", self.start_drag)
self.canvas.tag_bind(self.ObjID, "<B1-Motion>", self.drag)
self.canvas.tag_bind(self.ObjID, "<ButtonRelease-1>", self.release)
def set_default_name(self):
data_frame = search(f"select objid from sim_objs where objType={self.ObjType}")
num = data_frame.size + 1
if isinstance(self, SimHost):
name = "SHO%d" % num
elif isinstance(self, SimRouter):
name = "SRO%d" % num
elif isinstance(self, SimSwitch):
name = "SWI%d" % num
else:
name = "SHUB%d" % num
return name
def release(self, event):
if self.click_x == event.x and self.click_y == event.y: # 鼠标左键单击
self.canvas.chose_obj = self
self.canvas.is_chose()
else:
self.update()
def start_drag(self, event):
self.canvas.tag_raise(self.ObjID) # 将 SimBase 组件置于最上层
self.start_x = event.x
self.start_y = event.y
self.click_x = event.x
self.click_y = event.y
def drag(self, event):
"""
移动图标
:param event:
:return:
"""
self.canvas.delete("rectangle")
self.canvas.chose_obj = None
dx = event.x - self.start_x
dy = event.y - self.start_y
# 移动范围限制, 超出移动范围则直接返回
if not (170 <= self.ObjX + dx <= 630 and 60 <= self.ObjY + dy <= 470):
return
self.ObjX += dx
self.ObjY += dy
self.canvas.move(self.ObjID, dx, dy) # 移动 SimBase 组件
self.canvas.move(self.ObjID + "text", dx, dy) # 移动 SimBase 组件
self.start_x = event.x
self.start_y = event.y
for conn in self.connections:
if isinstance(conn, AllSimConnect):
conn.update_line()
def create_img(self):
"""
创建图片
:return:
"""
self.canvas.delete("L")
id = self.canvas.create_image(self.ObjX - 30, self.ObjY - 30,
image=self.img if self.ConfigCorrect == 1 else self.img_tm, anchor="nw",
tags=self.ObjID)
self.canvas.dtag("L", "L")
self.canvas.create_text(self.ObjX, self.ObjY - 40, text=self.ObjLabel, font=("", 12, "bold"),
fill="#7030a0", tags=self.ObjID + "text", anchor="nw")
self.canvas.tag_raise(id)
self.bind_event()
def config(self, interface):
"""
网络配置方法，
:param interface: 传入配置数据
:return:
"""
for key, value in interface.items():
self.interface[key - 1] = {key: value.get() if not value.get() == "" else "NULL" for key, value in value.items()}
self.ConfigCorrect = 1
self.create_img()
for conn in self.connections:
if isinstance(conn, AllSimConnect):
index = self.connections.index(conn) + 1
conn.update_node_config(self, index)
self.update()
def set_default_config(self):
sql = f"select * from conn_config where node_id='{self.ObjID}'"
conn_data = search(sql)
for index, conn in conn_data.iterrows():
self.interface[int(conn["node_ifs"]) - 1] = {"ip": conn["ip"],
"mac": conn["mac"],
"conn_port": conn["conn_port"],
"addr": conn["addr"]}
def get_config(self):
sql = f"select * from conn_config where node_id='{self.ObjID}'"
conn_data = search(sql)
return conn_data
def save(self):
"""
将对象存储至mysql当中
:return:
"""
sql = f"insert into sim_objs values ('{self.ObjID}', {self.ObjType}, '{self.ObjLabel}'," \
f"{self.ObjX}, {self.ObjY}, {self.ConfigCorrect})"
execute_sql(sql)
def update(self):
"""
当坐标发生改变时修改数据库数据
:return:
"""
self.canvas.delete(self.ObjID + "text")
self.canvas.create_text(self.ObjX, self.ObjY - 40, text=self.ObjLabel, font=("", 12, "bold"),
fill="#7030a0", tags=self.ObjID + "text", anchor="nw")
sql = f"update sim_objs set objLabel='{self.ObjLabel}', ObjX={self.ObjX}," \
f"ObjY={self.ObjY}, ConfigCorrect={self.ConfigCorrect} where ObjID='{self.ObjID}'"
execute_sql(sql)
def transfer_animate(self, status, packet, error_message=None):
if status:
text = f"目的IP: {str(packet.destination_ip)}\n" \
f"目的MAC: {packet.destination_mac}\n" \
f"消息内容: {packet.message}"
self.canvas.create_rectangle(self.ObjX + 30, self.ObjY - 30, self.ObjX + 160, self.ObjY + 20, outline="#92d050",
width=3, fill="#92d050", tags=self.ObjID + "packetData")
self.canvas.create_text(self.ObjX + 35, self.ObjY - 25, text=text, anchor="nw",
font=('', 10), tags=self.ObjID + "packetData") # 显示文字
self.canvas.update()
sleep(2)
self.canvas.delete(self.ObjID + "packetData") # 删除展示的数据包内容
else:
text = f"传输失败\n" if error_message is None else error_message
self.canvas.create_rectangle(self.ObjX + 30, self.ObjY - 30, self.ObjX + 160, self.ObjY, outline="red",
width=3, fill="red", tags=self.ObjID + "packetData")
self.canvas.create_text(self.ObjX + 35, self.ObjY - 25, text=text, anchor="nw",
font=('', 10), tags=self.ObjID + "packetData") # 显示文字
self.canvas.update()
sleep(2)
self.canvas.delete(self.ObjID + "packetData") # 删除展示的数据包内容
def __str__(self):
str = ""
config = self.get_config()
for index, data in config.iterrows():
str += f"【接口{data['node_ifs']}】\n"
str += f"IP: {data['ip']}\n"
str += f"MAC: {data['mac']}\n"
return str
class SimPacket():
# todo: 数据包类
"""
数据包类
"""
def __init__(self, source_ip, source_mac, destination_ip, destination_mac, message):
"""
:param source_mac: 源主机mac地址
:param source_ip: 源主机ip地址
:param destination_mac: 目的主机mac地址
:param destination_ip: 目的主机ip地址
:param message: 数据
"""
self.source_mac = source_mac
self.source_ip = source_ip
self.destination_mac = destination_mac
self.destination_ip = destination_ip
self.message = message
self.up_jump = None # 上一跳连接对象
self.img = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/packet.png").resize((30, 30)))
self.id = str(uuid4())
def move(self, cv, target_x, target_y, duration):
start_x, start_y = cv.coords(self.id)
distance_x = target_x - start_x
distance_y = target_y - start_y
steps = duration // 10 # 以10毫秒为间隔进行移动
step_x = distance_x / steps
step_y = distance_y / steps
self._move_step(cv, start_x, start_y, target_x, target_y, step_x, step_y, steps)
def _move_step(self, cv, start_x, start_y, target_x, target_y, step_x, step_y, steps):
if steps > 0:
new_x = start_x + step_x
new_y = start_y + step_y
cv.coords(self.id, new_x, new_y)
cv.update() # 更新画布显示
sleep(0.01) # 添加延迟以控制动画速度
self._move_step(cv, new_x, new_y, target_x, target_y, step_x, step_y, steps - 1)
else:
cv.coords(self.id, target_x, target_y)
cv.delete(self.id)
def transfer_packet(self, cv: Canvas, nodex_tag: SimBase, nodey_tag: SimBase):
cv.create_image(nodex_tag.ObjX - 15, nodex_tag.ObjY - 15, image=self.img, anchor="nw", tags=self.id)
self.move(cv, nodey_tag.ObjX - 15, nodey_tag.ObjY - 15, 500)
class AllSimConnect():
# todo: 连接类
def __init__(self, canvas: Canvas, nodex: SimBase, nodex_ifs, nodey: SimBase, nodey_ifs, config=None):
"""
连接对象
:param nodex: 节点
:param nodex_ifs: 节点接口
:param nodey: 节点
:param nodey_ifs: 节点接口
"""
self.canvas = canvas
self.ConfigCorrect = 0 if config is None else config
self.NobjS = nodex
self.NobjE = nodey
self.IfsS = nodex_ifs
self.IfsE = nodey_ifs
self.width = 1 if self.ConfigCorrect == 0 else 4 # 线的粗细
self.draw_line()
def is_connected_to(self, other):
"""
判断两个连接对象是否已经连接
:param other: 另一个连接对象
:return: 如果已连接，则返回True；否则返回False
"""
return (
self.NobjS.ObjID == other.NobjS.ObjID
and self.NobjE.ObjID == other.NobjE.ObjID
)
def update_node_config(self, node, ifs):
"""
当节点进行网络配置后将节点的接口配置保存
:return:
"""
try:
nodex_update_sql = f"""
update conn_config set
ip='{node.interface[ifs - 1]["ip"]}',
mac='{node.interface[ifs - 1]["mac"]}',
conn_port='{node.interface[ifs - 1]["conn_port"]}',
addr='{node.interface[ifs - 1]["addr"]}'
where node_id='{node.ObjID}' and node_ifs={ifs}
"""
execute_sql(nodex_update_sql)
self.ConfigCorrect = 1
self.check_config()
update_sql = f"""
update sim_conn set ConfigCorrect={self.ConfigCorrect}
where conn_id in (
select distinct conn_id from conn_config where node_id='{node.ObjID}'
)
"""
execute_sql(update_sql)
except Exception as E:
pass
def check_config(self):
"""
检查两边节点的配置是否正确
:return:
"""
if self.NobjS.interface[self.IfsS - 1]["mac"] != "" and self.NobjS.interface[self.IfsS - 1][
"ip"] != "" \
and self.NobjE.interface[self.IfsE - 1]["mac"] != "" and self.NobjE.interface[self.IfsE - 1][
"ip"] != "":
self.width = 4
self.draw_line()
else:
self.width = 1
self.draw_line()
def transfer(self, source_node, packet: SimPacket):
"""
传输数据
:param packet: 数据包
:return:
"""
if source_node == self.NobjS:
packet.transfer_packet(self.canvas, self.NobjS, self.NobjE)
self.NobjE.receive(packet)
else:
packet.transfer_packet(self.canvas, self.NobjE, self.NobjS)
self.NobjS.receive(packet)
def draw_line(self):
line = self.canvas.create_line(self.NobjS.ObjX, self.NobjS.ObjY, self.NobjE.ObjX, self.NobjE.ObjY,
width=self.width, fill="#5b9bd5", tags=self.NobjS.ObjID + self.NobjE.ObjID + "line")
self.canvas.tag_lower(line)
self.analyseIFS(self.NobjS.ObjX, self.NobjS.ObjY, self.NobjS.ObjLabel, self.IfsS,
self.NobjE.ObjX, self.NobjE.ObjY, self.NobjE.ObjLabel, self.IfsE)
def update_line(self):
"""
当组件移动时重新绘制线
:return:
"""
self.canvas.delete(self.NobjS.ObjID + self.NobjE.ObjID + "line")
self.canvas.delete(self.NobjS.ObjID + self.NobjE.ObjID)
self.draw_line()
def update_info(self, obj_id):
"""
修改数据库中的连接信息
:return:
"""
update_sql = f"""
update sim_conn set conn_id='{self.NobjS.ObjLabel}-{self.NobjE.ObjLabel}', ConfigCorrect={self.ConfigCorrect}
where conn_id in (
select distinct conn_id from conn_config where node_id='{obj_id}'
)
"""
execute_sql(update_sql)
def analyseIFS(self, SX, SY, SLabel, IfsS, EX, EY, ELabel, IfsE): # NobjS的x,y;NobjE的x,y; #分析接口在哪个位置，再显示
'''
:param SX: S对象的x坐标
:param SY: S对象的y坐标
:param SLabel: S对象的标签
:param IfsS: S对象要显示的接口号
'''
if (EX - SX) == 0: # 即垂直的时候,NobjS在NobjE对象的正上方和正下方，x坐标无变化，只需要y变化
R = 18
x = 0 # x无偏移量
y = R
else:
k = (EY - SY) / (EX - SX) # ey-sy/sx-ex
reat = math.atan(k) # 根据斜率计算弧度
# 连线与图标交接点坐标
R = 18
x = abs(math.cos(reat) * R) + 6 # python math中三角函数中的数值是弧度，而计算器中的数值是角度
y = abs(math.sin(reat) * R) + 6
if SX <= EX and SY <= EY: # NobjS在NobjE的左上角，NobjS的右下角连接NobjE的左上角
x_S = SX + x
y_S = SY + y # NobjS的连接点坐标
x_E = EX - x
y_E = EY - y # NobjE的连接点坐标
# 显示S接口号
self.canvas.create_text((x_S - 5, y_S - 5), text=IfsS, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
# 显示E接口号
self.canvas.create_text(x_E - 10, y_E - 10, text=IfsE, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
elif SX < EX and SY > EY: # NobjS在NobjE的左下角，NobjS的右上角连接NobjE的左下角
x_S = SX + x
y_S = SY - y # NobjS的连接点坐标
x_E = EX - x
y_E = EY + y # NobjE的连接点坐标
# 显示S接口号
self.canvas.create_text(x_S + 5, y_S - 10, text=IfsS, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
# 显示E接口号
self.canvas.create_text(x_E - 5, y_E, text=IfsE, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
elif SX > EX and SY < EY: # NobjS在NobjE的右上角，NobjS的左下角连接NobjE的右上角
x_S = SX - x
y_S = SY + y # NobjS的连接点坐标
x_E = EX + x
y_E = EY - y # NobjE的连接点坐标
# 显示S接口号
self.canvas.create_text(x_S - 5, y_S, text=IfsS, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
# 显示E接口号
self.canvas.create_text(x_E + 5, y_E, text=IfsE, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
elif SX >= EX and SY >= EY: # NobjS在NobjE的右下角，NobjS的左上角连接NobjE的右下角
x_S = SX - x
y_S = SY - y # NobjS的连接点坐标
x_E = EX + x
y_E = EY + y # NobjE的连接点坐标
# 显示S接口号
self.canvas.create_text(x_S - 5, y_S - 15, text=IfsS, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
# 显示E接口号
self.canvas.create_text(x_E - 5, y_E - 5, text=IfsE, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
def save(self):
"""
将连接对象保存至数据库
:return:
"""
conn_id = self.NobjS.ObjLabel + "-" + self.NobjE.ObjLabel
sql = f"insert into sim_conn values ('{conn_id}', {self.ConfigCorrect})"
execute_sql(sql)
execute_sql(
f"insert into conn_config values ('{conn_id}', '{self.NobjS.ObjID}', {self.IfsS}, '', '', '', '')")
execute_sql(
f"insert into conn_config values ('{conn_id}', '{self.NobjE.ObjID}', {self.IfsE}, '', '', '', '')")
def delete_line(self):
self.canvas.delete(self.NobjS.ObjID + self.NobjE.ObjID)
self.canvas.delete(self.NobjS.ObjID + self.NobjE.ObjID + "line")
def __eq__(self, other):
"""
重写equals方法，判断两个连接对象是否相同
:param other: 连接对象
:return: Boolean
"""
if isinstance(other, AllSimConnect):
other_ids = other.NobjS.ObjID + other.NobjE.ObjID
return (self.NobjS.ObjID in other_ids
and self.NobjE.ObjID in other_ids)
return False
class SimHost(SimBase):
# todo: 主机类
"""
主机类
"""
def __init__(self, canvas: Canvas, x, y, id=None, config=None, label=None):
self.ObjID = str(uuid4()) if id is None else id
super().__init__(canvas, x, y, self.ObjID, config, label)
self.ObjType = 1
self.ObjLabel = label if label is not None else self.set_default_name()
self.interface = [{}]
self.connections = [None]
self.img = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/主机.png").resize((60, 60)))
self.img_tm = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/主机_tm.png").resize((60, 60)))
self.set_default_config()
self.create_img()
def create_packet(self, ip, mac, message):
"""
创建数据包
:param ip: 目的主机ip
:param mac: 目的主机mac
:param message: 消息
:return:
"""
packet = SimPacket(self.interface[0]["ip"], self.interface[0]["mac"], ip, mac, message)
print(f"创建数据包成功,数据包由{packet.source_ip} 发往 {packet.destination_ip}")
self.send(packet)
def send(self, packet):
"""
发送数据包
:param packet:
:return:
"""
connection: AllSimConnect = self.connections[0]
print(f"数据包从 {self.ObjLabel} 发出")
packet.up_jump = connection
connection.transfer(self, packet)
def receive(self, packet: SimPacket):
"""
接收数据
:param packet: 数据包
:return:
"""
print(f"主机{self.ObjLabel}接受到数据{packet.message}")
if packet.destination_ip == self.interface[0]["ip"]:
self.transfer_animate(True, packet)
else:
self.transfer_animate(False, packet)
def __str__(self):
str = ""
config = self.get_config()
for index, data in config.iterrows():
str += f"【接口{data['node_ifs']}】\n"
str += f"AppAddr: /Root\n"
str += f"PORT: {data['conn_port']}\n"
str += f"IP: {data['ip']}\n"
str += f"MAC: {data['mac']}\n"
return str
class SimRouter(SimBase):
# todo: 路由类
"""
路由类
"""
def __init__(self, canvas: Canvas, x, y, id=None, config=None, label=None, *args):
self.ObjID = str(uuid4()) if id is None else id
super().__init__(canvas, x, y, self.ObjID, config, label)
self.ObjType = 2
self.ObjLabel = label if label is not None else self.set_default_name()
self.img = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/路由器.png").resize((60, 60)))
self.img_tm = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/路由器_tm.png").resize((60, 60)))
self.create_img()
self.router_table = {}
self.set_default_router_table()
def set_default_router_table(self):
"""
将数据库中的路由表信息提取
:return:
"""
sql = f"select * from router_table where obj_id='{self.ObjID}'"
router_tables = search(sql)
for index, router_table in router_tables.iterrows():
if router_table["node_ifs"] in self.router_table:
self.router_table[router_table["node_ifs"]].append(router_table["segment"])
else:
self.router_table[router_table["node_ifs"]] = [router_table["segment"]]
def check_destination_ip(self, destination_ip, network):
"""
检查目标ip是否属于网段范围内
:param destination_ip: 目标ip
:param network: 网段
:return:10.2.3.0/24
"""
ip = ipaddress.ip_address(destination_ip)
network = ipaddress.ip_network(network)
if ip in network:
return True
if network == "0.0.0.0/24": # 如果网段为 0.0.0.0/24 则为默认路由
return True
def transmit(self, packet: SimPacket):
"""
转发数据包
:return:
"""
flag = False
next_hop_ifs = None
for conn in self.connections:
if isinstance(conn, AllSimConnect):
if conn.ConfigCorrect == 0:
continue
if conn == packet.up_jump:
continue
ifs = self.connections.index(conn) + 1
for network in self.router_table[ifs]:
if self.check_destination_ip(packet.destination_ip, network):
flag = True
next_hop_ifs = ifs
if flag:
conn = self.connections[next_hop_ifs - 1]
packet.up_jump = conn
conn.transfer(self, packet)
else:
for conn in self.connections:
if isinstance(conn, AllSimConnect):
if conn == packet.up_jump:
continue
if conn.NobjS != self:
if conn.NobjE.ObjType == 1:
conn.transfer(self, packet)
break
error_message = "路由寻址失败"
self.transfer_animate(False, packet, error_message)
def receive(self, packet):
"""
接收数据
:param packet: 数据包
:return:
"""
print(f"{self.ObjLabel}-路由器接受到数据{packet.message}")
self.transmit(packet)
def add_config(self, router, router_ifs):
sql = f"insert into router_table values ('{self.ObjID}', {router_ifs}, '{router}')"
execute_sql(sql)
def delete_config(self, ifs, network):
sql = f"delete from router_table where obj_id='{self.ObjID}' and node_ifs={ifs} and segment='{network}'"
execute_sql(sql)
def get_table_config(self):
"""
返回对象的路由表配置信息,用于展示
:return:
"""
str = ""
sql = f"select * from router_table where obj_id='{self.ObjID}'"
router_tables = search(sql)
for index, router in router_tables.iterrows():
str += f"网段号: {router['segment']}\n"
str += f"接口号: {router['node_ifs']}\n"
return str
class SimSwitch(SimBase):
# todo: 交换机类
"""
交换机类
"""
def __init__(self, canvas: Canvas, x, y, id=None, config=None, label=None, *args):
self.ObjID = str(uuid4()) if id is None else id
super().__init__(canvas, x, y, self.ObjID, config, label)
self.ObjType = 3
self.ObjLabel = label if label is not None else self.set_default_name()
self.img = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/交换机.png").resize((60, 60)))
self.img_tm = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/交换机_tm.png").resize((60, 60)))
self.create_img()
self.mac_table = {}
self.set_default_mac_table()
def set_default_mac_table(self):
"""
将数据库中的路由表信息提取
:return:
"""
sql = f"select * from mac_table where obj_id='{self.ObjID}'"
router_tables = search(sql)
for index, router_table in router_tables.iterrows():
if router_table["node_ifs"] in self.mac_table:
self.mac_table[router_table["node_ifs"]].append(router_table["mac"])
else:
self.mac_table[router_table["node_ifs"]] = [router_table["mac"]]
def add_config(self, router, router_ifs):
sql = f"insert into mac_table values ('{self.ObjID}', {router_ifs}, '{router}')"
execute_sql(sql)
def delete_config(self, ifs, mac):
sql = f"delete from mac_table where obj_id='{self.ObjID}' and node_ifs={ifs} and mac='{mac}'"
execute_sql(sql)
def get_table_config(self):
"""
返回对象的交换表配置信息,用于展示
:return:
"""
str = ""
sql = f"select * from mac_table where obj_id='{self.ObjID}'"
router_tables = search(sql)
for index, router in router_tables.iterrows():
str += f"网段号: {router['mac']}\n"
str += f"接口号: {router['node_ifs']}\n"
return str
def transmit(self, packet: SimPacket):
"""
转发数据包
:return:
"""
flag = False
next_hub_ifs = None
for conn in self.connections:
if isinstance(conn, AllSimConnect):
if conn.ConfigCorrect == 0:
continue
ifs = self.connections.index(conn) + 1
if packet.destination_mac in self.mac_table.get(ifs, []):
flag = True
next_hub_ifs = ifs
if flag:
conn = self.connections[next_hub_ifs - 1]
packet.up_jump = conn
conn.transfer(self, packet)
return
for conn in self.connections: # 将数据包往所有接口进行转发
if isinstance(conn, AllSimConnect):
if conn == packet.up_jump:
continue
if conn.ConfigCorrect == 0:
continue
new_packet = SimPacket(packet.source_ip,
packet.source_mac,
packet.destination_ip,
packet.destination_mac,
packet.message)
new_packet.up_jump = conn
threading.Thread(target=conn.transfer, args=(self, new_packet)).start()
def receive(self, packet: SimPacket):
"""
接收数据
:param packet: 数据包
:return:
"""
print(f"交换机{self.ObjLabel}接受到数据{packet.message}")
self.transmit(packet)
class SimHub(SimBase):
# todo: 集线器类
"""
集线器类
"""
def __init__(self, canvas: Canvas, x, y, id=None, config=None, label=None, *args):
self.ObjID = str(uuid4()) if id is None else id
super().__init__(canvas, x, y, self.ObjID, config, label)
self.ObjType = 4
self.ObjLabel = label if label is not None else self.set_default_name()
self.img = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/集线器.png").resize((60, 60)))
self.img_tm = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/集线器_tm.png").resize((60, 60)))
self.create_img()
def transmit(self, packet: SimPacket):
"""
集线器转发数据包
:return:
"""
for conn in self.connections: # 将数据包往所有接口进行转发
if isinstance(conn, AllSimConnect):
if conn == packet.up_jump:
continue
if conn.ConfigCorrect == 0:
continue
new_packet = SimPacket(packet.source_ip,
packet.source_mac,
packet.destination_ip,
packet.destination_mac,
packet.message)
new_packet.up_jump = conn
threading.Thread(target=conn.transfer, args=(self, new_packet)).start()
def receive(self, packet: SimPacket):
"""
接收数据
:param packet: 数据包
:return:
"""
print(f"集线器-{self.ObjLabel}接受到数据，将进行转发！")
self.transmit(packet)

42
NetworkAnalog/SimObjs.py
Unescape View File

@ -17,7 +17,7 @@ class SimBase():
"""
图标类，所有组件的父类
"""
def __init__(self, canvas: Canvas, x, y, id, config=None, label=None):
def __init__(self, canvas, x, y, id, config=None, label=None):
self.ConfigCorrect = 0 if config is None else config # 是否进行配置
self.ObjType = None # 组件类型 1->主机，2->路由器，3->交换机，4->集线器
self.ObjLabel = ""
@ -73,8 +73,10 @@ class SimBase():
self.canvas.chose_obj = None
dx = event.x - self.start_x
dy = event.y - self.start_y
width = int(self.canvas.cget("width"))
height = int(self.canvas.cget("height"))
# 移动范围限制, 超出移动范围则直接返回
if not (170 <= self.ObjX + dx <= 630 and 60 <= self.ObjY + dy <= 470):
if not (0 + 20 <= self.ObjX + dx <= width * 0.8 - 20 and 20 <= self.ObjY + dy <= height * 0.85 - 20):
return
self.ObjX += dx
self.ObjY += dy
@ -96,7 +98,7 @@ class SimBase():
image=self.img if self.ConfigCorrect == 1 else self.img_tm, anchor="nw",
tags=self.ObjID)
self.canvas.dtag("L", "L")
self.canvas.create_text(self.ObjX, self.ObjY - 40, text=self.ObjLabel, font=("", 12, "bold"),
self.canvas.create_text(self.ObjX, self.ObjY - 40, text=self.ObjLabel, font=("", 16, "bold"),
fill="#7030a0", tags=self.ObjID + "text", anchor="nw")
self.canvas.tag_raise(id)
self.bind_event()
@ -146,7 +148,7 @@ class SimBase():
:return:
"""
self.canvas.delete(self.ObjID + "text")
self.canvas.create_text(self.ObjX, self.ObjY - 40, text=self.ObjLabel, font=("", 12, "bold"),
self.canvas.create_text(self.ObjX, self.ObjY - 40, text=self.ObjLabel, font=("", 16, "bold"),
fill="#7030a0", tags=self.ObjID + "text", anchor="nw")
sql = f"update sim_objs set objLabel='{self.ObjLabel}', ObjX={self.ObjX}," \
f"ObjY={self.ObjY}, ConfigCorrect={self.ConfigCorrect} where ObjID='{self.ObjID}'"
@ -235,7 +237,7 @@ class SimPacket():
class AllSimConnect():
# todo: 连接类
def __init__(self, canvas: Canvas, nodex: SimBase, nodex_ifs, nodey: SimBase, nodey_ifs, config=None):
def __init__(self, canvas, nodex: SimBase, nodex_ifs, nodey: SimBase, nodey_ifs, config=None):
"""
连接对象
:param nodex: 节点
@ -250,7 +252,6 @@ class AllSimConnect():
self.IfsS = nodex_ifs
self.IfsE = nodey_ifs
self.width = 1 if self.ConfigCorrect == 0 else 4 # 线的粗细
self.draw_line()
def is_connected_to(self, other):
"""
@ -371,10 +372,10 @@ class AllSimConnect():
x_E = EX - x
y_E = EY - y # NobjE的连接点坐标
# 显示S接口号
self.canvas.create_text((x_S - 5, y_S - 5), text=IfsS, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
self.canvas.create_text((x_S - 5, y_S - 5), text=IfsS, anchor="nw", font=("幼圆", 16, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
# 显示E接口号
self.canvas.create_text(x_E - 10, y_E - 10, text=IfsE, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
self.canvas.create_text(x_E - 10, y_E - 10, text=IfsE, anchor="nw", font=("幼圆", 16, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
elif SX < EX and SY > EY: # NobjS在NobjE的左下角，NobjS的右上角连接NobjE的左下角
x_S = SX + x
@ -382,10 +383,10 @@ class AllSimConnect():
x_E = EX - x
y_E = EY + y # NobjE的连接点坐标
# 显示S接口号
self.canvas.create_text(x_S + 5, y_S - 10, text=IfsS, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
self.canvas.create_text(x_S + 5, y_S - 10, text=IfsS, anchor="nw", font=("幼圆", 16, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
# 显示E接口号
self.canvas.create_text(x_E - 5, y_E, text=IfsE, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
self.canvas.create_text(x_E - 5, y_E, text=IfsE, anchor="nw", font=("幼圆", 16, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
elif SX > EX and SY < EY: # NobjS在NobjE的右上角，NobjS的左下角连接NobjE的右上角
x_S = SX - x
@ -393,10 +394,10 @@ class AllSimConnect():
x_E = EX + x
y_E = EY - y # NobjE的连接点坐标
# 显示S接口号
self.canvas.create_text(x_S - 5, y_S, text=IfsS, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
self.canvas.create_text(x_S - 5, y_S, text=IfsS, anchor="nw", font=("幼圆", 16, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
# 显示E接口号
self.canvas.create_text(x_E + 5, y_E, text=IfsE, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
self.canvas.create_text(x_E + 5, y_E, text=IfsE, anchor="nw", font=("幼圆", 16, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
elif SX >= EX and SY >= EY: # NobjS在NobjE的右下角，NobjS的左上角连接NobjE的右下角
x_S = SX - x
@ -404,10 +405,10 @@ class AllSimConnect():
x_E = EX + x
y_E = EY + y # NobjE的连接点坐标
# 显示S接口号
self.canvas.create_text(x_S - 5, y_S - 15, text=IfsS, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
self.canvas.create_text(x_S - 5, y_S - 15, text=IfsS, anchor="nw", font=("幼圆", 16, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
# 显示E接口号
self.canvas.create_text(x_E - 5, y_E - 5, text=IfsE, anchor="nw", font=("幼圆", 12, "bold"), fill="red",
self.canvas.create_text(x_E - 5, y_E - 5, text=IfsE, anchor="nw", font=("幼圆", 16, "bold"), fill="red",
tag=self.NobjS.ObjID + self.NobjE.ObjID) # 显示文字
def save(self):
@ -445,7 +446,7 @@ class SimHost(SimBase):
"""
主机类
"""
def __init__(self, canvas: Canvas, x, y, id=None, config=None, label=None):
def __init__(self, canvas, x, y, id=None, config=None, label=None):
self.ObjID = str(uuid4()) if id is None else id
super().__init__(canvas, x, y, self.ObjID, config, label)
self.ObjType = 1
@ -457,7 +458,6 @@ class SimHost(SimBase):
self.img_tm = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/主机_tm.png").resize((60, 60)))
self.set_default_config()
self.create_img()
def create_packet(self, ip, mac, message):
"""
@ -511,7 +511,7 @@ class SimRouter(SimBase):
"""
路由类
"""
def __init__(self, canvas: Canvas, x, y, id=None, config=None, label=None, *args):
def __init__(self, canvas, x, y, id=None, config=None, label=None, *args):
self.ObjID = str(uuid4()) if id is None else id
super().__init__(canvas, x, y, self.ObjID, config, label)
self.ObjType = 2
@ -520,7 +520,6 @@ class SimRouter(SimBase):
Image.open(sys.path[0] + "/../datas/images/路由器.png").resize((60, 60)))
self.img_tm = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/路由器_tm.png").resize((60, 60)))
self.create_img()
self.router_table = {}
self.set_default_router_table()
@ -621,7 +620,7 @@ class SimSwitch(SimBase):
"""
交换机类
"""
def __init__(self, canvas: Canvas, x, y, id=None, config=None, label=None, *args):
def __init__(self, canvas, x, y, id=None, config=None, label=None, *args):
self.ObjID = str(uuid4()) if id is None else id
super().__init__(canvas, x, y, self.ObjID, config, label)
self.ObjType = 3
@ -630,7 +629,6 @@ class SimSwitch(SimBase):
Image.open(sys.path[0] + "/../datas/images/交换机.png").resize((60, 60)))
self.img_tm = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/交换机_tm.png").resize((60, 60)))
self.create_img()
self.mac_table = {}
self.set_default_mac_table()
@ -717,7 +715,7 @@ class SimHub(SimBase):
"""
集线器类
"""
def __init__(self, canvas: Canvas, x, y, id=None, config=None, label=None, *args):
def __init__(self, canvas, x, y, id=None, config=None, label=None, *args):
self.ObjID = str(uuid4()) if id is None else id
super().__init__(canvas, x, y, self.ObjID, config, label)
self.ObjType = 4
@ -726,7 +724,6 @@ class SimHub(SimBase):
Image.open(sys.path[0] + "/../datas/images/集线器.png").resize((60, 60)))
self.img_tm = ImageTk.PhotoImage(
Image.open(sys.path[0] + "/../datas/images/集线器_tm.png").resize((60, 60)))
self.create_img()
def transmit(self, packet: SimPacket):
"""
@ -755,3 +752,4 @@ class SimHub(SimBase):
"""
print(f"集线器-{self.ObjLabel}接受到数据，将进行转发！")
self.transmit(packet)

BIN
NetworkAnalog/__pycache__/NetworkAnalog.cpython-37.pyc
View File

Binary file not shown.

BIN
NetworkAnalog/__pycache__/SimObjs.cpython-37.pyc
View File

Binary file not shown.

BIN
NetworkAnalog/network.db
View File

Binary file not shown.

18
NetworkAnalog/x1_test.py
Unescape View File

@ -0,0 +1,18 @@
from tkinter import Tk
from NetworkAnalog import *
from SimObjs import *
if __name__ == '__main__':
Tk()
AllSimObjs = {}
host1 = SimHost(None, 50, 100)
AllSimObjs[host1.ObjID] = host1
host2 = SimHost(None, 200, 100)
AllSimObjs[host2.ObjID] = host2
router = SimRouter(None, 100, 100)
AllSimObjs[router.ObjID] = router
host1_router_connect = AllSimConnect(None, host1, 1, router, 1)
host2_router_connect = AllSimConnect(None, host2, 1, router, 2)
Write Preview
Loading…
Cancel
Save