| # 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() |
