bindings/python/bt2/bt2/notification.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.clock_class_priority_map
  25 import bt2.clock_value
  26 import bt2.packet
  27 import bt2.stream
  28 import bt2.event
  29 import copy
  30 import bt2
  31
  32
  33 def _create_from_ptr(ptr):
  34     notif_type = native_bt.notification_get_type(ptr)
  35     cls = None
  36
  37     if notif_type not in _NOTIF_TYPE_TO_CLS:
  38         raise bt2.Error('unknown notification type: {}'.format(notif_type))
  39
  40     return _NOTIF_TYPE_TO_CLS[notif_type]._create_from_ptr(ptr)
  41
  42
  43 def _notif_types_from_notif_classes(notification_types):
  44     if notification_types is None:
  45         notif_types = None
  46     else:
  47         for notif_cls in notification_types:
  48             if notif_cls not in _NOTIF_TYPE_TO_CLS.values():
  49                 raise ValueError("'{}' is not a notification class".format(notif_cls))
  50
  51         notif_types = [notif_cls._TYPE for notif_cls in notification_types]
  52
  53     return notif_types
  54
  55
  56 class _Notification(object._Object):
  57     pass
  58
  59
  60 class _CopyableNotification(_Notification):
  61     def __copy__(self):
  62         return self._copy(lambda obj: obj)
  63
  64     def __deepcopy__(self, memo):
  65         cpy = self._copy(copy.deepcopy)
  66         memo[id(self)] = cpy
  67         return cpy
  68
  69
  70 class EventNotification(_CopyableNotification):
  71     _TYPE = native_bt.NOTIFICATION_TYPE_EVENT
  72
  73     def __init__(self, event, cc_prio_map=None):
  74         utils._check_type(event, bt2.event._Event)
  75
  76         if cc_prio_map is not None:
  77             utils._check_type(cc_prio_map, bt2.clock_class_priority_map.ClockClassPriorityMap)
  78             cc_prio_map_ptr = cc_prio_map._ptr
  79         else:
  80             cc_prio_map_ptr = None
  81
  82         ptr = native_bt.notification_event_create(event._ptr, cc_prio_map_ptr)
  83
  84         if ptr is None:
  85             raise bt2.CreationError('cannot create event notification object')
  86
  87         super().__init__(ptr)
  88
  89     @property
  90     def event(self):
  91         event_ptr = native_bt.notification_event_get_event(self._ptr)
  92         assert(event_ptr)
  93         return bt2.event._create_from_ptr(event_ptr)
  94
  95     @property
  96     def clock_class_priority_map(self):
  97         cc_prio_map_ptr = native_bt.notification_event_get_clock_class_priority_map(self._ptr)
  98         assert(cc_prio_map_ptr)
  99         return bt2.clock_class_priority_map.ClockClassPriorityMap._create_from_ptr(cc_prio_map_ptr)
 100
 101     def __eq__(self, other):
 102         if type(other) is not type(self):
 103             return False
 104
 105         if self.addr == other.addr:
 106             return True
 107
 108         self_props = (
 109             self.event,
 110             self.clock_class_priority_map,
 111         )
 112         other_props = (
 113             other.event,
 114             other.clock_class_priority_map,
 115         )
 116         return self_props == other_props
 117
 118     def _copy(self, copy_func):
 119         # We can always use references here because those properties are
 120         # frozen anyway if they are part of a notification. Since the
 121         # user cannot modify them after copying the notification, it's
 122         # useless to copy/deep-copy them.
 123         return EventNotification(self.event, self.clock_class_priority_map)
 124
 125
 126 class PacketBeginningNotification(_CopyableNotification):
 127     _TYPE = native_bt.NOTIFICATION_TYPE_PACKET_BEGIN
 128
 129     def __init__(self, packet):
 130         utils._check_type(packet, bt2.packet._Packet)
 131         ptr = native_bt.notification_packet_begin_create(packet._ptr)
 132
 133         if ptr is None:
 134             raise bt2.CreationError('cannot create packet beginning notification object')
 135
 136         super().__init__(ptr)
 137
 138     @property
 139     def packet(self):
 140         packet_ptr = native_bt.notification_packet_begin_get_packet(self._ptr)
 141         assert(packet_ptr)
 142         return bt2.packet._Packet._create_from_ptr(packet_ptr)
 143
 144     def __eq__(self, other):
 145         if type(other) is not type(self):
 146             return False
 147
 148         if self.addr == other.addr:
 149             return True
 150
 151         return self.packet == other.packet
 152
 153     def _copy(self, copy_func):
 154         # We can always use references here because those properties are
 155         # frozen anyway if they are part of a notification. Since the
 156         # user cannot modify them after copying the notification, it's
 157         # useless to copy/deep-copy them.
 158         return PacketBeginningNotification(self.packet)
 159
 160
 161 class PacketEndNotification(_CopyableNotification):
 162     _TYPE = native_bt.NOTIFICATION_TYPE_PACKET_END
 163
 164     def __init__(self, packet):
 165         utils._check_type(packet, bt2.packet._Packet)
 166         ptr = native_bt.notification_packet_end_create(packet._ptr)
 167
 168         if ptr is None:
 169             raise bt2.CreationError('cannot create packet end notification object')
 170
 171         super().__init__(ptr)
 172
 173     @property
 174     def packet(self):
 175         packet_ptr = native_bt.notification_packet_end_get_packet(self._ptr)
 176         assert(packet_ptr)
 177         return bt2.packet._Packet._create_from_ptr(packet_ptr)
 178
 179     def __eq__(self, other):
 180         if type(other) is not type(self):
 181             return False
 182
 183         if self.addr == other.addr:
 184             return True
 185
 186         return self.packet == other.packet
 187
 188     def _copy(self, copy_func):
 189         # We can always use references here because those properties are
 190         # frozen anyway if they are part of a notification. Since the
 191         # user cannot modify them after copying the notification, it's
 192         # useless to copy/deep-copy them.
 193         return PacketEndNotification(self.packet)
 194
 195
 196 class StreamBeginningNotification(_CopyableNotification):
 197     _TYPE = native_bt.NOTIFICATION_TYPE_STREAM_BEGIN
 198
 199     def __init__(self, stream):
 200         utils._check_type(stream, bt2.stream._Stream)
 201         ptr = native_bt.notification_stream_begin_create(stream._ptr)
 202
 203         if ptr is None:
 204             raise bt2.CreationError('cannot create stream beginning notification object')
 205
 206         super().__init__(ptr)
 207
 208     @property
 209     def stream(self):
 210         stream_ptr = native_bt.notification_stream_begin_get_stream(self._ptr)
 211         assert(stream_ptr)
 212         return bt2.stream._create_from_ptr(stream_ptr)
 213
 214     def __eq__(self, other):
 215         if type(other) is not type(self):
 216             return False
 217
 218         if self.addr == other.addr:
 219             return True
 220
 221         return self.stream == other.stream
 222
 223     def _copy(self, copy_func):
 224         # We can always use references here because those properties are
 225         # frozen anyway if they are part of a notification. Since the
 226         # user cannot modify them after copying the notification, it's
 227         # useless to copy/deep-copy them.
 228         return StreamBeginningNotification(self.stream)
 229
 230
 231 class StreamEndNotification(_CopyableNotification):
 232     _TYPE = native_bt.NOTIFICATION_TYPE_STREAM_END
 233
 234     def __init__(self, stream):
 235         utils._check_type(stream, bt2.stream._Stream)
 236         ptr = native_bt.notification_stream_end_create(stream._ptr)
 237
 238         if ptr is None:
 239             raise bt2.CreationError('cannot create stream end notification object')
 240
 241         super().__init__(ptr)
 242
 243     @property
 244     def stream(self):
 245         stream_ptr = native_bt.notification_stream_end_get_stream(self._ptr)
 246         assert(stream_ptr)
 247         return bt2.stream._create_from_ptr(stream_ptr)
 248
 249     def __eq__(self, other):
 250         if type(other) is not type(self):
 251             return False
 252
 253         if self.addr == other.addr:
 254             return True
 255
 256         return self.stream == other.stream
 257
 258     def _copy(self, copy_func):
 259         # We can always use references here because those properties are
 260         # frozen anyway if they are part of a notification. Since the
 261         # user cannot modify them after copying the notification, it's
 262         # useless to copy/deep-copy them.
 263         return StreamEndNotification(self.stream)
 264
 265
 266 class InactivityNotification(_CopyableNotification):
 267     _TYPE = native_bt.NOTIFICATION_TYPE_INACTIVITY
 268
 269     def __init__(self, cc_prio_map=None):
 270         if cc_prio_map is not None:
 271             utils._check_type(cc_prio_map, bt2.clock_class_priority_map.ClockClassPriorityMap)
 272             cc_prio_map_ptr = cc_prio_map._ptr
 273         else:
 274             cc_prio_map_ptr = None
 275
 276         ptr = native_bt.notification_inactivity_create(cc_prio_map_ptr)
 277
 278         if ptr is None:
 279             raise bt2.CreationError('cannot create inactivity notification object')
 280
 281         super().__init__(ptr)
 282
 283     @property
 284     def clock_class_priority_map(self):
 285         cc_prio_map_ptr = native_bt.notification_inactivity_get_clock_class_priority_map(self._ptr)
 286         assert(cc_prio_map_ptr)
 287         return bt2.clock_class_priority_map.ClockClassPriorityMap._create_from_ptr(cc_prio_map_ptr)
 288
 289     def clock_value(self, clock_class):
 290         utils._check_type(clock_class, bt2.ClockClass)
 291         clock_value_ptr = native_bt.notification_inactivity_get_clock_value(self._ptr,
 292                                                                             clock_class._ptr)
 293
 294         if clock_value_ptr is None:
 295             return
 296
 297         clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
 298         return clock_value
 299
 300     def add_clock_value(self, clock_value):
 301         utils._check_type(clock_value, bt2.clock_value._ClockValue)
 302         ret = native_bt.notification_inactivity_set_clock_value(self._ptr,
 303                                                                 clock_value._ptr)
 304         utils._handle_ret(ret, "cannot set inactivity notification object's clock value")
 305
 306     def _get_clock_values(self):
 307         clock_values = {}
 308
 309         for clock_class in self.clock_class_priority_map:
 310             clock_value = self.clock_value(clock_class)
 311
 312             if clock_value is None:
 313                 continue
 314
 315             clock_values[clock_class] = clock_value
 316
 317         return clock_values
 318
 319     def __eq__(self, other):
 320         if type(other) is not type(self):
 321             return False
 322
 323         if self.addr == other.addr:
 324             return True
 325
 326         self_props = (
 327             self.clock_class_priority_map,
 328             self._get_clock_values(),
 329         )
 330         other_props = (
 331             other.clock_class_priority_map,
 332             other._get_clock_values(),
 333         )
 334         return self_props == other_props
 335
 336     def __copy__(self):
 337         cpy = InactivityNotification(self.clock_class_priority_map)
 338
 339         for clock_class in self.clock_class_priority_map:
 340             clock_value = self.clock_value(clock_class)
 341
 342             if clock_value is None:
 343                 continue
 344
 345             cpy.add_clock_value(clock_value)
 346
 347         return cpy
 348
 349     def __deepcopy__(self, memo):
 350         cc_prio_map_cpy = copy.deepcopy(self.clock_class_priority_map)
 351         cpy = InactivityNotification(cc_prio_map_cpy)
 352
 353         # copy clock values
 354         for orig_clock_class in self.clock_class_priority_map:
 355             orig_clock_value = self.clock_value(orig_clock_class)
 356
 357             if orig_clock_value is None:
 358                 continue
 359
 360             # find equivalent, copied clock class in CC priority map copy
 361             for cpy_clock_class in cc_prio_map_cpy:
 362                 if cpy_clock_class == orig_clock_class:
 363                     break
 364
 365             # create copy of clock value from copied clock class
 366             clock_value_cpy = cpy_clock_class(orig_clock_value.cycles)
 367
 368             # set copied clock value in notification copy
 369             cpy.add_clock_value(clock_value_cpy)
 370
 371         memo[id(self)] = cpy
 372         return cpy
 373
 374
 375 class _DiscardedElementsNotification(_Notification):
 376     def __eq__(self, other):
 377         if type(other) is not type(self):
 378             return False
 379
 380         if self.addr == other.addr:
 381             return True
 382
 383         self_props = (
 384             self.count,
 385             self.stream,
 386             self.beginning_clock_value,
 387             self.end_clock_value,
 388         )
 389         other_props = (
 390             other.count,
 391             other.stream,
 392             other.beginning_clock_value,
 393             other.end_clock_value,
 394         )
 395         return self_props == other_props
 396
 397
 398 class _DiscardedPacketsNotification(_DiscardedElementsNotification):
 399     _TYPE = native_bt.NOTIFICATION_TYPE_DISCARDED_PACKETS
 400
 401     @property
 402     def count(self):
 403         count = native_bt.notification_discarded_packets_get_count(self._ptr)
 404         assert(count >= 0)
 405         return count
 406
 407     @property
 408     def stream(self):
 409         stream_ptr = native_bt.notification_discarded_packets_get_stream(self._ptr)
 410         assert(stream_ptr)
 411         return bt2.stream._create_from_ptr(stream_ptr)
 412
 413     @property
 414     def beginning_clock_value(self):
 415         clock_value_ptr = native_bt.notification_discarded_packets_get_begin_clock_value(self._ptr)
 416
 417         if clock_value_ptr is None:
 418             return
 419
 420         clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
 421         return clock_value
 422
 423     @property
 424     def end_clock_value(self):
 425         clock_value_ptr = native_bt.notification_discarded_packets_get_end_clock_value(self._ptr)
 426
 427         if clock_value_ptr is None:
 428             return
 429
 430         clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
 431         return clock_value
 432
 433
 434 class _DiscardedEventsNotification(_DiscardedElementsNotification):
 435     _TYPE = native_bt.NOTIFICATION_TYPE_DISCARDED_EVENTS
 436
 437     @property
 438     def count(self):
 439         count = native_bt.notification_discarded_events_get_count(self._ptr)
 440         assert(count >= 0)
 441         return count
 442
 443     @property
 444     def stream(self):
 445         stream_ptr = native_bt.notification_discarded_events_get_stream(self._ptr)
 446         assert(stream_ptr)
 447         return bt2.stream._create_from_ptr(stream_ptr)
 448
 449     @property
 450     def beginning_clock_value(self):
 451         clock_value_ptr = native_bt.notification_discarded_events_get_begin_clock_value(self._ptr)
 452
 453         if clock_value_ptr is None:
 454             return
 455
 456         clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
 457         return clock_value
 458
 459     @property
 460     def end_clock_value(self):
 461         clock_value_ptr = native_bt.notification_discarded_events_get_end_clock_value(self._ptr)
 462
 463         if clock_value_ptr is None:
 464             return
 465
 466         clock_value = bt2.clock_value._create_clock_value_from_ptr(clock_value_ptr)
 467         return clock_value
 468
 469
 470 _NOTIF_TYPE_TO_CLS = {
 471     native_bt.NOTIFICATION_TYPE_EVENT: EventNotification,
 472     native_bt.NOTIFICATION_TYPE_PACKET_BEGIN: PacketBeginningNotification,
 473     native_bt.NOTIFICATION_TYPE_PACKET_END: PacketEndNotification,
 474     native_bt.NOTIFICATION_TYPE_STREAM_BEGIN: StreamBeginningNotification,
 475     native_bt.NOTIFICATION_TYPE_STREAM_END: StreamEndNotification,
 476     native_bt.NOTIFICATION_TYPE_INACTIVITY: InactivityNotification,
 477     native_bt.NOTIFICATION_TYPE_DISCARDED_PACKETS: _DiscardedPacketsNotification,
 478     native_bt.NOTIFICATION_TYPE_DISCARDED_EVENTS: _DiscardedEventsNotification,
 479 }