# Debian / Ubuntu
sudo apt install python3-dbus python3-gi gir1.2-glib-2.0

# Fedora / RHEL
sudo dnf install python3-dbus python3-gobject glib2-devel

# pip 安装 pydbus（需要系统级 GLib）
pip3 install pydbus

#!/usr/bin/env python3
"""dbus-python 基本连接"""

import dbus

# 连接到 Session Bus
bus = dbus.SessionBus()

# 连接到 System Bus
sys_bus = dbus.SystemBus()

print(f"Session Bus: {bus}")
print(f"System Bus:  {sys_bus}")

#!/usr/bin/env python3
"""使用 dbus-python 调用 D-Bus 方法"""

import dbus

bus = dbus.SessionBus()

# 获取代理对象
proxy = bus.get_object(
    'org.freedesktop.DBus',    # bus name
    '/org/freedesktop/DBus'    # object path
)

# 获取接口
iface = dbus.Interface(proxy, 'org.freedesktop.DBus')

# 调用方法
names = iface.ListNames()
print("总线名称:")
for name in names:
    print(f"  {name}")

# 使用 Properties 接口
props = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')
version = props.Get('org.freedesktop.DBus', 'Version')
print(f"\nD-Bus 版本: {version}")

#!/usr/bin/env python3
"""发送复杂 D-Bus 类型"""

import dbus

bus = dbus.SessionBus()

# 发送 a{sv} 字典（最常用）
data = dbus.Dictionary({
    'name': dbus.String('test'),
    'count': dbus.Int32(42),
    'pi': dbus.Double(3.14),
    'tags': dbus.Array(['a', 'b'], signature='s'),
}, signature='sv')

# 发送结构体 (si)
struct = dbus.Struct(('hello', 42), signature='si')

# 发送 VARIANT
variant = dbus.Variant(dbus.String('hello'))

# 发送 OBJECT_PATH
path = dbus.ObjectPath('/org/example/MyObj')

#!/usr/bin/env python3
"""使用 dbus-python 创建 D-Bus 服务"""

import dbus
import dbus.service
import dbus.mainloop.glib
from gi.repository import GLib

# 必须在创建总线之前设置主循环
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

class CalculatorService(dbus.service.Object):
    """D-Bus 计算器服务"""
    
    def __init__(self, bus_name, object_path):
        super().__init__(bus_name, object_path)
        self._history = []
    
    # 方法：两数相加
    @dbus.service.method(
        dbus_interface='org.example.Calculator',
        in_signature='dd',      # 输入：两个 double
        out_signature='d'       # 输出：一个 double
    )
    def Add(self, a, b):
        result = a + b
        self._history.append(f"{a} + {b} = {result}")
        self.CalculationPerformed('add', result)
        return result
    
    # 方法：两数相乘
    @dbus.service.method(
        dbus_interface='org.example.Calculator',
        in_signature='dd',
        out_signature='d'
    )
    def Multiply(self, a, b):
        result = a * b
        self._history.append(f"{a} * {b} = {result}")
        self.CalculationPerformed('multiply', result)
        return result
    
    # 方法：获取历史记录
    @dbus.service.method(
        dbus_interface='org.example.Calculator',
        in_signature='',
        out_signature='as'
    )
    def GetHistory(self):
        return self._history
    
    # 方法：清空历史
    @dbus.service.method(
        dbus_interface='org.example.Calculator',
        in_signature='',
        out_signature=''
    )
    def ClearHistory(self):
        self._history.clear()
    
    # 信号：计算完成
    @dbus.service.signal(
        dbus_interface='org.example.Calculator',
        signature='sd'
    )
    def CalculationPerformed(self, operation, result):
        pass
    
    # 属性：获取历史（通过 Properties 接口）
    @dbus.service.method(
        dbus_interface='org.freedesktop.DBus.Properties',
        in_signature='ss',
        out_signature='v'
    )
    def Get(self, interface_name, property_name):
        if interface_name == 'org.example.Calculator':
            if property_name == 'History':
                return dbus.Array(self._history, signature='s')
        raise dbus.exceptions.DBusException(
            f"未知属性: {property_name}"
        )
    
    @dbus.service.method(
        dbus_interface='org.freedesktop.DBus.Properties',
        in_signature='s',
        out_signature='a{sv}'
    )
    def GetAll(self, interface_name):
        if interface_name == 'org.example.Calculator':
            return {
                'History': dbus.Array(self._history, signature='s'),
            }
        return {}

# 创建总线名称和对象
bus = dbus.SessionBus()
bus_name = dbus.service.BusName('org.example.Calculator', bus)
service = CalculatorService(bus_name, '/org/example/Calculator')

print("计算器服务已启动")
print("总线名称: org.example.Calculator")
print("对象路径: /org/example/Calculator")
print("等待客户端调用...")

loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    print("\n服务已停止")

#!/usr/bin/env python3
"""管理多个 D-Bus 对象"""

import dbus
import dbus.service
import dbus.mainloop.glib
from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

class DeviceObject(dbus.service.Object):
    """设备对象"""
    
    def __init__(self, bus_name, object_path, device_name):
        super().__init__(bus_name, object_path)
        self._name = device_name
        self._connected = False
    
    @dbus.service.method(
        dbus_interface='org.example.Device',
        in_signature='',
        out_signature='s'
    )
    def GetName(self):
        return self._name
    
    @dbus.service.method(
        dbus_interface='org.example.Device',
        in_signature='',
        out_signature='b'
    )
    def Connect(self):
        self._connected = True
        self.StatusChanged(True)
        return True
    
    @dbus.service.signal(
        dbus_interface='org.example.Device',
        signature='b'
    )
    def StatusChanged(self, connected):
        pass

# 创建多个设备
bus = dbus.SessionBus()
bus_name = dbus.service.BusName('org.example.Devices', bus)

devices = [
    DeviceObject(bus_name, '/org/example/Devices/0', 'Keyboard'),
    DeviceObject(bus_name, '/org/example/Devices/1', 'Mouse'),
    DeviceObject(bus_name, '/org/example/Devices/2', 'Monitor'),
]

print("设备服务已启动，共注册 3 个设备")
loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    print("\n服务已停止")

#!/usr/bin/env python3
"""D-Bus 客户端调用示例"""

import dbus

bus = dbus.SessionBus()

# 获取服务代理
proxy = bus.get_object('org.example.Calculator', '/org/example/Calculator')
iface = dbus.Interface(proxy, 'org.example.Calculator')

# 调用方法
result = iface.Add(3.14, 2.72)
print(f"3.14 + 2.72 = {result}")

result = iface.Multiply(5.0, 6.0)
print(f"5.0 * 6.0 = {result}")

# 获取历史
history = iface.GetHistory()
print("\n计算历史:")
for entry in history:
    print(f"  {entry}")

#!/usr/bin/env python3
"""监听 D-Bus 信号"""

import dbus
import dbus.mainloop.glib
from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

bus = dbus.SessionBus()
loop = GLib.MainLoop()

def on_calculation(operation, result):
    print(f"收到信号: {operation} = {result}")

bus.add_signal_receiver(
    on_calculation,
    signal_name='CalculationPerformed',
    dbus_interface='org.example.Calculator',
    bus_name='org.example.Calculator',
    path='/org/example/Calculator',
)

print("正在监听计算信号...")
try:
    loop.run()
except KeyboardInterrupt:
    print("\n监听已停止")

#!/usr/bin/env python3
"""通过 Properties 接口访问属性"""

import dbus

bus = dbus.SessionBus()
proxy = bus.get_object('org.example.Calculator', '/org/example/Calculator')
props = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')

# 获取单个属性
history = props.Get('org.example.Calculator', 'History')
print(f"历史记录: {list(history)}")

# 获取所有属性
all_props = props.GetAll('org.example.Calculator')
for key, value in all_props.items():
    print(f"  {key}: {value}")

#!/usr/bin/env python3
"""使用 pydbus 的客户端"""

from pydbus import SessionBus

bus = SessionBus()

# 获取服务（自动创建代理）
calc = bus.get("org.example.Calculator", "/org/example/Calculator")

# 调用方法（像普通 Python 方法一样）
result = calc.Add(3.14, 2.72)
print(f"3.14 + 2.72 = {result}")

result = calc.Multiply(5.0, 6.0)
print(f"5.0 * 6.0 = {result}")

# 读取属性
print(f"历史: {calc.History}")

# 监听信号
def on_calc(op, result):
    print(f"信号: {op} = {result}")

calc.CalculationPerformed.connect(on_calc)

# 保持运行
from gi.repository import GLib
loop = GLib.MainLoop()
loop.run()

#!/usr/bin/env python3
"""使用 pydbus 创建服务"""

from pydbus import SessionBus
from gi.repository import GLib

bus = SessionBus()

class Calculator:
    """D-Bus 计算器服务（pydbus 风格）"""
    
    # 接口定义（XML）
    dbus = """
    <node>
        <interface name='org.example.Calculator'>
            <method name='Add'>
                <arg type='d' direction='in'/>
                <arg type='d' direction='in'/>
                <arg type='d' direction='out'/>
            </method>
            <method name='Multiply'>
                <arg type='d' direction='in'/>
                <arg type='d' direction='in'/>
                <arg type='d' direction='out'/>
            </method>
            <signal name='CalculationPerformed'>
                <arg type='s'/>
                <arg type='d'/>
            </signal>
            <property name='History' type='as' access='read'/>
        </interface>
    </node>
    """
    
    def __init__(self):
        self._history = []
    
    def Add(self, a, b):
        result = a + b
        self._history.append(f"{a} + {b} = {result}")
        self.CalculationPerformed('add', result)
        return result
    
    def Multiply(self, a, b):
        result = a * b
        self._history.append(f"{a} * {b} = {result}")
        self.CalculationPerformed('multiply', result)
        return result
    
    @property
    def History(self):
        return self._history

# 注册服务
calc = Calculator()
bus.publish("org.example.Calculator", calc)

print("计算器服务已启动（pydbus）")
loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    print("\n服务已停止")

#!/usr/bin/env python3
"""dbus-python 异步方法调用"""

import dbus
import dbus.mainloop.glib
from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

bus = dbus.SessionBus()
loop = GLib.MainLoop()

def on_reply(names):
    print("异步收到结果:")
    for name in names[:5]:
        print(f"  {name}")
    print(f"  ... 共 {len(names)} 个")
    loop.quit()

def on_error(error):
    print(f"错误: {error}")
    loop.quit()

proxy = bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus')
iface = dbus.Interface(proxy, 'org.freedesktop.DBus')

# 异步调用
iface.ListNames(
    reply_handler=on_reply,
    error_handler=on_error,
)

print("等待异步结果...")
loop.run()

#!/usr/bin/env python3
"""使用 dasbus + asyncio 进行异步编程"""

import asyncio
from dasbus.connection import SessionMessageBus

async def main():
    bus = SessionMessageBus()
    proxy = bus.get_proxy(
        "org.freedesktop.DBus",
        "/org/freedesktop/DBus"
    )
    
    # dasbus 支持同步调用
    names = proxy.ListNames()
    print("总线名称:")
    for name in names:
        print(f"  {name}")
    
    bus.disconnect()

asyncio.run(main())

#!/usr/bin/env python3
"""
系统监控 D-Bus 服务
提供 CPU/内存使用率、系统运行时间等信息
"""

import os
import time
import psutil
import dbus
import dbus.service
import dbus.mainloop.glib
from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

class SystemMonitor(dbus.service.Object):
    
    def __init__(self, bus_name, object_path):
        super().__init__(bus_name, object_path)
        # 定期发送信号
        GLib.timeout_add_seconds(5, self._emit_update)
    
    @dbus.service.method(
        dbus_interface='org.example.SystemMonitor',
        in_signature='',
        out_signature='a{sv}'
    )
    def GetStats(self):
        return dbus.Dictionary({
            'cpu_percent': dbus.Double(psutil.cpu_percent()),
            'memory_percent': dbus.Double(psutil.virtual_memory().percent),
            'memory_used': dbus.Int64(psutil.virtual_memory().used),
            'memory_total': dbus.Int64(psutil.virtual_memory().total),
            'uptime': dbus.Int64(int(time.time() - psutil.boot_time())),
            'hostname': dbus.String(os.uname().nodename),
        }, signature='sv')
    
    @dbus.service.method(
        dbus_interface='org.example.SystemMonitor',
        in_signature='',
        out_signature='a{sv}'
    )
    def GetCPUInfo(self):
        cpu_freq = psutil.cpu_freq()
        return dbus.Dictionary({
            'physical_cores': dbus.Int32(psutil.cpu_count(logical=False)),
            'logical_cores': dbus.Int32(psutil.cpu_count(logical=True)),
            'current_freq': dbus.Double(cpu_freq.current if cpu_freq else 0),
            'max_freq': dbus.Double(cpu_freq.max if cpu_freq else 0),
        }, signature='sv')
    
    @dbus.service.signal(
        dbus_interface='org.example.SystemMonitor',
        signature='a{sv}'
    )
    def StatsUpdated(self, stats):
        pass
    
    def _emit_update(self):
        stats = self.GetStats()
        self.StatsUpdated(stats)
        return True  # 继续定时器

bus = dbus.SessionBus()
bus_name = dbus.service.BusName('org.example.SystemMonitor', bus)
monitor = SystemMonitor(bus_name, '/org/example/SystemMonitor')

print("系统监控服务已启动")
print("  总线名称: org.example.SystemMonitor")
print("  对象路径: /org/example/SystemMonitor")
print("  每 5 秒发送 StatsUpdated 信号")

loop = GLib.MainLoop()
try:
    loop.run()
except KeyboardInterrupt:
    print("\n服务已停止")

#!/usr/bin/env python3
"""系统监控客户端"""

import dbus
import dbus.mainloop.glib
from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()
loop = GLib.MainLoop()

# 获取服务
proxy = bus.get_object('org.example.SystemMonitor', '/org/example/SystemMonitor')
iface = dbus.Interface(proxy, 'org.example.SystemMonitor')

# 一次性获取
stats = iface.GetStats()
print("系统状态:")
for key, value in stats.items():
    print(f"  {key}: {value}")

# 监听信号
def on_update(stats):
    cpu = stats.get('cpu_percent', 0)
    mem = stats.get('memory_percent', 0)
    print(f"\rCPU: {cpu}%  内存: {mem}%", end='', flush=True)

bus.add_signal_receiver(
    on_update,
    signal_name='StatsUpdated',
    dbus_interface='org.example.SystemMonitor',
)

print("\n监控中...")
loop.run()

# 1. 列出服务的所有接口
import dbus
bus = dbus.SessionBus()
proxy = bus.get_object('org.example.Service', '/org/example')
iface = dbus.Interface(proxy, 'org.freedesktop.DBus.Introspectable')
print(iface.Introspect())

# 2. 检查服务是否存在
print(bus.name_has_owner('org.example.Service'))

# 3. 捕获详细错误
import dbus.exceptions
try:
    iface.SomeMethod()
except dbus.exceptions.DBusException as e:
    print(f"错误名称: {e.get_dbus_name()}")
    print(f"错误消息: {e.get_dbus_message()}")