mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-18 07:23:03 +08:00
70 lines
1.8 KiB
Python
70 lines
1.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Copyright (c) 2025 - 2026 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 queue
|
|
import datetime
|
|
|
|
|
|
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.
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
def _showMsg(
|
|
self,
|
|
msg: str
|
|
):
|
|
|
|
self._output_queue.put(f"[{self._class_name:<15}] >>> : {msg}")
|
|
|
|
|
|
def _showTrace(
|
|
self,
|
|
msg: str
|
|
):
|
|
|
|
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
|
|
self._output_queue.put(f"{timestamp}-[{self._class_name:<15}] : {msg}")
|
|
|
|
|
|
def _waitMsg(
|
|
self,
|
|
timeout: float = 1.0
|
|
) -> str:
|
|
|
|
try:
|
|
msg = self._input_queue.get(timeout=timeout)
|
|
return msg
|
|
except queue.Empty:
|
|
return None
|