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