如何使用 pydbus 访问 bluez-obexd 属性(状态)?

How to access bluez-obexd properties(status) using pydbus?

我正在使用 BlueZ-obexd 和 pydbus 创建一个 python obex 客户端用于 opp 文件传输。到目前为止,我已经能够进行文件传输,并且我正在尝试从 API 实施 RemoveSession 以仅在传输完成或失败后触发。我知道 API 中也有状态 属性,但我完全不知道如何使用它。我对 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)

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

监视此状态的理想方法是使用来自 GLib.MainLoop 的事件循环并观察 SendFile.[=18= 返回的 DBus 对象路径上的 PropertiesChanged 信号]

此外,通过获取配对设备信息,有一个 BlueZ DBus API 用于获取该信息,而无需对 bluetoothctl 进行 subprocess 调用。

我试着举了一个例子:

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()