0

我正在使用 BlueZ-obexd 和 pydbus 创建一个用于 opp 文件传输的 python obex 客户端。到目前为止，我已经能够使文件传输正常工作，并且我正在尝试从 API 实现 RemoveSession 以仅在传输完成或失败后触发。我知道 API 中也有一个 status 属性，但我完全不知道如何使用它。我对 dbus 或 pydbus 没有太多经验，所以任何帮助都会很棒。我的代码如下。

import pydbus
import time
import subprocess

# Setup of device specific values
address = subprocess.check_output("bluetoothctl paired-devices", shell=True)
dev_id = str(address)[9:26]
TRANSFER_IFACE = 'org.bluez.obex.Transfer1'
ses = pydbus.SessionBus()
obex = ses.get('org.bluez.obex', '/org/bluez/obex')
ses1 = obex.CreateSession(dev_id, {'Target': pydbus.Variant('s', 'OPP')})
ses1_dbus = ses.get('org.bluez.obex', ses1)
props = ses1_dbus.SendFile('/home/pi/Desktop/image.jpg')

obex.RemoveSession(ses1)
4

1 回答 1

1

的文档SendFile说返回了一个 DBus 对象路径和一个属性字典。返回的对象的接口是org.bluez.obex.Transfer1允许Status监视的。

监视此状态的理想方法是使用 GLib.MainLoop 中的事件循环并观察PropertiesChanged由SendFile.

此外，通过获取配对设备信息，有一个 BlueZ DBus API 可用于获取该信息，而无需subprocess调用bluetoothctl.

我试图做一个例子：

import pydbus
from gi.repository import GLib
from time import sleep

OBEX_TRANSFER = 'org.bluez.obex.Transfer1'
BLUEZ_DEVICE = 'org.bluez.Device1'

ses_bus = pydbus.SessionBus()
sys_bus = pydbus.SystemBus()
mngr = sys_bus.get('org.bluez', '/')

mainloop = GLib.MainLoop()

def paired_devices():
    mngd_objs = mngr.GetManagedObjects()
    for path, iface in mngd_objs.items():
        if BLUEZ_DEVICE in iface:
            addr = iface.get(BLUEZ_DEVICE, {}).get('Address')
            name = iface.get(BLUEZ_DEVICE, {}).get('Name', addr)
            paired = iface.get(BLUEZ_DEVICE, {}).get('Paired')
            if paired:
                return addr
            

def exit_session():
    print('Exit session')
    mainloop.quit()
    obex.RemoveSession(ses1)

def transfer_status_handler(iface, props_changed, props_removed):
    if iface == OBEX_TRANSFER:
        status = props_changed.get('Status')
        if status == 'complete':
            print('Transfer complete')
            exit_session()
        elif status == 'queued':
            print('Still queued')
        elif status == 'active':
            print('transferring')
        elif status == 'suspended':
            print('Suspended')
        elif status == 'error':
            print('error')
            exit_session()

obex = ses_bus.get('org.bluez.obex', '/org/bluez/obex')
my_addr = paired_devices()
ses1 = obex.CreateSession(my_addr, {'Target': pydbus.Variant('s', 'OPP')})
ses1_dbus = ses_bus.get('org.bluez.obex', ses1)
path, props = ses1_dbus.SendFile('/home/pi/.bashrc')
transfer = ses_bus.get('org.bluez.obex', path)
transfer.onPropertiesChanged = transfer_status_handler
try:
    mainloop.run()
except KeyboardInterrupt:
    exit_session()
于 2021-07-27T08:33:51.463 回答