def doctor(
json: bool = typer.Option(False, "--json", help="Output results as JSON"),
) -> None:
"""Check Marianne environment health.
Validates Python version, Marianne installation, conductor status,
available instruments, and safety configuration. Use this after
installation to verify everything is set up correctly.
"""
out = default_console
# Collect all check results
checks: list[dict[str, Any]] = []
warnings: list[str] = []
errors: list[str] = []
# --- Python version ---
py_version = platform.python_version()
py_ok = sys.version_info >= (3, 11)
checks.append({
"name": "Python",
"status": "ok" if py_ok else "error",
"detail": py_version,
"hint": "Python 3.11+ required" if not py_ok else None,
})
if not py_ok:
errors.append("Python 3.11+ required")
# --- Marianne version ---
checks.append({
"name": "Marianne",
"status": "ok",
"detail": f"v{__version__}",
"hint": None,
})
# --- Conductor status ---
conductor_status, conductor_pid = _check_conductor_status()
conductor_ok = conductor_status == "running"
detail = f"running (pid {conductor_pid})" if conductor_ok else "not running"
checks.append({
"name": "Conductor",
"status": "ok" if conductor_ok else "warning",
"detail": detail,
"hint": "Start with: mzt start" if not conductor_ok else None,
})
if not conductor_ok:
warnings.append("Conductor not running")
# --- Instruments ---
all_profiles = _get_all_profiles()
instrument_results: list[dict[str, Any]] = []
ready_count = 0
for profile in all_profiles.values():
available, binary_path = _check_instrument_binary(profile)
if profile.kind == "http":
# HTTP instruments: report as available (we don't probe endpoints)
status = "ok"
detail_str = f"{profile.display_name}"
if profile.http and profile.http.base_url:
detail_str += f" ({profile.http.base_url})"
ready_count += 1
elif available:
status = "ok"
detail_str = f"{binary_path}" if binary_path else profile.display_name
ready_count += 1
else:
status = "optional"
executable = profile.cli.command.executable if profile.cli else "unknown"
detail_str = f"not found ({executable})"
instrument_results.append({
"name": profile.name,
"display_name": profile.display_name,
"kind": profile.kind,
"status": status,
"detail": detail_str,
})
# --- Safety checks ---
safety_warnings: list[str] = []
# Check for cost limits configuration
# The default CostLimitConfig has enabled=False
safety_warnings.append("No cost limits configured. Recommend: cost_limits.max_cost_per_job")
# --- JSON output ---
if json:
result: dict[str, Any] = {
"python_version": py_version,
"marianne_version": __version__,
"conductor": {
"status": conductor_status,
"pid": conductor_pid,
},
"instruments": instrument_results,
"safety_warnings": safety_warnings,
"warnings_count": len(warnings) + len(safety_warnings),
"errors_count": len(errors),
}
out.print_json(json_mod.dumps(result))
return
# --- Rich output ---
out.print()
out.print("[bold]Marianne Doctor[/bold]")
out.print()
# Core checks
for check in checks:
icon = _status_icon(check["status"])
line = f" {icon} {check['name']:<24} {check['detail']}"
out.print(line)
if check.get("hint"):
out.print(f" [dim]{check['hint']}[/dim]")
# Instruments section
out.print()
out.print(" [bold]Instruments:[/bold]")
for inst in instrument_results:
icon = _status_icon(inst["status"])
out.print(f" {icon} {inst['name']:<24} {inst['detail']}")
# Safety section
if safety_warnings:
out.print()
out.print(" [bold]Safety:[/bold]")
for warn in safety_warnings:
out.print(f" [yellow]![/yellow] {warn}")
warnings.append(warn)
# Summary
out.print()
total_warnings = len(warnings)
total_errors = len(errors)
if total_errors > 0:
out.print(
f"[red]{total_errors} error(s), {total_warnings} warning(s). "
f"Marianne is not ready.[/red]"
)
elif total_warnings > 0:
out.print(
f"[yellow]{total_warnings} warning(s).[/yellow] Marianne is ready."
)
else:
out.print("[green]No issues found. Marianne is ready.[/green]")
out.print()