Spaces:
Paused
Paused
File size: 3,007 Bytes
05fcd0f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
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()
# ANSI-style formatting for different message types
self._formats = {
"INFO": "ℹ️", # Info icon
"SUCCESS": "✅", # Checkmark
"WARNING": "⚠️", # Warning icon
"ERROR": "❌", # Error icon
}
def add_message(self, message: str, message_type: str = "INFO") -> None:
"""Add a new message with minimal timestamp and icon."""
# Only show hours:minutes for timestamps
timestamp = datetime.now().strftime("%H:%M")
icon = self._formats.get(message_type, "•")
# Format filename paths to be more readable
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:
# Extract just the filename from temp path
filename = message.split("\\")[-1]
return message.split(":")[0] + ": " + filename
elif "batch_" in message:
# Shorten batch folder path
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:
# Add a blank line between different message types for readability
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("") # Add spacing between different types
formatted.append(msg)
last_type = current_type
return "\n".join(formatted)
def clear(self) -> None:
"""Clear all messages."""
with self._lock:
self._messages.clear()
|