Update/results

sharc/results.py

@@ -84,7 +84,14 @@ def __init__(self):
 
         self.system_dl_coupling_loss = SampleList()
         self.system_dl_interf_power = SampleList()
-        # Interference Power [dBm]
+        # Interference Power [dBm/MHz]
+        # NOTE: this may not be what you want for a correct
+        # protection criteria analysis since it is
+        # a mean value. If you have both cochannel
+        # and adjacent channel, the adjacent channel interference
+        # will always drag the mean down
+        self.system_dl_interf_power_per_mhz = SampleList()
+        self.system_ul_interf_power_per_mhz = SampleList()
 
         self.system_inr = SampleList()
         self.system_pfd = SampleList()
@@ -185,7 +192,13 @@ def write_files(self, snapshot_number: int):
             self.overwrite_sample_files = False
 
     @staticmethod
-    def load_many_from_dir(root_dir: str, *, only_latest=True) -> list["Results"]:
+    def load_many_from_dir(
+        root_dir: str,
+        *,
+        only_latest=True,
+        only_samples: list[str] = None,
+        filter_fn=None
+    ) -> list["Results"]:
         output_dirs = list(glob.glob(f"{root_dir}/output_*"))
 
         if len(output_dirs) == 0:
@@ -194,21 +207,27 @@ def load_many_from_dir(root_dir: str, *, only_latest=True) -> list["Results"]:
         if only_latest:
             output_dirs = Results.get_most_recent_outputs_for_each_prefix(output_dirs)
 
+        if filter_fn:
+            output_dirs = filter(filter_fn, output_dirs)
+
         all_res = []
         for output_dir in output_dirs:
             res = Results()
-            res.load_from_dir(output_dir)
+            res.load_from_dir(output_dir, only_samples=only_samples)
             all_res.append(res)
 
         return all_res
 
-    def load_from_dir(self, abs_path: str) -> "Results":
+    def load_from_dir(self, abs_path: str, *, only_samples: list[str] = None) -> "Results":
         self.output_directory = abs_path
 
         self_dict = self.__dict__
-        results_relevant_attr_names = filter(
-            lambda x: isinstance(getattr(self, x), SampleList), self_dict
-        )
+        if only_samples is not None:
+            results_relevant_attr_names = only_samples
+        else:
+            results_relevant_attr_names = filter(
+                lambda x: isinstance(getattr(self, x), SampleList), self_dict
+            )
 
         for attr_name in results_relevant_attr_names:
             file_path = os.path.join(abs_path, f"{attr_name}.csv")

sharc/simulation_downlink.py

@@ -273,6 +273,9 @@ def collect_results(self, write_to_file: bool, snapshot_number: int):
             self.results.system_dl_interf_power.extend(
                 [self.system.rx_interference],
             )
+            self.results.system_dl_interf_power_per_mhz.extend(
+                [self.system.rx_interference - 10 * math.log10(self.system.bandwidth)],
+            )
             # TODO: generalize this a bit more if needed (same conditional as above)
             if hasattr(self.system.antenna[0], "effective_area") and self.system.num_stations == 1:
                 self.results.system_pfd.extend([self.system.pfd])

sharc/simulation_uplink.py

@@ -283,6 +283,9 @@ def collect_results(self, write_to_file: bool, snapshot_number: int):
             self.results.system_ul_interf_power.extend(
                 [self.system.rx_interference],
             )
+            self.results.system_ul_interf_power_per_mhz.extend(
+                [self.system.rx_interference - 10 * math.log10(self.system.bandwidth)],
+            )
             # TODO: generalize this a bit more if needed
             if hasattr(self.system.antenna[0], "effective_area") and self.system.num_stations == 1:
                 self.results.system_pfd.extend([self.system.pfd])

tests/test_results.py

@@ -15,25 +15,39 @@ def setUp(self):
     def test_flush_to_and_load_from_file(self):
         arr1 = [1., 2., 3., 4., 5., 6., 7., 8., 9., 100.]
         self.results.imt_coupling_loss.extend(arr1)
+        self.results.imt_bs_antenna_gain.extend(arr1)
         self.assertGreater(len(self.results.imt_coupling_loss), 0)
+        self.assertGreater(len(self.results.imt_bs_antenna_gain), 0)
         # Results should flush
         self.results.write_files(1)
         # check that no results are left in arr
         self.assertEqual(len(self.results.imt_coupling_loss), 0)
+        self.assertEqual(len(self.results.imt_bs_antenna_gain), 0)
 
         arr2 = [101., 102., 103., 104., 105., 106., 107., 108., 109.]
         self.results.imt_coupling_loss.extend(arr2)
+        self.results.imt_bs_antenna_gain.extend(arr2)
         self.assertGreater(len(self.results.imt_coupling_loss), 0)
+        self.assertGreater(len(self.results.imt_bs_antenna_gain), 0)
         self.results.write_files(2)
         # check that no results are left in arr
         self.assertEqual(len(self.results.imt_coupling_loss), 0)
+        self.assertEqual(len(self.results.imt_bs_antenna_gain), 0)
 
         results_recuperated_from_file = Results().load_from_dir(self.results.output_directory)
 
         results_arr = arr1
         results_arr.extend(arr2)
 
         self.assertEqual(results_recuperated_from_file.imt_coupling_loss, results_arr)
+        self.assertEqual(results_recuperated_from_file.imt_bs_antenna_gain, results_arr)
+
+        results_recuperated_from_file = Results().load_from_dir(
+            self.results.output_directory, only_samples=["imt_bs_antenna_gain"]
+        )
+
+        self.assertEqual(results_recuperated_from_file.imt_coupling_loss, [])
+        self.assertEqual(results_recuperated_from_file.imt_bs_antenna_gain, results_arr)
 
     def test_get_most_recent_dirs(self):
         dir_2024_01_01_04 = "caminho_abs/prefixo_2024-01-01_04"