Skip to content

rs_workflows/on_demand/sentinel3/s3_processing_utils.md

<< Back to index

Code shared by the Sentinel-3 OLCI processing flows.

emit_olci_quicklook_event(level, owner_id, products)

Request quicklooks for published OLCI products, linked to the current flow run.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_processing_utils.py 
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
def emit_olci_quicklook_event(
    level: Literal["l1", "l2"],
    owner_id: str,
    products: list[dict[str, Any]],
) -> None:
    """Request quicklooks for published OLCI products, linked to the current flow run."""
    if not products:
        return

    flow_run_id = str(runtime.flow_run.id or "unknown")
    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_logger().warning(
            "Quicklook-inputs-ready event was not emitted: event=%s, flow_run_id=%s",
            event_name,
            flow_run_id,
        )
    else:
        get_logger().info(
            "Emitted event=%s, event_id=%s, product_count=%d",
            event_name,
            event.id,
            len(products),
        )

process_olci(level, env_args, settings, input_products, external_variables, workflow) async

Run the OLCI processing of a level, then request the quicklooks of the generated products.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_processing_utils.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
async def process_olci(
    level: Literal["l1", "l2"],
    env_args: FlowEnvArgs,
    settings: ProcessorSettings,
    input_products: list[FlowInputProduct],
    external_variables: dict[str, Any],
    workflow: WorkflowType,
) -> list[dict[str, Any]]:
    """Run the OLCI processing of a level, then request the quicklooks of the generated products."""
    products = await call_dpr_flow(
        env=env_args,
        settings=settings,
        input_products=input_products,
        external_variables=external_variables,
        workflow=workflow,
    )
    emit_olci_quicklook_event(level, env_args.owner_id, products)
    return products