我们从Python开源项目中，提取了以下34个代码示例，用于说明如何使用dbus.Dictionary()。
def unwrap(self, val): if isinstance(val, dbus.ByteArray): return "".join([str(x) for x in val]) if isinstance(val, (dbus.Array, list, tuple)): return [self.unwrap(x) for x in val] if isinstance(val, (dbus.Dictionary, dict)): return dict([(self.unwrap(x), self.unwrap(y)) for x, y in val.items()]) if isinstance(val, dbus.ObjectPath): if val.startswith('/org/freedesktop/NetworkManager/'): classname = val.split('/')[4] classname = { 'Settings': 'Connection', 'Devices': 'Device', }.get(classname, classname) return globals()[classname](val) if isinstance(val, (dbus.Signature, dbus.String)): return unicode(val) if isinstance(val, dbus.Boolean): return bool(val) if isinstance(val, (dbus.Int16, dbus.UInt16, dbus.Int32, dbus.UInt32, dbus.Int64, dbus.UInt64)): return int(val) if isinstance(val, dbus.Byte): return bytes([int(val)]) return val
def get_properties(self): properties = dict() properties['Type'] = self.ad_type if self.service_uuids is not None: properties['ServiceUUIDs'] = dbus.Array(self.service_uuids, signature='s') if self.solicit_uuids is not None: properties['SolicitUUIDs'] = dbus.Array(self.solicit_uuids, signature='s') if self.manufacturer_data is not None: properties['ManufacturerData'] = dbus.Dictionary( self.manufacturer_data, signature='qv') if self.service_data is not None: properties['ServiceData'] = dbus.Dictionary(self.service_data, signature='sv') if self.include_tx_power is not None: properties['IncludeTxPower'] = dbus.Boolean(self.include_tx_power) return {LE_ADVERTISEMENT_IFACE: properties}
def dict_to_string(self, d): # Try to trivially translate a dictionary's elements into nice string # formatting. dstr="" for key in d: val=d[key] str_val="" add_string=True if type(val)==type(dbus.Array([])): for elt in val: if type(elt)==type(dbus.Byte(1)): str_val+="%s " % int(elt) elif type(elt)==type(dbus.String("")): str_val+="%s" % elt elif type(val)==type(dbus.Dictionary({})): dstr+=self.dict_to_string(val) add_string=False else: str_val=val if add_string: dstr+="%s: %s\n" % ( key, str_val) return dstr
def create_interface(self, interface, bridge_interface=None, driver=None, config_file=None): try: wpa_interface = self.__get_interface() except dbus.exceptions.DBusException as error: raise ServiceError(error) else: args = {"Ifname": interface} if bridge_interface is not None: args["BridgeIfname"] = bridge_interface if driver is not None: args["Driver"] = driver if config_file is not None: args["ConfigFile"] = config_file try: return wpa_interface.CreateInterface(dbus.Dictionary(args, 'sv')) except dbus.exceptions.DBusException as error: raise InterfaceError(error)
def Set(self, interface_name, property_name, value, *args, **kwargs): """Standard D-Bus API for setting a property value""" try: iface_props = self.props[interface_name] except KeyError: raise dbus.exceptions.DBusException( 'no such interface ' + interface_name, name=self.interface + '.UnknownInterface') if property_name not in iface_props: raise dbus.exceptions.DBusException( 'no such property ' + property_name, name=self.interface + '.UnknownProperty') iface_props[property_name] = value self.PropertiesChanged(interface_name, dbus.Dictionary({property_name: value}, signature='sv'), dbus.Array([], signature='s'))
def _return_credentials(self): c = dbus.Dictionary( { dbus.String(u'consumer_secret'): dbus.String( self._random_unicode_string(30)), dbus.String(u'token'): dbus.String( self._random_unicode_string(50)), dbus.String(u'consumer_key'): dbus.String( self._random_unicode_string(7)), dbus.String(u'name'): dbus.String('Ubuntu Software Center @ ' + self._random_unicode_string(6)), dbus.String(u'token_secret'): dbus.String( self._random_unicode_string(50)) }, signature=dbus.Signature('ss') ) return c
def convert(dbus_obj): """Converts dbus_obj from dbus type to python type. :param dbus_obj: dbus object. :returns: dbus_obj in python type. """ _isinstance = partial(isinstance, dbus_obj) ConvertType = namedtuple('ConvertType', 'pytype dbustypes') pyint = ConvertType(int, (dbus.Byte, dbus.Int16, dbus.Int32, dbus.Int64, dbus.UInt16, dbus.UInt32, dbus.UInt64)) pybool = ConvertType(bool, (dbus.Boolean, )) pyfloat = ConvertType(float, (dbus.Double, )) pylist = ConvertType(lambda _obj: list(map(convert, dbus_obj)), (dbus.Array, )) pytuple = ConvertType(lambda _obj: tuple(map(convert, dbus_obj)), (dbus.Struct, )) types_str = (dbus.ObjectPath, dbus.Signature, dbus.String) pystr = ConvertType(str, types_str) pydict = ConvertType( lambda _obj: dict(list(zip(list(map(convert, dbus_obj.keys())), list(map(convert, dbus_obj.values())) )) ), (dbus.Dictionary, ) ) for conv in (pyint, pybool, pyfloat, pylist, pytuple, pystr, pydict): if any(map(_isinstance, conv.dbustypes)): return conv.pytype(dbus_obj) else: return dbus_obj
def add_manufacturer_data(self, manuf_code, data): if not self.manufacturer_data: self.manufacturer_data = dbus.Dictionary({}, signature='qv') self.manufacturer_data[manuf_code] = dbus.Array(data, signature='y')
def add_service_data(self, uuid, data): if not self.service_data: self.service_data = dbus.Dictionary({}, signature='sv') self.service_data[uuid] = dbus.Array(data, signature='y')
def convert(dbus_obj): """Converts dbus_obj from dbus type to python type. :param dbus_obj: dbus object. :returns: dbus_obj in python type. """ _isinstance = partial(isinstance, dbus_obj) ConvertType = namedtuple('ConvertType', 'pytype dbustypes') pyint = ConvertType(int, (dbus.Byte, dbus.Int16, dbus.Int32, dbus.Int64, dbus.UInt16, dbus.UInt32, dbus.UInt64)) pybool = ConvertType(bool, (dbus.Boolean, )) pyfloat = ConvertType(float, (dbus.Double, )) pylist = ConvertType(lambda _obj: list(map(convert, dbus_obj)), (dbus.Array, )) pytuple = ConvertType(lambda _obj: tuple(map(convert, dbus_obj)), (dbus.Struct, )) types_str = (dbus.ObjectPath, dbus.Signature, dbus.String) pystr = ConvertType(str, types_str) pydict = ConvertType( lambda _obj: dict(zip(map(convert, dbus_obj.keys()), map(convert, dbus_obj.values()) ) ), (dbus.Dictionary, ) ) for conv in (pyint, pybool, pyfloat, pylist, pytuple, pystr, pydict): if any(map(_isinstance, conv.dbustypes)): return conv.pytype(dbus_obj) else: return dbus_obj
def base_to_python(val): if isinstance(val, dbus.ByteArray): return "".join([str(x) for x in val]) if isinstance(val, (dbus.Array, list, tuple)): return [fixups.base_to_python(x) for x in val] if isinstance(val, (dbus.Dictionary, dict)): return dict([(fixups.base_to_python(x), fixups.base_to_python(y)) for x, y in val.items()]) if isinstance(val, dbus.ObjectPath): for obj in (NetworkManager, Settings, AgentManager): if val == obj.object_path: return obj if val.startswith('/org/freedesktop/NetworkManager/'): classname = val.split('/')[4] classname = { 'Settings': 'Connection', 'Devices': 'Device', }.get(classname, classname) try: return globals()[classname](val) except ObjectVanished: return None if val == '/': return None if isinstance(val, (dbus.Signature, dbus.String)): return six.text_type(val) if isinstance(val, dbus.Boolean): return bool(val) if isinstance(val, (dbus.Int16, dbus.UInt16, dbus.Int32, dbus.UInt32, dbus.Int64, dbus.UInt64)): return int(val) if isinstance(val, dbus.Byte): return six.int2byte(int(val)) return val
def test_dbus_wrap_dict(self): value = {'a' : 3, 'b': 7.0} dbus_value = dbus_mqtt.wrap_dbus_value(value) print(dbus_value) self.assertIsInstance(dbus_value, dbus.Dictionary) self.assertEqual(dbus.Dictionary({ dbus.String('a', variant_level=1): dbus.Int32(3, variant_level=1), dbus.String('b', variant_level=1): dbus.Double(7.0, variant_level=1)}, variant_level=1), dbus_value)
def test_dbus_unwrap_dict(self): dbus_value = dbus.Dictionary({ dbus.String('a', variant_level=1): dbus.Double(3.2, variant_level=1), dbus.String('b', variant_level=1): dbus.Double(3.7, variant_level=1)}, variant_level=1) value = dbus_mqtt.unwrap_dbus_value(dbus_value) self.assertIsInstance(value, dict) self.assertEqual({'a':3.2, 'b':3.7}, value)
def scan(self): interface = self.__get_interface() try: return interface.Scan(dbus.Dictionary({"Type": "passive"}, 'sv')) except dbus.exceptions.DBusException as error: raise ServiceError(error)
def add_network(self, network): interface = self.__get_interface() try: return interface.AddNetwork(dbus.Dictionary(network, 'sv')) except dbus.exceptions.DBusException as error: raise ServiceError(error)
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 Set(self, interface_name, property_name, value, *args, **kwargs): """Standard D-Bus API for setting a property value""" if property_name not in self.props[constants.GATT_CHRC_IFACE]: raise dbus.exceptions.DBusException( 'no such property ' + property_name, name=constants.GATT_CHRC_IFACE + '.UnknownProperty') self.props[constants.GATT_CHRC_IFACE][property_name] = value return self.PropertiesChanged(interface_name, dbus.Dictionary({property_name: value}, signature='sv'), dbus.Array([], signature='s'))
def Set(self, interface_name, property_name, value, *args, **kwargs): """Standard D-Bus API for setting a property value""" if property_name not in self.props[constants.GATT_DESC_IFACE]: raise dbus.exceptions.DBusException( 'no such property ' + property_name, name=constants.GATT_DESC_IFACE + '.UnknownProperty') self.props[interface_name][property_name] = value return self.PropertiesChanged(interface_name, dbus.Dictionary({property_name: value}, signature='sv'), dbus.Array([], signature='s'))
def get_properties(self): """Return a dictionary of the advert properties. The dictionary has the following keys: - Type: the advertisement type. - ServiceUUIDS: UUIDs of services to advertise - SolicitUUIDS: - ManufacturerData: dictionary of manufacturer data - ServiceData: dictionary of service data - IncludeTxPower: """ properties = dict() properties['Type'] = self.ad_type if self.service_uuids is not None: properties['ServiceUUIDs'] = dbus.Array(self.service_uuids, signature='s') if self.solicit_uuids is not None: properties['SolicitUUIDs'] = dbus.Array(self.solicit_uuids, signature='s') if self.manufacturer_data is not None: properties['ManufacturerData'] = dbus.Dictionary( self.manufacturer_data, signature='qay') if self.service_data is not None: properties['ServiceData'] = dbus.Dictionary(self.service_data, signature='say') if self.include_tx_power is not None: properties['IncludeTxPower'] = dbus.Boolean(self.include_tx_power) return {constants.LE_ADVERTISEMENT_IFACE: properties}
def GetAll(self, interface_name): """Return the advertisement properties. This method is registered with the D-Bus at ``org.freedesktop.DBus.Properties`` :param interface: interface to get the properties of. The interface must be ``org.bluez.LEAdvertisement1`` otherwise an exception is raised. """ if interface_name != constants.LE_ADVERTISEMENT_IFACE: raise InvalidArgsException() response = {} response['Type'] = self.props[interface_name]['Type'] if self.props[interface_name]['ServiceUUIDs'] is not None: response['ServiceUUIDs'] = dbus.Array( self.props[interface_name]['ServiceUUIDs'], signature='s') if self.props[interface_name]['ServiceData'] is not None: response['ServiceData'] = dbus.Dictionary( self.props[interface_name]['ServiceData'], signature='sv') if self.props[interface_name]['ManufacturerData'] is not None: response['ManufacturerData'] = dbus.Dictionary( self.props[interface_name]['ManufacturerData'], signature='qv') if self.props[interface_name]['SolicitUUIDs'] is not None: response['SolicitUUIDs'] = dbus.Array( self.props[interface_name]['SolicitUUIDs'], signature='s') response['IncludeTxPower'] = dbus.Boolean( self.props[interface_name]['IncludeTxPower']) return response
def start(self): self.assertEquals(interface.do_start(dbus.Dictionary({'uuid': self.uuid , 'resume': 'No'}))[0],"started:"+self.uuid)
def test_done_current(self): other_task = Task(self.tw, description="task to be done") other_task.save() uuid2 = other_task['uuid'] self.assertEquals(interface.do_start(dbus.Dictionary({'uuid': uuid2 , 'resume': 'No'}))[0],"started:"+uuid2) prog = re.compile('started .* left.*') self.assertTrue(prog.match(interface.do_fsm("status")[0])) self.assertEquals(interface.done_current()[0],"ok") self.assertTrue(prog.match(interface.do_fsm("status")[0])) self.assertEquals(interface.done_current()[0],"no active task")
def test_resume(self): self.start() self.assertEquals(interface.do_fsm("stop")[0],"ok") self.assertEquals(interface.do_fsm("start")[0],"ok") prog = re.compile('started .* left.*') self.assertTrue(prog.match(interface.do_fsm("status")[0])) self.assertEquals(interface.do_fsm("pause")[0],"ok") prog = re.compile('paused .* left.*') self.assertTrue(prog.match(interface.do_fsm("status")[0])) self.assertEquals(interface.do_fsm("start")[0],"ok") self.assertEquals(interface.do_fsm("start")[0],"Already started") # test done and resume other_task = Task(self.tw, description="task 2 for the test") other_task.save() uuid2 = other_task['uuid'] self.assertEquals(interface.do_start(dbus.Dictionary({'uuid': uuid2 , 'resume': 'No'}))[0],"started:"+uuid2) prog = re.compile('started .* left.*') self.assertTrue(prog.match(interface.do_fsm("status")[0])) self.assertEquals(interface.do_fsm("stop")[0],"ok") other_task.done() self.assertEquals(interface.do_fsm("start")[0],"ok") self.assertTrue(prog.match(interface.do_fsm("status")[0])) # test delete and resume other_task = Task(self.tw, description="task 3 for the test") other_task.save() uuid2 = other_task['uuid'] self.assertEquals(interface.do_start(dbus.Dictionary({'uuid': uuid2 , 'resume': 'No'}))[0],"started:"+uuid2) prog = re.compile('started .* left.*') self.assertTrue(prog.match(interface.do_fsm("status")[0])) self.assertEquals(interface.do_fsm("stop")[0],"ok") other_task.delete() self.assertEquals(interface.do_fsm("start")[0],"ok") self.assertTrue(prog.match(interface.do_fsm("status")[0]))
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 get_metadata(self): metadata = { 'mpris:trackid': self.state_service.source, 'mpris:length': 60000 } return dbus.Dictionary(metadata, signature = 'sv')
def Do (self, params, ack=None, error=None): fuzz = dbus.Dictionary(signature="sv") fuzz.update(**params) self.Phase('put', fuzz); print "DO!", params self.Q.put((params, (ack, error))) return "OK"
def get_properties(self): properties = dict() properties['Type'] = self.ad_type if self.service_uuids is not None: properties['ServiceUUIDs'] = dbus.Array(self.service_uuids, signature='s') if self.solicit_uuids is not None: properties['SolicitUUIDs'] = dbus.Array(self.solicit_uuids, signature='s') if self.manufacturer_data is not None: properties['ManufacturerData'] = dbus.Dictionary(self.manufacturer_data, signature='qv') if self.service_data is not None: properties['ServiceData'] = dbus.Dictionary(self.service_data, signature='sv') if self.include_tx_power is not None: properties['IncludeTxPower'] = dbus.Boolean(self.include_tx_power) return {LE_ADVERTISEMENT_IFACE: properties}
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': 'AnimeWatch', 'Identity': 'animewatch', }, 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 __init__(self): bus = dbus.service.BusName( AW_MPRIS_BUS_NAME, bus=dbus.SessionBus()) super().__init__(bus, MPRIS_OBJECT_PATH) self._properties = dbus.Dictionary({ 'DesktopEntry': 'AnimeWatch', 'Identity': 'animewatch', }, 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, }, signature='sv', variant_level=2)
def run (self, manager, Q): print "running", self, self.running while self.running or not Q.empty( ): item, callbacks = Q.get( ) on_ack, on_error = callbacks print "GOT FROM Q", item results = None # print "ALL SCHEDULED", manager.master.scheduler.schedules trigger = get_trigger_for(item.get('trigger', None), manager.master.scheduler) print "FROM TRIGGER", trigger # http://stackoverflow.com/questions/5943249/python-argparse-and-controlling-overriding-the-exit-status-code fuzz = dbus.Dictionary(item, signature="sv") # fuzz.update(**item) manager.Phase('get', fuzz); update_phase(trigger, 'Running') try: app = DoTool.Make(item) manager.Phase('make', fuzz); print "GOT ITEM", item, app results = app( ) update_phase(trigger, 'Success') manager.Phase('success', fuzz); on_ack(results) # bad time.sleep(0.150) except (Exception, SystemExit), e: print "EXCEPTINO!!!!" print e manager.Phase('error', fuzz); update_phase(trigger, 'Error') time.sleep(0.150) on_error( ) if e: traceback.print_exc(file=sys.stdout) # traceback.print_last( ) finally: # update_phase(trigger, 'Done') # update_phase(trigger, 'Finish') update_phase(trigger, 'Remove') print "DONE", results Q.task_done( )
