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
python类Dictionary()的实例源码
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 _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 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')
