cellular/pseudomodem/dbus_std_ifaces.py

# Lint as: python2, python3
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""
Python implementation of the standard interfaces:
  - org.freedesktop.DBus.Properties
  - org.freedesktop.DBus.Introspectable (TODO(armansito): May not be necessary)
  - org.freedesktop.DBus.ObjectManager

"""

import dbus
import dbus.service
import dbus.types
import logging

from autotest_lib.client.cros.cellular import mm1_constants
from autotest_lib.client.cros.cellular.pseudomodem import pm_errors
from autotest_lib.client.cros.cellular.pseudomodem import utils

class MMPropertyError(pm_errors.MMError):
    """
    MMPropertyError is raised by DBusProperties methods
    to indicate that a value for the given interface or
    property could not be found.

    """

    UNKNOWN_PROPERTY = 0
    UNKNOWN_INTERFACE = 1

    def __init__(self, errno, *args, **kwargs):
        super(MMPropertyError, self).__init__(errno, args, kwargs)


    def _Setup(self):
        self._error_name_base = mm1_constants.I_MODEM_MANAGER
        self._error_name_map = {
            self.UNKNOWN_PROPERTY : '.UnknownProperty',
            self.UNKNOWN_INTERFACE : '.UnknownInterface'
        }


class DBusProperties(dbus.service.Object):
    """
    == org.freedesktop.DBus.Properties ==

    This serves as the abstract base class for all objects that expose
    properties. Each instance holds a mapping from DBus interface names to
    property-value mappings, which are provided by the subclasses.

    """

    def __init__(self, path, bus=None, config=None):
        """
        @param bus: The pydbus bus object.
        @param path: The DBus object path of this object.
        @param config: This is an optional dictionary that can be used to
                initialize the property dictionary with values other than the
                ones provided by |_InitializeProperties|. The dictionary has to
                contain a mapping from DBus interfaces to property-value pairs,
                and all contained keys must have been initialized during
                |_InitializeProperties|, i.e. if config contains any keys that
                have not been already set in the internal property dictionary,
                an error will be raised (see DBusProperties.Set).

        """
        if not path:
            raise TypeError(('A value for "path" has to be provided that is '
                'not "None".'))
        if bus:
          dbus.service.Object.__init__(self, bus, path)
        else:
          dbus.service.Object.__init__(self, None, None)
        self.path = path
        self.bus = bus
        self._properties = self._InitializeProperties()

        if config:
            for key, props in config:
                for prop, val in props:
                    self.Set(key, prop, val)


    @property
    def properties(self):
        """
        @returns: The property dictionary.

        """
        return self._properties


    def SetBus(self, bus):
        """
        Sets the pydbus bus object that this instance of DBusProperties should
        be exposed on. Call this method only if |bus| is not already set.

        @param bus: The pydbus bus object to assign.

        """
        self.bus = bus
        self.add_to_connection(bus, self.path)


    def SetPath(self, path):
        """
        Exposes this object on a new DBus path. This method fails with an
        Exception by default, since exposing an object on multiple paths is
        disallowed by default.

        Subclasses can change this behavior by setting the
        SUPPORTS_MULTIPLE_OBJECT_PATHS class variable to True.

        @param path: The new path to assign to this object.

        """
        self.path = path
        self.add_to_connection(self.bus, path)


    def SetUInt32(self, interface_name, property_name, value):
        """
        Sets the given uint32 value matching the given property and interface.
        Wraps the given value inside a dbus.types.UInt32.

        @param interface_name: The DBus interface name.
        @param property_name: The property name.
        @param value: Value to set.
        @raises: MMPropertyError, if the given |interface_name| or
                |property_name| is not exposed by this object.
        Emits:
            PropertiesChanged

        """
        self.Set(interface_name, property_name, dbus.types.UInt32(value))


    def SetInt32(self, interface_name, property_name, value):
        """
        Sets the given int32 value matching the given property and interface.
        Wraps the given value inside a dbus.types.Int32.

        @param interface_name: The DBus interface name.
        @param property_name: The property name.
        @param value: Value to set.
        @raises: MMPropertyError, if the given |interface_name| or
                |property_name| is not exposed by this object.
        Emits:
            PropertiesChanged

        """
        self.Set(interface_name, property_name, dbus.types.Int32(value))


    @utils.log_dbus_method()
    @dbus.service.method(mm1_constants.I_PROPERTIES, in_signature='ss',
                         out_signature='v')
    def Get(self, interface_name, property_name):
        """
        Returns the value matching the given property and interface.

        @param interface_name: The DBus interface name.
        @param property_name: The property name.
        @returns: The value matching the given property and interface.
        @raises: MMPropertyError, if the given |interface_name| or
                |property_name| is not exposed by this object.

        """
        logging.info(
            '%s: Get(%s, %s)',
            self.path,
            interface_name,
            property_name)
        val = self.GetAll(interface_name).get(property_name, None)
        if val is None:
            message = ("Property '%s' not implemented for interface '%s'." %
                (property_name, interface_name))
            logging.info(message)
            raise MMPropertyError(
                MMPropertyError.UNKNOWN_PROPERTY, message)
        return val


    @utils.log_dbus_method()
    @dbus.service.method(mm1_constants.I_PROPERTIES, in_signature='ssv')
    def Set(self, interface_name, property_name, value):
        """
        Sets the value matching the given property and interface.

        @param interface_name: The DBus interface name.
        @param property_name: The property name.
        @param value: The value to set.
        @raises: MMPropertyError, if the given |interface_name| or
                |property_name| is not exposed by this object.
        Emits:
            PropertiesChanged

        """
        logging.info(
            '%s: Set(%s, %s)',
            self.path,
            interface_name,
            property_name)
        props = self.GetAll(interface_name)
        if property_name not in props:
            raise MMPropertyError(
                MMPropertyError.UNKNOWN_PROPERTY,
                ("Property '%s' not implemented for "
                "interface '%s'.") %
                (property_name, interface_name))
        if props[property_name] == value:
            logging.info("Property '%s' already has value '%s'. Ignoring.",
                         property_name,
                         value)
            return
        props[property_name] = value
        changed = { property_name : value }
        inv = self._InvalidatedPropertiesForChangedValues(changed)
        self.PropertiesChanged(interface_name, changed, inv)


    @utils.log_dbus_method()
    @dbus.service.method(mm1_constants.I_PROPERTIES,
                         in_signature='s', out_signature='a{sv}')
    def GetAll(self, interface_name):
        """
        Returns all property-value pairs that match the given interface.

        @param interface_name: The DBus interface name.
        @returns: A dictionary, containing the properties of the given DBus
                interface and their values.
        @raises: MMPropertyError, if the given |interface_name| or
                |property_name| is not exposed by this object.

        """
        logging.info(
            '%s: GetAll(%s)',
            self.path,
            interface_name)
        props = self._properties.get(interface_name, None)
        if props is None:
            raise MMPropertyError(
                MMPropertyError.UNKNOWN_INTERFACE,
                "Object does not implement interface '%s'." %
                interface_name)
        return props


    @dbus.service.signal(mm1_constants.I_PROPERTIES, signature='sa{sv}as')
    def PropertiesChanged(
            self,
            interface_name,
            changed_properties,
            invalidated_properties):
        """
        This signal is emitted by Set, when the value of a property is changed.

        @param interface_name: The interface the changed properties belong to.
        @param changed_properties: Dictionary containing the changed properties
                                   and their new values.
        @param invalidated_properties: List of properties that were invalidated
                                       when properties changed.

        """
        logging.info(('Properties Changed on interface: %s Changed Properties:'
            ' %s InvalidatedProperties: %s.', interface_name,
            str(changed_properties), str(invalidated_properties)))


    def SetAll(self, interface, properties):
        """
        Sets the entire property dictionary for the given interface.

        @param interface: String specifying the DBus interface.
        @param properties: Dictionary containing the properties to set.
        Emits:
            PropertiesChanged

        """
        old_props = self._properties.get(interface, None)
        if old_props:
            invalidated = list(old_props.keys())
        else:
            invalidated = []
        self._properties[interface] = properties
        self.PropertiesChanged(interface, properties, invalidated)


    def GetInterfacesAndProperties(self):
        """
        Returns all DBus properties of this object.

        @returns: The complete property dictionary. The returned dict is a tree,
                where the keys are DBus interfaces and the values are
                dictionaries that map properties to values.
        """
        return self._properties


    def _InvalidatedPropertiesForChangedValues(self, changed):
        """
        Called by Set, returns the list of property names that should become
        invalidated given the properties and their new values contained in
        changed. Subclasses can override this method; the default implementation
        returns an empty list.

        """
        return []


    def _InitializeProperties(self):
        """
        Called at instantiation. Subclasses have to override this method and
        return a dictionary containing mappings from implemented interfaces to
        dictionaries of property-value mappings.

        """
        raise NotImplementedError()


class DBusObjectManager(dbus.service.Object):
    """
    == org.freedesktop.DBus.ObjectManager ==

    This interface, included in rev. 0.17 of the DBus specification, allows a
    generic way to control the addition and removal of Modem objects, as well
    as the addition and removal of interfaces in the given objects.

    """

    def __init__(self, bus, path):
        dbus.service.Object.__init__(self, bus, path)
        self.devices = []
        self.bus = bus
        self.path = path


    def Add(self, device):
        """
        Adds a device to the list of devices that are managed by this modem
        manager.

        @param device: Device to add.
        Emits:
            InterfacesAdded

        """
        self.devices.append(device)
        device.manager = self
        self.InterfacesAdded(device.path, device.GetInterfacesAndProperties())


    def Remove(self, device):
        """
        Removes a device from the list of devices that are managed by this
        modem manager.

        @param device: Device to remove.
        Emits:
            InterfacesRemoved

        """
        if device in self.devices:
            self.devices.remove(device)
        interfaces = list(device.GetInterfacesAndProperties().keys())
        self.InterfacesRemoved(device.path, interfaces)
        device.remove_from_connection()


    @utils.log_dbus_method()
    @dbus.service.method(mm1_constants.I_OBJECT_MANAGER,
                         out_signature='a{oa{sa{sv}}}')
    def GetManagedObjects(self):
        """
        @returns: A dictionary containing all objects and their properties. The
                keys to the dictionary are object paths which are mapped to
                dictionaries containing mappings from DBus interface names to
                property-value pairs.

        """
        results = {}
        for device in self.devices:
            results[dbus.types.ObjectPath(device.path)] = (
                    device.GetInterfacesAndProperties())
        logging.info('%s: GetManagedObjects: %s', self.path,
                     ', '.join(list(results.keys())))
        return results


    @dbus.service.signal(mm1_constants.I_OBJECT_MANAGER,
                         signature='oa{sa{sv}}')
    def InterfacesAdded(self, object_path, interfaces_and_properties):
        """
        The InterfacesAdded signal is emitted when either a new object is added
        or when an existing object gains one or more interfaces.

        @param object_path: Path of the added object.
        @param interfaces_and_properties: The complete property dictionary that
                belongs to the recently added object.

        """
        logging.info((self.path + ': InterfacesAdded(' + object_path +
                     ', ' + str(interfaces_and_properties)) + ')')


    @dbus.service.signal(mm1_constants.I_OBJECT_MANAGER, signature='oas')
    def InterfacesRemoved(self, object_path, interfaces):
        """
        The InterfacesRemoved signal is emitted whenever an object is removed
        or it loses one or more interfaces.

        @param object_path: Path of the remove object.
        @param interfaces_and_properties: The complete property dictionary that
                belongs to the recently removed object.

        """
        logging.info((self.path + ': InterfacesRemoved(' + object_path +
                     ', ' + str(interfaces) + ')'))