Here is a simple connection project in python.
I posted 2 codes. I have bolded the lines where I face the problem.
The bolded line of first code ,calls the def connect (self, addrport) function of the second(library) code, and in the bolded line of second code it hangs. The device seems to be connected (it appears the bluetooth symbol on screen) and after a while the connection is lost and exeption is thrown.
I have no idea what the bt.connect (self._sockfd, addr, port) does and why it hangs.
GLM_50c.py
import bluetooth # install pybluez before importing
import struct
import binascii
class GLMxxC(object):
device_name = ''
socket = None
port = 0x0005 # depends on model type
bluetooth_address = None
connected = False
cmds = {
'measure': b'\xC0\x40\x00\xEE',
'laser_on': b'\xC0\x41\x00\x96',
'laser_off': b'\xC0\x42\x00\x1E',
'backlight_on': b'\xC0\x47\x00\x20',
'backlight_off': b'\xC0\x48\x00\x62'
}
status = {
0: 'ok',
1: 'communication timeout',
3: 'checksum error',
4: 'unknown command',
5: 'invalid access level',
8: 'hardware error',
10: 'device not ready',
}
#Initializing or Constructor
def __init__(self, bluetooth_address=None):
if bluetooth_address is None:
self.find_GLMxxC()
else:
self.bluetooth_address = bluetooth_address
self.connect()
#Finding the available bluetooth devices
def find_GLMxxC(self):
print('Searching for BOSCH GLMxxC ...')
nearby_devices = bluetooth.discover_devices(
duration=8, lookup_names=True, flush_cache=True, lookup_class=False)
print('0')
for index, val in enumerate(nearby_devices):
addr, name = val
if 'BOSCH GLM' in name.upper():
self.bluetooth_address = addr
print('Found ', name.upper(), ' @', self.bluetooth_address)
self.device_name = name.upper()
if 'GLM50' in self.device_name:
self.port = 0x0005
return
def connect(self):
try:
self.socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
self.socket.connect((self.bluetooth_address, self.port))
self.connected = True
except:
self.socket.close()
self.conencted = False
def measure(self):
self.socket.send(self.cmds['measure'])
data = self.socket.recv(1024)
print('received:', int(binascii.hexlify((data[0]))))
if self.status[int(binascii.hexlify((data[0])))] == 'ok':
try:
# distance to object from top of device
distance = int(struct.unpack("<L", data[2:6])[0])*0.05
return distance
except:
return -1
else:
return -1
def find_bluetooth_services(self):
services = bluetooth.find_service(address=self.bluetooth_address)
if len(services) > 0:
print("found %d services on %s" % (len(services),self.bluetooth_address))
print(services)
else:
print("no services found")
if __name__ == "__main__":
# Add argparse in a future.
try:
#device = GLMxxC(bluetooth_address='00:13:43:A4:93:07')
device = GLMxxC()
except:
print('No devices GLM100C found')
print('after classes built')
# connecting can be speeded up when the mac address of the device is known, e.g.:
# device = GLM100C(bluetooth_address='54:6C:0E:29:92:2F')
try:
print("Trying to connect with "+ device.__class__.__name__)
device.connect()
except ConnectionError:
print ('Can\'t connect with ' + device.__class__.__name__)
# print('')
#device.find_bluetooth_services()
# print('')
if device.connected:
print('Connected BOSCH '+ device.__class__.__name__+'@',device.bluetooth_address)
try:
print('\ntype \'m\' to measure, \n\'lon\' or \'loff\' to turn laser on/off, \n\'bon\' or \'boff\' to turn backlight on/off,\n\'x\' to exit\n')
while True:
data = input()
if data == 'm':
distance = device.measure()
if distance > 0:
print(distance, 'mm from top of device')
print(distance+40.0, 'mm from tripod socket')
print(distance+110.0, 'mm from back of device')
elif data == 'lon':
device.turn_laser_on()
elif data == 'loff':
device.turn_laser_off()
elif data == 'bon':
device.turn_backlight_on()
elif data == 'boff':
device.turn_backlight_off()
elif data == 'x':
device.close()
print('Connection to BOSCH ' + device.__class__.__name__ + ' closed')
break
except KeyboardInterrupt:
device.close()
print('Connection to '+ device.__class__.__name__+' closed')
else:
print('Could not connect to '+ device.__class__.__name__ )
Here is the msbt.py file from python39 Libraries, where the code it hangs.
from bluetooth import *
import bluetooth._msbt as bt
bt.initwinsock ()
# ============== SDP service registration and unregistration ============
def discover_devices (duration=8, flush_cache=True, lookup_names=False,
lookup_class=False, device_id=-1):
#this is order of items in C-code
btAddresIndex = 0
namesIndex = 1
classIndex = 2
try:
devices = bt.discover_devices(duration=duration, flush_cache=flush_cache)
except OSError:
return []
ret = list()
for device in devices:
item = [device[btAddresIndex],]
if lookup_names:
item.append(device[namesIndex])
if lookup_class:
item.append(device[classIndex])
if len(item) == 1: # in case of address-only we return string not tuple
ret.append(item[0])
else:
ret.append(tuple(i for i in item))
return ret
def read_local_bdaddr():
return bt.list_local()
def lookup_name (address, timeout=10):
if not is_valid_address (address):
raise ValueError ("Invalid Bluetooth address")
try:
return bt.lookup_name(address)
except OSError:
return None
class BluetoothSocket:
def __init__ (self, proto = RFCOMM, sockfd = None):
if proto not in [ RFCOMM ]:
raise ValueError ("invalid protocol")
if sockfd:
self._sockfd = sockfd
else:
self._sockfd = bt.socket (bt.SOCK_STREAM, bt.BTHPROTO_RFCOMM)
self._proto = proto
# used by advertise_service and stop_advertising
self._sdp_handle = None
self._raw_sdp_record = None
# used to track if in blocking or non-blocking mode (FIONBIO appears
# write only)
self._blocking = True
self._timeout = False
@property
def family (self):
return bt.AF_BTH
@property
def type (self):
return bt.SOCK_STREAM
@property
def proto (self):
return bt.BTHPROTO_RFCOMM
def bind (self, addrport):
if self._proto == RFCOMM:
addr, port = addrport
if port == 0: port = bt.BT_PORT_ANY
bt.bind (self._sockfd, addr, port)
def listen (self, backlog):
bt.listen (self._sockfd, backlog)
def accept (self):
clientfd, addr, port = bt.accept (self._sockfd)
client = BluetoothSocket (self._proto, sockfd=clientfd)
return client, (addr, port)
def connect (self, addrport):
addr, port = addrport
bt.connect (self._sockfd, addr, port)
def send (self, data):
return bt.send (self._sockfd, data)
def recv (self, numbytes):
return bt.recv (self._sockfd, numbytes)
def close (self):
return bt.close (self._sockfd)
def getsockname (self):
return bt.getsockname (self._sockfd)
def getpeername (self):
return bt.getpeername (self._sockfd)
getpeername.__doc__ = bt.getpeername.__doc__
def setblocking (self, blocking):
bt.setblocking (self._sockfd, blocking)
self._blocking = blocking
def settimeout (self, timeout):
if timeout < 0: raise ValueError ("invalid timeout")
if timeout == 0:
self.setblocking (False)
else:
self.setblocking (True)
bt.settimeout (self._sockfd, timeout)
self._timeout = timeout
def gettimeout (self):
if self._blocking and not self._timeout: return None
return bt.gettimeout (self._sockfd)
def fileno (self):
return self._sockfd
def dup (self):
return BluetoothSocket (self._proto, sockfd=bt.dup (self._sockfd))
def makefile (self):
# TODO
raise Exception("Not yet implemented")
def advertise_service (sock, name, service_id = "", service_classes = [], \
profiles = [], provider = "", description = "", protocols = []):
if service_id != "" and not is_valid_uuid (service_id):
raise ValueError ("invalid UUID specified for service_id")
for uuid in service_classes:
if not is_valid_uuid (uuid):
raise ValueError ("invalid UUID specified in service_classes")
for uuid, version in profiles:
if not is_valid_uuid (uuid) or version < 0 or version > 0xFFFF:
raise ValueError ("Invalid Profile Descriptor")
for uuid in protocols:
if not is_valid_uuid (uuid):
raise ValueError ("invalid UUID specified in protocols")
if sock._raw_sdp_record is not None:
raise OSError("service already advertised")
avpairs = []
# service UUID
if len (service_id) > 0:
avpairs.append (("UInt16", SERVICE_ID_ATTRID))
avpairs.append (("UUID", service_id))
# service class list
if len (service_classes) > 0:
seq = [ ("UUID", svc_class) for svc_class in service_classes ]
avpairs.append (("UInt16", SERVICE_CLASS_ID_LIST_ATTRID))
avpairs.append (("ElemSeq", seq))
# set protocol and port information
assert sock._proto == RFCOMM
addr, port = sock.getsockname ()
avpairs.append (("UInt16", PROTOCOL_DESCRIPTOR_LIST_ATTRID))
l2cap_pd = ("ElemSeq", (("UUID", L2CAP_UUID),))
rfcomm_pd = ("ElemSeq", (("UUID", RFCOMM_UUID), ("UInt8", port)))
proto_list = [ l2cap_pd, rfcomm_pd ]
for proto_uuid in protocols:
proto_list.append (("ElemSeq", (("UUID", proto_uuid),)))
avpairs.append (("ElemSeq", proto_list))
# make the service publicly browseable
avpairs.append (("UInt16", BROWSE_GROUP_LIST_ATTRID))
avpairs.append (("ElemSeq", (("UUID", PUBLIC_BROWSE_GROUP),)))
# profile descriptor list
if len (profiles) > 0:
seq = [ ("ElemSeq", (("UUID",uuid), ("UInt16",version))) \
for uuid, version in profiles ]
avpairs.append (("UInt16",
BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRID))
avpairs.append (("ElemSeq", seq))
# service name
avpairs.append (("UInt16", SERVICE_NAME_ATTRID))
avpairs.append (("String", name))
# service description
if len (description) > 0:
avpairs.append (("UInt16", SERVICE_DESCRIPTION_ATTRID))
avpairs.append (("String", description))
# service provider
if len (provider) > 0:
avpairs.append (("UInt16", PROVIDER_NAME_ATTRID))
avpairs.append (("String", provider))
sock._raw_sdp_record = sdp_make_data_element ("ElemSeq", avpairs)
# pr = sdp_parse_raw_record (sock._raw_sdp_record)
# for attrid, val in pr.items ():
# print "%5s: %s" % (attrid, val)
# print binascii.hexlify (sock._raw_sdp_record)
# print repr (sock._raw_sdp_record)
sock._sdp_handle = bt.set_service_raw (sock._raw_sdp_record, True)
def stop_advertising (sock):
if sock._raw_sdp_record is None:
raise OSError("service isn't advertised, " \
"but trying to un-advertise")
bt.set_service_raw (sock._raw_sdp_record, False, sock._sdp_handle)
sock._raw_sdp_record = None
sock._sdp_handle = None
def find_service (name = None, uuid = None, address = None):
if address is not None:
addresses = [ address ]
else:
addresses = discover_devices (lookup_names = False)
results = []
for addr in addresses:
uuidstr = uuid or PUBLIC_BROWSE_GROUP
if not is_valid_uuid (uuidstr): raise ValueError ("invalid UUID")
uuidstr = to_full_uuid (uuidstr)
dresults = bt.find_service (addr, uuidstr)
for dict in dresults:
raw = dict["rawrecord"]
record = sdp_parse_raw_record (raw)
if SERVICE_CLASS_ID_LIST_ATTRID in record:
svc_class_id_list = [ t[1] for t in \
record[SERVICE_CLASS_ID_LIST_ATTRID] ]
dict["service-classes"] = svc_class_id_list
else:
dict["services-classes"] = []
if BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRID in record:
pdl = []
for profile_desc in \
record[BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRID]:
uuidpair, versionpair = profile_desc[1]
pdl.append ((uuidpair[1], versionpair[1]))
dict["profiles"] = pdl
else:
dict["profiles"] = []
dict["provider"] = record.get (PROVIDER_NAME_ATTRID, None)
dict["service-id"] = record.get (SERVICE_ID_ATTRID, None)
# XXX the C version is buggy (retrieves an extra byte or two),
# so get the service name here even though it may have already
# been set
dict["name"] = record.get (SERVICE_NAME_ATTRID, None)
dict["handle"] = record.get (SERVICE_RECORD_HANDLE_ATTRID, None)
# if LANGUAGE_BASE_ATTRID_LIST_ATTRID in record:
# for triple in record[LANGUAGE_BASE_ATTRID_LIST_ATTRID]:
# code_ISO639, encoding, base_offset = triple
#
# if SERVICE_DESCRIPTION_ATTRID in record:
# service_description = record[SERVICE_DESCRIPTION_ATTRID]
if name is None:
results.extend (dresults)
else:
results.extend ([ d for d in dresults if d["name"] == name ])
return results
# =============== DeviceDiscoverer ==================
class DeviceDiscoverer:
def __init__ (self):
raise NotImplementedError
