1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
# Copyright 2021-2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import collections
import ctypes
import logging
import os
import socket
import struct
from bumble.transport.common import ParserSource, Transport
# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logger = logging.getLogger(__name__)
# -----------------------------------------------------------------------------
async def open_hci_socket_transport(spec: str | None) -> Transport:
'''
Open an HCI Socket (only available on some platforms).
The parameter string is either empty (to use the first/default Bluetooth adapter)
or a 0-based integer to indicate the adapter number.
'''
HCI_CHANNEL_USER = 1 # pylint: disable=invalid-name
# Create a raw HCI socket
try:
hci_socket = socket.socket(
socket.AF_BLUETOOTH, # type: ignore[attr-defined]
socket.SOCK_RAW | socket.SOCK_NONBLOCK, # type: ignore[attr-defined]
socket.BTPROTO_HCI, # type: ignore[attr-defined]
)
except AttributeError as error:
# Not supported on this platform
logger.info("HCI sockets not supported on this platform")
raise Exception(
'Bluetooth HCI sockets not supported on this platform'
) from error
# Compute the adapter index
adapter_index = int(spec) if spec else 0
# Bind the socket
# NOTE: since Python doesn't support binding with the required address format (yet),
# we need to go directly to the C runtime...
try:
ctypes.cdll.LoadLibrary('libc.so.6')
libc = ctypes.CDLL('libc.so.6', use_errno=True)
except OSError as error:
logger.info("HCI sockets not supported on this platform")
raise Exception(
'Bluetooth HCI sockets not supported on this platform'
) from error
libc.bind.argtypes = (ctypes.c_int, ctypes.POINTER(ctypes.c_char), ctypes.c_int)
libc.bind.restype = ctypes.c_int
bind_address = struct.pack(
# pylint: disable=no-member
'<HHH',
socket.AF_BLUETOOTH, # type: ignore[attr-defined]
adapter_index,
HCI_CHANNEL_USER,
)
if (
libc.bind(
hci_socket.fileno(),
ctypes.create_string_buffer(bind_address),
len(bind_address),
)
!= 0
):
raise OSError(ctypes.get_errno(), os.strerror(ctypes.get_errno()))
class HciSocketSource(ParserSource):
def __init__(self, hci_socket):
super().__init__()
self.socket = hci_socket
asyncio.get_running_loop().add_reader(
self.socket.fileno(), self.recv_until_would_block
)
def recv_until_would_block(self):
logger.debug('recv until would block +++')
while True:
try:
packet = self.socket.recv(4096)
logger.debug(f'received packet {len(packet)} bytes')
self.parser.feed_data(packet)
except BlockingIOError:
logger.debug('recv would block')
break
def close(self):
asyncio.get_running_loop().remove_reader(self.socket.fileno())
class HciSocketSink:
def __init__(self, hci_socket):
self.socket = hci_socket
self.packets = collections.deque()
self.writer_added = False
def send_until_would_block(self):
logger.debug('send until would block ---')
while self.packets:
packet = self.packets.pop()
logger.debug('sending packet')
try:
bytes_written = self.socket.send(packet)
except BlockingIOError:
bytes_written = 0
if bytes_written != len(packet):
# Note: we assume here that there are no partial writes
logger.debug('send would block')
break
if self.packets:
# There's still something to send, ensure that we are monitoring the
# socket
if not self.writer_added:
asyncio.get_running_loop().add_writer(
# pylint: disable=no-member
self.socket.fileno(),
self.send_until_would_block,
)
self.writer_added = True
else:
# Nothing left to send, stop monitoring the socket
if self.writer_added:
asyncio.get_running_loop().remove_writer(self.socket.fileno())
self.writer_added = False
def on_packet(self, packet):
self.packets.appendleft(packet)
self.send_until_would_block()
def close(self):
if self.writer_added:
asyncio.get_running_loop().remove_writer(self.socket.fileno())
class HciSocketTransport(Transport):
def __init__(self, hci_socket, source, sink):
super().__init__(source, sink)
self.socket = hci_socket
async def close(self):
logger.debug('closing HCI socket transport')
self.source.close()
self.sink.close()
self.socket.close()
packet_source = HciSocketSource(hci_socket)
packet_sink = HciSocketSink(hci_socket)
return HciSocketTransport(hci_socket, packet_source, packet_sink)
|
