[harness-automation] add rf shield device support (#2172)

tools/harness-automation/autothreadharness/harness_case.py

@@ -47,6 +47,7 @@ from autothreadharness.harness_controller import HarnessController
 from autothreadharness.helpers import HistoryHelper
 from autothreadharness.open_thread_controller import OpenThreadController
 from autothreadharness.pdu_controller_factory import PduControllerFactory
+from autothreadharness.rf_shield_controller import get_rf_shield_controller
 
 logger = logging.getLogger(__name__)
 
@@ -280,6 +281,18 @@ class HarnessCase(unittest.TestCase):
         self._browser.close()
         self._browser = None
 
+    def _init_rf_shield(self):
+        if getattr(settings, 'SHIELD_CONTROLLER_TYPE', None) and getattr(settings, 'SHIELD_CONTROLLER_PARAMS', None):
+            self.rf_shield = get_rf_shield_controller(
+                shield_type=settings.SHIELD_CONTROLLER_TYPE,
+                params=settings.SHIELD_CONTROLLER_PARAMS
+            )
+        else:
+            self.rf_shield = None
+
+    def _destroy_rf_shield(self):
+        self.rf_shield = None
+
     def setUp(self):
         """Prepare to run test case.
 
@@ -306,6 +319,7 @@ class HarnessCase(unittest.TestCase):
         self._init_harness()
         self._init_devices()
         self._init_dut()
+        self._init_rf_shield()
 
     def tearDown(self):
         """Clean up after each case.
@@ -319,6 +333,7 @@ class HarnessCase(unittest.TestCase):
         self._destroy_harness()
         self._destroy_browser()
         self._destroy_dut()
+        self._destroy_rf_shield()
 
     def _setup_page(self):
         """Do sniffer settings and general settings
@@ -791,14 +806,22 @@ class HarnessCase(unittest.TestCase):
             inp.send_keys(ml64)
 
         elif title.startswith('Shield Devices') or title.startswith('Sheild DUT'):
-            if self.dut and settings.SHIELD_SIMULATION:
+            if self.rf_shield:
+                logger.info('Shielding devices')
+                with self.rf_shield:
+                    self.rf_shield.shield()
+            elif self.dut and settings.SHIELD_SIMULATION:
                 self.dut.channel = (self.channel == THREAD_CHANNEL_MAX
                                     and THREAD_CHANNEL_MIN) or (self.channel + 1)
             else:
                 raw_input('Shield DUT and press enter to continue..')
 
         elif title.startswith('Unshield Devices') or title.startswith('Bring DUT Back to network'):
-            if self.dut and settings.SHIELD_SIMULATION:
+            if self.rf_shield:
+                logger.info('Unshielding devices')
+                with self.rf_shield:
+                    self.rf_shield.unshield()
+            elif self.dut and settings.SHIELD_SIMULATION:
                 self.dut.channel = self.channel
             else:
                 raw_input('Bring DUT and press enter to continue..')

tools/harness-automation/autothreadharness/rf_shield_controller.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python
+#
+# Copyright (c) 2017, The OpenThread Authors.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+# 1. Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+# 3. Neither the name of the copyright holder nor the
+#    names of its contributors may be used to endorse or promote products
+#    derived from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+#
+
+import sys
+import abc
+import logging
+import serial
+import time
+
+
+ABC = abc.ABC if sys.version_info >= (3, 4) else abc.ABCMeta('ABC', (), {})
+logger = logging.getLogger(__name__)
+
+
+class RfShieldController(ABC):
+
+    @abc.abstractmethod
+    def shield(self):
+        pass
+
+    @abc.abstractmethod
+    def unshield(self):
+        pass
+
+    @abc.abstractmethod
+    def __enter__(self):
+        pass
+
+    @abc.abstractmethod
+    def __exit__(self, *args, **kwargs):
+        pass
+
+
+class RfSwitchController(RfShieldController):
+
+    def __init__(self, channel, port):
+        self._channel = channel
+        self._port = port
+        self._conn = None
+
+    def shield(self):
+        self._display_string('CLOSE {}'.format(self._channel))
+        self._write('CLOSE (@{})'.format(self._channel))
+
+    def unshield(self):
+        self._display_string('OPEN {}'.format(self._channel))
+        self._write('OPEN (@{})'.format(self._channel))
+
+    def _write(self, data):
+        return self._conn.write('{}\r\n'.format(data))
+
+    def _display_string(self, string):
+        self._write('DIAGNOSTIC:DISPLAY \"{}\"'.format(string))
+
+    def __enter__(self):
+        self._conn = serial.Serial(self._port, 9600)
+        if not self._conn.isOpen():
+            self._conn.open()
+        self._write('')
+        time.sleep(1)
+        return self
+
+    def __exit__(self, *args, **kwargs):
+        if self._conn:
+            self._conn.close()
+            self._conn = None
+
+
+CONTROLLERS = {
+    'RF_SWITCH': RfSwitchController,
+}
+
+
+def get_rf_shield_controller(shield_type, params):
+    if shield_type in CONTROLLERS:
+        return CONTROLLERS[shield_type](**params)
+    logger.exception('Unknown RF shield controller type: {}'.format(shield_type))

tools/harness-automation/autothreadharness/settings_sample.py

@@ -116,3 +116,20 @@ Example parameters for the 'APC_PDU_CONTROLLER':
 Example parameters for the 'NORDIC_BOARD_PDU_CONTOLLER':
     {'boards_serial_numbers': ('12345123', ...)}
 """
+
+SHIELD_CONTROLLER_TYPE = None
+"""str: Type of connected RF Shield controller.
+
+Keep this None if no RF Shield controller available.
+
+Types of supported RF Shield controllers:
+    - None - when no RF Shield controller connected
+    - 'RF_SWITCH' - when RF Switch connected
+"""
+
+SHIELD_CONTROLLER_PARAMS = None
+"""dict: Parameters passed to the "__init__" method of RF Shield controller.
+
+Example parameters for the 'RF_SWITCH':
+    {'channel': 200, 'port': 'COM50'}
+"""