Bulkfraction engine for pyaeroval

pyaerocom/aeroval/bulkfraction_engine.py

@@ -0,0 +1,214 @@
+import logging
+
+from pyaerocom import ColocatedData
+from pyaerocom.aeroval._processing_base import HasColocator, ProcessingEngine
+from pyaerocom.aeroval.obsentry import ObsEntry
+from pyaerocom.aeroval.modelentry import ModelEntry
+from pyaerocom.aeroval.coldatatojson_engine import ColdataToJsonEngine
+
+
+from pyaerocom.colocation.colocator import Colocator
+from pyaerocom.colocation.colocation_setup import ColocationSetup
+
+
+logger = logging.getLogger(__name__)
+
+
+class BulkFractionEngine(ProcessingEngine, HasColocator):
+    def run(self, var_list: list[str] | str | None, model_name: str, obs_name: str):
+        self.sobs_cfg = self.cfg.obs_cfg.get_entry(obs_name)
+        self.smodel_cfg = self.cfg.model_cfg.get_entry(model_name)
+
+        if var_list is None:
+            var_list = self.sobs_cfg.obs_vars
+        elif isinstance(var_list, str):
+            var_list = [var_list]
+        elif not isinstance(var_list, list):
+            raise ValueError(f"invalid input for var_list: {var_list}.")
+
+        files_to_convert = []
+        for var_name in var_list:
+            bulk_vars = self._get_bulk_vars(var_name, self.sobs_cfg)
+
+            freq = self.sobs_cfg.ts_type
+            cd, fp = self._run_var(
+                model_name, obs_name, var_name, bulk_vars, freq, self.sobs_cfg, self.smodel_cfg
+            )
+
+            files_to_convert.append(fp)
+
+        engine = ColdataToJsonEngine(self.cfg)
+        engine.run(files_to_convert)
+
+    def _get_bulk_vars(self, var_name: str, cfg: ObsEntry) -> list:
+        bulk_vars = cfg.bulk_options
+        if var_name not in bulk_vars:
+            raise KeyError(f"Could not find bulk vars entry for {var_name}")
+
+        if len(bulk_vars[var_name].vars) != 2:
+            raise ValueError(
+                f"(Only) 2 entries must be present for bulk vars to calculate fraction for {var_name}. Found {bulk_vars[var_name]}"
+            )
+
+        return bulk_vars[var_name].vars
+
+    def _run_var(
+        self,
+        model_name: str,
+        obs_name: str,
+        var_name: str,
+        bulk_vars: list,
+        freq: str,
+        obs_entry: ObsEntry,
+        model_entry: ModelEntry,
+    ) -> tuple[ColocatedData, str]:
+        model_exists = obs_entry.bulk_options[var_name].model_exists
+
+        cols = self.get_colocators(bulk_vars, var_name, freq, model_name, obs_name, model_exists)
+
+        coldatas = []
+        for col in cols:
+            if len(list(col.keys())) != 1:
+                raise ValueError(
+                    "Found more than one colocated object when trying to run bulk variable"
+                )
+            bv = list(col.keys())[0]
+            coldatas.append(col[bv].run(bv))
+
+        num_name, denum_name = bulk_vars[0], bulk_vars[1]
+        num_col = cols[0][num_name].run(num_name)
+        denum_col = cols[1][denum_name].run(denum_name)
+
+        model_num_name, model_denum_name = self._get_model_var_names(
+            var_name, bulk_vars, model_exists, model_entry
+        )
+
+        cd = self._combine_coldatas(
+            num_col[model_num_name][num_name],
+            denum_col[model_denum_name][denum_name],
+            var_name,
+            obs_entry,
+        )
+
+        num_colocator = cols[0][num_name]
+
+        fp = cd.to_netcdf(
+            out_dir=num_colocator.output_dir,
+            savename=cd._aerocom_savename(
+                var_name,
+                obs_name,
+                var_name,
+                model_name,
+                num_colocator.get_start_str(),
+                num_colocator.get_stop_str(),
+                freq,
+                num_colocator.colocation_setup.filter_name,
+                None,  # cd.data.attrs["vert_code"],
+            ),
+        )
+        return cd, fp
+
+    def _combine_coldatas(
+        self,
+        num_coldata: ColocatedData,
+        denum_coldata: ColocatedData,
+        var_name: str,
+        obs_entry: ObsEntry,
+    ) -> ColocatedData:
+        mode = obs_entry.bulk_options[var_name].mode
+        model_exists = obs_entry.bulk_options[var_name].model_exists
+        units = obs_entry.bulk_options[var_name].units
+
+        if mode == "fraction":
+            new_data = num_coldata.data / denum_coldata.data
+
+        elif mode == "product":
+            new_data = num_coldata.data * denum_coldata.data
+        else:
+            raise ValueError(f"Mode must be either fraction of product, and not {mode}")
+        if model_exists:
+            # TODO: Unsure if this works!!!
+            new_data[1] = num_coldata.data[1].where(new_data[1])
+
+        cd = ColocatedData(new_data)
+
+        cd.data.attrs = num_coldata.data.attrs
+        cd.data.attrs["var_name"] = [var_name, var_name]
+        cd.data.attrs["var_units"] = [units, units]
+        cd.metadata["var_name_input"] = [var_name, var_name]
+        return cd
+
+    def _get_model_var_names(
+        self, var_name: str, bulk_vars: list[str], model_exists: bool, model_entry: ModelEntry
+    ) -> tuple[str]:
+        num_name, denum_name = bulk_vars[0], bulk_vars[1]
+        if model_exists:
+            num_name, denum_name = var_name, var_name
+
+        model_use_vars = model_entry.model_use_vars
+        if model_use_vars != {}:
+            num_name, denum_name = model_use_vars[num_name], model_use_vars[denum_name]
+
+        return num_name, denum_name
+
+    def get_colocators(
+        self,
+        bulk_vars: list,
+        var_name: str,
+        freq: str,
+        model_name: str = None,
+        obs_name: str = None,
+        model_exists: bool = False,
+    ) -> list[dict[str | Colocator]]:
+        """
+        Instantiate colocation engine
+
+        Parameters
+        ----------
+        model_name : str, optional
+            name of model. The default is None.
+        obs_name : str, optional
+            name of obs. The default is None.
+
+        Returns
+        -------
+        Colocator
+
+        """
+        col_cfg = {**self.cfg.colocation_opts.model_dump()}
+        outdir = self.cfg.path_manager.get_coldata_dir()
+        col_cfg["basedir_coldata"] = outdir
+
+        if model_name:
+            mod_cfg = self.cfg.get_model_entry(model_name)
+            col_cfg["model_cfg"] = mod_cfg
+
+            # Hack and at what lowlevel_helpers's import_from was doing
+            for key, val in mod_cfg.items():
+                if key in ColocationSetup.model_fields:
+                    col_cfg[key] = val
+        if obs_name:
+            obs_cfg = self.cfg.get_obs_entry(obs_name)
+            pyaro_config = obs_cfg["obs_config"] if "obs_config" in obs_cfg else None
+            col_cfg["obs_config"] = pyaro_config
+
+            # Hack and at what lowlevel_helpers's import_from was doing
+            for key, val in obs_cfg.model_dump().items():
+                if key in ColocationSetup.model_fields:
+                    col_cfg[key] = val
+
+            col_cfg["add_meta"].update(diurnal_only=self.cfg.get_obs_entry(obs_name).diurnal_only)
+        cols = []
+        col_cfg["ts_type"] = freq
+        for bulk_var in bulk_vars:
+            col_cfg["obs_vars"] = bulk_var
+            if model_exists:
+                col_cfg["model_use_vars"] = {
+                    bulk_var: var_name,
+                }
+            col_stp = ColocationSetup(**col_cfg)
+            col = Colocator(col_stp)
+
+            cols.append({bulk_var: col})
+
+        return cols

pyaerocom/aeroval/data/var_scale_colmap.ini

@@ -440,4 +440,29 @@ colmap = coolwarm
 
 [vcdno2]
 scale = [0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120]
+colmap = coolwarm
+
+
+[fractioneBCbb]
+scale = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
+colmap = coolwarm
+
+[fractioneBCff]
+scale = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
+colmap = coolwarm
+
+[concebc]
+scale = [0.0, 0.05, 0.1, 0.15, 0.2, 0.25, 0.3, 0.35, 0.4, 0.5, 1.0, 2.0]
+colmap = coolwarm
+
+[concebcem]
+scale = [0.0, 0.05, 0.1, 0.15, 0.2, 0.25, 0.3, 0.35, 0.4, 0.5, 1.0, 2.0]
+colmap = coolwarm
+
+[fractioneBCbbem]
+scale = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
+colmap = coolwarm
+
+[fractioneBCffem]
+scale = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
 colmap = coolwarm

pyaerocom/aeroval/data/var_web_info.ini

@@ -847,3 +847,52 @@ category = Vertical column density
 menu_name = VCD CO
 vertical_type = 3D
 category = Vertical column density
+
+
+; # Fraction tests
+; [pmfraction]
+; menu_name = PM Fraction
+; vertical_type = 3D
+; category = Particle ratio
+
+; [pmfraction2]
+; menu_name = PM Fraction 2
+; vertical_type = 3D
+; category = Particle ratio
+
+[concwetrdn]
+menu_name = RDN in Precip
+vertical_type = 3D
+category = Particle concentrations
+
+
+# EC
+; [fractionECFine]
+; menu_name = EC Fine Fraction
+; vertical_type = 3D
+; category = Particle ratio
+
+; [fractionECCoarse]
+; menu_name = EC Fine Fraction
+; vertical_type = 3D
+; category = Particle ratio
+
+; [fractionEC]
+; menu_name = EC Fraction
+; vertical_type = 3D
+; category = Particle ratio
+
+[fractioneBCbb]
+menu_name = eBC Residential Fraction
+vertical_type = 3D
+category = Particle ratio
+
+[fractioneBCff]
+menu_name = eBC Non-Residential Fraction
+vertical_type = 3D
+category = Particle ratio
+
+[concebc]
+menu_name = Equivalent BC
+vertical_type = 3D
+category = Particle concentrations

pyaerocom/aeroval/experiment_processor.py

@@ -8,6 +8,7 @@
 from pyaerocom.aeroval.helpers import delete_dummy_model, make_dummy_model
 from pyaerocom.aeroval.modelmaps_engine import ModelMapsEngine
 from pyaerocom.aeroval.superobs_engine import SuperObsEngine
+from pyaerocom.aeroval.bulkfraction_engine import BulkFractionEngine
 
 logger = logging.getLogger(__name__)
 
@@ -65,6 +66,10 @@ def _run_single_entry(self, model_name, obs_name, var_list):
                 engine = ColdataToJsonEngine(self.cfg)
                 engine.run(files_to_convert)
 
+        elif ocfg.is_bulk:
+            engine = BulkFractionEngine(self.cfg)
+            engine.run(var_list, model_name, obs_name)
+
         else:
             # If a var_list is given, only run on the obs networks which contain that variable
             if var_list: