如何在 Kivy 中将 android.bluetooth.socket 输入流转换为 python 字符串?

2022-01-15 00:00:00 kivy android java

我正在开发 Kivy 应用程序.

I'm working on Kivy app.

因为我想从蓝牙适配器获取数据，所以我使用了下面的代码.

Since I want to get data from bluetooth adapter, I used code below.

from kivy.app import App
from kivy.uix.label import Label
from kivy.uix.scatter import Scatter
from kivy.properties import ObjectProperty,NumericProperty
from kivy.clock import Clock
from kivy.lang import Builder
from jnius import cast,autoclass
from kivy.logger import Logger

BluetoothAdapter = autoclass('android.bluetooth.BluetoothAdapter')
bufferedreader = autoclass('android.bluetooth.BluetoothAdapter')
BluetoothDevice = autoclass('android.bluetooth.BluetoothDevice')
BluetoothSocket = autoclass('android.bluetooth.BluetoothSocket')
InputStreamReader = autoclass('java.io.InputStreamReader')
BufferedReader = autoclass('java.io.BufferedReader')
UUID = autoclass('java.util.UUID')
StringBuilder = autoclass('java.lang.StringBuilder')

Builder.load_string('''
<bluetooth>:
    Button:
        pos:root.width/3,root.height/2
        text: root.data
        size: (300,100)


''')

class bluetooth(Scatter):
    socket = ObjectProperty(None,allownone = True)
    data = ObjectProperty('getting data',allownone = True)
    recv = ObjectProperty(None,allownone = True)
    counter = NumericProperty(0)

    def change_data(self,dt):
        Logger.info('Im in the change_data!!')
        self.data = 'change_data'
        paired_devices = BluetoothAdapter.getDefaultAdapter().getBondedDevices().toArray()
        for device in paired_devices:
            self.data = str(device.getName())
            Logger.info('Im in the loop!!'+str(device))
            if device.getName() == 'HC-06':

                self.socket = device.createRfcommSocketToServiceRecord(UUID.fromString('00001101-0000-1000-8000-00805F9B34FB'))
                bufferedreader = BufferedReader(InputStreamReader(self.socket.getInputStream(),"UTF-8"))

                StringBuilder.append(bufferedreader.read())
                self.data = StringBuilder.toString()

        #if self.socket == None:
        #   pass
        #else:
        #   self.socket.connect()
class myApp(App):
    def build(self):
        bt = bluetooth()
        Clock.schedule_interval(bt.change_data,1)
        return bt
myApp().run()

也许我错过了一些代码..我不知道如何将 bluetooth.socket 输入流转换为 python 字符串.有人可以帮忙吗?

Maybe I missed some code.. I can't find out how to get bluetooth.socket inputstream to python string. Can someone please help?

推荐答案

我终于找到了一个似乎可行的解决方案.我有一个 Kivy 应用程序通过蓝牙与基于 Arduino 的设备通信.在 Arduino 上，我使用 SerialCommand 库来接收自定义命令并做出相应的响应.当命令在主线程中发送到我的 Arduino 时，我有一个带有循环的第二个线程，它从我的蓝牙套接字读取 InputStream.来自 Arduino 的响应包含在 <> 中，当我得到正确的响应时，我会提取括号之间的文本并将其发送到我的主线程中的一个函数.我希望这对你有帮助.

I have finally find a solution that seems to work. I have a Kivy app communicating with an Arduino based device over Bluetooth. On the Arduino I use the SerialCommand library to recieve custom commands and respond accordingly. While the commands is send to my Arduino in the main thread, I have a second thread with a loop that reads the InputStream from my Bluetooth socket. The response from Arduino is enclosed with <>, and when I get a proper response I extract the text between the brackets and send it to a function in my mainthread. I hope this is helpful for you.

from kivy.clock import mainthread
import threading
import jnius

def get_socket_stream(self, name):
    paired_devices =  self.BluetoothAdapter.getDefaultAdapter().getBondedDevices().toArray()
    socket = None
    for device in paired_devices:
        if device.getName() == name:
            socket = device.createRfcommSocketToServiceRecord(
            self.UUID.fromString("00001101-0000-1000-8000-00805F9B34FB"))
            reader = self.InputStreamReader(socket.getInputStream(), 'US-ASCII')
            recv_stream = self.BufferedReader(reader)
            send_stream = socket.getOutputStream()
            break
    socket.connect()
    return recv_stream, send_stream

def connect(self, *args):
    device = self.config.get('bluetooth', 'bt_name')
    try:
        self.recv_stream, self.send_stream = self.get_socket_stream(device)
    except AttributeError as e:
        print e.message
        return False
    except jnius.JavaException as e:
        print e.message
        return False
    except:
        print sys.exc_info()[0]
        return False

    threading.Thread(target=self.stream_reader).start()

def stream_reader(self, *args):
    stream = ''
    while True:
        if self.stop.is_set():
            jnius.detach()
            return
        if self.recv_stream.ready():
            try:
                stream = self.recv_stream.readLine()
            except self.IOException as e:
                print "IOException: ", e.message
            except jnius.JavaException as e:
                print "JavaException: ", e.message
            except:
                print "Misc error: ", sys.exc_info()[0]

            try:
                start = stream.rindex("<") + 1
                end = stream.rindex(">", start)
                self.got_response(stream[start:end])
            except ValueError:
                pass

@mainthread
def got_response(self, response):
    do something...

相关文章