masonhuang/KNEO-Academy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor: Clean up codebase and improve documentation

Browse Source 
- Remove unused files (model_controller.py, model_service.py, device_connection_popup.py)
- Clean up commented code in device_service.py, device_popup.py, config.py
- Update docstrings and comments across all modules
- Improve code organization and readability
This commit is contained in:
HuangMason320 2025-12-30 16:47:31 +08:00
parent 09156cce94
commit 17deba3bdb
18 changed files with 1388 additions and 867 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
main.py
View File

@ -1,3 +1,11 @@
"""
main.py - Kneron Academy Application Entry Point
This module serves as the main entry point for the application.
It initializes the PyQt5 application and uses AppController to manage
navigation and screen transitions.
"""
import sys
from PyQt5.QtWidgets import QApplication, QStackedWidget
from src.views.mainWindows import MainWindow
@ -6,7 +14,25 @@ from src.views.login_screen import LoginScreen
from src.views.utilities_screen import UtilitiesScreen
from src.config import APP_NAME, WINDOW_SIZE
class AppController:
"""
Application Controller Class
Manages the entire application lifecycle, including:
- Initializing all screens
- Setting up signal connections between screens
- Controlling screen navigation logic
Attributes:
app (QApplication): The PyQt5 application instance
stack (QStackedWidget): Stacked widget container for managing multiple screens
selection_screen (SelectionScreen): The selection/home screen
login_screen (LoginScreen): The login screen
utilities_screen (UtilitiesScreen): The utilities screen
main_window (MainWindow): The main demo application window
"""
def __init__(self):
self.app = QApplication(sys.argv)
self.stack = QStackedWidget()
@ -23,53 +49,81 @@ class AppController:
self.show_selection_screen()
def init_screens(self):
# Selection screen
"""
Initialize all application screens.
Creates instances of each screen and adds them to the stacked widget.
The order of addition determines the default index of each screen.
"""
self.selection_screen = SelectionScreen()
self.stack.addWidget(self.selection_screen)
# Login screen
self.login_screen = LoginScreen()
self.stack.addWidget(self.login_screen)
# Utilities screen
self.utilities_screen = UtilitiesScreen()
self.stack.addWidget(self.utilities_screen)
# Demo app (main window)
self.main_window = MainWindow()
self.stack.addWidget(self.main_window)
def connect_signals(self):
# Selection screen signals
"""
Connect signals between screens for navigation.
Sets up Qt signal-slot connections to enable screen transitions:
- Selection screen -> Login screen (for utilities access)
- Selection screen -> Demo app
- Login screen -> Utilities screen (on successful login)
- Login screen -> Selection screen (back navigation)
- Utilities screen -> Selection screen (back navigation)
"""
self.selection_screen.open_utilities.connect(self.show_login_screen)
self.selection_screen.open_demo_app.connect(self.show_demo_app)
# Login screen signals
self.login_screen.login_success.connect(self.show_utilities_screen)
self.login_screen.back_to_selection.connect(self.show_selection_screen)
# Utilities screen signals
self.utilities_screen.back_to_selection.connect(self.show_selection_screen)
def show_selection_screen(self):
"""Switch to the selection/home screen."""
self.stack.setCurrentWidget(self.selection_screen)
def show_login_screen(self):
"""Switch to the login screen."""
self.stack.setCurrentWidget(self.login_screen)
def show_utilities_screen(self):
"""Switch to the utilities screen."""
self.stack.setCurrentWidget(self.utilities_screen)
def show_demo_app(self):
"""Switch to the main demo application window."""
self.stack.setCurrentWidget(self.main_window)
def run(self):
"""
Start the application event loop.
Returns:
int: The exit code from the Qt application event loop.
"""
self.stack.show()
return self.app.exec_()
def main():
"""
Application entry point function.
Creates the AppController and starts the application event loop.
Returns:
int: The exit code from the application.
"""
controller = AppController()
return controller.run()
if __name__ == "__main__":
main()

37
src/config.py
View File

@ -1,15 +1,26 @@
"""
config.py - Application Configuration
This module contains all global configuration constants, paths, and styles
for the Kneron Academy AI Playground application.
"""
from enum import Enum
import os
# Application data path (platform-specific)
APPDATA_PATH = os.environ.get("LOCALAPPDATA")
# APPDATA_PATH = "/Users/mason/Developer/Kneron-Academy/test_images"
# 取得專案根目錄的絕對路徑並設定 UXUI_ASSETS 為絕對路徑
# Get project root directory and set UXUI_ASSETS as absolute path
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
UXUI_ASSETS = os.path.join(PROJECT_ROOT, "uxui", "")
# 新版路徑結構 (不需要獨立的 models 和 scripts 資料夾)
# Directory structure for utilities, scripts, uploads, and firmware
UTILS_DIR = os.path.join(APPDATA_PATH, "Kneron_Academy", "utils")
SCRIPT_CONFIG = os.path.join(UTILS_DIR, "config.json")
UPLOAD_DIR = os.path.join(APPDATA_PATH, "Kneron_Academy", "uploads")
FW_DIR = os.path.join(APPDATA_PATH, "Kneron_Academy", "firmware")
# Global Constants
APP_NAME = "Innovedus AI Playground"
WINDOW_SIZE = (1200, 900)
@ -45,22 +56,13 @@ FIRMWARE_PATHS = {
"ncpu": "../../res/firmware/fw_ncpu.bin",
}
# MODEL_BUTTON = [
# ('Face Detection', self.run_face_detection),
# ('Gender/Age Detection', self.run_gender_age_detection),
# ('Object Detection', self.run_object_detection),
# ('Mask Detection', self.run_mask_detection),
# ('Image Project', self.start_image_project),
# ('Upload Model', self.upload_model)
# ]
# Model Inference Parameter
MODEL_TIMEOUT = 5000
# TODO: Mapping of the values
# Device type enumeration for Kneron hardware
class DeviceType(Enum):
"""Enumeration of supported Kneron device types with their product IDs."""
KL520 = 256
# KL720 = 720
KL720 = 1824
KL720_L = 512
KL530 = 530
@ -69,11 +71,14 @@ class DeviceType(Enum):
KL630 = 630
KL540 = 540
# Mapping from product_id hex string to device model name
DongleModelMap = {
"0x100": "KL520", # product_id "0x100" 對應到 520 系列
"0x720": "KL720", # product_id "0x720" 對應到 720 系列
"0x100": "KL520", # product_id "0x100" maps to KL520 series
"0x720": "KL720", # product_id "0x720" maps to KL720 series
}
# Mapping from product_id hex string to device icon filename
DongleIconMap = {
"0x100": "ic_dongle_520.png",
"0x720": "ic_dongle_720.png"

169
src/controllers/device_controller.py
View File

@ -1,45 +1,75 @@
# src/controllers/device_controller.py
"""
device_controller.py - Device Controller
This module handles Kneron device connection, selection, and management.
It provides functionality for scanning, connecting, and disconnecting devices.
"""
from PyQt5.QtWidgets import QWidget, QListWidgetItem
from PyQt5.QtGui import QPixmap, QIcon
from PyQt5.QtCore import Qt
import os
import kp # 新增 kp 模組的引入
import kp
from src.services.device_service import check_available_device
from src.config import UXUI_ASSETS, DongleModelMap, DongleIconMap, FW_DIR
class DeviceController:
"""
Controller class for managing Kneron device connections.
Attributes:
main_window: Reference to the main application window
selected_device: Currently selected device
connected_devices: List of connected devices
device_group: Kneron Plus device group for connected devices
"""
def __init__(self, main_window):
"""
Initialize the DeviceController.
Args:
main_window: Reference to the main application window.
"""
self.main_window = main_window
self.selected_device = None
self.connected_devices = []
self.device_group = None # 新增儲存連接的 device_group
self.device_group = None # Stores connected device group
def refresh_devices(self):
"""Refresh the list of connected devices"""
"""
Refresh the list of connected devices.
Scans for available Kneron devices and updates the UI.
Returns:
bool: True if devices were found, False otherwise.
"""
try:
print("[CTRL] Refreshing devices...")
device_descriptors = check_available_device()
print("[CTRL] check_available_device 已返回")
print(f"[CTRL] device_descriptors 類型: {type(device_descriptors)}")
print("[CTRL] check_available_device returned")
print(f"[CTRL] device_descriptors type: {type(device_descriptors)}")
# 分開訪問屬性以便調試
print("[CTRL] 嘗試訪問 device_descriptor_number...")
# Access attributes separately for debugging
print("[CTRL] Accessing device_descriptor_number...")
desc_num = device_descriptors.device_descriptor_number
print(f"[CTRL] device_descriptor_number: {desc_num}")
self.connected_devices = []
if device_descriptors.device_descriptor_number > 0:
print("[DEBUG] 開始 parse_and_store_devices...")
print("[DEBUG] Starting parse_and_store_devices...")
self.parse_and_store_devices(device_descriptors.device_descriptor_list)
print("[DEBUG] parse_and_store_devices 完成")
print("[DEBUG] 開始 display_devices...")
print("[DEBUG] parse_and_store_devices completed")
print("[DEBUG] Starting display_devices...")
self.display_devices(device_descriptors.device_descriptor_list)
print("[DEBUG] display_devices 完成")
print("[DEBUG] display_devices completed")
return True
else:
print("[DEBUG] 沒有檢測到設備")
print("[DEBUG] No devices detected")
self.main_window.show_no_device_gif()
return False
except Exception as e:
@ -49,7 +79,12 @@ class DeviceController:
return False
def parse_and_store_devices(self, devices):
"""Parse device information and store it"""
"""
Parse device information and store it in connected_devices list.
Args:
devices: List of device descriptors from the scanner.
"""
for device in devices:
try:
product_id = hex(device.product_id).strip().lower()
@ -77,7 +112,12 @@ class DeviceController:
print(f"Error processing device: {e}")
def display_devices(self, devices):
"""Display the connected devices in the UI"""
"""
Display the connected devices in the UI list widget.
Args:
devices: List of device descriptors to display.
"""
try:
if not hasattr(self.main_window, 'device_list_widget'):
print("Warning: main_window does not have device_list_widget attribute")
@ -113,7 +153,12 @@ class DeviceController:
print(f"Error in display_devices: {e}")
def get_devices(self):
"""Get the list of connected devices"""
"""
Get the list of connected devices.
Returns:
list: List of device descriptors, or empty list if no devices found.
"""
try:
device_descriptors = check_available_device()
if device_descriptors.device_descriptor_number > 0:
@ -126,33 +171,50 @@ class DeviceController:
return []
def get_selected_device(self):
"""Get the currently selected device"""
"""
Get the currently selected device.
Returns:
The selected device object, or None if no device is selected.
"""
return self.selected_device
def select_device(self, device, list_item, list_widget):
"""選擇設備（不自動連接和載入 firmware）"""
self.selected_device = device
print("選擇 dongle:", device)
"""
Select a device (does not automatically connect or load firmware).
# 更新列表項目的視覺選擇
Args:
device: The device object to select.
list_item: The QListWidgetItem representing the device.
list_widget: The QListWidget containing the devices.
"""
self.selected_device = device
print("Selected dongle:", device)
# Update visual selection for list items
for index in range(list_widget.count()):
item = list_widget.item(index)
widget = list_widget.itemWidget(item)
if widget: # 檢查 widget 是否存在再設定樣式
if widget: # Check if widget exists before setting style
widget.setStyleSheet("background: none;")
list_item_widget = list_widget.itemWidget(list_item)
if list_item_widget: # 檢查 widget 是否存在再設定樣式
if list_item_widget: # Check if widget exists before setting style
list_item_widget.setStyleSheet("background-color: lightblue;")
def connect_device(self):
"""連接選定的設備並上傳固件"""
"""
Connect to the selected device and upload firmware.
Returns:
bool: True if connection successful, False otherwise.
"""
if not self.selected_device:
print("未選擇設備，無法連接")
print("No device selected, cannot connect")
return False
try:
# 取得 USB port ID
# Get USB port ID
if isinstance(self.selected_device, dict):
usb_port_id = self.selected_device.get("usb_port_id", 0)
product_id = self.selected_device.get("product_id", 0)
@ -160,7 +222,7 @@ class DeviceController:
usb_port_id = getattr(self.selected_device, "usb_port_id", 0)
product_id = getattr(self.selected_device, "product_id", 0)
# 將 product_id 轉換為小寫十六進制字串
# Convert product_id to lowercase hex string
if isinstance(product_id, int):
product_id = hex(product_id).lower()
elif isinstance(product_id, str) and not product_id.startswith('0x'):
@ -169,42 +231,37 @@ class DeviceController:
except ValueError:
pass
# 對應 product_id 到 dongle 類型
# Map product_id to dongle type
dongle = DongleModelMap.get(product_id, "unknown")
print(f"連接設備: product_id={product_id}, mapped to={dongle}")
print(f"Connecting device: product_id={product_id}, mapped to={dongle}")
# 設置固件路徑
# Set firmware paths
scpu_path = os.path.join(FW_DIR, dongle, "fw_scpu.bin")
ncpu_path = os.path.join(FW_DIR, dongle, "fw_ncpu.bin")
# 確認固件文件是否存在
# Check if firmware files exist
if not os.path.exists(scpu_path) or not os.path.exists(ncpu_path):
print(f"固件文件不存在: {scpu_path} 或 {ncpu_path}")
print(f"Firmware files not found: {scpu_path} or {ncpu_path}")
return False
# 連接設備
print('[連接設備]')
# Connect to device
print('[Connecting device]')
self.device_group = kp.core.connect_devices(usb_port_ids=[usb_port_id])
print(' - 連接成功')
print(' - Connection successful')
# # 設置超時
# print('[設置超時]')
# kp.core.set_timeout(device_group=self.device_group, milliseconds=10000)
# print(' - 設置成功')
# 上傳固件
print('[上傳固件]')
# Upload firmware
print('[Uploading firmware]')
kp.core.load_firmware_from_file(
device_group=self.device_group,
scpu_fw_path=scpu_path,
ncpu_fw_path=ncpu_path
)
print(' - 上傳成功')
print(' - Upload successful')
return True
except Exception as e:
print(f"連接設備時發生錯誤: {e}")
# 發生錯誤時嘗試清理
print(f"Error connecting to device: {e}")
# Try to clean up on error
if self.device_group:
try:
kp.core.disconnect_devices(device_group=self.device_group)
@ -214,20 +271,30 @@ class DeviceController:
return False
def disconnect_device(self):
"""中斷與設備的連接"""
"""
Disconnect from the currently connected device.
Returns:
bool: True if disconnection successful, False otherwise.
"""
if self.device_group:
try:
print('[中斷設備連接]')
print('[Disconnecting device]')
kp.core.disconnect_devices(device_group=self.device_group)
print(' - 已中斷連接')
print(' - Disconnected')
self.device_group = None
return True
except Exception as e:
print(f"中斷設備連接時發生錯誤: {e}")
print(f"Error disconnecting device: {e}")
self.device_group = None
return False
return True # 如果沒有連接的設備，視為成功
return True # If no connected device, treat as success
def get_device_group(self):
"""獲取已連接的設備群組"""
"""
Get the connected device group.
Returns:
The Kneron Plus device group object, or None if not connected.
"""
return self.device_group

181
src/controllers/inference_controller.py
View File

@ -1,15 +1,44 @@
# src/controllers/inference_controller.py
import os, queue, cv2, json
"""
inference_controller.py - Inference Controller
This module handles AI model inference operations including tool selection,
model loading, and frame processing for the Kneron AI Playground.
"""
import os
import queue
import cv2
import json
from PyQt5.QtWidgets import QMessageBox, QApplication
from PyQt5.QtCore import QTimer, Qt
import kp # 新增 kp 模組的引入
import kp
from src.models.inference_worker import InferenceWorkerThread
from src.models.custom_inference_worker import CustomInferenceWorkerThread
from src.config import UTILS_DIR, FW_DIR, DongleModelMap
class InferenceController:
"""
Controller class for managing AI model inference operations.
Attributes:
main_window: Reference to the main application window
device_controller: Reference to the device controller
inference_worker: Thread worker for running inference
inference_queue: Queue for frames to be processed
current_tool_config: Current AI tool configuration
model_descriptor: Loaded model descriptor
"""
def __init__(self, main_window, device_controller):
"""
Initialize the InferenceController.
Args:
main_window: Reference to the main application window.
device_controller: Reference to the device controller.
"""
self.main_window = main_window
self.device_controller = device_controller
self.inference_worker = None
@ -17,22 +46,30 @@ class InferenceController:
self.current_tool_config = None
self.previous_tool_config = None
self._camera_was_active = False
# 儲存原始影格尺寸，用於邊界框縮放計算
self.original_frame_width = 640 # 預設值
self.original_frame_height = 480 # 預設值
# Store original frame dimensions for bounding box scaling
self.original_frame_width = 640 # Default value
self.original_frame_height = 480 # Default value
self.model_descriptor = None
def select_tool(self, tool_config):
"""選擇AI工具並配置推論"""
"""
Select an AI tool and configure inference.
Args:
tool_config (dict): Configuration dictionary for the AI tool.
Returns:
bool: True if tool selection successful, False otherwise.
"""
try:
print("選擇工具:", tool_config.get("display_name"))
print("Selecting tool:", tool_config.get("display_name"))
self.current_tool_config = tool_config
# 獲取模式和模型名稱
# Get mode and model name
mode = tool_config.get("mode", "")
model_name = tool_config.get("model_name", "")
# 載入詳細模型配置
# Load detailed model configuration
model_path = os.path.join(UTILS_DIR, mode, model_name)
model_config_path = os.path.join(model_path, "config.json")
@ -42,31 +79,31 @@ class InferenceController:
detailed_config = json.load(f)
tool_config = {**tool_config, **detailed_config}
except Exception as e:
print(f"讀取模型配置時發生錯誤: {e}")
print(f"Error reading model config: {e}")
# 獲取工具輸入類型
# Get tool input type
input_info = tool_config.get("input_info", {})
tool_type = input_info.get("type", "video")
once_mode = True if tool_type == "image" else False
# 檢查是否從視訊模式切換到圖片模式，或從圖片模式切換到視訊模式
# Check if switching from video mode to image mode or vice versa
previous_tool_type = "video"
if hasattr(self, 'previous_tool_config') and self.previous_tool_config:
previous_input_info = self.previous_tool_config.get("input_info", {})
previous_tool_type = previous_input_info.get("type", "video")
# 清空推論佇列，確保在模式切換時不會使用舊數據
# Clear inference queue to avoid using old data when switching modes
self._clear_inference_queue()
# 儲存當前工具類型以供下次比較
# Store current tool type for next comparison
self.previous_tool_config = tool_config
# 準備輸入參數
# Prepare input parameters
input_params = tool_config.get("input_parameters", {}).copy()
# 取得連接的設備群組
# Get connected device group
device_group = self.device_controller.get_device_group()
# 新增設備群組到輸入參數
# Add device group to input parameters
input_params["device_group"] = device_group
# Configure device-related settings
@ -108,7 +145,7 @@ class InferenceController:
msgBox.exec_()
return False
# 僅添加固件路徑作為參考，實際不再使用 (因為裝置已經在選擇時連接)
# Add firmware paths as reference only (device already connected during selection)
scpu_path = os.path.join(FW_DIR, dongle, "fw_scpu.bin")
ncpu_path = os.path.join(FW_DIR, dongle, "fw_ncpu.bin")
input_params["scpu_path"] = scpu_path
@ -147,20 +184,20 @@ class InferenceController:
model_file_path = os.path.join(model_path, model_file)
input_params["model"] = model_file_path
# 上傳模型 (新增)
# Upload model to device
if device_group:
try:
print('[上傳模型]')
print('[Uploading model]')
self.model_descriptor = kp.core.load_model_from_file(
device_group=device_group,
file_path=model_file_path
)
print(' - 上傳成功')
print(' - Upload successful')
# 將模型描述符添加到輸入參數
# Add model descriptor to input parameters
input_params["model_descriptor"] = self.model_descriptor
except Exception as e:
print(f"上傳模型時發生錯誤: {e}")
print(f"Error uploading model: {e}")
self.model_descriptor = None
print("Input parameters:", input_params)
@ -210,110 +247,124 @@ class InferenceController:
return True
except Exception as e:
print(f"選擇工具時發生錯誤: {e}")
print(f"Error selecting tool: {e}")
import traceback
print(traceback.format_exc())
return False
def _clear_inference_queue(self):
"""清空推論佇列中的所有數據"""
"""Clear all data from the inference queue."""
try:
# 清空現有佇列
# Clear existing queue
while not self.inference_queue.empty():
try:
self.inference_queue.get_nowait()
except queue.Empty:
break
print("推論佇列已清空")
print("Inference queue cleared")
except Exception as e:
print(f"清空推論佇列時發生錯誤: {e}")
print(f"Error clearing inference queue: {e}")
def add_frame_to_queue(self, frame):
"""將影格添加到推論佇列"""
"""
Add a frame to the inference queue.
Args:
frame: The image frame to add (numpy array).
"""
try:
# 更新原始影格尺寸
# Update original frame dimensions
if frame is not None and hasattr(frame, 'shape'):
height, width = frame.shape[:2]
self.original_frame_width = width
self.original_frame_height = height
# 添加到佇列
# Add to queue
if not self.inference_queue.full():
self.inference_queue.put(frame)
except Exception as e:
print(f"添加影格到佇列時發生錯誤: {e}")
print(f"Error adding frame to queue: {e}")
import traceback
print(traceback.format_exc())
def stop_inference(self):
"""Stop the inference worker"""
"""Stop the inference worker thread."""
if self.inference_worker:
self.inference_worker.stop()
self.inference_worker = None
def process_uploaded_image(self, file_path):
"""處理上傳的圖片並進行推論"""
"""
Process an uploaded image and run inference.
Args:
file_path (str): Path to the uploaded image file.
"""
try:
if not os.path.exists(file_path):
print(f"錯誤: 檔案不存在 {file_path}")
print(f"Error: File does not exist {file_path}")
return
# 清空推論佇列，確保只處理最新的圖片
# Clear inference queue to ensure only the latest image is processed
self._clear_inference_queue()
# 讀取圖片
# Read image
img = cv2.imread(file_path)
if img is None:
print(f"錯誤: 無法讀取圖片 {file_path}")
print(f"Error: Unable to read image {file_path}")
return
# 更新推論工作器參數
# Update inference worker parameters
if self.inference_worker:
self.inference_worker.input_params["file_path"] = file_path
# 將圖片添加到推論佇列
# Add image to inference queue
if not self.inference_queue.full():
self.inference_queue.put(img)
print(f"已將圖片 {file_path} 添加到推論佇列")
print(f"Added image {file_path} to inference queue")
else:
print("警告: 推論佇列已滿")
print("Warning: Inference queue is full")
else:
print("錯誤: 推論工作器未初始化")
print("Error: Inference worker not initialized")
except Exception as e:
print(f"處理上傳圖片時發生錯誤: {e}")
print(f"Error processing uploaded image: {e}")
import traceback
print(traceback.format_exc())
def select_custom_tool(self, tool_config):
"""
選擇自訂模型工具並配置推論
Select a custom model tool and configure inference.
Args:
tool_config: 包含自訂模型配置的字典，必須包含:
- custom_model_path: .nef 模型檔案路徑
- custom_scpu_path: SCPU firmware 路徑
- custom_ncpu_path: NCPU firmware 路徑
tool_config (dict): Configuration dictionary containing:
- custom_model_path: Path to .nef model file
- custom_scpu_path: Path to SCPU firmware
- custom_ncpu_path: Path to NCPU firmware
- custom_labels: Optional list of class labels
Returns:
bool: True if custom tool selection successful, False otherwise.
"""
try:
print("選擇自訂模型:", tool_config.get("display_name"))
print("Selecting custom model:", tool_config.get("display_name"))
self.current_tool_config = tool_config
# 清空推論佇列
# Clear inference queue
self._clear_inference_queue()
# 儲存當前工具類型以供下次比較
# Store current tool type for next comparison
self.previous_tool_config = tool_config
# 準備輸入參數
# Prepare input parameters
input_params = tool_config.get("input_parameters", {}).copy()
# 添加自訂模型路徑
# Add custom model paths
input_params["custom_model_path"] = tool_config.get("custom_model_path")
input_params["custom_scpu_path"] = tool_config.get("custom_scpu_path")
input_params["custom_ncpu_path"] = tool_config.get("custom_ncpu_path")
input_params["custom_labels"] = tool_config.get("custom_labels")
# 取得設備相關設定
# Get device-related settings
selected_device = self.device_controller.get_selected_device()
if selected_device:
if isinstance(selected_device, dict):
@ -324,19 +375,19 @@ class InferenceController:
devices = self.device_controller.connected_devices
if devices and len(devices) > 0:
input_params["usb_port_id"] = devices[0].get("usb_port_id", 0)
print("警告: 未選擇特定設備，使用第一個可用設備")
print("Warning: No device selected, using first available device")
else:
input_params["usb_port_id"] = 0
print("警告: 沒有已連接的設備，使用預設 usb_port_id 0")
print("Warning: No connected devices, using default usb_port_id 0")
print("自訂模型輸入參數:", input_params)
print("Custom model input parameters:", input_params)
# 停止現有的推論工作器
# Stop existing inference worker
if self.inference_worker:
self.inference_worker.stop()
self.inference_worker = None
# 創建新的自訂推論工作器
# Create new custom inference worker
self.inference_worker = CustomInferenceWorkerThread(
self.inference_queue,
min_interval=0.5,
@ -347,23 +398,23 @@ class InferenceController:
self.main_window.handle_inference_result
)
self.inference_worker.start()
print("自訂模型推論工作器已啟動")
print("Custom model inference worker started")
# 啟動相機 (自訂模型預設使用視訊模式)
# Start camera (custom model defaults to video mode)
tool_type = tool_config.get("input_info", {}).get("type", "video")
if tool_type == "video":
if self._camera_was_active and self.main_window.media_controller.video_thread is not None:
self.main_window.media_controller.video_thread.change_pixmap_signal.connect(
self.main_window.media_controller.update_image
)
print("相機已重新連接用於視訊處理")
print("Camera reconnected for video processing")
else:
self.main_window.media_controller.start_camera()
return True
except Exception as e:
print(f"選擇自訂模型時發生錯誤: {e}")
print(f"Error selecting custom model: {e}")
import traceback
print(traceback.format_exc())
return False

124
src/controllers/media_controller.py
View File

@ -1,3 +1,10 @@
"""
media_controller.py - Media Controller
This module handles media operations including camera capture, video/audio
recording, screenshot capture, and image display with bounding box overlay.
"""
import cv2
import os
from PyQt5.QtWidgets import QFileDialog
@ -7,8 +14,28 @@ from PyQt5.QtCore import Qt, QRect
from src.models.video_thread import VideoThread
from src.utils.image_utils import qimage_to_numpy
class MediaController:
"""
Controller class for managing media operations.
Attributes:
main_window: Reference to the main application window
inference_controller: Reference to the inference controller
video_thread: Thread for camera video capture
recording: Flag indicating if video recording is active
recording_audio: Flag indicating if audio recording is active
recorded_frames: List of recorded video frames
"""
def __init__(self, main_window, inference_controller):
"""
Initialize the MediaController.
Args:
main_window: Reference to the main application window.
inference_controller: Reference to the inference controller.
"""
self.main_window = main_window
self.inference_controller = inference_controller
self.video_thread = None
@ -16,14 +43,18 @@ class MediaController:
self.recording_audio = False
self.recorded_frames = []
self._signal_was_connected = False # Track if signal was previously connected
self._inference_paused = False # 追蹤推論是否暫停
self._inference_paused = False # Track if inference is paused
def start_camera(self):
"""啟動相機進行視訊擷取"""
"""
Start the camera for video capture.
Initializes the video thread and connects the signal for frame updates.
"""
try:
if self.video_thread is None:
print("初始化相機執行緒...")
# 先清除畫布上的任何文字或圖像
print("Initializing camera thread...")
# Clear any text or image on the canvas
if hasattr(self.main_window, 'canvas_label'):
self.main_window.canvas_label.clear()
@ -32,94 +63,103 @@ class MediaController:
try:
self.video_thread.change_pixmap_signal.connect(self.update_image)
self._signal_was_connected = True
print("相機信號連接成功")
print("Camera signal connected successfully")
except Exception as e:
print(f"連接相機信號時發生錯誤: {e}")
print(f"Error connecting camera signal: {e}")
# 啟動相機執行緒
# Start camera thread
self.video_thread.start()
print("相機執行緒啟動成功")
print("Camera thread started successfully")
else:
print("相機已經在運行中")
print("Camera is already running")
except Exception as e:
print(f"啟動相機時發生錯誤: {e}")
print(f"Error starting camera: {e}")
import traceback
print(traceback.format_exc())
def stop_camera(self):
"""停止相機"""
"""
Stop the camera and release resources.
Disconnects signal connections and stops the video thread.
"""
try:
if self.video_thread is not None:
print("停止相機執行緒")
# 確保先斷開信號連接
print("Stopping camera thread")
# Disconnect signal connection first
if self._signal_was_connected:
try:
self.video_thread.change_pixmap_signal.disconnect()
self._signal_was_connected = False
print("已斷開相機信號連接")
print("Camera signal disconnected")
except Exception as e:
print(f"斷開信號連接時發生錯誤: {e}")
print(f"Error disconnecting signal: {e}")
# 停止執行緒
# Stop thread
self.video_thread.stop()
self.video_thread = None
print("相機已完全停止")
print("Camera completely stopped")
except Exception as e:
print(f"停止相機時發生錯誤: {e}")
print(f"Error stopping camera: {e}")
import traceback
print(traceback.format_exc())
def update_image(self, qt_image):
"""更新圖像顯示並處理推論"""
"""
Update the image display and process inference.
Args:
qt_image (QImage): The image frame from the camera.
"""
try:
# 更新畫布上的圖像
# Update image on canvas
if hasattr(self.main_window, 'canvas_label'):
pixmap = QPixmap.fromImage(qt_image)
# 如果有邊界框，繪製它們
# If there are bounding boxes, draw them
if hasattr(self.main_window, 'current_bounding_boxes') and self.main_window.current_bounding_boxes is not None:
painter = QPainter(pixmap)
pen = QPen(Qt.red)
pen.setWidth(2)
painter.setPen(pen)
# 遍歷並繪製所有邊界框
# Iterate and draw all bounding boxes
for bbox_info in self.main_window.current_bounding_boxes:
# 確保邊界框資訊是合法的
# Ensure bounding box info is valid
if isinstance(bbox_info, list) and len(bbox_info) >= 4:
# 繪製矩形
# Draw rectangle
x1, y1, x2, y2 = bbox_info[0], bbox_info[1], bbox_info[2], bbox_info[3]
painter.drawRect(QRect(x1, y1, x2 - x1, y2 - y1))
# 如果有標籤（邊界框的第5個元素），繪製它
# If there's a label (5th element of bounding box), draw it
if len(bbox_info) > 4 and bbox_info[4]:
font = QFont()
font.setPointSize(10)
painter.setFont(font)
painter.setPen(QColor(255, 0, 0)) # 紅色
painter.setPen(QColor(255, 0, 0)) # Red color
# 計算標籤位置（邊界框上方）
# Calculate label position (above bounding box)
label_x = x1
label_y = y1 - 10
# 確保標籤在畫布範圍內
# Ensure label is within canvas bounds
if label_y < 10:
label_y = y2 + 15 # 如果上方空間不足，放在底部
label_y = y2 + 15 # If not enough space above, place below
painter.drawText(label_x, label_y, str(bbox_info[4]))
painter.end()
# 顯示圖像
# Display image
self.main_window.canvas_label.setPixmap(pixmap)
# 只有在推論未暫停時才將影格添加到推論佇列
# Only add frame to inference queue if inference is not paused
if not self._inference_paused:
frame_np = qimage_to_numpy(qt_image)
self.inference_controller.add_frame_to_queue(frame_np)
except Exception as e:
print(f"更新圖像時發生錯誤: {e}")
print(f"Error updating image: {e}")
import traceback
print(traceback.format_exc())
@ -256,23 +296,33 @@ class MediaController:
print(f"Error taking screenshot: {e}")
def toggle_inference_pause(self):
"""切換推論暫停狀態"""
"""
Toggle the inference pause state.
Returns:
bool: The new pause state (True if paused, False if running).
"""
try:
self._inference_paused = not self._inference_paused
if self._inference_paused:
# 暫停時清除邊界框
# Clear bounding boxes when paused
self.main_window.current_bounding_boxes = None
else:
# 恢復推論時，確保相機仍在運行
# When resuming inference, ensure camera is still running
if self.video_thread is None or not self.video_thread.isRunning():
self.start_camera()
return self._inference_paused
except Exception as e:
print(f"切換推論暫停狀態時發生錯誤: {e}")
print(f"Error toggling inference pause state: {e}")
import traceback
print(traceback.format_exc())
return self._inference_paused
def is_inference_paused(self):
"""檢查推論是否暫停"""
"""
Check if inference is paused.
Returns:
bool: True if inference is paused, False otherwise.
"""
return self._inference_paused

0
src/controllers/model_controller.py
View File

147
src/models/custom_inference_worker.py
View File

@ -1,7 +1,8 @@
"""
Custom Inference Worker
使用使用者上傳的自訂模型進行推論
前後處理使用 script.py 中定義的 YOLO V5 處理邏輯
custom_inference_worker.py - Custom Inference Worker
This module provides a worker thread for running inference using user-uploaded
custom models. It uses YOLO V5 pre/post-processing logic for object detection.
"""
import os
import time
@ -15,7 +16,7 @@ import kp
from kp.KPBaseClass.ValueBase import ValueRepresentBase
# COCO 數據集的類別名稱
# COCO dataset class names (80 classes)
COCO_CLASSES = [
'person', 'bicycle', 'car', 'motorcycle', 'airplane', 'bus', 'train', 'truck', 'boat',
'traffic light', 'fire hydrant', 'stop sign', 'parking meter', 'bench', 'bird', 'cat',
@ -80,7 +81,7 @@ class ExampleYoloResult(ValueRepresentBase):
return member_variable_dict
# YOLO 常數
# YOLO constants
YOLO_V3_CELL_BOX_NUM = 3
NMS_THRESH_YOLOV5 = 0.5
YOLO_MAX_DETECTION_PER_CLASS = 100
@ -246,26 +247,27 @@ def post_process_yolo_v5(inference_float_node_output_list: List[kp.InferenceFloa
def preprocess_frame(frame, target_size=640):
"""
預處理影像
Preprocess image frame for YOLO inference.
Args:
frame: 原始 BGR 影像
target_size: 目標大小 (default 640 for YOLO)
frame: Original BGR image (numpy array).
target_size (int): Target size for resizing (default 640 for YOLO).
Returns:
processed_frame: 處理後的影像 (BGR565 格式)
original_width: 原始寬度
original_height: 原始高度
tuple: (processed_frame in BGR565 format, original_width, original_height)
Raises:
Exception: If input frame is None.
"""
if frame is None:
raise Exception("輸入的 frame 為 None")
raise Exception("Input frame is None")
original_height, original_width = frame.shape[:2]
# 調整大小
# Resize to target size
resized_frame = cv2.resize(frame, (target_size, target_size))
# 轉換為 BGR565 格式
# Convert to BGR565 format
frame_bgr565 = cv2.cvtColor(resized_frame, cv2.COLOR_BGR2BGR565)
return frame_bgr565, original_width, original_height
@ -273,18 +275,18 @@ def preprocess_frame(frame, target_size=640):
def postprocess(output_list, hw_preproc_info, original_width, original_height, target_size=640, thresh=0.2):
"""
後處理 YOLO 輸出
Post-process YOLO model output.
Args:
output_list: 模型輸出節點列表
hw_preproc_info: 硬體預處理資訊
original_width: 原始影像寬度
original_height: 原始影像高度
target_size: 縮放目標大小
thresh: 閾值
output_list: List of model output nodes.
hw_preproc_info: Hardware preprocessing info from Kneron device.
original_width (int): Original image width.
original_height (int): Original image height.
target_size (int): Resize target size used during preprocessing.
thresh (float): Detection confidence threshold.
Returns:
yolo_result: YOLO 偵測結果
ExampleYoloResult: YOLO detection results with bounding boxes.
"""
yolo_result = post_process_yolo_v5(
inference_float_node_output_list=output_list,
@ -292,7 +294,7 @@ def postprocess(output_list, hw_preproc_info, original_width, original_height, t
thresh_value=thresh
)
# 調整邊界框座標以符合原始尺寸
# Adjust bounding box coordinates to match original dimensions
width_ratio = original_width / target_size
height_ratio = original_height / target_size
@ -307,12 +309,29 @@ def postprocess(output_list, hw_preproc_info, original_width, original_height, t
class CustomInferenceWorkerThread(QThread):
"""
自訂模型推論工作線程
使用使用者上傳的模型和韌體進行推論
Custom model inference worker thread.
Uses user-uploaded model and firmware to perform inference on video frames.
Attributes:
inference_result_signal: Signal emitted with inference results
frame_queue: Queue containing frames to process
device_group: Connected Kneron device group
model_descriptor: Loaded model descriptor
custom_labels: Custom class labels (optional)
"""
inference_result_signal = pyqtSignal(object)
def __init__(self, frame_queue, min_interval=0.5, mse_threshold=500):
"""
Initialize the CustomInferenceWorkerThread.
Args:
frame_queue: Queue containing frames to process.
min_interval (float): Minimum seconds between inferences.
mse_threshold (float): MSE threshold for detecting frame changes.
"""
super().__init__()
self.frame_queue = frame_queue
self.min_interval = min_interval
@ -323,63 +342,76 @@ class CustomInferenceWorkerThread(QThread):
self.cached_result = None
self.input_params = {}
# 設備和模型相關
# Device and model related
self.device_group = None
self.model_descriptor = None
self.is_initialized = False
# 自訂標籤
# Custom labels
self.custom_labels = None
def initialize_device(self):
"""初始化設備、上傳韌體和模型"""
"""
Initialize device, upload firmware and model.
Returns:
bool: True if initialization successful, False otherwise.
"""
try:
model_path = self.input_params.get("custom_model_path")
scpu_path = self.input_params.get("custom_scpu_path")
ncpu_path = self.input_params.get("custom_ncpu_path")
port_id = self.input_params.get("usb_port_id", 0)
# 載入自訂標籤
# Load custom labels
self.custom_labels = self.input_params.get("custom_labels")
if self.custom_labels:
print(f'[自訂標籤] 已載入 {len(self.custom_labels)} 個類別')
print(f'[Custom Labels] Loaded {len(self.custom_labels)} classes')
else:
print('[自訂標籤] 未提供，使用預設 COCO 類別')
print('[Custom Labels] Not provided, using default COCO classes')
if not all([model_path, scpu_path, ncpu_path]):
print("缺少必要的檔案路徑")
print("Missing required file paths")
return False
# 連接設備
print('[連接裝置]')
# Connect to device
print('[Connecting device]')
self.device_group = kp.core.connect_devices(usb_port_ids=[port_id])
kp.core.set_timeout(device_group=self.device_group, milliseconds=5000)
print(' - 連接成功')
print(' - Connection successful')
# 上傳韌體
print('[上傳韌體]')
# Upload firmware
print('[Uploading firmware]')
kp.core.load_firmware_from_file(self.device_group, scpu_path, ncpu_path)
print(' - 韌體上傳成功')
print(' - Firmware upload successful')
# 上傳模型
print('[上傳模型]')
# Upload model
print('[Uploading model]')
self.model_descriptor = kp.core.load_model_from_file(
self.device_group,
file_path=model_path
)
print(' - 模型上傳成功')
print(' - Model upload successful')
self.is_initialized = True
return True
except Exception as e:
print(f"初始化設備時發生錯誤: {e}")
print(f"Error initializing device: {e}")
import traceback
print(traceback.format_exc())
return False
def run_single_inference(self, frame):
"""執行單次推論"""
"""
Execute a single inference on the given frame.
Args:
frame: Input image frame (numpy array in BGR format).
Returns:
dict: Inference results with bounding boxes and labels, or None on error.
"""
try:
if not self.is_initialized:
if not self.initialize_device():
@ -425,12 +457,12 @@ class CustomInferenceWorkerThread(QThread):
original_height
)
# 轉換為標準格式
# Convert to standard format
bounding_boxes = [
[box.x1, box.y1, box.x2, box.y2] for box in yolo_result.box_list
]
# 使用自訂標籤或預設 COCO 類別
# Use custom labels or default COCO classes
labels_to_use = self.custom_labels if self.custom_labels else COCO_CLASSES
results = []
@ -447,13 +479,18 @@ class CustomInferenceWorkerThread(QThread):
}
except Exception as e:
print(f"推論時發生錯誤: {e}")
print(f"Error during inference: {e}")
import traceback
print(traceback.format_exc())
return None
def run(self):
"""主執行循環"""
"""
Main execution loop.
Continuously processes frames from the queue, runs inference,
and emits results. Uses MSE-based frame change detection.
"""
while self._running:
try:
frame = self.frame_queue.get(timeout=0.1)
@ -464,7 +501,7 @@ class CustomInferenceWorkerThread(QThread):
if current_time - self.last_inference_time < self.min_interval:
continue
# MSE 檢測以優化效能
# MSE detection to optimize performance
if self.last_frame is not None:
if frame.shape != self.last_frame.shape:
self.last_frame = None
@ -476,11 +513,11 @@ class CustomInferenceWorkerThread(QThread):
self.inference_result_signal.emit(self.cached_result)
continue
except Exception as e:
print(f"計算 MSE 時發生錯誤: {e}")
print(f"Error calculating MSE: {e}")
self.last_frame = None
self.cached_result = None
# 執行推論
# Execute inference
result = self.run_single_inference(frame)
self.last_inference_time = current_time
@ -490,22 +527,22 @@ class CustomInferenceWorkerThread(QThread):
if result is not None:
self.inference_result_signal.emit(result)
# 斷開設備連接
# Disconnect device
self.cleanup()
self.quit()
def cleanup(self):
"""清理資源"""
"""Clean up resources and disconnect device."""
try:
if self.device_group is not None:
kp.core.disconnect_devices(self.device_group)
print('[已斷開裝置]')
print('[Device disconnected]')
self.device_group = None
except Exception as e:
print(f"清理資源時發生錯誤: {e}")
print(f"Error cleaning up resources: {e}")
def stop(self):
"""停止工作線程"""
"""Stop the worker thread and clean up resources."""
self._running = False
self.wait()
self.cleanup()

77
src/models/inference_worker.py
View File

@ -1,9 +1,30 @@
import os, time, queue, numpy as np, importlib.util
"""
inference_worker.py - Inference Worker Thread
This module provides a QThread-based worker for running AI model inference
on video frames. It supports dynamic module loading and frame caching.
"""
import os
import time
import queue
import numpy as np
import importlib.util
from PyQt5.QtCore import QThread, pyqtSignal
from src.config import UTILS_DIR
def load_inference_module(mode, model_name):
"""Dynamically load an inference module"""
"""
Dynamically load an inference module from the utils directory.
Args:
mode (str): The inference mode/category (e.g., 'object_detection').
model_name (str): The name of the model.
Returns:
module: The loaded Python module containing the inference function.
"""
script_path = os.path.join(UTILS_DIR, mode, model_name, "script.py")
module_name = f"{mode}_{model_name}"
spec = importlib.util.spec_from_file_location(module_name, script_path)
@ -11,10 +32,38 @@ def load_inference_module(mode, model_name):
spec.loader.exec_module(module)
return module
class InferenceWorkerThread(QThread):
"""
Worker thread for running AI model inference on frames.
This thread processes frames from a queue, runs inference using a
dynamically loaded module, and emits results via Qt signals.
Attributes:
inference_result_signal: Signal emitted with inference results
frame_queue: Queue containing frames to process
mode: Inference mode/category
model_name: Name of the model to use
min_interval: Minimum time between inferences (seconds)
mse_threshold: MSE threshold for frame change detection
once_mode: If True, stop after one inference
"""
inference_result_signal = pyqtSignal(object)
def __init__(self, frame_queue, mode, model_name, min_interval=0.5, mse_threshold=500, once_mode=False):
"""
Initialize the InferenceWorkerThread.
Args:
frame_queue: Queue containing frames to process.
mode (str): Inference mode/category.
model_name (str): Name of the model.
min_interval (float): Minimum seconds between inferences.
mse_threshold (float): MSE threshold for detecting frame changes.
once_mode (bool): If True, stop after processing one frame.
"""
super().__init__()
self.frame_queue = frame_queue
self.mode = mode
@ -32,6 +81,13 @@ class InferenceWorkerThread(QThread):
self.inference_module = load_inference_module(mode, model_name)
def run(self):
"""
Main thread execution loop.
Continuously processes frames from the queue, runs inference,
and emits results. Uses MSE-based frame change detection to
optimize performance by skipping similar frames.
"""
while self._running:
try:
frame = self.frame_queue.get(timeout=0.1)
@ -43,18 +99,18 @@ class InferenceWorkerThread(QThread):
continue
if self.last_frame is not None:
# 檢查當前幀與上一幀的尺寸是否相同
# Check if current frame and previous frame have same dimensions
if frame.shape != self.last_frame.shape:
print(f"幀尺寸變更: 從 {self.last_frame.shape} 變更為 {frame.shape}")
# 尺寸不同時，重置上一幀和緩存結果
print(f"Frame size changed: from {self.last_frame.shape} to {frame.shape}")
# Reset last frame and cached result when dimensions differ
self.last_frame = None
self.cached_result = None
else:
# 只有在尺寸相同時才進行 MSE 計算
# Only calculate MSE when dimensions are the same
try:
mse = np.mean((frame.astype(np.float32) - self.last_frame.astype(np.float32)) ** 2)
if mse < self.mse_threshold and self.cached_result is not None:
# 只有在結果不為 None 時才發送信號
# Only emit signal if result is not None
if self.cached_result is not None:
self.inference_result_signal.emit(self.cached_result)
if self.once_mode:
@ -62,8 +118,8 @@ class InferenceWorkerThread(QThread):
break
continue
except Exception as e:
print(f"計算 MSE 時發生錯誤: {e}")
# 發生錯誤時重置上一幀和緩存結果
print(f"Error calculating MSE: {e}")
# Reset last frame and cached result on error
self.last_frame = None
self.cached_result = None
@ -77,7 +133,7 @@ class InferenceWorkerThread(QThread):
self.last_frame = frame.copy()
self.cached_result = result
# 只有在結果不為 None 時才發送信號
# Only emit signal if result is not None
if result is not None:
self.inference_result_signal.emit(result)
@ -88,5 +144,6 @@ class InferenceWorkerThread(QThread):
self.quit()
def stop(self):
"""Stop the inference worker thread and wait for it to finish."""
self._running = False
self.wait()

91
src/models/video_thread.py
View File

@ -1,22 +1,51 @@
"""
video_thread.py - Video Capture Thread
This module provides a QThread-based video capture worker for webcam streaming.
It handles camera connection, frame capture, and automatic reconnection.
"""
import cv2
import time
import threading
from PyQt5.QtCore import QThread, pyqtSignal
from PyQt5.QtGui import QImage
class VideoThread(QThread):
"""
Thread for capturing video frames from a webcam.
Emits frames as QImage objects via Qt signals for display in the UI.
Handles camera connection with timeout and automatic reconnection.
Signals:
change_pixmap_signal: Emitted with each new frame (QImage)
camera_error_signal: Emitted when camera errors occur (str)
"""
change_pixmap_signal = pyqtSignal(QImage)
camera_error_signal = pyqtSignal(str) # 新增：相機錯誤信號
camera_error_signal = pyqtSignal(str) # Camera error signal
def __init__(self):
"""Initialize the VideoThread with default settings."""
super().__init__()
self._run_flag = True
self._camera_open_attempts = 0
self._max_attempts = 3
self._camera_timeout = 5 # 相機開啟超時秒數
self._camera_timeout = 5 # Camera open timeout in seconds
def _open_camera_with_timeout(self, camera_index, backend=None):
"""使用超時機制開啟相機"""
"""
Open camera with timeout mechanism.
Args:
camera_index (int): Index of the camera to open.
backend: OpenCV video capture backend (e.g., cv2.CAP_DSHOW).
Returns:
cv2.VideoCapture: Opened camera object, or None if failed.
"""
cap = None
open_success = False
@ -29,17 +58,17 @@ class VideoThread(QThread):
cap = cv2.VideoCapture(camera_index)
open_success = cap is not None and cap.isOpened()
except Exception as e:
print(f"開啟相機時發生異常: {e}")
print(f"Exception while opening camera: {e}")
open_success = False
# 在單獨的線程中嘗試開啟相機
# Try to open camera in a separate thread
thread = threading.Thread(target=try_open)
thread.daemon = True
thread.start()
thread.join(timeout=self._camera_timeout)
if thread.is_alive():
print(f"相機開啟超時 ({self._camera_timeout}秒)")
print(f"Camera open timeout ({self._camera_timeout} seconds)")
return None
if open_success:
@ -50,73 +79,79 @@ class VideoThread(QThread):
return None
def run(self):
# 嘗試多次開啟相機
"""
Main thread execution loop.
Attempts to open the camera with multiple backends and retries.
Captures frames and emits them as QImage signals.
"""
# Try multiple times to open camera
while self._camera_open_attempts < self._max_attempts and self._run_flag:
self._camera_open_attempts += 1
print(f"嘗試開啟相機 (嘗試 {self._camera_open_attempts}/{self._max_attempts})...")
print(f"Attempting to open camera (attempt {self._camera_open_attempts}/{self._max_attempts})...")
# 嘗試使用DirectShow後端，通常在Windows上更快
# Try DirectShow backend first (usually faster on Windows)
cap = self._open_camera_with_timeout(0, cv2.CAP_DSHOW)
if cap is None:
print("無法使用DirectShow開啟相機，嘗試預設後端")
print("Unable to open camera with DirectShow, trying default backend")
cap = self._open_camera_with_timeout(0)
if cap is None:
print(f"無法使用任何後端開啟相機，等待1秒後重試...")
print("Unable to open camera with any backend, retrying in 1 second...")
time.sleep(1)
continue
# 設置相機屬性以獲得更好的性能
# 降低解析度以提高啟動速度和幀率
# Set camera properties for better performance
# Lower resolution for faster startup and higher frame rate
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
cap.set(cv2.CAP_PROP_FPS, 30)
# 設置緩衝區大小為1，減少延遲
# Set buffer size to 1 to reduce latency
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
# 預熱相機，丟棄前幾幀以加快穩定速度
# Warm up camera by discarding first few frames
for _ in range(5):
cap.read()
# 相機開啟成功，重置嘗試計數
# Camera opened successfully, reset attempt count
self._camera_open_attempts = 0
# 主循環
# Main capture loop
while self._run_flag:
ret, frame = cap.read()
if ret:
# 轉換為RGB格式
# Convert to RGB format
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
height, width, channel = frame.shape
bytes_per_line = channel * width
qt_image = QImage(frame.data, width, height, bytes_per_line, QImage.Format_RGB888)
self.change_pixmap_signal.emit(qt_image)
else:
print("無法讀取相機幀，相機可能已斷開連接")
print("Unable to read camera frame, camera may be disconnected")
break
# 釋放相機資源
# Release camera resources
cap.release()
# 如果是因為停止信號而退出循環，則不再重試
# If stopped by stop signal, don't retry
if not self._run_flag:
break
print("相機連接中斷，嘗試重新連接...")
print("Camera connection lost, attempting to reconnect...")
if self._camera_open_attempts >= self._max_attempts:
print("達到最大嘗試次數，無法開啟相機")
print("Maximum attempts reached, unable to open camera")
def stop(self):
"""停止執行緒"""
"""Stop the video capture thread."""
try:
print("正在停止相機執行緒...")
print("Stopping camera thread...")
self._run_flag = False
# 等待執行緒完成
# Wait for thread to finish
if self.isRunning():
self.wait()
print("相機執行緒已停止")
print("Camera thread stopped")
except Exception as e:
print(f"停止相機執行緒時發生錯誤: {e}")
print(f"Error stopping camera thread: {e}")
import traceback
print(traceback.format_exc())

96
src/services/device_service.py
View File

@ -1,20 +1,37 @@
"""
device_service.py - Device Service
This module provides device scanning functionality with timeout mechanism
for detecting connected Kneron devices.
"""
import kp
import threading
class EmptyDescriptor:
"""
Empty device descriptor placeholder.
Used when no devices are found or when device scanning fails.
"""
def __init__(self):
"""Initialize with empty device list."""
self.device_descriptor_number = 0
self.device_descriptor_list = []
def check_available_device(timeout=0.5):
"""
掃描可用設備，帶有超時機制
Scan for available Kneron devices with timeout mechanism.
Args:
timeout: 超時秒數，預設 5 秒
timeout (float): Timeout in seconds (default 0.5).
Returns:
設備描述符
Device descriptor object containing found devices,
or EmptyDescriptor if no devices found or scan failed.
"""
result = [None]
error = [None]
@ -26,32 +43,32 @@ def check_available_device(timeout=0.5):
error[0] = e
try:
print("[SCAN] 開始掃描設備...")
print("[SCAN] Starting device scan...")
# 在單獨的線程中執行掃描
# Execute scan in a separate thread
thread = threading.Thread(target=scan_devices)
thread.daemon = True
print("[SCAN] 啟動掃描線程...")
print("[SCAN] Starting scan thread...")
thread.start()
print(f"[SCAN] 等待掃描完成 (超時: {timeout}秒)...")
print(f"[SCAN] Waiting for scan to complete (timeout: {timeout}s)...")
thread.join(timeout=timeout)
if thread.is_alive():
print(f"[SCAN] 設備掃描超時 ({timeout}秒)")
print(f"[SCAN] Device scan timeout ({timeout}s)")
return EmptyDescriptor()
print("[SCAN] 掃描線程已完成")
print("[SCAN] Scan thread completed")
if error[0]:
print(f"[SCAN] Error scanning devices: {error[0]}")
return EmptyDescriptor()
if result[0] is None:
print("[SCAN] 結果為 None")
print("[SCAN] Result is None")
return EmptyDescriptor()
print("[SCAN] device_descriptors:", result[0])
print("[SCAN] 準備返回結果...")
print("[SCAN] Preparing to return results...")
import sys
sys.stdout.flush()
return result[0]
@ -59,60 +76,3 @@ def check_available_device(timeout=0.5):
except Exception as e:
print(f"[SCAN] Error scanning devices: {e}")
return EmptyDescriptor()
# def check_available_device():
# # 模擬設備描述符
# print("checking available devices")
# class EmptyDescriptor:
# def __init__(self):
# self.device_descriptor_number = 0
# self.device_descriptor_list = [{
# "usb_port_id": 4,
# "vendor_id": "0x3231",
# "product_id": "0x720",
# "link_speed": "UsbSpeed.KP_USB_SPEED_SUPER",
# "kn_number": "0xB306224C",
# "is_connectable": True,
# "usb_port_path": "4-1",
# "firmware": "KDP2 Comp/F"
# },
# {
# "usb_port_id": 5,
# "vendor_id": "0x3231",
# "product_id": "0x520",
# "link_speed": "UsbSpeed.KP_USB_SPEED_SUPER",
# "kn_number": "0xB306224C",
# "is_connectable": True,
# "usb_port_path": "4-1",
# "firmware": "KDP2 Comp/F"
# }]
# return EmptyDescriptor()
# device_descriptors = [
# {
# "usb_port_id": 4,
# "vendor_id": "0x3231",
# "product_id": "0x720",
# "link_speed": "UsbSpeed.KP_USB_SPEED_SUPER",
# "kn_number": "0xB306224C",
# "is_connectable": True,
# "usb_port_path": "4-1",
# "firmware": "KDP2 Comp/F"
# },
# {
# "usb_port_id": 5,
# "vendor_id": "0x3231",
# "product_id": "0x520",
# "link_speed": "UsbSpeed.KP_USB_SPEED_SUPER",
# "kn_number": "0xB306224C",
# "is_connectable": True,
# "usb_port_path": "4-1",
# "firmware": "KDP2 Comp/F"
# }
# ]
# return device_descriptors
# def get_dongle_type(self, product_id):
# for dongle_type in self.K_:
# if dongle_type.value == product_id:
# return dongle_type
# return None

121
src/services/file_service.py
View File

@ -1,3 +1,10 @@
"""
file_service.py - File Service
This module provides file upload functionality including image file handling
and integration with the inference system.
"""
import os
import shutil
from PyQt5.QtWidgets import QFileDialog, QMessageBox, QApplication
@ -5,111 +12,139 @@ from PyQt5.QtCore import Qt
from PyQt5.QtGui import QImage, QPixmap
import cv2
class FileService:
"""
Service class for handling file uploads.
Manages file selection, copying, and integration with the camera
and inference systems.
Attributes:
main_window: Reference to the main application window
upload_dir: Directory path for storing uploaded files
destination: Path to the most recently uploaded file
"""
def __init__(self, main_window, upload_dir):
"""
Initialize the FileService.
Args:
main_window: Reference to the main application window.
upload_dir (str): Directory path for storing uploaded files.
"""
self.main_window = main_window
self.upload_dir = upload_dir
self.destination = None
self._camera_was_active = False # Track if camera was active before upload
def upload_file(self):
"""處理檔案上傳流程"""
"""
Handle the file upload process.
Opens a file dialog, copies the selected file to the upload directory,
and triggers inference processing if a tool is configured.
Returns:
str: Path to the uploaded file, or None if upload failed/cancelled.
"""
try:
# 1. 先完全停止相機（如果正在運行）
# 1. Pause camera if running
if hasattr(self.main_window, 'media_controller') and self.main_window.media_controller.video_thread is not None:
print("上傳前停止相機")
print("Stopping camera before upload")
try:
# 儲存狀態以指示相機正在運行
# Save state to indicate camera was running
self._camera_was_active = True
# 顯示上傳中提示
# Show upload preparation message
if hasattr(self.main_window, 'canvas_label'):
self.main_window.canvas_label.setText("準備上傳檔案...")
self.main_window.canvas_label.setText("Preparing to upload file...")
self.main_window.canvas_label.setAlignment(Qt.AlignCenter)
self.main_window.canvas_label.setStyleSheet("color: white; font-size: 24px;")
# 確保 UI 更新
# Ensure UI updates
QApplication.processEvents()
# 只暫停推論，不完全停止相機
# Only pause inference, don't completely stop camera
if not self.main_window.media_controller._inference_paused:
self.main_window.media_controller.toggle_inference_pause()
# 斷開信號連接但不停止相機執行緒
# Disconnect signal but don't stop camera thread
if hasattr(self.main_window.media_controller, '_signal_was_connected') and self.main_window.media_controller._signal_was_connected:
try:
self.main_window.media_controller.video_thread.change_pixmap_signal.disconnect()
self.main_window.media_controller._signal_was_connected = False
print("已暫時斷開相機信號連接")
print("Temporarily disconnected camera signal")
except Exception as e:
print(f"斷開信號連接時發生錯誤: {e}")
print(f"Error disconnecting signal: {e}")
except Exception as e:
print(f"準備上傳時發生錯誤: {e}")
print(f"Error preparing for upload: {e}")
import traceback
print(traceback.format_exc())
# 即使發生錯誤，也繼續嘗試上傳
# Continue trying to upload even if error occurs
else:
self._camera_was_active = False
print("呼叫 QFileDialog.getOpenFileName")
print("Calling QFileDialog.getOpenFileName")
options = QFileDialog.Options()
file_path, _ = QFileDialog.getOpenFileName(
self.main_window,
"上傳檔案",
"Upload File",
"",
"所有檔案 (*)",
"All Files (*)",
options=options
)
print("檔案路徑取得:", file_path)
print("File path obtained:", file_path)
if file_path:
print("檢查上傳目錄是否存在")
print("Checking if upload directory exists")
if not os.path.exists(self.upload_dir):
os.makedirs(self.upload_dir)
print(f"建立上傳目錄: {self.upload_dir}")
print(f"Created upload directory: {self.upload_dir}")
print("檢查原始檔案是否存在:", file_path)
print("Checking if source file exists:", file_path)
if not os.path.exists(file_path):
self.show_message(QMessageBox.Critical, "錯誤", "選擇的檔案不存在")
self.show_message(QMessageBox.Critical, "Error", "Selected file does not exist")
return None
file_name = os.path.basename(file_path)
self.destination = os.path.join(self.upload_dir, file_name)
print("目標路徑:", self.destination)
print("Destination path:", self.destination)
# 檢查目標路徑是否可寫入
# Check if destination is writable
try:
print("測試檔案寫入權限")
print("Testing file write permissions")
with open(self.destination, 'wb') as test_file:
pass
os.remove(self.destination)
print("測試檔案建立和刪除成功")
print("Test file creation and deletion successful")
except PermissionError:
self.show_message(QMessageBox.Critical, "錯誤", "無法寫入目標目錄")
self.show_message(QMessageBox.Critical, "Error", "Cannot write to destination directory")
return None
print("開始檔案複製")
print("Starting file copy")
try:
shutil.copy2(file_path, self.destination)
print("檔案複製成功")
print("File copy successful")
# 更新主視窗目的地
# Update main window destination
self.main_window.destination = self.destination
print(f"更新主視窗目的地: {self.main_window.destination}")
print(f"Updated main window destination: {self.main_window.destination}")
# 處理上傳的影像
# Process uploaded image
if self.main_window.inference_controller.current_tool_config:
print("使用推論控制器處理上傳的影像")
# 先在畫布上顯示影像
print("Processing uploaded image with inference controller")
# First display image on canvas
try:
# 載入和顯示影像
# Load and display image
image = cv2.imread(self.destination)
if image is not None:
# 轉換為 RGB 顯示
# Convert to RGB for display
image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
height, width, channel = image_rgb.shape
bytes_per_line = channel * width
qt_image = QImage(image_rgb.data, width, height, bytes_per_line, QImage.Format_RGB888)
# 將影像縮放以適應畫布
# Scale image to fit canvas
canvas_size = self.main_window.canvas_label.size()
scaled_image = qt_image.scaled(
int(canvas_size.width() * 0.95),
@ -118,28 +153,28 @@ class FileService:
Qt.SmoothTransformation
)
self.main_window.canvas_label.setPixmap(QPixmap.fromImage(scaled_image))
print("影像顯示在畫布上")
print("Image displayed on canvas")
except Exception as e:
print(f"顯示影像時發生錯誤: {e}")
print(f"Error displaying image: {e}")
import traceback
print(traceback.format_exc())
# 然後使用推論處理它
# Then process it with inference
self.main_window.inference_controller.process_uploaded_image(self.destination)
return self.destination
except Exception as e:
import traceback
print("檔案複製過程中發生錯誤:\n", traceback.format_exc())
self.show_message(QMessageBox.Critical, "錯誤", f"上傳錯誤: {str(e)}")
print("Error during file copy:\n", traceback.format_exc())
self.show_message(QMessageBox.Critical, "Error", f"Upload error: {str(e)}")
return None
return None
except Exception as e:
import traceback
print("上傳過程中發生錯誤:\n", traceback.format_exc())
self.show_message(QMessageBox.Critical, "錯誤", f"上傳錯誤: {str(e)}")
print("Error during upload:\n", traceback.format_exc())
self.show_message(QMessageBox.Critical, "Error", f"Upload error: {str(e)}")
return None
finally:
# 如果相機之前是活動的，嘗試恢復相機連接

14
src/services/model_service.py
View File

@ -1,14 +0,0 @@
import kp
import cv2, os, shutil, sys
from PyQt5.QtWidgets import (QApplication, QWidget, QVBoxLayout, QLabel, QPushButton,
QComboBox, QFileDialog, QMessageBox, QHBoxLayout, QDialog, QListWidget,
QScrollArea, QFrame, QListWidgetItem, QTextEdit)
from PyQt5.QtSvg import QSvgWidget
from PyQt5.QtMultimedia import QCamera, QCameraImageCapture, QCameraInfo, QMediaRecorder, QAudioRecorder
from PyQt5.QtMultimediaWidgets import QVideoWidget
from PyQt5.QtGui import QPixmap, QMovie
from PyQt5.QtCore import Qt, QTimer, QUrl
from ..config import UXUI_ASSETS, WINDOW_SIZE, BACKGROUND_COLOR
def show_error_popup(self, message):
error_dialog = QMessageBox.critical(self, "Error", message)

95
src/views/components/device_popup.py
View File

@ -87,69 +87,6 @@ def create_device_popup(parent, device_controller):
# Store reference to this list widget for later use
parent.device_list_widget_popup = device_list_widget_popup
# Comment out the device details section
"""
# Device details section (initially hidden)
device_details = QFrame(popup)
device_details.setObjectName("device-details")
device_details.setProperty("class", "device-info")
device_details.setVisible(False) # Initially hidden
details_layout = QGridLayout(device_details)
details_layout.setColumnStretch(1, 1)
# 增加行間距
details_layout.setVerticalSpacing(2)
# Device Type
device_type_label_title = QLabel("Device Type:")
device_type_label_title.setStyleSheet("font-size: 14px;")
details_layout.addWidget(device_type_label_title, 0, 0)
device_type_label = QLabel("-")
device_type_label.setObjectName("device-type")
device_type_label.setStyleSheet("font-size: 14px;")
details_layout.addWidget(device_type_label, 0, 1)
# Port ID
port_id_label_title = QLabel("Port ID:")
port_id_label_title.setStyleSheet("font-size: 14px;")
details_layout.addWidget(port_id_label_title, 1, 0)
port_id_label = QLabel("-")
port_id_label.setObjectName("port-id")
port_id_label.setStyleSheet("font-size: 14px;")
details_layout.addWidget(port_id_label, 1, 1)
# KN Number
kn_number_label_title = QLabel("KN Number:")
kn_number_label_title.setStyleSheet("font-size: 14px;")
details_layout.addWidget(kn_number_label_title, 2, 0)
kn_number_label = QLabel("-")
kn_number_label.setObjectName("kn-number")
kn_number_label.setStyleSheet("font-size: 14px;")
details_layout.addWidget(kn_number_label, 2, 1)
# Status
status_label_title = QLabel("Status:")
status_label_title.setStyleSheet("font-size: 14px;")
details_layout.addWidget(status_label_title, 3, 0)
status_label = QLabel("-")
status_label.setObjectName("status")
status_label.setStyleSheet("font-size: 14px;")
details_layout.addWidget(status_label, 3, 1)
# Store references to labels
parent.device_detail_labels = {
"device-type": device_type_label,
"port-id": port_id_label,
"kn-number": kn_number_label,
"status": status_label,
"frame": device_details
}
# Connect item selection to show details
device_list_widget_popup.itemClicked.connect(lambda item: show_device_details(parent, item))
list_layout.addWidget(device_details)
"""
popup_layout.addWidget(list_section)
# Button area
@ -178,31 +115,6 @@ def create_device_popup(parent, device_controller):
print(f"Error in create_device_popup: {e}")
return QWidget(parent)
# Also need to comment out the show_device_details function since it's no longer used
# def show_device_details(parent, item):
# """Show details for the selected device"""
# try:
# # Get the device data from the item
# device_data = item.data(Qt.UserRole)
# print("device_data", device_data)
# if not device_data:
# return
# # Update the detail labels
# labels = parent.device_detail_labels
# # Set values
# labels["device-type"].setText(device_data.get("dongle", "-"))
# labels["port-id"].setText(str(device_data.get("usb_port_id", "-")))
# labels["kn-number"].setText(str(device_data.get("kn_number", "-")))
# labels["status"].setText("Connected" if device_data.get("is_connectable", True) else "Not Available")
# # Show the details frame
# labels["frame"].setVisible(True)
# except Exception as e:
# print(f"Error showing device details: {e}")
def refresh_devices(parent, device_controller):
"""Refresh the device list and update the UI"""
@ -215,13 +127,6 @@ def refresh_devices(parent, device_controller):
# Clear the list
parent.device_list_widget_popup.clear()
# Comment out the details frame visibility setting
"""
# Hide details frame
if hasattr(parent, "device_detail_labels") and "frame" in parent.device_detail_labels:
parent.device_detail_labels["frame"].setVisible(False)
"""
# Track unique device models to avoid duplicates
seen_models = set()

70
src/views/device_connection_popup.py
View File

@ -1,70 +0,0 @@
# srcs/views/device_connection_popup.py
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QListWidget, QListWidgetItem
from PyQt5.QtSvg import QSvgWidget
from PyQt5.QtCore import Qt
from ..config import UXUI_ASSETS, SECONDARY_COLOR, BUTTON_STYLE
import os
class DeviceConnectionPopup(QWidget):
def __init__(self, parent=None):
super().__init__(parent)
self.parent = parent
self.init_ui()
def init_ui(self):
# 設定彈出視窗的大小和樣式
self.setFixedSize(int(self.parent.width() * 0.67), int(self.parent.height() * 0.67))
self.setStyleSheet(f"""
QWidget {{
background-color: {SECONDARY_COLOR};
border-radius: 20px;
padding: 20px;
}}
""")
layout = QVBoxLayout(self)
# 標題列
title_layout = QHBoxLayout()
title_layout.setAlignment(Qt.AlignCenter)
# 容器來放置圖示和標籤
title_container = QWidget()
container_layout = QHBoxLayout(title_container)
container_layout.setSpacing(10)
# 添加圖示
device_icon = QSvgWidget(os.path.join(UXUI_ASSETS, "Assets_svg/ic_window_device.svg"))
device_icon.setFixedSize(35, 35)
container_layout.addWidget(device_icon)
# 添加標籤
popup_label = QLabel("Device Connection")
popup_label.setStyleSheet("color: white; font-size: 25px;")
container_layout.addWidget(popup_label)
# 設置容器的對齊方式
container_layout.setAlignment(Qt.AlignCenter)
# 將容器添加到標題布局中
title_layout.addWidget(title_container)
layout.addLayout(title_layout)
# 設備列表
self.device_list_widget = QListWidget(self)
layout.addWidget(self.device_list_widget)
# 按鈕區域
button_layout = QHBoxLayout()
refresh_button = QPushButton("Refresh")
refresh_button.clicked.connect(self.parent.refresh_devices)
refresh_button.setStyleSheet(BUTTON_STYLE)
button_layout.addWidget(refresh_button)
done_button = QPushButton("Done")
done_button.setStyleSheet(BUTTON_STYLE)
done_button.clicked.connect(self.parent.hide_device_popup)
button_layout.addWidget(done_button)
layout.addLayout(button_layout)

68
src/views/login_screen.py
View File

@ -1,23 +1,68 @@
from PyQt5.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel,
QLineEdit, QComboBox, QFrame, QMessageBox)
"""
login_screen.py - User Login Screen
This module contains the LoginScreen class which provides the authentication
interface for accessing the utilities section of the application.
"""
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel,
QLineEdit, QComboBox, QFrame, QMessageBox
)
from PyQt5.QtCore import Qt, pyqtSignal
from PyQt5.QtGui import QPixmap, QFont
import os
from src.config import UXUI_ASSETS, WINDOW_SIZE, BACKGROUND_COLOR
class LoginScreen(QWidget):
"""
Login Screen Class
Provides user authentication interface with:
- Server authentication type selection
- Username and password input fields
- Login and back navigation buttons
Signals:
login_success: Emitted when login is successful
back_to_selection: Emitted when user clicks back button
Attributes:
server_combo (QComboBox): Authentication type dropdown
username_input (QLineEdit): Username input field
password_input (QLineEdit): Password input field (masked)
error_label (QLabel): Error message display label
"""
# Signals for navigation
login_success = pyqtSignal()
back_to_selection = pyqtSignal()
def __init__(self, parent=None):
"""
Initialize the LoginScreen.
Args:
parent: Optional parent widget.
"""
super().__init__(parent)
self.init_ui()
def init_ui(self):
"""
Initialize the user interface.
Creates the login form layout with:
- Header with Kneron logo
- Authentication type selection dropdown
- Username and password input fields
- Login and back buttons
- Footer with copyright information
"""
# Basic window setup
self.setGeometry(100, 100, *WINDOW_SIZE)
self.setStyleSheet(f"background-color: #F5F7FA;") # Light gray background
self.setStyleSheet("background-color: #F5F7FA;")
# Main layout
layout = QVBoxLayout(self)
@ -211,17 +256,30 @@ class LoginScreen(QWidget):
layout.addWidget(footer_label)
def attempt_login(self):
"""
Attempt to log in with the provided credentials.
Validates that both username and password are provided.
In production, this would authenticate against a server.
Emits login_success signal on successful validation.
"""
username = self.username_input.text()
password = self.password_input.text()
# For demo purposes, use a simple validation
# Basic validation - ensure both fields are filled
if not username or not password:
self.show_error("Please enter both username and password")
return
# Simulate login success (in a real app, you would validate with your server)
# Simulate login success (in production, validate with server)
self.login_success.emit()
def show_error(self, message):
"""
Display an error message to the user.
Args:
message (str): The error message to display.
"""
self.error_label.setText(message)
self.error_label.show()

183
src/views/mainWindows.py
View File

@ -1,5 +1,19 @@
import os, sys, json, queue, numpy as np
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QLabel, QGridLayout, QFrame, QMessageBox, QApplication, QListWidgetItem
"""
mainWindows.py - Main Application Window
This module contains the MainWindow class which serves as the primary UI
container for the Kneron Academy AI Playground application.
"""
import os
import sys
import json
import queue
import numpy as np
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QGridLayout,
QFrame, QMessageBox, QApplication, QListWidgetItem
)
from PyQt5.QtCore import Qt, QTimer
from PyQt5.QtGui import QPixmap, QMovie
@ -16,7 +30,27 @@ from src.views.components.media_panel import create_media_panel
from src.views.components.device_popup import create_device_popup, refresh_devices
from src.views.components.custom_model_block import create_custom_model_block
class MainWindow(QWidget):
"""
Main Application Window Class
The primary UI container for the Kneron Academy AI Playground.
Manages the overall layout, device connections, media display, and AI inference.
Attributes:
device_controller (DeviceController): Handles Kneron device connections
inference_controller (InferenceController): Manages AI model inference
media_controller (MediaController): Controls camera and media playback
file_service (FileService): Handles file upload operations
config_utils (ConfigUtils): Manages configuration settings
destination: Current output destination
current_bounding_boxes: Stores current bounding box detection results
canvas_label (QLabel): Display area for camera/media content
device_popup: Device connection popup dialog
overlay: Semi-transparent overlay for popup dialogs
"""
def __init__(self):
super().__init__()
@ -31,13 +65,18 @@ class MainWindow(QWidget):
# Set up UI and configuration
self.destination = None
self.current_bounding_boxes = None # 儲存當前的邊界框資訊（改為複數形式）
self.current_bounding_boxes = None # Stores current bounding box info (plural form)
self.generate_global_config()
self.init_ui()
def init_ui(self):
"""
Initialize the user interface.
Sets up the main window geometry, title, and background color.
Shows a welcome screen first, then transitions to the main page.
"""
try:
# Basic window setup
self.setGeometry(100, 100, *WINDOW_SIZE)
self.setWindowTitle('Kneron Academy AI Playground')
self.setStyleSheet(f"background-color: {BACKGROUND_COLOR};")
@ -56,6 +95,7 @@ class MainWindow(QWidget):
print(f"Error in init_ui: {e}")
def show_welcome_label(self):
"""Display the welcome screen with Kneron logo."""
try:
welcome_label = QLabel(self)
logo_path = os.path.join(UXUI_ASSETS, "Assets_png/kneron_logo.png")
@ -77,8 +117,13 @@ class MainWindow(QWidget):
print(f"Error in show_welcome_label: {e}")
def show_device_popup_and_main_page(self):
"""
Transition from welcome screen to main page.
Clears the welcome screen, creates the main page layout,
sets up the device popup overlay, and auto-starts the camera.
"""
try:
# Clear welcome screen
self.clear_layout()
# 1. Initialize main page
@ -93,25 +138,32 @@ class MainWindow(QWidget):
# 3. Refresh devices - do this after UI is fully set up
QTimer.singleShot(100, self.device_controller.refresh_devices)
# # 4. Show popup
# 4. Show popup
self.show_device_popup()
# 5. 延遲啟動相機，讓 UI 先完全顯示
# 5. Delay camera start to let UI fully display first
QTimer.singleShot(500, self.auto_start_camera)
print("Popup window setup complete")
# # 5. Start camera automatically after a short delay
# QTimer.singleShot(100, self.auto_start_camera)
except Exception as e:
print(f"Error in show_device_popup_and_main_page: {e}")
def create_main_page(self):
"""
Create the main page layout.
Builds a two-column layout:
- Left column: Logo, device list, and custom model block
- Right column: Canvas area for media display and media panel controls
Returns:
QWidget: The main page widget containing all UI components.
"""
try:
main_page = QWidget(self)
main_layout = QHBoxLayout(main_page)
main_page.setLayout(main_layout)
# Left layout - 使用固定寬度的容器，避免滾輪
# Left layout - fixed width container to avoid scrolling
left_container = QWidget()
left_container.setFixedWidth(260)
left_layout = QVBoxLayout(left_container)
@ -131,11 +183,11 @@ class MainWindow(QWidget):
self.device_frame, self.device_list_widget = create_device_layout(self, self.device_controller)
left_layout.addWidget(self.device_frame)
# 使用 Custom Model Block 取代原本的 AI Toolbox
# Use Custom Model Block instead of original AI Toolbox
self.custom_model_frame = create_custom_model_block(self, self.inference_controller)
left_layout.addWidget(self.custom_model_frame)
# 添加彈性空間，確保元件不會被拉伸
# Add stretch to prevent components from being stretched
left_layout.addStretch()
# Right layout
@ -161,6 +213,12 @@ class MainWindow(QWidget):
return QWidget(self)
def device_popup_mask_setup(self):
"""
Set up the device popup overlay and mask.
Creates a semi-transparent dark overlay with a centered device
connection popup dialog.
"""
try:
print("Setting up popup mask")
# Add transparent overlay
@ -184,25 +242,35 @@ class MainWindow(QWidget):
print(f"Error in device_popup_mask_setup: {e}")
def show_device_popup(self):
"""
Show the device connection popup.
Displays the overlay and triggers a delayed device list refresh
to allow the UI to render first.
"""
try:
# 先顯示彈窗
self.overlay.show()
# 延遲刷新設備列表，讓 UI 先顯示
# Delay device list refresh to let UI display first
from src.views.components.device_popup import refresh_devices
QTimer.singleShot(100, lambda: refresh_devices(self, self.device_controller))
except Exception as e:
print(f"Error in show_device_popup: {e}")
def hide_device_popup(self):
"""Hide the device connection popup overlay."""
try:
self.overlay.hide()
except Exception as e:
print(f"Error in hide_device_popup: {e}")
def clear_layout(self):
"""
Clear all widgets from the main layout.
Removes and deletes all child widgets from the layout.
"""
try:
# Clear all widgets
while self.layout.count():
child = self.layout.takeAt(0)
if child.widget():
@ -211,72 +279,82 @@ class MainWindow(QWidget):
print(f"Error in clear_layout: {e}")
def handle_inference_result(self, result):
"""處理來自推論工作器的結果"""
"""
Handle inference results from the inference worker.
Processes the inference result and either:
- Updates bounding boxes for drawing on canvas (for detection results)
- Shows a popup dialog (for classification/other results)
Args:
result: The inference result, can be dict with 'bounding box',
'bounding boxes', or other classification results.
"""
try:
# 如果結果為 None，直接返回不處理
# If result is None, skip processing
if result is None:
print("收到空的推論結果 (None)，略過處理")
print("Received empty inference result (None), skipping")
return
# 輸出結果到控制台
print("收到推論結果:", result)
print("Received inference result:", result)
# 檢查結果是否包含邊界框資訊（支持新舊兩種格式）
# Check if result contains bounding box info (supports both old and new formats)
if isinstance(result, dict) and "bounding box" in result:
# 舊格式兼容: 單一邊界框
# Legacy format compatibility: single bounding box
self.current_bounding_boxes = [result["bounding box"]]
# 如果結果中有標籤，將其添加到邊界框中
# If result contains a label, add it to the bounding box
if "result" in result:
# 確保邊界框列表至少有5個元素
# Ensure bounding box list has at least 5 elements
while len(self.current_bounding_boxes[0]) < 5:
self.current_bounding_boxes[0].append(None)
# 設置第5個元素為結果標籤
# Set the 5th element as the result label
self.current_bounding_boxes[0][4] = result["result"]
# 不需要顯示彈窗，因為邊界框會直接繪製在畫面上
# No popup needed as bounding boxes are drawn directly on canvas
return
# 新格式: 多邊界框
# New format: multiple bounding boxes
elif isinstance(result, dict) and "bounding boxes" in result:
bboxes = result["bounding boxes"]
results = result.get("results", [])
# 確保邊界框是列表格式
# Ensure bounding boxes is in list format
if not isinstance(bboxes, list):
print("錯誤: 'bounding boxes' 必須是列表格式")
print("Error: 'bounding boxes' must be a list")
return
# 如果只有邊界框座標沒有標籤
# If only bounding box coordinates without labels
if not results:
self.current_bounding_boxes = bboxes
else:
# 結合邊界框和標籤
# Combine bounding boxes with labels
self.current_bounding_boxes = []
for i, bbox in enumerate(bboxes):
# 創建包含座標的新列表
# Create new list containing coordinates
new_bbox = list(bbox)
# 添加標籤（如果有）
# Add label if available
if i < len(results):
new_bbox.append(results[i])
else:
new_bbox.append(None)
self.current_bounding_boxes.append(new_bbox)
# 不需要顯示彈窗，因為邊界框會直接繪製在畫面上
# No popup needed as bounding boxes are drawn directly on canvas
return
# 對於非邊界框結果，使用彈窗顯示 (限制每次只顯示一個彈窗)
# 檢查是否有其他消息框正在顯示
# For non-bounding-box results, show popup (limit to one popup at a time)
# Check if another message box is already showing
for widget in QApplication.topLevelWidgets():
if isinstance(widget, QMessageBox) and widget.isVisible():
print("已有彈窗顯示中，跳過此次結果顯示")
print("Popup already showing, skipping this result")
return
# 創建QMessageBox
# Create QMessageBox
msgBox = QMessageBox(self)
msgBox.setWindowTitle("推論結果")
msgBox.setWindowTitle("Inference Result")
# 根據結果類型格式化文字
# Format text based on result type
if isinstance(result, dict):
result_str = "\n".join(f"{key}: {value}" for key, value in result.items())
else:
@ -285,17 +363,15 @@ class MainWindow(QWidget):
msgBox.setText(result_str)
msgBox.setStandardButtons(QMessageBox.Ok)
# 設置樣式
msgBox.setStyleSheet("""
QLabel {
color: white;
}
""")
# 顯示彈窗
msgBox.exec_()
except Exception as e:
print(f"處理推論結果時發生錯誤: {e}")
print(f"Error handling inference result: {e}")
import traceback
print(traceback.format_exc())
@ -337,20 +413,25 @@ class MainWindow(QWidget):
return self.config_utils.generate_global_config()
def auto_start_camera(self):
"""自動啟動相機並顯示預覽"""
print("開始啟動相機...")
"""
Automatically start the camera and display preview.
Shows a loading message on the canvas while the camera initializes,
then starts the camera in a separate thread to avoid blocking the UI.
"""
print("Starting camera...")
try:
# 先清除畫布，顯示加載中的信息
# Clear canvas and show loading message
if hasattr(self, 'canvas_label'):
self.canvas_label.setText("相機啟動中...")
self.canvas_label.setText("Starting camera...")
self.canvas_label.setAlignment(Qt.AlignCenter)
self.canvas_label.setStyleSheet("color: white; font-size: 24px;")
# 確保UI更新
# Ensure UI updates
QApplication.processEvents()
# 在單獨的線程中啟動相機，避免阻塞UI
# Start camera in separate thread to avoid blocking UI
self.media_controller.start_camera()
except Exception as e:
print(f"啟動相機時發生錯誤: {e}")
print(f"Error starting camera: {e}")
import traceback
print(traceback.format_exc())

36
src/views/selection_screen.py
View File

@ -1,22 +1,56 @@
"""
selection_screen.py - Application Selection Screen
This module contains the SelectionScreen class which provides the initial
screen where users can choose between the Utilities or Demo App functions.
"""
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QFrame
from PyQt5.QtCore import Qt, pyqtSignal
from PyQt5.QtGui import QPixmap, QFont, QIcon
import os
from src.config import UXUI_ASSETS, WINDOW_SIZE, BACKGROUND_COLOR, APP_NAME
class SelectionScreen(QWidget):
"""
Selection Screen Class
The initial screen displayed to users, allowing them to choose between:
- Device Utilities: For managing and configuring Kneron devices
- Demo App: For testing AI inference capabilities
Signals:
open_utilities: Emitted when user selects the utilities option
open_demo_app: Emitted when user selects the demo app option
"""
# Signals for navigation
open_utilities = pyqtSignal()
open_demo_app = pyqtSignal()
def __init__(self, parent=None):
"""
Initialize the SelectionScreen.
Args:
parent: Optional parent widget.
"""
super().__init__(parent)
self.init_ui()
def init_ui(self):
"""
Initialize the user interface.
Creates a card-based layout with:
- Header containing the Kneron logo
- Content area with title and two selection cards
- Footer with copyright information
"""
# Basic window setup
self.setGeometry(100, 100, *WINDOW_SIZE)
self.setStyleSheet(f"background-color: #F8F9FA;") # Slightly lighter background
self.setStyleSheet("background-color: #F8F9FA;")
# Main layout
layout = QVBoxLayout(self)

356
src/views/utilities_screen.py
View File

@ -1,6 +1,16 @@
from PyQt5.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel,
"""
utilities_screen.py - Device Utilities Screen
This module contains the UtilitiesScreen class which provides device management
functionality including device scanning, firmware updates, driver installation,
and purchased items management.
"""
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel,
QFrame, QMessageBox, QScrollArea, QTableWidget, QTableWidgetItem,
QHeaderView, QProgressBar, QLineEdit, QAbstractItemView)
QHeaderView, QProgressBar, QLineEdit, QAbstractItemView
)
from PyQt5.QtCore import Qt, pyqtSignal, QTimer
from PyQt5.QtGui import QPixmap, QFont, QIcon, QColor
import os
@ -10,20 +20,53 @@ from src.controllers.device_controller import DeviceController
from src.services.device_service import check_available_device
from ..config import FW_DIR
class UtilitiesScreen(QWidget):
"""
Utilities Screen Class
Provides device management functionality with two main pages:
1. Utilities Page: Device connection, firmware updates, driver installation
2. Purchased Items Page: Download and manage purchased AI models
Signals:
back_to_selection: Emitted when user clicks back button
Attributes:
device_controller (DeviceController): Controller for device operations
current_page (str): Current page being displayed ("utilities" or "purchased_items")
device_table (QTableWidget): Table displaying connected devices
purchased_table (QTableWidget): Table displaying purchased items
progress_bar (QProgressBar): Progress bar for operations
status_label (QLabel): Status message display
"""
# Signals for navigation
back_to_selection = pyqtSignal()
def __init__(self, parent=None):
"""
Initialize the UtilitiesScreen.
Args:
parent: Optional parent widget.
"""
super().__init__(parent)
self.device_controller = DeviceController(self)
self.current_page = "utilities" # 追蹤當前頁面: "utilities" 或 "purchased_items"
self.current_page = "utilities" # Track current page: "utilities" or "purchased_items"
self.init_ui()
def init_ui(self):
"""
Initialize the user interface.
Creates the main layout with:
- Header with navigation buttons and logo
- Content container with switchable pages (utilities and purchased items)
"""
# Basic window setup
self.setGeometry(100, 100, *WINDOW_SIZE)
self.setStyleSheet(f"background-color: #F5F7FA;") # Light gray background
self.setStyleSheet("background-color: #F5F7FA;")
# Main layout
self.main_layout = QVBoxLayout(self)
@ -34,7 +77,7 @@ class UtilitiesScreen(QWidget):
header_frame = self.create_header()
self.main_layout.addWidget(header_frame)
# 創建主要內容容器
# Create main content container
self.content_container = QFrame(self)
self.content_container.setStyleSheet("""
QFrame {
@ -47,32 +90,43 @@ class UtilitiesScreen(QWidget):
content_layout.setContentsMargins(20, 20, 20, 20)
content_layout.setSpacing(20)
# 創建兩個頁面的容器
# Create containers for both pages
self.utilities_page = QWidget()
self.purchased_items_page = QWidget()
# 設置 utilities 頁面
# Set up utilities page
self.setup_utilities_page()
# 設置 purchased items 頁面
# Set up purchased items page
self.setup_purchased_items_page()
# 添加頁面到內容容器
# Add pages to content container
content_layout.addWidget(self.utilities_page)
content_layout.addWidget(self.purchased_items_page)
# 初始顯示 utilities 頁面
# Initially show utilities page
self.utilities_page.show()
self.purchased_items_page.hide()
# 添加內容容器到主佈局
# Add content container to main layout
self.main_layout.addWidget(self.content_container, 1)
# Initialize with device refresh (暫時禁用自動刷新，避免阻塞)
# Note: Auto-refresh disabled to prevent blocking
# QTimer.singleShot(500, self.refresh_devices)
def create_header(self):
"""Create the header with back button and logo"""
"""
Create the header section with navigation elements.
Builds a header frame containing:
- Back button for returning to selection screen
- Title label showing current page
- Navigation buttons for switching between Utilities and Purchased Items pages
- Kneron logo on the right side
Returns:
QFrame: The header frame widget.
"""
header_frame = QFrame(self)
header_frame.setStyleSheet("background-color: #2C3E50; border-radius: 0px;")
header_frame.setFixedHeight(60)
@ -167,7 +221,13 @@ class UtilitiesScreen(QWidget):
return header_frame
def setup_utilities_page(self):
"""Create the utilities page"""
"""
Set up the utilities page layout.
Creates the utilities page with:
- Device connection section for managing connected devices
- Status section showing operation progress and device status
"""
utilities_layout = QVBoxLayout(self.utilities_page)
utilities_layout.setContentsMargins(0, 0, 0, 0)
utilities_layout.setSpacing(20)
@ -181,17 +241,27 @@ class UtilitiesScreen(QWidget):
utilities_layout.addWidget(status_section)
def setup_purchased_items_page(self):
"""Create the purchased items page"""
"""
Set up the purchased items page layout.
Creates the page for displaying and downloading purchased AI models
and packages from the Kneron store.
"""
purchased_items_layout = QVBoxLayout(self.purchased_items_page)
purchased_items_layout.setContentsMargins(0, 0, 0, 0)
purchased_items_layout.setSpacing(20)
# 已購買項目區域
# Purchased items section
purchased_items_section = self.create_purchased_items_section()
purchased_items_layout.addWidget(purchased_items_section)
def create_purchased_items_section(self):
"""創建已購買項目區域"""
"""
Create the purchased items section.
Returns:
QFrame: A frame containing the purchased items table and action buttons.
"""
purchased_section = QFrame()
purchased_section.setStyleSheet("""
QFrame {
@ -205,36 +275,35 @@ class UtilitiesScreen(QWidget):
purchased_layout.setContentsMargins(15, 15, 15, 15)
purchased_layout.setSpacing(15)
# 標題
# Title
title_label = QLabel("Your Purchased Items", purchased_section)
title_label.setStyleSheet("font-size: 18px; font-weight: bold; color: #2C3E50;")
purchased_layout.addWidget(title_label)
# 描述
# Description
desc_label = QLabel("Select items to download to your device", purchased_section)
desc_label.setStyleSheet("font-size: 14px; color: #7F8C8D;")
purchased_layout.addWidget(desc_label)
# 項目表格
# Items table (5 columns, "Action" column removed)
self.purchased_table = QTableWidget()
# 修改為只有5列，移除 "Action" 列
self.purchased_table.setColumnCount(5)
self.purchased_table.setHorizontalHeaderLabels([
"Select", "Product", "Model", "Current Version", "Compatible Dongles"
])
# 設置行寬
# Set column widths
header = self.purchased_table.horizontalHeader()
header.setSectionResizeMode(0, QHeaderView.ResizeToContents) # 勾選框列
header.setSectionResizeMode(0, QHeaderView.ResizeToContents) # Checkbox column
header.setSectionResizeMode(1, QHeaderView.Stretch)
header.setSectionResizeMode(2, QHeaderView.Stretch)
header.setSectionResizeMode(3, QHeaderView.ResizeToContents)
header.setSectionResizeMode(4, QHeaderView.Stretch)
# 設置表格高度
# Set table height
self.purchased_table.setMinimumHeight(300)
# 啟用整行選擇
# Enable row selection
self.purchased_table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.purchased_table.setStyleSheet("""
@ -263,10 +332,10 @@ class UtilitiesScreen(QWidget):
""")
purchased_layout.addWidget(self.purchased_table)
# 添加一些模擬數據
# Add mock data for demonstration
self.populate_mock_purchased_items()
# 下載按鈕
# Download buttons
button_layout = QHBoxLayout()
button_layout.setSpacing(10)
@ -319,11 +388,17 @@ class UtilitiesScreen(QWidget):
return purchased_section
def populate_mock_purchased_items(self):
"""填充模擬的已購買項目數據"""
# 清空表格
"""
Populate the purchased items table with mock data.
Clears the existing table and fills it with sample purchased items
for demonstration purposes. In production, this would fetch real
data from the server.
"""
# Clear the table
self.purchased_table.setRowCount(0)
# 模擬數據
# Mock data for demonstration
mock_items = [
{
"product": "KL720 AI Package",
@ -357,11 +432,11 @@ class UtilitiesScreen(QWidget):
}
]
# 添加數據到表格
# Add data to table
for i, item in enumerate(mock_items):
self.purchased_table.insertRow(i)
# 創建勾選框
# Create checkbox item
checkbox_item = QTableWidgetItem()
checkbox_item.setFlags(Qt.ItemIsUserCheckable | Qt.ItemIsEnabled | Qt.ItemIsSelectable)
checkbox_item.setCheckState(Qt.Unchecked)
@ -373,31 +448,47 @@ class UtilitiesScreen(QWidget):
self.purchased_table.setItem(i, 4, QTableWidgetItem(item["dongles"]))
def download_item(self, row):
"""下載特定項目"""
"""
Download a specific item from the purchased items list.
Args:
row (int): The row index of the item to download.
"""
product = self.purchased_table.item(row, 1).text()
model = self.purchased_table.item(row, 2).text()
# 顯示進度條
# Show progress bar
self.show_progress(f"Downloading {product} - {model}...", 0)
# 模擬下載過程
# Simulate download process
for i in range(1, 11):
progress = i * 10
QTimer.singleShot(i * 300, lambda p=progress: self.update_progress(p))
# 完成下載
# Complete download
QTimer.singleShot(3000, lambda: self.handle_download_complete(product, model))
def handle_download_complete(self, product, model):
"""處理下載完成"""
"""
Handle download completion for a single item.
Args:
product (str): The product name that was downloaded.
model (str): The model name that was downloaded.
"""
self.hide_progress()
QMessageBox.information(self, "Download Complete", f"{product} - {model} has been downloaded successfully!")
def download_selected_items(self):
"""下載所有勾選的項目"""
"""
Download all selected/checked items from the purchased items list.
Iterates through the table to find checked items and initiates
download for each. Shows progress bar during the download process.
"""
selected_rows = set()
# 檢查勾選的項目
# Check for selected items
for row in range(self.purchased_table.rowCount()):
if self.purchased_table.item(row, 0).checkState() == Qt.Checked:
selected_rows.add(row)
@ -406,38 +497,57 @@ class UtilitiesScreen(QWidget):
QMessageBox.warning(self, "No Selection", "Please select at least one item to download")
return
# 顯示進度條
# Show progress bar
self.show_progress(f"Downloading {len(selected_rows)} items...", 0)
# 模擬下載過程
# Simulate download process
total_items = len(selected_rows)
for i, row in enumerate(selected_rows):
product = self.purchased_table.item(row, 1).text()
model = self.purchased_table.item(row, 2).text()
progress = int((i / total_items) * 100)
# 更新進度條
# Update progress bar
self.update_progress(progress)
self.progress_title.setText(f"Downloading {product} - {model}... ({i+1}/{total_items})")
# 模擬下載延遲
# Simulate download delay
QTimer.singleShot((i+1) * 1000, lambda p=product, m=model: self.status_label.setText(f"Downloaded {p} - {m}"))
# 完成所有下載
# Complete all downloads
QTimer.singleShot((total_items+1) * 1000, self.handle_all_downloads_complete)
def update_download_progress(self, progress, message):
"""更新下載進度"""
"""
Update the download progress display.
Args:
progress (int): The progress percentage (0-100).
message (str): The status message to display.
"""
self.update_progress(progress)
self.progress_title.setText(message)
def handle_all_downloads_complete(self):
"""處理所有下載完成"""
"""
Handle completion of all selected downloads.
Hides the progress bar and shows a success message to the user.
"""
self.hide_progress()
QMessageBox.information(self, "Downloads Complete", "All selected items have been downloaded successfully!")
def create_device_section(self):
"""Create the device connection section"""
"""
Create the device connection section.
Builds a section containing:
- Device table showing connected Kneron devices with their details
- Action buttons for refresh, register, update firmware, and install drivers
Returns:
QFrame: The device section frame widget.
"""
device_section = QFrame(self)
device_section.setStyleSheet("""
QFrame {
@ -618,7 +728,16 @@ class UtilitiesScreen(QWidget):
return device_section
def create_status_section(self):
"""Create the status section"""
"""
Create the status section for displaying operation status.
Builds a section containing:
- Status label showing current device status or operation result
- Progress bar for long-running operations (hidden by default)
Returns:
QFrame: The status section frame widget.
"""
status_section = QFrame(self)
status_section.setStyleSheet("""
QFrame {
@ -686,7 +805,16 @@ class UtilitiesScreen(QWidget):
return status_section
def refresh_devices(self):
"""Refresh the list of devices"""
"""
Refresh and scan for connected Kneron devices.
Clears the device table and scans for available devices using the
Kneron Plus SDK. Populates the table with device information including
model, port ID, firmware version, KN number, link speed, and status.
Returns:
bool: True if devices were found, False otherwise.
"""
try:
# Clear the table
self.device_table.setRowCount(0)
@ -719,32 +847,32 @@ class UtilitiesScreen(QWidget):
usb_id = QTableWidgetItem(port_id)
self.device_table.setItem(i, 1, usb_id)
# 嘗試獲取 system_info 中的 firmware_version
# Try to get firmware_version from system_info
firmware_version = "-"
try:
if device.is_connectable:
# 連接設備並獲取系統信息
# Connect to device and get system info
device_group = kp.core.connect_devices(usb_port_ids=[port_id])
system_info = kp.core.get_system_info(
device_group=device_group,
usb_port_id=device.usb_port_id
)
# 從 system_info 對象獲取固件版本
# Get firmware version from system_info object
if system_info and hasattr(system_info, 'firmware_version'):
# firmware_version 是一個對象，獲取其字符串表示
# firmware_version is an object, get its string representation
fw_version = system_info.firmware_version
if hasattr(fw_version, 'firmware_version'):
# 提取版本號，移除字典格式
# Extract version number, remove dict format
version_str = fw_version.firmware_version
# 如果版本是字典格式，提取其中的值
# If version is in dict format, extract the value
if isinstance(version_str, dict) and 'firmware_version' in version_str:
firmware_version = version_str['firmware_version']
else:
firmware_version = version_str
else:
# 將對象轉換為字符串並清理格式
# Convert object to string and clean up format
version_str = str(fw_version)
# 嘗試從字符串中提取版本號
# Try to extract version number from string
import re
match = re.search(r'"firmware_version":\s*"([^"]+)"', version_str)
if match:
@ -765,7 +893,7 @@ class UtilitiesScreen(QWidget):
# Link Speed
link_speed_str = "Unknown"
if hasattr(device, 'link_speed'):
# 從完整的 link_speed 字符串中提取 SPEED_XXX 部分
# Extract SPEED_XXX part from full link_speed string
full_speed = str(device.link_speed)
if "SUPER" in full_speed:
link_speed_str = "SUPER"
@ -774,7 +902,7 @@ class UtilitiesScreen(QWidget):
elif "FULL" in full_speed:
link_speed_str = "FULL"
else:
# 嘗試提取 KP_USB_SPEED_XXX 部分
# Try to extract KP_USB_SPEED_XXX part
import re
match = re.search(r'KP_USB_SPEED_(\w+)', full_speed)
if match:
@ -810,7 +938,13 @@ class UtilitiesScreen(QWidget):
return False
def register_device(self):
"""Register the selected device"""
"""
Register the currently selected device.
Checks if a device is selected and initiates the registration process.
Currently shows a placeholder message as this feature is planned for
a future update.
"""
selected_rows = self.device_table.selectedItems()
if not selected_rows:
QMessageBox.warning(self, "Warning", "Please select a device to register")
@ -820,64 +954,78 @@ class UtilitiesScreen(QWidget):
QMessageBox.information(self, "Info", "Device registration functionality will be implemented in a future update")
def update_firmware(self):
"""Update firmware for the selected device"""
"""
Update firmware for the currently selected device.
Loads and flashes the SCPU and NCPU firmware files to the selected
Kneron device. Shows progress during the update process.
Raises:
Displays error message if no device is selected, firmware files
are not found, or update fails.
"""
try:
# 檢查是否有選擇設備
# Check if a device is selected
selected_rows = self.device_table.selectionModel().selectedRows()
if not selected_rows:
QMessageBox.warning(self, "警告", "請選擇要更新韌體的設備")
QMessageBox.warning(self, "Warning", "Please select a device to update firmware")
return
# 獲取選擇的設備資訊
# Get selected device information
row_index = selected_rows[0].row()
device_model = self.device_table.item(row_index, 0).text() # 設備型號
device_model = self.device_table.item(row_index, 0).text() # Device model
port_id = self.device_table.item(row_index, 1).text() # Port ID
# 顯示進度條
self.show_progress(f"正在更新 {device_model} 的韌體...", 0)
# Show progress bar
self.show_progress(f"Updating firmware for {device_model}...", 0)
# 連接設備
# Connect to device
device_group = kp.core.connect_devices(usb_port_ids=[int(port_id)])
# 構建韌體檔案路徑
# Build firmware file paths
scpu_fw_path = os.path.join(FW_DIR, device_model, "fw_scpu.bin")
ncpu_fw_path = os.path.join(FW_DIR, device_model, "fw_ncpu.bin")
# 檢查韌體檔案是否存在
# Check if firmware files exist
if not os.path.exists(scpu_fw_path) or not os.path.exists(ncpu_fw_path):
self.hide_progress()
QMessageBox.critical(self, "錯誤", f"找不到 {device_model} 的韌體檔案")
QMessageBox.critical(self, "Error", f"Firmware files not found for {device_model}")
return
# 更新進度
# Update progress
self.update_progress(30)
# 載入韌體
# Load firmware
kp.core.load_firmware_from_file(
device_group=device_group,
scpu_fw_path=scpu_fw_path,
ncpu_fw_path=ncpu_fw_path
)
# 更新進度
# Update progress
self.update_progress(100)
# 顯示成功訊息
QMessageBox.information(self, "成功", f"{device_model} 的韌體已成功更新")
# Show success message
QMessageBox.information(self, "Success", f"Firmware for {device_model} has been updated successfully")
except Exception as e:
self.hide_progress()
QMessageBox.critical(self, "錯誤", f"更新韌體時發生錯誤: {str(e)}")
QMessageBox.critical(self, "Error", f"Error updating firmware: {str(e)}")
finally:
self.hide_progress()
def install_drivers(self):
"""Install drivers for all supported Kneron devices"""
"""
Install drivers for all supported Kneron devices.
Iterates through all supported Kneron product IDs and installs
the corresponding Windows drivers. Shows progress during installation.
"""
try:
# 顯示進度條
# Show progress bar
self.show_progress("Installing Kneron Device Drivers...", 0)
# 列出所有產品 ID
# List all product IDs
product_ids = [
kp.ProductId.KP_DEVICE_KL520,
kp.ProductId.KP_DEVICE_KL720_LEGACY,
@ -890,30 +1038,30 @@ class UtilitiesScreen(QWidget):
success_count = 0
total_count = len(product_ids)
# 安裝每個驅動程式
# Install each driver
for i, product_id in enumerate(product_ids):
try:
# 更新進度條
# Update progress bar
progress = int((i / total_count) * 100)
self.update_progress(progress)
self.progress_title.setText(f"Installing [{product_id.name}] driver...")
# 安裝驅動程式
# Install driver
kp.core.install_driver_for_windows(product_id=product_id)
success_count += 1
# 更新狀態訊息
# Update status message
self.status_label.setText(f"Successfully installed {product_id.name} driver")
except kp.ApiKPException as exception:
error_msg = f"Error: install {product_id.name} driver failed, error msg: [{str(exception)}]"
self.status_label.setText(error_msg)
QMessageBox.warning(self, "Driver Installation Error", error_msg)
# 完成安裝
# Complete installation
self.update_progress(100)
self.hide_progress()
# 顯示結果訊息
# Show result message
if success_count == total_count:
QMessageBox.information(self, "Success", "All Kneron device drivers installed successfully!")
else:
@ -927,21 +1075,37 @@ class UtilitiesScreen(QWidget):
QMessageBox.critical(self, "Error", error_msg)
def show_progress(self, title, value):
"""Show the progress bar with the given title and value"""
"""
Show the progress section with the specified title and initial value.
Args:
title (str): The title/message to display above the progress bar.
value (int): The initial progress value (0-100).
"""
self.progress_title.setText(title)
self.progress_bar.setValue(value)
self.progress_section.setVisible(True)
def update_progress(self, value):
"""Update the progress bar value"""
"""
Update the progress bar to the specified value.
Args:
value (int): The progress value (0-100).
"""
self.progress_bar.setValue(value)
def hide_progress(self):
"""Hide the progress section"""
"""Hide the progress section and reset it to default state."""
self.progress_section.setVisible(False)
def on_device_selection_changed(self):
"""Handle device selection changes"""
"""
Handle device selection changes in the device table.
Updates the status label to show the currently selected device
information and ensures the entire row is highlighted.
"""
selected_rows = self.device_table.selectionModel().selectedRows()
if selected_rows:
# Get the selected row index
@ -956,6 +1120,12 @@ class UtilitiesScreen(QWidget):
self.device_table.selectRow(row_index)
def show_utilities_page(self):
"""
Switch to the utilities page view.
Updates button styles to show utilities as active and purchased items
as inactive. Shows the utilities page and hides the purchased items page.
"""
self.utilities_button.setStyleSheet("""
QPushButton {
background-color: #3498DB;
@ -996,6 +1166,12 @@ class UtilitiesScreen(QWidget):
self.current_page = "utilities"
def show_purchased_items_page(self):
"""
Switch to the purchased items page view.
Updates button styles to show purchased items as active and utilities
as inactive. Shows the purchased items page and hides the utilities page.
"""
self.purchased_items_button.setStyleSheet("""
QPushButton {
background-color: #3498DB;