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