rs_workflows/on_demand/adf/convert_adf_set.md

<< Back to index

Convert a set of ADF data.

adf_conversion_scheduled(env, adf_type, cql2_filter_without_date, period, auxiliary_product_to_collection_identifier) async

Flow to convert ADF data for a scheduled period. The period is defined by the scheduling rule of the flow and the period parameter, which defines the length of the period to convert starting from the flow run start time.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/adf/convert_adf_set.py

@flow(name="convert-adf-scheduled")
async def adf_conversion_scheduled(
    env: FlowEnvArgs,
    adf_type: str,
    cql2_filter_without_date: dict,
    period: str,
    auxiliary_product_to_collection_identifier: list[AuxiliaryProductMapping],
) -> None:
    """
    Flow to convert ADF data for a scheduled period. The period is defined by the scheduling rule of the flow and
    the `period` parameter, which defines the length of the period to convert starting from the flow run start time.
    """
    logger = get_run_logger()

    decoded_period: timedelta = timedelta(seconds=int(period))

    logger.info(f"Starting the conversion for {adf_type}")
    start: datetime = flow_run.scheduled_start_time - decoded_period
    stop: datetime = flow_run.scheduled_start_time
    cql2_filter = substitute_values(
        cql2_filter_without_date,
        {
            "start_datetime": strftime_millis(start),
            "end_datetime": strftime_millis(stop),
        },
    )
    logger.debug(f"🧹 The CQL2 used for the conversion is {cql2_filter}")

    flow_parameters: AdfProcessIn = AdfProcessIn(
        env=env,
        adf_type=adf_type,
        auxiliary_product_to_collection_identifier=auxiliary_product_to_collection_identifier,
        cql2_filter=cql2_filter,
    )
    await adf_conversion_task.with_options(name=f"convert {adf_type} on the period [{start}-{stop}]")(flow_parameters)

compute_cql2(cql2_query_name, dta, dtb, satellite)

Compute the CQL2 filter content by reading the configuration file and substituting the values.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/adf/convert_adf_set.py

def compute_cql2(cql2_query_name: str, dta: int, dtb: int, satellite: str | None) -> dict:
    """Compute the CQL2 filter content by reading the configuration file and substituting the values."""
    logger = get_run_logger()

    try:
        # Read the file and load its content into a variable
        with open(CQL2_FILTERS_PATH, encoding="utf-8") as file:
            cql2_json = json.load(file)  # json_data is a Python dictionary

    except FileNotFoundError:
        logger.error(f"❌ Error: The file '{CQL2_FILTERS_PATH}' does not exist.")
    except json.JSONDecodeError as e:
        logger.error(f"❌ Error: The file '{CQL2_FILTERS_PATH}' is not valid JSON. Details: {e}")
    except OSError as e:
        logger.error(f"❌ OS error while reading file '{CQL2_FILTERS_PATH}': {e}")

    logger.debug(f"🧹 Read CQL2 filter content from file '{CQL2_FILTERS_PATH}' : {cql2_json}")

    # Find the filter
    cql2_temp = next(entry for entry in cql2_json if entry["name"] == cql2_query_name)
    if cql2_temp is None:
        raise RuntimeError(f"❌ cql2 query '{cql2_query_name}' not found on configuration.")
    cql2_json = {
        "filter": cql2_temp["stac"]["filter"],
        "sortby": cql2_temp["stac"]["sortby"],
        "limit": cql2_temp["stac"]["limit"],
    }

    return substitute_values(cql2_json, {"dTa": dta, "dTb": dtb, "satellite": satellite})

convert_adf_group(period_start_datetime, period_end_datetime, configuration, owner='copernicus') async

Convert ADF format

Convert a set of ADF (Auxiliary Data Files) retrieved from ADGS or rs-catalog.

---

Workflow Steps

1. Download — Retrieves the old format of ADF files from ADGS or rs-catalog.
2. Convert — Immediately performs the conversion for past periods.
3. Schedule — Schedules a conversion flow for future periods.

---

Parameters

Parameter	Type	Default	Description
period_start_datetime	datetime	required	Period start in UTC
period_end_datetime	datetime	required	Period end in UTC
configuration	dict	required	JSON configuration (see format below)
owner	str	copernicus	Name of the user triggering the flow

Note: If a period traverses the current period, the part of the period in the past is processed immediately, the part of the period in the future is scheduled for later execution.

---

Configuration Format example

{
    "satellite": "all",
    "aux-to-be-generated": [
        {
            "product_type": "S00__ADF_ECMWA",
            "cql2_query_name": "ValIntersect",
            "dTa": 43200,
            "dTb": 43200,
            "period_in_hours": 24
        }
    ],
    "auxiliary-product-to-collection-identifier": [
        {
            "product_type": "AX___MA1_AX",
            "collection_name": "adfs_old_format",
            "source": "catalog"
        },
        {
            "product_type": "ADF_ECMWA",
            "collection_name": "adfs_new_format"
        }
    ]
}

By default source is set to AUXIP and it is only taken into account for input data.

---

Source code in docs/rs-client-libraries/rs_workflows/on_demand/adf/convert_adf_set.py

@flow(name="convert-adf-group")
async def convert_adf_group(
    period_start_datetime: datetime,
    period_end_datetime: datetime,
    configuration: dict,
    owner: str = "copernicus",
) -> None:
    """
    # Convert ADF format

    Convert a set of ADF (Auxiliary Data Files) retrieved from ADGS or rs-catalog.

    ---

    ## Workflow Steps

    1. *Download* — Retrieves the old format of ADF files from ADGS or rs-catalog.
    2. *Convert* — Immediately performs the conversion for past periods.
    3. *Schedule* — Schedules a conversion flow for future periods.

    ---

    ## Parameters
    Parameter | Type | Default | Description |
    |---|---|---|---|
    | `period_start_datetime` | `datetime` | *required* | Period start in UTC |
    | `period_end_datetime` | `datetime` | *required* | Period end in UTC |
    | `configuration` | `dict` | *required* | JSON configuration (see format below) |
    | `owner` | `str` | `copernicus` | Name of the user triggering the flow |

    > *Note:* If a period traverses the current period, the part of the period in
    > the past is processed immediately, the part of the period in the future is
    > scheduled for later execution.

    ---
    ## Configuration Format example

    ```json
    {
        "satellite": "all",
        "aux-to-be-generated": [
            {
                "product_type": "S00__ADF_ECMWA",
                "cql2_query_name": "ValIntersect",
                "dTa": 43200,
                "dTb": 43200,
                "period_in_hours": 24
            }
        ],
        "auxiliary-product-to-collection-identifier": [
            {
                "product_type": "AX___MA1_AX",
                "collection_name": "adfs_old_format",
                "source": "catalog"
            },
            {
                "product_type": "ADF_ECMWA",
                "collection_name": "adfs_new_format"
            }
        ]
    }
    ```
      > By default `source` is set to `AUXIP` and it is only taken into
      > account for input data.
    ---
    """
    logger = get_run_logger()

    # Check input chronology
    if period_start_datetime >= period_end_datetime:
        raise ValueError(
            "❌ period_start_datetime should be before period_end_datetime",
            f" ( here {period_start_datetime} >= {period_end_datetime})",
        )

    # Some checks
    if not isinstance(configuration, dict):
        raise ValueError("❌ Configuration has got an invalid format.")

    satellite: str | None = configuration["satellite"]
    logger.debug(f"read from the 'configuration' : satellite = {satellite}")

    aux_to_be_generated: list = configuration["aux-to-be-generated"]
    logger.debug(f"read from 'configuration' : aux_to_be_generated = {aux_to_be_generated}")

    auxiliary_product_to_collection_identifier: list[AuxiliaryProductMapping] = configuration[
        "auxiliary-product-to-collection-identifier"
    ]
    logger.debug(
        "read from 'configuration' : auxiliary_product_to_collection_identifier = "
        f"{auxiliary_product_to_collection_identifier}",
    )

    # Split the problem in two : past and future period.
    now_utc = datetime.now(timezone.utc)

    # 1) Let's start with future period
    if period_end_datetime > now_utc:
        schedule_start = now_utc if period_start_datetime < now_utc else period_start_datetime
        logger.info(f"Scheduling ADF conversion for the period [{schedule_start} - {period_end_datetime}]")
        for item in aux_to_be_generated:
            await schedule_adf_conversion.with_options(name=f"schedule {item['product_type']}  ")(
                owner,
                item["product_type"],
                item["cql2_query_name"],
                item.get("dTa", 0),
                item.get("dTb", 0),
                item["period_in_hours"],
                schedule_start,
                period_end_datetime,
                auxiliary_product_to_collection_identifier,
                satellite,
            )
    else:
        logger.info("No AUX data to be retrieved for the future period. No flow will be scheduled.")

    # 2) Continue with transformation on the past
    if period_start_datetime < now_utc:
        retrieve_past_start = period_start_datetime
        retrieve_past_end = now_utc if period_end_datetime > now_utc else period_end_datetime

        logger.info(f"Convert ADF for the period [{retrieve_past_start} - {retrieve_past_end}]")

        tasks = [
            past_adf_conversion.with_options(name=f"convert {item['product_type']}  ")(
                owner,
                item["product_type"],
                item["cql2_query_name"],
                item.get("dTa", 0),
                item.get("dTb", 0),
                item["period_in_hours"],
                retrieve_past_start,
                retrieve_past_end,
                auxiliary_product_to_collection_identifier,
                satellite,
            )
            for item in aux_to_be_generated
        ]
        await asyncio.gather(*tasks)
    else:
        logger.info("No AUX data to retrieve in the past.")

past_adf_conversion(owner_identifier, product_type, cql2_query_name, dta, dtb, period_in_hours, period_start, period_end, auxiliary_product_to_collection_identifier, satellite) async

Convert ADF data for a period in the past by splitting it into sub-periods of length period_in_hours and running the conversion flow on each of them. If period_in_hours is equal to 0, then the conversion is run on the whole period at once.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/adf/convert_adf_set.py

@task(name="conversion from the past")
async def past_adf_conversion(
    owner_identifier: str,
    product_type: str,
    cql2_query_name: str,
    dta: int,
    dtb: int,
    period_in_hours: int,
    period_start: datetime,
    period_end: datetime,
    auxiliary_product_to_collection_identifier: list[AuxiliaryProductMapping],
    satellite: str | None,
) -> None:
    """
    Convert ADF data for a period in the past by splitting it into
    sub-periods of length `period_in_hours` and running the conversion flow on each of them.
    If `period_in_hours` is equal to 0, then the conversion is run on the whole period at once.
    """
    logger = get_run_logger()
    logger.info("🧹 Computing cql2_filter without start_datetime and end_datetime...")
    cql2_filter_without_date = compute_cql2(cql2_query_name, dta, dtb, satellite)

    # Scheduling according to the period_in_hours
    flow_parameters: AdfProcessIn
    if period_in_hours == 0:
        # special case where a single run is requested
        logger.info(
            f"Run the conversion task with time range [{period_start}-{period_end}].",
        )
        cql2_filter = substitute_values(
            cql2_filter_without_date,
            {
                "start_datetime": strftime_millis(period_start),
                "end_datetime": strftime_millis(period_end),
            },
        )
        logger.debug(f"🧹 Associated cql2 filter is: {cql2_filter}")
        flow_parameters = AdfProcessIn(
            env=FlowEnvArgs(owner_id=owner_identifier),
            adf_type=product_type,
            auxiliary_product_to_collection_identifier=auxiliary_product_to_collection_identifier,
            cql2_filter=cql2_filter,
        )
        await adf_conversion_task.with_options(
            name=f"convert {product_type} on the period [{period_start}-{period_end}]",
        )(flow_parameters)

    else:
        start = period_start
        duration: timedelta = timedelta(hours=int(period_in_hours))
        while start < period_end:
            stop = min(start + duration, period_end)
            logger.info(
                f"Run the conversion task time range [{start}-{stop}].",
            )
            cql2_filter = substitute_values(
                cql2_filter_without_date,
                {
                    "start_datetime": strftime_millis(start),
                    "end_datetime": strftime_millis(stop),
                },
            )
            logger.debug(f"🧹 ( past ) Associated cql2 filter is: {cql2_filter}")
            flow_parameters = AdfProcessIn(
                env=FlowEnvArgs(owner_id=owner_identifier),
                adf_type=product_type,
                auxiliary_product_to_collection_identifier=auxiliary_product_to_collection_identifier,
                cql2_filter=cql2_filter,
            )
            await adf_conversion_task.with_options(name=f"convert {product_type} on the period [{start}-{stop}]")(
                flow_parameters,
            )
            start += duration

schedule_adf_conversion(owner_identifier, product_type, cql2_query_name, dta, dtb, period_in_hours, period_start, period_end, auxiliary_product_to_collection_identifier, satellite) async

Convert a single ADF data. - product_type: type of the product to convert. - start: start datetime of the period to convert. - stop: end datetime of the period to convert.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/adf/convert_adf_set.py

@task(name="schedule conversion")
async def schedule_adf_conversion(
    owner_identifier: str,
    product_type: str,
    cql2_query_name: str,
    dta: int,
    dtb: int,
    period_in_hours: int,
    period_start: datetime,
    period_end: datetime,
    auxiliary_product_to_collection_identifier: list[AuxiliaryProductMapping],
    satellite: str | None,
) -> None:
    """
    Convert a single ADF data.
     - product_type: type of the product to convert.
     - start: start datetime of the period to convert.
     - stop: end datetime of the period to convert.

    """
    logger = get_run_logger()
    logger.info("🧹 Computing cql2_filter without start_datetime and end_datetime...")
    cql2_filter_without_date = compute_cql2(cql2_query_name, dta, dtb, satellite)

    # Scheduling according to the period_in_hours
    rule: str
    if period_in_hours == 0:
        # special case where a single run is requested
        logger.info(
            f"Schedule the flow conversion to start at {period_start} for a time range [{period_start}-{period_end}].",
        )
        logger.debug(f"🧹 Associated cql2 filter is: {cql2_filter_without_date}")
        rule = (
            f"DTSTART:{period_start.strftime("%Y%m%dT%H%M%SZ")}\n"
            f"FREQ=HOURLY;UNTIL={period_start.strftime("%Y%m%dT%H%M%SZ")}"
        )

        await schedule_conversion_flow(
            owner_identifier,
            rule,
            cql2_filter_without_date,
            product_type,
            period_end - period_start,
            auxiliary_product_to_collection_identifier,
        )

    else:
        period_corrected: timedelta = min(period_end - period_start, timedelta(hours=period_in_hours))
        logger.debug(f"period_corrected = {period_corrected}")
        rule = (
            f"DTSTART:{period_start.strftime("%Y%m%dT%H%M%SZ")}\n"
            f"FREQ=HOURLY;INTERVAL={period_in_hours};UNTIL={period_end.strftime("%Y%m%dT%H%M%SZ")}"
        )
        logger.debug(f"rule = {rule}")

        logger.info(
            f"Schedule the flow conversion to start at {period_start} for a time range [{period_start}-{period_end}].",
        )
        logger.debug(f"🧹 Associated cql2 filter is: {cql2_filter_without_date}")
        await schedule_conversion_flow(
            owner_identifier,
            rule,
            cql2_filter_without_date,
            product_type,
            period_corrected,
            auxiliary_product_to_collection_identifier,
        )

schedule_conversion_flow(owner_identifier, rule, cql2_filter_without_date, product_type, period, auxiliary_product_to_collection_identifier) async

Schedule the conversion flow with the given parameters and scheduling rule.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/adf/convert_adf_set.py

async def schedule_conversion_flow(
    owner_identifier: str,
    rule: str,
    cql2_filter_without_date: dict,
    product_type: str,
    period: timedelta,
    auxiliary_product_to_collection_identifier: list[AuxiliaryProductMapping],
) -> None:
    """Schedule the conversion flow with the given parameters and scheduling rule."""

    logger = get_run_logger()

    # Retrieve the name of the workpool, GitHub URL and Branch
    work_pool_name: str | None = None
    github_repository: str
    github_branch: str | None = None
    async with get_client() as client:
        deployment = await client.read_deployment(runtime.deployment.id)
        work_pool_name = deployment.work_pool_name
        pull_steps = deployment.pull_steps or []

        for step in pull_steps:
            for step_name, step_config in step.items():
                if "git_clone" in step_name:
                    github_repository = step_config.get("repository")
                    github_branch = step_config.get("branch", "develop")  # "develop" by default
        logger.info(
            "Schedule the flow 'adf_conversion_scheduled' on the Work pool "
            f"name: {work_pool_name}, GitHub repository: {github_repository}, GitHub branch: {github_branch}",
        )

    flow_obj = await cast(
        Awaitable[Any],
        flow.from_source(
            source=GitRepository(url=github_repository, branch=github_branch),
            entrypoint=FLOW_TO_BE_SCHEDULED,
        ),
    )

    encoded_period: str = str(int(period.total_seconds()))
    await flow_obj.deploy(
        name=f"Convert ADF {product_type}",
        work_pool_name=work_pool_name,
        rrule=rule,
        tags=["auxip", "conversion"],
        parameters={
            "env": FlowEnvArgs(owner_id=owner_identifier),
            "adf_type": product_type,
            "cql2_filter_without_date": cql2_filter_without_date,
            "period": encoded_period,
            "auxiliary_product_to_collection_identifier": auxiliary_product_to_collection_identifier,
        },
    )