xref: /external/autotest/server/cros/network/rf_switch/rf_switch.py

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

"""Interface to control RF Switch.

Helper module to control RF Switch. Common commands to control the relays
are made available as methods that can be called.

Please refer go/rf-switch for more info on the switch.
"""
import contextlib
import logging

from autotest_lib.server.cros.network.rf_switch import scpi


class RfSwitchException(Exception):
    """Exception for RfSwitch Errors."""


class RfSwitch(scpi.Scpi):
    """RF Switch Controller."""

    _ALL_AP_RELAYS = 'k1_1:k1_24'  # Each AP uses 6 relays
    _ALL_CLIENT_RELAYS = 'k1_25:k1_48'  # Each client uses 6 relays
    _MIN_ENCLOSURE = 1  # Starting enclosure number
    _MAX_ENCLOSURE = 4  # Last enclosure number
    _ALL_ENCLOSURE = 0  # All the enclosures
    _RELAYS_PER_ENCLOSURE = 6  # Number of relays in an enclosure

    # Each AP enclosure uses 6 relays and a R relay to set attenuation
    _AP_ATTENUATOR_RELAYS = {
        1: ['k1_49', 'k1_50', 'k1_51', 'k1_52', 'k1_53', 'k1_54', 'R1_9'],
        2: ['k1_55', 'k1_56', 'k1_57', 'k1_58', 'k1_59', 'k1_60', 'R1_10'],
        3: ['k1_61', 'k1_62', 'k1_63', 'k1_64', 'k1_65', 'k1_66', 'R1_11'],
        4: ['k1_67', 'k1_68', 'k1_69', 'k1_70', 'k1_71', 'k1_72', 'R1_12'],
    }
    # Shorter version
    _AP_ATTENUATOR_RELAYS_SHORT = {
        1: 'k1_49:k1_54,R1_9',
        2: 'k1_55:k1_60,R1_10',
        3: 'k1_61:k1_66,R1_11',
        4: 'k1_67:k1_72,R1_12',
    }

    _CMD_CLOSE_RELAYS = 'ROUT:CLOS'
    _CMD_CHECK_RELAYS_CLOSED = 'ROUT:CLOS?'
    _CMD_OPEN_RELAYS = 'ROUT:OPEN'
    _CMD_OPEN_ALL_RELAYS = 'ROUT:OPEN:ALL'
    _CMD_SET_VERIFY = 'ROUT:CHAN:VER'
    _CMD_GET_VERIFY = 'ROUT:CHAN:VER?'
    _CMD_SET_VERIFY_INVERTED = 'ROUT:CHAN:VER:POL'
    _CMD_GET_VERIFY_INVERTED = 'ROUT:CHAN:VER:POL?'
    _CMD_GET_VERIFY_STATE = 'ROUT:CHAN:VER:POS:STAT?'
    _CMD_CHECK_BUSY = 'ROUT:MOD:BUSY?'
    _CMD_WAIT = 'ROUT:MOD:WAIT'

    def __init__(self, host, port=scpi.Scpi.SCPI_PORT):
        """Controller for RF Switch.

        @param host: Hostname or IP address of RF Switch
        @param port: Int SCPI port number (default 5025)

        """
        scpi.Scpi.__init__(self, host, port)

    def send_cmd_check_error(self, cmd):
        """Send command to switch and check for any error.

        @param cmd: string cmd to send to switch

        """
        self.write(cmd)
        code, error = self.error_query()
        if code:
            raise RfSwitchException('Error on command: "%s" - code: %s,'
                                    ' message: %s', cmd, code, error)
        self.wait()

    def close_relays(self, relays):
        """Close relays.

        @param relays: relays to close (, to separate and : for range)

        """
        self.send_cmd_check_error('%s (@%s)\n'
                                  % (self._CMD_CLOSE_RELAYS, relays))

    def relays_closed(self, relays):
        """Get open/closed status of relays.

        @param relays: relays to check (, to separate and : for range)

        @returns tuple of relay status, status is true if a relay is closed.

        """

        status = self.query('%s (@%s)\n' % (self._CMD_CHECK_RELAYS_CLOSED,
                                            relays)).strip().split(',')
        return tuple(bool(int(x)) for x in status)

    def open_relays(self, relays):
        """Open relays.

        @param relays: string relays to close (, to separate and : for range)

        """
        self.send_cmd_check_error('%s (@%s)\n' % (
                self._CMD_OPEN_RELAYS, relays))

    def open_all_relays(self):
        """Open all relays.

        This will open all relays including the attenuator, which will set it
        to Max. Please remember to set your attenuation to right level after
        this call.

        """
        self.send_cmd_check_error('%s\n' % self._CMD_OPEN_ALL_RELAYS)

    def set_verify(self, relays, on):
        """Configure Close? to return indicator state instead of driven state.

        @param relays: relays to verify (, to separate and : for range)
        @param on: string state to verify (on(True)/off(False))

        """
        self.send_cmd_check_error('%s %s,(@%s)\n' % (
            self._CMD_SET_VERIFY, int(bool(on)), relays))

    def get_verify(self, relays):
        """Get the verify mode of relays.

        @param relays: relays to verify (, to separate and : for range)
        @returns tuple of verify mode for relays

        """
        status = self.query('%s (@%s)\n' % (
            self._CMD_GET_VERIFY, relays)).strip().split(',')
        return tuple(bool(int(x)) for x in status)

    def set_verify_inverted(self, relays, inverted):
        """Set the polarity confidence of relay to be inverted.

        @param relays: relays to set (, to separate and : for range)
        @param inverted: Boolean True if INV, False for NORM

        """
        self.send_cmd_check_error('%s %s,(@%s)\n' % (
                self._CMD_SET_VERIFY_INVERTED,
                'INV' if inverted else 'NORM', relays))

    def get_verify_inverted(self, relays):
        """Get the confidence polarity. 1 is inverted, 0 is value as-is.

        @param relays: relays to get (, to separate and : for range)
        @returns tuple of status where 1 in inverted and 0 for value as-is

        """
        status = self.query('%s (@%s)\n' % (self._CMD_GET_VERIFY_INVERTED,
                                            relays)).strip().split(',')
        return tuple(bool(int(x)) for x in status)

    def get_verify_state(self, relays):
        """If verify set get driven state, else confidence state.

        @param relays: relays to get (, to separate and : for range)

        @returns tuple of verify status for given relays

        """
        status = self.query('%s (@%s)\n' % (self._CMD_GET_VERIFY_STATE,
                                            relays)).strip().split(',')
        return tuple(bool(int(x)) for x in status)

    @property
    def busy(self):
        """Check relays are still settling.

        @returns Boolean True if relays are settling, False if not.

        """
        return bool(int(self.query('%s\n' % self._CMD_CHECK_BUSY).strip()))

    def wait(self):
        """Wait for relays to debounce."""
        self.write('%s\n' % self._CMD_WAIT)

    def get_attenuation(self, ap_enclosure):
        """Get Attenuation for an AP enclosure.

        Attenuation is set by turning on/off the relays.  Each relay specifies
        a bit in the binary value of attenuation.  Find the relay status and
        build the binary value, then find the decimal value from it.

        @param ap_enclosure: Int 1/2/3/4 (AP enclosure index)

        @returns attenuation value in int

        @raises ValueError: on bad ap_enclosure value

        """
        if (ap_enclosure < self._MIN_ENCLOSURE or
                ap_enclosure > self._MAX_ENCLOSURE):
            raise ValueError('Invalid AP enclosure number: %s.', ap_enclosure)
        else:
            a_status = self.relays_closed(
                self._AP_ATTENUATOR_RELAYS_SHORT[ap_enclosure])
            status = a_status[::-1]  # reverse for Endian transform
            logging.debug('attenuator status: %s', status)

            # build the binary value
            binary = ''.join(['0' if x else '1' for x in status])
            return int(binary, 2)

    def get_all_attenuation(self):
        """Get attenuation value for all AP enclosures.

        @returns tuple of attenuation value for each enclosure

        """
        attenuations = []
        for x in xrange(self._MIN_ENCLOSURE, self._MAX_ENCLOSURE + 1):
            attenuations.append(self.get_attenuation(x))
        return tuple(attenuations)

    def set_attenuation(self, ap_enclosure, attenuation):
        """Set attenuation for an AP enclosure.

        @param ap_enclosure: Int 0,1,2,3,4 AP enclosure number. 0 for all
        @param attenuation: Int Attenuation value to set

        @raises ValueError: on bad ap_enclosure value

        """
        if ap_enclosure == self._ALL_ENCLOSURE:
            # set attenuation on all
            for x in xrange(self._MIN_ENCLOSURE, self._MAX_ENCLOSURE + 1):
                self.set_attenuation(x, attenuation)
        elif (ap_enclosure < self._MIN_ENCLOSURE or
              ap_enclosure > self._MAX_ENCLOSURE):
            raise ValueError('Bad AP enclosure value: %s' % ap_enclosure)
        else:
            # convert attenuation decimal value to binary
            bin_array = [int(x) for x in list('{0:07b}'.format(attenuation))]
            # For endian
            reverse_bits = self._AP_ATTENUATOR_RELAYS[ap_enclosure][:: -1]

            # determine which bits should be set
            relays_to_close = [
                reverse_bits[i] for i, j in enumerate(bin_array) if not j
            ]

            # open all relay for attenuator & then close the ones we need
            relays = ','.join(self._AP_ATTENUATOR_RELAYS[ap_enclosure])
            self.open_relays(relays)
            logging.debug('Attenuator relays opened: %s', relays)
            relays = ','.join(relays_to_close)
            self.close_relays(relays)
            logging.debug('Attenuator relays closed: %s', relays)

    def get_ap_connections(self):
        """Get a list of AP to client connections.

        @returns tuple of dict of connections ({'AP': 1, 'Client': 1}, ...)

        """
        # Get the closed status for relays connected to AP enclosures.
        ap_connections = self.relays_closed(self._ALL_AP_RELAYS)

        # Find out the connections
        connections = []
        for index, relay_num in enumerate(ap_connections):
            # if the relay is closed, there is a connection
            if relay_num:
                connection = {
                    'AP': (index / self._RELAYS_PER_ENCLOSURE) + 1,
                    'Client': (index % self._RELAYS_PER_ENCLOSURE) + 1
                }
                connections.append(connection)
        return tuple(connections)

    def get_client_connections(self):
        """Get a list of AP to client connections.

        @returns tuple of dict of connections ({'AP': 1, 'Client': 1}, ...)

        """
        # Get the closed status for relays connected to client enclosures.
        c_connections = self.relays_closed(self._ALL_CLIENT_RELAYS)

        # Find out the connections
        connections = []
        for index, relay_num in enumerate(c_connections):
            # if the relay is closed, there is a connection
            if relay_num:
                connection = {
                    'AP': (index % self._RELAYS_PER_ENCLOSURE) + 1,
                    'Client': (index / self._RELAYS_PER_ENCLOSURE) + 1
                }
                connections.append(connection)
        return tuple(connections)

    def connect_ap_client(self, ap, client):
        """Connect AP and a Client enclosure.

        @param ap: Int 1/2/3/4 AP enclosure index
        @param client: Int 1/2/3/4 Client enclosure index

        @raises ValueError: when ap / client value is not 1/2/3/4

        """
        if (ap < self._MIN_ENCLOSURE or ap > self._MAX_ENCLOSURE):
            raise ValueError(
                'AP enclosure should be 1/2/3/4. Given: %s' % ap)
        if (client < self._MIN_ENCLOSURE or
                client > self._MAX_ENCLOSURE):
            raise ValueError('Client enclosure should be 1/2/3/4. Given: %s' %
                             client)

        # Relay index start from 0 so subtract 1 for index
        relays = 'k1_%s,k1_%s' % (
            ((int(ap) - 1) * self._RELAYS_PER_ENCLOSURE) + int(client),
            (((int(client) - 1) + self._MAX_ENCLOSURE) *
             self._RELAYS_PER_ENCLOSURE) + int(ap))
        logging.debug('Relays to close: %s', relays)
        self.close_relays(relays)


@contextlib.contextmanager
def RfSwitchSession(host, port=scpi.Scpi.SCPI_PORT):
    """Start a RF Switch session and close it when done.

    @param host: Hostname or IP address of RF Switch
    @param port: Int SCPI port number (default 5025)

    @yields: an instance of InteractiveSsh
    """
    session = RfSwitch(host, port)
    try:
        yield session
    finally:
        session.close()