#!/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 MyService(dbus.service.Object):
    """自定义 D-Bus 服务，发布信号"""
    
    def __init__(self, bus_name, object_path):
        super().__init__(bus_name, object_path)
        self._counter = 0
    
    # 声明信号（使用 dbus.service.signal 装饰器）
    @dbus.service.signal(dbus_interface='org.example.Signals', signature='si')
    def StatusChanged(self, status, code):
        """状态变化信号，参数: (string status, int32 code)"""
        print(f"发送信号: StatusChanged({status}, {code})")
    
    @dbus.service.signal(dbus_interface='org.example.Signals', signature='a{sv}')
    def DataUpdated(self, data):
        """数据更新信号，参数: (dict data)"""
        print(f"发送信号: DataUpdated({data})")
    
    @dbus.service.method(dbus_interface='org.example.Signals', in_signature='', out_signature='')
    def TriggerSignal(self):
        """方法：触发信号发送"""
        self._counter += 1
        self.StatusChanged("running", self._counter)
        self.DataUpdated({
            'timestamp': dbus.Int64(1234567890),
            'value': dbus.Double(3.14),
            'name': 'test',
        })

# 设置总线
bus = dbus.SessionBus()
bus_name = dbus.service.BusName('org.example.Signals', bus)
service = MyService(bus_name, '/org/example/Signals')

print("信号服务已启动，等待客户端触发...")
print("在另一个终端运行: busctl call --user org.example.Signals /org/example/Signals org.example.Signals TriggerSignal")

loop = GLib.MainLoop()
loop.run()

#include <gio/gio.h>

/* 服务端发布信号 */
static void emit_signal(GDBusConnection *conn) {
    GError *error = NULL;
    
    g_dbus_connection_emit_signal(
        conn,
        NULL,                            /* destination (NULL = broadcast) */
        "/org/example/Signals",          /* object path */
        "org.example.Signals",           /* interface */
        "StatusChanged",                 /* signal name */
        g_variant_new("(si)", "running", 42),  /* parameters */
        &error
    );
    
    if (error) {
        g_printerr("发送信号失败: %s\n", error->message);
        g_error_free(error);
    }
}

# 使用 dbus-send 发送自定义信号
dbus-send --session \
  --type=signal \
  /org/example/Signals \
  org.example.Signals.StatusChanged \
  string:"running" \
  int32:42

# 注意：dbus-send 发送的信号没有总线名称前缀
# 因此 sender 字段将是发送者的唯一名称（如 :1.99）

#!/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_status_changed(status, code):
    """处理 StatusChanged 信号"""
    print(f"收到信号: StatusChanged(status={status}, code={code})")

def on_data_updated(data):
    """处理 DataUpdated 信号"""
    print(f"收到信号: DataUpdated(data={data})")

def on_properties_changed(interface, changed, invalidated):
    """处理 PropertiesChanged 信号（标准接口）"""
    print(f"属性变化: interface={interface}")
    for key, value in changed.items():
        print(f"  {key} = {value}")
    if invalidated:
        print(f"  已失效: {invalidated}")

# 方式 1：通过信号名匹配
bus.add_signal_receiver(
    on_status_changed,
    signal_name='StatusChanged',
    dbus_interface='org.example.Signals',
    bus_name='org.example.Signals',
    path='/org/example/Signals'
)

bus.add_signal_receiver(
    on_data_updated,
    signal_name='DataUpdated',
    dbus_interface='org.example.Signals',
    bus_name='org.example.Signals',
    path='/org/example/Signals'
)

# 方式 2：监听所有 PropertiesChanged 信号
bus.add_signal_receiver(
    on_properties_changed,
    signal_name='PropertiesChanged',
    dbus_interface='org.freedesktop.DBus.Properties',
)

print("正在监听信号，按 Ctrl+C 退出...")
loop.run()

# 方式 1：使用 busctl monitor
busctl monitor --user

# 方式 2：使用 busctl monitor + 匹配规则
busctl monitor --user \
  --match="type='signal',interface='org.example.Signals'"

# 方式 3：使用 dbus-monitor
dbus-monitor --session \
  "type='signal',interface='org.example.Signals'"

# 方式 4：使用 gdbus
gdbus monitor --session --dest org.example.Signals

#include <gio/gio.h>

static void on_signal(GDBusConnection *conn,
                      const gchar *sender,
                      const gchar *object_path,
                      const gchar *interface_name,
                      const gchar *signal_name,
                      GVariant *parameters,
                      gpointer user_data) {
    const gchar *status;
    gint32 code;
    g_variant_get(parameters, "(&si)", &status, &code);
    g_print("收到信号: StatusChanged(%s, %d)\n", status, code);
}

int main(void) {
    GMainLoop *loop = g_main_loop_new(NULL, FALSE);
    GError *error = NULL;
    
    GDBusConnection *conn = g_bus_get_sync(G_BUS_TYPE_SESSION, NULL, &error);
    
    /* 订阅信号 */
    guint sub_id = g_dbus_connection_signal_subscribe(
        conn,
        "org.example.Signals",       /* sender */
        "org.example.Signals",       /* interface */
        "StatusChanged",             /* signal name */
        "/org/example/Signals",      /* object path */
        NULL,                        /* arg0 */
        G_DBUS_SIGNAL_FLAGS_NONE,
        on_signal,
        loop,                        /* user_data */
        NULL                         /* GDestroyNotify */
    );
    
    g_print("正在监听信号...\n");
    g_main_loop_run(loop);
    
    g_dbus_connection_signal_unsubscribe(conn, sub_id);
    g_object_unref(conn);
    g_main_loop_unref(loop);
    return 0;
}

# 通过 D-Bus 方法添加匹配规则
dbus-send --session --dest=org.freedesktop.DBus \
  --type=method_call \
  /org/freedesktop/DBus \
  org.freedesktop.DBus.AddMatch \
  string:"type='signal',sender='org.freedesktop.NetworkManager',member='StateChanged'"

# 删除匹配规则
dbus-send --session --dest=org.freedesktop.DBus \
  --type=method_call \
  /org/freedesktop/DBus \
  org.freedesktop.DBus.RemoveMatch \
  string:"type='signal',sender='org.freedesktop.NetworkManager',member='StateChanged'"

# 匹配所有 PropertiesChanged 信号
"type='signal',member='PropertiesChanged'"

# 匹配 NetworkManager 的所有信号
"type='signal',sender='org.freedesktop.NetworkManager'"

# 匹配特定路径下的接口变化
"type='signal',path_namespace='/org/freedesktop/NetworkManager',member='DeviceAdded'"

# 匹配 arg0 为特定值的信号
"type='signal',interface='org.freedesktop.login1.Manager',member='SessionNew',arg0='c2'"

# 匹配某个路径上所有接口的所有信号
"type='signal',path='/org/example/MyObject'"

#!/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()

def on_state_changed(status, code):
    # 客户端侧过滤：只处理 code > 0 的信号
    if code > 0:
        print(f"处理状态变化: {status} (code={code})")
    else:
        print(f"忽略信号: code={code}")

# 订阅时使用路径前缀过滤
bus.add_signal_receiver(
    on_state_changed,
    signal_name='StatusChanged',
    dbus_interface='org.example.Signals',
    path_keyword='path',     # 将路径作为 keyword 参数传入回调
    sender_keyword='sender', # 将发送方作为 keyword 参数传入回调
)

print("监听中（带过滤）...")
loop.run()

# eavesdrop 允许监听所有消息（包括发给其他客户端的）
# 需要特权（通常是 root 或 dbus-daemon 配置允许）
dbus-monitor --session "eavesdrop='true',type='signal'"

#!/usr/bin/env python3
"""信号接收端节流"""

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

last_time = 0
MIN_INTERVAL = 0.1  # 最小间隔 100ms

def on_data_changed(data):
    global last_time
    now = time.monotonic()
    if now - last_time < MIN_INTERVAL:
        return  # 跳过过于频繁的信号
    last_time = now
    print(f"处理数据: {data}")

bus.add_signal_receiver(
    on_data_changed,
    signal_name='DataUpdated',
    dbus_interface='org.example.Signals',
)

print("节流监听中...")
loop.run()

#!/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()

pending_data = None
timer_id = None

def process_batch():
    """定时处理批量数据"""
    global pending_data, timer_id
    if pending_data is not None:
        print(f"批量处理: {pending_data}")
        pending_data = None
    timer_id = None
    return False  # 不重复定时器

def on_data_changed(data):
    global pending_data, timer_id
    pending_data = data
    # 重置定时器
    if timer_id is not None:
        GLib.source_remove(timer_id)
    timer_id = GLib.timeout_add(200, process_batch)  # 200ms 后处理

bus.add_signal_receiver(
    on_data_changed,
    signal_name='DataUpdated',
    dbus_interface='org.example.Signals',
)

print("防抖监听中...")
loop.run()

# 服务端：定期批量发送，而不是每次变化都发送
class BatchService(dbus.service.Object):
    def __init__(self, bus_name, object_path):
        super().__init__(bus_name, object_path)
        self._pending_updates = {}
        GLib.timeout_add(500, self._flush_updates)  # 每 500ms 发送一次
    
    @dbus.service.signal(dbus_interface='org.example.Signals', signature='a{sv}')
    def BatchUpdated(self, data):
        pass
    
    def update(self, key, value):
        self._pending_updates[key] = value
    
    def _flush_updates(self):
        if self._pending_updates:
            self.BatchUpdated(self._pending_updates)
            self._pending_updates = {}
        return True  # 继续定时器

#!/usr/bin/env python3
"""监控 USB 设备插拔"""

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

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

def interfaces_added(path, interfaces):
    for iface_name, props in interfaces.items():
        if 'org.freedesktop.login1.Seat' in iface_name:
            continue
        print(f"设备添加: {path}")
        for k, v in props.items():
            print(f"  {k} = {v}")

def interfaces_removed(path, interfaces):
    print(f"设备移除: {path}")
    print(f"  接口: {interfaces}")

# 使用 ObjectManager 的信号
bus.add_signal_receiver(
    interfaces_added,
    signal_name='InterfacesAdded',
    dbus_interface='org.freedesktop.DBus.ObjectManager',
    bus_name='org.freedesktop.login1',
)

bus.add_signal_receiver(
    interfaces_removed,
    signal_name='InterfacesRemoved',
    dbus_interface='org.freedesktop.DBus.ObjectManager',
    bus_name='org.freedesktop.login1',
)

print("监控设备插拔...")
loop.run()

#!/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.SystemBus()
loop = GLib.MainLoop()

def on_prepare_for_shutdown(starting):
    if starting:
        print("⚠️ 系统即将关机！正在执行清理...")
        # 在这里执行清理逻辑
    else:
        print("关机准备取消")

def on_prepare_for_sleep(starting):
    if starting:
        print("💤 系统即将休眠！正在保存状态...")
        # 在这里保存状态
    else:
        print("系统唤醒")

bus.add_signal_receiver(
    on_prepare_for_shutdown,
    signal_name='PrepareForShutdown',
    dbus_interface='org.freedesktop.login1.Manager',
)

bus.add_signal_receiver(
    on_prepare_for_sleep,
    signal_name='PrepareForSleep',
    dbus_interface='org.freedesktop.login1.Manager',
)

print("监控电源事件...")
loop.run()