xmipy

Package for the eXtendend Model Interface. It provides abstract classes, as well as the implementation XmiWrapper

View Source

 1"""
 2Package for the eXtendend Model Interface.
 3It provides abstract classes, as well as the implementation `XmiWrapper`
 4"""
 5
 6# exports
 7from bmipy.bmi import Bmi
 8
 9from xmipy.xmi import Xmi
10from xmipy.xmiwrapper import XmiWrapper
11
12__all__ = ["Bmi", "Xmi", "XmiWrapper"]
13
14__version__ = "1.5.0"

class XmiWrapper(xmipy.Xmi): View Source

 43class XmiWrapper(Xmi):
 44    """The implementation of the XMI"""
 45
 46    def __init__(
 47        self,
 48        lib_path: Union[str, PathLike[Any]],
 49        lib_dependency: Union[str, PathLike[Any], None] = None,
 50        working_directory: Union[str, PathLike[Any], None] = None,
 51        timing: bool = False,
 52        logger_level: Union[str, int] = 0,
 53    ):
 54        """
 55        Constructor of `XmiWrapper`
 56
 57        Next to wrapping C functions of a library exposing XMI,
 58        it also adds timing and logging functionality.
 59        An example for logging can be seen below.
 60
 61        ```
 62        In [1]: from xmipy import XmiWrapper
 63
 64        In [2]: mf6 = XmiWrapper("/path/to/libmf6.so", working_directory="/path/to/sim", logger_level="DEBUG")
 65
 66        In [3]: mf6.initialize()
 67        DEBUG:libmf6.so: execute function: initialize(b'') returned 0
 68
 69        In [4]: mf6.get_start_time()
 70        DEBUG:libmf6.so: execute function: get_start_time(&c_double(0.0)) returned 0
 71        Out[4]: 0.0
 72
 73        In [5]: mf6.get_end_time()
 74        DEBUG:libmf6.so: execute function: get_end_time(&c_double(504.0)) returned 0
 75        Out[5]: 504.0
 76
 77        In [6]: mf6.get_grid_rank(1)
 78        DEBUG:libmf6.so: execute function: get_grid_rank(&c_int(1), &c_int(2)) returned 0
 79        Out[6]: 2
 80
 81        In [7]: mf6.get_value('SLN_1/MXITER')
 82        DEBUG:libmf6.so: execute function: get_var_rank(c_char_p(b'SLN_1/MXITER'), &c_int(0)) returned 0
 83        DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
 84        DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
 85        DEBUG:libmf6.so: execute function: get_value_ptr_int(c_char_p(b'SLN_1/MXITER'), &ndpointer_<i4_1d_1_C) returned 0
 86        DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
 87        DEBUG:libmf6.so: execute function: get_value_ptr_int(c_char_p(b'SLN_1/MXITER'), &ndpointer_<i4_1d_1_C) returned 0
 88        Out[7]: array([25], dtype=int32)
 89        ```
 90
 91        Parameters
 92        ----------
 93        lib_path : Union[str, PathLike]
 94            Path to the shared library
 95
 96        lib_dependency : Union[str, PathLike, None], optional
 97            Path to the dependencies of the shared library, by default None
 98
 99        working_directory : Union[str, PathLike, None], optional
100            The working directory the shared library expects when being called,
101            by default None
102
103        timing : bool, optional
104            Whether timing should be activated, by default False
105
106        logger_level : str, int, optional
107            Logger level, default 0 ("NOTSET"). Accepted values are
108            "DEBUG" (10), "INFO" (20), "WARNING" (30), "ERROR" (40) or
109            "CRITICAL" (50).
110        """
111
112        self._state = State.UNINITIALIZED
113        self.libname = Path(lib_path).name
114        self.logger = get_logger(self.libname, logger_level)
115
116        if lib_dependency:
117            self._add_lib_dependency(lib_dependency)
118        # LoadLibraryEx flag (py38+): LOAD_WITH_ALTERED_SEARCH_PATH 0x08
119        # -> uses the altered search path for resolving dll dependencies
120        # `winmode` has no effect while running on Linux or macOS
121        # Note: this could make xmipy less secure (dll-injection)
122        # Can we get it to work without this flag?
123        self.lib = CDLL(str(lib_path), winmode=0x08)
124
125        if working_directory:
126            self.working_directory = Path(working_directory)
127        else:
128            self.working_directory = Path().cwd()
129        self.timing = timing
130
131        if self.timing:
132            self.timer = Timer(
133                name=self.libname,
134                text="Elapsed time for {name}.{fn_name}: {seconds:0.4f} seconds",
135            )
136
137    def __del__(self) -> None:
138        if self._state == State.INITIALIZED:
139            self.finalize()
140
141    @staticmethod
142    def _add_lib_dependency(lib_dependency: Union[str, PathLike[Any]]) -> None:
143        lib_dependency = str(Path(lib_dependency).absolute())
144        if platform.system() == "Windows":
145            os.environ["PATH"] = lib_dependency + os.pathsep + os.environ["PATH"]
146        else:
147            # Assume a Unix-like system
148            if "LD_LIBRARY_PATH" in os.environ:
149                os.environ["LD_LIBRARY_PATH"] = (
150                    lib_dependency + os.pathsep + os.environ["LD_LIBRARY_PATH"]
151                )
152            else:
153                os.environ["LD_LIBRARY_PATH"] = lib_dependency
154
155    def report_timing_totals(self) -> float:
156        if self.timing:
157            total = self.timer.report_totals()
158            with show_logger_message(self.logger):
159                self.logger.info(
160                    "Total elapsed time for %s: %0.4f seconds",
161                    self.libname,
162                    total,
163                )
164            return total
165        else:
166            raise TimerError("Timing not activated")
167
168    def get_constant_int(self, name: str) -> int:
169        c_var = c_int.in_dll(self.lib, name)
170        return c_var.value
171
172    def set_int(self, name: str, value: int) -> None:
173        c_var = c_int.in_dll(self.lib, name)
174        c_var.value = value
175
176    def initialize(self, config_file: Union[str, PathLike[Any]] = "") -> None:
177        if self._state == State.UNINITIALIZED:
178            with cd(self.working_directory):
179                self._execute_function(self.lib.initialize, os.fsencode(config_file))
180                self._state = State.INITIALIZED
181        else:
182            raise InputError("The library is already initialized")
183
184    def initialize_mpi(self, value: int) -> None:
185        if self._state == State.UNINITIALIZED:
186            with cd(self.working_directory):
187                comm = c_int(value)
188                self._execute_function(self.lib.initialize_mpi, byref(comm))
189                self._state = State.INITIALIZED
190        else:
191            raise InputError("The library is already initialized")
192
193    def update(self) -> None:
194        with cd(self.working_directory):
195            self._execute_function(self.lib.update)
196
197    def update_until(self, time: float) -> None:
198        with cd(self.working_directory):
199            self._execute_function(self.lib.update_until, c_double(time))
200
201    def finalize(self) -> None:
202        if self._state == State.INITIALIZED:
203            with cd(self.working_directory):
204                self._execute_function(self.lib.finalize)
205                self._state = State.UNINITIALIZED
206        else:
207            raise InputError("The library is not initialized yet")
208
209    def get_current_time(self) -> float:
210        current_time = c_double(0.0)
211        self._execute_function(self.lib.get_current_time, byref(current_time))
212        return current_time.value
213
214    def get_start_time(self) -> float:
215        start_time = c_double(0.0)
216        self._execute_function(self.lib.get_start_time, byref(start_time))
217        return start_time.value
218
219    def get_end_time(self) -> float:
220        end_time = c_double(0.0)
221        self._execute_function(self.lib.get_end_time, byref(end_time))
222        return end_time.value
223
224    def get_time_step(self) -> float:
225        dt = c_double(0.0)
226        self._execute_function(self.lib.get_time_step, byref(dt))
227        return dt.value
228
229    def get_component_name(self) -> str:
230        len_name = self.get_constant_int("BMI_LENCOMPONENTNAME")
231        component_name = create_string_buffer(len_name)
232        self._execute_function(self.lib.get_component_name, byref(component_name))
233        return component_name.value.decode("ascii")
234
235    def get_version(self) -> str:
236        len_version = self.get_constant_int("BMI_LENVERSION")
237        version = create_string_buffer(len_version)
238        self._execute_function(self.lib.get_version, byref(version))
239        return version.value.decode("ascii")
240
241    def get_input_item_count(self) -> int:
242        count = c_int(0)
243        self._execute_function(self.lib.get_input_item_count, byref(count))
244        return count.value
245
246    def get_output_item_count(self) -> int:
247        count = c_int(0)
248        self._execute_function(self.lib.get_output_item_count, byref(count))
249        return count.value
250
251    def get_input_var_names(self) -> Tuple[str]:
252        len_address = self.get_constant_int("BMI_LENVARADDRESS")
253        nr_input_vars = self.get_input_item_count()
254        len_names = nr_input_vars * len_address
255        names = create_string_buffer(len_names)
256
257        # get a (1-dim) char array (char*) containing the input variable
258        # names as \x00 terminated sub-strings
259        self._execute_function(self.lib.get_input_var_names, byref(names))
260
261        # decode
262        input_vars: Tuple[str] = tuple(
263            names[i * len_address : (i + 1) * len_address]  # type: ignore
264            .split(b"\0", 1)[0]
265            .decode("ascii")
266            for i in range(nr_input_vars)
267        )
268        return input_vars
269
270    def get_output_var_names(self) -> Tuple[str]:
271        len_address = self.get_constant_int("BMI_LENVARADDRESS")
272        nr_output_vars = self.get_output_item_count()
273        len_names = nr_output_vars * len_address
274        names = create_string_buffer(len_names)
275
276        # get a (1-dim) char array (char*) containing the output variable
277        # names as \x00 terminated sub-strings
278        self._execute_function(self.lib.get_output_var_names, byref(names))
279
280        # decode
281        output_vars: Tuple[str] = tuple(
282            names[i * len_address : (i + 1) * len_address]  # type: ignore
283            .split(b"\0", 1)[0]
284            .decode("ascii")
285            for i in range(nr_output_vars)
286        )
287        return output_vars
288
289    def get_var_grid(self, name: str) -> int:
290        grid_id = c_int(0)
291        self._execute_function(
292            self.lib.get_var_grid,
293            c_char_p(name.encode()),
294            byref(grid_id),
295        )
296        return grid_id.value
297
298    def get_var_type(self, name: str) -> str:
299        len_var_type = self.get_constant_int("BMI_LENVARTYPE")
300        var_type = create_string_buffer(len_var_type)
301        self._execute_function(
302            self.lib.get_var_type,
303            c_char_p(name.encode()),
304            byref(var_type),
305        )
306        return var_type.value.decode()
307
308    # strictly speaking not BMI...
309    def get_var_shape(self, name: str) -> NDArray[np.int32]:
310        rank = self.get_var_rank(name)
311        array = np.zeros(rank, dtype=np.int32)
312        self._execute_function(
313            self.lib.get_var_shape,
314            c_char_p(name.encode()),
315            c_void_p(array.ctypes.data),
316        )
317        return array
318
319    def get_var_rank(self, name: str) -> int:
320        rank = c_int(0)
321        self._execute_function(
322            self.lib.get_var_rank,
323            c_char_p(name.encode()),
324            byref(rank),
325        )
326        return rank.value
327
328    def get_var_units(self, name: str) -> str:
329        raise NotImplementedError
330
331    def get_var_itemsize(self, name: str) -> int:
332        item_size = c_int(0)
333        self._execute_function(
334            self.lib.get_var_itemsize,
335            c_char_p(name.encode()),
336            byref(item_size),
337        )
338        return item_size.value
339
340    def get_var_nbytes(self, name: str) -> int:
341        nbytes = c_int(0)
342        self._execute_function(
343            self.lib.get_var_nbytes,
344            c_char_p(name.encode()),
345            byref(nbytes),
346        )
347        return nbytes.value
348
349    def get_var_location(self, name: str) -> str:
350        raise NotImplementedError
351
352    def get_time_units(self) -> str:
353        raise NotImplementedError
354
355    def get_value(
356        self, name: str, dest: Union[NDArray[Any], None] = None
357    ) -> NDArray[Any]:
358        # make sure that optional array is of correct layout:
359        if dest is not None and not dest.flags["C"]:
360            raise InputError("Array should have C layout")
361
362        # first deal with scalars
363        rank = self.get_var_rank(name)
364        var_type = self.get_var_type(name)
365        var_type_lower = var_type.lower()
366
367        if rank == 0:
368            if var_type_lower.startswith("string"):
369                ilen = self.get_var_nbytes(name)
370                strtype = "<S" + str(ilen + 1)
371                if dest is None:
372                    dest = np.empty(1, dtype=strtype, order="C")
373                self._execute_function(
374                    self.lib.get_value,
375                    c_char_p(name.encode()),
376                    byref(dest.ctypes.data_as(POINTER(c_char))),
377                )
378                dest[0] = dest[0].decode("ascii").strip()
379                return dest.astype(str)
380            else:
381                src = self.get_value_ptr_scalar(name)
382                if dest is None:
383                    return self.get_value_ptr_scalar(name).copy()
384                else:
385                    dest[0] = src[0]
386                    return dest
387
388        var_shape = self.get_var_shape(name)
389
390        if var_type_lower.startswith("double"):
391            if dest is None:
392                dest = np.empty(shape=var_shape, dtype=np.float64, order="C")
393            self._execute_function(
394                self.lib.get_value,
395                c_char_p(name.encode()),
396                byref(dest.ctypes.data_as(POINTER(c_double))),
397            )
398        elif var_type_lower.startswith("int"):
399            if dest is None:
400                dest = np.empty(shape=var_shape, dtype=np.int32, order="C")
401            self._execute_function(
402                self.lib.get_value,
403                c_char_p(name.encode()),
404                byref(dest.ctypes.data_as(POINTER(c_int))),
405            )
406        elif var_type_lower.startswith("string"):
407            if dest is None:
408                if var_shape[0] == 0:
409                    return np.empty((0,), "U1")
410                ilen = self.get_var_nbytes(name) // var_shape[0]
411                strtype = "<S" + str(ilen + 1)
412                dest = np.empty(var_shape[0], dtype=strtype, order="C")
413            self._execute_function(
414                self.lib.get_value,
415                c_char_p(name.encode()),
416                byref(dest.ctypes.data_as(POINTER(c_char))),
417            )
418            for i, x in enumerate(dest):
419                dest[i] = x.decode("ascii").strip()
420            return dest.astype(str)
421        else:
422            raise InputError(f"Unsupported value type {var_type!r}")
423
424        return dest
425
426    def get_value_ptr(self, name: str) -> NDArray[Any]:
427        # first scalars
428        rank = self.get_var_rank(name)
429        if rank == 0:
430            return self.get_value_ptr_scalar(name)
431
432        var_type = self.get_var_type(name)
433        var_type_lower = var_type.lower()
434        shape_array = self.get_var_shape(name)
435
436        # convert shape array to python tuple
437        shape_tuple = tuple(np.trim_zeros(shape_array))
438        ndim = len(shape_tuple)
439
440        if var_type_lower.startswith("double"):
441            arraytype = np.ctypeslib.ndpointer(
442                dtype=np.float64, ndim=ndim, shape=shape_tuple, flags="C"
443            )
444        elif var_type_lower.startswith("float"):
445            arraytype = np.ctypeslib.ndpointer(
446                dtype=np.float32, ndim=ndim, shape=shape_tuple, flags="C"
447            )
448        elif var_type_lower.startswith("int"):
449            arraytype = np.ctypeslib.ndpointer(
450                dtype=np.int32, ndim=ndim, shape=shape_tuple, flags="C"
451            )
452        else:
453            raise InputError(f"Unsupported value type {var_type!r}")
454        values = arraytype()
455        self._execute_function(
456            self.lib.get_value_ptr,
457            c_char_p(name.encode()),
458            byref(values),
459            detail="for variable " + name,
460        )
461        return values.contents
462
463    def get_value_ptr_scalar(self, name: str) -> NDArray[Any]:
464        var_type = self.get_var_type(name)
465        var_type_lower = var_type.lower()
466        if var_type_lower.startswith("double"):
467            arraytype = np.ctypeslib.ndpointer(
468                dtype=np.float64, ndim=1, shape=(1,), flags="C"
469            )
470        elif var_type_lower.startswith("float"):
471            arraytype = np.ctypeslib.ndpointer(
472                dtype=np.float32, ndim=1, shape=(1,), flags="C"
473            )
474        elif var_type_lower.startswith("int"):
475            arraytype = np.ctypeslib.ndpointer(
476                dtype=np.int32, ndim=1, shape=(1,), flags="C"
477            )
478        else:
479            raise InputError(f"Unsupported value type {var_type!r}")
480        values = arraytype()
481        self._execute_function(
482            self.lib.get_value_ptr,
483            c_char_p(name.encode()),
484            byref(values),
485            detail="for variable " + name,
486        )
487        return values.contents
488
489    def get_value_at_indices(
490        self, name: str, dest: NDArray[Any], inds: NDArray[np.int32]
491    ) -> NDArray[Any]:
492        raise NotImplementedError
493
494    def set_value(self, name: str, values: NDArray[Any]) -> None:
495        if not values.flags["C"]:
496            raise InputError("Array should have C layout")
497        var_type = self.get_var_type(name)
498        var_type_lower = var_type.lower()
499        if var_type_lower.startswith("double"):
500            if values.dtype != np.float64:
501                raise InputError("Array should have float64 elements")
502            self._execute_function(
503                self.lib.set_value,
504                c_char_p(name.encode()),
505                byref(values.ctypes.data_as(POINTER(c_double))),
506            )
507        elif var_type_lower.startswith("int"):
508            if values.dtype != np.int32:
509                raise InputError("Array should have int32 elements")
510            self._execute_function(
511                self.lib.set_value,
512                c_char_p(name.encode()),
513                byref(values.ctypes.data_as(POINTER(c_int))),
514            )
515        else:
516            raise InputError("Unsupported value type")
517
518    def set_value_at_indices(
519        self, name: str, inds: NDArray[Any], src: NDArray[Any]
520    ) -> None:
521        raise NotImplementedError
522
523    def get_grid_rank(self, grid: int) -> int:
524        grid_rank = c_int(0)
525        c_grid = c_int(grid)
526        self._execute_function(
527            self.lib.get_grid_rank,
528            byref(c_grid),
529            byref(grid_rank),
530        )
531        return grid_rank.value
532
533    def get_grid_size(self, grid: int) -> int:
534        grid_size = c_int(0)
535        c_grid = c_int(grid)
536        self._execute_function(
537            self.lib.get_grid_size,
538            byref(c_grid),
539            byref(grid_size),
540        )
541        return grid_size.value
542
543    def get_grid_type(self, grid: int) -> str:
544        len_grid_type = self.get_constant_int("BMI_LENGRIDTYPE")
545        grid_type = create_string_buffer(len_grid_type)
546        c_grid = c_int(grid)
547        self._execute_function(
548            self.lib.get_grid_type,
549            byref(c_grid),
550            byref(grid_type),
551        )
552        return grid_type.value.decode()
553
554    def get_grid_shape(self, grid: int, shape: NDArray[np.int32]) -> NDArray[np.int32]:
555        c_grid = c_int(grid)
556        self._execute_function(
557            self.lib.get_grid_shape,
558            byref(c_grid),
559            c_void_p(shape.ctypes.data),
560        )
561        return shape
562
563    def get_grid_spacing(
564        self, grid: int, spacing: NDArray[np.int32]
565    ) -> NDArray[np.int32]:
566        raise NotImplementedError
567
568    def get_grid_origin(
569        self, grid: int, origin: NDArray[np.int32]
570    ) -> NDArray[np.int32]:
571        raise NotImplementedError
572
573    def get_grid_x(self, grid: int, x: NDArray[np.float64]) -> NDArray[np.float64]:
574        c_grid = c_int(grid)
575        self._execute_function(
576            self.lib.get_grid_x,
577            byref(c_grid),
578            c_void_p(x.ctypes.data),
579        )
580        return x
581
582    def get_grid_y(self, grid: int, y: NDArray[np.float64]) -> NDArray[np.float64]:
583        c_grid = c_int(grid)
584        self._execute_function(
585            self.lib.get_grid_y,
586            byref(c_grid),
587            c_void_p(y.ctypes.data),
588        )
589        return y
590
591    def get_grid_z(self, grid: int, z: NDArray[np.float64]) -> NDArray[np.float64]:
592        c_grid = c_int(grid)
593        self._execute_function(
594            self.lib.get_grid_z,
595            byref(c_grid),
596            c_void_p(z.ctypes.data),
597        )
598        return z
599
600    def get_grid_node_count(self, grid: int) -> int:
601        grid_node_count = c_int(0)
602        c_grid = c_int(grid)
603        self._execute_function(
604            self.lib.get_grid_node_count,
605            byref(c_grid),
606            byref(grid_node_count),
607        )
608        return grid_node_count.value
609
610    def get_grid_edge_count(self, grid: int) -> int:
611        raise NotImplementedError
612
613    def get_grid_face_count(self, grid: int) -> int:
614        grid_face_count = c_int(0)
615        c_grid = c_int(grid)
616        self._execute_function(
617            self.lib.get_grid_face_count,
618            byref(c_grid),
619            byref(grid_face_count),
620        )
621        return grid_face_count.value
622
623    def get_grid_edge_nodes(
624        self, grid: int, edge_nodes: NDArray[np.int32]
625    ) -> NDArray[np.int32]:
626        raise NotImplementedError
627
628    def get_grid_face_edges(
629        self, grid: int, face_edges: NDArray[np.int32]
630    ) -> NDArray[np.int32]:
631        raise NotImplementedError
632
633    def get_grid_face_nodes(
634        self, grid: int, face_nodes: NDArray[np.int32]
635    ) -> NDArray[np.int32]:
636        c_grid = c_int(grid)
637        self._execute_function(
638            self.lib.get_grid_face_nodes,
639            byref(c_grid),
640            c_void_p(face_nodes.ctypes.data),
641        )
642        return face_nodes
643
644    def get_grid_nodes_per_face(
645        self, grid: int, nodes_per_face: NDArray[np.int32]
646    ) -> NDArray[np.int32]:
647        c_grid = c_int(grid)
648        self._execute_function(
649            self.lib.get_grid_nodes_per_face,
650            byref(c_grid),
651            c_void_p(nodes_per_face.ctypes.data),
652        )
653        return nodes_per_face
654
655    # ===========================
656    # here starts the XMI
657    # ===========================
658    def prepare_time_step(self, dt: float) -> None:
659        with cd(self.working_directory):
660            c_dt = c_double(dt)
661            self._execute_function(self.lib.prepare_time_step, byref(c_dt))
662
663    def do_time_step(self) -> None:
664        with cd(self.working_directory):
665            self._execute_function(self.lib.do_time_step)
666
667    def finalize_time_step(self) -> None:
668        with cd(self.working_directory):
669            self._execute_function(self.lib.finalize_time_step)
670
671    def get_subcomponent_count(self) -> int:
672        count = c_int(0)
673        self._execute_function(self.lib.get_subcomponent_count, byref(count))
674        return count.value
675
676    def prepare_solve(self, component_id: int = 1) -> None:
677        cid = c_int(component_id)
678        with cd(self.working_directory):
679            self._execute_function(self.lib.prepare_solve, byref(cid))
680
681    def solve(self, component_id: int = 1) -> bool:
682        cid = c_int(component_id)
683        has_converged = c_int(0)
684        with cd(self.working_directory):
685            self._execute_function(self.lib.solve, byref(cid), byref(has_converged))
686        return has_converged.value == 1
687
688    def finalize_solve(self, component_id: int = 1) -> None:
689        cid = c_int(component_id)
690
691        with cd(self.working_directory):
692            self._execute_function(self.lib.finalize_solve, byref(cid))
693
694    def get_var_address(
695        self, var_name: str, component_name: str, subcomponent_name: str = ""
696    ) -> str:
697        len_var_address = self.get_constant_int("BMI_LENVARADDRESS")
698        var_address = create_string_buffer(len_var_address)
699        self._execute_function(
700            self.lib.get_var_address,
701            c_char_p(component_name.upper().encode()),
702            c_char_p(subcomponent_name.upper().encode()),
703            c_char_p(var_name.upper().encode()),
704            byref(var_address),
705        )
706
707        return var_address.value.decode()
708
709    def _execute_function(
710        self, function: Callable[[Any], int], *args: Any, **kwargs: Any
711    ) -> None:
712        """
713        Utility function to execute a BMI function in the kernel and checks its status
714        """
715
716        if self.timing:
717            self.timer.start(function.__name__)
718
719        try:
720            # Execute library function
721            result = function(*args)
722
723            if self.logger.isEnabledFor(logging.DEBUG):
724                self.logger.debug(
725                    "execute function: %s returned %s",
726                    repr_function_call(function.__name__, *args),
727                    result,
728                )
729
730            if result != Status.SUCCESS:
731                msg = "BMI exception in "
732                msg += repr_function_call(function.__name__, *args)
733
734                # try to get detailed error msg, beware:
735                # directly call CDLL methods to avoid recursion
736                try:
737                    len_err_msg = self.get_constant_int("BMI_LENERRMESSAGE")
738                    err_msg = create_string_buffer(len_err_msg)
739                    self.lib.get_last_bmi_error(byref(err_msg))
740
741                    len_name = self.get_constant_int("BMI_LENCOMPONENTNAME")
742                    component_name = create_string_buffer(len_name)
743                    self.lib.get_component_name(byref(component_name))
744
745                    if "detail" in kwargs:
746                        detail = f", details : '{kwargs['detail']}'"
747                    else:
748                        detail = ""
749                    msg += (
750                        f": Message from {component_name.value.decode()} "
751                        + f"'{err_msg.value.decode()}'"
752                        + detail
753                    )
754                except AttributeError:
755                    self.logger.error("Couldn't extract error message")
756
757                raise XMIError(msg)
758
759        finally:
760            if self.timing:
761                self.timer.stop(function.__name__)

The implementation of the XMI

XmiWrapper( lib_path: Union[str, os.PathLike[Any]], lib_dependency: Union[str, os.PathLike[Any], NoneType] = None, working_directory: Union[str, os.PathLike[Any], NoneType] = None, timing: bool = False, logger_level: Union[str, int] = 0) View Source

 46    def __init__(
 47        self,
 48        lib_path: Union[str, PathLike[Any]],
 49        lib_dependency: Union[str, PathLike[Any], None] = None,
 50        working_directory: Union[str, PathLike[Any], None] = None,
 51        timing: bool = False,
 52        logger_level: Union[str, int] = 0,
 53    ):
 54        """
 55        Constructor of `XmiWrapper`
 56
 57        Next to wrapping C functions of a library exposing XMI,
 58        it also adds timing and logging functionality.
 59        An example for logging can be seen below.
 60
 61        ```
 62        In [1]: from xmipy import XmiWrapper
 63
 64        In [2]: mf6 = XmiWrapper("/path/to/libmf6.so", working_directory="/path/to/sim", logger_level="DEBUG")
 65
 66        In [3]: mf6.initialize()
 67        DEBUG:libmf6.so: execute function: initialize(b'') returned 0
 68
 69        In [4]: mf6.get_start_time()
 70        DEBUG:libmf6.so: execute function: get_start_time(&c_double(0.0)) returned 0
 71        Out[4]: 0.0
 72
 73        In [5]: mf6.get_end_time()
 74        DEBUG:libmf6.so: execute function: get_end_time(&c_double(504.0)) returned 0
 75        Out[5]: 504.0
 76
 77        In [6]: mf6.get_grid_rank(1)
 78        DEBUG:libmf6.so: execute function: get_grid_rank(&c_int(1), &c_int(2)) returned 0
 79        Out[6]: 2
 80
 81        In [7]: mf6.get_value('SLN_1/MXITER')
 82        DEBUG:libmf6.so: execute function: get_var_rank(c_char_p(b'SLN_1/MXITER'), &c_int(0)) returned 0
 83        DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
 84        DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
 85        DEBUG:libmf6.so: execute function: get_value_ptr_int(c_char_p(b'SLN_1/MXITER'), &ndpointer_<i4_1d_1_C) returned 0
 86        DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
 87        DEBUG:libmf6.so: execute function: get_value_ptr_int(c_char_p(b'SLN_1/MXITER'), &ndpointer_<i4_1d_1_C) returned 0
 88        Out[7]: array([25], dtype=int32)
 89        ```
 90
 91        Parameters
 92        ----------
 93        lib_path : Union[str, PathLike]
 94            Path to the shared library
 95
 96        lib_dependency : Union[str, PathLike, None], optional
 97            Path to the dependencies of the shared library, by default None
 98
 99        working_directory : Union[str, PathLike, None], optional
100            The working directory the shared library expects when being called,
101            by default None
102
103        timing : bool, optional
104            Whether timing should be activated, by default False
105
106        logger_level : str, int, optional
107            Logger level, default 0 ("NOTSET"). Accepted values are
108            "DEBUG" (10), "INFO" (20), "WARNING" (30), "ERROR" (40) or
109            "CRITICAL" (50).
110        """
111
112        self._state = State.UNINITIALIZED
113        self.libname = Path(lib_path).name
114        self.logger = get_logger(self.libname, logger_level)
115
116        if lib_dependency:
117            self._add_lib_dependency(lib_dependency)
118        # LoadLibraryEx flag (py38+): LOAD_WITH_ALTERED_SEARCH_PATH 0x08
119        # -> uses the altered search path for resolving dll dependencies
120        # `winmode` has no effect while running on Linux or macOS
121        # Note: this could make xmipy less secure (dll-injection)
122        # Can we get it to work without this flag?
123        self.lib = CDLL(str(lib_path), winmode=0x08)
124
125        if working_directory:
126            self.working_directory = Path(working_directory)
127        else:
128            self.working_directory = Path().cwd()
129        self.timing = timing
130
131        if self.timing:
132            self.timer = Timer(
133                name=self.libname,
134                text="Elapsed time for {name}.{fn_name}: {seconds:0.4f} seconds",
135            )

Constructor of XmiWrapper

Next to wrapping C functions of a library exposing XMI, it also adds timing and logging functionality. An example for logging can be seen below.

In [1]: from xmipy import XmiWrapper

In [2]: mf6 = XmiWrapper("/path/to/libmf6.so", working_directory="/path/to/sim", logger_level="DEBUG")

In [3]: mf6.initialize()
DEBUG:libmf6.so: execute function: initialize(b'') returned 0

In [4]: mf6.get_start_time()
DEBUG:libmf6.so: execute function: get_start_time(&c_double(0.0)) returned 0
Out[4]: 0.0

In [5]: mf6.get_end_time()
DEBUG:libmf6.so: execute function: get_end_time(&c_double(504.0)) returned 0
Out[5]: 504.0

In [6]: mf6.get_grid_rank(1)
DEBUG:libmf6.so: execute function: get_grid_rank(&c_int(1), &c_int(2)) returned 0
Out[6]: 2

In [7]: mf6.get_value('SLN_1/MXITER')
DEBUG:libmf6.so: execute function: get_var_rank(c_char_p(b'SLN_1/MXITER'), &c_int(0)) returned 0
DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
DEBUG:libmf6.so: execute function: get_value_ptr_int(c_char_p(b'SLN_1/MXITER'), &ndpointer_<i4_1d_1_C) returned 0
DEBUG:libmf6.so: execute function: get_var_type(c_char_p(b'SLN_1/MXITER'), &c_char_Array_51(b'INTEGER')) returned 0
DEBUG:libmf6.so: execute function: get_value_ptr_int(c_char_p(b'SLN_1/MXITER'), &ndpointer_<i4_1d_1_C) returned 0
Out[7]: array([25], dtype=int32)

Parameters

lib_path : Union[str, PathLike] Path to the shared library

lib_dependency : Union[str, PathLike, None], optional Path to the dependencies of the shared library, by default None

working_directory : Union[str, PathLike, None], optional The working directory the shared library expects when being called, by default None

timing : bool, optional Whether timing should be activated, by default False

logger_level : str, int, optional Logger level, default 0 ("NOTSET"). Accepted values are "DEBUG" (10), "INFO" (20), "WARNING" (30), "ERROR" (40) or "CRITICAL" (50).

libname

logger

lib

timing

def report_timing_totals(self) -> float: View Source

155    def report_timing_totals(self) -> float:
156        if self.timing:
157            total = self.timer.report_totals()
158            with show_logger_message(self.logger):
159                self.logger.info(
160                    "Total elapsed time for %s: %0.4f seconds",
161                    self.libname,
162                    total,
163                )
164            return total
165        else:
166            raise TimerError("Timing not activated")

Logs and returns total time spent

Returns

float Total time spent

Raises

TimerError Raised if timing is not activated

def get_constant_int(self, name: str) -> int: View Source

168    def get_constant_int(self, name: str) -> int:
169        c_var = c_int.in_dll(self.lib, name)
170        return c_var.value

Get a constant integer

Parameters

name : str Name of the constant

Returns

int Constant to be returned

def set_int(self, name: str, value: int) -> None: View Source

172    def set_int(self, name: str, value: int) -> None:
173        c_var = c_int.in_dll(self.lib, name)
174        c_var.value = value

Set integer

Parameters

name : str Integer to be set value : int Value to set the integer to

def initialize(self, config_file: Union[str, os.PathLike[Any]] = '') -> None: View Source

176    def initialize(self, config_file: Union[str, PathLike[Any]] = "") -> None:
177        if self._state == State.UNINITIALIZED:
178            with cd(self.working_directory):
179                self._execute_function(self.lib.initialize, os.fsencode(config_file))
180                self._state = State.INITIALIZED
181        else:
182            raise InputError("The library is already initialized")

Perform startup tasks for the model.

Perform all tasks that take place before entering the model's time loop, including opening files and initializing the model state. Model inputs are read from a text-based configuration file, specified by config_file.

Parameters

config_file : str, optional The path to the model configuration file.

Notes

Models should be refactored, if necessary, to use a configuration file. CSDMS does not impose any constraint on how configuration files are formatted, although YAML is recommended. A template of a model's configuration file with placeholder values is used by the BMI.

def initialize_mpi(self, value: int) -> None: View Source

184    def initialize_mpi(self, value: int) -> None:
185        if self._state == State.UNINITIALIZED:
186            with cd(self.working_directory):
187                comm = c_int(value)
188                self._execute_function(self.lib.initialize_mpi, byref(comm))
189                self._state = State.INITIALIZED
190        else:
191            raise InputError("The library is already initialized")

def update(self) -> None: View Source

193    def update(self) -> None:
194        with cd(self.working_directory):
195            self._execute_function(self.lib.update)

Advance model state by one time step.

Perform all tasks that take place within one pass through the model's time loop. This typically includes incrementing all of the model's state variables. If the model's state variables don't change in time, then they can be computed by the initialize() method and this method can return with no action.

def update_until(self, time: float) -> None: View Source

197    def update_until(self, time: float) -> None:
198        with cd(self.working_directory):
199            self._execute_function(self.lib.update_until, c_double(time))

Advance model state until the given time.

Parameters

time : float A model time later than the current model time.

def finalize(self) -> None: View Source

201    def finalize(self) -> None:
202        if self._state == State.INITIALIZED:
203            with cd(self.working_directory):
204                self._execute_function(self.lib.finalize)
205                self._state = State.UNINITIALIZED
206        else:
207            raise InputError("The library is not initialized yet")

Perform tear-down tasks for the model.

Perform all tasks that take place after exiting the model's time loop. This typically includes deallocating memory, closing files and printing reports.

def get_current_time(self) -> float: View Source

209    def get_current_time(self) -> float:
210        current_time = c_double(0.0)
211        self._execute_function(self.lib.get_current_time, byref(current_time))
212        return current_time.value

Return the current time of the model.

Returns

float The current model time.

def get_start_time(self) -> float: View Source

214    def get_start_time(self) -> float:
215        start_time = c_double(0.0)
216        self._execute_function(self.lib.get_start_time, byref(start_time))
217        return start_time.value

Start time of the model.

Model times should be of type float.

Returns

float The model start time.

def get_end_time(self) -> float: View Source

219    def get_end_time(self) -> float:
220        end_time = c_double(0.0)
221        self._execute_function(self.lib.get_end_time, byref(end_time))
222        return end_time.value

End time of the model.

Returns

float The maximum model time.

def get_time_step(self) -> float: View Source

224    def get_time_step(self) -> float:
225        dt = c_double(0.0)
226        self._execute_function(self.lib.get_time_step, byref(dt))
227        return dt.value

Return the current time step of the model.

The model time step should be of type float.

Returns

float The time step used in model.

def get_component_name(self) -> str: View Source

229    def get_component_name(self) -> str:
230        len_name = self.get_constant_int("BMI_LENCOMPONENTNAME")
231        component_name = create_string_buffer(len_name)
232        self._execute_function(self.lib.get_component_name, byref(component_name))
233        return component_name.value.decode("ascii")

Name of the component.

Returns

str The name of the component.

def get_version(self) -> str: View Source

235    def get_version(self) -> str:
236        len_version = self.get_constant_int("BMI_LENVERSION")
237        version = create_string_buffer(len_version)
238        self._execute_function(self.lib.get_version, byref(version))
239        return version.value.decode("ascii")

Get the version of the kernel.

def get_input_item_count(self) -> int: View Source

241    def get_input_item_count(self) -> int:
242        count = c_int(0)
243        self._execute_function(self.lib.get_input_item_count, byref(count))
244        return count.value

Count of a model's input variables.

Returns

int The number of input variables.

def get_output_item_count(self) -> int: View Source

246    def get_output_item_count(self) -> int:
247        count = c_int(0)
248        self._execute_function(self.lib.get_output_item_count, byref(count))
249        return count.value

Count of a model's output variables.

Returns

int The number of output variables.

def get_input_var_names(self) -> Tuple[str]: View Source

251    def get_input_var_names(self) -> Tuple[str]:
252        len_address = self.get_constant_int("BMI_LENVARADDRESS")
253        nr_input_vars = self.get_input_item_count()
254        len_names = nr_input_vars * len_address
255        names = create_string_buffer(len_names)
256
257        # get a (1-dim) char array (char*) containing the input variable
258        # names as \x00 terminated sub-strings
259        self._execute_function(self.lib.get_input_var_names, byref(names))
260
261        # decode
262        input_vars: Tuple[str] = tuple(
263            names[i * len_address : (i + 1) * len_address]  # type: ignore
264            .split(b"\0", 1)[0]
265            .decode("ascii")
266            for i in range(nr_input_vars)
267        )
268        return input_vars

List of a model's input variables.

Input variable names must be CSDMS Standard Names, also known as long variable names.

Returns

list of str The input variables for the model.

Notes

Standard Names enable the CSDMS framework to determine whether an input variable in one model is equivalent to, or compatible with, an output variable in another model. This allows the framework to automatically connect components.

Standard Names do not have to be used within the model.

def get_output_var_names(self) -> Tuple[str]: View Source

270    def get_output_var_names(self) -> Tuple[str]:
271        len_address = self.get_constant_int("BMI_LENVARADDRESS")
272        nr_output_vars = self.get_output_item_count()
273        len_names = nr_output_vars * len_address
274        names = create_string_buffer(len_names)
275
276        # get a (1-dim) char array (char*) containing the output variable
277        # names as \x00 terminated sub-strings
278        self._execute_function(self.lib.get_output_var_names, byref(names))
279
280        # decode
281        output_vars: Tuple[str] = tuple(
282            names[i * len_address : (i + 1) * len_address]  # type: ignore
283            .split(b"\0", 1)[0]
284            .decode("ascii")
285            for i in range(nr_output_vars)
286        )
287        return output_vars

List of a model's output variables.

Output variable names must be CSDMS Standard Names, also known as long variable names.

Returns

list of str The output variables for the model.

def get_var_grid(self, name: str) -> int: View Source

289    def get_var_grid(self, name: str) -> int:
290        grid_id = c_int(0)
291        self._execute_function(
292            self.lib.get_var_grid,
293            c_char_p(name.encode()),
294            byref(grid_id),
295        )
296        return grid_id.value

Get grid identifier for the given variable.

Parameters

name : str An input or output variable name, a CSDMS Standard Name.

Returns

int The grid identifier.

def get_var_type(self, name: str) -> str: View Source

298    def get_var_type(self, name: str) -> str:
299        len_var_type = self.get_constant_int("BMI_LENVARTYPE")
300        var_type = create_string_buffer(len_var_type)
301        self._execute_function(
302            self.lib.get_var_type,
303            c_char_p(name.encode()),
304            byref(var_type),
305        )
306        return var_type.value.decode()

Get data type of the given variable.

Parameters

name : str An input or output variable name, a CSDMS Standard Name.

Returns

str The Python variable type; e.g., str, int, float.

def get_var_shape(self, name: str) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]: View Source

309    def get_var_shape(self, name: str) -> NDArray[np.int32]:
310        rank = self.get_var_rank(name)
311        array = np.zeros(rank, dtype=np.int32)
312        self._execute_function(
313            self.lib.get_var_shape,
314            c_char_p(name.encode()),
315            c_void_p(array.ctypes.data),
316        )
317        return array

def get_var_rank(self, name: str) -> int: View Source

319    def get_var_rank(self, name: str) -> int:
320        rank = c_int(0)
321        self._execute_function(
322            self.lib.get_var_rank,
323            c_char_p(name.encode()),
324            byref(rank),
325        )
326        return rank.value

def get_var_units(self, name: str) -> str: View Source

328    def get_var_units(self, name: str) -> str:
329        raise NotImplementedError

Get units of the given variable.

Standard unit names, in lower case, should be used, such as meters or seconds. Standard abbreviations, like m for meters, are also supported. For variables with compound units, each unit name is separated by a single space, with exponents other than 1 placed immediately after the name, as in m s-1 for velocity, W m-2 for an energy flux, or km2 for an area.

Parameters

name : str An input or output variable name, a CSDMS Standard Name.

Returns

str The variable units.

Notes

CSDMS uses the UDUNITS standard from Unidata.

def get_var_itemsize(self, name: str) -> int: View Source

331    def get_var_itemsize(self, name: str) -> int:
332        item_size = c_int(0)
333        self._execute_function(
334            self.lib.get_var_itemsize,
335            c_char_p(name.encode()),
336            byref(item_size),
337        )
338        return item_size.value

Get memory use for each array element in bytes.

Parameters

name : str An input or output variable name, a CSDMS Standard Name.

Returns

int Item size in bytes.

def get_var_nbytes(self, name: str) -> int: View Source

340    def get_var_nbytes(self, name: str) -> int:
341        nbytes = c_int(0)
342        self._execute_function(
343            self.lib.get_var_nbytes,
344            c_char_p(name.encode()),
345            byref(nbytes),
346        )
347        return nbytes.value

Get size, in bytes, of the given variable.

Parameters

name : str An input or output variable name, a CSDMS Standard Name.

Returns

int The size of the variable, counted in bytes.

def get_var_location(self, name: str) -> str: View Source

349    def get_var_location(self, name: str) -> str:
350        raise NotImplementedError

Get the grid element type that the a given variable is defined on.

The grid topology can be composed of nodes, edges, and faces.

node A point that has a coordinate pair or triplet: the most basic element of the topology.

edge A line or curve bounded by two nodes.

face A plane or surface enclosed by a set of edges. In a 2D horizontal application one may consider the word “polygon”, but in the hierarchy of elements the word “face” is most common.

Parameters

name : str An input or output variable name, a CSDMS Standard Name.

Returns

str The grid location on which the variable is defined. Must be one of "node", "edge", or "face".

Notes

CSDMS uses the ugrid conventions to define unstructured grids.

def get_time_units(self) -> str: View Source

352    def get_time_units(self) -> str:
353        raise NotImplementedError

Time units of the model.

Returns

str The model time unit; e.g., days or s.

Notes

CSDMS uses the UDUNITS standard from Unidata.

def get_value( self, name: str, dest: Optional[numpy.ndarray[Any, numpy.dtype[Any]]] = None) -> numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]: View Source

355    def get_value(
356        self, name: str, dest: Union[NDArray[Any], None] = None
357    ) -> NDArray[Any]:
358        # make sure that optional array is of correct layout:
359        if dest is not None and not dest.flags["C"]:
360            raise InputError("Array should have C layout")
361
362        # first deal with scalars
363        rank = self.get_var_rank(name)
364        var_type = self.get_var_type(name)
365        var_type_lower = var_type.lower()
366
367        if rank == 0:
368            if var_type_lower.startswith("string"):
369                ilen = self.get_var_nbytes(name)
370                strtype = "<S" + str(ilen + 1)
371                if dest is None:
372                    dest = np.empty(1, dtype=strtype, order="C")
373                self._execute_function(
374                    self.lib.get_value,
375                    c_char_p(name.encode()),
376                    byref(dest.ctypes.data_as(POINTER(c_char))),
377                )
378                dest[0] = dest[0].decode("ascii").strip()
379                return dest.astype(str)
380            else:
381                src = self.get_value_ptr_scalar(name)
382                if dest is None:
383                    return self.get_value_ptr_scalar(name).copy()
384                else:
385                    dest[0] = src[0]
386                    return dest
387
388        var_shape = self.get_var_shape(name)
389
390        if var_type_lower.startswith("double"):
391            if dest is None:
392                dest = np.empty(shape=var_shape, dtype=np.float64, order="C")
393            self._execute_function(
394                self.lib.get_value,
395                c_char_p(name.encode()),
396                byref(dest.ctypes.data_as(POINTER(c_double))),
397            )
398        elif var_type_lower.startswith("int"):
399            if dest is None:
400                dest = np.empty(shape=var_shape, dtype=np.int32, order="C")
401            self._execute_function(
402                self.lib.get_value,
403                c_char_p(name.encode()),
404                byref(dest.ctypes.data_as(POINTER(c_int))),
405            )
406        elif var_type_lower.startswith("string"):
407            if dest is None:
408                if var_shape[0] == 0:
409                    return np.empty((0,), "U1")
410                ilen = self.get_var_nbytes(name) // var_shape[0]
411                strtype = "<S" + str(ilen + 1)
412                dest = np.empty(var_shape[0], dtype=strtype, order="C")
413            self._execute_function(
414                self.lib.get_value,
415                c_char_p(name.encode()),
416                byref(dest.ctypes.data_as(POINTER(c_char))),
417            )
418            for i, x in enumerate(dest):
419                dest[i] = x.decode("ascii").strip()
420            return dest.astype(str)
421        else:
422            raise InputError(f"Unsupported value type {var_type!r}")
423
424        return dest

Get a copy of values of the given variable.

This is a getter for the model, used to access the model's current state. It returns a copy of a model variable, with the return type, size and rank dependent on the variable.

Parameters

name : str An input or output variable name, a CSDMS Standard Name. dest : ndarray A numpy array into which to place the values.

Returns

ndarray The same numpy array that was passed as an input buffer.

def get_value_ptr(self, name: str) -> numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]: View Source

426    def get_value_ptr(self, name: str) -> NDArray[Any]:
427        # first scalars
428        rank = self.get_var_rank(name)
429        if rank == 0:
430            return self.get_value_ptr_scalar(name)
431
432        var_type = self.get_var_type(name)
433        var_type_lower = var_type.lower()
434        shape_array = self.get_var_shape(name)
435
436        # convert shape array to python tuple
437        shape_tuple = tuple(np.trim_zeros(shape_array))
438        ndim = len(shape_tuple)
439
440        if var_type_lower.startswith("double"):
441            arraytype = np.ctypeslib.ndpointer(
442                dtype=np.float64, ndim=ndim, shape=shape_tuple, flags="C"
443            )
444        elif var_type_lower.startswith("float"):
445            arraytype = np.ctypeslib.ndpointer(
446                dtype=np.float32, ndim=ndim, shape=shape_tuple, flags="C"
447            )
448        elif var_type_lower.startswith("int"):
449            arraytype = np.ctypeslib.ndpointer(
450                dtype=np.int32, ndim=ndim, shape=shape_tuple, flags="C"
451            )
452        else:
453            raise InputError(f"Unsupported value type {var_type!r}")
454        values = arraytype()
455        self._execute_function(
456            self.lib.get_value_ptr,
457            c_char_p(name.encode()),
458            byref(values),
459            detail="for variable " + name,
460        )
461        return values.contents

Get a reference to values of the given variable.

This is a getter for the model, used to access the model's current state. It returns a reference to a model variable, with the return type, size and rank dependent on the variable.

Parameters

name : str An input or output variable name, a CSDMS Standard Name.

Returns

array_like A reference to a model variable.

def get_value_ptr_scalar(self, name: str) -> numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]: View Source

463    def get_value_ptr_scalar(self, name: str) -> NDArray[Any]:
464        var_type = self.get_var_type(name)
465        var_type_lower = var_type.lower()
466        if var_type_lower.startswith("double"):
467            arraytype = np.ctypeslib.ndpointer(
468                dtype=np.float64, ndim=1, shape=(1,), flags="C"
469            )
470        elif var_type_lower.startswith("float"):
471            arraytype = np.ctypeslib.ndpointer(
472                dtype=np.float32, ndim=1, shape=(1,), flags="C"
473            )
474        elif var_type_lower.startswith("int"):
475            arraytype = np.ctypeslib.ndpointer(
476                dtype=np.int32, ndim=1, shape=(1,), flags="C"
477            )
478        else:
479            raise InputError(f"Unsupported value type {var_type!r}")
480        values = arraytype()
481        self._execute_function(
482            self.lib.get_value_ptr,
483            c_char_p(name.encode()),
484            byref(values),
485            detail="for variable " + name,
486        )
487        return values.contents

def get_value_at_indices( self, name: str, dest: numpy.ndarray[typing.Any, numpy.dtype[typing.Any]], inds: numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]) -> numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]: View Source

489    def get_value_at_indices(
490        self, name: str, dest: NDArray[Any], inds: NDArray[np.int32]
491    ) -> NDArray[Any]:
492        raise NotImplementedError

Get values at particular indices.

Parameters

name : str An input or output variable name, a CSDMS Standard Name. dest : ndarray A numpy array into which to place the values. inds : array_like The indices into the variable array.

Returns

array_like Value of the model variable at the given location.

def set_value( self, name: str, values: numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]) -> None: View Source

494    def set_value(self, name: str, values: NDArray[Any]) -> None:
495        if not values.flags["C"]:
496            raise InputError("Array should have C layout")
497        var_type = self.get_var_type(name)
498        var_type_lower = var_type.lower()
499        if var_type_lower.startswith("double"):
500            if values.dtype != np.float64:
501                raise InputError("Array should have float64 elements")
502            self._execute_function(
503                self.lib.set_value,
504                c_char_p(name.encode()),
505                byref(values.ctypes.data_as(POINTER(c_double))),
506            )
507        elif var_type_lower.startswith("int"):
508            if values.dtype != np.int32:
509                raise InputError("Array should have int32 elements")
510            self._execute_function(
511                self.lib.set_value,
512                c_char_p(name.encode()),
513                byref(values.ctypes.data_as(POINTER(c_int))),
514            )
515        else:
516            raise InputError("Unsupported value type")

Specify a new value for a model variable.

This is the setter for the model, used to change the model's current state. It accepts, through src, a new value for a model variable, with the type, size and rank of src dependent on the variable.

Parameters

name : str An input or output variable name, a CSDMS Standard Name. src : array_like The new value for the specified variable.

def set_value_at_indices( self, name: str, inds: numpy.ndarray[typing.Any, numpy.dtype[typing.Any]], src: numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]) -> None: View Source

518    def set_value_at_indices(
519        self, name: str, inds: NDArray[Any], src: NDArray[Any]
520    ) -> None:
521        raise NotImplementedError

Specify a new value for a model variable at particular indices.

Parameters

name : str An input or output variable name, a CSDMS Standard Name. inds : array_like The indices into the variable array. src : array_like The new value for the specified variable.

def get_grid_rank(self, grid: int) -> int: View Source

523    def get_grid_rank(self, grid: int) -> int:
524        grid_rank = c_int(0)
525        c_grid = c_int(grid)
526        self._execute_function(
527            self.lib.get_grid_rank,
528            byref(c_grid),
529            byref(grid_rank),
530        )
531        return grid_rank.value

Get number of dimensions of the computational grid.

Parameters

grid : int A grid identifier.

Returns

int Rank of the grid.

def get_grid_size(self, grid: int) -> int: View Source

533    def get_grid_size(self, grid: int) -> int:
534        grid_size = c_int(0)
535        c_grid = c_int(grid)
536        self._execute_function(
537            self.lib.get_grid_size,
538            byref(c_grid),
539            byref(grid_size),
540        )
541        return grid_size.value

Get the total number of elements in the computational grid.

Parameters

grid : int A grid identifier.

Returns

int Size of the grid.

def get_grid_type(self, grid: int) -> str: View Source

543    def get_grid_type(self, grid: int) -> str:
544        len_grid_type = self.get_constant_int("BMI_LENGRIDTYPE")
545        grid_type = create_string_buffer(len_grid_type)
546        c_grid = c_int(grid)
547        self._execute_function(
548            self.lib.get_grid_type,
549            byref(c_grid),
550            byref(grid_type),
551        )
552        return grid_type.value.decode()

Get the grid type as a string.

Parameters

grid : int A grid identifier.

Returns

str Type of grid as a string.

def get_grid_shape( self, grid: int, shape: numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]: View Source

554    def get_grid_shape(self, grid: int, shape: NDArray[np.int32]) -> NDArray[np.int32]:
555        c_grid = c_int(grid)
556        self._execute_function(
557            self.lib.get_grid_shape,
558            byref(c_grid),
559            c_void_p(shape.ctypes.data),
560        )
561        return shape

Get dimensions of the computational grid.

Parameters

grid : int A grid identifier. shape : ndarray of int, shape (ndim,) A numpy array into which to place the shape of the grid.

Returns

ndarray of int The input numpy array that holds the grid's shape.

def get_grid_spacing( self, grid: int, spacing: numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]: View Source

563    def get_grid_spacing(
564        self, grid: int, spacing: NDArray[np.int32]
565    ) -> NDArray[np.int32]:
566        raise NotImplementedError

Get distance between nodes of the computational grid.

Parameters

grid : int A grid identifier. spacing : ndarray of float, shape (ndim,) A numpy array to hold the spacing between grid rows and columns.

Returns

ndarray of float The input numpy array that holds the grid's spacing.

def get_grid_origin( self, grid: int, origin: numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]: View Source

568    def get_grid_origin(
569        self, grid: int, origin: NDArray[np.int32]
570    ) -> NDArray[np.int32]:
571        raise NotImplementedError

Get coordinates for the lower-left corner of the computational grid.

Parameters

grid : int A grid identifier. origin : ndarray of float, shape (ndim,) A numpy array to hold the coordinates of the lower-left corner of the grid.

Returns

ndarray of float The input numpy array that holds the coordinates of the grid's lower-left corner.

def get_grid_x( self, grid: int, x: numpy.ndarray[typing.Any, numpy.dtype[numpy.float64]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.float64]]: View Source

573    def get_grid_x(self, grid: int, x: NDArray[np.float64]) -> NDArray[np.float64]:
574        c_grid = c_int(grid)
575        self._execute_function(
576            self.lib.get_grid_x,
577            byref(c_grid),
578            c_void_p(x.ctypes.data),
579        )
580        return x

Get coordinates of grid nodes in the x direction.

Parameters

grid : int A grid identifier. x : ndarray of float, shape (nrows,) A numpy array to hold the x-coordinates of the grid node columns.

Returns

ndarray of float The input numpy array that holds the grid's column x-coordinates.

def get_grid_y( self, grid: int, y: numpy.ndarray[typing.Any, numpy.dtype[numpy.float64]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.float64]]: View Source

582    def get_grid_y(self, grid: int, y: NDArray[np.float64]) -> NDArray[np.float64]:
583        c_grid = c_int(grid)
584        self._execute_function(
585            self.lib.get_grid_y,
586            byref(c_grid),
587            c_void_p(y.ctypes.data),
588        )
589        return y

Get coordinates of grid nodes in the y direction.

Parameters

grid : int A grid identifier. y : ndarray of float, shape (ncols,) A numpy array to hold the y-coordinates of the grid node rows.

Returns

ndarray of float The input numpy array that holds the grid's row y-coordinates.

def get_grid_z( self, grid: int, z: numpy.ndarray[typing.Any, numpy.dtype[numpy.float64]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.float64]]: View Source

591    def get_grid_z(self, grid: int, z: NDArray[np.float64]) -> NDArray[np.float64]:
592        c_grid = c_int(grid)
593        self._execute_function(
594            self.lib.get_grid_z,
595            byref(c_grid),
596            c_void_p(z.ctypes.data),
597        )
598        return z

Get coordinates of grid nodes in the z direction.

Parameters

grid : int A grid identifier. z : ndarray of float, shape (nlayers,) A numpy array to hold the z-coordinates of the grid nodes layers.

Returns

ndarray of float The input numpy array that holds the grid's layer z-coordinates.

def get_grid_node_count(self, grid: int) -> int: View Source

600    def get_grid_node_count(self, grid: int) -> int:
601        grid_node_count = c_int(0)
602        c_grid = c_int(grid)
603        self._execute_function(
604            self.lib.get_grid_node_count,
605            byref(c_grid),
606            byref(grid_node_count),
607        )
608        return grid_node_count.value

Get the number of nodes in the grid.

Parameters

grid : int A grid identifier.

Returns

int The total number of grid nodes.

def get_grid_edge_count(self, grid: int) -> int: View Source

610    def get_grid_edge_count(self, grid: int) -> int:
611        raise NotImplementedError

Get the number of edges in the grid.

Parameters

grid : int A grid identifier.

Returns

int The total number of grid edges.

def get_grid_face_count(self, grid: int) -> int: View Source

613    def get_grid_face_count(self, grid: int) -> int:
614        grid_face_count = c_int(0)
615        c_grid = c_int(grid)
616        self._execute_function(
617            self.lib.get_grid_face_count,
618            byref(c_grid),
619            byref(grid_face_count),
620        )
621        return grid_face_count.value

Get the number of faces in the grid.

Parameters

grid : int A grid identifier.

Returns

int The total number of grid faces.

def get_grid_edge_nodes( self, grid: int, edge_nodes: numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]: View Source

623    def get_grid_edge_nodes(
624        self, grid: int, edge_nodes: NDArray[np.int32]
625    ) -> NDArray[np.int32]:
626        raise NotImplementedError

Get the edge-node connectivity.

Parameters

grid : int A grid identifier. edge_nodes : ndarray of int, shape (2 x nnodes,) A numpy array to place the edge-node connectivity. For each edge, connectivity is given as node at edge tail, followed by node at edge head.

Returns

ndarray of int The input numpy array that holds the edge-node connectivity.

def get_grid_face_edges( self, grid: int, face_edges: numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]: View Source

628    def get_grid_face_edges(
629        self, grid: int, face_edges: NDArray[np.int32]
630    ) -> NDArray[np.int32]:
631        raise NotImplementedError

Get the face-edge connectivity.

Parameters

grid : int A grid identifier. face_edges : ndarray of int A numpy array to place the face-edge connectivity.

Returns

ndarray of int The input numpy array that holds the face-edge connectivity.

def get_grid_face_nodes( self, grid: int, face_nodes: numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]: View Source

633    def get_grid_face_nodes(
634        self, grid: int, face_nodes: NDArray[np.int32]
635    ) -> NDArray[np.int32]:
636        c_grid = c_int(grid)
637        self._execute_function(
638            self.lib.get_grid_face_nodes,
639            byref(c_grid),
640            c_void_p(face_nodes.ctypes.data),
641        )
642        return face_nodes

Get the face-node connectivity.

Parameters

grid : int A grid identifier. face_nodes : ndarray of int A numpy array to place the face-node connectivity. For each face, the nodes (listed in a counter-clockwise direction) that form the boundary of the face.

Returns

ndarray of int The input numpy array that holds the face-node connectivity.

def get_grid_nodes_per_face( self, grid: int, nodes_per_face: numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]) -> numpy.ndarray[typing.Any, numpy.dtype[numpy.int32]]: View Source

644    def get_grid_nodes_per_face(
645        self, grid: int, nodes_per_face: NDArray[np.int32]
646    ) -> NDArray[np.int32]:
647        c_grid = c_int(grid)
648        self._execute_function(
649            self.lib.get_grid_nodes_per_face,
650            byref(c_grid),
651            c_void_p(nodes_per_face.ctypes.data),
652        )
653        return nodes_per_face

Get the number of nodes for each face.

Parameters

grid : int A grid identifier. nodes_per_face : ndarray of int, shape (nfaces,) A numpy array to place the number of nodes per face.

Returns

ndarray of int The input numpy array that holds the number of nodes per face.

def prepare_time_step(self, dt: float) -> None: View Source

658    def prepare_time_step(self, dt: float) -> None:
659        with cd(self.working_directory):
660            c_dt = c_double(dt)
661            self._execute_function(self.lib.prepare_time_step, byref(c_dt))

Prepare a single time step.

Read data from input files and calculate the current time step length and the simulation time at the end of the current time step.

Parameters

dt : float Model time step length for the current time step.

def do_time_step(self) -> None: View Source

663    def do_time_step(self) -> None:
664        with cd(self.working_directory):
665            self._execute_function(self.lib.do_time_step)

Perform a single time step.

Build and solve a time step to completion. This method encapsulates the prepare_solve, solve, and finalize_solve methods.

def finalize_time_step(self) -> None: View Source

667    def finalize_time_step(self) -> None:
668        with cd(self.working_directory):
669            self._execute_function(self.lib.finalize_time_step)

Finalize the time step.

Write messages and output after model convergence has been achieved.

def get_subcomponent_count(self) -> int: View Source

671    def get_subcomponent_count(self) -> int:
672        count = c_int(0)
673        self._execute_function(self.lib.get_subcomponent_count, byref(count))
674        return count.value

Get the number of components in the simulation.

For most applications, this number will be equal to 1.

Returns

int The number of components in the simulation.

def prepare_solve(self, component_id: int = 1) -> None: View Source

676    def prepare_solve(self, component_id: int = 1) -> None:
677        cid = c_int(component_id)
678        with cd(self.working_directory):
679            self._execute_function(self.lib.prepare_solve, byref(cid))

Prepare for solving the system of equations.

This preparation mostly consists of advancing model states.

Parameters

component_id : int Component id number.

def solve(self, component_id: int = 1) -> bool: View Source

681    def solve(self, component_id: int = 1) -> bool:
682        cid = c_int(component_id)
683        has_converged = c_int(0)
684        with cd(self.working_directory):
685            self._execute_function(self.lib.solve, byref(cid), byref(has_converged))
686        return has_converged.value == 1

Build and solve the linear system of equations.

Formulate the system of equations for this outer (Picard) iteration and solve. New data used when formulating the system of equations can be injected prior to solve(). Before calling this, a matching call to prepare_solve() should be performed.

Parameters

component_id : int Component id number.

Returns

bool Boolean indicating if convergence has been achieved.

def finalize_solve(self, component_id: int = 1) -> None: View Source

688    def finalize_solve(self, component_id: int = 1) -> None:
689        cid = c_int(component_id)
690
691        with cd(self.working_directory):
692            self._execute_function(self.lib.finalize_solve, byref(cid))

Finalize the model solve for the current time step.

Finalize model variables prior to calling finalize_time_step(). This method should always be called after calls to prepare_solve() and solve()

Parameters

component_id : int Component id number.

def get_var_address( self, var_name: str, component_name: str, subcomponent_name: str = '') -> str: View Source

694    def get_var_address(
695        self, var_name: str, component_name: str, subcomponent_name: str = ""
696    ) -> str:
697        len_var_address = self.get_constant_int("BMI_LENVARADDRESS")
698        var_address = create_string_buffer(len_var_address)
699        self._execute_function(
700            self.lib.get_var_address,
701            c_char_p(component_name.upper().encode()),
702            c_char_p(subcomponent_name.upper().encode()),
703            c_char_p(var_name.upper().encode()),
704            byref(var_address),
705        )
706
707        return var_address.value.decode()

Get the address of a given variable

Parameters

var_name : str The variable name component_name : str The name of the component subcomponent_name : str, optional If applicable the name of the subcomponent, by default ""

Returns

str The variable address