API reference

The public surface is intentionally smaller than at_py.readwrite: the top-level at_py package re-exports common parsers and at_py.client.ATClient. For the full list of parser and writer symbols, see Read/write parsers. For additional names not re-exported at the top level (e.g. at_py.readwrite.load_mat_normalized_path(), at_py.readwrite.vector.parse_readvector_lines()), import from at_py.readwrite.

class at_py.client.ATClient(target: str = 'localhost:50051', runner_mode: Literal['remote', 'docker'] = 'remote', docker_image: str = 'ghcr.io/jgebbie/at-runner:v0.3.0', docker_host_port: int = 50051, backend: RunnerBackend | None = None)

Bases: object

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 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.

Parameters:

target – gRPC address when runner_mode="remote". Ignored for Docker mode after the container is up (the backend uses localhost:docker_host_port).

backend: RunnerBackend | None = None

bellhop(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run the bellhop executable (same arguments as run()).

bellhop3d(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run bellhop3d (same arguments as run()).

bounce(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run bounce (same arguments as run()).

close() → None

Stop the ephemeral Docker container when runner_mode is "docker".

docker_host_port: int = 50051

docker_image: str = 'ghcr.io/jgebbie/at-runner:v0.3.0'

field(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run field (same arguments as run()).

field3d(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run field3d (same arguments as run()).

kraken(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run kraken (same arguments as run()).

krakenc(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run krakenc (same arguments as run()).

run(*, 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).

Parameters:

• model – Executable name (e.g. "bellhop", "kraken").

• file_root – Basename without extension for input files in the runner workspace.

• inputs – Optional map of filename → file body (bytes or str).

• timeout – Optional gRPC deadline in seconds (backend-specific).

runner_mode: Literal['remote', 'docker'] = 'remote'

scooter(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run scooter (same arguments as run()).

sparc(*, file_root: str, inputs: dict[str, bytes | str] | None = None, timeout: int | None = None) → RunResult

Run sparc (same arguments as run()).

target: str = 'localhost:50051'

class at_py.backend.AtRunnerBackend(use_docker: bool = False, docker_image: str = 'ghcr.io/jgebbie/at-runner:v0.3.0', host_port: int = 50051, docker_grpc_port: int = 50051, connect_host: str = '127.0.0.1')

Bases: object

Backend that calls at_runner.run_sync (lazy import).

Remote (default): use ATClient(target="host:port"); this backend forwards that gRPC address to the installed at_runner package.

Docker: set use_docker=True to start the official image (DEFAULT_AT_RUNNER_IMAGE) on first use, map host_port to the container gRPC port, then connect to localhost:host_port. Requires the docker CLI on PATH. Call close() (or use ATClient as a context manager) to stop the container.

close() → None

Stop the Docker container started for use_docker=True, if any.

connect_host: str = '127.0.0.1'

docker_grpc_port: int = 50051

docker_image: str = 'ghcr.io/jgebbie/at-runner:v0.3.0'

host_port: int = 50051

run_sync(*, target: str, model: str, file_root: str, inputs: dict[str, bytes] | None = None, timeout: int | None = None) → RunResult

Call at_runner.run_sync on target (or Docker localhost when configured).

use_docker: bool = False

class at_py.backend.RunnerBackend(*args, **kwargs)

Bases: Protocol

run_sync(*, target: str, model: str, file_root: str, inputs: dict[str, bytes] | None = None, timeout: int | None = None) → RunResult

Run one model synchronously; implementations wrap at_runner.run_sync.

Helpers to run the official at-runner Docker image for local gRPC.

Used by AtRunnerBackend when use_docker=True and by tests.

at_py.runner_docker.assert_docker_available() → None

Raise RuntimeError if the docker CLI is not on PATH.

at_py.runner_docker.grpc_target_for_host_port(host: str, port: int) → str

Format host:port for gRPC client target arguments.

at_py.runner_docker.start_at_runner_container(*, image: str = 'ghcr.io/jgebbie/at-runner:v0.3.0', host_port: int = 50051, container_port: int = 50051) → str

Start a detached container; return container id (for stop_at_runner_container()).

Maps host_port:container_port for the gRPC service. Raises on failure.

at_py.runner_docker.stop_at_runner_container(container_id: str, *, timeout_s: float = 120.0) → None

Run docker stop on the given container id (ignores non-zero exit).

at_py.runner_docker.wait_tcp(host: str, port: int, *, timeout_s: float = 60.0) → None

Block until host:port accepts a TCP connection or timeout_s elapses.

class at_py.types.RunResult(status: str, exit_code: int, stdout: str, stderr: str, elapsed: float, files: dict[str, bytes])

Bases: object

Outcome of a single at-runner model run (mirrors at_runner gRPC response).

Fields follow the remote API: process status and exit_code, captured stdout / stderr, wall-clock elapsed seconds, and files mapping output basenames to raw bytes.

elapsed: float

exit_code: int

files: dict[str, bytes]

status: str

stderr: str

stdout: str