16367912334

Committed 18 Jul 2025 10:06AM UTC coverage: 76.934% (+0.1%) from 76.806%

Build # 16367912334

Build Type

push

github

Committed by

web-flow

Commit Message

ci: use Ruff as new formatter and linter (#233)

* wip

* pycodestyle

* update dependencies

* skl2onnx

* use ruff

* apply formatter

* apply lint auto fixes

* manually apply lints

* change check

* ruff ci from branch

Run Details

2805 of 3646 relevant lines covered (76.93%)

0.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.48

geoengine/workflow_builder/blueprints.py

"""This module contains blueprints for workflows."""

from . import operators


def sentinel2_band(band_name, provider="5779494c-f3a2-48b3-8a2d-5fbba8c5b6c5", utm_zone="UTM32N"):
    """Creates a workflow for a band from Sentinel 2 data."""
    band_source = operators.GdalSource(f"_:{provider}:`{utm_zone}:{band_name}`")
    return band_source


def sentinel2_cloud_free_band(
    band_name: str, provider: str = "5779494c-f3a2-48b3-8a2d-5fbba8c5b6c5", utm_zone: str = "UTM32N"
):
    """Creates a workflow for a cloud free band from Sentinel 2 data."""

    valid_sentinel_bands = ["B01", "B02", "B03", "B04", "B05", "B06", "B07", "B08", "B8A", "B11", "B12"]
    if band_name.upper() not in valid_sentinel_bands:
        raise ValueError(f"Invalid band name {band_name}. Valid band names are: {valid_sentinel_bands}")

    band_id = f"_:{provider}:`{utm_zone}:{band_name.upper()}`"
    scl_id = f"_:{provider}:`{utm_zone}:SCL`"
    return sentinel2_cloud_free_band_custom_input(band_id, scl_id)


def sentinel2_cloud_free_ndvi(provider="5779494c-f3a2-48b3-8a2d-5fbba8c5b6c5", utm_zone="UTM32N"):
    """Creates a workflow for a cloud free NDVI from Sentinel 2 data."""

    nir_id = f"_:{provider}:`{utm_zone}:B08`"
    red_id = f"_:{provider}:`{utm_zone}:B04`"
    scl_id = f"_:{provider}:`{utm_zone}:SCL`"

    return sentinel2_cloud_free_ndvi_custom_input(nir_id, red_id, scl_id)


def sentinel2_cloud_free_band_custom_input(band_dataset: str, scl_dataset: str):
    """Creates a workflow for a cloud free band from Sentinel 2 data provided by the inputs."""
    band_source = operators.GdalSource(dataset=band_dataset)
    scl_source = operators.GdalSource(dataset=scl_dataset)

    scl_source_u16 = operators.RasterTypeConversion(source=scl_source, output_data_type="U16")

    # [sen2_mask == 3 |sen2_mask == 7 |sen2_mask == 8 | sen2_mask == 9 |sen2_mask == 10 |sen2_mask == 11 ]
    cloud_free = operators.Expression(
        expression="if (B == 3 || (B >= 7 && B <= 11)) { NODATA } else { A }",
        output_type="U16",
        source=operators.RasterStacker([band_source, scl_source_u16]),
    )

    return cloud_free


def sentinel2_cloud_free_ndvi_custom_input(nir_dataset: str, red_dataset: str, scl_dataset: str):
    """Creates a workflow for a cloud free NDVI from Sentinel 2 data provided by the inputs."""
    nir_source = operators.GdalSource(dataset=nir_dataset)
    red_source = operators.GdalSource(dataset=red_dataset)
    scl_source = operators.GdalSource(dataset=scl_dataset)
    scl_source_u16 = operators.RasterTypeConversion(source=scl_source, output_data_type="U16")

    # [sen2_mask == 3 |sen2_mask == 7 |sen2_mask == 8 | sen2_mask == 9 |sen2_mask == 10 |sen2_mask == 11 ]
    cloud_free = operators.Expression(
        expression="if (C == 3 || (C >= 7 && C <= 11)) { NODATA } else { (A - B) / (A + B) }",
        output_type="F32",
        source=operators.RasterStacker([nir_source, red_source, scl_source_u16]),
    )

    return cloud_free


def s2_cloud_free_aggregated_band(
    band,
    provider="5779494c-f3a2-48b3-8a2d-5fbba8c5b6c5",
    utm_zone="UTM32N",
    granularity="days",
    window_size=1,
    aggregation_type="mean",
):
    # pylint: disable=too-many-arguments,too-many-positional-arguments
    """Creates a workflow for a cloud free monthly band (or NDVI) from Sentinel 2 data."""
    valid_sentinel_bands = ["B01", "B02", "B03", "B04", "B05", "B06", "B07", "B08", "B8A", "B11", "B12"]

    band_upper = band.upper()

    if band_upper not in valid_sentinel_bands and band_upper != "NDVI":
        raise ValueError(f"Invalid band: {band_upper}")

    if band_upper == "NDVI":
        s2_cloud_free_operator = sentinel2_cloud_free_ndvi(provider, utm_zone)
    else:
        s2_cloud_free_operator = sentinel2_cloud_free_band(band_upper, provider, utm_zone)

    return operators.TemporalRasterAggregation(
        source=s2_cloud_free_operator,
        aggregation_type=aggregation_type,
        granularity=granularity,
        window_size=window_size,
        ignore_no_data=True,
        output_type="F32",
    )


def s2_cloud_free_aggregated_band_custom_input(
    band_id: str, scl_id: str, granularity="days", window_size=1, aggregation_type="mean"
):
    """Creates a workflow for a cloud free monthly band from Sentinel 2 data provided by the inputs."""

    s2_cloud_free_operator = sentinel2_cloud_free_band_custom_input(band_id, scl_id)

    # We could also leave out the scaling and use the I64 data directly
    # s2_cloud_free_operator = geoengine.unstable.workflow_operators.RasterTypeConversion(
    #    source=s2_cloud_free_operator,
    #    output_data_type="F32"
    # )
    # s2_cloud_free_operator = geoengine.unstable.workflow_operators.RasterScaling(
    #    source=s2_cloud_free_operator,
    #    slope = 0.0001,
    #    offset = -0.1, # this should be -0.1 but the values are too small?
    #    scaling_mode="mulSlopeAddOffset"
    # )

    monthly_s2_cloud_free_operator = operators.TemporalRasterAggregation(
        source=s2_cloud_free_operator,
        aggregation_type=aggregation_type,
        granularity=granularity,
        window_size=window_size,
        ignore_no_data=True,
        output_type="F32",
    )

    return monthly_s2_cloud_free_operator


def s2_cloud_free_aggregated_ndvi_custom_input(
    nir_dataset: str, red_dataset: str, scl_dataset: str, granularity="days", window_size=1, aggregation_type="mean"
):
    # pylint: disable=too-many-arguments,too-many-positional-arguments
    """Creates a workflow for a cloud free monthly NDVI from Sentinel 2 data provided by the inputs."""

    s2_cloud_free_operator = sentinel2_cloud_free_ndvi_custom_input(nir_dataset, red_dataset, scl_dataset)

    monthly_s2_cloud_free_operator = operators.TemporalRasterAggregation(
        source=s2_cloud_free_operator,
        aggregation_type=aggregation_type,
        granularity=granularity,
        window_size=window_size,
        ignore_no_data=True,
        output_type="F32",
    )

    return monthly_s2_cloud_free_operator

1	"""This module contains blueprints for workflows."""
2
3	from . import operators	1✔
4
5
6	def sentinel2_band(band_name, provider="5779494c-f3a2-48b3-8a2d-5fbba8c5b6c5", utm_zone="UTM32N"):	1✔
7	"""Creates a workflow for a band from Sentinel 2 data."""
8	band_source = operators.GdalSource(f"_:{provider}:`{utm_zone}:{band_name}`")	1✔
9	return band_source	1✔
10
11
12	def sentinel2_cloud_free_band(	1✔
13	band_name: str, provider: str = "5779494c-f3a2-48b3-8a2d-5fbba8c5b6c5", utm_zone: str = "UTM32N"
14	):
15	"""Creates a workflow for a cloud free band from Sentinel 2 data."""
16
17	valid_sentinel_bands = ["B01", "B02", "B03", "B04", "B05", "B06", "B07", "B08", "B8A", "B11", "B12"]	1✔
18	if band_name.upper() not in valid_sentinel_bands:	1✔
19	raise ValueError(f"Invalid band name {band_name}. Valid band names are: {valid_sentinel_bands}")	×
20
21	band_id = f"_:{provider}:`{utm_zone}:{band_name.upper()}`"	1✔
22	scl_id = f"_:{provider}:`{utm_zone}:SCL`"	1✔
23	return sentinel2_cloud_free_band_custom_input(band_id, scl_id)	1✔
24
25
26	def sentinel2_cloud_free_ndvi(provider="5779494c-f3a2-48b3-8a2d-5fbba8c5b6c5", utm_zone="UTM32N"):	1✔
27	"""Creates a workflow for a cloud free NDVI from Sentinel 2 data."""
28
29	nir_id = f"_:{provider}:`{utm_zone}:B08`"	1✔
30	red_id = f"_:{provider}:`{utm_zone}:B04`"	1✔
31	scl_id = f"_:{provider}:`{utm_zone}:SCL`"	1✔
32
33	return sentinel2_cloud_free_ndvi_custom_input(nir_id, red_id, scl_id)	1✔
34
35
36	def sentinel2_cloud_free_band_custom_input(band_dataset: str, scl_dataset: str):	1✔
37	"""Creates a workflow for a cloud free band from Sentinel 2 data provided by the inputs."""
38	band_source = operators.GdalSource(dataset=band_dataset)	1✔
39	scl_source = operators.GdalSource(dataset=scl_dataset)	1✔
40
41	scl_source_u16 = operators.RasterTypeConversion(source=scl_source, output_data_type="U16")	1✔
42
43	# [sen2_mask == 3 \|sen2_mask == 7 \|sen2_mask == 8 \| sen2_mask == 9 \|sen2_mask == 10 \|sen2_mask == 11 ]
44	cloud_free = operators.Expression(	1✔
45	expression="if (B == 3 \|\| (B >= 7 && B <= 11)) { NODATA } else { A }",
46	output_type="U16",
47	source=operators.RasterStacker([band_source, scl_source_u16]),
48	)
49
50	return cloud_free	1✔
51
52
53	def sentinel2_cloud_free_ndvi_custom_input(nir_dataset: str, red_dataset: str, scl_dataset: str):	1✔
54	"""Creates a workflow for a cloud free NDVI from Sentinel 2 data provided by the inputs."""
55	nir_source = operators.GdalSource(dataset=nir_dataset)	1✔
56	red_source = operators.GdalSource(dataset=red_dataset)	1✔
57	scl_source = operators.GdalSource(dataset=scl_dataset)	1✔
58	scl_source_u16 = operators.RasterTypeConversion(source=scl_source, output_data_type="U16")	1✔
59
60	# [sen2_mask == 3 \|sen2_mask == 7 \|sen2_mask == 8 \| sen2_mask == 9 \|sen2_mask == 10 \|sen2_mask == 11 ]
61	cloud_free = operators.Expression(	1✔
62	expression="if (C == 3 \|\| (C >= 7 && C <= 11)) { NODATA } else { (A - B) / (A + B) }",
63	output_type="F32",
64	source=operators.RasterStacker([nir_source, red_source, scl_source_u16]),
65	)
66
67	return cloud_free	1✔
68
69
70	def s2_cloud_free_aggregated_band(	1✔
71	band,
72	provider="5779494c-f3a2-48b3-8a2d-5fbba8c5b6c5",
73	utm_zone="UTM32N",
74	granularity="days",
75	window_size=1,
76	aggregation_type="mean",
77	):
78	# pylint: disable=too-many-arguments,too-many-positional-arguments
79	"""Creates a workflow for a cloud free monthly band (or NDVI) from Sentinel 2 data."""
80	valid_sentinel_bands = ["B01", "B02", "B03", "B04", "B05", "B06", "B07", "B08", "B8A", "B11", "B12"]	1✔
81
82	band_upper = band.upper()	1✔
83
84	if band_upper not in valid_sentinel_bands and band_upper != "NDVI":	1✔
85	raise ValueError(f"Invalid band: {band_upper}")	×
86
87	if band_upper == "NDVI":	1✔
88	s2_cloud_free_operator = sentinel2_cloud_free_ndvi(provider, utm_zone)	×
89	else:
90	s2_cloud_free_operator = sentinel2_cloud_free_band(band_upper, provider, utm_zone)	1✔
91
92	return operators.TemporalRasterAggregation(	1✔
93	source=s2_cloud_free_operator,
94	aggregation_type=aggregation_type,
95	granularity=granularity,
96	window_size=window_size,
97	ignore_no_data=True,
98	output_type="F32",
99	)
100
101
102	def s2_cloud_free_aggregated_band_custom_input(	1✔
103	band_id: str, scl_id: str, granularity="days", window_size=1, aggregation_type="mean"
104	):
105	"""Creates a workflow for a cloud free monthly band from Sentinel 2 data provided by the inputs."""
106
107	s2_cloud_free_operator = sentinel2_cloud_free_band_custom_input(band_id, scl_id)	1✔
108
109	# We could also leave out the scaling and use the I64 data directly
110	# s2_cloud_free_operator = geoengine.unstable.workflow_operators.RasterTypeConversion(
111	# source=s2_cloud_free_operator,
112	# output_data_type="F32"
113	# )
114	# s2_cloud_free_operator = geoengine.unstable.workflow_operators.RasterScaling(
115	# source=s2_cloud_free_operator,
116	# slope = 0.0001,
117	# offset = -0.1, # this should be -0.1 but the values are too small?
118	# scaling_mode="mulSlopeAddOffset"
119	# )
120
121	monthly_s2_cloud_free_operator = operators.TemporalRasterAggregation(	1✔
122	source=s2_cloud_free_operator,
123	aggregation_type=aggregation_type,
124	granularity=granularity,
125	window_size=window_size,
126	ignore_no_data=True,
127	output_type="F32",
128	)
129
130	return monthly_s2_cloud_free_operator	1✔
131
132
133	def s2_cloud_free_aggregated_ndvi_custom_input(	1✔
134	nir_dataset: str, red_dataset: str, scl_dataset: str, granularity="days", window_size=1, aggregation_type="mean"
135	):
136	# pylint: disable=too-many-arguments,too-many-positional-arguments
137	"""Creates a workflow for a cloud free monthly NDVI from Sentinel 2 data provided by the inputs."""
138
139	s2_cloud_free_operator = sentinel2_cloud_free_ndvi_custom_input(nir_dataset, red_dataset, scl_dataset)	1✔
140
141	monthly_s2_cloud_free_operator = operators.TemporalRasterAggregation(	1✔
142	source=s2_cloud_free_operator,
143	aggregation_type=aggregation_type,
144	granularity=granularity,
145	window_size=window_size,
146	ignore_no_data=True,
147	output_type="F32",
148	)
149
150	return monthly_s2_cloud_free_operator	1✔

geo-engine / geoengine-python / 16367912334

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous