Coverage for kwave/options/simulation_execution_options.py: 37%
147 statements
« prev ^ index » next coverage.py v7.7.1, created at 2025-03-24 12:06 -0700
1import os
2import warnings
3from pathlib import Path
4from typing import Optional, Union
6from kwave import BINARY_DIR, PLATFORM
7from kwave.ksensor import kSensor
10class SimulationExecutionOptions:
11 """
12 A class to manage and configure the execution options for k-Wave simulations.
13 """
15 def __init__(
16 self,
17 is_gpu_simulation: bool = False,
18 binary_path: Optional[str] = None,
19 binary_dir: Optional[str] = None,
20 binary_name: Optional[str] = None,
21 kwave_function_name: Optional[str] = "kspaceFirstOrder3D",
22 delete_data: bool = True,
23 device_num: Optional[int] = None,
24 num_threads: Optional[int] = None,
25 thread_binding: Optional[bool] = None,
26 system_call: Optional[str] = None,
27 verbose_level: int = 0,
28 auto_chunking: Optional[bool] = True,
29 show_sim_log: bool = True,
30 ):
31 self.is_gpu_simulation = is_gpu_simulation
32 self._binary_path = binary_path
33 self._binary_name = binary_name
34 self._binary_dir = binary_dir
35 self.kwave_function_name = kwave_function_name
36 self.delete_data = delete_data
37 self.device_num = device_num
38 self.num_threads = num_threads
39 self.thread_binding = thread_binding
40 self.system_call = system_call
41 self.verbose_level = verbose_level
42 self.auto_chunking = auto_chunking
43 self.show_sim_log = show_sim_log
45 @property
46 def num_threads(self) -> Union[int, str]:
47 return self._num_threads
49 @num_threads.setter
50 def num_threads(self, value: Union[int, str]):
51 cpu_count = os.cpu_count()
52 if cpu_count is None: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 raise RuntimeError("Unable to determine the number of CPUs on this system. Please specify the number of threads explicitly.")
55 if value == "all": 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 warnings.warn(
57 "The 'all' option is deprecated. The value of None sets the maximal number of threads (excluding Windows).",
58 DeprecationWarning,
59 )
60 value = cpu_count
62 if value is None: 62 ↛ 65line 62 didn't jump to line 65 because the condition on line 62 was always true
63 value = cpu_count
65 if not isinstance(value, int): 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true
66 raise ValueError("Got {value}. Number of threads must be 'all' or a positive integer")
68 if value <= 0 or value > cpu_count: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true
69 raise ValueError(f"Number of threads {value} must be a positive integer and less than total threads on the system {cpu_count}.")
71 self._num_threads = value
73 @property
74 def verbose_level(self) -> int:
75 return self._verbose_level
77 @verbose_level.setter
78 def verbose_level(self, value: int):
79 if not (isinstance(value, int) and 0 <= value <= 2): 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 raise ValueError("Verbose level must be between 0 and 2")
81 self._verbose_level = value
83 @property
84 def is_gpu_simulation(self) -> Optional[bool]:
85 return self._is_gpu_simulation
87 @is_gpu_simulation.setter
88 def is_gpu_simulation(self, value: Optional[bool]):
89 "Set the flag to enable default GPU simulation. This option will supersede custom binary paths."
90 self._is_gpu_simulation = value
91 # Automatically update the binary name based on the GPU simulation flag
92 if value is not None: 92 ↛ exitline 92 didn't return from function 'is_gpu_simulation' because the condition on line 92 was always true
93 self._binary_name = None
95 @property
96 def binary_name(self) -> str:
97 valid_binary_names = ["kspaceFirstOrder-OMP", "kspaceFirstOrder-CUDA"]
98 if self._binary_name is None:
99 # set default binary name based on GPU simulation value
100 if self.is_gpu_simulation is None:
101 raise ValueError("`is_gpu_simulation` must be set to either True or False before determining the binary name.")
103 if self.is_gpu_simulation:
104 self._binary_name = "kspaceFirstOrder-CUDA"
105 else:
106 self._binary_name = "kspaceFirstOrder-OMP"
108 if PLATFORM == "windows":
109 self._binary_name += ".exe"
110 valid_binary_names = [name + ".exe" for name in valid_binary_names]
112 elif self._binary_name not in valid_binary_names:
113 warnings.warn("Custom binary name set. Ignoring `is_gpu_simulation` state.")
114 return self._binary_name
116 @binary_name.setter
117 def binary_name(self, value: str):
118 self._binary_name = value
120 @property
121 def binary_path(self) -> Path:
122 if self._binary_path is not None:
123 return self._binary_path
125 binary_dir = BINARY_DIR if self._binary_dir is None else self._binary_dir
127 if binary_dir is None:
128 raise ValueError("Binary directory is not specified.")
130 path = Path(binary_dir) / self.binary_name
131 if PLATFORM == "windows" and not path.name.endswith(".exe"):
132 path = path.with_suffix(".exe")
133 return path
135 @binary_path.setter
136 def binary_path(self, value: str):
137 # check if the binary path is a valid path
138 if not os.path.exists(value):
139 raise FileNotFoundError(
140 f"Binary path {value} does not exist. If you are trying to set `binary_dir`, use the `binary_dir` attribute instead."
141 )
142 self._binary_path = value
144 @property
145 def binary_dir(self) -> str:
146 return BINARY_DIR if self._binary_dir is None else self._binary_dir
148 @binary_dir.setter
149 def binary_dir(self, value: str):
150 # check if binary_dir is a directory
151 if not os.path.isdir(value):
152 raise NotADirectoryError(
153 f"{value} is not a directory. If you are trying to set the `binary_path`, use the `binary_path` attribute instead."
154 )
155 self._binary_dir = Path(value)
157 @property
158 def device_num(self) -> Optional[int]:
159 return self._device_num
161 @device_num.setter
162 def device_num(self, value: Optional[int]):
163 if value is not None and value < 0: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 raise ValueError("Device number must be non-negative")
165 self._device_num = value
167 def as_list(self, sensor: kSensor) -> list[str]:
168 options_list = []
170 if self.device_num is not None:
171 options_list.append("-g")
172 options_list.append(str(self.device_num))
174 if self._num_threads is not None and PLATFORM != "windows":
175 options_list.append("-t")
176 options_list.append(str(self._num_threads))
178 if self.verbose_level > 0:
179 options_list.append("--verbose")
180 options_list.append(str(self.verbose_level))
182 record_options_map = {
183 "p": "p_raw",
184 "p_max": "p_max",
185 "p_min": "p_min",
186 "p_rms": "p_rms",
187 "p_max_all": "p_max_all",
188 "p_min_all": "p_min_all",
189 "p_final": "p_final",
190 "u": "u_raw",
191 "u_max": "u_max",
192 "u_min": "u_min",
193 "u_rms": "u_rms",
194 "u_max_all": "u_max_all",
195 "u_min_all": "u_min_all",
196 "u_final": "u_final",
197 }
199 if sensor.record is not None:
200 matching_keys = sorted(set(sensor.record).intersection(record_options_map.keys()))
201 options_list.extend([f"--{record_options_map[key]}" for key in matching_keys])
203 if "u_non_staggered" in sensor.record or "I_avg" in sensor.record or "I" in sensor.record:
204 options_list.append("--u_non_staggered_raw")
206 if ("I_avg" in sensor.record or "I" in sensor.record) and ("p" not in sensor.record):
207 options_list.append("--p_raw")
208 else:
209 options_list.append("--p_raw")
211 if sensor.record_start_index is not None:
212 options_list.append("-s")
213 options_list.append(f"{sensor.record_start_index}")
215 return options_list
217 def get_options_string(self, sensor: kSensor) -> str:
218 # raise a deprecation warning
219 warnings.warn("This method is deprecated. Use `as_list` method instead.", DeprecationWarning)
220 options_list = self.as_list(sensor)
222 return " ".join(options_list)
224 @property
225 def env_vars(self) -> dict:
226 env = os.environ
228 if PLATFORM != "darwin":
229 env.update({"OMP_PLACES": "cores"})
231 if self.thread_binding is not None:
232 if PLATFORM == "darwin":
233 raise ValueError("Thread binding is not supported in MacOS.")
234 # read the parameters and update the system options
235 if self.thread_binding:
236 env.update({"OMP_PROC_BIND": "SPREAD"})
237 else:
238 env.update({"OMP_PROC_BIND": "CLOSE"})
239 else:
240 if PLATFORM != "darwin":
241 env.update({"OMP_PROC_BIND": "SPREAD"})
243 return env