| 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() |
|
|