Skip to content

rs_workflows/on_demand/sentinel3/s3_processing_utils.md

<< Back to index

Shared configuration and product-mapping helpers for Sentinel-3 processing.

S3ProcessingOrchestrationSettings

Bases: BaseModel

Environment-specific deployment and collection names for the S3 chain.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_processing_utils.py 
26
27
28
29
30
31
32
33
34
class S3ProcessingOrchestrationSettings(BaseModel):
    """Environment-specific deployment and collection names for the S3 chain."""

    cadip_staging_deployment: str = Field(min_length=1)
    s3_l0_deployment: str = Field(min_length=1)
    s3_l1_olci_deployment: str = Field(min_length=1)
    cadip_collection: str = Field(min_length=1)
    staging_catalog_collection: str = Field(min_length=1)
    s3_l0_output_collection: str = Field(min_length=1)

build_olci_l1_input_products(l0_products, input_collection)

Convert full STAC items or compact event products into OLCI L1 inputs.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_processing_utils.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def build_olci_l1_input_products(
    l0_products: list[dict[str, Any]],
    input_collection: str,
) -> list[dict[str, str]]:
    """Convert full STAC items or compact event products into OLCI L1 inputs."""

    normalized_products: list[tuple[str, str]] = []
    for product in l0_products:
        if "id" in product and "properties" in product:
            product_type = product["properties"]["product:type"]
            item_id = product["id"]
        elif len(product) == 1:
            product_type, item_id = next(iter(product.items()))
        else:
            raise ValueError(
                "Invalid L0 product: expected a STAC item or a single {product_type: item_id} entry",
            )
        if not isinstance(product_type, str) or not isinstance(item_id, str):
            raise ValueError("Invalid L0 product: product type and item ID must be strings")
        normalized_products.append((product_type, item_id))

    olci_item_ids = [item_id for product_type, item_id in normalized_products if product_type == "S03OLCL0_"]
    nav_item_ids = [item_id for product_type, item_id in normalized_products if product_type == "S03NATL0_"]

    if len(olci_item_ids) < 3:
        raise ValueError(f"Expected at least 3 S03OLCL0_ products from L0, found {len(olci_item_ids)}")
    if not nav_item_ids:
        raise ValueError("Expected at least 1 S03NATL0_ product from L0, found 0")

    input_products = [
        {
            "name": f"S3OLCIL0_{index}",
            "item_id": item_id,
            "collection_name": input_collection,
        }
        for index, item_id in enumerate(olci_item_ids[:3], start=1)
    ]
    input_products.append(
        {
            "name": "S3NAVL0_1",
            "item_id": nav_item_ids[0],
            "collection_name": input_collection,
        },
    )
    return input_products

read_s3_orchestration_settings() async

Load and validate the orchestration section of the unified S3 variable.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_processing_utils.py 
37
38
39
40
41
42
43
async def read_s3_orchestration_settings() -> S3ProcessingOrchestrationSettings:
    """Load and validate the ``orchestration`` section of the unified S3 variable."""
    raw_settings = await cast(Awaitable[Any], Variable.get(S3_PROCESSING_CONFIGURATION, default={}))
    if not isinstance(raw_settings, dict):
        raise ValueError(f"Prefect variable {S3_PROCESSING_CONFIGURATION!r} must contain a dictionary")

    return S3ProcessingOrchestrationSettings.model_validate(raw_settings.get("orchestration"))