Merge "[RTT] Update RTT ACTS to select 'best' APs for ranging" into pi-dev

5 files changed, 48 insertions, 14 deletions

diff --git a/acts/framework/acts/test_utils/wifi/rtt/rtt_test_utils.py b/acts/framework/acts/test_utils/wifi/rtt/rtt_test_utils.py
index 29fac81d28..c24b406893 100644
--- a/acts/framework/acts/test_utils/wifi/rtt/rtt_test_utils.py
+++ b/acts/framework/acts/test_utils/wifi/rtt/rtt_test_utils.py
@@ -143,6 +143,33 @@ def scan_with_rtt_support_constraint(dut, support_rtt, repeat=0):
   return []
 
 
+def select_best_scan_results(scans, select_count, lowest_rssi=-80):
+  """Select the strongest 'select_count' scans in the input list based on
+  highest RSSI. Exclude all very weak signals, even if results in a shorter
+  list.
+
+  Args:
+    scans: List of scan results.
+    select_count: An integer specifying how many scans to return at most.
+    lowest_rssi: The lowest RSSI to accept into the output.
+  Returns: a list of the strongest 'select_count' scan results from the scans
+           list.
+  """
+  def takeRssi(element):
+    return element['level']
+
+  result = []
+  scans.sort(key=takeRssi, reverse=True)
+  for scan in scans:
+    if len(result) == select_count:
+      break
+    if scan['level'] < lowest_rssi:
+      break # rest are lower since we're sorted
+    result.append(scan)
+
+  return result
+
+
 def validate_ap_result(scan_result, range_result):
   """Validate the range results:
   - Successful if AP (per scan result) support 802.11mc (allowed to fail
diff --git a/acts/tests/google/wifi/rtt/functional/RangeApNonSupporting11McTest.py b/acts/tests/google/wifi/rtt/functional/RangeApNonSupporting11McTest.py
index 813102b4be..65b67d29b2 100644
--- a/acts/tests/google/wifi/rtt/functional/RangeApNonSupporting11McTest.py
+++ b/acts/tests/google/wifi/rtt/functional/RangeApNonSupporting11McTest.py
@@ -14,15 +14,12 @@
 #   See the License for the specific language governing permissions and
 #   limitations under the License.
 
-import queue
-
 from acts import asserts
 from acts.test_decorators import test_tracker_info
-from acts.test_utils.wifi import wifi_test_utils as wutils
+from acts.test_utils.wifi.WifiBaseTest import WifiBaseTest
 from acts.test_utils.wifi.rtt import rtt_const as rconsts
 from acts.test_utils.wifi.rtt import rtt_test_utils as rutils
 from acts.test_utils.wifi.rtt.RttBaseTest import RttBaseTest
-from acts.test_utils.wifi.WifiBaseTest import WifiBaseTest
 
 
 class RangeApNonSupporting11McTest(WifiBaseTest, RttBaseTest):
@@ -47,7 +44,8 @@ class RangeApNonSupporting11McTest(WifiBaseTest, RttBaseTest):
   def test_rtt_non_80211mc_supporting_aps(self):
     """Scan for APs and perform RTT on non-IEEE 802.11mc supporting APs"""
     dut = self.android_devices[0]
-    non_rtt_aps = rutils.scan_with_rtt_support_constraint(dut, False)
+    non_rtt_aps = rutils.select_best_scan_results(
+      rutils.scan_with_rtt_support_constraint(dut, False), select_count=1)
     dut.log.debug("Visible non-IEEE 802.11mc APs=%s", non_rtt_aps)
     asserts.assert_true(len(non_rtt_aps) > 0, "Need at least one AP!")
     events = rutils.run_ranging(dut, non_rtt_aps, self.NUM_ITER,
@@ -84,7 +82,8 @@ class RangeApNonSupporting11McTest(WifiBaseTest, RttBaseTest):
     device not having privilege access (expect failures)."""
     dut = self.android_devices[0]
     rutils.config_privilege_override(dut, True)
-    non_rtt_aps = rutils.scan_with_rtt_support_constraint(dut, False)
+    non_rtt_aps = rutils.select_best_scan_results(
+      rutils.scan_with_rtt_support_constraint(dut, False), select_count=1)
     dut.log.debug("Visible non-IEEE 802.11mc APs=%s", non_rtt_aps)
     asserts.assert_true(len(non_rtt_aps) > 0, "Need at least one AP!")
     events = rutils.run_ranging(dut, non_rtt_aps, self.NUM_ITER,
@@ -114,7 +113,8 @@ class RangeApNonSupporting11McTest(WifiBaseTest, RttBaseTest):
     that get an error result.
     """
     dut = self.android_devices[0]
-    non_rtt_aps = rutils.scan_with_rtt_support_constraint(dut, False)
+    non_rtt_aps = rutils.select_best_scan_results(
+      rutils.scan_with_rtt_support_constraint(dut, False), select_count=1)
     dut.log.debug("Visible non-IEEE 802.11mc APs=%s", non_rtt_aps)
     asserts.assert_true(len(non_rtt_aps) > 0, "Need at least one AP!")
     non_rtt_aps = non_rtt_aps[0:1] # pick first
diff --git a/acts/tests/google/wifi/rtt/functional/RangeApSupporting11McTest.py b/acts/tests/google/wifi/rtt/functional/RangeApSupporting11McTest.py
index 98586cbd48..d889a22c49 100644
--- a/acts/tests/google/wifi/rtt/functional/RangeApSupporting11McTest.py
+++ b/acts/tests/google/wifi/rtt/functional/RangeApSupporting11McTest.py
@@ -42,8 +42,9 @@ class RangeApSupporting11McTest(RttBaseTest):
   def test_rtt_80211mc_supporting_aps(self):
     """Scan for APs and perform RTT only to those which support 802.11mc"""
     dut = self.android_devices[0]
-    rtt_supporting_aps = rutils.scan_with_rtt_support_constraint(dut, True,
-                                                                 repeat=10)
+    rtt_supporting_aps = rutils.select_best_scan_results(
+      rutils.scan_with_rtt_support_constraint(dut, True, repeat=10),
+      select_count=2)
     dut.log.debug("RTT Supporting APs=%s", rtt_supporting_aps)
     events = rutils.run_ranging(dut, rtt_supporting_aps, self.NUM_ITER,
                                 self.TIME_BETWEEN_ITERATIONS)
@@ -87,8 +88,9 @@ class RangeApSupporting11McTest(RttBaseTest):
     """Scan for APs and perform RTT only to those which support 802.11mc - using
     the LEGACY API!"""
     dut = self.android_devices[0]
-    rtt_supporting_aps = rutils.scan_with_rtt_support_constraint(dut, True,
-                                                                 repeat=10)
+    rtt_supporting_aps = rutils.select_best_scan_results(
+      rutils.scan_with_rtt_support_constraint(dut, True, repeat=10),
+      select_count=2)
     dut.log.debug("RTT Supporting APs=%s", rtt_supporting_aps)
 
     rtt_configs = []
diff --git a/acts/tests/google/wifi/rtt/functional/RttDisableTest.py b/acts/tests/google/wifi/rtt/functional/RttDisableTest.py
index cb8ade3ae8..1816cd5f14 100644
--- a/acts/tests/google/wifi/rtt/functional/RttDisableTest.py
+++ b/acts/tests/google/wifi/rtt/functional/RttDisableTest.py
@@ -50,7 +50,8 @@ class RttDisableTest(WifiBaseTest, RttBaseTest):
     asserts.assert_true(dut.droid.wifiIsRttAvailable(), "RTT is not available")
 
     # scan to get some APs to be used later
-    all_aps = rutils.scan_networks(dut)
+    all_aps = rutils.select_best_scan_results(rutils.scan_networks(dut),
+                                              select_count=1)
     asserts.assert_true(len(all_aps) > 0, "Need at least one visible AP!")
 
     # disable RTT and validate broadcast & API
diff --git a/acts/tests/google/wifi/rtt/functional/RttRequestManagementTest.py b/acts/tests/google/wifi/rtt/functional/RttRequestManagementTest.py
index 8370290e42..82c1058bc2 100644
--- a/acts/tests/google/wifi/rtt/functional/RttRequestManagementTest.py
+++ b/acts/tests/google/wifi/rtt/functional/RttRequestManagementTest.py
@@ -57,7 +57,9 @@ class RttRequestManagementTest(RttBaseTest):
     all_uids = [1000, 20, 30] # 1000 = System Server (makes requests foreground)
     some_uids = [20, 30]
 
-    aps = rutils.scan_with_rtt_support_constraint(dut, True, repeat=10)
+    aps = rutils.select_best_scan_results(
+      rutils.scan_with_rtt_support_constraint(dut, True, repeat=10),
+      select_count=1)
     dut.log.info("RTT Supporting APs=%s", aps)
 
     asserts.assert_true(
@@ -112,7 +114,9 @@ class RttRequestManagementTest(RttBaseTest):
     # background uid will be throttled on the next run of this script
     fake_uid = [random.randint(10, 9999)]
 
-    aps = rutils.scan_with_rtt_support_constraint(dut, True, repeat=10)
+    aps = rutils.select_best_scan_results(
+      rutils.scan_with_rtt_support_constraint(dut, True, repeat=10),
+      select_count=1)
     dut.log.info("RTT Supporting APs=%s", aps)
 
     asserts.assert_true(