[RTT] Update RTT ACTS to select 'best' APs for ranging

Current testing scanned for all APs of a particular type (e.g. RTT
supporting APs) and then ranged to all of them (possibly restricting
to the number of allowed parallel requests). This could result (in
an open environment) with picking ranging to distant APs over ranging
to nearer APs.

Modify tests to:
- Select the best (strongest RSSI) APs first.
- Limit the selected APs to have RSSI above a reasonable threshold.
- Limit the number of selected APs

Bug: 80319317
Test: all modified tests pass
Change-Id: Id08923066c415f3d3cf9a4a12951fbec117f0f1e

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 36722888a7..e99a914f45 100644
--- a/acts/tests/google/wifi/rtt/functional/RttDisableTest.py
+++ b/acts/tests/google/wifi/rtt/functional/RttDisableTest.py
@@ -46,7 +46,8 @@ class RttDisableTest(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(