from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
from at_py.backend import AtRunnerBackend, RunnerBackend
from at_py.runner_docker import DEFAULT_AT_RUNNER_GRPC_PORT, DEFAULT_AT_RUNNER_IMAGE
from at_py.types import RunResult
RunnerMode = Literal["remote", "docker"]
def _ensure_bytes_map(inputs: dict[str, bytes | str] | None) -> dict[str, bytes] | None:
"""Normalize optional input map to UTF-8 ``bytes`` values for the gRPC backend."""
if inputs is None:
return None
out: dict[str, bytes] = {}
for name, content in inputs.items():
if isinstance(content, bytes):
out[name] = content
elif isinstance(content, str):
out[name] = content.encode("utf-8")
else:
raise TypeError(f"inputs[{name!r}] must be bytes or str; got {type(content)}")
return out
[docs]
@dataclass
class ATClient:
"""Execute Acoustics Toolbox models via **at-runner** (gRPC).
**Connecting to a runner**
* ``runner_mode="remote"`` (default): call an existing at-runner process. Set
``target`` to its ``host:port`` (default ``localhost:50051``).
* ``runner_mode="docker"``: start the official container image on first use
(requires the ``docker`` CLI on ``PATH``), map ``docker_host_port`` to the
container gRPC port, then connect. Stop the container with :meth:`close` or
use ``with ATClient(...) as client:``.
Install the **Python gRPC client** (import name ``at_runner``) from the pinned
public at-runner tag documented in ``README.md``. Docstrings in this package
intentionally avoid heavy Sphinx ``versionadded`` / “parity with Matlab …”
boilerplate; see the **Port status** section in ``README.md`` for Matlab file
mapping.
:param target: gRPC address when ``runner_mode="remote"``. Ignored for Docker
mode after the container is up (the backend uses ``localhost:docker_host_port``).
"""
target: str = "localhost:50051"
runner_mode: RunnerMode = "remote"
docker_image: str = DEFAULT_AT_RUNNER_IMAGE
docker_host_port: int = DEFAULT_AT_RUNNER_GRPC_PORT
backend: RunnerBackend | None = None
def __post_init__(self) -> None:
"""Construct the default :class:`~at_py.backend.AtRunnerBackend` when none is injected."""
if self.backend is None:
object.__setattr__(
self,
"backend",
AtRunnerBackend(
use_docker=(self.runner_mode == "docker"),
docker_image=self.docker_image,
host_port=self.docker_host_port,
),
)
[docs]
def close(self) -> None:
"""Stop the ephemeral Docker container when ``runner_mode`` is ``\"docker\"``."""
be = self.backend
if isinstance(be, AtRunnerBackend):
be.close()
def __enter__(self) -> ATClient:
"""Enter context: return ``self`` (see :meth:`close`)."""
return self
def __exit__(self, *exc: object) -> None:
"""Leave context: always call :meth:`close`."""
self.close()
[docs]
def run(
self,
*,
model: str,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run an Acoustics Toolbox executable via at-runner (see package docstring for install).
:param model: Executable name (e.g. ``\"bellhop\"``, ``\"kraken\"``).
:param file_root: Basename without extension for input files in the runner workspace.
:param inputs: Optional map of filename → file body (``bytes`` or ``str``).
:param timeout: Optional gRPC deadline in seconds (backend-specific).
"""
backend = self.backend
assert backend is not None # set in __post_init__
return backend.run_sync(
target=self.target,
model=model,
file_root=file_root,
inputs=_ensure_bytes_map(inputs),
timeout=timeout,
)
# --- Model convenience wrappers (matching AT executables) ---
[docs]
def bellhop(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run the ``bellhop`` executable (same arguments as :meth:`run`)."""
return self.run(model="bellhop", file_root=file_root, inputs=inputs, timeout=timeout)
[docs]
def bellhop3d(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run ``bellhop3d`` (same arguments as :meth:`run`)."""
return self.run(model="bellhop3d", file_root=file_root, inputs=inputs, timeout=timeout)
[docs]
def kraken(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run ``kraken`` (same arguments as :meth:`run`)."""
return self.run(model="kraken", file_root=file_root, inputs=inputs, timeout=timeout)
[docs]
def krakenc(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run ``krakenc`` (same arguments as :meth:`run`)."""
return self.run(model="krakenc", file_root=file_root, inputs=inputs, timeout=timeout)
[docs]
def scooter(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run ``scooter`` (same arguments as :meth:`run`)."""
return self.run(model="scooter", file_root=file_root, inputs=inputs, timeout=timeout)
[docs]
def sparc(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run ``sparc`` (same arguments as :meth:`run`)."""
return self.run(model="sparc", file_root=file_root, inputs=inputs, timeout=timeout)
[docs]
def bounce(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run ``bounce`` (same arguments as :meth:`run`)."""
return self.run(model="bounce", file_root=file_root, inputs=inputs, timeout=timeout)
[docs]
def field(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run ``field`` (same arguments as :meth:`run`)."""
return self.run(model="field", file_root=file_root, inputs=inputs, timeout=timeout)
[docs]
def field3d(
self,
*,
file_root: str,
inputs: dict[str, bytes | str] | None = None,
timeout: int | None = None,
) -> RunResult:
"""Run ``field3d`` (same arguments as :meth:`run`)."""
return self.run(model="field3d", file_root=file_root, inputs=inputs, timeout=timeout)