src/bindings/python/bt2/bt2/query_executor.py

   1 # SPDX-License-Identifier: MIT
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4
   5 from bt2 import native_bt, object, utils
   6 from bt2 import interrupter as bt2_interrupter
   7 from bt2 import value as bt2_value
   8 import bt2
   9
  10
  11 def _bt2_component():
  12     from bt2 import component as bt2_component
  13
  14     return bt2_component
  15
  16
  17 class _QueryExecutorCommon:
  18     @property
  19     def _common_ptr(self):
  20         return self._as_query_executor_ptr()
  21
  22     @property
  23     def is_interrupted(self):
  24         is_interrupted = native_bt.query_executor_is_interrupted(self._common_ptr)
  25         return bool(is_interrupted)
  26
  27     @property
  28     def logging_level(self):
  29         return native_bt.query_executor_get_logging_level(self._common_ptr)
  30
  31
  32 class QueryExecutor(object._SharedObject, _QueryExecutorCommon):
  33     _get_ref = staticmethod(native_bt.query_executor_get_ref)
  34     _put_ref = staticmethod(native_bt.query_executor_put_ref)
  35
  36     def _as_query_executor_ptr(self):
  37         return self._ptr
  38
  39     def __init__(self, component_class, object_name, params=None, method_obj=None):
  40         if not isinstance(component_class, _bt2_component()._ComponentClassConst):
  41             err = False
  42
  43             try:
  44                 if not issubclass(component_class, _bt2_component()._UserComponent):
  45                     err = True
  46             except TypeError:
  47                 err = True
  48
  49             if err:
  50                 o = component_class
  51                 raise TypeError("'{}' is not a component class object".format(o))
  52
  53         utils._check_str(object_name)
  54
  55         if params is None:
  56             params_ptr = native_bt.value_null
  57         else:
  58             params = bt2.create_value(params)
  59             params_ptr = params._ptr
  60
  61         cc_ptr = component_class._bt_component_class_ptr()
  62         assert cc_ptr is not None
  63
  64         if method_obj is not None and not native_bt.bt2_is_python_component_class(
  65             cc_ptr
  66         ):
  67             raise ValueError(
  68                 "cannot pass a Python object to a non-Python component class"
  69             )
  70
  71         ptr = native_bt.bt2_query_executor_create(
  72             cc_ptr, object_name, params_ptr, method_obj
  73         )
  74
  75         if ptr is None:
  76             raise bt2._MemoryError("cannot create query executor object")
  77
  78         super().__init__(ptr)
  79
  80         # Keep a reference of `method_obj` as the native query executor
  81         # does not have any. This ensures that, when this object's
  82         # query() method is called, the Python object still exists.
  83         self._method_obj = method_obj
  84
  85     def add_interrupter(self, interrupter):
  86         utils._check_type(interrupter, bt2_interrupter.Interrupter)
  87         native_bt.query_executor_add_interrupter(self._ptr, interrupter._ptr)
  88
  89     @property
  90     def default_interrupter(self):
  91         ptr = native_bt.query_executor_borrow_default_interrupter(self._ptr)
  92         return bt2_interrupter.Interrupter._create_from_ptr_and_get_ref(ptr)
  93
  94     def _set_logging_level(self, log_level):
  95         utils._check_log_level(log_level)
  96         status = native_bt.query_executor_set_logging_level(self._ptr, log_level)
  97         utils._handle_func_status(status, "cannot set query executor's logging level")
  98
  99     logging_level = property(
 100         fget=_QueryExecutorCommon.logging_level, fset=_set_logging_level
 101     )
 102
 103     @property
 104     def is_interrupted(self):
 105         is_interrupted = native_bt.query_executor_is_interrupted(self._ptr)
 106         return bool(is_interrupted)
 107
 108     def query(self):
 109         status, result_ptr = native_bt.query_executor_query(self._ptr)
 110         utils._handle_func_status(status, "cannot query component class")
 111         assert result_ptr is not None
 112         return bt2_value._create_from_const_ptr(result_ptr)
 113
 114
 115 class _PrivateQueryExecutor(_QueryExecutorCommon):
 116     def __init__(self, ptr):
 117         self._ptr = ptr
 118
 119     def _check_validity(self):
 120         if self._ptr is None:
 121             raise RuntimeError("this object is not valid anymore")
 122
 123     def _as_query_executor_ptr(self):
 124         self._check_validity()
 125         return native_bt.private_query_executor_as_query_executor_const(self._ptr)
 126
 127     def _invalidate(self):
 128         self._ptr = None