Skip to content

rs_workflows/on_demand/sentinel3/s3_l0.md

<< Back to index

sentinel 3 Level-0 processing.

process_s3l0(session, flow_params=None, verbose=False) async

Sentinel-3 L0 processing. The session should have been staged before.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_l0.py 
 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
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@flow(
    name="process-s3-l0",
    flow_run_name="s3-l0-{session}",
    persist_result=True,
    result_storage=S3_L0_RESULT_STORAGE,
)
async def process_s3l0(
    session: str,
    flow_params: Level0FlowParams | None = None,
    verbose: bool = False,
) -> list[dict[str, Any]]:
    """
    Sentinel-3 L0 processing.
    The session should have been staged before.
    """

    # Resolve values from the s3-l0-default-setting Prefect variable before
    # using them to build the DPR input. Explicit flow parameters still win.
    resolved_flow_params = await (flow_params or Level0FlowParams()).resolve("3")

    logger = get_run_logger()
    input_products = [
        FlowInputProduct(
            name="S3ACADUS",
            item_id=session,
            collection_name=resolved_flow_params.session_collection,
        ),
    ]
    products = await process_l0_last_steps(
        mission="3",
        session=session,
        flow_params=resolved_flow_params,
        input_products=input_products,
        verbose=verbose,
    )
    event_products = [
        {
            product["properties"]["product:type"]: product["id"],
        }
        for product in products
    ]

    flow_run_id = str(runtime.flow_run.id or "unknown")
    event_name = products_ready_event_name(mission="3", level="0")
    # Prefect event payloads are limited to 1.5 MB. Send only product type and
    # item ID instead of the complete STAC items with links and assets.
    emitted_event = emit_event(
        event=event_name,
        resource={
            "prefect.resource.id": f"rs-python.s3-l0-result.{flow_run_id}",
            "prefect.resource.name": session,
            "rs-python.session-id": session,
        },
        related=[
            {
                "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
                "prefect.resource.role": "flow-run",
            },
        ],
        payload={
            "flow_run_id": flow_run_id,
            "session_id": session,
            "products": event_products,
        },
    )
    if emitted_event is None:
        logger.warning(
            "Products-ready event was not emitted: event=%s, flow_run_id=%s, session=%s",
            event_name,
            flow_run_id,
            session,
        )
    else:
        logger.info(
            "Emitted event=%s, event_id=%s, flow_run_id=%s, session=%s, product_count=%d",
            event_name,
            emitted_event.id,
            flow_run_id,
            session,
            len(event_products),
        )

    return products

process_s3l0_task(*args, **kwargs) async

See: dpr_processing

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_l0.py 
115
116
117
118
@task(name="process-s3-l0")
async def process_s3l0_task(*args, **kwargs) -> list[dict[str, Any]]:
    """See: dpr_processing"""
    return await process_s3l0.fn(*args, **kwargs)