[PT-5056] Parameterized job templates (#627)

* Parameterized job templates

* Drop changelog reformatting

* Apply hooks

* Tech debt comment

CHANGELOG.md

@@ -29,6 +29,12 @@ You can check your current version with the following command:
     ```
 
 For more information, see [UP42 Python package description](https://pypi.org/project/up42-py/).
+## 1.0.4a11
+
+**Jun 10, 2024**
+
+- Added job templates for `SpaceptAugmentation`, `NSUpsamling`, `Pansharpening` to `templates.py`
+
 ## 1.0.4a10
 
 **Jun 7, 2024**

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "1.0.4a10"
+version = "1.0.4a11"
 description = "Python SDK for UP42, the geospatial marketplace and developer platform."
 authors = ["UP42 GmbH <[email protected]>"]
 license = "https://github.com/up42/up42-py/blob/master/LICENSE"

tests/test_templates.py

@@ -1,3 +1,5 @@
+import random
+from typing import Optional
 from unittest import mock
 
 import pystac
@@ -55,3 +57,58 @@ def test_should_construct_multi_item_template(self, template_class):
         template = template_class(title=TITLE, items=[item], workspace_id=constants.WORKSPACE_ID)
         assert template.is_valid and template.cost == COST
         assert template.inputs == {"title": TITLE, "items": [ITEM_URL]}
+
+
+class TestSpaceptAugmentation:
+    def test_should_construct_template(self):
+        denoising_factor = random.randint(0, 100)
+        colour_denoising_factor = random.randint(0, 100)
+        template = templates.SpaceptAugmentation(
+            title=TITLE,
+            item=item,
+            denoising_factor=denoising_factor,
+            colour_denoising_factor=colour_denoising_factor,
+            workspace_id=constants.WORKSPACE_ID,
+        )
+        assert template.is_valid and template.cost == COST
+        assert template.inputs == {
+            "title": TITLE,
+            "item": ITEM_URL,
+            "denoising_factor": denoising_factor,
+            "colour_denoising_factor": colour_denoising_factor,
+        }
+
+
+class TestNSUpsampling:
+    def test_should_construct_template(self):
+        ned, rgb = random.choices([True, False], k=2)
+        template = templates.NSUpsamling(
+            title=TITLE,
+            item=item,
+            ned=ned,
+            rgb=rgb,
+            workspace_id=constants.WORKSPACE_ID,
+        )
+        assert template.is_valid and template.cost == COST
+        assert template.inputs == {
+            "title": TITLE,
+            "item": ITEM_URL,
+            "NED": ned,
+            "RGB": rgb,
+        }
+
+
+class TestPansharpening:
+    @pytest.mark.parametrize("grey_weight", [templates.GreyWeight(band="red", weight=random.random()), None])
+    def test_should_construct_template(self, grey_weight: Optional[templates.GreyWeight]):
+        template = templates.Pansharpening(
+            title=TITLE,
+            item=item,
+            grey_weights=[grey_weight] if grey_weight else None,
+            workspace_id=constants.WORKSPACE_ID,
+        )
+        grey_weights = (
+            {"greyWeights": [{"band": grey_weight.band, "weight": grey_weight.weight}]} if grey_weight else {}
+        )
+        assert template.is_valid and template.cost == COST
+        assert template.inputs == {"title": TITLE, "item": ITEM_URL, **grey_weights}

up42/templates.py

@@ -1,9 +1,10 @@
 import dataclasses
-from typing import Union
+from typing import List, Optional, Union
 
 from up42 import base, processing
 
 
+# TODO: drop these with Python 3.10 kw_only=True data classes
 @dataclasses.dataclass
 class WorkspaceIdSingleItemTemplate(processing.SingleItemJobTemplate):
     workspace_id: Union[str, base.WorkspaceId] = dataclasses.field(default=base.WorkspaceId())
@@ -77,3 +78,52 @@ class HypervergePleiadesChangeDetection(WorkspaceIdMultiItemTemplate):
 @dataclasses.dataclass
 class HypervergeSpotChangeDetection(WorkspaceIdMultiItemTemplate):
     process_id = "detection-change-spot-hyperverge"
+
+
+@dataclasses.dataclass
+class SpaceptAugmentation(processing.SingleItemJobTemplate):
+    denoising_factor: int = 0
+    colour_denoising_factor: int = 10
+    workspace_id: Union[str, base.WorkspaceId] = dataclasses.field(default=base.WorkspaceId())
+
+    @property
+    def inputs(self):
+        return {
+            **super().inputs,
+            "denoising_factor": self.denoising_factor,
+            "colour_denoising_factor": self.colour_denoising_factor,
+        }
+
+
+@dataclasses.dataclass
+class NSUpsamling(processing.SingleItemJobTemplate):
+    ned: bool = False
+    rgb: bool = True
+    workspace_id: Union[str, base.WorkspaceId] = dataclasses.field(default=base.WorkspaceId())
+
+    @property
+    def inputs(self):
+        return {
+            **super().inputs,
+            "NED": self.ned,
+            "RGB": self.rgb,
+        }
+
+
+@dataclasses.dataclass
+class GreyWeight:
+    band: str
+    weight: float
+
+
+@dataclasses.dataclass
+class Pansharpening(processing.SingleItemJobTemplate):
+    grey_weights: Optional[List[GreyWeight]] = None
+    workspace_id: Union[str, base.WorkspaceId] = dataclasses.field(default=base.WorkspaceId())
+
+    @property
+    def inputs(self):
+        weights = (
+            {"greyWeights": [dataclasses.asdict(weight) for weight in self.grey_weights]} if self.grey_weights else {}
+        )
+        return {**super().inputs, **weights}