def _prepair(self):
'''Try to connect to the given dbus services. If successful it will
return a callable dbus proxy and those arguments.
'''
try:
sessionbus = dbus.SessionBus()
systembus = dbus.SystemBus()
except:
return (None, None)
for dbus_props in self.DBUS_SHUTDOWN.values():
try:
if dbus_props['bus'] == SESSION_BUS:
bus = sessionbus
else:
bus = systembus
interface = bus.get_object(dbus_props['service'],
dbus_props['objectPath'])
proxy = interface.get_dbus_method(dbus_props['method'],
dbus_props['interface'])
return (proxy, dbus_props['arguments'])
except dbus.exceptions.DBusException:
continue
return (None, None)
python类SessionBus()的实例源码
def __init__(self):
self.bm = BacklightManagerForGNOME()
self.desktop = os.getenv('DESKTOP_SESSION', 'gnome')
properties = DBUS_PROPS[self.desktop]
self.callback = None
self.bus = dbus.SessionBus()
bus = dbus.SessionBus()
proxy = bus.get_object(properties['service'], properties['path'])
self.properties_manager = dbus.Interface(
proxy, 'org.freedesktop.DBus.Properties')
self.dbus_interface = dbus.Interface(
proxy, dbus_interface=properties['interface'])
self.get_value = self.dbus_interface.get_dbus_method(
properties['method-get'])
self.set_value = self.dbus_interface.get_dbus_method(
properties['method-set'])
self.callback = None
def notify(summary, body='', app_name='', app_icon='',
timeout=3000, actions=[], hints=[], replaces_id=0):
"""Send a system notification"""
_bus_name = 'org.freedesktop.Notifications'
_object_path = '/org/freedesktop/Notifications'
_interface_name = _bus_name
session_bus = dbus.SessionBus()
obj = session_bus.get_object(_bus_name, _object_path)
interface = dbus.Interface(obj, _interface_name)
interface.Notify(app_name, replaces_id, app_icon,
summary, body, actions, hints, timeout)
## Main ##
# Call battery_check shell script to get info
def notify(summary, body='', app_name='', app_icon='',
timeout=3000, actions=[], hints=[], replaces_id=0):
"""Send a system notification"""
_bus_name = 'org.freedesktop.Notifications'
_object_path = '/org/freedesktop/Notifications'
_interface_name = _bus_name
session_bus = dbus.SessionBus()
obj = session_bus.get_object(_bus_name, _object_path)
interface = dbus.Interface(obj, _interface_name)
interface.Notify(app_name, replaces_id, app_icon,
summary, body, actions, hints, timeout)
## Main ##
# Call battery_check shell script to get info
def notify(summary, body='', app_name='', app_icon='',
timeout=5000, actions=[], hints=[], replaces_id=0):
"""Send a system notification"""
_bus_name = 'org.freedesktop.Notifications'
_object_path = '/org/freedesktop/Notifications'
_interface_name = _bus_name
session_bus = dbus.SessionBus()
obj = session_bus.get_object(_bus_name, _object_path)
interface = dbus.Interface(obj, _interface_name)
interface.Notify(app_name, replaces_id, app_icon,
summary, body, actions, hints, timeout)
## Main ##
# Call battery_check shell script to get info
# TODO(nate): parse text more cleanly
def create_dbus_server(cls, session_bus=False):
'''Return a D-BUS server backend instance.
Normally this connects to the system bus. Set session_bus to True to
connect to the session bus (for testing).
'''
backend = Backend()
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
if session_bus:
backend.bus = dbus.SessionBus()
backend.enforce_polkit = False
else:
backend.bus = dbus.SystemBus()
try:
backend.dbus_name = dbus.service.BusName(DBUS_BUS_NAME, backend.bus)
except dbus.exceptions.DBusException as msg:
logging.error("Exception when spawning dbus service")
logging.error(msg)
return None
return backend
#
# Internal methods
#
def run(self):
GObject.threads_init()
dbus.mainloop.glib.threads_init()
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus_name = dbus.service.BusName(
"com.spoppy",
dbus.SessionBus()
)
super(SpoppyDBusService, self).__init__(
bus_name, "/com/spoppy"
)
self._loop = GObject.MainLoop()
while self.running:
try:
logger.debug('Starting dbus loop')
self._loop.run()
except KeyboardInterrupt:
logger.debug('Loop interrupted, will restart')
def _notify(self, **kwargs):
summary = kwargs.get('title', "title")
body = kwargs.get('message', "body")
app_name = kwargs.get('app_name', '')
app_icon = kwargs.get('app_icon', '')
timeout = kwargs.get('timeout', 10)
actions = kwargs.get('actions', [])
hints = kwargs.get('hints', [])
replaces_id = kwargs.get('replaces_id', 0)
_bus_name = 'org.freedesktop.Notifications'
_object_path = '/org/freedesktop/Notifications'
_interface_name = _bus_name
import dbus
session_bus = dbus.SessionBus()
obj = session_bus.get_object(_bus_name, _object_path)
interface = dbus.Interface(obj, _interface_name)
interface.Notify(app_name, replaces_id, app_icon,
summary, body, actions, hints, timeout * 1000)
def __init__(self):
if dbus.SessionBus().request_name("es.atareao.google-tasks-indicator") != dbus.bus.REQUEST_NAME_REPLY_PRIMARY_OWNER:
print("application already running")
exit(0)
self.indicator = appindicator.Indicator.new('Google-Tasks-Indicator', 'Google-Tasks-Indicator', appindicator.IndicatorCategory.APPLICATION_STATUS)
self.notification = Notify.Notification.new('','', None)
self.tasks = googletasksapi.TaskAlone()
self.tasks.restore()
if self.tasks.tasklists == {}:
tld = TaskListDialog()
if tld.run() == Gtk.ResponseType.ACCEPT:
tld.hide()
self.tasks.create_tasklist(tld.get_title())
tld.destroy()
self.read_preferences()
self.set_menu()
self.menu_update()
def __init__(self):
if dbus.SessionBus().request_name("es.atareao.google-tasks-indicator") != dbus.bus.REQUEST_NAME_REPLY_PRIMARY_OWNER:
print("application already running")
exit(0)
self.indicator = appindicator.Indicator.new('Google-Tasks-Indicator', 'Google-Tasks-Indicator', appindicator.IndicatorCategory.APPLICATION_STATUS)
self.notification = Notify.Notification.new('','', None)
self.tasks = googletasksapi.TaskAlone()
self.read_preferences()
self.tasks.restore()
'''
if self.tasklist_id is None and len(self.tasks.tasklists.values())>0:
for item in self.tasks.tasklists.values():
self.tasklist_id = item['id']
break
'''
self.create_menu()
self.menu_update()
def connected(self, service):
if self.handle >= 0:
return True
bus = dbus.SessionBus()
wId = 0
try:
remote_obj = bus.get_object(self.bus_name, self.object_path)
self.iface = dbus.Interface(remote_obj, 'org.kde.KWallet')
self.handle = self.iface.open(
self.iface.networkWallet(), wId, self.appid)
except dbus.DBusException:
self.handle = -1
if self.handle < 0:
return False
self._migrate(service)
return True
def dbus_method(path, method, *arguments):
"""Invokes dbus method
Invokes method with arguments via dbus.
@param method Dbus method
@param arguments Dictionary of arguments for the method
@return Always None
"""
try:
bus = dbus.SessionBus()
bus_name = dbus.service.BusName(path, bus=bus)
obj = bus.get_object(bus_name.get_name(), "/{}".format(path.replace(".", "/")))
remote_method = obj.get_dbus_method(method, path)
remote_method(*arguments)
except Exception as e:
logger.error('common.swm: dbus_method(%s, %s): Exception: %s', path, method, e)
return None
def _get_region_geoclue(self):
""" return the dict with at least countrycode,country from a geoclue
provider
"""
bus = dbus.SessionBus()
master = bus.get_object(
'org.freedesktop.Geoclue.Master',
'/org/freedesktop/Geoclue/Master')
client = bus.get_object(
'org.freedesktop.Geoclue.Master', master.Create())
client.SetRequirements(AccuracyLevel.COUNTRY, # (i) accuracy_level
0, # (i) time
False, # (b) require_updates
AllowedResources.ALL) # (i) allowed_resoures
# this is crucial
client.AddressStart()
# now get the data
time, address_res, accuracy = client.GetAddress()
return address_res
def _get_region_geoclue(self):
""" return the dict with at least countrycode,country from a geoclue
provider
"""
bus = dbus.SessionBus()
master = bus.get_object(
'org.freedesktop.Geoclue.Master',
'/org/freedesktop/Geoclue/Master')
client = bus.get_object(
'org.freedesktop.Geoclue.Master', master.Create())
client.SetRequirements(AccuracyLevel.COUNTRY, # (i) accuracy_level
0, # (i) time
False, # (b) require_updates
AllowedResources.ALL) # (i) allowed_resoures
# this is crucial
client.AddressStart()
# now get the data
time, address_res, accuracy = client.GetAddress()
return address_res
def im_command(queues: Dict[bytes, 'Queue']) -> None:
"""Loop that executes commands on IM client."""
bus = dbus.SessionBus(private=True)
obj = bus.get_object("im.pidgin.purple.PurpleService", "/im/pidgin/purple/PurpleObject")
purple = dbus.Interface(obj, "im.pidgin.purple.PurpleInterface")
account = purple.PurpleAccountsGetAllActive()[0]
queue = queues[NH_TO_IM_QUEUE]
while True:
with ignored(dbus.exceptions.DBusException, EOFError, KeyboardInterrupt):
while queue.qsize() == 0:
time.sleep(0.01)
command = queue.get()
if command[:2] in [UNENCRYPTED_SCREEN_CLEAR, UNENCRYPTED_SCREEN_RESET]:
contact = command[2:]
new_conv = purple.PurpleConversationNew(1, account, contact)
purple.PurpleConversationClearMessageHistory(new_conv)
def get_current_location2():
'''Gets the current location from geolocation via IP (only method
currently supported)
'''
latitude = 0
longitude = 0
bus = dbus.SessionBus()
# For now we default to the UbuntuGeoIP provider and we fall back to
# Hostip. We should probably be cleverer about provider detection, but
# this solution works for now and does not rely solely on UbuntuGeoIP,
# which means qreator can run on other distros
try:
geoclue = bus.get_object(
'org.freedesktop.Geoclue.Providers.UbuntuGeoIP',
'/org/freedesktop/Geoclue/Providers/UbuntuGeoIP')
position_info = geoclue.GetPosition(
dbus_interface='org.freedesktop.Geoclue.Position')
latitude = convert(position_info[2])
longitude = convert(position_info[3])
except dbus.exceptions.DBusException as e:
print('Error 1', e)
try:
geoclue = bus.get_object(
'org.freedesktop.Geoclue.Providers.Hostip',
'/org/freedesktop/Geoclue/Providers/Hostip')
position_info = geoclue.GetPosition(
dbus_interface='org.freedesktop.Geoclue.Position')
latitude = convert(position_info[2])
longitude = convert(position_info[3])
except dbus.exceptions.DBusException as e:
print('Error 2', e)
return latitude, longitude
def _get_dbus_player_status(pl, bus_name, player_path, iface_prop,
iface_player, player_name='player'):
bus = dbus.SessionBus()
try:
player = bus.get_object(bus_name, player_path)
iface = dbus.Interface(player, iface_prop)
info = iface.Get(iface_player, 'Metadata')
status = iface.Get(iface_player, 'PlaybackStatus')
except dbus.exceptions.DBusException:
return
if not info:
return
album = info.get('xesam:album')
title = info.get('xesam:title')
artist = info.get('xesam:artist')
state = _convert_state(status)
if album:
album = out_u(album)
if title:
title = out_u(title)
if artist:
artist = out_u(artist[0])
return {
'state': state,
'album': album,
'artist': artist,
'title': title,
'total': _convert_seconds(info.get('mpris:length') / 1e6),
}
def main():
if dbus.SessionBus().request_name(
'es.atareao.PomodoroIndicator') !=\
dbus.bus.REQUEST_NAME_REPLY_PRIMARY_OWNER:
print("application already running")
exit(0)
GObject.threads_init()
Gst.init(None)
Gst.init_check(None)
Notify.init('pomodoro-indicator')
Pomodoro_Indicator()
Gtk.main()
def __init__(self, user=False):
self.bus = SessionBus() if user else SystemBus()
systemd = self.bus.get_object(SYSTEMD_BUSNAME, SYSTEMD_PATH)
self.manager = Interface(systemd, dbus_interface=SYSTEMD_MANAGER_INTERFACE)
def get_dbus( bus_type, obj, path, interface, method, arg):
""" utility: executes dbus method on specific interface"""
if bus_type == "session":
bus = dbus.SessionBus()
if bus_type == "system":
bus = dbus.SystemBus()
proxy = bus.get_object(obj, path)
method = proxy.get_dbus_method(method, interface)
if arg:
return method(arg)
else:
return method()
def get_dbus_property( bus_type, obj, path, iface, prop):
""" utility:reads properties defined on specific dbus interface"""
if bus_type == "session":
bus = dbus.SessionBus()
if bus_type == "system":
bus = dbus.SystemBus()
proxy = bus.get_object(obj, path)
aux = 'org.freedesktop.DBus.Properties'
props_iface = dbus.Interface(proxy, aux)
try:
props = props_iface.Get(iface, prop)
return props
except:
return None
def setUpClass(cls):
cls.server = SimlabServerThread()
cls.server.setDaemon(True)
cls.server.start()
time.sleep(1)
cls.bus = dbus.SessionBus()
cls.session = cls.bus.get_object("org.sim.simlab", "/org/sim/simlab")
cls.interface = dbus.Interface(cls.session, "org.sim.simlab")
def __init__(self, taskQueue, resultQueue):
self.taskQueue = taskQueue
self.resultQueue = resultQueue
bus_name = dbus.service.BusName('org.sim.simlab', bus=dbus.SessionBus())
dbus.service.Object.__init__(self, bus_name, '/org/sim/simlab')
def run(self):
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus_name = dbus.service.BusName("com.prozacville.steam_monitor", dbus.SessionBus())
dbus.service.Object.__init__(self, bus_name, "/com/prozacville/steam_monitor")
self._loop = gobject.MainLoop()
print "Service running..."
self._loop.run()
print "Service stopped"
def __init__(self):
bus = dbus.SessionBus()
service = bus.get_object('com.prozacville.steam_monitor', "/com/prozacville/steam_monitor")
self._start = service.get_dbus_method('steam_start', 'com.prozacville.steam_monitor.Start')
self._status = service.get_dbus_method('steam_status', 'com.prozacville.steam_monitor.Status')
self._stop = service.get_dbus_method('steam_stop', 'com.prozacville.steam_monitor.Stop')
self._kill = service.get_dbus_method('steam_kill', 'com.prozacville.steam_monitor.Kill')
self._quit = service.get_dbus_method('quit', 'com.example.service.Quit')
def connect_to_buffy():
if not dbus: return None
session_bus = dbus.SessionBus()
try:
buffy_obj = session_bus.get_object("org.enricozini.buffy", "/buffy")
return dbus.Interface(buffy_obj, dbus_interface="org.enricozini.buffy")
except:
return None
def handle_krunner_plasma_desktopskill_intent(self, message):
utterance = message.data.get('utterance').lower()
utterance = utterance.replace(
message.data.get('KrunnerPlasmaDesktopSkillKeyword'), '')
searchString = utterance
bus = dbus.SessionBus()
remote_object = bus.get_object("org.kde.krunner","/App")
remote_object.query(searchString + ' ', dbus_interface = "org.kde.krunner.App")
self.speak_dialog("krunner.search", data={'Query': searchString})
def handle_krunner_plasma_recentskill_intent(self, message):
bus = dbus.SessionBus()
remote_object = bus.get_object("org.kde.krunner","/App")
remote_object.query('recent' + ' ', dbus_interface = "org.kde.krunner.App")
self.speak_dialog("krunner.recent")
def __init__(self, mqtt_server=None, ca_cert=None, user=None, passwd=None, dbus_address=None,
keep_alive_interval=None, init_broker=False):
self._dbus_address = dbus_address
self._dbus_conn = (dbus.SessionBus() if 'DBUS_SESSION_BUS_ADDRESS' in os.environ else dbus.SystemBus()) \
if dbus_address is None \
else dbus.bus.BusConnection(dbus_address)
self._dbus_conn.add_signal_receiver(self._dbus_name_owner_changed, signal_name='NameOwnerChanged')
self._connected_to_cloud = False
# @todo EV Get portal ID from com.victronenergy.system?
self._system_id = get_vrm_portal_id()
# Key: D-BUS Service + path, value: topic
self._topics = {}
# Key: topic, value: last value seen on D-Bus
self._values = {}
# Key: service_type/device_instance, value: D-Bus service name
self._services = {}
# Key: short D-Bus service name (eg. 1:31), value: full D-Bus service name (eg. com.victronenergy.settings)
self._service_ids = {}
if init_broker:
self._registrator = MosquittoBridgeRegistrator(self._system_id)
self._registrator.register()
else:
self._registrator = None
self._dbus_conn.add_signal_receiver(self._on_dbus_value_changed,
dbus_interface='com.victronenergy.BusItem', signal_name='PropertiesChanged', path_keyword='path',
sender_keyword='service_id')
services = self._dbus_conn.list_names()
for service in services:
if service.startswith('com.victronenergy.'):
self._service_ids[self._dbus_conn.get_name_owner(service)] = service
self._scan_dbus_service(service, publish=False)
# Bus scan may take a log time, so start keep alive after scan
self._keep_alive_interval = keep_alive_interval
self._keep_alive_timer = None
MqttGObjectBridge.__init__(self, mqtt_server, "ve-dbus-mqtt-py", ca_cert, user, passwd)
def update(self, widgets):
try:
bus = dbus.SessionBus()
spotify = bus.get_object("org.mpris.MediaPlayer2.spotify", "/org/mpris/MediaPlayer2")
spotify_iface = dbus.Interface(spotify, 'org.freedesktop.DBus.Properties')
props = spotify_iface.Get('org.mpris.MediaPlayer2.Player', 'Metadata')
self._song = self._format.format(album=str(props.get('xesam:album')),
title=str(props.get('xesam:title')),
artist=','.join(props.get('xesam:artist')),
trackNumber=str(props.get('xesam:trackNumber')))
except Exception:
self._song = ""
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
