Remove stale trace-ir test files

[babeltrace.git] / bindings / python / babeltrace / babeltrace / reader_trace_handle.py

# The MIT License (MIT)
#
# Copyright (c) 2013-2017 Jérémie Galarneau <jeremie.galarneau@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.

import bt2
import itertools
from babeltrace import reader_event_declaration


class TraceHandle:
    """
    A :class:`TraceHandle` is a handle allowing the user to manipulate
    a specific trace directly. It is a unique identifier representing a
    trace, and is not meant to be instantiated by the user.
    """

    def __init__(self):
        raise NotImplementedError("TraceHandle cannot be instantiated")

    def __repr__(self):
        # TODO print an id or some information about component / query result?
        return "Babeltrace TraceHandle: trace_id('{0}')".format(self._id)

    def __hash__(self):
        return hash((self.path, self.id))

    def __eq__(self, other):
        if type(other) is not type(self):
            return False

        return (self.path, self.id) == (other.path, other.id)

    @property
    def id(self):
        """
        Numeric ID of this trace handle.
        """

        return self._id

    @property
    def path(self):
        """
        Path of the underlying trace.
        """

        return self._path

    def _query_trace_info(self):
        try:
            result = bt2.QueryExecutor().query(self._trace_collection._fs_comp_cls,
                                               'trace-info', {'path': self._path})
        except:
            raise ValueError

        assert(len(result) == 1)
        return result

    @property
    def timestamp_begin(self):
        """
        Buffers creation timestamp (nanoseconds since Epoch) of the
        underlying trace.
        """

        result = self._query_trace_info()

        try:
            return int(result[0]['range-ns']['begin'])
        except:
            raise ValueError

    @property
    def timestamp_end(self):
        """
        Buffers destruction timestamp (nanoseconds since Epoch) of the
        underlying trace.
        """

        result = self._query_trace_info()

        try:
            return int(result[0]['range-ns']['end'])
        except:
            raise ValueError

    @property
    def _has_intersection(self):
        result = self._query_trace_info()

        try:
            return 'intersection-range-ns' in result[0]
        except:
            raise ValueError

    def _get_event_declarations(self):
        notif_iter = bt2.TraceCollectionNotificationIterator([
            bt2.ComponentSpec('ctf', 'fs', self._path)
        ])

        # raises if the trace contains no streams
        first_notif = next(notif_iter)
        assert(type(first_notif) is bt2.StreamBeginningNotification)
        trace = first_notif.stream.stream_class.trace
        ec_iters = [sc.values() for sc in trace.values()]
        return map(reader_event_declaration._create_event_declaration,
                   itertools.chain(*ec_iters))

    @property
    def events(self):
        """
        Generates all the :class:`EventDeclaration` objects of the
        underlying trace.
        """

        try:
            return self._get_event_declarations()
        except:
            return