我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用dbus.SessionBus()。
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)
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=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 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
def main(): print('=== starting Backlight Indicator ===') if dbus.SessionBus().request_name('es.atareao.BacklightIndicator') != \ dbus.bus.REQUEST_NAME_REPLY_PRIMARY_OWNER: print("application already running") exit(0) Notify.init('BacklightIndicator') BacklightIndicator() print('=== started Backlight Indicator === ') Gtk.main()
def notify(summary, body='', app_name='', app_icon='', timeout=DEFAULT_TIMEOUT, actions=[], hints=[], replaces_id=0): _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) # If run as a script, just display the argv as summary
def __init__(self, ui, home, tr_ay, new_tr): global tray, new_tray_widget tray = tr_ay new_tray_widget = new_tr bus = dbus.service.BusName( AW_MPRIS_BUS_NAME, bus=dbus.SessionBus()) super().__init__(bus, MPRIS_OBJECT_PATH) self._properties = dbus.Dictionary({ 'DesktopEntry': 'kawaii-player', 'Identity': 'kawaii-player', }, signature='sv') self._player_properties = dbus.Dictionary({ 'Metadata': dbus.Dictionary({ 'mpris:artUrl': '', 'xesam:artist': ['None'], 'xesam:title': 'None', 'xesam:album': 'None' }, signature='sv', variant_level=1), 'CanGoNext': True, 'CanGoPrevious': True, 'CanPause': True, 'CanPlay': True, 'CanControl': True, 'CanStop': True, }, signature='sv', variant_level=2) self.ui = ui self.home = home
def linux_status(): try: session_bus = dbus.SessionBus() spotify_bus = session_bus.get_object("org.mpris.MediaPlayer2.spotify", "/org/mpris/MediaPlayer2") spotify_properties = dbus.Interface(spotify_bus, "org.freedesktop.DBus.Properties") status = spotify_properties.Get("org.mpris.MediaPlayer2.Player", "PlaybackStatus") return status except: return "Paused"
def song_info_linux(): if linux_status() == "Playing": session_bus = dbus.SessionBus() spotify_bus = session_bus.get_object("org.mpris.MediaPlayer2.spotify", "/org/mpris/MediaPlayer2") spotify_properties = dbus.Interface(spotify_bus, "org.freedesktop.DBus.Properties") metadata = spotify_properties.Get( "org.mpris.MediaPlayer2.Player", "Metadata") song_info = metadata['xesam:title'] return song_info else: return "There is nothing playing at this moment"
def artist_info_linux(): if linux_status() == "Playing": session_bus = dbus.SessionBus() spotify_bus = session_bus.get_object("org.mpris.MediaPlayer2.spotify", "/org/mpris/MediaPlayer2") spotify_properties = dbus.Interface(spotify_bus, "org.freedesktop.DBus.Properties") metadata = spotify_properties.Get( "org.mpris.MediaPlayer2.Player", "Metadata") artist_info = metadata['xesam:artist'][0] return artist_info else: return "There is nothing playing at this moment"
def next(): if platform.system() == 'Linux': bus = dbus.SessionBus() proxy = bus.get_object( 'org.mpris.MediaPlayer2.spotify', '/org/mpris/MediaPlayer2') interface = dbus.Interface( proxy, dbus_interface='org.mpris.MediaPlayer2.Player') interface.Next() elif platform.system() == 'Windows': win32api.keybd_event(Media_Next, hwcode(Media_Next))
def previous(): if platform.system() == 'Linux': bus = dbus.SessionBus() proxy = bus.get_object( 'org.mpris.MediaPlayer2.spotify', '/org/mpris/MediaPlayer2') interface = dbus.Interface( proxy, dbus_interface='org.mpris.MediaPlayer2.Player') interface.Previous() elif platform.system() == 'Windows': win32api.keybd_event(Media_Previous, hwcode(Media_Previous))
def play(): if platform.system() == 'Linux': bus = dbus.SessionBus() proxy = bus.get_object( 'org.mpris.MediaPlayer2.spotify', '/org/mpris/MediaPlayer2') interface = dbus.Interface( proxy, dbus_interface='org.mpris.MediaPlayer2.Player') interface.PlayPause() elif platform.system() == 'Windows': win32api.keybd_event(Media_Pause, hwcode(Media_Pause))
def __init__(self): self.tw = TaskWarrior() builder.add_from_file("gui/timeout.glade") self.wTimeout = builder.get_object("wTimeout") self.pbTimeout = builder.get_object("pbTimeout") self.wContinue = builder.get_object("wContinue") self.lsbReminders = builder.get_object("lsbReminders") self.bus = dbus.SessionBus() self.session_bus = self.bus.get_object('org.liloman.pomodoro', "/daemon") self.interface = dbus.Interface(self.session_bus, "org.liloman.pomodoroInterface") ################ # Set events # ################ self.btYes = builder.get_object("btYes") self.btYes.connect("clicked",self.onYesPressed) self.btNo = builder.get_object("btNo") self.btNo.connect("clicked",self.onNoPressed) self.wTimeout.connect("delete-event",self.onDeleteWindow) self.btBack = builder.get_object("btBackWork") self.btBack.connect("clicked",self.onBackWorkPressed) self.pbTimeout = builder.get_object("pbTimeout") DATEFORMAT='%d/%m/%Y %H:%M' for task in self.tw.tasks.filter('+READY +reminder'): #get all fields in task task.refresh() self.addReminder(task['description'],task['due'].strftime(DATEFORMAT)) ############### # Reminders # ###############
def doCommand(self,comm): self.com = comm #Quit daemon and systray if self.com == "quit": #close systray when closing the daemon as well self.interface.do_quit(True) print("pomodoro daemon halted") # Show the change task gui elif self.com == "systray": # use busConnection better than bus = dbus.SessionBus() to work with systemd for example self.bus = dbus.bus.BusConnection(self.dbus_path) self.session_bus = self.bus.get_object('org.liloman.pomodoro.systray', "/systray") systray = dbus.Interface(self.session_bus, "org.liloman.pomodoro.systrayInterface") systray.show_change_task() # Start a uuid task elif self.com == "do_start": if len(sys.argv) != 3: print("must pass a valid uuid") sys.exit(1) dic = dbus.Dictionary({'uuid': sys.argv[2] , 'resume': 'No' } , signature = 'ss' ) try: print ("reply:"+self.interface.do_start(dic)[0]) except: print("Incorrect uuid:"+sys.argv[2]+" .The task couldn't be started") # Do any other command else: print(u''.join(self.interface.do_fsm(self.com)[0]).encode('utf-8').strip())
def __init__(self, rc = '~/.task'): self.tw = TaskWarrior(data_location=rc, create=True) self.status_icon = Gtk.StatusIcon() self.status_icon.set_from_file("images/iconStarted-0.png") self.status_icon.connect("popup-menu", self.right_click_event) self.status_icon.connect("activate", self.left_click_event) # systray daemon name = dbus.service.BusName(self.bus_name, bus=dbus.SessionBus(),do_not_queue=True, replace_existing=False, allow_replacement=False ) dbus.service.Object.__init__(self, name, '/systray') # client for daemon bus = dbus.SessionBus(private = True) daemon_client = bus.get_object('org.liloman.pomodoro', "/daemon") self.interface = dbus.Interface(daemon_client, "org.liloman.pomodoroInterface")
def check_instance(self): ''' Check whether another instance of the setup tool is running already ''' if (dbus.SessionBus().request_name("org.ibus.typingbooster") != dbus.bus.REQUEST_NAME_REPLY_PRIMARY_OWNER): self.__run_message_dialog( _("Another instance of this app is already running."), Gtk.MessageType.ERROR) sys.exit(1) else: return False
def __init__(self): bus_name = dbus.service.BusName( 'org.ibus.typingbooster', bus=dbus.SessionBus()) dbus.service.Object.__init__(self, bus_name, '/org/ibus/typingbooster')
def __init__(self): busname = dbus.service.BusName('org.documentroot.AAuname', bus=dbus.SessionBus()) dbus.service.Object.__init__(self, busname, '/AAuname')
def __init__(self, sony_av_indicator, device_service, state_service, command_service): threading.Thread.__init__(self) self.sony_av_indicator = sony_av_indicator self.device_service = device_service self.state_service = state_service self.command_service = command_service self.properties = { ROOT_INTERFACE: self._get_root_iface_properties(), PLAYER_INTERFACE: self._get_player_iface_properties() } self.main_loop = dbus.mainloop.glib.DBusGMainLoop(set_as_default = True) # self.main_loop = GObject.MainLoop() self.bus = dbus.SessionBus(mainloop = self.main_loop) self.bus_name = self._connect_to_dbus() dbus.service.Object.__init__(self, self.bus_name, OBJECT_PATH)
def _connect_to_dbus(self): # bus_type = self.config['mpris']['bus_type'] self.bus = dbus.SessionBus() bus_name = dbus.service.BusName(BUS_NAME, self.bus) return bus_name
def __init__(self, window): self.window = window self.session = dbus.SessionBus() self.menu_items = dict() self.menu_actions = dict() self.explore_menu_paths() self.explore_menu_items()
def reboot(self, menu, widget): session_bus = dbus.SessionBus() obj = session_bus.get_object('org.gnome.SessionManager','/org/gnome/SessionManager') interface = dbus.Interface(obj,'org.gnome.SessionManager') interface.RequestReboot() self.ukuiMenuWin.hide() Gdk.flush()
