feat: python script to deploy OGC App Packages

ogc-application-packages/cwl_dag.json

@@ -0,0 +1,74 @@
+{
+  "executionUnit": {
+    "image": "ghcr.io/unity-sds/unity-sps/sps-docker-cwl:2.1.0",
+    "type": "docker"
+  },
+  "processDescription": {
+    "description": "This process executes any CWL workflow.",
+    "id": "cwl_dag",
+    "inputs": {
+      "cwl_args": {
+        "description": "The URL of the CWL workflow's YAML parameters file",
+        "maxOccurs": 1,
+        "minOccurs": 1,
+        "schema": {
+          "format": "uri",
+          "type": "string"
+        },
+        "title": "CWL Workflow Parameters URL"
+      },
+      "cwl_workflow": {
+        "description": "The URL of the CWL workflow",
+        "maxOccurs": 1,
+        "minOccurs": 1,
+        "schema": {
+          "format": "uri",
+          "type": "string"
+        },
+        "title": "CWL Workflow URL"
+      },
+      "request_cpu": {
+        "description": "The number of CPU cores requested for the job",
+        "maxOccurs": 1,
+        "minOccurs": 1,
+        "schema": {
+          "type": "string"
+        },
+        "title": "Requested CPU"
+      },
+      "request_memory": {
+        "default": "8Gi",
+        "description": "The amount of memory requested for the job",
+        "maxOccurs": 1,
+        "minOccurs": 1,
+        "schema": {
+          "type": "string"
+        },
+        "title": "Requested Memory"
+      },
+      "request_storage": {
+        "description": "The amount of storage requested for the job",
+        "maxOccurs": 1,
+        "minOccurs": 1,
+        "schema": {
+          "type": "string"
+        },
+        "title": "Requested Storage"
+      }
+    },
+    "jobControlOptions": [
+      "async-execute"
+    ],
+    "outputs": {
+      "result": {
+        "description": "The result of the SBG Preprocess Workflow execution",
+        "schema": {
+          "$ref": "some-ref"
+        },
+        "title": "Process Result"
+      }
+    },
+    "title": "Generic CWL Process",
+    "version": "1.0.0"
+  }
+}

ogc-application-packages/karpenter_test.json

@@ -0,0 +1,36 @@
+{
+  "executionUnit": {
+    "image": "busybox",
+    "type": "docker"
+  },
+  "processDescription": {
+    "description": "This process tests Karpenter node provisioning with different instance types.",
+    "id": "karpenter_test",
+    "inputs": {
+      "placeholder": {
+        "default": 1,
+        "description": "A placeholder parameter",
+        "maxOccurs": 1,
+        "minOccurs": 1,
+        "schema": {
+          "type": "integer"
+        },
+        "title": "Placeholder"
+      }
+    },
+    "jobControlOptions": [
+      "async-execute"
+    ],
+    "outputs": {
+      "result": {
+        "description": "The result of the Karpenter test execution",
+        "schema": {
+          "$ref": "some-ref"
+        },
+        "title": "Process Result"
+      }
+    },
+    "title": "Karpenter Test Process",
+    "version": "1.0.0"
+  }
+}

ogc-application-packages/sbg_preprocess_cwl_dag.json

@@ -0,0 +1,48 @@
+{
+  "executionUnit": {
+    "image": "ghcr.io/unity-sds/unity-sps/sps-docker-cwl:2.2.0-rc-1",
+    "type": "docker"
+  },
+  "processDescription": {
+    "description": "This process executes the SBG Preprocess Workflow using CWL.",
+    "id": "sbg_preprocess_cwl_dag",
+    "inputs": {
+      "cwl_args": {
+        "default": "https://raw.githubusercontent.com/unity-sds/sbg-workflows/main/preprocess/sbg-preprocess-workflow.dev.yml",
+        "description": "The SBG Pre-process YAML parameters URL",
+        "maxOccurs": 1,
+        "minOccurs": 1,
+        "schema": {
+          "format": "uri",
+          "type": "string"
+        },
+        "title": "CWL Workflow Parameters"
+      },
+      "cwl_workflow": {
+        "default": "https://raw.githubusercontent.com/unity-sds/sbg-workflows/main/preprocess/sbg-preprocess-workflow.cwl",
+        "description": "The SBG Pre-process CWL workflow URL",
+        "maxOccurs": 1,
+        "minOccurs": 1,
+        "schema": {
+          "format": "uri",
+          "type": "string"
+        },
+        "title": "CWL Workflow"
+      }
+    },
+    "jobControlOptions": [
+      "async-execute"
+    ],
+    "outputs": {
+      "result": {
+        "description": "The result of the SBG Preprocess Workflow execution",
+        "schema": {
+          "$ref": "some-ref"
+        },
+        "title": "Process Result"
+      }
+    },
+    "title": "SBG Preprocess CWL Workflow",
+    "version": "1.0.0"
+  }
+}

pyproject.toml

@@ -44,7 +44,9 @@ test = [
     "kubernetes==29.0.0",
     "boto3==1.34.46",
     "backoff==2.2.1",
-    "apache-airflow-providers-cncf-kubernetes==8.4.1"
+    "apache-airflow-providers-cncf-kubernetes==8.4.1",
+    "unity_sps_ogc_processes_api_python_client @ git+https://github.com/unity-sds/unity-sps-ogc-processes-api-client-python.git@129-arbitrary-parameters#egg=129-arbitrary-parameters"
+
 ]
 experiment = []
 lambda-airflow-dag-trigger = [

utils/deploy_ogc_app_packages.py

@@ -0,0 +1,93 @@
+import argparse
+import json
+import logging
+import os
+
+import unity_sps_ogc_processes_api_python_client
+from unity_sps_ogc_processes_api_python_client.models.ogcapppkg import Ogcapppkg
+from unity_sps_ogc_processes_api_python_client.rest import ApiException
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+
+
+def register_process(api_instance, proc, ogcapppkg_instance):
+    """
+    Register a process with the OGC API.
+
+    Args:
+        api_instance: The API client instance.
+        proc: The process identifier.
+        ogcapppkg_instance: The Ogcapppkg instance containing the process description.
+
+    Returns:
+        None
+    """
+    try:
+        # Deploy a process
+        api_instance.deploy_processes_post(w=proc, ogcapppkg=ogcapppkg_instance)
+        logging.info(f"Successfully registered process: {proc}")
+    except ApiException as e:
+        logging.error(f"Exception when calling DRUApi->deploy_processes_post for process {proc}: {e}")
+    except Exception as e:
+        logging.error(f"Unexpected error for process {proc}: {e}")
+
+
+def main():
+    """
+    Main function to deploy processes to the OGC API.
+
+    Parses command-line arguments, reads process descriptions from JSON files,
+    and registers each process with the OGC API.
+
+    Args:
+        None
+
+    Returns:
+        None
+    """
+    parser = argparse.ArgumentParser(description="Deploy processes to OGC API")
+    parser.add_argument("ogc_api_processes", help="OGC API Processes URL")
+    parser.add_argument(
+        "ogc_app_packages_dir", help="Directory containing JSON files with application packages"
+    )
+    args = parser.parse_args()
+
+    OGC_API_PROCESSES = args.ogc_api_processes
+    ogc_app_packages_dir = args.ogc_app_packages_dir
+
+    # Configure the API client
+    configuration = unity_sps_ogc_processes_api_python_client.Configuration(host=OGC_API_PROCESSES)
+
+    with unity_sps_ogc_processes_api_python_client.ApiClient(configuration) as api_client:
+        api_instance = unity_sps_ogc_processes_api_python_client.DRUApi(api_client)
+
+        # Iterate through all JSON files in the specified directory
+        for filename in os.listdir(ogc_app_packages_dir):
+            if filename.endswith(".json"):
+                json_file = os.path.join(ogc_app_packages_dir, filename)
+                try:
+                    # Read the process description from the JSON file
+                    with open(json_file, "r") as f:
+                        process_data = json.load(f)
+
+                    # Extract the process ID from the JSON data
+                    proc = process_data.get("processDescription", {}).get("id")
+                    if not proc:
+                        logging.error(f"Process ID not found in JSON file: {json_file}")
+                        continue
+
+                    # Create an instance of Ogcapppkg from the JSON data
+                    ogcapppkg_instance = Ogcapppkg.from_dict(process_data)
+                    logging.info(f"Registering process: {proc}")
+                    register_process(api_instance, proc, ogcapppkg_instance)
+                except FileNotFoundError:
+                    logging.error(f"JSON file not found: {json_file}")
+                except json.JSONDecodeError:
+                    logging.error(f"Error decoding JSON file: {json_file}")
+                except Exception as e:
+                    logging.error(f"Unexpected error processing file {json_file}: {e}")
+
+
+if __name__ == "__main__":
+    main()