src/bindings/python/bt2/bt2/utils.py

   1 # SPDX-License-Identifier: MIT
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4
   5
   6 from bt2 import error as bt2_error
   7 from bt2 import logging as bt2_logging
   8 from bt2 import native_bt
   9
  10
  11 class UnknownObject(Exception):
  12     """
  13     Raised when a component class handles a query for an object it doesn't
  14     know about.
  15     """
  16
  17     pass
  18
  19
  20 class _OverflowError(bt2_error._Error, OverflowError):
  21     pass
  22
  23
  24 class TryAgain(Exception):
  25     pass
  26
  27
  28 class Stop(StopIteration):
  29     pass
  30
  31
  32 def _check_bool(o):
  33     if not isinstance(o, bool):
  34         raise TypeError("'{}' is not a 'bool' object".format(o.__class__.__name__))
  35
  36
  37 def _check_int(o):
  38     if not isinstance(o, int):
  39         raise TypeError("'{}' is not an 'int' object".format(o.__class__.__name__))
  40
  41
  42 def _check_float(o):
  43     if not isinstance(o, float):
  44         raise TypeError("'{}' is not a 'float' object".format(o.__class__.__name__))
  45
  46
  47 def _check_str(o):
  48     if not isinstance(o, str):
  49         raise TypeError("'{}' is not a 'str' object".format(o.__class__.__name__))
  50
  51
  52 def _check_type(o, expected_type):
  53     if not isinstance(o, expected_type):
  54         raise TypeError(
  55             "'{}' is not a '{}' object".format(o.__class__.__name__, expected_type)
  56         )
  57
  58
  59 def _is_in_int64_range(v):
  60     assert isinstance(v, int)
  61     return v >= -(2**63) and v <= (2**63 - 1)
  62
  63
  64 def _is_int64(v):
  65     if not isinstance(v, int):
  66         return False
  67
  68     return _is_in_int64_range(v)
  69
  70
  71 def _is_in_uint64_range(v):
  72     assert isinstance(v, int)
  73     return v >= 0 and v <= (2**64 - 1)
  74
  75
  76 def _is_uint64(v):
  77     if not isinstance(v, int):
  78         return False
  79
  80     return _is_in_uint64_range(v)
  81
  82
  83 def _check_int64(v, msg=None):
  84     _check_int(v)
  85
  86     if not _is_in_int64_range(v):
  87         if msg is None:
  88             msg = "expecting a signed 64-bit integral value"
  89
  90         msg += " (got {})".format(v)
  91         raise ValueError(msg)
  92
  93
  94 def _check_uint64(v, msg=None):
  95     _check_int(v)
  96
  97     if not _is_in_uint64_range(v):
  98         if msg is None:
  99             msg = "expecting an unsigned 64-bit integral value"
 100
 101         msg += " (got {})".format(v)
 102         raise ValueError(msg)
 103
 104
 105 def _is_m1ull(v):
 106     return v == 18446744073709551615
 107
 108
 109 def _is_pow2(v):
 110     return v != 0 and ((v & (v - 1)) == 0)
 111
 112
 113 def _check_alignment(a):
 114     _check_uint64(a)
 115
 116     if not _is_pow2(a):
 117         raise ValueError("{} is not a power of two".format(a))
 118
 119
 120 def _check_log_level(log_level):
 121     _check_int(log_level)
 122
 123     log_levels = (
 124         bt2_logging.LoggingLevel.TRACE,
 125         bt2_logging.LoggingLevel.DEBUG,
 126         bt2_logging.LoggingLevel.INFO,
 127         bt2_logging.LoggingLevel.WARNING,
 128         bt2_logging.LoggingLevel.ERROR,
 129         bt2_logging.LoggingLevel.FATAL,
 130         bt2_logging.LoggingLevel.NONE,
 131     )
 132
 133     if log_level not in log_levels:
 134         raise ValueError("'{}' is not a valid logging level".format(log_level))
 135
 136
 137 def _handle_func_status(status, msg=None):
 138     if status == native_bt.__BT_FUNC_STATUS_OK:
 139         # no error
 140         return
 141
 142     if status == native_bt.__BT_FUNC_STATUS_ERROR:
 143         assert msg is not None
 144         raise bt2_error._Error(msg)
 145     elif status == native_bt.__BT_FUNC_STATUS_MEMORY_ERROR:
 146         assert msg is not None
 147         raise bt2_error._MemoryError(msg)
 148     elif status == native_bt.__BT_FUNC_STATUS_END:
 149         if msg is None:
 150             raise Stop
 151         else:
 152             raise Stop(msg)
 153     elif status == native_bt.__BT_FUNC_STATUS_AGAIN:
 154         if msg is None:
 155             raise TryAgain
 156         else:
 157             raise TryAgain(msg)
 158     elif status == native_bt.__BT_FUNC_STATUS_OVERFLOW_ERROR:
 159         raise _OverflowError(msg)
 160     elif status == native_bt.__BT_FUNC_STATUS_UNKNOWN_OBJECT:
 161         if msg is None:
 162             raise UnknownObject
 163         else:
 164             raise UnknownObject(msg)
 165     else:
 166         assert False
 167
 168
 169 class _ListenerHandle:
 170     def __init__(self, addr):
 171         self._addr = addr
 172         self._listener_id = None
 173
 174     def _set_listener_id(self, listener_id):
 175         self._listener_id = listener_id
 176
 177     def _invalidate(self):
 178         self._listener_id = None