# Copyright (c) 2021 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.

"""
This module provides bindings for ModemManager1.

"""

import dbus
import dbus.mainloop.glib
import logging
import time

from autotest_lib.client.bin import utils
from autotest_lib.client.cros.cellular import mm1_constants


def _is_unknown_dbus_binding_exception(e):
    return (isinstance(e, dbus.exceptions.DBusException) and
            e.get_dbus_name() in [mm1_constants.DBUS_SERVICE_UNKNOWN,
                                  mm1_constants.DBUS_UNKNOWN_METHOD,
                                  mm1_constants.DBUS_UNKNOWN_OBJECT,
                                  mm1_constants.DBUS_UNKNOWN_INTERFACE])


class ModemManager1ProxyError(Exception):
    """Exceptions raised by ModemManager1ProxyError and it's children."""
    pass


class ModemManager1Proxy(object):
    """A wrapper around a DBus proxy for ModemManager1."""

    # Amount of time to wait between attempts to connect to ModemManager1.
    CONNECT_WAIT_INTERVAL_SECONDS = 0.2

    @classmethod
    def get_proxy(cls, bus=None, timeout_seconds=10):
        """Connect to ModemManager1 over DBus, retrying if necessary.

        After connecting to ModemManager1, this method will verify that
        ModemManager1 is answering RPCs.

        @param bus: D-Bus bus to use, or specify None and this object will
            create a mainloop and bus.
        @param timeout_seconds: float number of seconds to try connecting
            A value <= 0 will cause the method to return immediately,
            without trying to connect.
        @return a ModemManager1Proxy instance if we connected, or None
            otherwise.
        @raise ModemManager1ProxyError if it fails to connect to
            ModemManager1.

        """
        def _connect_to_mm1(bus):
            try:
                # We create instance of class on which this classmethod was
                # called. This way, calling
                # SubclassOfModemManager1Proxy.get_proxy() will get a proxy of
                # the right type.
                return cls(bus=bus)
            except dbus.exceptions.DBusException as e:
                if _is_unknown_dbus_binding_exception(e):
                    return None
                raise ModemManager1ProxyError(
                    'Error connecting to ModemManager1. DBus error: |%s|',
                    repr(e))

        utils.poll_for_condition(
            lambda: _connect_to_mm1(bus) is not None,
            exception=ModemManager1ProxyError(
                'Timed out connecting to ModemManager1'),
            timeout=timeout_seconds,
            sleep_interval=ModemManager1Proxy.CONNECT_WAIT_INTERVAL_SECONDS)
        connection = _connect_to_mm1(bus)

        # Check to make sure ModemManager1 is responding to DBus requests by
        # setting the logging to debug.
        connection.manager.SetLogging('DEBUG', timeout=timeout_seconds)

        return connection

    def __init__(self, bus=None):
        if bus is None:
            dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
            bus = dbus.SystemBus()
        self._bus = bus
        self._manager = dbus.Interface(
            self._bus.get_object(mm1_constants.I_MODEM_MANAGER,
                                 mm1_constants.MM1),
            mm1_constants.I_MODEM_MANAGER)
        self._device = None

    @property
    def manager(self):
        """@return the DBus ModemManager1 Manager object."""
        return self._manager

    def inhibit_device(self, inhibit):
        """

        Uses Modem Manager InhibitDevice DBus API to inhibit/uninhibit
        @param inhibit: true to inhibit the modem and false to uninhibit it.

        InhibitDevice API:
        @uid: the unique ID of the physical device, given in the
              #org.freedesktop.ModemManager1.Modem:Device property.
        @inhibit: %TRUE to inhibit the modem and %FALSE to uninhibit it.

        Inhibit or uninhibit the device.

        When the modem is inhibited ModemManager will close all its ports and
        unexport it from the bus, so that users of the interface are no longer
        able to operate with it.

        This operation binds the inhibition request to the existence of the
        caller in the DBus bus. If the caller disappears from the bus, the
        inhibition will automatically removed.
        """
        try:
            if not self._manager:
                raise ModemManager1ProxyError(
                    'Failed to obtain dbus manager object.- No manager')
            if not inhibit and not self._device:
                raise ModemManager1ProxyError(
                    'Uninhibit called before inhibit %s', self._device)

            if inhibit:
                modem = self.get_modem()
                if not modem:
                    raise ModemManager1ProxyError(
                        'Failed to to obtain dbus manager object. - No modem')

                self._device = modem.properties(
                mm1_constants.I_MODEM).get('Device')

            logging.debug('device to be inhibited/uninhibited %s', self._device)
            self._manager.InhibitDevice(dbus.String(self._device), inhibit)

            logging.debug('inhibit=%r done with %s', inhibit, self._device)

            if inhibit:
                time.sleep(mm1_constants.MM_INHIBIT_PROCESSING_TIME)
            else:
                result = self.wait_for_modem(
                    mm1_constants.MM_UNINHIBIT_PROCESSING_TIME)

                time.sleep(mm1_constants.MM_REPROBE_PROCESSING_TIME)
                if result is None:
                    raise ModemManager1ProxyError('No modem after uninhibit')
        except dbus.exceptions.DBusException as e:
            raise ModemManager1ProxyError(
                'Failed to to obtain dbus object for the modem.'
                'DBus error: |%s|', repr(e))

    def get_modem(self):
        """
        Return the one and only modem object.

        This method distinguishes between no modem and more than one modem.
        In the former, this could happen if the modem has not yet surfaced and
        is not really considered an error. The caller can wait for the modem
        by repeatedly calling this method. In the latter, it is a clear error
        condition and an exception will be raised.

        Every call to |get_modem| obtains a fresh DBus proxy for the modem. So,
        if the modem DBus object has changed between two calls to this method,
        the proxy returned will be for the currently exported modem.

        @return a ModemProxy object.  Return None if no modem is found.
        @raise ModemManager1ProxyError unless exactly one modem is found.

        """
        try:
            object_manager = dbus.Interface(
                self._bus.get_object(mm1_constants.I_MODEM_MANAGER,
                                     mm1_constants.MM1),
                mm1_constants.I_OBJECT_MANAGER)
            modems = object_manager.GetManagedObjects()
        except dbus.exceptions.DBusException as e:
            raise ModemManager1ProxyError(
                'Failed to list the available modems. DBus error: |%s|',
                repr(e))

        if not modems:
            return None
        elif len(modems) > 1:
            raise ModemManager1ProxyError(
                'Expected one modem object, found %d', len(modems))

        modem_proxy = ModemProxy(self._bus, modems.keys()[0])
        # Check that this object is valid
        try:
            modem_proxy.modem.GetAll(mm1_constants.I_MODEM,
                                     dbus_interface=mm1_constants.I_PROPERTIES)
            return modem_proxy
        except dbus.exceptions.DBusException as e:
            if _is_unknown_dbus_binding_exception(e):
                return None
            raise ModemManager1ProxyError(
                'Failed to obtain dbus object for the modem. DBus error: '
                '|%s|', repr(e))

    def wait_for_modem(self, timeout_seconds):
        """
        Wait for the modem to appear.

        @param timeout_seconds: Number of seconds to wait for modem to appear.
        @return a ModemProxy object.
        @raise ModemManager1ProxyError if no modem is found within the timeout
                or if more than one modem is found. NOTE: This method does not
                wait for a second modem. The exception is raised if there is
                more than one modem at the time of polling.

        """
        return utils.poll_for_condition(
            self.get_modem,
            exception=ModemManager1ProxyError('No modem found'),
            timeout=timeout_seconds)


class ModemProxy(object):
    """A wrapper around a DBus proxy for ModemManager1 modem object."""

    # Amount of time to wait for a state transition.
    STATE_TRANSITION_WAIT_SECONDS = 10

    def __init__(self, bus, path):
        self._bus = bus
        self._modem = self._bus.get_object(mm1_constants.I_MODEM_MANAGER, path)

    @property
    def modem(self):
        """@return the DBus modem object."""
        return self._modem

    @property
    def iface_modem(self):
        """@return org.freedesktop.ModemManager1.Modem DBus interface."""
        return dbus.Interface(self._modem, mm1_constants.I_MODEM)

    @property
    def iface_simple_modem(self):
        """@return org.freedesktop.ModemManager1.Simple DBus interface."""
        return dbus.Interface(self._modem, mm1_constants.I_MODEM_SIMPLE)

    @property
    def iface_gsm_modem(self):
        """@return org.freedesktop.ModemManager1.Modem3gpp DBus interface."""
        return dbus.Interface(self._modem, mm1_constants.I_MODEM_3GPP)

    @property
    def iface_cdma_modem(self):
        """@return org.freedesktop.ModemManager1.ModemCdma DBus interface."""
        return dbus.Interface(self._modem, mm1_constants.I_MODEM_CDMA)

    @property
    def iface_properties(self):
        """@return org.freedesktop.DBus.Properties DBus interface."""
        return dbus.Interface(self._modem, dbus.PROPERTIES_IFACE)

    def properties(self, iface):
        """Return the properties associated with the specified interface.

        @param iface: Name of interface to retrieve the properties from.
        @return array of properties.

        """
        return self.iface_properties.GetAll(iface)

    def get_sim(self):
        """
        Return the SIM proxy object associated with this modem.

        @return SimProxy object or None if no SIM exists.

        """
        sim_path = self.properties(mm1_constants.I_MODEM).get('Sim')
        if not sim_path:
            return None
        sim_proxy = SimProxy(self._bus, sim_path)
        # Check that this object is valid
        try:
            sim_proxy.properties(mm1_constants.I_SIM)
            return sim_proxy
        except dbus.exceptions.DBusException as e:
            if _is_unknown_dbus_binding_exception(e):
                return None
            raise ModemManager1ProxyError(
                'Failed to obtain dbus object for the SIM. DBus error: '
                '|%s|', repr(e))

    def get_sim_slots(self):
        """
        The list of SIM slots available in the system, including the SIM object
        paths if the cards are present. If a given SIM slot at a given index
        doesn't have a SIM card available, an empty object path will be given.

        The length of this array of objects will be equal to the amount of
        available SIM slots in the system, and the index in the array is the
        slot index.

        This list includes the SIM object considered as primary active SIM slot
        (#org.freedesktop.ModemManager1.Modem.Sim) at index
        #org.freedesktop.ModemManager1.Modem.ActiveSimSlot.

        @return list of SimSlot paths

        """
        return self.properties(mm1_constants.I_MODEM).get('SimSlots')

    def get_primary_sim_slot(self):
        """
        The index of the primary active SIM slot in the
        #org.freedesktop.ModemManager1.Modem.SimSlots array, given in the [1,N]
        range.

        If multiple SIM slots aren't supported, this property will report None

        In a Multi SIM Single Standby setup, this index identifies the only SIM
        that is currently active. All the remaining slots will be inactive.

        In a Multi SIM Multi Standby setup, this index identifies the active SIM
        that is considered primary, i.e. the one that will be used when a data
        connection is setup.

        @return current primary slot index

        """
        return self.properties(mm1_constants.I_MODEM).get('PrimarySimSlot')

    def set_primary_slot(self, sim_slot):
        """
        Selects which SIM slot to be considered as primary, on devices that
        expose multiple slots in the #org.freedesktop.ModemManager1.Modem
        :SimSlots property.

        When the switch happens the modem may require a full device reprobe,
        so the modem object in DBus will get removed, and recreated once the
        selected SIM slot is in use.

        There is no limitation on which SIM slot to select, so the user may
        also set as primary a slot that doesn't currently have any valid SIM
        card inserted.

        @param: sim_slot: SIM slot number to set as primary.
        @return: success or raise error

        """
        self.iface_modem.SetPrimarySimSlot(dbus.UInt32(sim_slot))

    def wait_for_states(self, states,
                        timeout_seconds=STATE_TRANSITION_WAIT_SECONDS):
        """
        Wait for the modem to transition to a state in |states|.

        This method does not support transitory states (eg. enabling,
        disabling, connecting, disconnecting, etc).

        @param states: List of states the modem can transition to.
        @param timeout_seconds: Max number of seconds to wait.
        @raise ModemManager1ProxyError if the modem does not transition to
            one of the accepted states.

        """
        for state in states:
            if state in [mm1_constants.MM_MODEM_STATE_INITIALIZING,
                         mm1_constants.MM_MODEM_STATE_DISABLING,
                         mm1_constants.MM_MODEM_STATE_ENABLING,
                         mm1_constants.MM_MODEM_STATE_SEARCHING,
                         mm1_constants.MM_MODEM_STATE_DISCONNECTING,
                         mm1_constants.MM_MODEM_STATE_CONNECTING]:
                raise ModemManager1ProxyError(
                    'wait_for_states() does not support transitory states.')

        utils.poll_for_condition(
            lambda: self.properties(mm1_constants.I_MODEM)[
                mm1_constants.MM_MODEM_PROPERTY_NAME_STATE] in states,
            exception=ModemManager1ProxyError(
                'Timed out waiting for modem to enter one of these '
                'states: %s, current state=%s',
                states,
                self.properties(mm1_constants.I_MODEM)[
                    mm1_constants.MM_MODEM_PROPERTY_NAME_STATE]),
            timeout=timeout_seconds)

class SimProxy(object):
    """A wrapper around a DBus proxy for ModemManager1 SIM object."""

    def __init__(self, bus, path):
        self._bus = bus
        self._sim = self._bus.get_object(mm1_constants.I_MODEM_MANAGER, path)

    @property
    def sim(self):
        """@return the DBus SIM object."""
        return self._sim

    @property
    def iface_properties(self):
        """@return org.freedesktop.DBus.Properties DBus interface."""
        return dbus.Interface(self._sim, dbus.PROPERTIES_IFACE)

    @property
    def iface_sim(self):
        """@return org.freedesktop.ModemManager1.Sim DBus interface."""
        return dbus.Interface(self._sim, mm1_constants.I_SIM)

    def properties(self, iface=mm1_constants.I_SIM):
        """Return the properties associated with the specified interface.

        @param iface: Name of interface to retrieve the properties from.
        @return array of properties.

        """
        return self.iface_properties.GetAll(iface)