fix: use iperf-cmd for example esp_ble_mesh/wifi_coexist

tools/ci/python_packages/idf_iperf_test_util/IperfUtility.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: 2022-2025 Espressif Systems (Shanghai) CO LTD
+# SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD
 # SPDX-License-Identifier: Apache-2.0
 import logging
 import os
@@ -13,7 +13,6 @@ from idf_iperf_test_util import LineChart
 
 try:
     from typing import Any
-    from typing import Tuple
 except ImportError:
     # Only used for type annotations
     pass
@@ -31,7 +30,7 @@ INVALID_HEAP_SIZE = 0xFFFFFFFF
 PC_IPERF_TEMP_LOG_FILE = '.tmp_iperf.log'
 
 
-class TestResult(object):
+class TestResult:
     """record, analysis test result and convert data to output format"""
 
     PC_BANDWIDTH_LOG_PATTERN = re.compile(r'(\d+\.\d+)\s*-\s*(\d+.\d+)\s+sec\s+[\d.]+\s+MBytes\s+([\d.]+)\s+Mbits\/sec')
@@ -119,24 +118,21 @@ class TestResult(object):
             if throughput == 0 and rssi > self.ZERO_POINT_THRESHOLD and fall_to_0_recorded < 1:
                 # throughput fall to 0 error. we only record 1 records for one test
                 self.error_list.append(
-                    '[Error][fall to 0][{}][att: {}][rssi: {}]: 0 throughput interval: {}-{}'.format(
-                        ap_ssid, att, rssi, result[0], result[1]
-                    )
+                    f'[Error][fall to 0][{ap_ssid}][att: {att}][rssi: {rssi}]: 0 '
+                    f'throughput interval: {result[0]}-{result[1]}'
                 )
                 fall_to_0_recorded += 1
 
         if len(throughput_list) < self.THROUGHPUT_QUALIFY_COUNT:
             self.error_list.append(
-                '[Error][Fatal][{}][att: {}][rssi: {}]: Only {} throughput values found, expected at least {}'.format(
-                    ap_ssid, att, rssi, len(throughput_list), self.THROUGHPUT_QUALIFY_COUNT
-                )
+                f'[Error][Fatal][{ap_ssid}][att: {att}][rssi: {rssi}]: '
+                f'Only {len(throughput_list)} throughput values found, '
+                f'expected at least {self.THROUGHPUT_QUALIFY_COUNT}'
             )
             max_throughput = 0.0
 
         if max_throughput == 0 and rssi > self.ZERO_THROUGHPUT_THRESHOLD:
-            self.error_list.append(
-                '[Error][Fatal][{}][att: {}][rssi: {}]: No throughput data found'.format(ap_ssid, att, rssi)
-            )
+            self.error_list.append(f'[Error][Fatal][{ap_ssid}][att: {att}][rssi: {rssi}]: No throughput data found')
 
         self._save_result(max_throughput, ap_ssid, att, rssi, heap_size)
 
@@ -167,9 +163,8 @@ class TestResult(object):
                     _percentage = result_dict[index_value] / result_dict[index_list[i]]
                     if _percentage < 1 - self.BAD_POINT_PERCENTAGE_THRESHOLD:
                         self.error_list.append(
-                            '[Error][Bad point][{}][{}: {}]: drop {:.02f}%'.format(
-                                ap_ssid, index_type, index_value, (1 - _percentage) * 100
-                            )
+                            f'[Error][Bad point][{ap_ssid}][{index_type}: {index_value}]: '
+                            f'drop {(1 - _percentage) * 100:.02f}%'
                         )
 
         analysis_bad_point(self.throughput_by_rssi, 'rssi')
@@ -193,14 +188,14 @@ class TestResult(object):
         else:
             raise AssertionError('draw type not supported')
         if isinstance(ap_ssid, list):
-            file_name = 'ThroughputVs{}_{}_{}.html'.format(type_name, self.proto, self.direction)
+            file_name = f'ThroughputVs{type_name}_{self.proto}_{self.direction}.html'
         else:
-            file_name = 'ThroughputVs{}_{}_{}.html'.format(type_name, self.proto, self.direction)
+            file_name = f'ThroughputVs{type_name}_{self.proto}_{self.direction}.html'
 
         LineChart.draw_line_chart(
             os.path.join(path, file_name),
-            'Throughput Vs {} ({} {})'.format(type_name, self.proto, self.direction),
-            '{} (dbm)'.format(type_name),
+            f'Throughput Vs {type_name} ({self.proto} {self.direction})',
+            f'{type_name} (dbm)',
             'Throughput (Mbps)',
             data,
             range_list,
@@ -241,15 +236,15 @@ class TestResult(object):
             )
             ret += 'Performance for each AP:\r\n'
             for ap_ssid in self.throughput_by_att:
-                ret += '[{}]: {:.02f} Mbps\r\n'.format(ap_ssid, max(self.throughput_by_att[ap_ssid].values()))
+                ret += f'[{ap_ssid}]: {max(self.throughput_by_att[ap_ssid].values()):.02f} Mbps\r\n'
             if self.heap_size != INVALID_HEAP_SIZE:
-                ret += 'Minimum heap size: {}'.format(self.heap_size)
+                ret += f'Minimum heap size: {self.heap_size}'
         else:
             ret = ''
         return ret
 
 
-class IperfTestUtility(object):
+class IperfTestUtility:
     """iperf test implementation"""
 
     def __init__(
@@ -282,7 +277,7 @@ class IperfTestUtility(object):
                 'udp_rx': TestResult('udp', 'rx', config_name),
             }
 
-    def setup(self) -> Tuple[str, int]:
+    def setup(self) -> tuple[str, int]:
         """
         setup iperf test:
 
@@ -299,23 +294,23 @@ class IperfTestUtility(object):
         self.dut.write('restart')
         self.dut.expect_exact("Type 'help' to get the list of commands.")
         self.dut.expect('iperf>')
-        self.dut.write('sta_scan {}'.format(self.ap_ssid))
+        self.dut.write(f'sta_scan {self.ap_ssid}')
         for _ in range(SCAN_RETRY_COUNT):
             try:
-                rssi = int(self.dut.expect(r'\[{}]\[rssi=(-\d+)]'.format(self.ap_ssid), timeout=SCAN_TIMEOUT).group(1))
+                rssi = int(self.dut.expect(rf'\[{self.ap_ssid}]\[rssi=(-\d+)]', timeout=SCAN_TIMEOUT).group(1))
                 break
             except pexpect.TIMEOUT:
                 continue
         else:
             raise AssertionError('Failed to scan AP')
-        self.dut.write('sta_connect {} {}'.format(self.ap_ssid, self.ap_password))
+        self.dut.write(f'sta_connect {self.ap_ssid} {self.ap_password}')
         dut_ip = self.dut.expect(r'sta ip: ([\d.]+), mask: ([\d.]+), gw: ([\d.]+)').group(1)
         return dut_ip, rssi
 
     def _save_test_result(self, test_case: str, raw_data: str, att: int, rssi: int, heap_size: int) -> Any:
         return self.test_result[test_case].add_result(raw_data, self.ap_ssid, att, rssi, heap_size)
 
-    def _test_once(self, proto: str, direction: str, bw_limit: int) -> Tuple[str, int, int]:
+    def _test_once(self, proto: str, direction: str, bw_limit: int) -> tuple[str, int, int]:
         """do measure once for one type"""
         # connect and scan to get RSSI
         dut_ip, rssi = self.setup()
@@ -333,9 +328,9 @@ class IperfTestUtility(object):
                         stderr=f,
                     )
                     if bw_limit > 0:
-                        self.dut.write('iperf -c {} -i 1 -t {} -b {}'.format(self.pc_nic_ip, TEST_TIME, bw_limit))
+                        self.dut.write(f'iperf -c {self.pc_nic_ip} -i 1 -t {TEST_TIME} -b {bw_limit}m')
                     else:
-                        self.dut.write('iperf -c {} -i 1 -t {}'.format(self.pc_nic_ip, TEST_TIME))
+                        self.dut.write(f'iperf -c {self.pc_nic_ip} -i 1 -t {TEST_TIME}')
                 else:
                     process = subprocess.Popen(
                         ['iperf', '-s', '-u', '-B', self.pc_nic_ip, '-t', str(TEST_TIME), '-i', '1', '-f', 'm'],
@@ -343,9 +338,9 @@ class IperfTestUtility(object):
                         stderr=f,
                     )
                     if bw_limit > 0:
-                        self.dut.write('iperf -c {} -u -i 1 -t {} -b {}'.format(self.pc_nic_ip, TEST_TIME, bw_limit))
+                        self.dut.write(f'iperf -c {self.pc_nic_ip} -u -i 1 -t {TEST_TIME} -b {bw_limit}m')
                     else:
-                        self.dut.write('iperf -c {} -u -i 1 -t {}'.format(self.pc_nic_ip, TEST_TIME))
+                        self.dut.write(f'iperf -c {self.pc_nic_ip} -u -i 1 -t {TEST_TIME}')
 
                 for _ in range(TEST_TIMEOUT):
                     if process.poll() is not None:
@@ -354,12 +349,12 @@ class IperfTestUtility(object):
                 else:
                     process.terminate()
 
-            with open(PC_IPERF_TEMP_LOG_FILE, 'r') as f:
+            with open(PC_IPERF_TEMP_LOG_FILE) as f:
                 pc_raw_data = server_raw_data = f.read()
         else:
             with open(PC_IPERF_TEMP_LOG_FILE, 'w') as f:
                 if proto == 'tcp':
-                    self.dut.write('iperf -s -i 1 -t {}'.format(TEST_TIME))
+                    self.dut.write(f'iperf -s -i 1 -t {TEST_TIME}')
                     # wait until DUT TCP server created
                     try:
                         self.dut.expect('Socket created', timeout=5)
@@ -384,7 +379,7 @@ class IperfTestUtility(object):
                         process.terminate()
                 else:
                     if bw_limit > 0:
-                        self.dut.write('iperf -s -u -i 1 -t {}'.format(TEST_TIME))
+                        self.dut.write(f'iperf -s -u -i 1 -t {TEST_TIME}')
                         # wait until DUT TCP server created
                         try:
                             self.dut.expect('Socket created', timeout=5)
@@ -408,7 +403,7 @@ class IperfTestUtility(object):
                         n = 10
                         step = int((stop_bw - start_bw) / n)
                         self.dut.write(
-                            'iperf -s -u -i 1 -t {}'.format(TEST_TIME + 4 * (n + 1))
+                            f'iperf -s -u -i 1 -t {TEST_TIME + 4 * (n + 1)}'
                         )  # 4 sec for each bw step instance start/stop
                         # wait until DUT TCP server created
                         try:
@@ -441,7 +436,7 @@ class IperfTestUtility(object):
                                 process.terminate()
 
             server_raw_data = self.dut.expect(pexpect.TIMEOUT, timeout=5).decode('utf-8')
-            with open(PC_IPERF_TEMP_LOG_FILE, 'r') as f:
+            with open(PC_IPERF_TEMP_LOG_FILE) as f:
                 pc_raw_data = f.read()
 
         if os.path.exists(PC_IPERF_TEMP_LOG_FILE):
@@ -452,7 +447,7 @@ class IperfTestUtility(object):
             f.write(
                 '## [{}] `{}`\r\n##### {}'.format(
                     self.config_name,
-                    '{}_{}'.format(proto, direction),
+                    f'{proto}_{direction}',
                     time.strftime('%m-%d %H:%M:%S', time.localtime(time.time())),
                 )
             )
@@ -476,17 +471,11 @@ class IperfTestUtility(object):
         heap_size = INVALID_HEAP_SIZE
         try:
             server_raw_data, rssi, heap_size = self._test_once(proto, direction, bw_limit)
-            throughput = self._save_test_result(
-                '{}_{}'.format(proto, direction), server_raw_data, atten_val, rssi, heap_size
-            )
-            logging.info(
-                '[{}][{}_{}][{}][{}]: {:.02f}'.format(
-                    self.config_name, proto, direction, rssi, self.ap_ssid, throughput
-                )
-            )
+            throughput = self._save_test_result(f'{proto}_{direction}', server_raw_data, atten_val, rssi, heap_size)
+            logging.info(f'[{self.config_name}][{proto}_{direction}][{rssi}][{self.ap_ssid}]: {throughput:.02f}')
             self.lowest_rssi_scanned = min(self.lowest_rssi_scanned, rssi)
         except (ValueError, IndexError):
-            self._save_test_result('{}_{}'.format(proto, direction), '', atten_val, rssi, heap_size)
+            self._save_test_result(f'{proto}_{direction}', '', atten_val, rssi, heap_size)
             logging.info('Fail to get throughput results.')
         except AssertionError:
             self.fail_to_scan += 1
@@ -504,9 +493,7 @@ class IperfTestUtility(object):
         self.run_test('udp', 'tx', atten_val, bw_limit)
         self.run_test('udp', 'rx', atten_val, bw_limit)
         if self.fail_to_scan > 10:
-            logging.info(
-                'Fail to scan AP for more than 10 times. Lowest RSSI scanned is {}'.format(self.lowest_rssi_scanned)
-            )
+            logging.info(f'Fail to scan AP for more than 10 times. Lowest RSSI scanned is {self.lowest_rssi_scanned}')
             raise AssertionError
 
     def wait_ap_power_on(self) -> bool:
@@ -520,8 +507,8 @@ class IperfTestUtility(object):
         self.dut.expect('iperf>')
         for _ in range(WAIT_AP_POWER_ON_TIMEOUT // SCAN_TIMEOUT):
             try:
-                self.dut.write('scan {}'.format(self.ap_ssid))
-                self.dut.expect(r'\[{}]\[rssi=(-\d+)]'.format(self.ap_ssid), timeout=SCAN_TIMEOUT)
+                self.dut.write(f'scan {self.ap_ssid}')
+                self.dut.expect(rf'\[{self.ap_ssid}]\[rssi=(-\d+)]', timeout=SCAN_TIMEOUT)
                 ret = True
                 break
             except pexpect.TIMEOUT: