"""
Structured summary models for run results.
Provides Pydantic models for single-run and batch-run summaries that are
populated during execution from events and execution metadata. Summaries
are deterministic, structured, and serializable to both human-readable
text and JSON.
Example:
>>> from thesis.core.output.summary import RunSummary, RunStatus
>>> summary = RunSummary(
... patient_id="P001",
... workflow="hcp",
... status=RunStatus.SUCCESS,
... elapsed_seconds=123.4,
... )
"""
from enum import Enum
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, ConfigDict, Field
__all__ = [
"RunStatus",
"RunResult",
"RunSummary",
"BatchSummary",
"build_bullets",
"format_duration",
]
# Maximum number of bullets shown in the compact summary. The full summary
# (``SummaryDetail.FULL``) renders all bullets uncapped.
_COMPACT_BULLET_CAP = 6
[docs]
class RunStatus(str, Enum):
"""Final status of a single patient run.
- SUCCESS: completed without errors.
- PARTIAL: some steps completed, others failed or were skipped.
- FAILED: processing failed with an error.
- BLOCKED: could not start (e.g. preflight check failure, missing data).
- SKIPPED: intentionally not run (e.g. dry run, user cancellation).
"""
SUCCESS = "success"
PARTIAL = "partial"
FAILED = "failed"
BLOCKED = "blocked"
SKIPPED = "skipped"
[docs]
class RunResult(BaseModel):
"""Detailed result for a single patient run.
Populated incrementally during execution and used to build the
final :class:`RunSummary`.
Attributes:
patient_id: Patient identifier.
status: Final run status.
elapsed_seconds: Wall-clock duration in seconds.
workflow: Workflow name that was executed.
error_message: Primary failure reason, if any.
error_type: Exception class name, if any.
error_history: Per-attempt failure messages in order (one entry per
failed retry attempt), used to show error progression across retries.
warnings: Warning messages collected during the run.
steps_completed: Names of successfully completed steps.
steps_failed: Names of steps that failed.
files_changed: Paths of files created or modified.
commands_run: External commands executed (e.g. FSL, ANTs).
retry_count: Number of retries attempted.
suggested_next_step: Actionable suggestion for the user.
metadata: Arbitrary extra data.
"""
model_config = ConfigDict(extra="forbid")
patient_id: str = ""
status: RunStatus = RunStatus.SUCCESS
elapsed_seconds: float = 0.0
workflow: str = ""
error_message: str = ""
error_type: str = ""
error_history: List[str] = Field(default_factory=list)
warnings: List[str] = Field(default_factory=list)
steps_completed: List[str] = Field(default_factory=list)
steps_failed: List[str] = Field(default_factory=list)
files_changed: List[str] = Field(default_factory=list)
commands_run: List[str] = Field(default_factory=list)
retry_count: int = 0
suggested_next_step: str = ""
metadata: Dict[str, Any] = Field(default_factory=dict)
[docs]
class RunSummary(BaseModel):
"""Structured summary of a single patient run.
Built from a :class:`RunResult` after execution completes. Contains
the information needed to render the compact footer shown to the user.
Attributes:
patient_id: Patient identifier.
workflow: Workflow name.
status: Final outcome.
elapsed_seconds: Wall-clock duration.
headline: One-line status headline (e.g. ``"SUCCESS hcp P001 2m 13s"``).
bullets: Key summary bullets (3-6 items).
next_step: Optional suggested next action.
result: Full result data for verbose/JSON output.
"""
model_config = ConfigDict(extra="forbid")
patient_id: str = ""
workflow: str = ""
status: RunStatus = RunStatus.SUCCESS
elapsed_seconds: float = 0.0
headline: str = ""
bullets: List[str] = Field(default_factory=list)
next_step: str = ""
result: Optional[RunResult] = None
[docs]
@classmethod
def from_result(cls, result: RunResult) -> "RunSummary":
"""Build a summary from a completed run result.
Auto-generates the headline and bullet points from the result data.
Args:
result: Completed run result.
Returns:
Populated RunSummary.
"""
headline = _build_headline(result)
bullets = build_bullets(result, cap=_COMPACT_BULLET_CAP)
next_step = result.suggested_next_step or _suggest_next_step(result)
return cls(
patient_id=result.patient_id,
workflow=result.workflow,
status=result.status,
elapsed_seconds=result.elapsed_seconds,
headline=headline,
bullets=bullets,
next_step=next_step,
result=result,
)
[docs]
class BatchSummary(BaseModel):
"""Aggregated summary for a batch of patient runs.
Collects child :class:`RunSummary` objects and computes aggregate
statistics for the batch-level footer.
Attributes:
workflow: Workflow name.
total: Total number of patient runs.
succeeded: Count of successful runs.
partial: Count of partially successful runs.
failed: Count of failed runs.
blocked: Count of blocked runs.
skipped: Count of skipped runs.
total_elapsed_seconds: Total wall-clock time for the batch.
avg_elapsed_seconds: Average per-patient duration.
retries: Total retries across all patients.
run_summaries: Per-patient summaries in execution order.
failure_reasons: Grouped failure reasons with patient lists.
suggested_follow_up: Actionable follow-up suggestion.
"""
model_config = ConfigDict(extra="forbid")
workflow: str = ""
total: int = 0
succeeded: int = 0
partial: int = 0
failed: int = 0
blocked: int = 0
skipped: int = 0
total_elapsed_seconds: float = 0.0
avg_elapsed_seconds: float = 0.0
retries: int = 0
run_summaries: List[RunSummary] = Field(default_factory=list)
failure_reasons: Dict[str, List[str]] = Field(default_factory=dict)
suggested_follow_up: str = ""
[docs]
@classmethod
def from_results(
cls,
results: List[RunResult],
workflow: str = "",
batch_elapsed: float = 0.0,
) -> "BatchSummary":
"""Build a batch summary from individual run results.
Args:
results: List of completed run results.
workflow: Workflow name.
batch_elapsed: Total wall-clock time for the batch.
Returns:
Populated BatchSummary.
"""
summaries = [RunSummary.from_result(r) for r in results]
succeeded = sum(1 for r in results if r.status == RunStatus.SUCCESS)
partial = sum(1 for r in results if r.status == RunStatus.PARTIAL)
failed = sum(1 for r in results if r.status == RunStatus.FAILED)
blocked = sum(1 for r in results if r.status == RunStatus.BLOCKED)
skipped = sum(1 for r in results if r.status == RunStatus.SKIPPED)
total_retries = sum(r.retry_count for r in results)
total_patient_time = sum(r.elapsed_seconds for r in results)
avg_time = total_patient_time / len(results) if results else 0.0
# Group failure reasons
failure_reasons: Dict[str, List[str]] = {}
for r in results:
if r.status in (RunStatus.FAILED, RunStatus.BLOCKED) and r.error_message:
em = r.error_message
msg = em if len(em) <= 80 else em[:77] + "..."
reason = f"{r.error_type}: {msg}" if r.error_type else msg
failure_reasons.setdefault(reason, []).append(r.patient_id)
follow_up = _suggest_batch_follow_up(results, failed, blocked, partial)
return cls(
workflow=workflow,
total=len(results),
succeeded=succeeded,
partial=partial,
failed=failed,
blocked=blocked,
skipped=skipped,
total_elapsed_seconds=batch_elapsed or total_patient_time,
avg_elapsed_seconds=avg_time,
retries=total_retries,
run_summaries=summaries,
failure_reasons=failure_reasons,
suggested_follow_up=follow_up,
)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _build_headline(result: RunResult) -> str:
"""Build a one-line status headline from a run result."""
status_label = result.status.value.upper()
parts = [status_label]
if result.workflow:
parts.append(result.workflow)
if result.patient_id:
parts.append(result.patient_id)
if result.elapsed_seconds > 0:
parts.append(format_duration(result.elapsed_seconds))
return " ".join(parts)
[docs]
def build_bullets(result: RunResult, cap: Optional[int] = _COMPACT_BULLET_CAP) -> List[str]:
"""Build key summary bullets from a run result.
Args:
result: Completed run result.
cap: Maximum number of bullets to return. ``None`` returns every
bullet (used by the full summary); the compact summary caps at
:data:`_COMPACT_BULLET_CAP`.
Returns:
Ordered list of summary bullets, truncated to ``cap`` when set.
"""
bullets: List[str] = []
# Steps info
n_completed = len(result.steps_completed)
n_failed = len(result.steps_failed)
if n_completed or n_failed:
total = n_completed + n_failed
bullets.append(f"Steps: {n_completed}/{total} completed")
# Files changed
n_files = len(result.files_changed)
if n_files:
bullets.append(f"Files created/modified: {n_files}")
# Commands run
n_commands = len(result.commands_run)
if n_commands:
bullets.append(f"Commands executed: {n_commands}")
# Warnings
n_warnings = len(result.warnings)
if n_warnings:
bullets.append(f"Warnings: {n_warnings}")
# Error
if result.error_message:
msg = result.error_message
if len(msg) > 100:
msg = msg[:97] + "..."
bullets.append(f"Error: {msg}")
# Retries
if result.retry_count > 0:
bullets.append(f"Retries: {result.retry_count}")
# Error progression across retries
if result.status == RunStatus.FAILED and len(result.error_history) > 1:
progression = "; ".join(result.error_history)
if len(progression) > 160:
progression = progression[:157] + "..."
bullets.append(f"Failed after {result.retry_count} retries. {progression}")
# Tract-similarity headline (populated by the CLI when running full_pipeline)
ts_bullet = _format_tract_similarity_bullet(result.metadata.get("tract_similarity"))
if ts_bullet:
bullets.append(ts_bullet)
# Ensure at least one bullet
if not bullets:
if result.status == RunStatus.SUCCESS:
bullets.append("Completed without errors")
elif result.status == RunStatus.SKIPPED:
bullets.append("Run was skipped (dry run)")
if cap is not None:
return bullets[:cap]
return bullets
def _format_tract_similarity_bullet(metrics: Optional[Dict[str, Any]]) -> str:
"""Render the tract_similarity headline bullet.
Returns an empty string when *metrics* is missing, not a dict, or
contains no non-null numeric values — the caller skips empty strings.
"""
if not isinstance(metrics, dict):
return ""
parts: List[str] = []
for key in ("dice", "pearson", "hausdorff95", "nmi"):
value = metrics.get(key)
if value is None:
continue
try:
formatted = f"{float(value):.2f}"
except (TypeError, ValueError):
continue
suffix = "mm" if key == "hausdorff95" else ""
parts.append(f"{key}={formatted}{suffix}")
if not parts:
return ""
return "tract_similarity: " + " ".join(parts)
def _suggest_next_step(result: RunResult) -> str:
"""Generate a context-aware next-step suggestion."""
if result.status == RunStatus.FAILED:
if result.error_type:
return f"Rerun with -v for details. Error type: {result.error_type}"
return "Rerun with -v for detailed error output"
if result.status == RunStatus.BLOCKED:
return "Check preflight requirements and input data"
if result.warnings:
return "Review warnings above; rerun with -v for full details"
return ""
def _suggest_batch_follow_up(
results: List[RunResult],
failed: int,
blocked: int,
partial: int,
) -> str:
"""Generate a follow-up suggestion for a batch run.
Args:
results: List of completed run results.
failed: Count of failed runs.
blocked: Count of blocked runs.
partial: Count of partially successful runs.
Returns:
A follow-up suggestion string, or an empty string when none applies.
"""
failed_ids = [r.patient_id for r in results if r.status == RunStatus.FAILED]
blocked_ids = [r.patient_id for r in results if r.status == RunStatus.BLOCKED]
if failed_ids:
ids_str = " -p ".join(failed_ids[:5])
suffix = f" (and {len(failed_ids) - 5} more)" if len(failed_ids) > 5 else ""
return f"Retry failed: thesis run -w <workflow> -p {ids_str}{suffix} -v"
if blocked_ids:
return "Check preflight requirements for blocked patients"
partial_ids = [r.patient_id for r in results if r.status == RunStatus.PARTIAL]
if partial_ids and not failed_ids and not blocked_ids:
return (
"Review partial results above; some steps may need manual "
"intervention or a rerun with adjusted parameters."
)
return ""
