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