metatomic-orca-client

#!/usr/bin/env python3
"""ORCA client that forwards external-tool jobs to a persistent Metatomic server."""

from __future__ import annotations

import json
import os
import sys
import traceback
import urllib.error
import urllib.request
from argparse import ArgumentParser

DEFAULT_BIND = "127.0.0.1:8888"


def send_to_server(host_port: str, arguments: list[str], *, working_directory: str) -> None:
    """Forward a calculation request to ``metatomic-orca-server``."""
    host, port = host_port.split(":", 1)
    url = f"http://{host}:{port}/calculate"
    payload = {"arguments": arguments, "directory": working_directory}
    request = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    try:
        with urllib.request.urlopen(request, timeout=None) as response:
            data = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        print(f"HTTP error {exc.code}: {body}", file=sys.stderr)
        raise SystemExit(1) from exc
    except urllib.error.URLError as exc:
        print(f"Connection error: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc
    except Exception as exc:
        print(f"Unexpected error: {type(exc).__name__}: {exc}", file=sys.stderr)
        traceback.print_exc()
        raise SystemExit(1) from exc

    print(data.get("stdout", ""), end="")
    if data.get("status") != "Success":
        print(
            f"Server error {data.get('error_type')}: {data.get('error_message')}.",
            file=sys.stderr,
        )
        if data.get("traceback"):
            print(data["traceback"], file=sys.stderr)
        raise SystemExit(1)


def build_client_parser() -> ArgumentParser:
    parser = ArgumentParser(
        prog="metatomic-orca-client",
        description="Forward ORCA external-tool jobs to a running metatomic-orca-server.",
    )
    parser.add_argument(
        "-b",
        "--bind",
        metavar="hostname:port",
        default=DEFAULT_BIND,
        dest="host_port",
        help=f"Server bind address and port. Default: {DEFAULT_BIND}.",
    )
    return parser


def main(argv: list[str] | None = None) -> None:
    parser = build_client_parser()
    args, remaining_args = parser.parse_known_args(argv)
    send_to_server(
        args.host_port,
        remaining_args,
        working_directory=os.getcwd(),
    )


if __name__ == "__main__":
    main()