blob: 3eea6a577b7a2fe66075cc630d1b14d9a6f0e408 [file] [log] [blame]
# Contributors Listed Below - COPYRIGHT 2016
# [+] International Business Machines Corp.
#
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied. See the License for the specific language governing
# permissions and limitations under the License.
import dbus
import dbus.service
import dbus.exceptions
OBJ_PREFIX = '/xyz/openbmc_project'
def is_unique(connection):
return connection[0] == ':'
def get_dbus():
return dbus.SystemBus()
class DbusProperties(dbus.service.Object):
def __init__(self, **kw):
self.validator = kw.pop('validator', None)
super(DbusProperties, self).__init__(**kw)
self.properties = {}
self._export = False
def unmask_signals(self):
self._export = True
inst = super(DbusProperties, self)
if hasattr(inst, 'unmask_signals'):
inst.unmask_signals()
def mask_signals(self):
self._export = False
inst = super(DbusProperties, self)
if hasattr(inst, 'mask_signals'):
inst.mask_signals()
@dbus.service.method(
dbus.PROPERTIES_IFACE,
in_signature='ss', out_signature='v')
def Get(self, interface_name, property_name):
d = self.GetAll(interface_name)
try:
v = d[property_name]
return v
except:
raise dbus.exceptions.DBusException(
"Unknown property: '{}'".format(property_name),
name="org.freedesktop.DBus.Error.UnknownProperty")
@dbus.service.method(
dbus.PROPERTIES_IFACE,
in_signature='s', out_signature='a{sv}')
def GetAll(self, interface_name):
try:
d = self.properties[interface_name]
return d
except:
raise dbus.exceptions.DBusException(
"Unknown interface: '{}'".format(interface_name),
name="org.freedesktop.DBus.Error.UnknownInterface")
@dbus.service.method(
dbus.PROPERTIES_IFACE,
in_signature='ssv')
def Set(self, interface_name, property_name, new_value):
if (interface_name not in self.properties):
self.properties[interface_name] = {}
if self.validator:
self.validator(interface_name, property_name, new_value)
try:
old_value = self.properties[interface_name][property_name]
if (old_value != new_value):
self.properties[interface_name][property_name] = new_value
if self._export:
self.PropertiesChanged(
interface_name, {property_name: new_value}, [])
except:
self.properties[interface_name][property_name] = new_value
if self._export:
self.PropertiesChanged(
interface_name, {property_name: new_value}, [])
@dbus.service.method(
"org.openbmc.Object.Properties", in_signature='sa{sv}')
def SetMultiple(self, interface_name, prop_dict):
if (interface_name not in self.properties):
self.properties[interface_name] = {}
value_changed = False
for property_name in prop_dict:
new_value = prop_dict[property_name]
try:
old_value = self.properties[interface_name][property_name]
if (old_value != new_value):
self.properties[interface_name][property_name] = new_value
value_changed = True
except:
self.properties[interface_name][property_name] = new_value
value_changed = True
if (value_changed is True and self._export):
self.PropertiesChanged(interface_name, prop_dict, [])
@dbus.service.signal(
dbus.PROPERTIES_IFACE, signature='sa{sv}as')
def PropertiesChanged(
self, interface_name, changed_properties, invalidated_properties):
pass
def add_interfaces_to_class(cls, ifaces):
"""
The built-in Introspect method in dbus-python doesn't find
interfaces if the @method or @signal decorators aren't used
(property-only interfaces). Use this method on a class
derived from dbus.service.Object to help the dbus-python provided
Introspect method find these interfaces.
Arguments:
cls -- The dbus.service.Object superclass to add interfaces to.
ifaces -- The property-only interfaces to add to the class.
"""
for iface in ifaces:
class_table_key = '{}.{}'.format(cls.__module__, cls.__name__)
cls._dbus_class_table[class_table_key].setdefault(iface, {})
def add_interfaces(ifaces):
"""
A class decorator for add_interfaces_to_class.
"""
def decorator(cls):
undecorated = cls.__init__
def ctor(obj, *a, **kw):
undecorated(obj, *a, **kw)
add_interfaces_to_class(cls, ifaces)
cls.__init__ = ctor
return cls
return decorator
class DbusObjectManager(dbus.service.Object):
def __init__(self, **kw):
super(DbusObjectManager, self).__init__(**kw)
self.objects = {}
self._export = False
def unmask_signals(self):
self._export = True
inst = super(DbusObjectManager, self)
if hasattr(inst, 'unmask_signals'):
inst.unmask_signals()
def mask_signals(self):
self._export = False
inst = super(DbusObjectManager, self)
if hasattr(inst, 'mask_signals'):
inst.mask_signals()
def add(self, object_path, obj):
self.objects[object_path] = obj
if self._export:
self.InterfacesAdded(object_path, obj.properties)
def remove(self, object_path):
obj = self.objects.pop(object_path, None)
obj.remove_from_connection()
if self._export:
self.InterfacesRemoved(object_path, obj.properties.keys())
def get(self, object_path, default=None):
return self.objects.get(object_path, default)
@dbus.service.method(
"org.freedesktop.DBus.ObjectManager",
in_signature='', out_signature='a{oa{sa{sv}}}')
def GetManagedObjects(self):
data = {}
for objpath in self.objects.keys():
data[objpath] = self.objects[objpath].properties
return data
@dbus.service.signal(
"org.freedesktop.DBus.ObjectManager", signature='oa{sa{sv}}')
def InterfacesAdded(self, object_path, properties):
pass
@dbus.service.signal(
"org.freedesktop.DBus.ObjectManager", signature='oas')
def InterfacesRemoved(self, object_path, interfaces):
pass