[babeltrace.git] / bindings / python / bt2 / bt2 / trace_collection_notification_iterator.py

# The MIT License (MIT)
#
# Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

from bt2 import utils
import bt2
import bt2.notification_iterator
import datetime
import collections.abc
from collections import namedtuple
import numbers


# a pair of source component and _SourceComponentSpec
_SourceComponentAndSpec = namedtuple('_SourceComponentAndSpec',
                                     ['comp', 'spec'])


class SourceComponentSpec:
    def __init__(self, plugin_name, component_class_name, params=None):
        utils._check_str(plugin_name)
        utils._check_str(component_class_name)
        self._plugin_name = plugin_name
        self._component_class_name = component_class_name

        if type(params) is str:
            self._params = bt2.create_value({'path': params})
        else:
            self._params = bt2.create_value(params)

    @property
    def plugin_name(self):
        return self._plugin_name

    @property
    def component_class_name(self):
        return self._component_class_name

    @property
    def params(self):
        return self._params


# datetime.datetime or integral to nanoseconds
def _get_ns(obj):
    if obj is None:
        return

    if isinstance(obj, numbers.Real):
        # consider that it's already in seconds
        s = obj
    elif isinstance(obj, datetime.datetime):
        # s -> ns
        s = obj.timestamp()
    else:
        raise TypeError('"{}" is not an integral number or a datetime.datetime object'.format(obj))

    return int(s * 1e9)


class TraceCollectionNotificationIterator(bt2.notification_iterator._NotificationIterator):
    def __init__(self, source_component_specs, notification_types=None,
                 stream_intersection_mode=False, begin=None,
                 end=None):
        utils._check_bool(stream_intersection_mode)
        self._stream_intersection_mode = stream_intersection_mode
        self._begin_ns = _get_ns(begin)
        self._end_ns = _get_ns(end)
        self._notification_types = notification_types
        self._src_comp_specs = source_component_specs
        self._next_suffix = 1
        self._connect_ports = False

        # set of _SourceComponentAndSpec
        self._src_comps_and_specs = []

        self._validate_source_component_specs()
        self._build_graph()

    def _validate_source_component_specs(self):
        for source_comp_spec in self._src_comp_specs:
            if type(source_comp_spec) is not SourceComponentSpec:
                raise TypeError('"{}" object is not a SourceComponentSpec'.format(type(source_comp_spec)))

    def __next__(self):
        return next(self._notif_iter)

    def _create_stream_intersection_trimmer(self, port):
        # find the original parameters specified by the user to create
        # this port's component to get the `path` parameter
        for src_comp_and_spec in self._src_comps_and_specs:
            if port.component == src_comp_and_spec.comp:
                params = src_comp_and_spec.spec.params
                break

        try:
            path = params['path']
        except:
            raise bt2.Error('all source components must be created with a "path" parameter in stream intersection mode')

        params = {'path': str(path)}

        # query the port's component for the `trace-info` object which
        # contains the stream intersection range for each exposed
        # trace
        query_exec = bt2.QueryExecutor()
        trace_info_res = query_exec.query(port.component.component_class,
                                          'trace-info', params)
        begin = None
        end = None

        # find the trace info for this port's trace by name's prefix
        try:
            for trace_info in trace_info_res:
                if port.name.startswith(str(trace_info['path'])):
                    range_ns = trace_info['intersection-range-ns']
                    begin = range_ns['begin']
                    end = range_ns['end']
                    break
        except:
            pass

        if begin is None or end is None:
            raise bt2.Error('cannot find stream intersection range for port "{}"'.format(port.name))

        name = 'trimmer-{}-{}'.format(port.component.name, port.name)
        return self._create_trimmer(begin, end, name)

    def _create_muxer(self):
        plugin = bt2.find_plugin('utils')

        if plugin is None:
            raise bt2.Error('cannot find "utils" plugin (needed for the muxer)')

        if 'muxer' not in plugin.filter_component_classes:
            raise bt2.Error('cannot find "muxer" filter component class in "utils" plugin')

        comp_cls = plugin.filter_component_classes['muxer']
        return self._graph.add_component(comp_cls, 'muxer')

    def _create_trimmer(self, begin, end, name):
        plugin = bt2.find_plugin('utils')

        if plugin is None:
            raise bt2.Error('cannot find "utils" plugin (needed for the trimmer)')

        if 'trimmer' not in plugin.filter_component_classes:
            raise bt2.Error('cannot find "trimmer" filter component class in "utils" plugin')

        params = {}

        if begin is not None:
            params['begin'] = begin

        if end is not None:
            params['end'] = end

        comp_cls = plugin.filter_component_classes['trimmer']
        return self._graph.add_component(comp_cls, name, params)

    def _get_unique_src_comp_name(self, comp_spec):
        name = '{}-{}'.format(comp_spec.plugin_name,
                              comp_spec.component_class_name)

        if name in [comp_and_spec.comp.name for comp_and_spec in self._src_comps_and_specs]:
            name += '-{}'.format(self._next_suffix)
            self._next_suffix += 1

        return name

    def _create_src_comp(self, comp_spec):
        plugin = bt2.find_plugin(comp_spec.plugin_name)

        if plugin is None:
            raise bt2.Error('no such plugin: {}'.format(comp_spec.plugin_name))

        if comp_spec.component_class_name not in plugin.source_component_classes:
            raise bt2.Error('no such source component class in "{}" plugin: {}'.format(comp_spec.plugin_name,
                                                                                       comp_spec.component_class_name))

        comp_cls = plugin.source_component_classes[comp_spec.component_class_name]
        name = self._get_unique_src_comp_name(comp_spec)
        comp = self._graph.add_component(comp_cls, name, comp_spec.params)
        return comp

    def _get_free_muxer_input_port(self):
        for port in self._muxer_comp.input_ports.values():
            if not port.is_connected:
                return port

    def _connect_src_comp_port(self, port):
        # if this trace collection iterator is in stream intersection
        # mode, we need this connection:
        #
        #     port -> trimmer -> muxer
        #
        # otherwise, simply:
        #
        #     port -> muxer
        if self._stream_intersection_mode:
            trimmer_comp = self._create_stream_intersection_trimmer(port)
            self._graph.connect_ports(port, trimmer_comp.input_ports['in'])
            port_to_muxer = trimmer_comp.output_ports['out']
        else:
            port_to_muxer = port

        self._graph.connect_ports(port_to_muxer, self._get_free_muxer_input_port())

    def _graph_port_added(self, port):
        if not self._connect_ports:
            return

        if type(port) is bt2._InputPort:
            return

        if port.component not in [comp.comp for comp in self._src_comps_and_specs]:
            # do not care about non-source components (muxer, trimmer, etc.)
            return

        self._connect_src_comp_port(port)

    def _build_graph(self):
        self._graph = bt2.Graph()
        self._graph.add_listener(bt2.GraphListenerType.PORT_ADDED,
                                 self._graph_port_added)
        self._muxer_comp = self._create_muxer()

        if self._begin_ns is not None or self._end_ns is not None:
            trimmer_comp = self._create_trimmer(self._begin_ns,
                                                self._end_ns, 'trimmer')
            self._graph.connect_ports(self._muxer_comp.output_ports['out'],
                                      trimmer_comp.input_ports['in'])
            notif_iter_port = trimmer_comp.output_ports['out']
        else:
            notif_iter_port = self._muxer_comp.output_ports['out']

        # Here we create the components, self._graph_port_added() is
        # called when they add ports, but the callback returns early
        # because self._connect_ports is False. This is because the
        # self._graph_port_added() could not find the associated source
        # component specification in self._src_comps_and_specs because
        # it does not exist yet (it needs the created component to
        # exist).
        for comp_spec in self._src_comp_specs:
            comp = self._create_src_comp(comp_spec)
            self._src_comps_and_specs.append(_SourceComponentAndSpec(comp, comp_spec))

        # Now we connect the ports which exist at this point. We allow
        # self._graph_port_added() to automatically connect _new_ ports.
        self._connect_ports = True

        for comp_and_spec in self._src_comps_and_specs:
            # Keep a separate list because comp_and_spec.output_ports
            # could change during the connection of one of its ports.
            # Any new port is handled by self._graph_port_added().
            out_ports = [port for port in comp_and_spec.comp.output_ports.values()]

            for out_port in out_ports:
                if not out_port.component or out_port.is_connected:
                    continue

                self._connect_src_comp_port(out_port)

        # create this trace collection iterator's notification iterator
        self._notif_iter = notif_iter_port.create_notification_iterator(self._notification_types)
Commit	Line	Data
85dcce24 PP	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 bt2.notification_iterator
	26	import datetime
	27	import collections.abc
	28	from collections import namedtuple
	29	import numbers
	30
	31
	32	# a pair of source component and _SourceComponentSpec
	33	_SourceComponentAndSpec = namedtuple('_SourceComponentAndSpec',
	34	['comp', 'spec'])
	35
	36
	37	class SourceComponentSpec:
	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 TraceCollectionNotificationIterator(bt2.notification_iterator._NotificationIterator):
80	def __init__(self, source_component_specs, notification_types=None,
81	stream_intersection_mode=False, begin=None,
82	end=None):
83	utils._check_bool(stream_intersection_mode)
84	self._stream_intersection_mode = stream_intersection_mode
85	self._begin_ns = _get_ns(begin)
86	self._end_ns = _get_ns(end)
87	self._notification_types = notification_types
88	self._src_comp_specs = source_component_specs
89	self._next_suffix = 1
90	self._connect_ports = False
91
92	# set of _SourceComponentAndSpec
93	self._src_comps_and_specs = []
94
95	self._validate_source_component_specs()
96	self._build_graph()
97
98	def _validate_source_component_specs(self):
99	for source_comp_spec in self._src_comp_specs:
100	if type(source_comp_spec) is not SourceComponentSpec:
101	raise TypeError('"{}" object is not a SourceComponentSpec'.format(type(source_comp_spec)))
102
103	def __next__(self):
104	return next(self._notif_iter)
105
106	def _create_stream_intersection_trimmer(self, port):
107	# find the original parameters specified by the user to create
108	# this port's component to get the `path` parameter
109	for src_comp_and_spec in self._src_comps_and_specs:
110	if port.component == src_comp_and_spec.comp:
111	params = src_comp_and_spec.spec.params
112	break
113
114	try:
115	path = params['path']
116	except:
117	raise bt2.Error('all source components must be created with a "path" parameter in stream intersection mode')
118
119	params = {'path': str(path)}
120
121	# query the port's component for the `trace-info` object which
122	# contains the stream intersection range for each exposed
123	# trace
124	query_exec = bt2.QueryExecutor()
125	trace_info_res = query_exec.query(port.component.component_class,
126	'trace-info', params)
127	begin = None
128	end = None
129
130	# find the trace info for this port's trace by name's prefix
87b768aa PP	131	try:
	132	for trace_info in trace_info_res:
	133	if port.name.startswith(str(trace_info['path'])):
	134	range_ns = trace_info['intersection-range-ns']
	135	begin = range_ns['begin']
	136	end = range_ns['end']
	137	break
	138	except:
	139	pass
85dcce24 PP	140
	141	if begin is None or end is None:
	142	raise bt2.Error('cannot find stream intersection range for port "{}"'.format(port.name))
	143
	144	name = 'trimmer-{}-{}'.format(port.component.name, port.name)
	145	return self._create_trimmer(begin, end, name)
	146
	147	def _create_muxer(self):
	148	plugin = bt2.find_plugin('utils')
	149
	150	if plugin is None:
	151	raise bt2.Error('cannot find "utils" plugin (needed for the muxer)')
	152
	153	if 'muxer' not in plugin.filter_component_classes:
	154	raise bt2.Error('cannot find "muxer" filter component class in "utils" plugin')
	155
	156	comp_cls = plugin.filter_component_classes['muxer']
	157	return self._graph.add_component(comp_cls, 'muxer')
	158
	159	def _create_trimmer(self, begin, end, name):
	160	plugin = bt2.find_plugin('utils')
	161
	162	if plugin is None:
	163	raise bt2.Error('cannot find "utils" plugin (needed for the trimmer)')
	164
	165	if 'trimmer' not in plugin.filter_component_classes:
	166	raise bt2.Error('cannot find "trimmer" filter component class in "utils" plugin')
	167
	168	params = {}
	169
	170	if begin is not None:
	171	params['begin'] = begin
	172
	173	if end is not None:
	174	params['end'] = end
	175
	176	comp_cls = plugin.filter_component_classes['trimmer']
	177	return self._graph.add_component(comp_cls, name, params)
	178
	179	def _get_unique_src_comp_name(self, comp_spec):
	180	name = '{}-{}'.format(comp_spec.plugin_name,
	181	comp_spec.component_class_name)
	182
	183	if name in [comp_and_spec.comp.name for comp_and_spec in self._src_comps_and_specs]:
	184	name += '-{}'.format(self._next_suffix)
	185	self._next_suffix += 1
	186
	187	return name
	188
	189	def _create_src_comp(self, comp_spec):
	190	plugin = bt2.find_plugin(comp_spec.plugin_name)
	191
	192	if plugin is None:
	193	raise bt2.Error('no such plugin: {}'.format(comp_spec.plugin_name))
	194
	195	if comp_spec.component_class_name not in plugin.source_component_classes:
	196	raise bt2.Error('no such source component class in "{}" plugin: {}'.format(comp_spec.plugin_name,
	197	comp_spec.component_class_name))
	198
	199	comp_cls = plugin.source_component_classes[comp_spec.component_class_name]
	200	name = self._get_unique_src_comp_name(comp_spec)
	201	comp = self._graph.add_component(comp_cls, name, comp_spec.params)
	202	return comp
	203
204	def _get_free_muxer_input_port(self):
205	for port in self._muxer_comp.input_ports.values():
206	if not port.is_connected:
207	return port
208
209	def _connect_src_comp_port(self, port):
210	# if this trace collection iterator is in stream intersection
211	# mode, we need this connection:
212	#
213	# port -> trimmer -> muxer
214	#
215	# otherwise, simply:
216	#
217	# port -> muxer
218	if self._stream_intersection_mode:
219	trimmer_comp = self._create_stream_intersection_trimmer(port)
220	self._graph.connect_ports(port, trimmer_comp.input_ports['in'])
221	port_to_muxer = trimmer_comp.output_ports['out']
222	else:
223	port_to_muxer = port
224
225	self._graph.connect_ports(port_to_muxer, self._get_free_muxer_input_port())
226
227	def _graph_port_added(self, port):
228	if not self._connect_ports:
229	return
230
231	if type(port) is bt2._InputPort:
232	return
233
234	if port.component not in [comp.comp for comp in self._src_comps_and_specs]:
235	# do not care about non-source components (muxer, trimmer, etc.)
236	return
237
238	self._connect_src_comp_port(port)
239
240	def _build_graph(self):
241	self._graph = bt2.Graph()
242	self._graph.add_listener(bt2.GraphListenerType.PORT_ADDED,
243	self._graph_port_added)
244	self._muxer_comp = self._create_muxer()
245
246	if self._begin_ns is not None or self._end_ns is not None:
247	trimmer_comp = self._create_trimmer(self._begin_ns,
248	self._end_ns, 'trimmer')
249	self._graph.connect_ports(self._muxer_comp.output_ports['out'],
250	trimmer_comp.input_ports['in'])
251	notif_iter_port = trimmer_comp.output_ports['out']
252	else:
253	notif_iter_port = self._muxer_comp.output_ports['out']
254
255	# Here we create the components, self._graph_port_added() is
256	# called when they add ports, but the callback returns early
257	# because self._connect_ports is False. This is because the
258	# self._graph_port_added() could not find the associated source
259	# component specification in self._src_comps_and_specs because
260	# it does not exist yet (it needs the created component to
261	# exist).
262	for comp_spec in self._src_comp_specs:
263	comp = self._create_src_comp(comp_spec)
264	self._src_comps_and_specs.append(_SourceComponentAndSpec(comp, comp_spec))
265
266	# Now we connect the ports which exist at this point. We allow
267	# self._graph_port_added() to automatically connect _new_ ports.
268	self._connect_ports = True
269
270	for comp_and_spec in self._src_comps_and_specs:
271	# Keep a separate list because comp_and_spec.output_ports
272	# could change during the connection of one of its ports.
273	# Any new port is handled by self._graph_port_added().
274	out_ports = [port for port in comp_and_spec.comp.output_ports.values()]
275
276	for out_port in out_ports:
277	if not out_port.component or out_port.is_connected:
278	continue
279
280	self._connect_src_comp_port(out_port)
281
282	# create this trace collection iterator's notification iterator
283	self._notif_iter = notif_iter_port.create_notification_iterator(self._notification_types)