Output System#

Structured output system for CLI runs. Provides event-driven messaging, verbosity-filtered rendering, progress tracking, and deterministic end-of-run summaries.

The live terminal output layer now coordinates renderer writes with active spinners and progress bars so workflow events do not bleed into the progress line. For Nipype execution, WorkflowProgress exposes an overall node-level progress bar that can be driven from scheduler status callbacks.

The package re-exports the most commonly used output classes, while the canonical API documentation lives on the concrete submodules below.

thesis.core.output.events#

Event system for structured output in the thesis framework.

Provides an event bus that decouples event emission from rendering. All executor/tool output is normalized into structured events, which are then filtered by verbosity and rendered by the output layer.

Example

>>> from thesis.core.output.events import get_event_bus, EventLevel
>>> bus = get_event_bus()
>>> bus.emit("Workflow started", level=EventLevel.IMPORTANT)
class thesis.core.output.events.EventLevel[source]#

Bases: IntEnum

Event importance levels, ordered from most to least critical.

  • ERROR: Failures that stop processing.

  • WARNING: Issues that may affect outcome but do not stop processing.

  • IMPORTANT: Key milestones – tool starts/completions, file changes, build/test results, skipped actions.

  • INFO: Detailed informational messages (full tool logs, timing, etc.).

  • DEBUG: Internal implementation details.

Default output mode shows ERROR, WARNING, IMPORTANT. Verbose mode shows everything. Quiet mode shows ERROR only plus the final result.

ERROR = 50#
WARNING = 40#
IMPORTANT = 30#
INFO = 20#
DEBUG = 10#
__new__(value)#
class thesis.core.output.events.Event[source]#

Bases: object

A structured output event emitted during pipeline execution.

Variables:

  • message – Human-readable event description.

  • level – Importance level controlling visibility.

  • timestamp – Unix timestamp when the event was created.

  • category – Optional grouping tag (e.g. "workflow", "tool", "file", "test").

  • patient_id – Patient this event relates to, if any.

  • metadata – Arbitrary key-value data attached to the event.

Parameters:
message: str#
level: EventLevel = 20#
timestamp: float#
category: str = ''#
patient_id: str = ''#
metadata: Dict[str, Any]#
__init__(message, level=EventLevel.INFO, timestamp=<factory>, category='', patient_id='', metadata=<factory>)#
Parameters:
Return type:

None

class thesis.core.output.events.EventBus[source]#

Bases: object

Central event pipeline that decouples emission from rendering.

Listeners subscribe to receive events and can filter by minimum level. The bus is thread-safe: events can be emitted from worker threads while the main thread renders output.

Example

>>> bus = EventBus()
>>> bus.subscribe(lambda e: print(e.message))
>>> bus.emit("hello")
Parameters:

max_events (int)

__init__(max_events=10000)[source]#
Parameters:

max_events (int)

Return type:

None

subscribe(listener)[source]#

Register a listener that will receive all emitted events.

Parameters:

listener (Callable[[Event], None]) – Callable that accepts an Event.

Return type:

None

unsubscribe(listener)[source]#

Remove a previously registered listener.

Parameters:

listener (Callable[[Event], None]) – The listener to remove. No-op if not found.

Return type:

None

emit(message, *, level=EventLevel.INFO, category='', patient_id='', metadata=None)[source]#

Create and dispatch an event to all listeners.

Parameters:

  • message (str) – Human-readable description of the event.

  • level (EventLevel) – Importance level.

  • category (str) – Optional grouping tag.

  • patient_id (str) – Patient identifier, if applicable.

  • metadata (Optional[Dict[str, Any]]) – Extra structured data.

Return type:

Event

Returns:

The created Event instance.

emit_event(event)[source]#

Dispatch a pre-built event to all listeners.

Parameters:

event (Event) – Event instance to dispatch.

Return type:

None

get_events(min_level=EventLevel.DEBUG, category=None, patient_id=None)[source]#

Return stored events matching the given filters.

Parameters:

  • min_level (EventLevel) – Minimum importance level to include.

  • category (Optional[str]) – If set, only return events with this category.

  • patient_id (Optional[str]) – If set, only return events for this patient.

Return type:

List[Event]

Returns:

List of matching events in chronological order.

clear()[source]#

Remove all stored events.

Return type:

None

property event_count: int#

Number of stored events.

thesis.core.output.events.get_event_bus()[source]#

Return the process-wide EventBus singleton.

The bus is created on first call and reused thereafter.

Return type:

EventBus

Returns:

The global EventBus instance.

thesis.core.output.events.reset_event_bus()[source]#

Reset the global event bus (primarily for testing).

Return type:

None

thesis.core.output.modes#

Output mode configuration for the thesis framework.

Defines the verbosity levels, summary detail, and output format settings that control what the user sees during and after a run.

Example

>>> from thesis.core.output.modes import OutputConfig, OutputMode
>>> cfg = OutputConfig(mode=OutputMode.VERBOSE)
>>> cfg.min_event_level
EventLevel.DEBUG
class thesis.core.output.modes.OutputMode[source]#

Bases: str, Enum

Verbosity level for console output.

  • QUIET: errors and final result only.

  • NORMAL: concise, human-friendly progress and key events.

  • VERBOSE: full informational logs, timing, debug context.

QUIET = 'quiet'#
NORMAL = 'normal'#
VERBOSE = 'verbose'#
__new__(value)#
class thesis.core.output.modes.SummaryDetail[source]#

Bases: str, Enum

Level of detail in the end-of-run summary.

  • OFF: no summary printed.

  • COMPACT: one-line status + 3-6 key bullets.

  • FULL: expanded summary with all metadata.

OFF = 'off'#
COMPACT = 'compact'#
FULL = 'full'#
__new__(value)#
class thesis.core.output.modes.OutputFormat[source]#

Bases: str, Enum

Output serialization format.

  • HUMAN: colored/styled text for interactive terminals.

  • JSON: machine-readable JSON lines (one object per event/summary).

HUMAN = 'human'#
JSON = 'json'#
__new__(value)#
class thesis.core.output.modes.OutputConfig[source]#

Bases: BaseModel

Configuration for the output system.

Controls verbosity, summary detail, progress display, and format. CLI flags override values set here.

Variables:

  • mode – Verbosity level.

  • summary – Summary detail level.

  • progress – Whether to show progress bars/spinners. None means auto-detect (enabled for TTY, disabled otherwise).

  • output_format – Serialization format.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

mode: OutputMode#
summary: SummaryDetail#
progress: bool | None#
output_format: OutputFormat#
property min_event_level: EventLevel#

Minimum event level to display in the current mode.

property show_progress: bool#

Whether animated progress UI should be shown.

Progress remains enabled across output modes unless the user explicitly disables it. The CLI and logging layers are responsible for routing non-progress output in a progress-safe way.

property is_verbose: bool#

Shorthand for checking verbose mode.

property is_quiet: bool#

Shorthand for checking quiet mode.

property is_json: bool#

Shorthand for checking JSON output format.

thesis.core.output.summary#

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,
... )
class thesis.core.output.summary.RunStatus[source]#

Bases: 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'#
__new__(value)#
class thesis.core.output.summary.RunResult[source]#

Bases: BaseModel

Detailed result for a single patient run.

Populated incrementally during execution and used to build the final RunSummary.

Variables:

  • 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.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

patient_id: str#
status: RunStatus#
elapsed_seconds: float#
workflow: str#
error_message: str#
error_type: str#
error_history: List[str]#
warnings: List[str]#
steps_completed: List[str]#
steps_failed: List[str]#
files_changed: List[str]#
commands_run: List[str]#
retry_count: int#
suggested_next_step: str#
metadata: Dict[str, Any]#
class thesis.core.output.summary.RunSummary[source]#

Bases: BaseModel

Structured summary of a single patient run.

Built from a RunResult after execution completes. Contains the information needed to render the compact footer shown to the user.

Variables:

  • 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.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

patient_id: str#
workflow: str#
status: RunStatus#
elapsed_seconds: float#
headline: str#
bullets: List[str]#
next_step: str#
result: RunResult | None#
classmethod from_result(result)[source]#

Build a summary from a completed run result.

Auto-generates the headline and bullet points from the result data.

Parameters:

result (RunResult) – Completed run result.

Return type:

RunSummary

Returns:

Populated RunSummary.

class thesis.core.output.summary.BatchSummary[source]#

Bases: BaseModel

Aggregated summary for a batch of patient runs.

Collects child RunSummary objects and computes aggregate statistics for the batch-level footer.

Variables:

  • 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.

Parameters:

data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

workflow: str#
total: int#
succeeded: int#
partial: int#
failed: int#
blocked: int#
skipped: int#
total_elapsed_seconds: float#
avg_elapsed_seconds: float#
retries: int#
run_summaries: List[RunSummary]#
failure_reasons: Dict[str, List[str]]#
suggested_follow_up: str#
classmethod from_results(results, workflow='', batch_elapsed=0.0)[source]#

Build a batch summary from individual run results.

Parameters:

  • results (List[RunResult]) – List of completed run results.

  • workflow (str) – Workflow name.

  • batch_elapsed (float) – Total wall-clock time for the batch.

Return type:

BatchSummary

Returns:

Populated BatchSummary.

thesis.core.output.summary.build_bullets(result, cap=6)[source]#

Build key summary bullets from a run result.

Parameters:

  • result (RunResult) – Completed run result.

  • cap (Optional[int]) – Maximum number of bullets to return. None returns every bullet (used by the full summary); the compact summary caps at _COMPACT_BULLET_CAP.

Return type:

List[str]

Returns:

Ordered list of summary bullets, truncated to cap when set.

thesis.core.output.summary.format_duration(seconds)[source]#

Format a duration in seconds to a human-readable string.

Parameters:

seconds (float) – Duration in seconds.

Return type:

str

Returns:

Formatted string like "1h 23m 45s" or "< 1s".

thesis.core.output.progress#

Progress tracking protocol for the thesis framework.

Provides a Protocol for node-level progress tracking that allows different progress bar implementations to be used interchangeably.

class thesis.core.output.progress.NodeProgressProtocol[source]#

Bases: Protocol

Protocol for node-level progress tracking.

Any class that implements node_started, node_finished, start, and stop can be used as a node progress tracker. This allows Click’s progressbar wrapper (ClickNodeProgress) to be used interchangeably.

start()[source]#

Start displaying the progress bar.

Return type:

None

stop()[source]#

Stop and clean up the progress bar.

Return type:

None

node_started(node_name)[source]#

Called when a node begins execution.

Parameters:

node_name (str)

Return type:

None

node_finished(node_name, status='')[source]#

Called when a node finishes execution.

Parameters:

  • node_name (str)

  • status (str)

Return type:

None

__init__(*args, **kwargs)#

thesis.core.output.renderer#

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 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)
class thesis.core.output.renderer.OutputRenderer[source]#

Bases: object

Renders structured output to the terminal or as JSON.

Connects to the EventBus and prints events that pass the verbosity filter. Also provides methods for rendering run and batch summaries.

Parameters:

  • config (Optional[OutputConfig]) – Output configuration controlling verbosity and format.

  • event_bus (Optional[EventBus]) – Event bus to subscribe to. Uses the global singleton if not provided.

  • file (Optional[IO[str]]) – Output stream (default: stderr).

__init__(config=None, event_bus=None, file=None)[source]#
Parameters:
Return type:

None

attach(bus=None)[source]#

Subscribe to an event bus to render events in real time.

Parameters:

bus (Optional[EventBus]) – Event bus to subscribe to. If None, uses the global singleton.

Return type:

None

detach()[source]#

Unsubscribe from the event bus.

Return type:

None

render_event(event)[source]#

Render a single event to the output stream.

Parameters:

event (Event) – The event to render.

Return type:

None

render_run_summary(summary)[source]#

Render a single-run summary footer.

Parameters:

summary (RunSummary) – The run summary to render.

Return type:

None

render_batch_summary(batch)[source]#

Render a batch summary footer.

Parameters:

batch (BatchSummary) – The batch summary to render.

Return type:

None

On this page