"""The single Rich console the whole tool prints through.
Every human-facing message goes through this module (CLAUDE.md rule), so output
styling, quiet modes, and testability stay in one place. Core logic never
prints -- the ``core/`` report objects arrive here fully computed and this
module only renders them.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from rich.console import Console
from rich.markup import escape
from rich.prompt import Confirm
from rich.table import Table
from matmelt.core.state import is_terminal
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
from matmelt.core.preflight import (
AiidaStatus,
DoctorReport,
EnumerationReport,
ModelStatus,
RequirementStatus,
)
from matmelt.core.state import InFlightJob, MaterialRecord, MaterialState
console = Console()
error_console = Console(stderr=True, style="bold red")
#: Advisory notes that must not land on stdout, because a ``--json`` run's
#: stdout has to stay parseable (the M10 lesson).
note_console = Console(stderr=True, style="yellow")
[docs]
def render_config_error(message: str) -> None:
"""One config problem report (the message already carries YAML paths).
Rendered verbatim: config errors quote user input and name lists in
square brackets (``missing [structure]``), which Rich would otherwise
swallow as markup tags.
Parameters
----------
message
The full ``ConfigError`` text, printed with Rich markup **off** so
bracketed content such as ``missing [structure]`` reaches the user
verbatim instead of being eaten as a style tag.
"""
error_console.print(f"Config error: {message}", markup=False)
[docs]
def render_legacy_config_note(legacy: str, preferred: str) -> None:
"""Say once that the pre-rename config filename is being used.
Parameters
----------
legacy
The pre-rename filename actually being read.
preferred
The current filename it should be renamed to.
"""
note_console.print(
f"NOTE: reading {legacy} -- that name is deprecated; rename it to {preferred}."
)
[docs]
def render_error(message: str) -> None:
"""A command-level problem that is not a config error (already actionable).
Parameters
----------
message
The message, which the caller has already made actionable.
"""
error_console.print(message)
[docs]
def render_init_created(path: str) -> None:
"""Announce the scaffolded config and hand off to the doctor checks.
Parameters
----------
path
Where the config was scaffolded.
"""
console.print(f"Wrote starter config: [bold]{path}[/bold]")
console.print(
"Edit run.elements and the backends/stages blocks for your machines; "
"re-run 'matmelt doctor' after any change."
)
console.print("Checking the environment against it now:")
[docs]
def print_json_payload(payload: dict[str, Any]) -> None:
"""Machine-readable output for ``--json`` flags: plain JSON on stdout.
Parameters
----------
payload
The document to print. It goes to stdout alone, with every human-facing
line routed to stderr, so the output stays machine-parseable.
"""
console.print_json(data=payload)
[docs]
def render_doctor(
report: DoctorReport,
enumeration: EnumerationReport,
model: ModelStatus | None,
aiida: AiidaStatus | None = None,
*,
ok: bool,
) -> None:
"""The full ``doctor`` diagnosis: backends, stage readiness, seeds, model, verdict.
Parameters
----------
report
Backend health and per-stage readiness.
enumeration
The seed-count report.
model
The screen model status, or ``None`` when no screen stage is configured.
aiida
The AiiDA status, or ``None`` when no calculate stage is configured.
ok
The overall verdict.
"""
backends = Table(title="Backend reachability")
backends.add_column("Backend")
backends.add_column("Kind")
backends.add_column("Status")
backends.add_column("Detail")
for health in report.backends:
backends.add_row(
health.backend,
health.kind,
"[green]healthy[/green]" if health.healthy else "[red]DOWN[/red]",
health.detail,
)
console.print(backends)
readiness = Table(title="Stage readiness")
readiness.add_column("Stage")
readiness.add_column("Needs")
readiness.add_column("Interpreter")
readiness.add_column("Where")
readiness.add_column("Status")
readiness.add_column("Detail")
for status in report.requirements:
requirement = status.requirement
needs = "stdlib only" if requirement is None else f"{requirement.module}"
readiness.add_row(
status.stage,
needs,
status.python,
status.where,
"[green]ok[/green]" if status.satisfied else "[red]MISSING[/red]",
status.detail,
)
console.print(readiness)
fixes = [status.install_command for status in report.requirements if status.install_command]
if fixes:
console.print("To fix the missing requirements, run:")
for fix in fixes:
console.print(f" {fix}")
render_enumeration(enumeration)
if model is not None:
label = "[green]ok[/green]" if model.ok else "[red]PROBLEM[/red]"
console.print(f"Screen model: {label} -- {model.detail}")
if aiida is not None:
label = "[green]ok[/green]" if aiida.ok else "[red]PROBLEM[/red]"
console.print(f"Calculate AiiDA: {label} -- {aiida.detail}")
if ok:
console.print("[green]doctor: all checks passed.[/green]")
else:
console.print("[red]doctor: problems found (see above).[/red]")
[docs]
def render_note(message: str) -> None:
"""A one-line informational message.
Parameters
----------
message
The text to print.
"""
console.print(message)
[docs]
def render_warning(message: str) -> None:
"""A one-line yellow warning: worth seeing, never fatal.
``markup=False`` for the same reason as ``render_config_error``: messages
carry user data (ids, paths) that must never be eaten as Rich markup.
"""
console.print(f"Warning: {message}", style="yellow", markup=False)
[docs]
def render_enumeration(enumeration: EnumerationReport) -> None:
"""The seed-enumeration report: the n-ary blast radius, before any work.
Parameters
----------
enumeration
The report, whose oversized sizes are named explicitly so a
silently-empty size is never a mystery.
"""
per_size = ", ".join(f"size {size}: {count}" for size, count in enumeration.per_size.items())
console.print(
f"Seed enumeration: {enumeration.n_elements} element(s) -> "
f"{enumeration.total} combination(s) ({per_size})."
)
if enumeration.oversized:
oversized = ", ".join(str(size) for size in enumeration.oversized)
console.print(
f"[yellow]Note:[/yellow] combination size(s) {oversized} exceed the "
f"element count and yield nothing."
)
[docs]
def render_unready(failures: Sequence[RequirementStatus]) -> None:
"""The preflight refusal: every unmet requirement and its exact fix.
Parameters
----------
failures
The unmet requirements. Each is printed with the exact install command
and the location it was probed at, so a remote gap is not mistaken for
a local one.
"""
error_console.print("run: not ready -- missing requirements:")
for status in failures:
module = status.requirement.module if status.requirement else "?"
error_console.print(f" {status.stage}: {module} ({status.detail}) [{status.where}]")
console.print("To fix, run:")
for status in failures:
if status.install_command:
console.print(f" {status.install_command}")
[docs]
def confirm_model_download(detail: str) -> bool:
"""The disclosed foundation-model consent prompt. EOF/no-TTY = declined.
Parameters
----------
detail
The disclosure: model size, cache location, and that nothing downloads
without consent.
Returns
-------
bool
Whether the user agreed. EOF or a non-interactive terminal reads as
**declined** -- an unattended run must never start a large download by
accident.
"""
console.print(detail)
try:
return Confirm.ask("Download it now?", console=console, default=False)
except EOFError:
return False
[docs]
def render_pass_summary(index: int, summary: Mapping[str, int]) -> None:
"""One progress line per pass with activity; silent when nothing happened.
Parameters
----------
index
The pass number.
summary
What the pass did. A pass with no activity prints nothing, so a long
drain does not scroll a wall of zeroes.
"""
activity = {key: value for key, value in summary.items() if value}
if activity:
pairs = ", ".join(f"{key}={value}" for key, value in activity.items())
console.print(f"pass {index}: {pairs}")
[docs]
def render_interrupted() -> None:
"""Clean Ctrl-C: everything durable, the same command resumes."""
console.print(
"\nInterrupted -- state and in-flight jobs are saved; "
"re-run the same command to resume where it left off."
)
[docs]
def render_run_result(
counts: Mapping[MaterialState, int],
*,
drained: bool,
only: str | None = None,
selective: bool = False,
stopped_early: str | None = None,
) -> None:
"""The final per-state tally and whether the funnel fully drained.
In only-mode (``only`` set) "drained" means the one stage has no work
left -- other states legitimately hold non-terminal materials awaiting
the rest of the funnel, so the full-run wording would mislead. The same
holds for a selective run (``selective``): only the chosen lineage was
driven, so unselected materials legitimately remain non-terminal. And
for a mid-funnel stop (``stopped_early`` = the stop_after stage name):
materials rest at that stage's output, resumable, never ``DONE``.
Parameters
----------
counts
The final per-state tally.
drained
Whether the funnel ran out of work.
only
The single stage this run executed, if any.
selective
Whether a material selection was active.
stopped_early
Whether a pass limit cut the run short.
"""
table = Table(title="Materials by state")
table.add_column("State")
table.add_column("Count", justify="right")
for state, count in sorted(counts.items(), key=lambda pair: pair[0].value):
table.add_row(state.value, str(count))
console.print(table)
if only is not None:
if drained:
console.print(
f"[green]--only {only}: stage drained; downstream stages were not run.[/green]"
)
else:
console.print(
f"[yellow]--only {only}: paused with work still pending -- "
f"re-run the same command to continue.[/yellow]"
)
elif selective:
if drained:
console.print(
"[green]--materials: the selected materials are settled; "
"unselected materials were not run.[/green]"
)
else:
console.print(
"[yellow]--materials: paused with selected work still pending -- "
"re-run the same command to continue.[/yellow]"
)
elif stopped_early is not None:
if drained:
console.print(
f"[green]stop_after {stopped_early}: stage drained; materials rest "
f"at its output, ready for the next stage -- raise or remove "
f"stop_after (or use --only) to continue them.[/green]"
)
else:
console.print(
f"[yellow]stop_after {stopped_early}: paused with work still "
f"pending -- re-run the same command to continue.[/yellow]"
)
elif drained:
console.print("[green]run complete: every material is settled.[/green]")
else:
pending = sum(count for state, count in counts.items() if not is_terminal(state))
console.print(
f"[yellow]run paused with {pending} material(s) still pending -- "
f"re-run the same command to continue.[/yellow]"
)
[docs]
def render_status(
run_name: str,
counts: Mapping[MaterialState, int],
failed: Sequence[MaterialRecord],
jobs: Sequence[InFlightJob],
) -> None:
"""The ``status`` view: state tally, failed materials, in-flight jobs.
Parameters
----------
run_name
The configured run name.
counts
The per-state tally.
failed
Terminally failed materials, with their stage, reason and attempts.
jobs
The in-flight ledger.
"""
console.print(f"Status of run [bold]{run_name}[/bold]:")
if not counts:
console.print("No materials yet -- 'matmelt run' seeds and starts the funnel.")
return
table = Table(title="Materials by state")
table.add_column("State")
table.add_column("Count", justify="right")
for state, count in sorted(counts.items(), key=lambda pair: pair[0].value):
table.add_row(state.value, str(count))
console.print(table)
if failed:
failures = Table(title="Failed materials")
failures.add_column("Material")
failures.add_column("Stage")
failures.add_column("Attempts", justify="right")
failures.add_column("Reason")
for record in failed:
failures.add_row(
record.material_id,
record.fail_stage or "?",
str(record.attempts),
record.fail_reason or "?",
)
console.print(failures)
if jobs:
console.print(f"In flight: {len(jobs)} job(s)")
for job in jobs:
materials = ", ".join(job.material_ids)
console.print(f" {job.stage} on {job.backend_name}: {materials} [{job.job_key}]")
[docs]
def render_state_materials(
run_name: str,
state: MaterialState,
records: Sequence[MaterialRecord],
) -> None:
"""The ``status --state`` view: the material ids sitting at one state.
This is the picking list for selective runs: the ids feed
``run --materials`` (parent ids shown so lineage selections are easy to
trace back to their seeds).
Parameters
----------
run_name
The configured run name.
state
The state being listed.
records
The materials at that state, with their parents where they have one.
"""
console.print(f"Run [bold]{run_name}[/bold]: {len(records)} material(s) at {state.value}.")
if not records:
return
table = Table(title=f"Materials at {state.value}")
table.add_column("Material")
table.add_column("Parent")
for record in records:
table.add_row(record.material_id, record.parent_id or "-")
console.print(table)
[docs]
def render_retry_preview(run_name: str, records: Sequence[MaterialRecord]) -> None:
"""Summarise the FAILED materials to be retried, grouped by stage.
The "ingenious UX" the user asked for: instead of a flat wall of ids, one row
per stage with its count -- the headline decision info ("128 failed at
structure") -- plus how many *distinct* reasons that stage saw and one
example. (Reasons often embed the material's own path, so clustering on the
exact reason string is useless; the distinct-count still flags whether a
stage failed one way or many.) Narrow with ``--stage`` if a group is not what
you meant. Stage/reason are ``escape``-d -- they carry user data (paths,
``<&>``) that is never Rich markup.
Parameters
----------
run_name
The configured run name.
records
The materials that would be reset.
"""
by_stage: dict[str, list[str]] = {}
for record in records:
by_stage.setdefault(record.fail_stage or "?", []).append(record.fail_reason or "?")
console.print(
f"Run [bold]{run_name}[/bold]: {len(records)} FAILED material(s) "
f"across {len(by_stage)} stage(s) to retry:"
)
table = Table(title="Failed materials by stage")
table.add_column("Stage")
table.add_column("Count", justify="right")
table.add_column("Distinct reasons", justify="right")
table.add_column("Example reason")
for stage, reasons in sorted(by_stage.items(), key=lambda kv: (-len(kv[1]), kv[0])):
example = reasons[0]
shown = example if len(example) <= 80 else example[:77] + "..."
table.add_row(escape(stage), str(len(reasons)), str(len(set(reasons))), escape(shown))
console.print(table)
[docs]
def confirm_retry(count: int) -> bool:
"""Confirm resetting ``count`` FAILED materials. EOF/no-TTY = declined.
Parameters
----------
count
How many materials would be reset.
Returns
-------
bool
Whether the user agreed. EOF or a non-interactive terminal reads as
declined.
"""
try:
return Confirm.ask(
f"Reset {count} failed material(s) to re-run?", console=console, default=False
)
except EOFError:
return False
[docs]
def render_retry_result(count: int) -> None:
"""Report how many materials were reset back to a claimable state.
Parameters
----------
count
How many materials were actually reset.
"""
if count == 0:
console.print("No materials were reset.")
return
console.print(
f"Reset {count} material(s) -- they will re-run at their failed stage on "
f"the next 'matmelt run'."
)
[docs]
def render_results(
run_name: str,
done: Sequence[MaterialRecord],
with_record: set[str],
) -> None:
"""The ``results`` view: every DONE material and whether it has a record.
Parameters
----------
run_name
The configured run name.
done
The finished materials.
with_record
Which of them have a ``record.json``. A finished material without one
is first-class, not an error.
"""
console.print(f"Results of run [bold]{run_name}[/bold]:")
if not done:
console.print("No DONE materials yet.")
return
table = Table(title="Finished materials")
table.add_column("Material")
table.add_column("Parent")
table.add_column("Record")
for material in done:
table.add_row(
material.material_id,
material.parent_id or "-",
"[green]yes[/green]" if material.material_id in with_record else "-",
)
console.print(table)
console.print(f"{len(done)} DONE material(s), {len(with_record)} with extracted record(s).")
[docs]
def render_cif_export(
run_name: str,
out_dir: str,
written: Sequence[Mapping[str, Any]],
skipped: Sequence[Mapping[str, str]],
failed: Sequence[Mapping[str, str]],
) -> None:
"""The ``cif`` view: what was exported, what was skipped, what failed.
Skips are grouped by reason (hundreds of "did not converge" rows help
nobody); failures are listed per material, because each one is something
the user may want to chase. Ids and reasons are ``escape``-d -- they are
user data, never Rich markup.
Parameters
----------
run_name
The configured run name.
out_dir
Where the files were written.
written
The exported files.
skipped
Ineligible materials, with reasons.
failed
Materials whose geometry could not be exported.
"""
console.print(
f"Run [bold]{run_name}[/bold]: exported {len(written)} CIF file(s) to "
f"[bold]{escape(out_dir)}[/bold]."
)
if skipped:
by_reason: dict[str, int] = {}
for entry in skipped:
by_reason[entry.get("reason", "?")] = by_reason.get(entry.get("reason", "?"), 0) + 1
table = Table(title="Skipped (no calculated structure)")
table.add_column("Reason")
table.add_column("Materials", justify="right")
for reason, count in sorted(by_reason.items(), key=lambda kv: (-kv[1], kv[0])):
shown = reason if len(reason) <= 80 else reason[:77] + "..."
table.add_row(escape(shown), str(count))
console.print(table)
if failed:
failures = Table(title="Could not export")
failures.add_column("Material")
failures.add_column("Problem")
for entry in failed:
problem = entry.get("error", "?")
shown = problem if len(problem) <= 90 else problem[:87] + "..."
failures.add_row(escape(entry.get("id", "?")), escape(shown))
console.print(failures)