[thci] add low power thci (#5835)

This commit adds THCI for Thread 1.2 low power features.

tools/harness-thci/OpenThread.py

@@ -2840,6 +2840,176 @@ class OpenThreadTHCI(object):
         else:
             return False
 
+    # Low power THCI
+    @API
+    def setCSLtout(self, tout=30):
+        self.log('call setCSLtout')
+        cmd = 'csl timeout %u' % tout
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    @API
+    def setCSLchannel(self, ch=11):
+        self.log('call setCSLchannel')
+        cmd = 'csl channel %u' % ch
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    @API
+    def setCSLperiod(self, period=500):
+        """set Csl Period
+        Args:
+            period: csl period in ms
+
+        note: OT command 'csl period' accepts parameter in unit of 10 symbols,
+        period is converted from unit ms to ten symbols (160us per 10 symbols).
+
+        """
+        self.log('call setCSLperiod')
+        cmd = 'csl period %u' % (period * 6.25)
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    def getForwardSeriesFlagsFromHexStr(self, flags):
+        hexFlags = int(flags, 16)
+        strFlags = ''
+        if hexFlags == 0:
+            strFlags = 'X'
+        else:
+            if hexFlags & 0x1 != 0:
+                strFlags += 'l'
+            if hexFlags & 0x2 != 0:
+                strFlags += 'd'
+            if hexFlags & 0x4 != 0:
+                strFlags += 'r'
+            if hexFlags & 0x8 != 0:
+                strFlags += 'a'
+
+        return strFlags
+
+    def mapMetricsHexToChar(self, metrics):
+        metricsFlagMap = {
+            0x40: 'p',
+            0x09: 'q',
+            0x0a: 'm',
+            0x0b: 'r',
+        }
+        return metricsFlagMap.get(metrics, '?')
+
+    def getMetricsFlagsFromHexStr(self, metrics):
+        if metrics.startswith('0x'):
+            metrics = metrics[2:]
+        hexMetricsArray = bytearray.fromhex(metrics)
+
+        strMetrics = ''
+        for metric in hexMetricsArray:
+            strMetrics += mapMetricsHexToChar(metric)
+
+        return strMetrics
+
+    @API
+    def LinkMetricsSingleReq(self, dst_addr, metrics):
+        self.log('call LinkMetricsSingleReq')
+        cmd = 'linkmetrics query %s single %s' % (dst_addr, getMetricsFlagsFromHexStr(metrics))
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    @API
+    def LinkMetricsMgmtReq(self, dst_addr, type_, flags, metrics, series_id):
+        self.log('call LinkMetricsMgmtReq')
+        cmd = 'linkmetrics mgmt %s ' % dst_addr
+        if type_ == 'FWD':
+            cmd += 'forward %d %s' % (series_id, getForwardSeriesFlagsFromHexStr(flags))
+            if flags != 0:
+                cmd += ' %s' % (getMetricsFlagsFromHexStr(metrics))
+        elif type_ == 'ENH':
+            cmd += 'enhanced-ack'
+            if flags != 0:
+                cmd += ' register'
+                metricsFlags = getMetricsFlagsFromHexStr(metrics)
+                if '?' in metricsFlags:
+                    cmd += ' %s r' % metricsFlags.replace('?', '')
+                else:
+                    cmd += ' %s' % metricsFlags
+            else:
+                cmd += ' clear'
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    @API
+    def LinkMetricsGetReport(self, dst_addr, series_id):
+        self.log('call LinkMetricsGetReport')
+        cmd = 'linkmetrics query %s forward %d' % (dst_addr, series_id)
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    #TODO: Series Id is not in this API.
+    @API
+    def LinkMetricsSendProbe(self, dst_addr, ack=True, size=0):
+        self.log('call LinkMetricsSendProbe')
+        cmd = 'linkmetrics probe %s %d' % (dst_addr, size)
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    @API
+    def setTxPower(self, level):
+        self.log('call setTxPower')
+        cmd = 'txpower '
+        if level == 'HIGH':
+            cmd += '127'
+        elif level == 'MEDIUM':
+            cmd += '0'
+        elif level == 'LOW':
+            cmd += '-128'
+        else:
+            print('wrong Tx Power level')
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    @API
+    def sendUdp(self, destination, port, payload='hello'):
+        self.log('call sendUdp')
+        assert payload is not None, 'payload should not be none'
+        cmd1 = 'udp open'
+        print(cmd1)
+        cmd2 = 'udp send %s %d %s' % (destination, port, payload)
+        print(cmd2)
+        return (self.__executeCommand(cmd1)[-1] == 'Done' and self.__executeCommand(cmd2)[-1] == 'Done')
+
+    @API
+    def send_udp(self, interface, destination, port, payload='12ABcd'):
+        ''' payload hexstring
+        '''
+        self.log('call send_udp')
+        assert payload is not None, 'payload should not be none'
+        cmd1 = 'udp open'
+        print(cmd1)
+        cmd2 = 'udp send %s %s -x %s' % (destination, port, payload)
+        print(cmd2)
+        return (self.__executeCommand(cmd1)[-1] == 'Done' and self.__executeCommand(cmd2)[-1] == 'Done')
+
+    @API
+    def sendMACcmd(self, enh=False):
+        self.log('call sendMACcmd')
+        cmd = 'mac send datarequest'
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    @API
+    def sendMACdata(self, enh=False):
+        self.log('call sendMACdata')
+        cmd = 'mac send emptydata'
+        print(cmd)
+        return self.__executeCommand(cmd)[-1] == 'Done'
+
+    @API
+    def setCSLsuspension(self, suspend):
+        self.log('call setCSLsuspension')
+        if suspend:
+            self.__setPollPeriod(1000000)
+        else:
+            self.__setPollPeriod(self.__sedPollPeriod)
+
     @staticmethod
     def __lstrip0x(s):
         """strip 0x at the beginning of a hex string if it exists