问题描述
摘要：我有一个微型位连接到rpi-zero。我对微比特进行了编码，当按下A button时，它将通过uart.write将数据发送到rpi-zero。
在此测试中，微比特将uart.write("Test")并向rpi-zero写入一个“测试”字。
我的最终目标是使用rpi-zero的BLE功能来充当控制设备，并通过微比特按钮发送指令。
我发现这个GATT Server Code用python为rpi编写。完全没有问题。
下面的代码将用于侦听微比特uart服务，并检查接收到的数据是否为"Test"：
import serial
serialPort = serial.Serial(port = "/dev/ttyACM0",baudrate=115200,bytesize=8,timeout=0.5,stopbits=serial.STOPBITS_ONE)
serialString = " "
(serialPort.in_waiting > 0)
while True:
serialString = serialPort.readline()
if serialString == b'Test':
print("Yes")
else:
print("F")
但是真正的问题是当我尝试将此循环代码实现为GATT服务器代码时。
我似乎无法理解如何将此值传递给self.send_tx
此外，GATT服务器代码中似乎已经存在一个全局循环。因此，我尝试使用线程来同时运行两个函数，但是当我添加self.send_tx("Test")时，它只会抛出错误Self is not defined。
对不起，我对编码一无所知，有人知道对此可能的解决方法吗？谢谢
这是完整的代码：
import sys
import threading
import dbus,dbus.mainloop.glib
import serial
from gi.repository import GLib
from example_advertisement import Advertisement
from example_advertisement import register_ad_cb,register_ad_error_cb
from example_gatt_server import Service,Characteristic
from example_gatt_server import register_app_cb,register_app_error_cb
BLUEZ_SERVICE_NAME = 'org.bluez'
DBUS_OM_IFACE = 'org.freedesktop.DBus.ObjectManager'
LE_ADVERTISING_MANAGER_IFACE = 'org.bluez.LEAdvertisingManager1'
GATT_MANAGER_IFACE = 'org.bluez.GattManager1'
GATT_CHRC_IFACE = 'org.bluez.GattCharacteristic1'
UART_SERVICE_UUID = '6e400001-b5a3-f393-e0a9-e50e24dcca9e'
UART_RX_CHARACTERISTIC_UUID = '6e400002-b5a3-f393-e0a9-e50e24dcca9e'
UART_TX_CHARACTERISTIC_UUID = '6e400003-b5a3-f393-e0a9-e50e24dcca9e'
LOCAL_NAME = 'rpi-gatt-server'
mainloop = None
serialPort = serial.Serial(port = "/dev/ttyACM0",timeout=0.8,stopbits=serial.STOPBITS_ONE)
serialString = " "
(serialPort.in_waiting > 0)
class TxCharacteristic(Characteristic):
def __init__(self,bus,index,service):
Characteristic.__init__(self,UART_TX_CHARACTERISTIC_UUID,['notify'],service)
self.notifying = False
GLib.io_add_watch(sys.stdin,GLib.IO_IN,self.on_console_input)
def on_console_input(self,fd,condition):
s = fd.readline()
if s.isspace():
pass
else:
self.send_tx(s)
return True
def send_tx(self,s):
if not self.notifying:
return
value = []
for c in s:
value.append(dbus.Byte(c.encode()))
self.PropertiesChanged(GATT_CHRC_IFACE,{'Value': value},[])
def StartNotify(self):
if self.notifying:
print("yes")
return
self.notifying = True
def StopNotify(self):
if not self.notifying:
print("no")
return
self.notifying = False
class RxCharacteristic(Characteristic):
def __init__(self,UART_RX_CHARACTERISTIC_UUID,['write'],service)
def WriteValue(self,value,options):
print('remote: {}'.format(bytearray(value).decode()))
class UartService(Service):
def __init__(self,index):
Service.__init__(self,UART_SERVICE_UUID,True)
self.add_characteristic(TxCharacteristic(bus,self))
self.add_characteristic(RxCharacteristic(bus,1,self))
class Application(dbus.service.Object):
def __init__(self,bus):
self.path = '/'
self.services = []
dbus.service.Object.__init__(self,self.path)
def get_path(self):
return dbus.ObjectPath(self.path)
def add_service(self,service):
self.services.append(service)
@dbus.service.method(DBUS_OM_IFACE,out_signature='a{oa{sa{sv}}}')
def GetManagedObjects(self):
response = {}
for service in self.services:
response[service.get_path()] = service.get_properties()
chrcs = service.get_characteristics()
for chrc in chrcs:
response[chrc.get_path()] = chrc.get_properties()
return response
class UartApplication(Application):
def __init__(self,bus):
Application.__init__(self,bus)
self.add_service(UartService(bus,0))
class UartAdvertisement(Advertisement):
def __init__(self,index):
Advertisement.__init__(self,'peripheral')
self.add_service_uuid(UART_SERVICE_UUID)
self.add_local_name(LOCAL_NAME)
self.include_tx_power = True
def find_adapter(bus):
remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME,'/'),DBUS_OM_IFACE)
objects = remote_om.GetManagedObjects()
for o,props in objects.items():
if LE_ADVERTISING_MANAGER_IFACE in props and GATT_MANAGER_IFACE in props:
return o
print('Skip adapter:',o)
return None
def check():
while True:
serialString = serialPort.readline()
if serialString == b'Test':
print("Okay,Test")
self.send_tx("Test")
else:
print("No")
def main():
global mainloop
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SystemBus()
adapter = find_adapter(bus)
if not adapter:
print('BLE adapter not found')
return
service_manager = dbus.Interface(
bus.get_object(BLUEZ_SERVICE_NAME,adapter),GATT_MANAGER_IFACE)
ad_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME,LE_ADVERTISING_MANAGER_IFACE)
app = UartApplication(bus)
adv = UartAdvertisement(bus,0)
mainloop = GLib.MainLoop()
service_manager.RegisterApplication(app.get_path(),{},reply_handler=register_app_cb,error_handler=register_app_error_cb)
ad_manager.RegisterAdvertisement(adv.get_path(),reply_handler=register_ad_cb,error_handler=register_ad_error_cb)
try:
mainloop.run()
except KeyboardInterrupt:
adv.Release()
if __name__ == '__main__':
p1 = threading.Thread(target=main)
p2 = threading.Thread(target=check)
p1.start()
p2.start()
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）