`ewatercycle.container`

Container utilities.

Module Contents

class ewatercycle.container.ContainerImage

Bases: str

Custom type for parsing and utilizing container images.

Given a docker url with the following structure

<repository>/<organisation>/<image_name>:<version>

The corresponding apptainer filename is assumed to be

<organisation>-<image_name>_<version>.sif

The repository in the docker url is optional.

Conversion from docker to apptainer is always possible Conversion from apptainer can lead to unexpected behaviour in some cases:

when image name contains ‘_’ but no version tag

when image name contains ‘-’ but no organisation

eWatercycle containers typically don’t have these issues.

Initialize self. See help(type(self)) for accurate signature.

property apptainer_filename: str: Return self as apptainer filename.

property docker_url: str: Return self as docker url.

property version: str

ewatercycle.container.start_container(work_dir: str | pathlib.Path, image: ContainerImage, input_dirs: Iterable[str] | None = None, image_port=55555, timeout=None, delay=0, wrappers: Sequence[type[Any]] = (MemoizedBmi, OptionalDestBmi)) → grpc4bmi.bmi_optionaldest.OptionalDestBmi

Start container with model inside.

The CFG.container_engine value determines the engine used to start a container.

Parameters:

work_dir – Work directory
image – Image name for container.
input_dirs – Additional directories to mount inside container.
image_port – Docker port inside container where grpc4bmi server is running.
timeout – Number of seconds to wait for grpc connection.
delay – Number of seconds to wait before connecting.
wrappers – List of classes to wrap around the grcp4bmi object from container. Order is important. The first wrapper is the most inner wrapper.

Raises:

ValueError – When unknown container technology is requested.
TimeoutError – When model inside container did not start quickly enough.

Returns:

When default wrappers are used then returns the Bmi object which wraps the container, has memoization and has optional dest arguments. When no wrappers are used then returns the Bmi object which wraps the container.

Example

Given CFG.container_engine == docker a marrmot container can be started with:

from ewatercycle.container import start_container

model = start_container(
    work_dir='.',
    image="ewatercycle/marrmot-grpc4bmi",
)

ewatercycle.container.start_apptainer_container(work_dir: str | pathlib.Path, image: ContainerImage, input_dirs: Iterable[str] = (), timeout: int | None = None, delay: int = 0) → bmipy.Bmi

Start Apptainer container with model inside.

Parameters:

work_dir – Work directory
image – Name of apptainer (sif file) or docker image (url). If a docker url is passed, will try to derive the apptainer filename following the format specified in apptainer manual.
input_dirs – Additional directories to mount inside container.
timeout – Number of seconds to wait for grpc connection.
delay – Number of seconds to wait before connecting.

Raises:: TimeoutError – When model inside container did not start quickly enough.
Returns:: Bmi object which wraps the container.

ewatercycle.container.start_docker_container(work_dir: str | pathlib.Path, image: ContainerImage, input_dirs: Iterable[str] = (), image_port=55555, timeout=None, delay=0)

Start Docker container with model inside.

Parameters:

work_dir – Work directory
image – Name of Docker image.
input_dirs – Additional directories to mount inside container.
image_port – Docker port inside container where grpc4bmi server is running.
timeout – Number of seconds to wait for grpc connection.
delay – Number of seconds to wait before connecting.

Raises:

TimeoutError – When model inside container did not start quickly enough.

Returns:

Bmi object which wraps the container.

class ewatercycle.container.BmiFromOrigin(origin: bmipy.Bmi)

Bases: Protocol

Protocol for a BMI that can be used as a BMI itself.

class ewatercycle.container.BmiProxy(origin: bmipy.Bmi)

Bases: bmipy.Bmi

Proxy for a BMI that can be used as a BMI itself.

Parameters:: origin – the BMI object to proxy

Example:

To overwrite the get_value method of a BMI class, you can use the following

>>> class MyBmi(BmiProxy):
...     def get_value(self, name: str, dest: np.ndarray) -> np.ndarray:
...         dest[:] = 1
...         return dest
>>> bmi = MyBmi(BmiImplementation())
>>> bmi.get_value("my_var", np.empty((2,3), dtype=np.float64))
array([[1., 1., 1.], [1., 1., 1.]])

All other methods are forwarded to the origin.

finalize() → None

get_component_name() → str

get_current_time() → float

get_end_time() → float

get_grid_edge_count(grid: int) → int

get_grid_edge_nodes(grid: int, edge_nodes: numpy.ndarray) → numpy.ndarray

get_grid_face_count(grid: int) → int

get_grid_face_edges(grid: int, face_edges: numpy.ndarray) → numpy.ndarray

get_grid_face_nodes(grid: int, face_nodes: numpy.ndarray) → numpy.ndarray

get_grid_node_count(grid: int) → int

get_grid_nodes_per_face(grid: int, nodes_per_face: numpy.ndarray) → numpy.ndarray

get_grid_origin(grid: int, shape: numpy.ndarray) → numpy.ndarray

get_grid_rank(grid: int) → int

get_grid_shape(grid: int, shape: numpy.ndarray) → numpy.ndarray

get_grid_size(grid: int) → int

get_grid_spacing(grid: int, shape: numpy.ndarray) → numpy.ndarray

get_grid_type(grid: int) → str

get_grid_x(grid: int, x: numpy.ndarray) → numpy.ndarray

get_grid_y(grid: int, y: numpy.ndarray) → numpy.ndarray

get_grid_z(grid: int, z: numpy.ndarray) → numpy.ndarray

get_input_item_count() → int

get_input_var_names() → Tuple[str, Ellipsis]

get_output_item_count() → int

get_output_var_names() → Tuple[str, Ellipsis]

get_start_time() → float

get_time_step() → float

get_time_units() → str

get_value(name: str, dest: numpy.ndarray) → numpy.ndarray

get_value_at_indices(name: str, dest: numpy.ndarray, inds: numpy.ndarray) → numpy.ndarray

get_value_ptr(name: str) → numpy.ndarray

get_var_itemsize(name: str) → int

get_var_grid(name: str) → int

get_var_location(name: str) → str

get_var_nbytes(name: str) → int

get_var_type(name: str) → str

get_var_units(name: str) → str

initialize(filename: str) → None

set_value(name: str, src: numpy.ndarray) → None

set_value_at_indices(name: str, inds: numpy.ndarray, src: numpy.ndarray) → None

update() → None

update_until(time: float) → None

ewatercycle.container

Module Contents

`ewatercycle.container`