Skip to content

rs_workflows/on_demand/sentinel3/s3_l1_olci.md

<< Back to index

sentinel 3 OLCI Level-1 processing.

process_s3l1_olci(flow_params=None, l0_products=None, source_l0_run_id='manual') async

Sentinel-3 OLCI L1 processing. The input_products should have been processed before by L0.

l0_products is the raw product list emitted by S3 L0. When supplied by a Prefect Automation, it is converted here into the four (or more) processor inputs expected by OLCI L1. source_l0_run_id provides a short upstream reference used in the L1 flow-run name.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_l1_olci.py 
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
@flow(
    name="process-s3-l1-olci",
    flow_run_name="s3-l1-olci-from-{source_l0_run_id}",
)
async def process_s3l1_olci(
    flow_params: Level1FlowParams | None = None,
    l0_products: list[dict[str, Any]] | None = None,
    source_l0_run_id: str = "manual",  # pylint: disable=unused-argument
) -> list[dict[str, Any]]:
    """
    Sentinel-3 OLCI L1 processing.
    The input_products should have been processed before by L0.

    ``l0_products`` is the raw product list emitted by S3 L0. When supplied by
    a Prefect Automation, it is converted here into the four (or more) processor inputs
    expected by OLCI L1. ``source_l0_run_id`` provides a short upstream
    reference used in the L1 flow-run name.
    """
    mission = "3"
    # how to use s3-l1-default-setting
    flow_parameters = await (flow_params or Level1FlowParams()).resolve(mission)

    if l0_products is not None:
        orchestration_settings = await read_s3_orchestration_settings()
        prepared_inputs = build_olci_l1_input_products(
            l0_products,
            orchestration_settings.s3_l0_output_collection,
        )
        flow_parameters.input_products = [FlowInputProduct.model_validate(product) for product in prepared_inputs]
        get_run_logger().info(
            "Built %d S3 L1 input product(s) from %d raw L0 product(s) received from Automation",
            len(flow_parameters.input_products),
            len(l0_products),
        )
    get_run_logger().info(f"Flow params: {flow_parameters}")
    # Call DPR flow
    return await call_dpr_flow(
        FlowEnvArgs(owner_id=flow_parameters.owner_identifier),
        input_products=flow_parameters.input_products,
        external_variables={
            "start_datetime": flow_parameters.start_datetime,
            "end_datetime": flow_parameters.end_datetime,
            "satellite": flow_parameters.satellite,
        },
        dask_cluster_label=flow_parameters.dask_cluster_label,
        processor_name=flow_parameters.processor_name,
        processor_version=flow_parameters.processor_version,
        pipeline=flow_parameters.pipeline,
        unit=flow_parameters.unit,
        priority=flow_parameters.priority,
        processing_mode=flow_parameters.processing_mode,
        workflow=flow_parameters.workflow,
        generated_product_to_collection_identifier=flow_parameters.generated_product_to_collection_identifier or [],
        auxiliary_product_to_collection_identifier=flow_parameters.auxiliary_product_to_collection_identifier or [],
        logging_level=flow_parameters.logging_level,
    )

process_s3l1_olci_task(*args, **kwargs) async

See: dpr_processing

Source code in docs/rs-client-libraries/rs_workflows/on_demand/sentinel3/s3_l1_olci.py 
91
92
93
94
@task(name="process-s3-l1-olci")
async def process_s3l1_olci_task(*args, **kwargs) -> list[dict[str, Any]]:
    """See: dpr_processing"""
    return await process_s3l1_olci.fn(*args, **kwargs)