"""
Output rendering for the thesis framework.
Renders events, summaries, and progress to the terminal or JSON.
Handles color, alignment, and non-TTY fallback behavior.
The renderer subscribes to the :class:`EventBus` and filters events
by the configured verbosity level before printing them.
Example:
>>> from thesis.core.output.renderer import OutputRenderer
>>> from thesis.core.output.modes import OutputConfig
>>> renderer = OutputRenderer(OutputConfig())
>>> renderer.render_run_summary(summary)
"""
import json
import os
import sys
from typing import IO, Optional
from .events import Event, EventBus, EventLevel
from .modes import OutputConfig, SummaryDetail
from .summary import (
BatchSummary,
RunResult,
RunStatus,
RunSummary,
build_bullets,
format_duration,
)
__all__ = [
"OutputRenderer",
]
# ---------------------------------------------------------------------------
# Status icons -- readable with and without color
# ---------------------------------------------------------------------------
_STATUS_ICONS = {
RunStatus.SUCCESS: "[OK]",
RunStatus.PARTIAL: "[PARTIAL]",
RunStatus.FAILED: "[FAIL]",
RunStatus.BLOCKED: "[BLOCKED]",
RunStatus.SKIPPED: "[SKIP]",
}
_EVENT_PREFIXES = {
EventLevel.ERROR: "[ERROR]",
EventLevel.WARNING: "[WARN]",
EventLevel.IMPORTANT: "[*]",
EventLevel.INFO: "[.]",
EventLevel.DEBUG: "[D]",
}
# ANSI color codes (applied only when color is enabled)
_COLORS = {
"reset": "\033[0m",
"bold": "\033[1m",
"dim": "\033[2m",
"red": "\033[31m",
"green": "\033[32m",
"yellow": "\033[33m",
"cyan": "\033[36m",
"white": "\033[37m",
}
_STATUS_COLORS = {
RunStatus.SUCCESS: "green",
RunStatus.PARTIAL: "yellow",
RunStatus.FAILED: "red",
RunStatus.BLOCKED: "red",
RunStatus.SKIPPED: "dim",
}
_LEVEL_COLORS = {
EventLevel.ERROR: "red",
EventLevel.WARNING: "yellow",
EventLevel.IMPORTANT: "cyan",
EventLevel.INFO: "dim",
EventLevel.DEBUG: "dim",
}
[docs]
class OutputRenderer:
"""Renders structured output to the terminal or as JSON.
Connects to the :class:`EventBus` and prints events that pass the
verbosity filter. Also provides methods for rendering run and batch
summaries.
Args:
config: Output configuration controlling verbosity and format.
event_bus: Event bus to subscribe to. Uses the global singleton
if not provided.
file: Output stream (default: stderr).
"""
[docs]
def __init__(
self,
config: Optional[OutputConfig] = None,
event_bus: Optional[EventBus] = None,
file: Optional[IO[str]] = None,
) -> None:
self._config = config or OutputConfig()
self._file = file or sys.stderr
self._color_enabled = self._detect_color()
self._bus = event_bus
self._subscribed = False
# ------------------------------------------------------------------
# Event bus integration
# ------------------------------------------------------------------
[docs]
def attach(self, bus: Optional[EventBus] = None) -> None:
"""Subscribe to an event bus to render events in real time.
Args:
bus: Event bus to subscribe to. If ``None``, uses the
global singleton.
"""
if bus is None:
from .events import get_event_bus
bus = get_event_bus()
self._bus = bus
if self._subscribed:
return
bus.subscribe(self._on_event)
self._subscribed = True
[docs]
def detach(self) -> None:
"""Unsubscribe from the event bus."""
if self._bus is not None and self._subscribed:
self._bus.unsubscribe(self._on_event)
self._subscribed = False
def _on_event(self, event: Event) -> None:
"""Event listener callback -- filter and render."""
if event.level < self._config.min_event_level:
return
self.render_event(event)
# ------------------------------------------------------------------
# Event rendering
# ------------------------------------------------------------------
[docs]
def render_event(self, event: Event) -> None:
"""Render a single event to the output stream.
Args:
event: The event to render.
"""
if self._config.is_json:
self._write_json_event(event)
return
prefix = _EVENT_PREFIXES.get(event.level, "[.]")
color_name = _LEVEL_COLORS.get(event.level, "white")
# Build the line
parts = []
if event.patient_id:
parts.append(f"[{event.patient_id}]")
parts.append(event.message)
body = " ".join(parts)
line = f" {self._color(prefix, color_name)} {body}"
self._write(line)
# ------------------------------------------------------------------
# Summary rendering
# ------------------------------------------------------------------
[docs]
def render_run_summary(self, summary: RunSummary) -> None:
"""Render a single-run summary footer.
Args:
summary: The run summary to render.
"""
if self._config.summary == SummaryDetail.OFF:
return
if self._config.is_json:
self._write_json({"type": "run_summary", "data": summary.model_dump(mode="json")})
return
self._write("") # blank line separator
self._write(self._separator())
# Headline
status_color = _STATUS_COLORS.get(summary.status, "white")
icon = _STATUS_ICONS.get(summary.status, "[?]")
headline = f" {self._color(icon, status_color)} {summary.headline}"
self._write(headline)
# Bullets (compact: capped at 6; full: all bullets, uncapped)
is_full = self._config.summary == SummaryDetail.FULL
if self._config.summary in (SummaryDetail.COMPACT, SummaryDetail.FULL):
if is_full and summary.result is not None:
bullets = build_bullets(summary.result, cap=None)
else:
bullets = summary.bullets
for bullet in bullets:
self._write(f" - {bullet}")
# Next step
if summary.next_step:
self._write(f" > {self._color(summary.next_step, 'dim')}")
# Full result details -- shown in verbose mode OR full summary detail,
# so `--summary full` is meaningfully more detailed than compact even
# without `-v`.
if (self._config.is_verbose or is_full) and summary.result:
self._render_verbose_result(summary.result)
self._write(self._separator())
[docs]
def render_batch_summary(self, batch: BatchSummary) -> None:
"""Render a batch summary footer.
Args:
batch: The batch summary to render.
"""
if self._config.summary == SummaryDetail.OFF:
return
if self._config.is_json:
self._write_json({"type": "batch_summary", "data": batch.model_dump(mode="json")})
return
self._write("")
self._write(self._separator("="))
# Top-level outcome
all_ok = batch.failed == 0 and batch.blocked == 0
overall_status = "SUCCESS" if all_ok else "FAILED"
overall_color = "green" if all_ok else "red"
overall_icon = "[OK]" if all_ok else "[FAIL]"
duration = format_duration(batch.total_elapsed_seconds)
headline = (
f" {self._color(overall_icon, overall_color)} "
f"Batch {overall_status}: {batch.succeeded}/{batch.total} succeeded "
f"{duration}"
)
self._write(headline)
# Counts line
counts = []
if batch.succeeded:
counts.append(self._color(f"{batch.succeeded} ok", "green"))
if batch.partial:
counts.append(self._color(f"{batch.partial} partial", "yellow"))
if batch.failed:
counts.append(self._color(f"{batch.failed} failed", "red"))
if batch.blocked:
counts.append(self._color(f"{batch.blocked} blocked", "red"))
if batch.skipped:
counts.append(f"{batch.skipped} skipped")
if counts:
self._write(f" {' | '.join(counts)}")
# Timing
if batch.avg_elapsed_seconds > 0:
avg = format_duration(batch.avg_elapsed_seconds)
self._write(f" Avg per patient: {avg}")
self._write(f" Retries configured: {batch.retries}")
# Per-run status list
self._write("")
show_limit = 50 if self._config.is_verbose else 20
for i, rs in enumerate(batch.run_summaries[:show_limit]):
icon = _STATUS_ICONS.get(rs.status, "[?]")
color = _STATUS_COLORS.get(rs.status, "white")
# Only show per-patient duration if it was actually measured
# (parallel runs don't have per-patient timing).
dur = format_duration(rs.elapsed_seconds) if rs.elapsed_seconds > 0 else ""
note = ""
if rs.result and rs.result.error_message and not self._config.is_quiet:
msg = rs.result.error_message
if len(msg) > 60:
msg = msg[:57] + "..."
note = f" {msg}"
self._write(f" {self._color(icon, color)} {rs.patient_id:<16} {dur:>10}{note}")
remaining = len(batch.run_summaries) - show_limit
if remaining > 0:
self._write(f" ... and {remaining} more (use -v to see all)")
# Failure reasons (grouped)
if batch.failure_reasons and not self._config.is_quiet:
self._write("")
self._write(" Failure reasons:")
for reason, pids in batch.failure_reasons.items():
truncated = reason if len(reason) <= 80 else reason[:77] + "..."
pid_str = ", ".join(pids[:5])
if len(pids) > 5:
pid_str += f" (+{len(pids) - 5})"
self._write(f" - {truncated}")
self._write(f" Patients: {pid_str}")
# Follow-up suggestion
if batch.suggested_follow_up:
self._write("")
self._write(f" > {self._color(batch.suggested_follow_up, 'dim')}")
self._write(self._separator("="))
# ------------------------------------------------------------------
# Verbose result details
# ------------------------------------------------------------------
def _render_verbose_result(self, result: RunResult) -> None:
"""Render full result details in verbose mode."""
self._write("")
if result.steps_completed:
self._write(" Steps completed:")
for step in result.steps_completed:
self._write(f" + {step}")
if result.steps_failed:
self._write(" Steps failed:")
for step in result.steps_failed:
self._write(f" x {step}")
if result.files_changed:
self._write(" Files changed:")
for f in result.files_changed[:20]:
self._write(f" {f}")
if len(result.files_changed) > 20:
self._write(f" ... and {len(result.files_changed) - 20} more")
if result.commands_run:
self._write(" Commands:")
for cmd in result.commands_run[:10]:
self._write(f" $ {cmd}")
if len(result.commands_run) > 10:
self._write(f" ... and {len(result.commands_run) - 10} more")
if result.warnings:
self._write(" Warnings:")
for w in result.warnings:
self._write(f" ! {w}")
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _write(self, text: str) -> None:
"""Write a line to the output stream."""
try:
self._file.write(text + "\n")
self._file.flush()
except (OSError, AttributeError):
pass
def _write_json(self, data: object) -> None:
"""Write a JSON object as a single line."""
try:
self._file.write(json.dumps(data, default=str) + "\n")
self._file.flush()
except (OSError, AttributeError, TypeError):
pass
def _write_json_event(self, event: Event) -> None:
"""Serialize an event as JSON."""
self._write_json(
{
"type": "event",
"level": event.level.name,
"message": event.message,
"category": event.category,
"patient_id": event.patient_id,
"timestamp": event.timestamp,
"metadata": event.metadata,
}
)
def _color(self, text: str, color_name: str) -> str:
"""Apply ANSI color if color is enabled."""
if not self._color_enabled:
return text
code = _COLORS.get(color_name, "")
reset = _COLORS["reset"]
return f"{code}{text}{reset}" if code else text
def _separator(self, char: str = "-") -> str:
"""Build a visual separator line."""
width = _terminal_width()
return f" {char * min(width - 4, 72)}"
def _detect_color(self) -> bool:
"""Detect whether color output should be enabled."""
if os.environ.get("NO_COLOR"):
return False
if self._config.is_json:
return False
try:
return self._file.isatty()
except AttributeError:
return False
def _terminal_width() -> int:
"""Get terminal width, defaulting to 80."""
try:
return os.get_terminal_size().columns
except (OSError, ValueError):
return 80
