diff --git a/src/openeo_grass_gis_driver/actinia_processing/mask_polygon_process.py b/src/openeo_grass_gis_driver/actinia_processing/mask_polygon_process.py index 650ff8f..3d78a54 100644 --- a/src/openeo_grass_gis_driver/actinia_processing/mask_polygon_process.py +++ b/src/openeo_grass_gis_driver/actinia_processing/mask_polygon_process.py @@ -48,7 +48,7 @@ def create_process_description(): ) p_value = Parameter( description="The value used to replace non-zero and `true` values with", - schema={"type": "object", "subtype": "string"}, + schema={"type": ["number", "boolean", "string", "null"]}, optional=True, ) p_inside = Parameter( diff --git a/src/openeo_grass_gis_driver/actinia_processing/mask_process.py b/src/openeo_grass_gis_driver/actinia_processing/mask_process.py index d9f1890..5a9c63c 100644 --- a/src/openeo_grass_gis_driver/actinia_processing/mask_process.py +++ b/src/openeo_grass_gis_driver/actinia_processing/mask_process.py @@ -3,14 +3,23 @@ from random import randint from typing import Tuple -from openeo_grass_gis_driver.models.process_graph_schemas import \ - ProcessGraphNode, ProcessGraph - -from openeo_grass_gis_driver.actinia_processing.base import \ - check_node_parents, DataObject, GrassDataType, \ - create_output_name -from openeo_grass_gis_driver.models.process_schemas import \ - Parameter, ProcessDescription, ReturnValue, ProcessExample +from openeo_grass_gis_driver.models.process_graph_schemas import ( + ProcessGraphNode, + ProcessGraph, +) + +from openeo_grass_gis_driver.actinia_processing.base import ( + check_node_parents, + DataObject, + GrassDataType, + create_output_name, +) +from openeo_grass_gis_driver.models.process_schemas import ( + Parameter, + ProcessDescription, + ReturnValue, + ProcessExample, +) from .base import PROCESS_DICT, PROCESS_DESCRIPTION_DICT, Node __license__ = "Apache License, Version 2.0" @@ -26,26 +35,25 @@ def create_process_description(): p_data = Parameter( description="Any openEO process object that returns raster datasets " "or space-time raster dataset", - schema={ - "type": "object", - "subtype": "raster-cube"}, - optional=False) + schema={"type": "object", "subtype": "raster-cube"}, + optional=False, + ) p_mask = Parameter( description="Any openEO process object that returns raster datasets " "or space-time raster dataset", - schema={ - "type": "object", - "subtype": "raster-cube"}, - optional=False) + schema={"type": "object", "subtype": "raster-cube"}, + optional=False, + ) p_value = Parameter( description="The value used to replace non-zero and `true` values with", - schema={ - "type": "object", - "subtype": "string"}, - optional=True) + schema={"type": ["number", "boolean", "string", "null"]}, + optional=True, + ) - rv = ReturnValue(description="Processed EO data.", - schema={"type": "object", "subtype": "raster-cube"}) + rv = ReturnValue( + description="Processed EO data.", + schema={"type": "object", "subtype": "raster-cube"}, + ) # Example arguments = { @@ -55,27 +63,23 @@ def create_process_description(): } node = ProcessGraphNode(process_id=PROCESS_NAME, arguments=arguments) graph = ProcessGraph( - title="title", - description="description", - process_graph={ - "mask_1": node}) + title="title", description="description", process_graph={"mask_1": node} + ) examples = [ ProcessExample( - title="Simple example", - description="Simple example", - process_graph=graph)] + title="Simple example", description="Simple example", process_graph=graph + ) + ] pd = ProcessDescription( id=PROCESS_NAME, description="Applies a mask to a raster data cube " " replacing pixels in data that are not null in mask with the new value.", summary="Applies a mask to a raster data cube", - parameters={ - "data": p_data, - "mask": p_mask, - "replacement": p_value}, + parameters={"data": p_data, "mask": p_mask, "replacement": p_value}, returns=rv, - examples=[examples]) + examples=[examples], + ) return json.loads(pd.to_json()) @@ -84,10 +88,11 @@ def create_process_description(): def create_process_chain_entry( - input_object: DataObject, - mask_object: DataObject, - mask_value: str, - output_object: DataObject): + input_object: DataObject, + mask_object: DataObject, + mask_value: str, + output_object: DataObject, +): """Create a Actinia command of the process chain that uses t.rast.mapcalc to mask raster values based on a mask dataset and a replacement value @@ -105,27 +110,30 @@ def create_process_chain_entry( pc = [] - p = {"id": "t_rast_mask_%i" % rn, - "module": "t.rast.mask", - "inputs": [{"param": "input", - "value": input_object.grass_name()}, - {"param": "mask", - "value": mask_object.grass_name()}, - {"param": "basename", - "value": output_object.name}, - {"param": "output", - "value": output_object.grass_name()}, - {"param": "value", - "value": mask_value}], - "flags": "i"} + p = { + "id": "t_rast_mask_%i" % rn, + "module": "t.rast.mask", + "inputs": [ + {"param": "input", "value": input_object.grass_name()}, + {"param": "mask", "value": mask_object.grass_name()}, + {"param": "basename", "value": output_object.name}, + {"param": "output", "value": output_object.grass_name()}, + {"param": "value", "value": mask_value}, + ], + "flags": "i", + } pc.append(p) - p = {"id": "t_info_%i" % rn, - "module": "t.info", - "inputs": [{"param": "input", "value": output_object.grass_name()}, - {"param": "type", "value": "strds"}], - "flags": 'g'} + p = { + "id": "t_info_%i" % rn, + "module": "t.info", + "inputs": [ + {"param": "input", "value": output_object.grass_name()}, + {"param": "type", "value": "strds"}, + ], + "flags": "g", + } pc.append(p) @@ -142,11 +150,8 @@ def get_process_list(node: Node) -> Tuple[list, list]: input_objects, process_list = check_node_parents(node=node) output_objects = [] - if "data" not in node.arguments or \ - "mask" not in node.arguments: - raise Exception( - "Process %s requires parameter data, mask" % - PROCESS_NAME) + if "data" not in node.arguments or "mask" not in node.arguments: + raise Exception("Process %s requires parameter data, mask" % PROCESS_NAME) if "replacement" in node.arguments: mask_value = node.arguments["replacement"] @@ -154,24 +159,16 @@ def get_process_list(node: Node) -> Tuple[list, list]: mask_value = "null" # Get the input and mask data separately - data_object = list( - node.get_parent_by_name( - parent_name="data").output_objects)[0] - mask_object = list( - node.get_parent_by_name( - parent_name="mask").output_objects)[0] + data_object = list(node.get_parent_by_name(parent_name="data").output_objects)[0] + mask_object = list(node.get_parent_by_name(parent_name="mask").output_objects)[0] output_object = DataObject( - name=create_output_name(data_object.name, node), - datatype=GrassDataType.STRDS) + name=create_output_name(data_object.name, node), datatype=GrassDataType.STRDS + ) output_objects.append(output_object) node.add_output(output_object=output_object) - pc = create_process_chain_entry( - data_object, - mask_object, - mask_value, - output_object) + pc = create_process_chain_entry(data_object, mask_object, mask_value, output_object) process_list.extend(pc) return output_objects, process_list