Refactor and optimize code structure with base widget and utility functions

- Create CustomWidgetBase for shared context menu and shortcut functionality
- Add utility functions for common tasks like text appending and file selection
- Simplify command processing with a generic send_command_and_process_response function
- Remove redundant comments and unused imports
- Improve code organization and readability
- Enhance modularity of custom widgets and utility methods
This commit is contained in:
2025-02-16 23:45:19 +03:00
parent a140b7d8a0
commit 7ebeb52808

View File

@@ -5,11 +5,6 @@
# Это программа для копирования конфигураций на коммутаторы
# ------------------------------------------------------------
# import argparse Использовался для получения аргументов из командной строки (не используется)
# import platform Использовался для получения списка сетевых адаптеров (не используется)
# import subprocess Использовался для получения списка сетевых адаптеров (не используется)
# import socket не используется
import json
import logging
import os
@@ -39,7 +34,6 @@ import serial.tools.list_ports
from serial.serialutil import SerialException
from about_window import AboutWindow
from TFTPServer import TFTPServer
# from TFTPServer import TFTPServerThread
import socket
from update_checker import UpdateChecker
@@ -56,12 +50,8 @@ os.makedirs("docs", exist_ok=True)
# Файл настроек находится в папке Settings
SETTINGS_FILE = os.path.join("Settings", "settings.json")
# ==========================
# Функции работы с настройками и логированием
# ==========================
# Настройка логирования с использованием RotatingFileHandler.
def setup_logging():
"""Настройка логирования с использованием RotatingFileHandler."""
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
log_path = os.path.join("Logs", "app.log")
@@ -72,8 +62,8 @@ def setup_logging():
handler.setFormatter(formatter)
logger.addHandler(handler)
# Загрузка настроек из JSON-файла или создание настроек по умолчанию.
def settings_load():
"""Загрузка настроек из JSON-файла или создание настроек по умолчанию."""
default_settings = {
"port": None, # Порт для подключения
"baudrate": 9600, # Скорость передачи данных
@@ -126,8 +116,8 @@ def settings_load():
logging.error(f"Ошибка загрузки файла настроек: {e}", exc_info=True)
return default_settings
# Сохранение настроек в JSON-файл
def settings_save(settings):
"""Сохранение настроек в JSON-файл."""
try:
with open(SETTINGS_FILE, "w", encoding="utf-8") as f:
json.dump(settings, f, indent=4, ensure_ascii=False)
@@ -135,18 +125,15 @@ def settings_save(settings):
except Exception as e:
logging.error(f"Ошибка при сохранении настроек: {e}", exc_info=True)
# Получение списка доступных последовательных портов
def list_serial_ports():
"""Получение списка доступных последовательных портов."""
ports = serial.tools.list_ports.comports()
logging.debug(f"Найдено {len(ports)} серийных портов.")
return [port.device for port in ports]
# ==========================
# Функции работы с сетевыми адаптерами (не используются)
# ==========================
# Получение списка IP-адресов из сетевых адаптеров
def get_network_adapters():
"""Получение списка сетевых адаптеров и их IP-адресов."""
adapters = []
try:
# Получаем имя хоста
@@ -177,12 +164,8 @@ def get_network_adapters():
return adapters
# ==========================
# Функции работы с COM-соединением
# ==========================
# Создание соединения с устройством через последовательный порт
def create_connection(settings):
"""Создание соединения с устройством через последовательный порт."""
try:
conn = serial.Serial(settings["port"], settings["baudrate"], timeout=1)
logging.info(f"Соединение установлено с {settings['port']} на {settings['baudrate']}.")
@@ -194,12 +177,8 @@ def create_connection(settings):
logging.error(f"Неизвестная ошибка подключения: {e}", exc_info=True)
return None
# Проверка наличия логина и пароля в настройках и отправка их на устройство
def send_login_password(serial_connection, login=None, password=None, is_gui=False):
"""Отправка логина и пароля на устройство."""
if not login:
if is_gui:
login = simpledialog.askstring("Login", "Введите логин:")
@@ -216,14 +195,8 @@ def send_login_password(serial_connection, login=None, password=None, is_gui=Fal
else:
password = getpass("Введите пароль: ")
# Чтение ответа от устройства с учётом таймаута.
def read_response(serial_connection, timeout, login=None, password=None, is_gui=False):
"""
Чтение ответа от устройства с учётом таймаута.
Если в последней строке появляется запрос логина или пароля, данные отправляются автоматически.
"""
response = b""
end_time = time.time() + timeout
decoded = ""
@@ -260,8 +233,8 @@ def read_response(serial_connection, timeout, login=None, password=None, is_gui=
time.sleep(0.1)
return decoded
# Генерация блоков команд для блочного копирования
def generate_command_blocks(lines, block_size):
"""Генерация блоков команд для блочного копирования."""
blocks = []
current_block = []
for line in lines:
@@ -287,6 +260,7 @@ def generate_command_blocks(lines, block_size):
blocks.append("\n".join(current_block))
return blocks
# Выполнение команд из файла конфигурации
def execute_commands_from_file(
serial_connection,
filename,
@@ -298,12 +272,6 @@ def execute_commands_from_file(
password=None,
is_gui=False,
):
"""
Выполнение команд из файла конфигурации.
Если передан log_callback, вывод будет отображаться в GUI.
Теперь при обнаружении ошибки (например, если в ответе присутствует символ '^')
команда будет отправляться повторно.
"""
try:
with open(filename, "r", encoding="utf-8") as file:
lines = [line for line in file if line.strip()]
@@ -442,18 +410,16 @@ def execute_commands_from_file(
log_callback(msg)
logging.error(f"Ошибка при выполнении команд из файла: {e}", exc_info=True)
# ==========================
# Улучшенные текстовые виджеты
# ==========================
class CustomText(tk.Text):
"""Улучшенный текстовый виджет с расширенной функциональностью копирования/вставки"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Базовый класс для кастомных виджетов с поддержкой контекстного меню и горячих клавиш
class CustomWidgetBase:
def __init__(self):
self.create_context_menu()
self.bind_shortcuts()
# Создание контекстного меню
def create_context_menu(self):
"""Создание контекстного меню"""
self.context_menu = tk.Menu(self, tearoff=0)
self.context_menu.add_command(label="Вырезать", command=self.cut)
self.context_menu.add_command(label="Копировать", command=self.copy)
@@ -462,90 +428,63 @@ class CustomText(tk.Text):
self.context_menu.add_command(label="Выделить всё", command=self.select_all)
self.bind("<Button-3>", self.show_context_menu)
def show_context_menu(self, event):
self.context_menu.post(event.x_root, event.y_root)
# Привязка горячих клавиш
def bind_shortcuts(self):
"""Привязка горячих клавиш"""
# Стандартные сочетания
self.bind("<Control-x>", lambda e: self.event_generate("<<Cut>>"))
self.bind("<Control-c>", lambda e: self.event_generate("<<Copy>>"))
self.bind("<Control-v>", lambda e: self.event_generate("<<Paste>>"))
self.bind("<Control-a>", self.select_all)
# Shift+Insert для вставки
# Дополнительные сочетания
self.bind("<Shift-Insert>", lambda e: self.event_generate("<<Paste>>"))
# Ctrl+Insert для копирования
self.bind("<Control-Insert>", lambda e: self.event_generate("<<Copy>>"))
# Shift+Delete для вырезания
self.bind("<Shift-Delete>", lambda e: self.event_generate("<<Cut>>"))
# Отображение контекстного меню
def show_context_menu(self, event):
"""Отображение контекстного меню"""
self.context_menu.post(event.x_root, event.y_root)
# Вырезание текста
def cut(self):
self.event_generate("<<Cut>>")
# Копирование текста
def copy(self):
self.event_generate("<<Copy>>")
# Вставка текста
def paste(self):
self.event_generate("<<Paste>>")
# Класс для текстового поля с поддержкой контекстного меню и горячих клавиш
class CustomText(tk.Text, CustomWidgetBase):
def __init__(self, *args, **kwargs):
tk.Text.__init__(self, *args, **kwargs)
CustomWidgetBase.__init__(self)
def select_all(self, event=None):
"""Выделение всего текста в Text виджете"""
self.tag_add(tk.SEL, "1.0", tk.END)
self.mark_set(tk.INSERT, "1.0")
self.see(tk.INSERT)
return "break"
class CustomEntry(ttk.Entry):
"""Улучшенное поле ввода с расширенной функциональностью копирования/вставки"""
# Класс для поля ввода с поддержкой контекстного меню и горячих клавиш
class CustomEntry(ttk.Entry, CustomWidgetBase):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.create_context_menu()
self.bind_shortcuts()
def create_context_menu(self):
self.context_menu = tk.Menu(self, tearoff=0)
self.context_menu.add_command(label="Вырезать", command=self.cut)
self.context_menu.add_command(label="Копировать", command=self.copy)
self.context_menu.add_command(label="Вставить", command=self.paste)
self.context_menu.add_separator()
self.context_menu.add_command(label="Выделить всё", command=self.select_all)
self.bind("<Button-3>", self.show_context_menu)
def show_context_menu(self, event):
self.context_menu.post(event.x_root, event.y_root)
def bind_shortcuts(self):
# Стандартные сочетания
self.bind("<Control-x>", lambda e: self.event_generate("<<Cut>>"))
self.bind("<Control-c>", lambda e: self.event_generate("<<Copy>>"))
self.bind("<Control-v>", lambda e: self.event_generate("<<Paste>>"))
self.bind("<Control-a>", self.select_all)
# Shift+Insert для вставки
self.bind("<Shift-Insert>", lambda e: self.event_generate("<<Paste>>"))
# Ctrl+Insert для копирования
self.bind("<Control-Insert>", lambda e: self.event_generate("<<Copy>>"))
# Shift+Delete для вырезания
self.bind("<Shift-Delete>", lambda e: self.event_generate("<<Cut>>"))
def cut(self):
self.event_generate("<<Cut>>")
def copy(self):
self.event_generate("<<Copy>>")
def paste(self):
self.event_generate("<<Paste>>")
ttk.Entry.__init__(self, *args, **kwargs)
CustomWidgetBase.__init__(self)
def select_all(self, event=None):
"""Выделение всего текста в Entry виджете"""
self.select_range(0, tk.END)
return "break"
# Класс для окна настроек
class SettingsWindow(tk.Toplevel):
def __init__(self, parent, settings, callback=None):
super().__init__(parent)
@@ -612,6 +551,7 @@ class SettingsWindow(tk.Toplevel):
# Центрируем окно
self.center_window()
# Центрирование окна
def center_window(self):
self.update_idletasks()
width = self.winfo_width()
@@ -620,12 +560,14 @@ class SettingsWindow(tk.Toplevel):
y = (self.winfo_screenheight() // 2) - (height // 2)
self.geometry(f"{width}x{height}+{x}+{y}")
# Обновление списка доступных последовательных портов
def update_ports(self):
ports = list_serial_ports()
self.port_combo["values"] = ports
if ports and not self.port_var.get():
self.port_var.set(ports[0])
# Сохранение настроек
def save_settings(self):
try:
self.settings.update({
@@ -644,6 +586,78 @@ class SettingsWindow(tk.Toplevel):
except ValueError as e:
messagebox.showerror("Ошибка", "Проверьте правильность введенных значений")
# Общая функция для добавления текста в текстовое поле
def append_text_to_widget(widget, text):
# Проверяем, заканчивается ли текст символом новой строки
if not text.endswith('\n'):
text += '\n'
widget.insert(END, text)
widget.see(END)
# Общая функция для выбора файла конфигурации
def select_config_file(self, var, save_to_settings=False):
filename = filedialog.askopenfilename(
title="Выберите файл конфигурации",
filetypes=[("Text files", "*.txt")]
)
if filename:
var.set(filename)
if save_to_settings:
self.settings["config_file"] = filename
settings_save(self.settings)
# Общая функция для отправки команды и обработки ответа
def send_command_and_process_response(
serial_connection,
cmd,
timeout,
max_attempts=3,
log_callback=None,
login=None,
password=None,
is_gui=False
):
attempt = 0
while attempt < max_attempts:
msg = f"\nОтправка команды: {cmd} (Попытка {attempt+1} из {max_attempts})\n"
if log_callback:
log_callback(msg)
serial_connection.write((cmd + "\n").encode())
logging.info(f"Отправлена команда: {cmd} (Попытка {attempt+1})")
response = read_response(serial_connection, timeout, login=login, password=password, is_gui=is_gui)
if response:
if '^' in response:
msg = (
f"[ERROR] Обнаружена ошибка при выполнении команды: {cmd}\n"
f"Ответ устройства:\n{response}\n"
f"Повторная отправка команды...\n"
)
if log_callback:
log_callback(msg)
logging.warning(f"Ошибка в команде: {cmd}. Повторная отправка.")
attempt += 1
time.sleep(1)
continue
else:
msg = f"Ответ устройства:\n{response}\n"
if log_callback:
log_callback(msg)
logging.info(f"Ответ устройства:\n{response}")
return True, response
else:
msg = f"Ответ не получен для команды: {cmd}\n"
if log_callback:
log_callback(msg)
logging.warning(f"Нет ответа для команды: {cmd}")
return False, None
msg = f"[ERROR] Команда не выполнена корректно после {max_attempts} попыток: {cmd}\n"
if log_callback:
log_callback(msg)
logging.error(msg)
return False, None
# Основной класс для графического интерфейса
class SerialAppGUI(tk.Tk):
def __init__(self, settings):
super().__init__()
@@ -675,6 +689,7 @@ class SerialAppGUI(tk.Tk):
# Проверка первого запуска
self.check_first_run()
# Создание меню
def create_menu(self):
menubar = tk.Menu(self)
self.config(menu=menubar)
@@ -693,8 +708,8 @@ class SerialAppGUI(tk.Tk):
help_menu.add_separator()
help_menu.add_command(label="О программе", command=self.open_about)
# Проверка наличия обновлений
def check_for_updates(self):
"""Проверка наличия обновлений"""
def on_update_check(update_available, error):
if error:
messagebox.showerror(
@@ -725,18 +740,18 @@ class SerialAppGUI(tk.Tk):
self.update_checker.check_updates(callback=on_update_check)
# Обработчик изменения настроек
def on_settings_changed(self):
"""Обработчик изменения настроек"""
self.settings = settings_load()
# Открытие окна настроек
def open_settings(self):
"""Открытие окна настроек"""
settings_window = SettingsWindow(self, self.settings, self.on_settings_changed)
settings_window.transient(self)
settings_window.grab_set()
# Проверка первого запуска программы
def check_first_run(self):
"""Проверка первого запуска программы"""
show_settings = False
# Проверяем существование файла настроек
@@ -765,6 +780,7 @@ class SerialAppGUI(tk.Tk):
if response:
self.open_settings()
# Создание вкладок
def create_tabs(self):
self.notebook = ttk.Notebook(self)
self.notebook.pack(fill=BOTH, expand=True, padx=5, pady=5)
@@ -785,7 +801,7 @@ class SerialAppGUI(tk.Tk):
self.create_config_editor_tab(config_editor_frame)
self.create_tftp_tab(tftp_frame)
# -------------- Вкладка "Интерактивный режим" --------------
# Создание вкладки "Интерактивный режим"
def create_interactive_tab(self, frame):
control_frame = ttk.Frame(frame)
control_frame.pack(fill=X, pady=5)
@@ -802,6 +818,7 @@ class SerialAppGUI(tk.Tk):
self.command_entry.pack(side=LEFT, padx=5)
ttk.Button(input_frame, text="Отправить", command=self.send_command).pack(side=LEFT, padx=5)
# Подключение к устройству
def connect_device(self):
if self.connection:
messagebox.showinfo("Информация", "Уже подключено.")
@@ -815,6 +832,7 @@ class SerialAppGUI(tk.Tk):
else:
self.append_interactive_text("[ERROR] Не удалось установить соединение.\n")
# Отключение от устройства
def disconnect_device(self):
if self.connection:
try:
@@ -826,6 +844,7 @@ class SerialAppGUI(tk.Tk):
else:
messagebox.showinfo("Информация", "Соединение не установлено.")
# Отправка команды
def send_command(self):
if not self.connection:
messagebox.showerror("Ошибка", "Сначала установите соединение!")
@@ -836,42 +855,19 @@ class SerialAppGUI(tk.Tk):
self.append_interactive_text(f"[INFO] Отправка команды: {cmd}\n")
threading.Thread(target=self.process_command, args=(cmd,), daemon=True).start()
# Обработка команды
def process_command(self, cmd):
try:
max_attempts = 3
attempt = 0
while attempt < max_attempts:
self.connection.write((cmd + "\n").encode())
logging.info(f"Отправлена команда: {cmd} (Попытка {attempt+1})")
response = read_response(
self.connection,
self.settings.get("timeout", 10),
login=self.settings.get("login"),
password=self.settings.get("password"),
is_gui=True,
)
if response:
if '^' in response:
self.append_interactive_text(
f"[ERROR] Обнаружена ошибка при выполнении команды: {cmd}\n"
f"Ответ устройства:\n{response}\n"
f"Повторная отправка команды...\n"
)
logging.warning(f"Ошибка в команде: {cmd}. Попытка повторной отправки.")
attempt += 1
time.sleep(1)
continue
else:
self.append_interactive_text(f"[INFO] Ответ устройства:\n{response}\n")
logging.info(f"Получен ответ:\n{response}")
break
else:
self.append_interactive_text("[WARN] Ответ не получен.\n")
logging.warning("Нет ответа от устройства в течение таймаута.")
break
if attempt == max_attempts:
self.append_interactive_text(f"[ERROR] Команда не выполнена корректно после {max_attempts} попыток: {cmd}\n")
logging.error(f"Команда не выполнена корректно после {max_attempts} попыток: {cmd}")
success, response = send_command_and_process_response(
self.connection,
cmd,
self.settings.get("timeout", 10),
max_attempts=3,
log_callback=self.append_interactive_text,
login=self.settings.get("login"),
password=self.settings.get("password"),
is_gui=True
)
except SerialException as e:
self.append_interactive_text(f"[ERROR] Ошибка при отправке команды: {e}\n")
logging.error(f"Ошибка отправки команды: {e}", exc_info=True)
@@ -879,11 +875,11 @@ class SerialAppGUI(tk.Tk):
self.append_interactive_text(f"[ERROR] Неизвестная ошибка: {e}\n")
logging.error(f"Неизвестная ошибка: {e}", exc_info=True)
# Добавление текста в текстовое поле
def append_interactive_text(self, text):
self.interactive_text.insert(END, text)
self.interactive_text.see(END)
append_text_to_widget(self.interactive_text, text)
# -------------- Вкладка "Выполнить команды из файла" --------------
# Создание вкладки "Выполнить команды из файла"
def create_file_exec_tab(self, frame):
file_frame = ttk.Frame(frame)
file_frame.pack(fill=X, pady=5)
@@ -895,11 +891,11 @@ class SerialAppGUI(tk.Tk):
self.file_exec_text = CustomText(frame, wrap="word", height=15)
self.file_exec_text.pack(fill=BOTH, expand=True, padx=5, pady=5)
# Выбор файла конфигурации для выполнения команд
def select_config_file_fileexec(self):
filename = filedialog.askopenfilename(title="Выберите файл конфигурации", filetypes=[("Text files", "*.txt")])
if filename:
self.file_exec_var.set(filename)
select_config_file(self, self.file_exec_var)
# Выполнение команд из файла
def execute_file_commands(self):
if not self.settings.get("port"):
messagebox.showerror("Ошибка", "COM-порт не выбран!")
@@ -929,10 +925,9 @@ class SerialAppGUI(tk.Tk):
).start()
def append_file_exec_text(self, text):
self.file_exec_text.insert(END, text)
self.file_exec_text.see(END)
append_text_to_widget(self.file_exec_text, text)
# -------------- Вкладка "Редактор конфигурационного файла" --------------
# Создание вкладки "Редактор конфигурационного файла"
def create_config_editor_tab(self, frame):
top_frame = ttk.Frame(frame)
top_frame.pack(fill=X, pady=5)
@@ -945,13 +940,11 @@ class SerialAppGUI(tk.Tk):
self.config_editor_text = CustomText(frame, wrap="word")
self.config_editor_text.pack(fill=BOTH, expand=True, padx=5, pady=5)
# Выбор файла конфигурации для редактирования
def select_config_file_editor(self):
filename = filedialog.askopenfilename(title="Выберите файл конфигурации", filetypes=[("Text files", "*.txt")])
if filename:
self.editor_file_var.set(filename)
self.settings["config_file"] = filename
settings_save(self.settings)
select_config_file(self, self.editor_file_var, save_to_settings=True)
# Загрузка файла конфигурации
def load_config_file(self):
filename = self.editor_file_var.get()
if not filename or not os.path.exists(filename):
@@ -967,6 +960,7 @@ class SerialAppGUI(tk.Tk):
logging.error(f"Ошибка загрузки файла: {e}", exc_info=True)
messagebox.showerror("Ошибка", f"Не удалось загрузить файл:\n{e}")
# Сохранение файла конфигурации
def save_config_file(self):
filename = self.editor_file_var.get()
if not filename:
@@ -981,13 +975,14 @@ class SerialAppGUI(tk.Tk):
logging.error(f"Ошибка сохранения файла: {e}", exc_info=True)
messagebox.showerror("Ошибка", f"Не удалось сохранить файл:\n{e}")
# Открытие окна "О программе"
def open_about(self):
about_window = AboutWindow(self)
about_window.transient(self)
about_window.grab_set()
# Создание вкладки TFTP сервера
def create_tftp_tab(self, frame):
"""Создание вкладки TFTP сервера."""
# Создаем фрейм для управления TFTP сервером
tftp_frame = ttk.Frame(frame)
tftp_frame.pack(fill=BOTH, expand=True, padx=5, pady=5)
@@ -1075,8 +1070,8 @@ class SerialAppGUI(tk.Tk):
self.tftp_server = None
self.tftp_server_thread = None
# Запуск TFTP сервера
def start_tftp_server(self):
"""Запуск TFTP сервера."""
try:
# Получаем выбранный IP-адрес
ip = self.tftp_ip_var.get()
@@ -1138,15 +1133,15 @@ class SerialAppGUI(tk.Tk):
self.append_tftp_log(f"[ERROR] Ошибка запуска сервера: {str(e)}")
messagebox.showerror("Ошибка", f"Не удалось запустить TFTP сервер: {str(e)}")
# Запуск TFTP сервера в отдельном потоке
def run_tftp_server(self, ip, port):
"""Запуск TFTP сервера в отдельном потоке."""
try:
self.tftp_server.start_server(ip, port)
except Exception as e:
self.append_tftp_log(f"[ERROR] Ошибка работы сервера: {str(e)}")
# Остановка TFTP сервера
def stop_tftp_server(self):
"""Остановка TFTP сервера."""
if self.tftp_server:
try:
# Отключаем кнопки на время остановки сервера
@@ -1190,13 +1185,11 @@ class SerialAppGUI(tk.Tk):
self.start_tftp_button.config(state="disabled")
self.stop_tftp_button.config(state="normal")
def append_tftp_log(self, message):
"""Добавление сообщения в лог TFTP сервера."""
self.tftp_log_text.insert(END, message + "\n")
self.tftp_log_text.see(END)
def append_tftp_log(self, text):
append_text_to_widget(self.tftp_log_text, text)
# Обновление информации об активных передачах
def update_transfers_info(self):
"""Обновление информации об активных передачах."""
if not self.tftp_server:
return
@@ -1237,15 +1230,15 @@ class SerialAppGUI(tk.Tk):
f"{elapsed_time:.1f}с"
))
# Периодическое обновление информации о передачах
def update_transfers_periodically(self):
"""Периодическое обновление информации о передачах."""
if self.tftp_server and self.tftp_server.running:
self.update_transfers_info()
# Планируем следующее обновление через 1 секунду
self.after(1000, self.update_transfers_periodically)
# Обновление списка сетевых адаптеров
def update_network_adapters(self):
"""Обновление списка сетевых адаптеров."""
adapters = get_network_adapters()
self.tftp_ip_combo["values"] = adapters
if not self.tftp_ip_var.get() in adapters: