def enable_networking(self, val):
"""
function enables/disables networking depending upon the 'val' argument
if val is True, networking is enabled
if val is False, networking is disabled
"""
try:
bus = dbus.SystemBus()
wifi = bus.get_object('org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager')
iface = dbus.Interface(wifi, dbus_interface='org.freedesktop.NetworkManager')
# enabling/disabling networking
m = iface.get_dbus_method("Enable", dbus_interface=None)
m(val)
except:
pass
python类Interface()的实例源码
def get_active_connection_info(self, ac_path):
bus = dbus.SystemBus()
wifi = bus.get_object('org.freedesktop.NetworkManager', ac_path)
iface = dbus.Interface(wifi, dbus_interface='org.freedesktop.DBus.Properties')
# creating proxy 'Get' method
m = iface.get_dbus_method("Get", dbus_interface=None)
# getting Id of active connection
Id = m("org.freedesktop.NetworkManager.Connection.Active", "Id")
# getting Type of active connection
Type = m("org.freedesktop.NetworkManager.Connection.Active", "Type")
# getting Uuid of active connection
Uuid = m("org.freedesktop.NetworkManager.Connection.Active", "Uuid")
# getting State of active connection
State = m("org.freedesktop.NetworkManager.Connection.Active", "State")
# NOTE:
# this function only returns properties like Id, Type, Uuid, State of an active connection
# However, other properties like Dhcp4Config, Dhcp6Config, Ip4Config, Ip6Config etc. can also be obtained
return (str(Id), str(Type), str(Uuid), int(State))
def get_access_point_brief_info(self, ap_path):
bus = dbus.SystemBus()
obj = bus.get_object('org.freedesktop.NetworkManager', ap_path)
iface = dbus.Interface(obj, dbus_interface='org.freedesktop.DBus.Properties')
m = iface.get_dbus_method("Get", dbus_interface=None)
# getting Ssid
dbusArray = m("org.freedesktop.NetworkManager.AccessPoint", "Ssid")
Ssid = ''.join([chr(character) for character in dbusArray])
# getting Strength
Strength = m("org.freedesktop.NetworkManager.AccessPoint", "Strength")
# getting HwAddress
HwAddress = m("org.freedesktop.NetworkManager.AccessPoint", "HwAddress")
# getting Mode
Mode = m("org.freedesktop.NetworkManager.AccessPoint", "Mode")
return (Ssid, int(Strength), str(HwAddress), int(Mode))
def __init__(self, object_path=None):
if isinstance(object_path, NMDbusInterface):
object_path = object_path.object_path
self.object_path = self.object_path or object_path
self.proxy = self.bus.get_object(self.dbus_service, self.object_path)
self.interface = dbus.Interface(self.proxy, self.interface_name)
properties = []
try:
properties = self.proxy.GetAll(self.interface_name,
dbus_interface='org.freedesktop.DBus.Properties')
except dbus.exceptions.DBusException as e:
if e.get_dbus_name() != 'org.freedesktop.DBus.Error.UnknownMethod':
raise
for p in properties:
p = str(p)
if not hasattr(self.__class__, p):
setattr(self.__class__, p, self._make_property(p))
def find_device_in_objects(objects, device_address, adapter_pattern=None):
bus = dbus.SystemBus()
path_prefix = ""
if adapter_pattern:
adapter = find_adapter_in_objects(objects, adapter_pattern)
path_prefix = adapter.object_path
for path, ifaces in objects.iteritems():
device = ifaces.get(DEVICE_INTERFACE)
if device is None:
continue
if (device["Address"] == device_address and
path.startswith(path_prefix)):
obj = bus.get_object(SERVICE_NAME, path)
return dbus.Interface(obj, DEVICE_INTERFACE)
raise Exception("Bluetooth device not found")
def advertising_main(mainloop, bus, adapter_name):
adapter = adapters.find_adapter(bus, LE_ADVERTISING_MANAGER_IFACE, adapter_name)
print('adapter: %s' % (adapter,))
if not adapter:
raise Exception('LEAdvertisingManager1 interface not found')
adapter_props = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
"org.freedesktop.DBus.Properties")
adapter_props.Set("org.bluez.Adapter1", "Powered", dbus.Boolean(1))
ad_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
LE_ADVERTISING_MANAGER_IFACE)
test_advertisement = TestAdvertisement(bus, 0)
ad_manager.RegisterAdvertisement(test_advertisement.get_path(), {},
reply_handler=register_ad_cb,
error_handler=functools.partial(register_ad_error_cb, mainloop))
def gatt_server_main(mainloop, bus, adapter_name):
adapter = adapters.find_adapter(bus, GATT_MANAGER_IFACE, adapter_name)
if not adapter:
raise Exception('GattManager1 interface not found')
service_manager = dbus.Interface(
bus.get_object(BLUEZ_SERVICE_NAME, adapter),
GATT_MANAGER_IFACE)
app = Application(bus)
print('Registering GATT application...')
service_manager.RegisterApplication(app.get_path(), {},
reply_handler=register_app_cb,
error_handler=functools.partial(register_app_error_cb, mainloop))
def __init__(self, adapter_name):
self.listener = None
self.adapter_name = adapter_name
self._bus = dbus.SystemBus()
try:
adapter_object = self._bus.get_object('org.bluez', '/org/bluez/' + adapter_name)
except dbus.exceptions.DBusException as e:
raise _error_from_dbus_error(e)
object_manager_object = self._bus.get_object("org.bluez", "/")
self._adapter = dbus.Interface(adapter_object, 'org.bluez.Adapter1')
self._adapter_properties = dbus.Interface(self._adapter, 'org.freedesktop.DBus.Properties')
self._object_manager = dbus.Interface(object_manager_object, "org.freedesktop.DBus.ObjectManager")
self._device_path_regex = re.compile('^/org/bluez/' + adapter_name + '/dev((_[A-Z0-9]{2}){6})$')
self._devices = {}
self._discovered_devices = {}
self._interface_added_signal = None
self._properties_changed_signal = None
self._main_loop = None
self.update_devices()
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 _initialize(self):
systemd_obj = self.bus.get_object(SYSTEMD_DBUS_SERVICE,
SYSTEMD_DBUS_OPATH)
self.sysd_manager = dbus.Interface(systemd_obj,
dbus_interface=SYSTEMD_MANAGER_DBUS_IFACE)
self.sysd_manager.Subscribe()
self.bus.add_signal_receiver(self._wpa_props_changed,
dbus_interface=WPAS_INTERFACE_DBUS_IFACE,
signal_name="PropertiesChanged",
path=WPAS_INTERFACE_DBUS_OPATH)
self.bus.add_signal_receiver(self._host_props_changed,
dbus_interface=DBUS_PROPERTIES_IFACE,
signal_name="PropertiesChanged",
path=HOSTAPD_DBUS_UNIT_OPATH)
self._register_local_callbacks()
self._set_initial_state()
def _check_permission(self, sender, action):
'''
Verifies if the specified action is permitted, and raises
an AccessDeniedException if not.
The caller should use ObtainAuthorization() to get permission.
'''
try:
if sender:
kit = dbus.SystemBus().get_object('org.freedesktop.PolicyKit1', '/org/freedesktop/PolicyKit1/Authority')
kit = dbus.Interface(kit, 'org.freedesktop.PolicyKit1.Authority')
(granted, _, details) = kit.CheckAuthorization(
('system-bus-name', {'name': sender}),
action, {}, dbus.UInt32(1), '', timeout=600)
if not granted:
raise AccessDeniedException('Session not authorized by PolicyKit')
except AccessDeniedException:
raise
except dbus.DBusException, ex:
raise AccessDeniedException(ex.message)
def backend(self):
'''Return D-BUS backend client interface.
This gets initialized lazily.
'''
if self._dbus_iface is None:
try:
bus = dbus.SystemBus()
self._dbus_iface = dbus.Interface(bus.get_object(DBUS_BUS_NAME,
'/RecoveryMedia'),
DBUS_INTERFACE_NAME)
except dbus.DBusException as msg:
self.dbus_exception_handler(msg)
sys.exit(1)
except Exception as msg:
self.show_alert(Gtk.MessageType.ERROR, "Exception", str(msg),
transient_for=self.tool_widgets.get_object('tool_selector'))
return self._dbus_iface
def props_changed_cb(self, iface, changed_props, invalidated_props):
"""
Callback indicating that properties have changed.
:param iface: Interface associated with the callback.
:param changed_props: Properties changed that triggered the callback.
:param invalidated_props: Unused.
"""
if iface != constants.GATT_CHRC_IFACE:
return
if not len(changed_props):
return
value = changed_props.get('Value', None)
if not value:
return
logger.debug('Properties changed: ', value)
def __init__(self, adapter_addr, device_addr, profile_uuid):
"""
Remote GATT Profile Initialisation.
:param profile_path: dbus path to the profile.
"""
self.profile_path = dbus_tools.get_profile_path(adapter_addr,
device_addr,
profile_uuid)
self.bus = dbus.SystemBus()
self.profile_object = self.bus.get_object(
constants.BLUEZ_SERVICE_NAME,
self.profile_path)
self.profile_methods = dbus.Interface(
self.profile_object,
constants.GATT_PROFILE_IFACE)
self.profile_props = dbus.Interface(self.profile_object,
dbus.PROPERTIES_IFACE)
def __init__(self, adapter_addr):
"""
GATT Manager Initialisation.
:param manager_path: dbus path to the GATT Manager.
"""
self.manager_path = dbus_tools.get_dbus_path(adapter_addr)
self.bus = dbus.SystemBus()
self.manager_obj = self.bus.get_object(
constants.BLUEZ_SERVICE_NAME,
self.manager_path)
self.manager_methods = dbus.Interface(
self.manager_obj,
constants.GATT_MANAGER_IFACE)
self.manager_props = dbus.Interface(self.manager_obj,
dbus.PROPERTIES_IFACE)
def list_adapters():
"""Return list of adapters address available on system."""
paths = []
addresses = []
bus = dbus.SystemBus()
manager = dbus.Interface(
bus.get_object(constants.BLUEZ_SERVICE_NAME, '/'),
constants.DBUS_OM_IFACE)
manager_obj = manager.GetManagedObjects()
for path, ifaces in manager_obj.items():
if constants.ADAPTER_INTERFACE in ifaces:
paths.append(path)
addresses.append(
manager_obj[path][constants.ADAPTER_INTERFACE]['Address'])
if len(paths) < 1:
raise AdapterError('No Bluetooth adapter found')
else:
return addresses
def add_service_type(self, type):
interface, protocol, domain = (
self.interface, self.protocol, self.domain)
# Are we already browsing this domain for this type?
if self.already_browsing(type):
return
logger.info("Browsing for services of type '%s' in domain '%s' on %s.%i ..." %
(type, domain, self.siocgifname(interface), protocol))
browser = self.server.ServiceBrowserNew(
interface, protocol, type, domain, dbus.UInt32(0))
bus = dbus.Interface(self.system_bus.get_object(
avahi.DBUS_NAME, browser), avahi.DBUS_INTERFACE_SERVICE_BROWSER)
bus.connect_to_signal('ItemNew', self.service_add)
bus.connect_to_signal('ItemRemove', self.service_remove)
self.service_browsers[(interface, protocol, type, domain)] = bus
def _start(self):
try:
self.server = dbus.Interface(self.system_bus.get_object(
avahi.DBUS_NAME, avahi.DBUS_PATH_SERVER), avahi.DBUS_INTERFACE_SERVER)
except:
logger.info("Check that the Avahi daemon is running!")
return
try:
self.use_host_names = self.server.IsNSSSupportAvailable()
except:
self.use_host_names = False
self.domain = self.server.GetDomainName()
logger.info("Starting discovery")
for service in self.services:
self.add_service_type(service)
def pair(self, address):
try:
device = bluezutils.find_device(address)
except (bluezutils.BluezUtilError,
dbus.exceptions.DBusException) as error:
print_error(str(error) + "\n")
return False
try:
props = dbus.Interface(
self._bus.get_object("org.bluez", device.object_path),
"org.freedesktop.DBus.Properties")
if not props.Get("org.bluez.Device1", "Paired"):
device.Pair()
except dbus.exceptions.DBusException as error:
print_error(str(error) + "\n")
return False
return True
def disconnect(self, address):
try:
device = bluezutils.find_device(address)
except (bluezutils.BluezUtilError,
dbus.exceptions.DBusException) as error:
print_error(str(error) + "\n")
return False
try:
props = dbus.Interface(
self._bus.get_object("org.bluez", device.object_path),
"org.freedesktop.DBus.Properties")
if props.Get("org.bluez.Device1", "Connected"):
device.Disconnect()
except dbus.exceptions.DBusException as error:
print_error(str(error) + "\n")
return False
return True
def trust(self, address):
try:
device = bluezutils.find_device(address)
except (bluezutils.BluezUtilError,
dbus.exceptions.DBusException) as error:
print_error(str(error) + "\n")
return False
try:
props = dbus.Interface(
self._bus.get_object("org.bluez", device.object_path),
"org.freedesktop.DBus.Properties")
if not props.Get("org.bluez.Device1", "Trusted"):
props.Set("org.bluez.Device1", "Trusted", dbus.Boolean(1))
except dbus.exceptions.DBusException as error:
print_error(str(error) + "\n")
return False
return True
def set_adapter_property(self, prop, value):
try:
adapter = bluezutils.find_adapter()
except (bluezutils.BluezUtilError,
dbus.exceptions.DBusException) as error:
print_error(str(error) + "\n")
return False
try:
props = dbus.Interface(
self._bus.get_object("org.bluez", adapter.object_path),
"org.freedesktop.DBus.Properties")
if props.Get("org.bluez.Adapter1", prop) != value:
props.Set("org.bluez.Adapter1", prop, value)
except dbus.exceptions.DBusException as error:
print_error(str(error) + "\n")
return False
return True
def get_adapter_property(self, prop):
try:
adapter = bluezutils.find_adapter()
except (bluezutils.BluezUtilError,
dbus.exceptions.DBusException) as error:
print_error(str(error) + "\n")
return None
try:
props = dbus.Interface(
self._bus.get_object("org.bluez", adapter.object_path),
"org.freedesktop.DBus.Properties")
return props.Get("org.bluez.Adapter1", prop)
except dbus.exceptions.DBusException as error:
print_error(str(error) + "\n")
return None
def set_device_property(self, address, prop, value):
try:
device = bluezutils.find_device(address)
except (bluezutils.BluezUtilError,
dbus.exceptions.DBusException) as error:
print_error(str(error) + "\n")
return False
try:
props = dbus.Interface(
self._bus.get_object("org.bluez", device.object_path),
"org.freedesktop.DBus.Properties")
if props.Get("org.bluez.Device1", prop) != value:
props.Set("org.bluez.Device1", prop, value)
except dbus.exceptions.DBusException as error:
print_error(str(error) + "\n")
return False
return True
def find_device_in_objects(objects, device_address, adapter_pattern=None):
bus = dbus.SystemBus()
path_prefix = ""
if adapter_pattern:
adapter = find_adapter_in_objects(objects, adapter_pattern)
path_prefix = adapter.object_path
for path, ifaces in objects.items():
device = ifaces.get(DEVICE_INTERFACE)
if device is None:
continue
if (device["Address"] == device_address and path.startswith(
path_prefix)):
obj = bus.get_object(SERVICE_NAME, path)
return dbus.Interface(obj, DEVICE_INTERFACE)
raise BluezUtilError("Bluetooth device not found")
def activate_menu_item(self, selection):
if not selection:
return
action = self.menu_actions.get(selection, 'sys.quit')
if 'sys.' in action:
self.window.close()
elif 'app.' in action:
app_object = self.session.get_object(self.window.bus_name, self.window.app_path)
app_iface = dbus.Interface(app_object, dbus_interface='org.gtk.Actions')
app_iface.Activate(action.replace('app.', ''), [], dict())
elif 'win.' in action:
win_object = self.session.get_object(self.window.bus_name, self.window.win_path)
win_iface = dbus.Interface(win_object, dbus_interface='org.gtk.Actions')
win_iface.Activate(action.replace('win.', ''), [], dict())
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)
