Merge pull request #914 from Wireless-Innovation-Forum/dpa-median-pathloss

Use median for DPA interference calc

Author: FW-ccaparanga

src/harness/reference_models/dpa/dpa_mgr.py

@@ -1,4 +1,4 @@
-#    Copyright 2018 SAS Project Authors. All Rights Reserved.
+#    Copyright 2024 SAS Project Authors. All Rights Reserved.
 #
 #    Licensed under the Apache License, Version 2.0 (the "License");
 #    you may not use this file except in compliance with the License.
@@ -132,8 +132,8 @@ class Dpa(object):
       OR the updated sequence (cata_indoor_dist, cata_indoor_6m_dist,
       cata_outdoor_dist, cata_outdoor_6m_dist, catb_dist, catb_6m_dist)
     monitor_type: The DPA monitoring category, either 'esc' or 'portal'.
-    apply_clutter_and_network_loss: Use updatd DPA pathloss which includes
-      clutter and network loss.
+    apply_clutter_network_loss_and_50_percent: Use updated DPA pathloss which includes
+      clutter,network loss, and median for interference calc
 
     move_lists: A list of move list (set of |CbsdGrantInfo|) per channel.
     nbor_lists: A list of neighbor list (set of |CbsdGrantInfo|) per channel.
@@ -176,7 +176,7 @@ def __init__(self, protected_points,
                freq_ranges_mhz=[DPA_DEFAULT_FREQ_RANGE],
                neighbor_distances=DPA_DEFAULT_DISTANCES,
                monitor_type='esc',
-               apply_clutter_and_network_loss=False):
+               apply_clutter_network_loss_and_50_percent=False):
     """Initialize the DPA attributes."""
     self.name = name
     self.geometry = geometry
@@ -187,7 +187,7 @@ def __init__(self, protected_points,
     self.beamwidth = beamwidth
     self.neighbor_distances = neighbor_distances
     self.monitor_type = monitor_type
-    self.apply_clutter_and_network_loss = apply_clutter_and_network_loss
+    self.apply_clutter_network_loss_and_50_percent = apply_clutter_network_loss_and_50_percent
     self._channels = None
     self._grants = []
     self._has_th_grants = False
@@ -291,7 +291,7 @@ def ComputeMoveLists(self):
           min_azimuth=self.azimuth_range[0],
           max_azimuth=self.azimuth_range[1],
           neighbor_distances=self.neighbor_distances,
-          apply_clutter_and_network_loss=self.apply_clutter_and_network_loss)
+          apply_clutter_network_loss_and_50_percent=self.apply_clutter_network_loss_and_50_percent)
 
       move_list, nbor_list = list(
           zip(*pool.map(moveListConstraint, self.protected_points)))
@@ -381,7 +381,7 @@ def CalcKeepListInterference(self, channel, num_iter=None):
         max_azimuth=self.azimuth_range[1],
         neighbor_distances=self.neighbor_distances,
         do_max=True,
-        apply_clutter_and_network_loss=self.apply_clutter_and_network_loss)
+        apply_clutter_network_loss_and_50_percent=self.apply_clutter_network_loss_and_50_percent)
 
     pool = mpool.Pool()
     max_interf = pool.map(interfCalculator,
@@ -392,7 +392,8 @@ def CheckInterference(self, sas_uut_active_grants, margin_db,
                         channel=None, num_iter=None,
                         do_abs_check_single_uut=False,
                         extensive_print=True,
-                        output_data=None):
+                        output_data=None,
+                        apply_clutter_network_loss_and_50_percent=False):
     """Checks interference of keep list of SAS UUT vs test harness.
 
     This compares the aggregated interference (within some margin) generated by:
@@ -427,6 +428,7 @@ def CheckInterference(self, sas_uut_active_grants, margin_db,
         interference results (|DpaInterferenceResult|) for each protected point
         and channel. Indexing is [chan_idx, point_idx], unless channel is unique,
         in which case indexing is [point_idx].
+      apply_clutter_network_loss_and_50_percent: if true, add signal and clutter loss and use median
 
     Returns:
       True if all SAS UUT aggregated interference are within margin, False otherwise
@@ -441,7 +443,7 @@ def CheckInterference(self, sas_uut_active_grants, margin_db,
       for k, chan in enumerate(self._channels):
         if not self.CheckInterference(sas_uut_active_grants, margin_db, chan, num_iter,
                                       do_abs_check_single_uut, extensive_print,
-                                      output_data[k]):
+                                      output_data[k], apply_clutter_network_loss_and_50_percent):
           test_passed = False
       return test_passed
 
@@ -463,13 +465,17 @@ def CheckInterference(self, sas_uut_active_grants, margin_db,
       margin_db = float(margin_db[idx1+1:idx2].strip())
 
     # Find the keep list component of TH: SAS UUT and peer SASes.
-    keep_list = self.GetKeepList(channel)
-    keep_list_th_other_sas = [
-        grant for grant in self._grants
-        if (not grant.is_managed_grant and grant in keep_list)]
-    keep_list_th_managing_sas = [
-        grant for grant in self._grants
-        if (grant.is_managed_grant and grant in keep_list)]
+    if apply_clutter_network_loss_and_50_percent:
+        keep_list_th_other_sas = []
+        keep_list_th_managing_sas = []
+    else:
+        keep_list = self.GetKeepList(channel)
+        keep_list_th_other_sas = [
+            grant for grant in self._grants
+            if (not grant.is_managed_grant and grant in keep_list)]
+        keep_list_th_managing_sas = [
+            grant for grant in self._grants
+            if (grant.is_managed_grant and grant in keep_list)]
 
     # Makes sure we have a list of SAS UUT active grants
     sas_uut_active_grants = list(sas_uut_active_grants)
@@ -515,7 +521,7 @@ def CheckInterference(self, sas_uut_active_grants, margin_db,
         num_iter=num_iter,
         azimuth_range=self.azimuth_range,
         neighbor_distances=self.neighbor_distances,
-        apply_clutter_and_network_loss=self.apply_clutter_and_network_loss,
+        apply_clutter_network_loss_and_50_percent=self.apply_clutter_network_loss_and_50_percent,
         threshold=hard_threshold
     )
     pool = mpool.Pool()
@@ -697,7 +703,7 @@ def _CalcTestPointInterfDiff(point,
                              num_iter,
                              azimuth_range,
                              neighbor_distances,
-                             apply_clutter_and_network_loss,
+                             apply_clutter_network_loss_and_50_percent,
                              threshold=None):
   """Calculate difference of aggregate interference between reference and SAS UUT.
 
@@ -733,7 +739,7 @@ def _CalcTestPointInterfDiff(point,
       or
       [cata_indoor_dist, cata_indoor_6m_dist, cata_outdoor_dist,
         cata_outdoor_6m_dist, catb_dist, catb_6m_dist]
-    apply_clutter_and_network_loss: if true, add signal and clutter loss
+    apply_clutter_network_loss_and_50_percent: if true, add signal and clutter loss and use median
     threshold: If set, do an absolute threshold check of SAS UUT interference against
       threshold. Otherwise compare against the reference model aggregated interference.
 
@@ -758,7 +764,7 @@ def _CalcTestPointInterfDiff(point,
         min_azimuth=azimuth_range[0],
         max_azimuth=azimuth_range[1],
         neighbor_distances=neighbor_distances,
-        apply_clutter_and_network_loss=apply_clutter_and_network_loss)
+        apply_clutter_network_loss_and_50_percent=apply_clutter_network_loss_and_50_percent)
     if threshold is not None:
       max_diff = np.max(uut_interferences - threshold)
       logging.debug('%s UUT interf @ %s Thresh %sdBm Diff %sdB: %s',
@@ -784,7 +790,7 @@ def _CalcTestPointInterfDiff(point,
         min_azimuth=azimuth_range[0],
         max_azimuth=azimuth_range[1],
         neighbor_distances=neighbor_distances,
-        apply_clutter_and_network_loss=apply_clutter_and_network_loss)
+        apply_clutter_network_loss_and_50_percent=apply_clutter_network_loss_and_50_percent)
 
     max_diff = np.max(uut_interferences - th_interferences)
     logging.debug(
@@ -808,7 +814,7 @@ def _CalcTestPointInterfDiff(point,
 
 
 def BuildDpa(dpa_name, protection_points_method=None, portal_dpa_filename=None,
-             use_updated_neighborhoods=False, apply_clutter_and_network_loss=False):
+             use_updated_neighborhoods=False, apply_clutter_network_loss_and_50_percent=False):
   """Builds a DPA parameterized correctly.
 
   The DPA special parameters are obtained from the DPA database.
@@ -840,8 +846,7 @@ def BuildDpa(dpa_name, protection_points_method=None, portal_dpa_filename=None,
       + other 'my_method (p1, p2, ..pk)': The 'my_method` will be checked against registered
         methods in which case it will be used, and passing to it the parameters p1, p2,...
     use_updated_neighborhoods: If true, use the updated set of 6 neighborhoods
-    apply_clutter_and_network_loss: Use updatd DPA pathloss which includes
-      clutter and network loss.
+    apply_clutter_network_loss_and_50_percent: if true, add signal and clutter loss and use median
 
   Returns:
     A Dpa object.
@@ -892,4 +897,4 @@ def BuildDpa(dpa_name, protection_points_method=None, portal_dpa_filename=None,
              freq_ranges_mhz=freq_ranges_mhz,
              neighbor_distances=neighbor_distances,
              monitor_type=monitor_type,
-             apply_clutter_and_network_loss=apply_clutter_and_network_loss)
+             apply_clutter_network_loss_and_50_percent=apply_clutter_network_loss_and_50_percent)

src/harness/reference_models/dpa/dpa_mgr_test.py

@@ -143,6 +143,81 @@ def test_computeMoveListAndCheckInterf(self):
                                     extensive_print=True)
     self.assertEqual(result, False)
 
+  def test_cbsdInsideDpaInMoveList_Rel1Ext(self):
+    dpa = dpa_mgr.BuildDpa('Alameda',
+                           protection_points_method='default(10,2,0,0)',
+                           use_updated_neighborhoods=True,
+                           apply_clutter_network_loss_and_50_percent=True)
+    dpa.ResetFreqRange([(3540, 3650)])
+    alameda_geom = zones.GetCoastalDpaZones()['Alameda'].geometry
+    # Assign grants inside the DPA and inband + OOB in mix of CatA and CatB
+    dpa.ResetFreqRange([(3550, 3650)])
+    np.random.seed(1234)
+    cbsds_a = entities.GenerateCbsdsInPolygon(
+        3, entities.CBSD_TEMPLATE_CAT_A_OUTDOOR, alameda_geom)
+    cbsds_b = entities.GenerateCbsdsInPolygon(
+        2, entities.CBSD_TEMPLATE_CAT_B, alameda_geom)
+    cbsds_a[1] = cbsds_a[1]._replace(eirp_dbm_mhz=-100)
+    cbsds_b[1] = cbsds_b[1]._replace(eirp_dbm_mhz=-100)
+    grants = []
+    grants.extend(entities.ConvertToCbsdGrantInfo(cbsds_a[0:2], 3550, 3560))
+    grants.extend(entities.ConvertToCbsdGrantInfo(cbsds_a[2:3], 3660, 3670))
+    grants.extend(entities.ConvertToCbsdGrantInfo(cbsds_b[0:1], 3550, 3570))
+    grants.extend(entities.ConvertToCbsdGrantInfo(cbsds_b[1:2], 3630, 3670))
+    dpa.SetGrantsFromList(grants)
+
+    dpa.ComputeMoveLists()
+
+    self.assertSetEqual(dpa.GetMoveList((3550, 3560)),
+                        {grants[0], grants[1], grants[3]})
+    self.assertSetEqual(dpa.GetMoveList((3640, 3650)),
+                        {grants[4]})
+
+  def test_computeMoveListAndCheckInterf_Rel1Ext(self):
+    protection_points = [ProtectionPoint(latitude=36.9400, longitude=-75.9989),
+                         ProtectionPoint(latitude=37.7579, longitude=-75.4105),
+                         ProtectionPoint(latitude=36.1044, longitude=-73.3147),
+                         ProtectionPoint(latitude=36.1211, longitude=-75.5939)]
+    regs = [json_load(os.path.join(TEST_DIR, 'RegistrationRequest_%d.json' % k))
+            for k in range(1, 7)]
+    grants = [json_load(os.path.join(TEST_DIR, 'GrantRequest_%d.json' % k))
+              for k in range(1, 7)]
+    grants_uut = data.getGrantsFromRequests(regs[:4], grants[:4])
+    grants_th = data.getGrantsFromRequests(regs[4:], grants[4:])
+    channel = (3600, 3610)
+    dpa = dpa_mgr.Dpa(protection_points,
+                      geometry=sgeo.Polygon([(pt.longitude, pt.latitude)
+                                             for pt in protection_points]),
+                      name='test(East1)',
+                      threshold=-144,
+                      beamwidth=3,
+                      radar_height=50,
+                      neighbor_distances=(150, 190, 0, 25),
+                      freq_ranges_mhz=[channel],
+                      apply_clutter_network_loss_and_50_percent=True)
+    dpa.SetGrantsFromList(grants_uut + grants_th)
+    expected_grants = set(grants_uut[0:3])
+    # Testing move list
+    dpa.ComputeMoveLists()
+    self.assertSetEqual(dpa.GetMoveList(channel),
+                        expected_grants)
+    # Testing interference calculation
+    interf = dpa.CalcKeepListInterference(channel)
+    self.assertLess(max(interf), -144)
+    # Testing check
+    result = dpa.CheckInterference([grants_uut[3], grants_th[1]],
+                                   margin_db=0.01,
+                                   do_abs_check_single_uut=True,
+                                   extensive_print=False,
+                                   apply_clutter_network_loss_and_50_percent=True)
+    self.assertEqual(result, True)
+    result = dpa.CheckInterference([grants_uut[2], grants_th[1]],
+                                   margin_db=0.01,
+                                   do_abs_check_single_uut=True,
+                                   extensive_print=True,
+                                   apply_clutter_network_loss_and_50_percent=True)
+    self.assertEqual(result, False)
+
 
 if __name__ == '__main__':
   unittest.main()