""" Palladium CLI implementation. Subcommands:: palladium test # Verify Athena connectivity palladium list # List TEI tool instances palladium reports # List TEI report templates palladium summary # Print a tool's financial summary palladium calculate # Trigger /calculate palladium export -o file # Save export envelope as JSON The CLI is invoked via ``python -m palladium`` (root shim) which calls :func:`main` here. """ from __future__ import annotations import argparse import json import logging import sys from collections.abc import Sequence from core.export import build_report_data, write_report_data from core.tei_client import AthenaAPIError, TEIClient def _configure_logging(verbosity: int) -> None: level = logging.WARNING if verbosity == 1: level = logging.INFO elif verbosity >= 2: level = logging.DEBUG logging.basicConfig( level=level, format="%(asctime)s %(levelname)s %(name)s: %(message)s" ) def _build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="palladium", description="Palladium — TEI Calculator CLI for Athena.", ) p.add_argument("-v", "--verbose", action="count", default=0) sub = p.add_subparsers(dest="command", required=True) sub.add_parser("test", help="Verify Athena API connectivity") sub.add_parser("list", help="List TEI tool instances") sub.add_parser("reports", help="List TEI report templates") s_summary = sub.add_parser("summary", help="Print a tool's financial summary") s_summary.add_argument("public_id") s_calc = sub.add_parser("calculate", help="Trigger calculation for a tool") s_calc.add_argument("public_id") s_export = sub.add_parser("export", help="Export a tool's report data as JSON") s_export.add_argument("public_id") s_export.add_argument( "-o", "--output", default="-", help="Output path (default: stdout)", ) s_export.add_argument( "--no-scenarios", action="store_true", help="Skip computing conservative/moderate/aggressive scenarios.", ) s_export.add_argument( "--study-slug", default=None, help="Optional study identifier to embed in metadata.", ) return p def _print_table(rows: list[dict], columns: list[tuple[str, str]]) -> None: """Tiny pure-stdlib pretty-printer.""" widths = [len(label) for _, label in columns] formatted: list[list[str]] = [] for r in rows: rec = [str(r.get(key, "") or "") for key, _ in columns] formatted.append(rec) for i, val in enumerate(rec): if len(val) > widths[i]: widths[i] = len(val) fmt = " ".join(f"{{:<{w}}}" for w in widths) print(fmt.format(*[label for _, label in columns])) print(fmt.format(*["-" * w for w in widths])) for rec in formatted: print(fmt.format(*rec)) def cmd_test(client: TEIClient, args) -> int: result = client.test_connection() print(json.dumps(result, indent=2)) return 0 if result.get("status") == "ok" else 1 def cmd_list(client: TEIClient, args) -> int: tools = client.list_tools() if not tools: print("(no TEI tools)") return 0 rows = [] for t in tools: report = t.get("report") if isinstance(report, dict): report_name = report.get("name", "") else: report_name = report or "" rows.append( { "id": t.get("id", ""), "name": t.get("name", ""), "report": report_name, "status": t.get("status", ""), "version": t.get("current_version", ""), "modified": t.get("modified_date", ""), } ) _print_table( rows, [ ("id", "PUBLIC_ID"), ("name", "NAME"), ("report", "REPORT"), ("status", "STATUS"), ("version", "VER"), ("modified", "MODIFIED"), ], ) return 0 def cmd_reports(client: TEIClient, args) -> int: reports = client.list_reports() if not reports: print("(no reports)") return 0 rows = [ { "id": r.get("id", ""), "name": r.get("name", ""), "vendor": r.get("vendor", ""), "version": r.get("version", ""), "fields": r.get("field_count", 0), "instances": r.get("instance_count", 0), "status": r.get("status", ""), } for r in reports ] _print_table( rows, [ ("id", "PUBLIC_ID"), ("name", "NAME"), ("vendor", "VENDOR"), ("version", "VER"), ("fields", "FIELDS"), ("instances", "TOOLS"), ("status", "STATUS"), ], ) return 0 def cmd_summary(client: TEIClient, args) -> int: client.print_summary(args.public_id) return 0 def cmd_calculate(client: TEIClient, args) -> int: client.calculate(args.public_id) print(f"Recalculated {args.public_id}.") client.print_summary(args.public_id) return 0 def cmd_export(client: TEIClient, args) -> int: envelope = build_report_data( client, args.public_id, include_scenarios=not args.no_scenarios, study_slug=args.study_slug, ) if args.output in ("-", ""): json.dump(envelope, sys.stdout, indent=2, default=str) sys.stdout.write("\n") else: path = write_report_data(envelope, args.output) print(f"Wrote {path}") return 0 COMMANDS = { "test": cmd_test, "list": cmd_list, "reports": cmd_reports, "summary": cmd_summary, "calculate": cmd_calculate, "export": cmd_export, } def main(argv: Sequence[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) _configure_logging(args.verbose) try: client = TEIClient() except ValueError as e: print(f"error: {e}", file=sys.stderr) return 2 handler = COMMANDS[args.command] try: return handler(client, args) except AthenaAPIError as e: print(f"Athena API error {e.status_code}: {e.detail}", file=sys.stderr) return 1 if __name__ == "__main__": # pragma: no cover raise SystemExit(main())