mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-18 07:23:03 +08:00
KenanZhu
5e898180c7
- GUI 模块统一 QtCore → QtGui → QtWidgets 导入排列，各类独占一行按字母排序 - 统一类间两空行、类内方法间一空行、函数间一空行的间距规范 - 统一方法排列顺序：__init__ → setupUi → connectSignals → public → Slot → private - 统一 _widgets 中 ConditionRowFrame/ActionStepFrame 方法命名（populate* / toScript / updateValueWidget） - LibTimeSelector 迁入 operators/abs 抽象层 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
99 lines
2.6 KiB
Python
99 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Copyright (c) 2025 KenanZhu.
|
|
All rights reserved.
|
|
|
|
This software is provided "as is", without any warranty of any kind.
|
|
You may use, modify, and distribute this file under the terms of the MIT License.
|
|
See the LICENSE file for details.
|
|
"""
|
|
import logging
|
|
import queue
|
|
import datetime
|
|
|
|
import managers.log.LogManager as LogManager
|
|
|
|
|
|
class MsgBase:
|
|
"""
|
|
Base class for message and trace abilities (thread-safe).
|
|
|
|
This class provides the foundation for message handling and tracing
|
|
abilities based on the provided input and output queues. It enables
|
|
thread-safe communication between components using queue-based messaging.
|
|
|
|
Args:
|
|
input_queue (queue.Queue): The input queue for receiving messages.
|
|
output_queue (queue.Queue): The output queue for sending messages.
|
|
|
|
Usage:
|
|
This class must be initialized with input and output queues. The queue
|
|
provider (the caller of this class or its subclasses) must explicitly
|
|
implement queue polling to retrieve and process messages.
|
|
"""
|
|
|
|
class TraceLevel:
|
|
"""
|
|
Enum class for trace levels.
|
|
|
|
This class provides the trace levels for the logger.
|
|
"""
|
|
DEBUG = logging.DEBUG
|
|
INFO = logging.INFO
|
|
WARNING = logging.WARNING
|
|
ERROR = logging.ERROR
|
|
CRITICAL = logging.CRITICAL
|
|
|
|
def __init__(
|
|
self,
|
|
input_queue: queue.Queue,
|
|
output_queue: queue.Queue
|
|
):
|
|
|
|
self._class_name = self.__class__.__name__
|
|
self._input_queue = input_queue
|
|
self._output_queue = output_queue
|
|
try:
|
|
self._logger = LogManager.getLogger(self._class_name)
|
|
except RuntimeError:
|
|
self._logger = None
|
|
|
|
def _showMsg(
|
|
self,
|
|
msg: str
|
|
):
|
|
|
|
self._output_queue.put(f"[{self._class_name:<15}] >>> : {msg}")
|
|
|
|
def _showTrace(
|
|
self,
|
|
msg: str,
|
|
level: int = logging.INFO,
|
|
no_log: bool = False
|
|
):
|
|
|
|
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
|
|
self._output_queue.put(f"{timestamp}-[{self._class_name:<15}] : {msg}")
|
|
if self._logger and not no_log:
|
|
self._logger.log(level, msg)
|
|
|
|
def _showLog(
|
|
self,
|
|
msg: str,
|
|
level: int = logging.INFO
|
|
):
|
|
|
|
if self._logger:
|
|
self._logger.log(level, msg)
|
|
|
|
def _waitMsg(
|
|
self,
|
|
timeout: float = 1.0
|
|
) -> str:
|
|
|
|
try:
|
|
msg = self._input_queue.get(timeout=timeout)
|
|
return msg
|
|
except queue.Empty:
|
|
return None
|