import odc 1.9 fixes from branch

Author: colourmeamused

.github/workflows/test.yml

@@ -19,11 +19,6 @@ jobs:
     - name: Pre-pull layers
       run: docker compose pull
 
-    - name: Activate Docker cache
-      uses: satackey/action-docker-layer-caching@v0.0.8
-      # Ignore the failure of a step and avoid terminating the job.
-      continue-on-error: true
-
     - name: Build dea-conflux image
       run: |
         docker compose build

dea_conflux/__main__.py

@@ -19,7 +19,6 @@
 import geopandas as gpd
 from datacube.ui import click as ui
 from rasterio.errors import RasterioIOError
-
 import dea_conflux.__version__
 import dea_conflux.db
 import dea_conflux.drill
@@ -70,7 +69,7 @@ def id_field_values_is_unique(shapefile_path: str, id_field) -> bool:
     has_s3 = "s3" in gpd.io.file._VALID_URLS
     gpd.io.file._VALID_URLS.discard("s3")
     logger.info(f"Attempting to read {shapefile_path} to check id field.")
-    gdf = gpd.read_file(shapefile_path, driver="ESRI Shapefile")
+    gdf = gpd.read_file(shapefile_path)
     if has_s3:
         gpd.io.file._VALID_URLS.add("s3")
     return len(set(gdf[id_field])) == len(gdf)
@@ -176,7 +175,7 @@ def load_and_reproject_shapefile(
     has_s3 = "s3" in gpd.io.file._VALID_URLS
     gpd.io.file._VALID_URLS.discard("s3")
     logger.info(f"Attempting to read {shapefile} to load polgyons.")
-    shapefile = gpd.read_file(shapefile, driver="ESRI Shapefile")
+    shapefile = gpd.read_file(shapefile)
     if has_s3:
         gpd.io.file._VALID_URLS.add("s3")
 
@@ -332,7 +331,6 @@ def run_one(
     Run dea-conflux on one scene.
     """
     logging_setup(verbose)
-
     # Read the plugin as a Python module.
     plugin = run_plugin(plugin)
     logger.info(f"Using plugin {plugin.__file__}")
@@ -703,7 +701,7 @@ def nrt_run_from_queue(
 
             logger.info(f"Processing {id_} ({i + 1}/{len(ids)})")
             logger.info(f"Processing {id_} ({i + 1}/{len(ids)})")
-            
+
             # if id_ is uuid
             if len(id_) != 36:
                 import json
@@ -1384,7 +1382,7 @@ def db_to_csv(
     has_s3 = "s3" in gpd.io.file._VALID_URLS
     gpd.io.file._VALID_URLS.discard("s3")
     logger.info(f"Attempting to read {shapefile} to load polgyons.")
-    shapefile = gpd.read_file(shapefile, driver="ESRI Shapefile")
+    shapefile = gpd.read_file(shapefile)
     if has_s3:
         gpd.io.file._VALID_URLS.add("s3")
 

dea_conflux/drill.py

@@ -389,12 +389,15 @@ def filter_dataset(dss, shapefile, worker_num=1):
     -------
     filtered_datasets: [str]
     """
-    with multiprocessing.Pool(processes=worker_num) as pool:
-        filtered_datasets = list(
-            tqdm.tqdm(pool.imap(partial(polygon_in_dataset, shapefile=shapefile), dss))
-        )
+    if __name__ == "__main__":
+        with multiprocessing.Pool(processes=worker_num) as pool:
+            filtered_datasets = list(
+                tqdm.tqdm(
+                    pool.imap(partial(polygon_in_dataset, shapefile=shapefile), dss)
+                )
+            )
 
-    return [e for e in filtered_datasets if e]
+        return [e for e in filtered_datasets if e]
 
 
 def polygon_in_dataset(ds, shapefile):

dea_conflux/stack.py

@@ -199,8 +199,10 @@ def remove_timeseries_with_duplicated(df: pd.DataFrame) -> pd.DataFrame:
     df = df.drop_duplicates("DAY", keep="first")
 
     if "date" in df.columns and len(df) > 1:
-        df['TIMEDIFF'] = pd.to_datetime(df['date'].shift(-1)) - pd.to_datetime(df['date'])
-        df = df[~(df['TIMEDIFF'] < timedelta(seconds=60))]
+        df["TIMEDIFF"] = pd.to_datetime(df["date"].shift(-1)) - pd.to_datetime(
+            df["date"]
+        )
+        df = df[~(df["TIMEDIFF"] < timedelta(seconds=60))]
         df = df.drop(columns=["TIMEDIFF"])
 
     # Remember to remove the temp column day in result_df
@@ -290,7 +292,6 @@ def save_df_as_csv(single_polygon_df, feature_id, outpath, remove_duplicated_dat
     single_polygon_df.drop(
         ["overall_veg_num", "veg_areas", "index"], axis=1, inplace=True
     )
-
     if not outpath.startswith("s3://"):
         os.makedirs(Path(filename).parent, exist_ok=True)
     with fsspec.open(filename, "w") as f:
@@ -315,9 +316,9 @@ def stack_wit_tooling_to_single_file(
 
     verbose : bool
     """
+
     polygon_df_list = []
     logger.info("Reading...")
-
     # Note: the stack_wit_tooling_to_single_file() input files are CSV file, which generate by save_df_as_csv()
     # then we assume they already had the norm_pv, norm_npv, norm_bs there.
     with tqdm(total=len(paths)) as bar:
@@ -391,7 +392,8 @@ def stack_wit_tooling(
     """
     wit_df_list = []
     logger.info("Reading...")
-
+    # Note: the stack_wit_tooling_to_single_file() input files are CSV file, which generate by save_df_as_csv()
+    # then we assume they already had the norm_pv, norm_npv, norm_bs there.
     with tqdm(total=len(paths)) as bar:
         with concurrent.futures.ThreadPoolExecutor(max_workers=32) as executor:
             wit_df_list = []
@@ -424,7 +426,6 @@ def stack_wit_tooling(
 
     # delete the temp result to release RAM
     del wit_result
-
     with tqdm(total=len(feature_ids)) as bar:
         with concurrent.futures.ThreadPoolExecutor(
             max_workers=multiprocessing.cpu_count()

examples/wit_ls9.conflux.py

@@ -73,7 +73,7 @@ def transform(inputs: xr.Dataset) -> xr.Dataset:
     # not apply poly_raster cause we will do it before summarise
 
     open_water = wo_ds.water & (1 << 7) > 0
-    
+
     # Thresholding
     # set wet pixels where not masked and above threshold of -350
     wet = tcw.where(mask) > -350

setup.py

@@ -20,10 +20,10 @@
             "pyarrow",
             "fsspec",
             "s3fs",
-            "moto[s3]",
+            "moto[all]",
             "tqdm",
             "SQLAlchemy",
-            # "python-geohash",
+            "python-geohash",
         ],
         entry_points={
             "console_scripts": ["dea-conflux=dea_conflux.__main__:main"],

tests/data/sum_pv_wet.conflux.py

@@ -1,19 +1,21 @@
 import xarray as xr
 
-product_name = 'sum_pv_wet'
-version = '0.0.1'
+product_name = "sum_pv_wet"
+version = "0.0.1"
 
 input_products = {
-    'ga_ls_wo_3': ['water'],
-    'ga_ls_fc_3': ['PV'],
+    "ga_ls_wo_3": ["water"],
+    "ga_ls_fc_3": ["PV"],
 }
 
 
 def transform(inputs: xr.Dataset) -> xr.Dataset:
-    return xr.Dataset({
-        'water': inputs.water == 128,
-        'pv': inputs.PV > 0.5,
-    })
+    return xr.Dataset(
+        {
+            "water": inputs.water == 128,
+            "pv": inputs.PV > 0.5,
+        }
+    )
 
 
 def summarise(inputs: xr.Dataset) -> xr.Dataset:

tests/data/sum_wet.conflux.py

@@ -1,12 +1,12 @@
 import xarray as xr
 
-product_name = 'sum_wet'
-version = '0.0.1'
+product_name = "sum_wet"
+version = "0.0.1"
 resolution = (-30, 30)
-output_crs = 'EPSG:3577'
+output_crs = "EPSG:3577"
 
 input_products = {
-    'ga_ls_wo_3': ['water'],
+    "ga_ls_wo_3": ["water"],
 }
 
 

tests/data/sum_wet_c3.conflux.py

@@ -1,13 +1,13 @@
 import xarray as xr
 
-product_name = 'sum_wet_c3'
-version = '0.0.1'
+product_name = "sum_wet_c3"
+version = "0.0.1"
 resolution = (-30, 30)
-output_crs = 'EPSG:3577'
-resampling = 'nearest'
+output_crs = "EPSG:3577"
+resampling = "nearest"
 
 input_products = {
-    'ga_ls_wo_3': ['water'],
+    "ga_ls_wo_3": ["water"],
 }
 
 

tests/data/sum_wet_missing_transform.conflux.py

@@ -1,10 +1,10 @@
 import xarray as xr
 
-product_name = 'sum_wet'
-version = '0.0.1'
+product_name = "sum_wet"
+version = "0.0.1"
 
 input_products = {
-    'ga_ls_wo_3': ['water'],
+    "ga_ls_wo_3": ["water"],
 }