Source code for grm.args

This module gives access to the ArgumentContainers exposed by GRM.

It is used to pass plotting data, settings and other data to GRM.

import numpy as np
from ctypes import c_int, c_uint, c_double, c_char_p, c_void_p
from ctypes import POINTER, create_string_buffer

from typing import Union, Tuple, List, Dict, Optional

from gr import _require_runtime_version, _RUNTIME_VERSION

from . import _grm, _encode_str_to_char_p

_ElemType = Union[
    List[Union[int, float]],
    List[Union[dict, "_ArgumentContainer"]],
    Tuple[Union[int, float], ...],
    Tuple[str, ...],
    Tuple[Union[dict, "_ArgumentContainer"], ...],

[docs]class _ArgumentContainer: def __init__(self, ptr: c_void_p, params: Optional[Dict[str, _ElemType]] = None) -> None: """ Initialize the class using the given pointer and optional params to insert directly. :param ptr: The pointer returned by grm_args_new :param params: The data to set after init """ self._ptr = ptr # type: Union[c_void_p, None] self._bufs = {} # type: Dict[str, Any] self._is_child = False if params is not None: self.update(params)
[docs] def update(self, params: Dict[str, _ElemType]) -> None: """ Update the argument container with the given dictionary params, by calling self.push(k, v) on each item. :param params: The data to set. On each element, self[k] = v is called, inserting the element. """ for k, v in params.items(): self[k] = v
@property def ptr(self) -> c_void_p: """ Return the internal pointer of the argument container. Should not be modified or otherwise dealt with, primarily for use of internal classes. :raises ValueError: if the container was already deleted. """ if self._ptr is None: raise ValueError("Pointer already dead!") return self._ptr
[docs] def clear(self) -> None: """ Clear the argument container and frees all resources held by bufs. :raises ValueError: if the container was already deleted. """ _grm.grm_args_clear(self.ptr) self._bufs = {}
[docs] def remove(self, name: str) -> None: """ Remove the given key `name` from the argument container, and frees the ressource held by it. `name in self` should be false after that. :param name: the key to remove :raises ValueError: if the container was already deleted. """ _grm.grm_args_remove(self.ptr, _encode_str_to_char_p(name)) del self._bufs[name]
[docs] def contains(self, name: str) -> bool: """ If the key `name` is contained in the argument, then return true. :param name: the key to check for. :raises ValueError: if the container was already deleted. """ return _grm.grm_args_contains(self.ptr, _encode_str_to_char_p(name)) == 1
def __setitem__(self, key: str, value: _ElemType) -> None: self.push(key, value) def __delitem__(self, key: str) -> None: self.remove(key) def __contains__(self, key: str) -> bool: return self.contains(key)
[docs] def delete(self) -> None: """ De-Initialises a argument container (e.g. clear and destroy). """ if not self._is_child: _grm.grm_args_delete(self.ptr) self._delete()
def _delete(self) -> None: """ Frees the internal bufs and passes _delete to subcontainers. """ if self._bufs is None: return for x in self._bufs.values(): if isinstance(x, tuple): for y in x[1]: y._delete() self._ptr = None del self._bufs
[docs] def push(self, name: str, values_to_insert: _ElemType) -> bool: """ Pushes the argument with name to the argument container args_ptr, which should have been created using args_new. This function also silently overwrites entries with the same name. You can always mix int values and float values, but they will then all be converted to floats. One-dimensional numpy.ndarray with either Float64 or Int32 can also be passed. Multi-dimensional numpy.ndarrays with either Float64 or Int32 can also be passed, they will be passed flattened, but :code:`name` _dims is populated with the shape beforehand. The exception is if the name is `absolute`, `relative` or `error` where ndarrays with a shape of [2, N] are passed as :code:`nDD`. :param name: The key to insert. :param values_to_insert: The data to insert. :raises TypeError: if name or values (or the child elements of values) are of no correct type. :raises ValueError: if one of the _ArgumentContainer elements is already a child of another or the container is already deleted. """ # Remove type annotation to silence mypy values = values_to_insert # type: Any if not isinstance(name, str): raise TypeError("Name must be a string!") if isinstance(values, (int, float, str, dict, _ArgumentContainer)): values = [values] if not isinstance(values, (tuple, list, np.ndarray)): raise TypeError( "Values must be int/int-array, float/float-array or string/string-array or dict/dict-array, _ArgumentContainer/_ArgumentContainer-array" ) values_orig = values if isinstance(values, np.ndarray): if values.ndim > 1: if values.ndim == 2 and values.shape[0] == 2 and name in ["error", "relative", "absolute"]: if values.dtype == np.float64: values_1 = values[0].ctypes.data_as(POINTER(c_double)) values_2 = values[1].ctypes.data_as(POINTER(c_double)) type_spec = create_string_buffer(b"nDD") elif values.dtype == np.int32: values_1 = values[0].ctypes.data_as(POINTER(c_int)) values_2 = values[1].ctypes.data_as(POINTER(c_int)) type_spec = create_string_buffer(b"nII") else: raise TypeError( "The ndarray has type " + + ", but it must be either int32 or float64" ) self._bufs[name] = [values_1, values_2] result = _grm.grm_args_push( self.ptr, _encode_str_to_char_p(name), type_spec, c_uint(values.shape[1]), values_1, values_2 ) return result != 0 # TODO: Exceptions self[name + "_dims"] = values.shape values_orig = values = values.ravel() if values.dtype == np.float64: type_spec = create_string_buffer(b"nD") values = values.ctypes.data_as(POINTER(c_double)) elif values.dtype == np.int32: type_spec = create_string_buffer(b"nI") values = values.ctypes.data_as(POINTER(c_int)) else: raise TypeError("The given ndarray does not have the correct type.") self._bufs[name] = values else: typ = None for x in values: if typ is None: typ = type(x) elif typ == type(x): pass elif typ == int and isinstance(x, float): typ = float elif typ == float and isinstance(x, int): pass elif typ == dict and isinstance(x, _ArgumentContainer): pass elif typ == _ArgumentContainer and isinstance(x, dict): typ = dict else: raise TypeError("All values in the array must be of the same type!") if typ == int: type_spec = create_string_buffer(b"nI") values = (c_int * len(values))(*values) self._bufs[name] = values elif typ == float: type_spec = create_string_buffer(b"nD") values = (c_double * len(values))(*values) self._bufs[name] = values elif typ == str: type_spec = create_string_buffer(b"nS") values = (c_char_p * len(values))(*[_encode_str_to_char_p(x) for x in values]) self._bufs[name] = values elif typ == _ArgumentContainer or typ == dict: values_orig = [new(x) if isinstance(x, dict) else x for x in values_orig] for x in values_orig: if x._is_child: raise ValueError("This ArgumentContainer is already a child of another!") x._is_child = True type_spec = create_string_buffer(b"nA") values = (c_void_p * len(values_orig))(*[x.ptr for x in values_orig]) self._bufs[name] = ( values, values_orig, ) # This also stores the ArgumentContainers, so if 'self' is destructed, they loose a reference, and can be destructed, too. else: raise TypeError("Unsupported type: " + repr(typ)) length = c_uint(len(values_orig)) result = _grm.grm_args_push(self.ptr, _encode_str_to_char_p(name), type_spec, length, values) if result == 0: return False # TODO: Exceptions? return True
def __del__(self) -> None: """ Destructor to optionally free resources and destroy the c container, if not already done. """ if self._ptr is not None: self.delete()
[docs]@_require_runtime_version(0, 47, 0) def new(params: Optional[Dict[str, _ElemType]] = None) -> _ArgumentContainer: """ Initialise a new argument container with optional initialization data. :param params: Each element in this dictionary is written into the container at initialization time. """ return _ArgumentContainer(_grm.grm_args_new(), params)
if _RUNTIME_VERSION >= (0, 47, 0, 0): _grm.grm_args_new.argtypes = [] _grm.grm_args_new.restype = c_void_p # Import note: `grm_args_push` is a variadic function and only non-variadic function arguments must be specified in # the `argtypes` attribute. The variadic part must not be declared here! Otherwise, calling `grm_args_push` will # fail on some platforms, e.g. Apple Silicon. For more information see # <>. _grm.grm_args_push.argtypes = [c_void_p, c_char_p, c_char_p] _grm.grm_args_push.restype = c_int _grm.grm_args_clear.argtypes = [c_void_p] _grm.grm_args_clear.restype = None _grm.grm_args_remove.argtypes = [c_void_p, c_char_p] _grm.grm_args_remove.restype = None _grm.grm_args_contains.argtypes = [c_void_p, c_char_p] _grm.grm_args_contains.restype = c_int _grm.grm_args_delete.argtypes = [c_void_p] _grm.grm_args_delete.restype = None __all__ = ["new"]