feat: added support for format

Author: JanssenBrm

app/platforms/base.py

@@ -1,6 +1,6 @@
 from abc import ABC, abstractmethod
 
-from app.schemas.enum import ProcessingStatusEnum
+from app.schemas.enum import OutputFormatEnum, ProcessingStatusEnum
 from app.schemas.unit_job import ServiceDetails
 
 from stac_pydantic import Collection
@@ -13,13 +13,16 @@ class BaseProcessingPlatform(ABC):
     """
 
     @abstractmethod
-    def execute_job(self, title: str, details: ServiceDetails, parameters: dict) -> str:
+    def execute_job(
+        self, title: str, details: ServiceDetails, parameters: dict, format: OutputFormatEnum
+    ) -> str:
         """
         Execute a processing job on the platform with the given service ID and parameters.
 
         :param title: The title of the job to be executed.
         :param details: The service details containing the service ID and application.
         :param parameters: The parameters required for the job execution.
+        :param format: Format of the output result.
         :return: Return the ID of the job that was created
         """
         pass

app/platforms/implementations/ogc_api_process.py

@@ -1,6 +1,6 @@
 from app.platforms.base import BaseProcessingPlatform
 from app.platforms.dispatcher import register_platform
-from app.schemas.enum import ProcessTypeEnum, ProcessingStatusEnum
+from app.schemas.enum import OutputFormatEnum, ProcessTypeEnum, ProcessingStatusEnum
 from app.schemas.unit_job import ServiceDetails
 from stac_pydantic import Collection
 
@@ -12,7 +12,9 @@ class OGCAPIProcessPlatform(BaseProcessingPlatform):
     This class handles the execution of processing jobs on the OGC API Process platform.
     """
 
-    def execute_job(self, title: str, details: ServiceDetails, parameters: dict) -> str:
+    def execute_job(
+        self, title: str, details: ServiceDetails, parameters: dict, format: OutputFormatEnum
+    ) -> str:
         raise NotImplementedError("OGC API Process job execution not implemented yet.")
 
     def get_job_status(

app/platforms/implementations/openeo.py

@@ -11,7 +11,7 @@
 
 from app.platforms.base import BaseProcessingPlatform
 from app.platforms.dispatcher import register_platform
-from app.schemas.enum import ProcessTypeEnum, ProcessingStatusEnum
+from app.schemas.enum import OutputFormatEnum, ProcessTypeEnum, ProcessingStatusEnum
 from app.schemas.unit_job import ServiceDetails
 from stac_pydantic import Collection
 
@@ -143,7 +143,9 @@ def _get_process_id(self, url: str) -> str:
 
         return process_id
 
-    def execute_job(self, title: str, details: ServiceDetails, parameters: dict) -> str:
+    def execute_job(
+        self, title: str, details: ServiceDetails, parameters: dict, format: OutputFormatEnum
+    ) -> str:
         try:
             process_id = self._get_process_id(details.application)
 
@@ -156,7 +158,7 @@ def execute_job(self, title: str, details: ServiceDetails, parameters: dict) ->
             service = connection.datacube_from_process(
                 process_id=process_id, namespace=details.application, **parameters
             )
-            job = service.create_job(title=title)
+            job = service.create_job(title=title, out_format=format)
             job.start()
 
             return job.job_id

app/routers/unit_jobs.py

@@ -4,7 +4,7 @@
 from sqlalchemy.orm import Session
 
 from app.database.db import get_db
-from app.schemas.enum import ProcessTypeEnum
+from app.schemas.enum import OutputFormatEnum, ProcessTypeEnum
 from app.schemas.unit_job import (
     BaseJobRequest,
     ProcessingJob,
@@ -53,6 +53,7 @@ async def create_unit_job(
                             "/32ea3c9a6fa24fe063cb59164cd318cceb7209b0/openeo_udp/variabilitymap/"
                             "variabilitymap.json",
                         ),
+                        format=OutputFormatEnum.GEOTIFF,
                         parameters={
                             "spatial_extent": {
                                 "type": "FeatureCollection",

app/routers/upscale_tasks.py

@@ -15,7 +15,7 @@
 from sqlalchemy.orm import Session
 
 from app.database.db import SessionLocal, get_db
-from app.schemas.enum import ProcessTypeEnum
+from app.schemas.enum import OutputFormatEnum, ProcessTypeEnum
 from app.schemas.unit_job import (
     ServiceDetails,
 )
@@ -66,6 +66,7 @@ async def create_upscale_task(
                             "/32ea3c9a6fa24fe063cb59164cd318cceb7209b0/openeo_udp/variabilitymap/"
                             "variabilitymap.json",
                         ),
+                        format=OutputFormatEnum.GEOTIFF,
                         parameters={
                             "temporal_extent": ["2025-05-01", "2025-05-01"],
                         },

app/schemas/enum.py

@@ -14,3 +14,9 @@ class ProcessingStatusEnum(str, Enum):
     CANCELED = "canceled"
     FAILED = "failed"
     UNKNOWN = "unknown"
+
+
+class OutputFormatEnum(str, Enum):
+    GEOJSON = "geojson"
+    GEOTIFF = "gtiff"
+    NETCDF = "netcdf"

app/schemas/unit_job.py

@@ -2,7 +2,7 @@
 
 from pydantic import BaseModel, Field
 
-from app.schemas.enum import ProcessingStatusEnum, ProcessTypeEnum
+from app.schemas.enum import OutputFormatEnum, ProcessingStatusEnum, ProcessTypeEnum
 
 
 class ServiceDetails(BaseModel):
@@ -77,3 +77,6 @@ class BaseJobRequest(BaseModel):
         ...,
         description="JSON representing the parameters for the service execution",
     )
+    format: OutputFormatEnum = Field(
+        ..., description="Expected format of the output results"
+    )

app/services/processing.py

@@ -41,13 +41,15 @@ def create_processing_job(
 
     try:
         job_id = platform.execute_job(
-            title=request.title, details=request.service, parameters=request.parameters
+            title=request.title,
+            details=request.service,
+            parameters=request.parameters,
+            format=request.format,
         )
     except Exception:
         job_id = None
         logger.exception(f"Could not create processing job for user {user}")
 
-    print(f"JOB IS EQUAL TO {job_id}")
     record = ProcessingJobRecord(
         title=request.title,
         label=request.label,
@@ -58,9 +60,7 @@ def create_processing_job(
         service=request.service.model_dump_json(),
         upscaling_task_id=upscaling_task_id,
     )
-    print(f"RECORD IS EQUAL TO {record.status}")
     record = save_job_to_db(database, record)
-    print(f"RECORD IS EQUAL TO {record.status}")
     return ProcessingJobSummary(
         id=record.id,
         title=record.title,

app/services/upscaling.py

@@ -52,6 +52,7 @@ def create_upscaling_processing_jobs(
                     label=request.label,
                     service=request.service,
                     parameters={**request.parameters, request.dimension.name: value},
+                    format=request.format,
                 ),
                 upscaling_task_id=upscaling_task_id,
             )