dev to main BLOCKED by issue #252

.github/workflows/cicd.yml

@@ -7,9 +7,7 @@ permissions:
 on:
   push:
     branches:
-      - main
       - dev
-      - production
   pull_request:
     branches:
       - main
@@ -36,12 +34,8 @@ jobs:
       - name: Set the environment based on the branch
         id: define_environment
         run: |
-          if [ "${{ github.ref }}" = "refs/heads/main" ]; then
-            echo "env_name=staging" >> $GITHUB_OUTPUT
-          elif [ "${{ github.ref }}" = "refs/heads/dev" ]; then
+          if [ "${{ github.ref }}" = "refs/heads/dev" ]; then
             echo "env_name=development" >> $GITHUB_OUTPUT
-          elif [ "${{ github.ref }}" = "refs/heads/production" ]; then
-            echo "env_name=production" >> $GITHUB_OUTPUT
           fi
       - name: Print the environment
         run: echo "The environment is ${{ steps.define_environment.outputs.env_name }}"
@@ -71,11 +65,6 @@ jobs:
           role-session-name: "veda-airflow-github-${{ needs.define-environment.outputs.env_name }}-deployment"
           aws-region: us-west-2
 
-      - name: Run MWAA deployment
-        uses: "./.github/actions/terraform-deploy"
-        with:
-          env_aws_secret_name: ${{ secrets.ENV_AWS_SECRET_NAME }}
-
       - name: Run SM2A deployment
         # Flag to deploy SM2A
         if: ${{ vars.DEPLOY_SM2A }} = "true"

dags/automated_transformation/README.md

@@ -0,0 +1,5 @@
+## Information about the folder
+This folder contains `automation DAG`. It contains the DAG which is essential for transforming the given dataset into COGs.
+The folder consists of two files-:
+- `automation_dag.py` - It contains the DAG architecture.
+- `transformation_pipeline.py` - It contains multiple functions to achieve the transformation of a given netCDF file to a COG. The file fetches checks whether the `transformation plugins` exist on `S3 bucket`, fetches the function and uses it for transformation. The statistics for the `netCDF` and `COG` files are also calculated and stored in a `JSON` file and pushed to `S3` along with the `transformed COG`.

dags/automated_transformation/automation_dag.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from airflow import DAG
+from airflow.decorators import task
+from airflow.models.param import Param
+from airflow.operators.dummy_operator import DummyOperator
+from slack_notifications import slack_fail_alert
+
+DAG_ID = "automate-cog-transformation"
+
+# Custom validation function
+
+
+dag_run_config = {
+    "data_acquisition_method": Param(
+        "s3", enum=["s3"]
+    ),  # To add Other protocols (HTTP, SFTP...)
+    "plugins_uri": Param(
+        "https://raw.githubusercontent.com/US-GHG-Center/ghgc-docs/refs/heads/main/",
+        type="string",
+    ),
+    "raw_data_bucket": "ghgc-data-store-develop",
+    "raw_data_prefix": Param(
+        "delivery/gpw",
+        type="string",
+        pattern="^[^/].*[^/]$",
+    ),
+    "dest_data_bucket": "ghgc-data-store-develop",
+    "data_prefix": Param("transformed_cogs", type="string", pattern="^[^/].*[^/]$"),
+    "collection_name": "gpw",
+    "nodata": Param(-9999, type="number"),
+    "ext": Param(".tif", type="string", pattern="^\\..*$"),
+}
+
+with DAG(
+    dag_id=DAG_ID,
+    schedule=None,
+    catchup=False,
+    tags=["Transformation", "Report"],
+    params=dag_run_config,
+    on_failure_callback=slack_fail_alert,
+) as dag:
+    start = DummyOperator(task_id="start", dag=dag)
+    end = DummyOperator(task_id="end", dag=dag)
+
+    @task
+    def check_function_exists(ti):
+        from dags.automated_transformation.transformation_pipeline import (
+            download_python_file,
+        )
+
+        config = ti.dag_run.conf
+        folder_name = "data_transformation_plugins"
+        file_name = f'{config.get("collection_name")}_transformation.py'
+        try:
+            plugin_url = f"{config['plugins_uri'].strip('/')}/{folder_name}/{file_name}"
+            download_python_file(uri=plugin_url, check_exist=True)
+            return f"The {file_name} exists in {folder_name} in this URL {plugin_url}."
+        except Exception as e:
+            raise Exception(f"Error checking file existence: {e}")
+
+    @task
+    def discover_files(ti):
+        from dags.automated_transformation.transformation_pipeline import (
+            get_all_s3_keys,
+        )
+
+        config = ti.dag_run.conf.copy()
+        bucket = config.get("raw_data_bucket")
+        model_name = config.get("raw_data_prefix")
+        ext = config.get("ext")  # .nc as well
+        generated_list = get_all_s3_keys(bucket, model_name, ext)
+        chunk_size = int(len(generated_list) / 900) + 1
+        return [
+            generated_list[i : i + chunk_size]
+            for i in range(0, len(generated_list), chunk_size)
+        ]
+
+    @task(max_active_tis_per_dag=1)
+    def process_files(file_url, **kwargs):
+        dag_run = kwargs.get("dag_run")
+        from dags.automated_transformation.transformation_pipeline import transform_cog
+
+        config = dag_run.conf.copy()
+        raw_bucket_name = config.get("raw_data_bucket")
+        dest_data_bucket = config.get("dest_data_bucket")
+        data_prefix = config.get("data_prefix")
+        nodata = config.get("nodata")
+        collection_name = config.get("collection_name")
+        print(f"The file I am processing is {file_url}")
+        print("len of files", len(file_url))
+        folder_name = "data_transformation_plugins"
+        file_name = f"{collection_name}_transformation.py"
+        plugin_url = f"{config['plugins_uri'].strip('/')}/{folder_name}/{file_name}"
+
+        file_status = transform_cog(
+            file_url,
+            plugin_url=plugin_url,
+            nodata=nodata,
+            raw_data_bucket=raw_bucket_name,
+            dest_data_bucket=dest_data_bucket,
+            data_prefix=data_prefix,
+            collection_name=collection_name,
+        )
+        return file_status
+
+    @task
+    def generate_report(reports, **kwargs):
+        dag_run = kwargs.get("dag_run")
+        collection_name = dag_run.conf.get("collection_name")
+        count, failed_files = 0, []
+        for report in reports:
+            if "failed" in report.values():
+                failed_files.append(report)
+            elif "success" in report.values():
+                count += 1
+
+        if failed_files:
+            raise Exception(f"Error generating COG file {failed_files}")
+        return {
+            "collection": collection_name,
+            "successes": count,
+            "failures": failed_files,
+        }
+
+    urls = start >> check_function_exists() >> discover_files()
+    report_data = process_files.expand(file_url=urls)
+    generate_report(reports=report_data) >> end