Skip to content

[Cleanup] formatter.py: remove 3 dead print methods, collapse log-strip regexes #1856

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
agu2347 wants to merge 1 commit into
base: master
Choose a base branch
from
+70 −182
Open

[Cleanup] formatter.py: remove 3 dead print methods, collapse log-strip regexes #1856

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
198 changes: 16 additions & 182 deletions swarms/utils/formatter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,9 @@
import time
import re
from typing import Any, Callable, Dict, List, Optional

from rich.console import Console
from rich.live import Live
from rich.panel import Panel
from rich.progress import (
Progress,
SpinnerColumn,
TextColumn,
)
from rich.table import Table
from rich.text import Text
from rich.spinner import Spinner
Expand Down Expand Up @@ -41,24 +35,12 @@ def _clean_output(self, output: str) -> str:
if not output:
return ""

# Remove log prefixes and timestamps
# Remove log prefixes and timestamps. The alternation covers every
# loguru level (not just the four originally hardcoded here), so a
# level like SUCCESS or TRACE is stripped the same as INFO/DEBUG.
output = re.sub(
r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| INFO.*?\|.*?\|",
"",
output,
)
output = re.sub(
r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| DEBUG.*?\|.*?\|",
"",
output,
)
output = re.sub(
r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| WARNING.*?\|.*?\|",
"",
output,
)
output = re.sub(
r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| ERROR.*?\|.*?\|",
r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \| "
r"(?:INFO|DEBUG|WARNING|ERROR|SUCCESS|TRACE|CRITICAL).*?\|.*?\|",
"",
output,
)
Expand All @@ -71,11 +53,12 @@ def _clean_output(self, output: str) -> str:
)
output = re.sub(rf"{spinner_chars} Loop \d+/\d+", "", output)

# Remove any remaining log messages
output = re.sub(r"INFO.*?\|.*?\|.*?\|", "", output)
output = re.sub(r"DEBUG.*?\|.*?\|.*?\|", "", output)
output = re.sub(r"WARNING.*?\|.*?\|.*?\|", "", output)
output = re.sub(r"ERROR.*?\|.*?\|.*?\|", "", output)
# Remove any remaining log messages (same level alternation as above).
output = re.sub(
r"(?:INFO|DEBUG|WARNING|ERROR|SUCCESS|TRACE|CRITICAL).*?\|.*?\|.*?\|",
"",
output,
)

# Clean up extra whitespace and empty lines
output = re.sub(r"\n\s*\n\s*\n", "\n\n", output)
Expand Down Expand Up @@ -449,20 +432,13 @@ def print_markdown(
title: str = "",
border_style: str = "blue",
) -> None:
"""Print content as markdown with syntax highlighting.
"""Alias for print_panel; kept for backward compatibility.

Args:
content (str): The content to display as markdown
title (str): The title of the panel
border_style (str): The border style for the panel
Historically this rendered markdown independently, but its body was
identical to print_panel's markdown branch, so it now just forwards
to print_panel instead of duplicating that logic.
"""
if self.markdown_handler:
self.markdown_handler.render_markdown_output(
content, title, border_style
)
else:
# Fallback to regular panel if markdown is disabled
self.print_panel(content, title, border_style)
self.print_panel(content, title, border_style)

def print_table(
self, title: str, data: Dict[str, List[str]]
Expand All @@ -484,69 +460,6 @@ def print_table(
self.console.print(f"\n🔥 {title}:", style="bold yellow")
self.console.print(table)

def print_progress(
self,
description: str,
task_fn: Callable,
*args: Any,
**kwargs: Any,
) -> Any:
"""
Prints a progress bar to the console and executes a task function.

Args:
description (str): The description of the task.
task_fn (Callable): The function to execute.
*args (Any): Arguments to pass to the task function.
**kwargs (Any): Keyword arguments to pass to the task function.

Returns:
Any: The result of the task function.
"""
with Progress(
SpinnerColumn(),
TextColumn("[progress.description]{task.description}"),
) as progress:
task = progress.add_task(description, total=None)
result = task_fn(*args, **kwargs)
progress.update(task, completed=True)
return result

def print_panel_token_by_token(
self,
tokens: str,
title: str = "Output",
style: str = "bold cyan",
delay: float = 0.01,
by_word: bool = False,
) -> None:
"""
Prints a string in real-time, token by token (character or word) inside a Rich panel.

Args:
tokens (str): The string to display in real-time.
title (str): Title of the panel.
style (str): Style for the text inside the panel.
delay (float): Delay in seconds between displaying each token.
by_word (bool): If True, display by words; otherwise, display by characters.
"""
text = Text(style=style)

# Split tokens into characters or words
token_list = tokens.split() if by_word else tokens

with Live(
Panel(text, title=title, border_style=style),
console=self.console,
refresh_per_second=10,
) as live:
for token in token_list:
text.append(token + (" " if by_word else ""))
live.update(
Panel(text, title=title, border_style=style)
)
time.sleep(delay)

def print_streaming_panel(
self,
streaming_response,
Expand Down Expand Up @@ -747,85 +660,6 @@ def stop_dashboard(self):
self.console.print() # Add blank line after stopping
self._dashboard_live = None

def print_plan_tree(
self,
task_description: str,
steps: List[Dict[str, Any]],
print_on: bool = True,
) -> None:
"""
Print the plan as a beautiful tree using Rich.

Args:
task_description: Description of the main task
steps: List of step dictionaries with step_id, description, priority, and optional dependencies
print_on: Whether to print to console (True) or just log (False)
"""
import logging

logger = logging.getLogger(__name__)

# Create root tree
tree = Tree(
f"[bold cyan]📋 Plan: {task_description}[/bold cyan]"
)

# Priority color mapping
priority_colors = {
"critical": "red",
"high": "yellow",
"medium": "blue",
"low": "green",
}

priority_icons = {
"critical": "🔴",
"high": "🟠",
"medium": "🟡",
"low": "🟢",
}

# Create a mapping of step_id to tree nodes for dependency handling
step_nodes = {}

# First pass: create all nodes
for step in steps:
step_id = step.get("step_id", "")
description = step.get("description", "")
priority = step.get("priority", "medium").lower()
dependencies = step.get("dependencies", [])

priority_color = priority_colors.get(priority, "white")
priority_icon = priority_icons.get(priority, "○")

# Create step label with priority indicator
step_label = (
f"[{priority_color}]{priority_icon} {step_id}[/{priority_color}]: "
f"{description}"
)

# Add dependencies info if present
if dependencies:
deps_text = ", ".join(dependencies)
step_label += f" [dim](depends on: {deps_text})[/dim]"

# Add node to tree
step_node = tree.add(step_label)
step_nodes[step_id] = step_node

# Print the tree
if print_on:
self.console.print("\n")
self.console.print(tree)
self.console.print("")
else:
# Even if print_on is False, log the tree structure
logger.info(f"Plan created: {task_description}")
for step in steps:
logger.info(
f" - {step.get('step_id')} ({step.get('priority')}): {step.get('description')}"
)

def display_hierarchy(
self,
director_name: str,
Expand Down
54 changes: 54 additions & 0 deletions tests/utils/test_formatter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from swarms.utils.formatter import Formatter
import pytest


def test_formatter():
Expand Down Expand Up @@ -124,3 +125,56 @@ def greet(self):

if __name__ == "__main__":
test_formatter()


@pytest.mark.parametrize(
"level",
[
"INFO",
"DEBUG",
"WARNING",
"ERROR",
"SUCCESS",
"TRACE",
"CRITICAL",
],
)
def test_clean_output_strips_every_log_level(level):
"""_clean_output's level alternation must cover every loguru level,
not just the four (INFO/DEBUG/WARNING/ERROR) originally hardcoded.
SUCCESS in particular is used throughout swarms (e.g. graph_workflow.py)
and previously passed through _clean_output untouched."""
handler = Formatter(md=True).markdown_handler
line = (
f"2026-08-09 12:00:00 | {level} | mymodule:myfunc:42 | "
"some log line"
)
content = line + "\nActual content that should survive"
cleaned = handler._clean_output(content)
assert level not in cleaned
assert "Actual content that should survive" in cleaned


def test_clean_output_handles_empty_string():
handler = Formatter(md=True).markdown_handler
assert handler._clean_output("") == ""


def test_dead_print_methods_are_removed():
"""print_progress, print_panel_token_by_token, and print_plan_tree had
zero callers anywhere in the codebase and are removed as dead code.
"""
formatter = Formatter(md=True)
assert not hasattr(formatter, "print_progress")
assert not hasattr(formatter, "print_panel_token_by_token")
assert not hasattr(formatter, "print_plan_tree")


def test_print_markdown_still_works_as_alias():
"""print_markdown now forwards to print_panel instead of duplicating
its markdown-rendering branch; it must still be callable with the same
signature and not raise."""
formatter = Formatter(md=True)
formatter.print_markdown(
"# Title\n\nBody", title="t", border_style="cyan"
)