mirror of
https://github.com/google/bumble.git
synced 2026-05-08 03:58:01 +00:00
@@ -5068,6 +5068,7 @@ class HCI_Event(HCI_Packet):
|
||||
hci_packet_type = HCI_EVENT_PACKET
|
||||
event_names: Dict[int, str] = {}
|
||||
event_classes: Dict[int, Type[HCI_Event]] = {}
|
||||
vendor_factory: Optional[Callable[[bytes], Optional[HCI_Event]]] = None
|
||||
|
||||
@staticmethod
|
||||
def event(fields=()):
|
||||
@@ -5125,37 +5126,41 @@ class HCI_Event(HCI_Packet):
|
||||
|
||||
return event_class
|
||||
|
||||
@staticmethod
|
||||
def from_bytes(packet: bytes) -> HCI_Event:
|
||||
@classmethod
|
||||
def from_bytes(cls, packet: bytes) -> HCI_Event:
|
||||
event_code = packet[1]
|
||||
length = packet[2]
|
||||
parameters = packet[3:]
|
||||
if len(parameters) != length:
|
||||
raise InvalidPacketError('invalid packet length')
|
||||
|
||||
cls: Any
|
||||
subclass: Any
|
||||
if event_code == HCI_LE_META_EVENT:
|
||||
# We do this dispatch here and not in the subclass in order to avoid call
|
||||
# loops
|
||||
subevent_code = parameters[0]
|
||||
cls = HCI_LE_Meta_Event.subevent_classes.get(subevent_code)
|
||||
if cls is None:
|
||||
subclass = HCI_LE_Meta_Event.subevent_classes.get(subevent_code)
|
||||
if subclass is None:
|
||||
# No class registered, just use a generic class instance
|
||||
return HCI_LE_Meta_Event(subevent_code, parameters)
|
||||
elif event_code == HCI_VENDOR_EVENT:
|
||||
subevent_code = parameters[0]
|
||||
cls = HCI_Vendor_Event.subevent_classes.get(subevent_code)
|
||||
if cls is None:
|
||||
# No class registered, just use a generic class instance
|
||||
return HCI_Vendor_Event(subevent_code, parameters)
|
||||
# Invoke all the registered factories to see if any of them can handle
|
||||
# the event
|
||||
if cls.vendor_factory:
|
||||
if event := cls.vendor_factory(parameters):
|
||||
return event
|
||||
|
||||
# No factory, or the factory could not create an instance,
|
||||
# return a generic vendor event
|
||||
return HCI_Event(event_code, parameters)
|
||||
else:
|
||||
cls = HCI_Event.event_classes.get(event_code)
|
||||
if cls is None:
|
||||
subclass = HCI_Event.event_classes.get(event_code)
|
||||
if subclass is None:
|
||||
# No class registered, just use a generic class instance
|
||||
return HCI_Event(event_code, parameters)
|
||||
|
||||
# Invoke the factory to create a new instance
|
||||
return cls.from_parameters(parameters) # type: ignore
|
||||
return subclass.from_parameters(parameters) # type: ignore
|
||||
|
||||
@classmethod
|
||||
def from_parameters(cls, parameters):
|
||||
@@ -5198,11 +5203,11 @@ HCI_Event.register_events(globals())
|
||||
# -----------------------------------------------------------------------------
|
||||
class HCI_Extended_Event(HCI_Event):
|
||||
'''
|
||||
HCI_Event subclass for events that has a subevent code.
|
||||
HCI_Event subclass for events that have a subevent code.
|
||||
'''
|
||||
|
||||
subevent_names: Dict[int, str] = {}
|
||||
subevent_classes: Dict[int, Type[HCI_Extended_Event]]
|
||||
subevent_classes: Dict[int, Type[HCI_Extended_Event]] = {}
|
||||
|
||||
@classmethod
|
||||
def event(cls, fields=()):
|
||||
@@ -5253,7 +5258,22 @@ class HCI_Extended_Event(HCI_Event):
|
||||
cls.subevent_names.update(cls.subevent_map(symbols))
|
||||
|
||||
@classmethod
|
||||
def from_parameters(cls, parameters):
|
||||
def subclass_from_parameters(
|
||||
cls, parameters: bytes
|
||||
) -> Optional[HCI_Extended_Event]:
|
||||
"""
|
||||
Factory method that parses the subevent code, finds a registered subclass,
|
||||
and creates an instance if found.
|
||||
"""
|
||||
subevent_code = parameters[0]
|
||||
if subclass := cls.subevent_classes.get(subevent_code):
|
||||
return subclass.from_parameters(parameters)
|
||||
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def from_parameters(cls, parameters: bytes) -> HCI_Extended_Event:
|
||||
"""Factory method for subclasses (the subevent code has already been parsed)"""
|
||||
self = cls.__new__(cls)
|
||||
HCI_Extended_Event.__init__(self, self.subevent_code, parameters)
|
||||
if fields := getattr(self, 'fields', None):
|
||||
@@ -5294,12 +5314,6 @@ class HCI_LE_Meta_Event(HCI_Extended_Event):
|
||||
HCI_LE_Meta_Event.register_subevents(globals())
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
class HCI_Vendor_Event(HCI_Extended_Event):
|
||||
event_code: int = HCI_VENDOR_EVENT
|
||||
subevent_classes = {}
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@HCI_LE_Meta_Event.event(
|
||||
[
|
||||
@@ -6173,8 +6187,9 @@ class HCI_Command_Complete_Event(HCI_Event):
|
||||
See Bluetooth spec @ 7.7.14 Command Complete Event
|
||||
'''
|
||||
|
||||
return_parameters = b''
|
||||
num_hci_command_packets: int
|
||||
command_opcode: int
|
||||
return_parameters = b''
|
||||
|
||||
def map_return_parameters(self, return_parameters):
|
||||
'''Map simple 'status' return parameters to their named constant form'''
|
||||
@@ -6710,6 +6725,14 @@ class HCI_Remote_Host_Supported_Features_Notification_Event(HCI_Event):
|
||||
'''
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
@HCI_Event.event([('data', "*")])
|
||||
class HCI_Vendor_Event(HCI_Event):
|
||||
'''
|
||||
See Bluetooth spec @ 5.4.4 HCI Event packet
|
||||
'''
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
class HCI_AclDataPacket(HCI_Packet):
|
||||
'''
|
||||
|
||||
Reference in New Issue
Block a user