dbussy_examples/list_bus_names at master · ldo/dbussy_examples · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
#!/usr/bin/python3
#+
# DBussy example -- ask the D-Bus daemon for a list of all registered
# names on a bus. This version uses an event loop and a proxy
# interface class generated from an Introspection with
# ravel.def_proxy_interface. Invoke this script as follows:
#
#     list_bus_names [--activatable] [--owners] [--queued] [--unique] session|system
#
# where --activatable specifies you want a list of activatable (as
# opposed to active) names, --owners shows the unique bus names owning
# the names, --queued shows bus peers in the ownership queue
# (excluding the actual owner if --owner is also specified), and
# --unique means you want the unique names (beginning with a colon and
# assigned by the D-Bus daemon) as opposed to the requested names.
# The single argument is the bus for which to list the names.
#
# Copyright 2017 by Lawrence D'Oliveiro <ldo@geek-central.gen.nz>. This
# script is licensed CC0
# <https://creativecommons.org/publicdomain/zero/1.0/>; do with it
# what you will.
#-

import sys
import pwd
import getopt
import dbussy as dbus
from dbussy import \
    DBUS
import ravel

activatable = False
show_owners = False
show_pid = False
show_queued = False
show_user = False
unique = False
opts, args = getopt.getopt \
  (
    sys.argv[1:],
    "",
    ["activatable", "owners", "pid", "queued", "unique", "user"]
  )
for keyword, value in opts :
    if keyword == "--activatable" :
        activatable = True
    elif keyword == "--owners" :
        show_owners = True
    elif keyword == "--pid" :
        show_pid = True
    elif keyword == "--queued" :
        show_queued = True
    elif keyword == "--unique" :
        unique = True
    elif keyword == "--user" :
        show_user = True
    #end if
#end for
if len(args) != 1 :
    raise getopt.GetoptError("usage: %s session|system" % sys.argv[0])
#end if

conn = dbus.Connection.bus_get \
  (
    {"session" : DBUS.BUS_SESSION, "system" : DBUS.BUS_SYSTEM}[args[0].lower()],
    private = False
  )
conn.attach_asyncio()
dbus_daemon = ravel.def_proxy_interface \
  (
    ravel.INTERFACE.CLIENT,
    name = "DBusDaemon",
    introspected = dbus.standard_interfaces[DBUS.INTERFACE_DBUS],
    is_async = True
  )(connection = conn, dest = DBUS.SERVICE_DBUS)["/"]
    # should the path be DBUS.PATH_DBUS? But when I introspect the daemon,
    # it doesn’t report the existence of any nodes for that path.

async def mainline() :
    reply = await (dbus_daemon.ListNames, dbus_daemon.ListActivatableNames)[activatable]()
    names, = reply
    for name in sorted(names) :
        if name.startswith(":") == unique :
            sys.stdout.write(name)
            if show_owners :
                reply = await dbus_daemon.GetNameOwner(name)
                owner, = reply
                sys.stdout.write(" -> %s" % owner)
            else :
                owner = None
            #end if
            if show_pid :
                reply = await dbus_daemon.GetConnectionUnixProcessID(name)
                pid, = reply
                sys.stdout.write(" pid %d" % pid)
            #end if
            if show_user :
                reply = await dbus_daemon.GetConnectionUnixUser(name)
                user, = reply
                sys.stdout.write(" user %s(%d)" % (pwd.getpwuid(user).pw_name, user))
            #end if
            if show_queued :
                reply = await dbus_daemon.ListQueuedOwners(name)
                queued, = reply
                sys.stdout.write(" <- [%s]" % ", ".join(repr(q) for q in queued if q != owner))
            #end if
            sys.stdout.write("\n")
        #end if
    #end for
#end mainline

conn.loop.run_until_complete(mainline())