Skip to content

rs_workflows/systematic/process_s3l1olci_systematic_flow.md

<< Back to index

Process systematic S3 OLCI L1 flow implementation

process_s3l1olci_systematic(payload) async

Process SYSTEMATIC S3 L1 OLCI

This flow is started from automation. It get the very first input from the 'event' that triggers the flow. Then it prepares the S3 L1 OLCI processing inputs. Finally it runs the processor with workflow set to "systematic".

Workflow Steps

  1. load the settings — reads the s3l1olci-default-setting Prefect variable and checks that the Dask clusters are running.
  2. compute input products — retrieve the stac items S3OLCIL0_ from rs-catalog. The first one is retrieve from the event that trigger this flow. The other ones are get by requesting contiguous period on the rs-catalog. They can be one or two.
  3. compute external variables — external variables are extracted from the first item properties.
  4. start the processing — calls the flow process-generic, then emits the events of the generated products.
  5. request the quicklooks — emits the rs-python.s3-l1.quicklook-inputs-ready event with the generated products.

Parameters

Parameter Type Default Description
payload dict[str, Any] required Payload of the product.S03OLCL0_.generated event.

Note: if the collection retrieved for S3OLCIL0_name is equal to s03olcl0_, S3NAVL0_ will be stored on the collection s03natl0_. Otherwise, S3NAVL0_ will be stored on the same collection than S3OLCIL0_.

Source code in docs/rs-client-libraries/rs_workflows/systematic/process_s3l1olci_systematic_flow.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
@flow(name="process-s3-l1-olci-systematic")
async def process_s3l1olci_systematic(payload: dict[str, Any]) -> list[dict[str, Any]]:
    """
    # Process SYSTEMATIC S3 L1 OLCI

    This flow is started from automation.
    It get the very first input from the 'event' that triggers the flow.
    Then it prepares the S3 L1 OLCI processing inputs.
    Finally it runs the processor with workflow set to "systematic".

    ---
    ## Workflow Steps

    1. *load the settings* — reads the `s3l1olci-default-setting` Prefect variable and checks that the Dask clusters
       are running.
    2. *compute input products* — retrieve the stac items `S3OLCIL0_` from rs-catalog.
       The first one is retrieve from the `event` that trigger this flow.
       The other ones are get by requesting contiguous period on the rs-catalog.
       They can be one or two.
    3. *compute external variables* — external variables are extracted from the first item properties.
    4. *start the processing* — calls the flow `process-generic`, then emits the events of the generated products.
    5. *request the quicklooks* — emits the `rs-python.s3-l1.quicklook-inputs-ready` event with the generated products.
    ---
    ## Parameters
    Parameter | Type | Default | Description |
    |---|---|---|---|
    | `payload` | dict[str, Any] | *required* | Payload of the `product.S03OLCL0_.generated` event. |

    > *Note:* if the collection retrieved for `S3OLCIL0_`name is equal to `s03olcl0_`, `S3NAVL0_` will be stored on
    the collection `s03natl0_`.
    Otherwise, `S3NAVL0_` will be stored on the same collection than `S3OLCIL0_`.
    ---
    """
    return await process_event_product(payload, ProcessingId.S3_OLCI_L1, process_s3l1olci_product)