Coverage for python / aubellhop / bellhop.py: 91%
108 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-24 14:11 +0000
1"""Provides BellhopSimulator class for interacting with bellhop models.
3This class is instantiated within `models.py` and supports the standard
4`bellhop.exe` and `bellhop3d.exe` Fortran interfaces.
6New classes could be written to replicate the interfaces if
7further models wished to be tested with different internals.
9Instances of BellhopSimulator are used as follows in `compute.py`:
11 >>> model_fn = Models.select(this_env, this_task, this_model, debug)
12 >>> fname_base = model_fn.write_env(this_env, this_task, fname_base)
13 >>> model_fn.run(this_task, fname_base, debug=debug),
15In the code above `model_fn` is the instance. `Models` is a utility
16cass which contains a global registry of BellhopSimulator instances.
17Internally `Models.select` uses `model_fn.supports()` to
18identify the BellhopSimulator model (instance) to use.
20Writing the environment file appears circuitous:
22 ~~bellhop.py~~ ~~environment.py~~ ~~writers.py~~
23 model_fn.write_env() → env.to_file() → EnvironmentWriter().write()
25These indirections are partially for modularity and partly for
26encapsulation.
27"""
29from __future__ import annotations
31import sys
32import os
33import subprocess
34import shutil
35from importlib.resources import files
37import tempfile
38from typing import Any, Dict, Tuple
40from .constants import ModelDefaults, BHStrings, FileExt
41from .environment import Environment
42from .readers import read_shd, read_arrivals, read_rays
45class BellhopSimulator:
46 """
47 Interface to the Bellhop underwater acoustics ray tracing propagation model.
49 The following methods are defined:
51 * `supports()`
52 * `write_env()`
53 * `run()`
55 Parameters
56 ----------
57 name : str
58 User-fancing name for the model
59 exe : str
60 Filename of Bellhop executable
61 dim : int
62 Number of dimensions in the model (`2` or `3`)
63 """
65 def __init__(self, name: str = ModelDefaults.name_2d,
66 exe: str = ModelDefaults.exe_2d,
67 dim: int = ModelDefaults.dim_2d,
68 ) -> None:
69 self.name: str = name
70 self.exe: str = exe
71 self.dim: int = dim
73 def supports(self, env: Environment | None = None,
74 task: str | None = None,
75 exe: str | None = None,
76 dim: int | None = None,
77 ) -> bool:
78 """Check whether the model supports the task.
80 This function is supposed to diagnose whether this combination of environment
81 and task is supported by the model."""
82 if env is not None:
83 dim = dim or env._dimension
84 which_bool = self._find_executable(exe or self.exe) is not None
85 task_bool = (task is None) or (task in self._taskmap)
86 dim_bool = (dim is None) or (dim == self.dim)
87 return (which_bool and task_bool and dim_bool)
89 def write_env(self, env: Environment,
90 task: str,
91 fname_base: str | None = None,
92 debug: bool = False,
93 overwrite: bool = False,
94 ) -> str:
95 """
96 Writes the environment to .env file prior to running the model.
98 Uses the `_taskmap` data structure to relate input flags to
99 processng stages, in particular how to select specific "tasks"
100 to be executed.
101 """
102 fname_base, fname = self._prepare_env_file(fname_base, overwrite=overwrite)
103 env.to_file(fname, task=task)
105 return fname_base
107 def run(self, task: str,
108 fname_base: str,
109 rm_files: bool = True,
110 debug: bool = False,
111 ) -> Any:
112 """
113 High-level interface function which runs the model.
114 """
115 load_task_data, task_ext = self._taskmap[task]
116 self._run_exe(fname_base)
117 results = load_task_data(fname_base + task_ext)
118 if rm_files:
119 if debug:
120 print('[DEBUG] Bellhop working files NOT deleted: '+fname_base+'.*')
121 else:
122 self._rm_files(fname_base)
123 return results
125 @property
126 def _taskmap(self) -> Dict[Any, list[Any]]:
127 """Dictionary which maps tasks to execution functions and their parameters"""
128 return {
129 BHStrings.arrivals: [read_arrivals, FileExt.arr],
130 BHStrings.eigenrays: [read_rays, FileExt.ray],
131 BHStrings.rays: [read_rays, FileExt.ray],
132 BHStrings.coherent: [read_shd, FileExt.shd],
133 BHStrings.incoherent: [read_shd, FileExt.shd],
134 BHStrings.semicoherent: [read_shd, FileExt.shd],
135 }
137 def _find_executable(self, exe_name: str) -> str | None:
138 """Find the bellhop executable.
140 First checks the package's bin directory (for installed wheels),
141 then falls back to searching PATH.
143 Parameters
144 ----------
145 exe_name : str
146 Name of the executable (e.g., 'bellhop.exe')
148 Returns
149 -------
150 str | None
151 Path to the executable, or None if not found
152 """
153 pkg_name = (__package__ or "unknown").split(".")[0]
154 try:
155 pkg_bin = files(pkg_name) / "bin" / exe_name
156 if pkg_bin.is_file():
157 return str(pkg_bin)
158 except Exception:
159 pass
161 return shutil.which(exe_name)
163 def _prepare_env_file(self,
164 fname_base: str | None,
165 overwrite: bool = False,
166 ) -> Tuple[str, str]:
167 """Opens a file for writing the .env file, in a temp location if necessary, and delete other files with same basename.
169 Parameters
170 ----------
171 fname_base : str, optional
172 Filename base (no extension) for writing -- if not specified a temporary file (and location) will be used instead
174 Returns
175 -------
176 fh : int
177 File descriptor
178 fname_base : str
179 Filename base
180 """
181 is_temp = fname_base is None
182 if fname_base is not None:
183 fname = os.path.abspath(fname_base + FileExt.env)
184 os.makedirs(os.path.dirname(fname), exist_ok=True)
185 open(fname, "w").close()
186 else:
187 tmp = tempfile.NamedTemporaryFile(suffix=FileExt.env, delete=False, mode="w")
188 fname = tmp.name
189 fname_base = fname[: -len(FileExt.env)]
190 tmp.close()
192 if fname_base is None:
193 raise RuntimeError("Internal error: fname_base is None after preparation")
194 if is_temp or overwrite:
195 self._rm_files(fname_base, not_env=True)
196 return fname_base, fname
198 def _rm_files(self, fname_base: str,
199 not_env: bool = False,
200 ) -> None:
201 """Remove files that would be constructed as bellhop inputs or created as bellhop outputs."""
202 all_ext = [v for k, v in vars(FileExt).items() if not k.startswith('_')]
203 if not_env:
204 all_ext.remove(FileExt.env)
205 for ext in all_ext:
206 self._unlink(fname_base + ext)
208 def _run_exe(self, fname_base: str,
209 args: str = "",
210 debug: bool = False,
211 exe: str | None = None,
212 ) -> None:
213 """Run the executable and raise exceptions if there are errors."""
215 exe_path = self._find_executable(exe or self.exe)
216 if exe_path is None:
217 raise FileNotFoundError(
218 f"Executable '{exe or self.exe}' not found in package bin directory or PATH.\n"
219 f"Please ensure the package is installed correctly or bellhop executables are in your PATH."
220 )
222 # Check macOS Gatekeeper / signature only on Darwin
223 if sys.platform == "darwin":
224 try:
225 check = subprocess.run(
226 ["spctl", "--assess", "--verbose=4", exe_path],
227 stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
228 )
229 if "rejected" in check.stdout or "rejected" in check.stderr:
230 print(f"Warning: {exe_path} is rejected by Gatekeeper. Trying ad-hoc codesign...")
231 subprocess.run(["codesign", "--force", "--sign", "-", exe_path], check=True)
232 except FileNotFoundError:
233 # spctl not found (e.g., not macOS), ignore
234 pass
237 runcmd = [exe_path, fname_base] + args.split()
238 if debug:
239 print("RUNNING:", " ".join(runcmd))
240 result = subprocess.run(runcmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, text=True)
242 if debug and result.stdout:
243 print(result.stdout.strip())
245 if result.returncode != 0:
246 err = self._check_error(fname_base)
247 raise RuntimeError(
248 f"Execution of '{exe_path}' failed with return code {result.returncode}.\n"
249 f"\nCommand: {' '.join(runcmd)}\n"
250 f"\nOutput:\n{result.stdout.strip()}\n"
251 f"\nExtract from PRT file:\n{err}"
252 )
254 def _check_error(self,
255 fname_base: str,
256 ) -> str:
257 """Extracts Bellhop error text from the .prt file"""
258 try:
259 err = ""
260 fatal = False
261 with open(fname_base + FileExt.prt, 'rt') as f:
262 for s in f:
263 if fatal and len(s.strip()) > 0:
264 err += '[FATAL] ' + s.strip() + '\n'
265 if '*** FATAL ERROR ***' in s:
266 fatal = True
267 except FileNotFoundError:
268 pass
269 return err
271 def _unlink(self, f: str) -> None:
272 """Delete file only if it exists"""
273 try:
274 os.unlink(f)
275 except FileNotFoundError:
276 pass