bt2: let components attach "user data" to ports
[babeltrace.git] / bindings / python / bt2 / bt2 / trace_collection_message_iterator.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 utils
24 import bt2
25 import itertools
26 import bt2.message_iterator
27 import datetime
28 import collections.abc
29 from collections import namedtuple
30 import numbers
31
32
33 # a pair of component and ComponentSpec
34 _ComponentAndSpec = namedtuple('_ComponentAndSpec', ['comp', 'spec'])
35
36
37 class ComponentSpec:
38 def __init__(self, plugin_name, component_class_name, params=None):
39 utils._check_str(plugin_name)
40 utils._check_str(component_class_name)
41 self._plugin_name = plugin_name
42 self._component_class_name = component_class_name
43
44 if type(params) is str:
45 self._params = bt2.create_value({'path': params})
46 else:
47 self._params = bt2.create_value(params)
48
49 @property
50 def plugin_name(self):
51 return self._plugin_name
52
53 @property
54 def component_class_name(self):
55 return self._component_class_name
56
57 @property
58 def params(self):
59 return self._params
60
61
62 # datetime.datetime or integral to nanoseconds
63 def _get_ns(obj):
64 if obj is None:
65 return
66
67 if isinstance(obj, numbers.Real):
68 # consider that it's already in seconds
69 s = obj
70 elif isinstance(obj, datetime.datetime):
71 # s -> ns
72 s = obj.timestamp()
73 else:
74 raise TypeError('"{}" is not an integral number or a datetime.datetime object'.format(obj))
75
76 return int(s * 1e9)
77
78
79 class _CompClsType:
80 SOURCE = 0
81 FILTER = 1
82
83
84 class TraceCollectionMessageIterator(bt2.message_iterator._MessageIterator):
85 def __init__(self, source_component_specs, filter_component_specs=None,
86 message_types=None, stream_intersection_mode=False,
87 begin=None, end=None):
88 utils._check_bool(stream_intersection_mode)
89 self._stream_intersection_mode = stream_intersection_mode
90 self._begin_ns = _get_ns(begin)
91 self._end_ns = _get_ns(end)
92 self._message_types = message_types
93
94 if type(source_component_specs) is ComponentSpec:
95 source_component_specs = [source_component_specs]
96
97 if type(filter_component_specs) is ComponentSpec:
98 filter_component_specs = [filter_component_specs]
99 elif filter_component_specs is None:
100 filter_component_specs = []
101
102 self._src_comp_specs = source_component_specs
103 self._flt_comp_specs = filter_component_specs
104 self._next_suffix = 1
105 self._connect_ports = False
106
107 # lists of _ComponentAndSpec
108 self._src_comps_and_specs = []
109 self._flt_comps_and_specs = []
110
111 self._validate_component_specs(source_component_specs)
112 self._validate_component_specs(filter_component_specs)
113 self._build_graph()
114
115 def _validate_component_specs(self, comp_specs):
116 for comp_spec in comp_specs:
117 if type(comp_spec) is not ComponentSpec:
118 raise TypeError('"{}" object is not a ComponentSpec'.format(type(comp_spec)))
119
120 def __next__(self):
121 return next(self._msg_iter)
122
123 def _create_stream_intersection_trimmer(self, port):
124 # find the original parameters specified by the user to create
125 # this port's component to get the `path` parameter
126 for src_comp_and_spec in self._src_comps_and_specs:
127 if port.component == src_comp_and_spec.comp:
128 params = src_comp_and_spec.spec.params
129 break
130
131 try:
132 path = params['path']
133 except:
134 raise bt2.Error('all source components must be created with a "path" parameter in stream intersection mode')
135
136 params = {'path': str(path)}
137
138 # query the port's component for the `trace-info` object which
139 # contains the stream intersection range for each exposed
140 # trace
141 query_exec = bt2.QueryExecutor()
142 trace_info_res = query_exec.query(port.component.component_class,
143 'trace-info', params)
144 begin = None
145 end = None
146
147 # find the trace info for this port's trace by name's prefix
148 try:
149 for trace_info in trace_info_res:
150 if port.name.startswith(str(trace_info['path'])):
151 range_ns = trace_info['intersection-range-ns']
152 begin = range_ns['begin']
153 end = range_ns['end']
154 break
155 except:
156 pass
157
158 if begin is None or end is None:
159 raise bt2.Error('cannot find stream intersection range for port "{}"'.format(port.name))
160
161 name = 'trimmer-{}-{}'.format(port.component.name, port.name)
162 return self._create_trimmer(begin, end, name)
163
164 def _create_muxer(self):
165 plugin = bt2.find_plugin('utils')
166
167 if plugin is None:
168 raise bt2.Error('cannot find "utils" plugin (needed for the muxer)')
169
170 if 'muxer' not in plugin.filter_component_classes:
171 raise bt2.Error('cannot find "muxer" filter component class in "utils" plugin')
172
173 comp_cls = plugin.filter_component_classes['muxer']
174 return self._graph.add_component(comp_cls, 'muxer')
175
176 def _create_trimmer(self, begin, end, name):
177 plugin = bt2.find_plugin('utils')
178
179 if plugin is None:
180 raise bt2.Error('cannot find "utils" plugin (needed for the trimmer)')
181
182 if 'trimmer' not in plugin.filter_component_classes:
183 raise bt2.Error('cannot find "trimmer" filter component class in "utils" plugin')
184
185 params = {}
186
187 if begin is not None:
188 params['begin'] = begin
189
190 if end is not None:
191 params['end'] = end
192
193 comp_cls = plugin.filter_component_classes['trimmer']
194 return self._graph.add_component(comp_cls, name, params)
195
196 def _get_unique_comp_name(self, comp_spec):
197 name = '{}-{}'.format(comp_spec.plugin_name,
198 comp_spec.component_class_name)
199 comps_and_specs = itertools.chain(self._src_comps_and_specs,
200 self._flt_comps_and_specs)
201
202 if name in [comp_and_spec.comp.name for comp_and_spec in comps_and_specs]:
203 name += '-{}'.format(self._next_suffix)
204 self._next_suffix += 1
205
206 return name
207
208 def _create_comp(self, comp_spec, comp_cls_type):
209 plugin = bt2.find_plugin(comp_spec.plugin_name)
210
211 if plugin is None:
212 raise bt2.Error('no such plugin: {}'.format(comp_spec.plugin_name))
213
214 if comp_cls_type == _CompClsType.SOURCE:
215 comp_classes = plugin.source_component_classes
216 else:
217 comp_classes = plugin.filter_component_classes
218
219 if comp_spec.component_class_name not in comp_classes:
220 cc_type = 'source' if comp_cls_type == _CompClsType.SOURCE else 'filter'
221 raise bt2.Error('no such {} component class in "{}" plugin: {}'.format(cc_type,
222 comp_spec.plugin_name,
223 comp_spec.component_class_name))
224
225 comp_cls = comp_classes[comp_spec.component_class_name]
226 name = self._get_unique_comp_name(comp_spec)
227 comp = self._graph.add_component(comp_cls, name, comp_spec.params)
228 return comp
229
230 def _get_free_muxer_input_port(self):
231 for port in self._muxer_comp.input_ports.values():
232 if not port.is_connected:
233 return port
234
235 def _connect_src_comp_port(self, port):
236 # if this trace collection iterator is in stream intersection
237 # mode, we need this connection:
238 #
239 # port -> trimmer -> muxer
240 #
241 # otherwise, simply:
242 #
243 # port -> muxer
244 if self._stream_intersection_mode:
245 trimmer_comp = self._create_stream_intersection_trimmer(port)
246 self._graph.connect_ports(port, trimmer_comp.input_ports['in'])
247 port_to_muxer = trimmer_comp.output_ports['out']
248 else:
249 port_to_muxer = port
250
251 self._graph.connect_ports(port_to_muxer, self._get_free_muxer_input_port())
252
253 def _graph_port_added(self, port):
254 if not self._connect_ports:
255 return
256
257 if type(port) is bt2._InputPort:
258 return
259
260 if port.component not in [comp.comp for comp in self._src_comps_and_specs]:
261 # do not care about non-source components (muxer, trimmer, etc.)
262 return
263
264 self._connect_src_comp_port(port)
265
266 def _build_graph(self):
267 self._graph = bt2.Graph()
268 self._graph.add_listener(bt2.GraphListenerType.PORT_ADDED,
269 self._graph_port_added)
270 self._muxer_comp = self._create_muxer()
271
272 if self._begin_ns is not None or self._end_ns is not None:
273 trimmer_comp = self._create_trimmer(self._begin_ns,
274 self._end_ns, 'trimmer')
275 self._graph.connect_ports(self._muxer_comp.output_ports['out'],
276 trimmer_comp.input_ports['in'])
277 msg_iter_port = trimmer_comp.output_ports['out']
278 else:
279 msg_iter_port = self._muxer_comp.output_ports['out']
280
281 # create extra filter components (chained)
282 for comp_spec in self._flt_comp_specs:
283 comp = self._create_comp(comp_spec, _CompClsType.FILTER)
284 self._flt_comps_and_specs.append(_ComponentAndSpec(comp, comp_spec))
285
286 # connect the extra filter chain
287 for comp_and_spec in self._flt_comps_and_specs:
288 in_port = list(comp_and_spec.comp.input_ports.values())[0]
289 out_port = list(comp_and_spec.comp.output_ports.values())[0]
290 self._graph.connect_ports(msg_iter_port, in_port)
291 msg_iter_port = out_port
292
293 # Here we create the components, self._graph_port_added() is
294 # called when they add ports, but the callback returns early
295 # because self._connect_ports is False. This is because the
296 # self._graph_port_added() could not find the associated source
297 # component specification in self._src_comps_and_specs because
298 # it does not exist yet (it needs the created component to
299 # exist).
300 for comp_spec in self._src_comp_specs:
301 comp = self._create_comp(comp_spec, _CompClsType.SOURCE)
302 self._src_comps_and_specs.append(_ComponentAndSpec(comp, comp_spec))
303
304 # Now we connect the ports which exist at this point. We allow
305 # self._graph_port_added() to automatically connect _new_ ports.
306 self._connect_ports = True
307
308 for comp_and_spec in self._src_comps_and_specs:
309 # Keep a separate list because comp_and_spec.output_ports
310 # could change during the connection of one of its ports.
311 # Any new port is handled by self._graph_port_added().
312 out_ports = [port for port in comp_and_spec.comp.output_ports.values()]
313
314 for out_port in out_ports:
315 if not out_port.component or out_port.is_connected:
316 continue
317
318 self._connect_src_comp_port(out_port)
319
320 # create this trace collection iterator's message iterator
321 self._msg_iter = msg_iter_port.create_message_iterator(self._message_types)
This page took 0.038437 seconds and 5 git commands to generate.