Skip to content

rs_workflows/on_demand/sentinel3/s3_processing_utils.md

<< Back to index

Shared configuration, product mapping and event 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 
73
74
75
76
77
78
79
80
81
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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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

emit_olci_quicklook_event(level, owner_id, products, flow_run_id)

Request quicklooks for published OLCI products, linked to their processing flow run.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_processing_utils.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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
def emit_olci_quicklook_event(
    level: Literal["l1", "l2"],
    owner_id: str,
    products: list[dict[str, Any]],
    flow_run_id: str,
) -> None:
    """Request quicklooks for published OLCI products, linked to their processing flow run."""
    if not products:
        return

    event_name = f"rs-python.s3-{level}.quicklook-inputs-ready"
    event = emit_event(
        event=event_name,
        resource={
            "prefect.resource.id": f"rs-python.s3-{level}-result.{flow_run_id}",
            "prefect.resource.name": f"S3 OLCI {level.upper()} products",
        },
        related=[
            {
                "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
                "prefect.resource.role": "flow-run",
            },
        ],
        payload={
            "owner_id": owner_id,
            # Only send catalog references; the quicklook flow reads the full items itself.
            "published_items": [{"id": product["id"], "collection": product["collection"]} for product in products],
        },
    )
    if event is None:
        # Missing quicklook events do not prevent returning the published processing results.
        get_run_logger().warning(
            "Quicklook-inputs-ready event was not emitted: event=%s, flow_run_id=%s",
            event_name,
            flow_run_id,
        )
    else:
        get_run_logger().info(
            "Emitted event=%s, event_id=%s, product_count=%d",
            event_name,
            event.id,
            len(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 
84
85
86
87
88
89
90
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"))