src/bindings/python/bt2/bt2/graph.py

   1 # The MIT License (MIT)
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4 #
   5 # Permission is hereby granted, free of charge, to any person obtaining a copy
   6 # of this software and associated documentation files (the "Software"), to deal
   7 # in the Software without restriction, including without limitation the rights
   8 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9 # copies of the Software, and to permit persons to whom the Software is
  10 # furnished to do so, subject to the following conditions:
  11 #
  12 # The above copyright notice and this permission notice shall be included in
  13 # all copies or substantial portions of the Software.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  21 # THE SOFTWARE.
  22
  23 from bt2 import native_bt, object, utils
  24 import bt2.connection
  25 import bt2.component
  26 import functools
  27 import bt2.port
  28 import bt2.logging
  29 import bt2
  30
  31
  32 def _graph_port_added_listener_from_native(
  33     user_listener, component_ptr, component_type, port_ptr, port_type
  34 ):
  35     component = bt2.component._create_component_from_ptr_and_get_ref(
  36         component_ptr, component_type
  37     )
  38     port = bt2.port._create_from_ptr_and_get_ref(port_ptr, port_type)
  39     user_listener(component, port)
  40
  41
  42 def _graph_ports_connected_listener_from_native(
  43     user_listener,
  44     upstream_component_ptr,
  45     upstream_component_type,
  46     upstream_port_ptr,
  47     downstream_component_ptr,
  48     downstream_component_type,
  49     downstream_port_ptr,
  50 ):
  51     upstream_component = bt2.component._create_component_from_ptr_and_get_ref(
  52         upstream_component_ptr, upstream_component_type
  53     )
  54     upstream_port = bt2.port._create_from_ptr_and_get_ref(
  55         upstream_port_ptr, native_bt.PORT_TYPE_OUTPUT
  56     )
  57     downstream_component = bt2.component._create_component_from_ptr_and_get_ref(
  58         downstream_component_ptr, downstream_component_type
  59     )
  60     downstream_port = bt2.port._create_from_ptr_and_get_ref(
  61         downstream_port_ptr, native_bt.PORT_TYPE_INPUT
  62     )
  63     user_listener(
  64         upstream_component, upstream_port, downstream_component, downstream_port
  65     )
  66
  67
  68 class Graph(object._SharedObject):
  69     _get_ref = staticmethod(native_bt.graph_get_ref)
  70     _put_ref = staticmethod(native_bt.graph_put_ref)
  71
  72     def __init__(self):
  73         ptr = native_bt.graph_create()
  74
  75         if ptr is None:
  76             raise bt2.MemoryError('cannot create graph object')
  77
  78         super().__init__(ptr)
  79
  80     def add_component(
  81         self,
  82         component_class,
  83         name,
  84         params=None,
  85         logging_level=bt2.logging.LoggingLevel.NONE,
  86     ):
  87         if isinstance(component_class, bt2.component._GenericSourceComponentClass):
  88             cc_ptr = component_class._ptr
  89             add_fn = native_bt.graph_add_source_component
  90             cc_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
  91         elif isinstance(component_class, bt2.component._GenericFilterComponentClass):
  92             cc_ptr = component_class._ptr
  93             add_fn = native_bt.graph_add_filter_component
  94             cc_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
  95         elif isinstance(component_class, bt2.component._GenericSinkComponentClass):
  96             cc_ptr = component_class._ptr
  97             add_fn = native_bt.graph_add_sink_component
  98             cc_type = native_bt.COMPONENT_CLASS_TYPE_SINK
  99         elif issubclass(component_class, bt2.component._UserSourceComponent):
 100             cc_ptr = component_class._bt_cc_ptr
 101             add_fn = native_bt.graph_add_source_component
 102             cc_type = native_bt.COMPONENT_CLASS_TYPE_SOURCE
 103         elif issubclass(component_class, bt2.component._UserSinkComponent):
 104             cc_ptr = component_class._bt_cc_ptr
 105             add_fn = native_bt.graph_add_sink_component
 106             cc_type = native_bt.COMPONENT_CLASS_TYPE_SINK
 107         elif issubclass(component_class, bt2.component._UserFilterComponent):
 108             cc_ptr = component_class._bt_cc_ptr
 109             add_fn = native_bt.graph_add_filter_component
 110             cc_type = native_bt.COMPONENT_CLASS_TYPE_FILTER
 111         else:
 112             raise TypeError(
 113                 "'{}' is not a component class".format(
 114                     component_class.__class__.__name__
 115                 )
 116             )
 117
 118         utils._check_str(name)
 119         utils._check_log_level(logging_level)
 120         params = bt2.create_value(params)
 121
 122         params_ptr = params._ptr if params is not None else None
 123
 124         status, comp_ptr = add_fn(self._ptr, cc_ptr, name, params_ptr, logging_level)
 125         utils._handle_func_status(status, 'cannot add component to graph')
 126         assert comp_ptr
 127         return bt2.component._create_component_from_ptr(comp_ptr, cc_type)
 128
 129     def connect_ports(self, upstream_port, downstream_port):
 130         utils._check_type(upstream_port, bt2.port._OutputPort)
 131         utils._check_type(downstream_port, bt2.port._InputPort)
 132         status, conn_ptr = native_bt.graph_connect_ports(
 133             self._ptr, upstream_port._ptr, downstream_port._ptr
 134         )
 135         utils._handle_func_status(status, 'cannot connect component ports within graph')
 136         assert conn_ptr
 137         return bt2.connection._Connection._create_from_ptr(conn_ptr)
 138
 139     def add_port_added_listener(self, listener):
 140         if not callable(listener):
 141             raise TypeError("'listener' parameter is not callable")
 142
 143         fn = native_bt.bt2_graph_add_port_added_listener
 144         listener_from_native = functools.partial(
 145             _graph_port_added_listener_from_native, listener
 146         )
 147
 148         listener_ids = fn(self._ptr, listener_from_native)
 149         if listener_ids is None:
 150             raise bt2.Error('cannot add listener to graph object')
 151
 152         return bt2._ListenerHandle(listener_ids, self)
 153
 154     def add_ports_connected_listener(self, listener):
 155         if not callable(listener):
 156             raise TypeError("'listener' parameter is not callable")
 157
 158         fn = native_bt.bt2_graph_add_ports_connected_listener
 159         listener_from_native = functools.partial(
 160             _graph_ports_connected_listener_from_native, listener
 161         )
 162
 163         listener_ids = fn(self._ptr, listener_from_native)
 164         if listener_ids is None:
 165             raise bt2.Error('cannot add listener to graph object')
 166
 167         return bt2._ListenerHandle(listener_ids, self)
 168
 169     def run(self):
 170         status = native_bt.graph_run(self._ptr)
 171
 172         try:
 173             utils._handle_func_status(
 174                 status, 'graph object stopped running because of an unexpected error'
 175             )
 176         except bt2.Stop:
 177             # done
 178             return
 179         except Exception:
 180             raise
 181
 182     def cancel(self):
 183         status = native_bt.graph_cancel(self._ptr)
 184         utils._handle_func_status(status, 'cannot cancel graph object')
 185
 186     @property
 187     def is_canceled(self):
 188         is_canceled = native_bt.graph_is_canceled(self._ptr)
 189         assert is_canceled >= 0
 190         return is_canceled > 0
 191
 192     def create_output_port_message_iterator(self, output_port):
 193         utils._check_type(output_port, bt2.port._OutputPort)
 194         msg_iter_ptr = native_bt.port_output_message_iterator_create(
 195             self._ptr, output_port._ptr
 196         )
 197
 198         if msg_iter_ptr is None:
 199             raise bt2.MemoryError('cannot create output port message iterator')
 200
 201         return bt2.message_iterator._OutputPortMessageIterator(msg_iter_ptr)