|
from datetime import datetime |
|
from typing import List, Optional |
|
import queue |
|
import threading |
|
|
|
class MessageManager: |
|
def __init__(self, max_messages: int = 100): |
|
self._messages: List[str] = [] |
|
self._max_messages = max_messages |
|
self._message_queue = queue.Queue() |
|
self._lock = threading.Lock() |
|
|
|
|
|
self._formats = { |
|
"INFO": "ℹ️", |
|
"SUCCESS": "✅", |
|
"WARNING": "⚠️", |
|
"ERROR": "❌", |
|
} |
|
|
|
def add_message(self, message: str, message_type: str = "INFO") -> None: |
|
"""Add a new message with minimal timestamp and icon.""" |
|
|
|
timestamp = datetime.now().strftime("%H:%M") |
|
icon = self._formats.get(message_type, "•") |
|
|
|
|
|
if "Processing file" in message or "Created batch folder" in message: |
|
message = self._format_path(message) |
|
|
|
formatted_message = f"{icon} {message}" |
|
|
|
with self._lock: |
|
self._messages.append(formatted_message) |
|
if len(self._messages) > self._max_messages: |
|
self._messages.pop(0) |
|
|
|
def _format_path(self, message: str) -> str: |
|
"""Format file paths to be more concise and readable.""" |
|
if "GRADIO_TEMP_DIR" in message: |
|
|
|
filename = message.split("\\")[-1] |
|
return message.split(":")[0] + ": " + filename |
|
elif "batch_" in message: |
|
|
|
return message.replace("../outputs/", "") |
|
return message |
|
|
|
def add_success(self, message: str) -> None: |
|
"""Add a success message.""" |
|
self.add_message(message, "SUCCESS") |
|
|
|
def add_warning(self, message: str) -> None: |
|
"""Add a warning message.""" |
|
self.add_message(message, "WARNING") |
|
|
|
def add_error(self, message: str) -> None: |
|
"""Add an error message.""" |
|
self.add_message(message, "ERROR") |
|
|
|
def get_messages(self) -> str: |
|
"""Get all messages as a single string with spacing between different types.""" |
|
with self._lock: |
|
|
|
formatted = [] |
|
last_type = None |
|
for msg in self._messages: |
|
current_type = next((t for t in self._formats if self._formats[t] in msg), None) |
|
if last_type and current_type != last_type: |
|
formatted.append("") |
|
formatted.append(msg) |
|
last_type = current_type |
|
return "\n".join(formatted) |
|
|
|
def clear(self) -> None: |
|
"""Clear all messages.""" |
|
with self._lock: |
|
self._messages.clear() |
|
|