import sys import os import socket import ipaddress import platform from concurrent.futures import ThreadPoolExecutor, as_completed from PyQt6.QtWidgets import ( QApplication, QWidget, QPushButton, QVBoxLayout, QHBoxLayout, QFileDialog, QTableWidget, QTableWidgetItem, QLabel, QProgressBar, QMessageBox, QGroupBox, QHeaderView, QLineEdit, QFrame, QCheckBox, QSpinBox, QScrollArea, QSizePolicy, QComboBox ) from PyQt6.QtCore import ( Qt, QThread, pyqtSignal, QPropertyAnimation, QAbstractAnimation, QParallelAnimationGroup, QRect ) from PyQt6.QtGui import QFont, QColor, QIcon, QAction import datetime from core.scanner import scan_network from core.workers import ScanThread from core.flash_new_worker import NewFlashThread from core.flash_update_worker import UpdateFlashThread from core.auto_flash_worker import AutoFlashWorker from utils.network import _resolve_hostname, get_default_network from utils.system import resource_path, get_machine_info, get_version from ui.components import CollapsibleGroupBox from ui.styles import STYLE, AUTO_STYLE class App(QWidget): def __init__(self): super().__init__() self.setWindowTitle(f"Mira Firmware Loader v{get_version()}") self.setWindowIcon(QIcon(resource_path("icon.ico"))) # Data state self.local_ip = "" self.gateway_ip = "" self.firmware = None self.all_devices = [] # raw list from scanner self.devices = [] # filtered list for table self.flashed_macs = {} # MAC -> (ip, mac, result, timestamp) self.scan_thread = None info = get_machine_info() self.local_ip = info["ip"] self.gateway_ip = self._guess_gateway(self.local_ip) layout = QVBoxLayout() layout.setSpacing(3) layout.setContentsMargins(8, 6, 8, 6) # ── Title (single line) ── title_row = QHBoxLayout() title_row.setSpacing(0) title = QLabel("⚡ Mira Firmware Loader") title.setObjectName("title") title_row.addWidget(title) title_row.addStretch() copyright_label = QLabel(f"© {datetime.datetime.now().year} Smatec — R&D Software Team. v{get_version()}") copyright_label.setStyleSheet( "color: #9399b2; font-size: 15px; font-weight: 500;" ) title_row.addWidget(copyright_label) layout.addLayout(title_row) # ── Machine Info (single compact row) ── info_group = CollapsibleGroupBox("🖥 Machine Info") info_row = QHBoxLayout() info_row.setSpacing(6) info_row.setContentsMargins(0, 0, 0, 0) for label, value in [("Host:", info["hostname"]), ("IP:", info["ip"]), ("OS:", info["os"]), ("MAC:", info["mac"])]: lbl = QLabel(label) lbl.setObjectName("info") info_row.addWidget(lbl) val = QLabel(value) val.setStyleSheet("color: #cdd6f4; font-weight: bold; font-size: 11px;") info_row.addWidget(val) if label != "MAC:": sep = QLabel("│") sep.setStyleSheet("color: #3d4a6b;") info_row.addWidget(sep) info_row.addStretch() info_group.set_content_layout(info_row) layout.addWidget(info_group) # ── Firmware + Network Scan (combined compact row) ── fw_scan_group = CollapsibleGroupBox("📦 FW & 📡 Scan") fw_scan_layout = QVBoxLayout() fw_scan_layout.setSpacing(4) fw_scan_layout.setContentsMargins(0, 0, 0, 0) # Row 1: FW selection + Scan top_row = QHBoxLayout() top_row.setSpacing(8) fw_lbl = QLabel("FW:") fw_lbl.setStyleSheet("font-weight: bold; font-size: 12px;") top_row.addWidget(fw_lbl) self.fw_label = QLabel("No firmware selected") self.fw_label.setObjectName("info") self.fw_label.setStyleSheet("font-size: 11px; color: #94a3b8;") top_row.addWidget(self.fw_label) btn_fw = QPushButton("📁") btn_fw.setFixedSize(32, 26) btn_fw.setToolTip("Select firmware file") btn_fw.clicked.connect(self.select_fw) top_row.addWidget(btn_fw) sep = QLabel("│") sep.setStyleSheet("color: #3d4a6b; font-size: 14px;") top_row.addWidget(sep) net_lbl = QLabel("Network:") net_lbl.setStyleSheet("font-weight: bold; font-size: 12px;") top_row.addWidget(net_lbl) self.net_input = QLineEdit(get_default_network(self.local_ip)) self.net_input.setPlaceholderText("e.g. 192.168.4.0/24") self.net_input.setMaximumWidth(170) self.net_input.setFixedHeight(26) self.net_input.setStyleSheet("QLineEdit { font-size: 12px; padding: 2px 8px; }") top_row.addWidget(self.net_input) self.btn_scan = QPushButton("🔍 Scan LAN") self.btn_scan.setObjectName("scan") self.btn_scan.clicked.connect(self.scan) self.btn_scan.setFixedHeight(26) self.btn_scan.setFixedWidth(100) top_row.addWidget(self.btn_scan) fw_scan_layout.addLayout(top_row) # Scan progress + status (hidden by default) self.scan_progress_bar = QProgressBar() self.scan_progress_bar.setObjectName("scan_bar") self.scan_progress_bar.setRange(0, 0) self.scan_progress_bar.setFormat("") self.scan_progress_bar.setFixedHeight(14) self.scan_progress_bar.setVisible(False) fw_scan_layout.addWidget(self.scan_progress_bar) self.scan_status = QLabel("") self.scan_status.setObjectName("info") self.scan_status.setStyleSheet("font-size: 10px;") fw_scan_layout.addWidget(self.scan_status) fw_scan_group.set_content_layout(fw_scan_layout) layout.addWidget(fw_scan_group) # ── Device Table (MAIN area — maximized) ── dev_group = QGroupBox("📋 Devices Found") dev_layout = QVBoxLayout() dev_layout.setSpacing(2) dev_layout.setContentsMargins(4, 12, 4, 4) self.table = QTableWidget() self.table.setColumnCount(4) self.table.setHorizontalHeaderLabels(["", "IP", "MAC", "Status"]) self.table.setAlternatingRowColors(True) header = self.table.horizontalHeader() header.setSectionResizeMode(0, QHeaderView.ResizeMode.Fixed) header.resizeSection(0, 32) header.setSectionResizeMode(1, QHeaderView.ResizeMode.Interactive) header.resizeSection(1, 115) header.setSectionResizeMode(2, QHeaderView.ResizeMode.Interactive) header.resizeSection(2, 135) header.setSectionResizeMode(3, QHeaderView.ResizeMode.Stretch) self.table.verticalHeader().setVisible(False) self.table.setSelectionBehavior( QTableWidget.SelectionBehavior.SelectRows ) dev_layout.addWidget(self.table) filter_row = QHBoxLayout() filter_row.setSpacing(4) self.device_count_label = QLabel("Total: 0 devices") self.device_count_label.setObjectName("info") filter_row.addWidget(self.device_count_label) filter_row.addStretch() btn_history = QPushButton("📋 Lịch sử nạp") btn_history.setFixedHeight(24) btn_history.clicked.connect(self._show_history) btn_history.setStyleSheet("background-color: #313244; color: #cdd6f4; font-size: 11px; padding: 2px 8px;") filter_row.addWidget(btn_history) btn_select_all = QPushButton("☑ All") btn_select_all.setFixedHeight(24) btn_select_all.setStyleSheet("font-size: 11px; padding: 2px 8px;") btn_select_all.clicked.connect(self._select_all_devices) filter_row.addWidget(btn_select_all) btn_deselect_all = QPushButton("☐ None") btn_deselect_all.setFixedHeight(24) btn_deselect_all.setStyleSheet("font-size: 11px; padding: 2px 8px;") btn_deselect_all.clicked.connect(self._deselect_all_devices) filter_row.addWidget(btn_deselect_all) self.show_all_cb = QCheckBox("Show all") self.show_all_cb.setChecked(False) self.show_all_cb.setToolTip("Include gateway & self") self.show_all_cb.stateChanged.connect(self._refresh_table) filter_row.addWidget(self.show_all_cb) dev_layout.addLayout(filter_row) dev_group.setLayout(dev_layout) layout.addWidget(dev_group, stretch=1) # ── Flash Controls (collapsible, compact) ── flash_group = CollapsibleGroupBox("🚀 Flash Controls") flash_layout = QVBoxLayout() flash_layout.setSpacing(6) flash_layout.setContentsMargins(8, 4, 8, 6) self.progress = QProgressBar() self.progress.setFormat("%v / %m devices (%p%)") self.progress.setFixedHeight(18) flash_layout.addWidget(self.progress) # Row 1: Mode + Method mode_method_row = QHBoxLayout() mode_method_row.setSpacing(10) mode_lbl = QLabel("Mode:") mode_lbl.setStyleSheet("font-size: 12px; font-weight: bold;") mode_method_row.addWidget(mode_lbl) self.mode_combo = QComboBox() self.mode_combo.addItem("⚡ New Flash (Factory Reset)", "new") self.mode_combo.addItem("🔄 Update FW (Pre-installed)", "update") self.mode_combo.setMinimumWidth(230) self.mode_combo.setFixedHeight(28) self.mode_combo.setStyleSheet(""" QComboBox { background-color: #2d3352; border: 1px solid #3d4a6b; border-radius: 4px; padding: 2px 8px; color: #ffffff; font-size: 12px; font-weight: bold; } """) self.mode_combo.currentIndexChanged.connect(self._on_mode_changed) mode_method_row.addWidget(self.mode_combo) sep_m = QLabel("│") sep_m.setStyleSheet("color: #3d4a6b;") mode_method_row.addWidget(sep_m) # Container cho method (ẩn khi Update mode) self.method_container = QWidget() mc_layout = QHBoxLayout(self.method_container) mc_layout.setContentsMargins(0, 0, 0, 0) mc_layout.setSpacing(6) meth_lbl = QLabel("Method:") meth_lbl.setStyleSheet("font-size: 12px; font-weight: bold;") mc_layout.addWidget(meth_lbl) self.method_combo = QComboBox() self.method_combo.addItem("🌐 API (LuCI)", "api") self.method_combo.addItem("💻 SSH", "ssh") self.method_combo.setMinimumWidth(140) self.method_combo.setFixedHeight(28) self.method_combo.setStyleSheet(""" QComboBox { background-color: #1e1e2e; border: 1px solid #3d4a6b; border-radius: 4px; padding: 2px 8px; color: #ffffff; font-size: 12px; font-weight: bold; } """) self.method_combo.currentIndexChanged.connect(self._on_method_changed) mc_layout.addWidget(self.method_combo) mode_method_row.addWidget(self.method_container) # Warning label for Update Mode (inline) self.update_warning_lbl = QLabel("⚠️ Update: SSH only, target 192.168.11.102") self.update_warning_lbl.setStyleSheet("color: #f38ba8; font-size: 11px; font-weight: bold;") self.update_warning_lbl.setVisible(False) mode_method_row.addWidget(self.update_warning_lbl) mode_method_row.addStretch() flash_layout.addLayout(mode_method_row) # SSH credentials (hidden by default) self.ssh_creds_widget = QWidget() self.ssh_creds_widget.setStyleSheet("background-color: #11121d; border-radius: 6px; border: 1px dashed #3d4a6b;") ssh_creds_layout = QHBoxLayout(self.ssh_creds_widget) ssh_creds_layout.setContentsMargins(10, 6, 10, 6) ssh_creds_layout.setSpacing(8) ssh_lbl1 = QLabel("User:") ssh_lbl1.setStyleSheet("font-size: 11px; font-weight: bold;") ssh_creds_layout.addWidget(ssh_lbl1) str_qlineedit = "QLineEdit { background-color: #1a1b2e; font-size: 12px; padding: 3px; border: 1px solid #3d4a6b; color: #ffffff; }" self.ssh_user_input = QLineEdit("root") self.ssh_user_input.setFixedWidth(80) self.ssh_user_input.setStyleSheet(str_qlineedit) ssh_creds_layout.addWidget(self.ssh_user_input) ssh_lbl2 = QLabel("Pass:") ssh_lbl2.setStyleSheet("font-size: 11px; font-weight: bold;") ssh_creds_layout.addWidget(ssh_lbl2) self.ssh_pass_input = QLineEdit("admin123a") self.ssh_pass_input.setEchoMode(QLineEdit.EchoMode.Password) self.ssh_pass_input.setFixedWidth(100) self.ssh_pass_input.setStyleSheet(str_qlineedit) ssh_creds_layout.addWidget(self.ssh_pass_input) self.set_passwd_cb = QCheckBox("Set passwd") self.set_passwd_cb.setChecked(True) self.set_passwd_cb.setStyleSheet("color: #94a3b8; font-size: 11px;") self.set_passwd_cb.setToolTip("Set password before flash (passwd → admin123a)") ssh_creds_layout.addWidget(self.set_passwd_cb) ssh_creds_layout.addStretch() self.ssh_creds_widget.setVisible(False) flash_layout.addWidget(self.ssh_creds_widget) # Row 2: Concurrent + Flash button action_row = QHBoxLayout() action_row.setSpacing(10) par_lbl = QLabel("Concurrent:") par_lbl.setStyleSheet("font-size: 12px; font-weight: bold;") action_row.addWidget(par_lbl) self.parallel_spin = QSpinBox() self.parallel_spin.setRange(0, 100) self.parallel_spin.setValue(10) self.parallel_spin.setSpecialValueText("∞") self.parallel_spin.setToolTip("0 = unlimited (all at once)") self.parallel_spin.setFixedWidth(65) self.parallel_spin.setFixedHeight(28) self.parallel_spin.setStyleSheet("QSpinBox { font-size: 13px; font-weight: bold; }") action_row.addWidget(self.parallel_spin) self.btn_flash = QPushButton("⚡ FLASH SELECTED DEVICES") self.btn_flash.setObjectName("flash") self.btn_flash.setFixedHeight(34) self.btn_flash.setStyleSheet("QPushButton#flash { font-size: 13px; font-weight: bold; letter-spacing: 1px; }") self.btn_flash.clicked.connect(self.flash_all) action_row.addWidget(self.btn_flash, 1) flash_layout.addLayout(action_row) flash_group.set_content_layout(flash_layout) layout.addWidget(flash_group) # ── Automation Button ── self.btn_auto = QPushButton("🤖 Tự động hóa nạp FW") self.btn_auto.setObjectName("auto") self.btn_auto.setFixedHeight(32) self.btn_auto.setStyleSheet(""" QPushButton#auto { background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #7c3aed, stop:1 #a78bfa); border-color: #7c3aed; color: #ffffff; font-size: 12px; font-weight: bold; letter-spacing: 1px; border-radius: 6px; } QPushButton#auto:hover { background: qlineargradient(x1:0, y1:0, x2:1, y2:0, stop:0 #8b5cf6, stop:1 #c4b5fd); } """) self.btn_auto.clicked.connect(self._open_auto_flash) layout.addWidget(self.btn_auto) self.setLayout(layout) # ── AUTO TAB (hidden by default) ── self.auto_widget = None # ── Helpers ── def _guess_gateway(self, ip): """Guess gateway IP (x.x.x.1) from local IP.""" try: parts = ip.split(".") return f"{parts[0]}.{parts[1]}.{parts[2]}.1" except Exception: return "" def _info_label(self, text): lbl = QLabel(text) lbl.setObjectName("info") return lbl def _info_value(self, text): lbl = QLabel(text) lbl.setStyleSheet("color: #cdd6f4; font-weight: bold;") return lbl def _get_filtered_devices(self): """Return devices filtered based on show_all checkbox.""" if self.show_all_cb.isChecked(): return list(self.all_devices) excluded = {self.local_ip, self.gateway_ip} return [d for d in self.all_devices if d["ip"] not in excluded] def _refresh_table(self): """Re-populate table based on current filter state.""" self.devices = self._get_filtered_devices() self.table.setRowCount(len(self.devices)) for i, d in enumerate(self.devices): mac_str = d["mac"].upper() # Checkbox column cb_item = QTableWidgetItem() # If device MAC is already flashed, mark it is_already_flashed = mac_str in self.flashed_macs if is_already_flashed and d["status"] in ["READY", ""]: d["status"] = "Already Flashed" # Checkbox logic if is_already_flashed: cb_item.setCheckState(Qt.CheckState.Unchecked) else: cb_item.setCheckState(Qt.CheckState.Checked) cb_item.setFlags(cb_item.flags() | Qt.ItemFlag.ItemIsUserCheckable) self.table.setItem(i, 0, cb_item) ip_item = QTableWidgetItem(d["ip"]) mac_item = QTableWidgetItem(mac_str) status_item = QTableWidgetItem(d["status"]) if is_already_flashed: status_item.setForeground(QColor("#a6e3a1")) ip_item.setForeground(QColor("#a6e3a1")) # Dim gateway & self if showing all if d["ip"] in {self.local_ip, self.gateway_ip}: for item in [ip_item, mac_item, status_item]: item.setForeground(QColor("#4a5568")) cb_item.setCheckState(Qt.CheckState.Unchecked) self.table.setItem(i, 1, ip_item) self.table.setItem(i, 2, mac_item) self.table.setItem(i, 3, status_item) total = len(self.devices) hidden = len(self.all_devices) - total label = f"Total: {total} devices" if hidden > 0: label += f" (hiding {hidden})" self.device_count_label.setText(label) def _select_all_devices(self): """Check all device checkboxes.""" for i in range(self.table.rowCount()): mac = self.table.item(i, 2).text().strip() # Prevent checking if already flashed or is a gateway (optional but good UI logic) if mac not in self.flashed_macs: item = self.table.item(i, 0) if item: item.setCheckState(Qt.CheckState.Checked) def _deselect_all_devices(self): """Uncheck all device checkboxes.""" for i in range(self.table.rowCount()): item = self.table.item(i, 0) if item: item.setCheckState(Qt.CheckState.Unchecked) def _show_history(self): """Hiển thị lịch sử thiết bị đã nạp trong phiên.""" if not self.flashed_macs: QMessageBox.information(self, "Lịch sử nạp", "Chưa có thiết bị nào được nạp trong phiên này.") else: lines = [] for mac in sorted(self.flashed_macs.keys()): ip, _, result, ts = self.flashed_macs[mac] icon = "✅" if result.startswith("DONE") else "❌" lines.append(f"[{ts}] {icon} {ip} ({mac}) — {result}") msg = f"Lịch sử nạp FW ({len(self.flashed_macs)} thiết bị):\n\n" + "\n".join(lines) QMessageBox.information(self, "Lịch sử nạp", msg) # ── Actions ── def select_fw(self): file, _ = QFileDialog.getOpenFileName( self, "Select Firmware", "", "Firmware Files (*.bin *.hex *.uf2);;All Files (*)" ) if file: self.firmware = file name = file.split("/")[-1] self.fw_label.setText(f"✅ {name}") self.fw_label.setStyleSheet("color: #a6e3a1; font-weight: bold; font-size: 11px;") def scan(self): network_str = self.net_input.text().strip() try: network = ipaddress.ip_network(network_str, strict=False) except ValueError: QMessageBox.warning( self, "Invalid Network", f"'{network_str}' is not a valid network.\n" "Example: 192.168.4.0/24" ) return self.scan_status.setText("⏳ Preparing scan...") self.scan_status.setStyleSheet("color: #f9e2af;") self.table.setRowCount(0) self.devices = [] self.scan_progress_bar.setRange(0, 0) self.scan_progress_bar.setFormat(" Starting...") self.scan_progress_bar.setVisible(True) self.btn_scan.setEnabled(False) self.scan_thread = ScanThread(network) self.scan_thread.finished.connect(self._on_scan_done) self.scan_thread.error.connect(self._on_scan_error) self.scan_thread.scan_progress.connect(self._on_scan_progress) self.scan_thread.stage.connect(self._on_scan_stage) self.scan_thread.start() def _on_scan_done(self, results): self.scan_progress_bar.setVisible(False) self.btn_scan.setEnabled(True) self.all_devices = [] for dev in results: dev["status"] = "READY" self.all_devices.append(dev) self.all_devices.sort(key=lambda d: ipaddress.ip_address(d["ip"])) # Apply filter and populate table self._refresh_table() total_all = len(self.all_devices) total_shown = len(self.devices) self.scan_status.setText(f"✅ Scan complete — {total_all} found, {total_shown} shown") self.scan_status.setStyleSheet("color: #a6e3a1;") def _on_scan_stage(self, stage): if stage == "ping": self.scan_progress_bar.setFormat("⚡ Pinging hosts... %v / %m") self.scan_status.setText("⏳ Pinging all hosts...") self.scan_status.setStyleSheet("color: #f9e2af;") elif stage == "arp": self.scan_progress_bar.setRange(0, 0) self.scan_progress_bar.setFormat(" Reading ARP cache...") self.scan_status.setText("⏳ Reading ARP cache...") elif stage == "scapy": self.scan_progress_bar.setRange(0, 0) self.scan_progress_bar.setFormat(" ARP broadcast scan...") self.scan_status.setText("⏳ Running ARP broadcast scan...") def _on_scan_progress(self, done, total): self.scan_progress_bar.setRange(0, total) self.scan_progress_bar.setValue(done) self.scan_status.setText(f"⏳ Pinging hosts... {done} / {total}") self.scan_status.setStyleSheet("color: #f9e2af;") def _on_scan_error(self, error_msg): self.scan_progress_bar.setVisible(False) self.btn_scan.setEnabled(True) self.scan_status.setText("❌ Scan failed") self.scan_status.setStyleSheet("color: #f38ba8;") QMessageBox.critical( self, "Scan Error", f"Failed to scan network:\n{error_msg}" ) def _on_mode_changed(self, index): """Show/hide method based on selected mode.""" mode = self.mode_combo.currentData() if mode == "update": self.method_container.setVisible(False) self.update_warning_lbl.setVisible(True) else: self.method_container.setVisible(True) self.update_warning_lbl.setVisible(False) self._on_method_changed(self.method_combo.currentIndex()) def _on_method_changed(self, index): """SSH credentials are always hidden; credentials are hardcoded.""" self.ssh_creds_widget.setVisible(False) def _get_selected_devices(self): """Return list of (table_row_index, device_dict) for checked devices.""" selected = [] for i in range(self.table.rowCount()): item = self.table.item(i, 0) if item and item.checkState() == Qt.CheckState.Checked: if i < len(self.devices): selected.append((i, self.devices[i])) return selected def flash_all(self): if not self.firmware: QMessageBox.warning( self, "No Firmware", "Please select a firmware file first." ) return if not self.devices: QMessageBox.information( self, "No Devices", "No devices found to flash.\n" "Please scan the network first." ) return selected = self._get_selected_devices() if not selected: QMessageBox.information( self, "No Selection", "No devices selected.\n" "Check the boxes next to the devices you want to flash." ) return total = len(selected) self.progress.setMaximum(total) self.progress.setValue(0) # Determine flash method and SSH credentials mode = self.mode_combo.currentData() # Validate: IP 192.168.11.102 chỉ được nạp ở chế độ Update FW if mode != "update": blocked = [dev["ip"] for _, dev in selected if dev["ip"] == "192.168.11.102"] if blocked: QMessageBox.warning( self, "Không được phép", "⚠️ Thiết bị 192.168.11.102 chỉ được nạp ở chế độ Update FW.\n" "Vui lòng bỏ chọn thiết bị này hoặc chuyển sang chế độ Update FW." ) return # Build list with row indices for UI updates self._flash_row_map = {} flash_devices = [] for idx, (row, dev) in enumerate(selected): self._flash_row_map[idx] = row flash_devices.append(dev) if mode == "update": # Hỏi xác nhận nếu có device nào khác 192.168.11.102 strange_ips = [dev["ip"] for _, dev in selected if dev["ip"] != "192.168.11.102"] if strange_ips: msg = (f"⚠️ Detected IP(s) other than [192.168.11.102] in your selection:\n" f"{', '.join(strange_ips[:3])}{'...' if len(strange_ips) > 3 else ''}\n\n" f"Are you SURE you want to update firmware on these devices?") reply = QMessageBox.question( self, "Confirm Update Target", msg, QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No ) if reply == QMessageBox.StandardButton.No: return method = "ssh" ssh_user = "root" ssh_password = "admin123a" ssh_backup_password = "admin" set_passwd = False else: method = self.method_combo.currentData() ssh_user = "root" ssh_password = "admin123a" ssh_backup_password = "admin123a" set_passwd = True if method == "ssh" else False # Chọn đúng worker theo mode và chạy trong background thread max_w = self.parallel_spin.value() if mode == "update": self.flash_thread = UpdateFlashThread( flash_devices, self.firmware, max_workers=max_w, ) else: self.flash_thread = NewFlashThread( flash_devices, self.firmware, max_workers=max_w, method=method, ssh_user=ssh_user, ssh_password=ssh_password, ssh_backup_password=ssh_backup_password, set_passwd=set_passwd, ) self.flash_thread.device_status.connect(self._on_flash_status) self.flash_thread.device_done.connect(self._on_flash_done) self.flash_thread.all_done.connect(self._on_flash_all_done) # Disable flash button trong khi đang flash self.btn_flash.setEnabled(False) self.flash_thread.start() def _on_flash_status(self, index, msg): """Update status column while flashing.""" row = self._flash_row_map.get(index, index) self.table.setItem(row, 3, QTableWidgetItem(f"⏳ {msg}")) def _on_flash_done(self, index, result): """One device finished flashing.""" row = self._flash_row_map.get(index, index) mac_item = self.table.item(row, 2) ip_item = self.table.item(row, 1) import datetime now_str = datetime.datetime.now().strftime("%H:%M:%S") mac_str = mac_item.text().strip() if mac_item else "" ip_str = ip_item.text().strip() if ip_item else "" if result.startswith("DONE"): item = QTableWidgetItem(f"✅ {result}") item.setForeground(QColor("#a6e3a1")) # Auto-uncheck so it won't be flashed again cb = self.table.item(row, 0) if cb: cb.setCheckState(Qt.CheckState.Unchecked) else: item = QTableWidgetItem(f"❌ {result}") item.setForeground(QColor("#f38ba8")) # Lưu vào lịch sử nạp if mac_str: self.flashed_macs[mac_str] = (ip_str, mac_str, result, now_str) self.table.setItem(row, 3, item) self.progress.setValue(self.progress.value() + 1) def _on_flash_all_done(self): """All flashing complete.""" self.btn_flash.setEnabled(True) QMessageBox.information(self, "Flash Complete", "All devices have been processed.") def _open_auto_flash(self): """Mở cửa sổ Tự động hóa nạp FW.""" if self.auto_widget is None or not self.auto_widget.isVisible(): self.auto_widget = AutoFlashWindow( firmware=self.firmware, network=self.net_input.text().strip(), local_ip=self.local_ip, gateway_ip=self.gateway_ip, parent_app=self, ) self.auto_widget.resize(700, 750) self.auto_widget.show() else: self.auto_widget.raise_() self.auto_widget.activateWindow() class AutoFlashWindow(QWidget): """Cửa sổ Tự động hóa nạp FW — scan + flash tự động.""" def __init__(self, firmware=None, network="", local_ip="", gateway_ip="", parent_app=None): super().__init__() self.setWindowTitle("🤖 Tự động hóa nạp FW") self.setWindowIcon(QIcon(resource_path("icon.ico"))) self.firmware = firmware self.local_ip = local_ip self.gateway_ip = gateway_ip self.parent_app = parent_app self.worker = None self._device_rows = {} # ip -> row index in result table self.setStyleSheet(AUTO_STYLE) layout = QVBoxLayout() layout.setSpacing(4) layout.setContentsMargins(10, 8, 10, 8) # ── Title row (compact) ── title = QLabel("🤖 Tự động hóa nạp FW") title.setObjectName("title") title.setAlignment(Qt.AlignmentFlag.AlignCenter) layout.addWidget(title) # ── Top bar: FW + Network + Config (compact, 2 rows) ── config_group = CollapsibleGroupBox("⚙️ Cấu hình nạp") config_layout = QVBoxLayout() config_layout.setSpacing(6) config_layout.setContentsMargins(8, 4, 8, 4) # Row 1: Firmware + Network row1 = QHBoxLayout() row1.setSpacing(12) fw_lbl = QLabel("FW:") fw_lbl.setStyleSheet("font-weight: bold; font-size: 12px;") row1.addWidget(fw_lbl) self.fw_label = QLabel(os.path.basename(firmware) if firmware else "Chưa chọn") self.fw_label.setStyleSheet( "color: #a6e3a1; font-weight: bold; font-size: 12px;" if firmware else "color: #f38ba8; font-size: 12px;" ) row1.addWidget(self.fw_label) btn_fw = QPushButton("📁") btn_fw.setFixedSize(32, 28) btn_fw.setToolTip("Chọn firmware") btn_fw.clicked.connect(self._select_firmware) row1.addWidget(btn_fw) sep1 = QLabel("│") sep1.setStyleSheet("color: #3d4a6b; font-size: 14px;") row1.addWidget(sep1) net_lbl = QLabel("Mạng:") net_lbl.setStyleSheet("font-weight: bold; font-size: 12px;") row1.addWidget(net_lbl) self.net_input = QLineEdit(network or get_default_network(self.local_ip)) self.net_input.setPlaceholderText("e.g. 192.168.4.0/24") self.net_input.setMaximumWidth(180) self.net_input.setFixedHeight(28) self.net_input.setStyleSheet("QLineEdit { font-size: 12px; padding: 3px 8px; }") row1.addWidget(self.net_input) row1.addStretch() config_layout.addLayout(row1) # Row 2: Target count + Method + Concurrent row2 = QHBoxLayout() row2.setSpacing(12) cnt_lbl = QLabel("Số lượng:") cnt_lbl.setStyleSheet("font-weight: bold; font-size: 12px;") row2.addWidget(cnt_lbl) self.target_spin = QSpinBox() self.target_spin.setRange(1, 500) self.target_spin.setValue(5) self.target_spin.setFixedWidth(70) self.target_spin.setFixedHeight(28) self.target_spin.setStyleSheet("QSpinBox { font-size: 13px; font-weight: bold; }") row2.addWidget(self.target_spin) sep2 = QLabel("│") sep2.setStyleSheet("color: #3d4a6b; font-size: 14px;") row2.addWidget(sep2) meth_lbl = QLabel("Phương thức:") meth_lbl.setStyleSheet("font-weight: bold; font-size: 12px;") row2.addWidget(meth_lbl) self.method_combo = QComboBox() self.method_combo.addItem("🌐 API (LuCI)", "api") self.method_combo.addItem("💻 SSH", "ssh") self.method_combo.setFixedHeight(28) self.method_combo.setMinimumWidth(140) row2.addWidget(self.method_combo) sep3 = QLabel("│") sep3.setStyleSheet("color: #3d4a6b; font-size: 14px;") row2.addWidget(sep3) par_lbl = QLabel("Song song:") par_lbl.setStyleSheet("font-weight: bold; font-size: 12px;") row2.addWidget(par_lbl) self.parallel_spin = QSpinBox() self.parallel_spin.setRange(0, 100) self.parallel_spin.setValue(10) self.parallel_spin.setSpecialValueText("∞") self.parallel_spin.setToolTip("0 = không giới hạn") self.parallel_spin.setFixedWidth(65) self.parallel_spin.setFixedHeight(28) self.parallel_spin.setStyleSheet("QSpinBox { font-size: 13px; font-weight: bold; }") row2.addWidget(self.parallel_spin) row2.addStretch() config_layout.addLayout(row2) config_group.set_content_layout(config_layout) layout.addWidget(config_group) # ── Control Buttons (compact) ── btn_row = QHBoxLayout() btn_row.setSpacing(8) self.btn_start = QPushButton("▶ XÁC NHẬN & BẮT ĐẦU") self.btn_start.setObjectName("start_btn") self.btn_start.setFixedHeight(36) self.btn_start.clicked.connect(self._on_start) btn_row.addWidget(self.btn_start) self.btn_stop = QPushButton("⏹ DỪNG") self.btn_stop.setObjectName("stop_btn") self.btn_stop.setFixedHeight(36) self.btn_stop.setEnabled(False) self.btn_stop.clicked.connect(self._on_stop) btn_row.addWidget(self.btn_stop) layout.addLayout(btn_row) # ── Status + Progress (inline) ── status_row = QHBoxLayout() status_row.setSpacing(8) self.status_label = QLabel("⏸ Chờ bắt đầu...") self.status_label.setStyleSheet("font-size: 12px; font-weight: bold; color: #94a3b8;") status_row.addWidget(self.status_label, 1) self.progress_bar = QProgressBar() self.progress_bar.setFormat("%v / %m (%p%)") self.progress_bar.setFixedHeight(18) self.progress_bar.setFixedWidth(250) self.progress_bar.setVisible(False) status_row.addWidget(self.progress_bar) layout.addLayout(status_row) # ── Device Table (MAIN area — stretch) ── dev_group = QGroupBox("📋 Danh sách thiết bị") dev_layout = QVBoxLayout() dev_layout.setSpacing(2) dev_layout.setContentsMargins(4, 12, 4, 4) self.result_table = QTableWidget() self.result_table.setColumnCount(4) self.result_table.setHorizontalHeaderLabels(["#", "IP", "MAC", "Kết quả"]) self.result_table.setAlternatingRowColors(True) header = self.result_table.horizontalHeader() header.setSectionResizeMode(0, QHeaderView.ResizeMode.Fixed) header.resizeSection(0, 35) header.setSectionResizeMode(1, QHeaderView.ResizeMode.Interactive) header.resizeSection(1, 120) header.setSectionResizeMode(2, QHeaderView.ResizeMode.Interactive) header.resizeSection(2, 140) header.setSectionResizeMode(3, QHeaderView.ResizeMode.Stretch) self.result_table.verticalHeader().setVisible(False) self.result_table.setStyleSheet("QTableWidget { font-size: 12px; }") dev_layout.addWidget(self.result_table) # Summary row below table summary_row = QHBoxLayout() summary_row.setSpacing(6) self.summary_label = QLabel("") self.summary_label.setStyleSheet("color: #94a3b8; font-size: 11px; padding: 2px;") summary_row.addWidget(self.summary_label, 1) btn_auto_history = QPushButton("📋 Lịch sử nạp") btn_auto_history.setFixedHeight(24) btn_auto_history.setStyleSheet("background-color: #313244; color: #cdd6f4; font-size: 11px; padding: 2px 8px;") btn_auto_history.clicked.connect(self._show_auto_history) summary_row.addWidget(btn_auto_history) dev_layout.addLayout(summary_row) dev_group.setLayout(dev_layout) layout.addWidget(dev_group, stretch=3) # ── Log (collapsible, compact) ── log_group = CollapsibleGroupBox("📝 Log") log_layout = QVBoxLayout() log_layout.setContentsMargins(4, 2, 4, 2) self.log_area = QScrollArea() self.log_content = QLabel("") self.log_content.setWordWrap(True) self.log_content.setAlignment(Qt.AlignmentFlag.AlignTop) self.log_content.setStyleSheet( "color: #cdd6f4; font-size: 10px; font-family: 'SF Mono', 'Menlo', monospace;" "padding: 4px; background-color: #11121d; border-radius: 4px;" ) self.log_content.setTextFormat(Qt.TextFormat.PlainText) self.log_area.setWidget(self.log_content) self.log_area.setWidgetResizable(True) self.log_area.setMinimumHeight(120) self.log_area.setMaximumHeight(280) self.log_area.setStyleSheet("QScrollArea { border: 1px solid #2d3748; border-radius: 4px; background-color: #11121d; }") log_layout.addWidget(self.log_area) log_group.set_content_layout(log_layout) layout.addWidget(log_group, stretch=1) self.setLayout(layout) self._log_lines = [] self._success_count = 0 self._fail_count = 0 self._auto_history = [] # list of (ip, mac, result, timestamp) def _show_auto_history(self): """Hiển thị lịch sử thiết bị đã nạp trong phiên tự động hóa.""" if not self._auto_history: QMessageBox.information(self, "Lịch sử nạp", "Chưa có thiết bị nào được nạp trong phiên này.") return lines = [] for ip, mac, result, ts in self._auto_history: icon = "✅" if result.startswith("DONE") else "❌" lines.append(f"[{ts}] {icon} {ip} ({mac}) — {result}") msg = f"Lịch sử nạp FW ({len(self._auto_history)} thiết bị):\n\n" + "\n".join(lines) QMessageBox.information(self, "Lịch sử nạp", msg) def _select_firmware(self): file, _ = QFileDialog.getOpenFileName( self, "Chọn Firmware", "", "Firmware Files (*.bin *.hex *.uf2);;All Files (*)" ) if file: self.firmware = file self.fw_label.setText(os.path.basename(file)) self.fw_label.setStyleSheet("color: #a6e3a1; font-weight: bold;") def _append_log(self, msg): self._log_lines.append(msg) # Giới hạn 500 dòng log if len(self._log_lines) > 500: self._log_lines = self._log_lines[-500:] self.log_content.setText("\n".join(self._log_lines)) # Scroll to bottom sb = self.log_area.verticalScrollBar() sb.setValue(sb.maximum()) def _on_start(self): if not self.firmware: QMessageBox.warning(self, "Chưa chọn FW", "Vui lòng chọn file firmware trước.") return network_str = self.net_input.text().strip() try: ipaddress.ip_network(network_str, strict=False) except ValueError: QMessageBox.warning(self, "Mạng không hợp lệ", f"'{network_str}' không phải network hợp lệ.\nVí dụ: 192.168.4.0/24") return target = self.target_spin.value() method = self.method_combo.currentData() max_workers = self.parallel_spin.value() # Confirm reply = QMessageBox.question( self, "Xác nhận", f"Bắt đầu tự động nạp FW?\n\n" f" Firmware: {os.path.basename(self.firmware)}\n" f" Mạng: {network_str}\n" f" Số lượng: {target} thiết bị\n" f" Phương thức: {method.upper()}\n" f" Song song: {max_workers if max_workers > 0 else 'Không giới hạn'}\n", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.Yes, ) if reply != QMessageBox.StandardButton.Yes: return # Reset state self._log_lines = [] self.log_content.setText("") self.result_table.setRowCount(0) self._device_rows = {} self._success_count = 0 self._fail_count = 0 self._auto_history.clear() self.summary_label.setText("") self.progress_bar.setVisible(False) self.btn_start.setEnabled(False) self.btn_stop.setEnabled(True) self.net_input.setEnabled(False) self.target_spin.setEnabled(False) self.method_combo.setEnabled(False) self.parallel_spin.setEnabled(False) self.status_label.setText("🔍 Đang scan mạng LAN...") self.status_label.setStyleSheet("font-size: 12px; font-weight: bold; color: #f9e2af;") self.worker = AutoFlashWorker( network=network_str, target_count=target, method=method, max_workers=max_workers, firmware_path=self.firmware, local_ip=self.local_ip, gateway_ip=self.gateway_ip, ) self.worker.log_message.connect(self._append_log) self.worker.scan_found.connect(self._on_scan_found) self.worker.devices_ready.connect(self._on_devices_ready) self.worker.device_status.connect(self._on_device_status) self.worker.device_done.connect(self._on_device_done) self.worker.flash_progress.connect(self._on_flash_progress) self.worker.all_done.connect(self._on_all_done) self.worker.scan_timeout.connect(self._on_scan_timeout) self.worker.stopped.connect(self._on_stopped) self.worker.start() def _on_stop(self): if self.worker: self.worker.stop() self.btn_stop.setEnabled(False) self.status_label.setText("⏳ Đang dừng...") def _on_scan_found(self, count): target = self.target_spin.value() self.status_label.setText(f"🔍 Scan: tìm thấy {count}/{target} thiết bị...") if count >= target: self.status_label.setText(f"⚡ Đủ {target} thiết bị — đang nạp FW...") self.status_label.setStyleSheet("font-size: 12px; font-weight: bold; color: #a6e3a1;") def _on_devices_ready(self, devices): """Pre-populate bảng kết quả trước khi bắt đầu flash.""" self.result_table.setRowCount(0) self._device_rows = {} for i, dev in enumerate(devices): self.result_table.insertRow(i) num_item = QTableWidgetItem(str(i + 1)) num_item.setTextAlignment(Qt.AlignmentFlag.AlignCenter) self.result_table.setItem(i, 0, num_item) self.result_table.setItem(i, 1, QTableWidgetItem(dev["ip"])) self.result_table.setItem(i, 2, QTableWidgetItem(dev.get("mac", "N/A").upper())) waiting_item = QTableWidgetItem("⏳ Đang chờ...") waiting_item.setForeground(QColor("#94a3b8")) self.result_table.setItem(i, 3, waiting_item) self._device_rows[dev["ip"]] = i self.summary_label.setText(f"Tổng: {len(devices)} thiết bị") def _on_device_status(self, ip, msg): row = self._device_rows.get(ip) if row is not None: item = QTableWidgetItem(f"⏳ {msg}") item.setForeground(QColor("#f9e2af")) self.result_table.setItem(row, 3, item) def _on_device_done(self, ip, mac, result): # Thêm dòng mới vào bảng kết quả nếu chưa có if ip not in self._device_rows: row = self.result_table.rowCount() self.result_table.insertRow(row) num_item = QTableWidgetItem(str(row + 1)) num_item.setTextAlignment(Qt.AlignmentFlag.AlignCenter) self.result_table.setItem(row, 0, num_item) self.result_table.setItem(row, 1, QTableWidgetItem(ip)) self.result_table.setItem(row, 2, QTableWidgetItem(mac.upper())) self._device_rows[ip] = row row = self._device_rows[ip] now_str = datetime.datetime.now().strftime("%H:%M:%S") if result.startswith("DONE"): item = QTableWidgetItem(f"✅ {result}") item.setForeground(QColor("#a6e3a1")) self._success_count += 1 # Lưu vào FlashHistory của cửa sổ chính if self.parent_app: self.parent_app.flashed_macs[mac.upper()] = (ip, mac.upper(), result, now_str) else: item = QTableWidgetItem(f"❌ {result}") item.setForeground(QColor("#f38ba8")) self._fail_count += 1 self.result_table.setItem(row, 3, item) self._auto_history.append((ip, mac.upper(), result, now_str)) total = len(self._device_rows) done = self._success_count + self._fail_count self.summary_label.setText( f"Tổng: {total} | Xong: {done} | ✅ {self._success_count} | ❌ {self._fail_count}" ) def _on_flash_progress(self, done, total): self.progress_bar.setVisible(True) self.progress_bar.setMaximum(total) self.progress_bar.setValue(done) self.status_label.setText(f"⚡ Nạp FW: {done}/{total} thiết bị...") def _on_all_done(self, success, fail): self._reset_controls() self.status_label.setText(f"🏁 Hoàn thành! ✅ {success} | ❌ {fail}") self.status_label.setStyleSheet("font-size: 12px; font-weight: bold; color: #a6e3a1;") QMessageBox.information( self, "Hoàn thành", f"Tự động nạp FW hoàn thành!\n\n" f"✅ Thành công: {success}\n" f"❌ Thất bại: {fail}", ) def _on_stopped(self): self._reset_controls() self.status_label.setText("⛔ Đã dừng bởi người dùng") self.status_label.setStyleSheet("font-size: 12px; font-weight: bold; color: #f38ba8;") def _on_scan_timeout(self, found, target): self._reset_controls() self.status_label.setText(f"⚠️ Scan hết lần: chỉ tìm thấy {found}/{target} thiết bị") self.status_label.setStyleSheet("font-size: 12px; font-weight: bold; color: #fab387;") QMessageBox.warning( self, "Không đủ thiết bị", f"Đã scan tối đa nhưng chỉ tìm thấy {found}/{target} thiết bị.\n\n" f"Vui lòng kiểm tra:\n" f" • Thiết bị đã bật và kết nối mạng chưa\n" f" • Dải mạng ({self.net_input.text()}) có đúng không\n" f" • Thử lại sau khi kiểm tra", ) def _reset_controls(self): self.btn_start.setEnabled(True) self.btn_stop.setEnabled(False) self.net_input.setEnabled(True) self.target_spin.setEnabled(True) self.method_combo.setEnabled(True) self.parallel_spin.setEnabled(True) self.worker = None if __name__ == "__main__": app = QApplication(sys.argv) app.setStyleSheet(STYLE) window = App() # 60% height, limited width (750px), centered screen = app.primaryScreen().availableGeometry() w = min(750, screen.width()) h = int(screen.height() * 0.6) x = (screen.width() - w) // 2 y = screen.y() + (screen.height() - h) // 2 window.setGeometry(x, y, w, h) window.show() sys.exit(app.exec())