Further modularised the level4

Author: tebadi

odc/stats/plugins/l34_utils/lc_lifeform.py

@@ -0,0 +1,22 @@
+from odc.stats._algebra import expr_eval
+import xarray as xr
+
+
+def lifeform(self, xx: xr.Dataset):
+
+    # 113 ----> 1 woody
+    # 114 ----> 2 herbaceous
+    lifeform_mask = expr_eval(
+        "where(a==113, 1, a)",
+        {"a": xx.woody_cover.data},
+        name="mark_lifeform",
+        dtype="uint8",
+    )
+    lifeform_mask = expr_eval(
+        "where(a==114, 2, a)",
+        {"a": lifeform_mask},
+        name="mark_lifeform",
+        dtype="uint8",
+    )
+
+    return lifeform_mask

odc/stats/plugins/l34_utils/lc_water_seasonality.py

@@ -0,0 +1,40 @@
+import xarray as xr
+
+from odc.stats._algebra import expr_eval
+from . import utils
+
+WATER_FREQ_NODATA = -999
+
+
+def water_seasonality(self, xx: xr.Dataset):
+    # >= 3 months ----> 1  Semi-permanent or permanent
+    # < 3 months  ----> 2 Temporary or seasonal
+
+    water_season_mask = expr_eval(
+        "where((a>watseas_trh)&(a<=12), 100, a)",
+        {"a": xx.water_frequency.data},
+        name="mark_water_season",
+        dtype="uint8",
+        **{"watseas_trh": self.water_seasonality_threshold},
+    )
+    water_season_mask = expr_eval(
+        "where((a<=watseas_trh)&(a<=12), 200, a)",
+        {"a": water_season_mask},
+        name="mark_water_season",
+        dtype="uint8",
+        **{"watseas_trh": self.water_seasonality_threshold},
+    )
+    water_season_mask = expr_eval(
+        "where((a==watersea_nodata), 255, a)",
+        {"a": water_season_mask},
+        name="mark_water_season",
+        dtype="uint8",
+        **{
+            "watseas_trh": self.water_seasonality_threshold,
+            "watersea_nodata": WATER_FREQ_NODATA,
+        },
+    )
+    mapping = {100: 1, 200: 2}
+    water_season_mask = utils.apply_mapping(water_season_mask, mapping)
+
+    return water_season_mask

odc/stats/plugins/lc_level34.py

@@ -7,11 +7,11 @@
 import numpy as np
 import xarray as xr
 
-from odc.stats._algebra import expr_eval
 from ._registry import StatsPluginInterface, register
 
 from .l34_utils import (
     l4_water_persistence,
+    lc_water_seasonality,
     l4_veg_cover,
     lc_level3,
     l4_cultivated,
@@ -20,12 +20,11 @@
     l4_surface,
     l4_bare_gradation,
     l4_water,
-    utils,
+    lc_lifeform,
 )
 
 
 NODATA = 255
-WATER_FREQ_NODATA = -999
 
 
 class StatsLccsLevel4(StatsPluginInterface):
@@ -66,74 +65,15 @@ def native_transform(self, xx):
     def fuser(self, xx):
         return xx
 
-    # @staticmethod
-    # def apply_mapping(data, class_mapping):
-    #     for o_val, n_val in class_mapping.items():
-    #         data = xr.where(data == o_val, n_val, data)
-    #     return data
-
-    def define_life_form(self, xx: xr.Dataset):
-
-        # 113 ----> 1 woody
-        # 114 ----> 2 herbaceous
-        lifeform_mask = expr_eval(
-            "where(a==113, 1, a)",
-            {"a": xx.woody_cover.data},
-            name="mark_lifeform",
-            dtype="uint8",
-        )
-        lifeform_mask = expr_eval(
-            "where(a==114, 2, a)",
-            {"a": lifeform_mask},
-            name="mark_lifeform",
-            dtype="uint8",
-        )
-
-        return lifeform_mask
-
-    def define_water_seasonality(self, xx: xr.Dataset):
-        # >= 3 months ----> 1  Semi-permanent or permanent
-        # < 3 months  ----> 2 Temporary or seasonal
-
-        water_season_mask = expr_eval(
-            "where((a>watseas_trh)&(a<=12), 100, a)",
-            {"a": xx.water_frequency.data},
-            name="mark_water_season",
-            dtype="uint8",
-            **{"watseas_trh": self.water_seasonality_threshold},
-        )
-        water_season_mask = expr_eval(
-            "where((a<=watseas_trh)&(a<=12), 200, a)",
-            {"a": water_season_mask},
-            name="mark_water_season",
-            dtype="uint8",
-            **{"watseas_trh": self.water_seasonality_threshold},
-        )
-        water_season_mask = expr_eval(
-            "where((a==watersea_nodata), 255, a)",
-            {"a": water_season_mask},
-            name="mark_water_season",
-            dtype="uint8",
-            **{
-                "watseas_trh": self.water_seasonality_threshold,
-                "watersea_nodata": WATER_FREQ_NODATA,
-            },
-        )
-        mapping = {100: 1, 200: 2}
-        water_season_mask = utils.apply_mapping(water_season_mask, mapping)
-
-        return water_season_mask
-
     def reduce(self, xx: xr.Dataset) -> xr.Dataset:
 
         intertidal_mask, level3 = lc_level3.lc_level3(xx)
 
         # Vegetation cover
         veg_cover = l4_veg_cover.canopyco_veg_con(xx, self.veg_threshold)
-        # # Define mapping from current output to expected a3 output
-        # veg_cover = utils.apply_mapping(veg_cover, self.veg_mapping)
+
         # Define life form
-        lifeform = self.define_life_form(xx)
+        lifeform = lc_lifeform.lifeform(xx)
 
         # Apply cultivated Level-4 classes (1-18)
         l4 = l4_cultivated.lc_l4_cultivated(
@@ -147,19 +87,13 @@ def reduce(self, xx: xr.Dataset) -> xr.Dataset:
         bare_gradation = l4_bare_gradation.bare_gradation(
             xx, self.bare_threshold, veg_cover
         )
-        # # Apply bare gradation expected output classes
-        # bare_gradation = utils.apply_mapping(bare_gradation, self.bs_mapping)
 
         # Water persistence
         water_persistence = l4_water_persistence.water_persistence(
             xx, self.watper_threshold
         )
-        # # Apply water persistence expcted classes
-        # water_persistence = utils.apply_mapping(
-        #     water_persistence, self.waterper_wat_mapping
-        # )
 
-        water_seasonality = self.define_water_seasonality(xx)
+        water_seasonality = lc_water_seasonality.water_seasonality(xx)
 
         l4 = l4_natural_aquatic.natural_auquatic_veg(
             l4, lifeform, veg_cover, water_seasonality