rs_workflows/dpr_flow.md

<< Back to index

DPR flow implementation

clean_paths(paths, logger)

Delete directories or files listed in paths, logging outcomes.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of filesystem paths to remove.	required
logger		Prefect logger for informational messages.	required

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

def clean_paths(paths: list[str], logger) -> None:
    """Delete directories or files listed in paths, logging outcomes.

    Args:
        paths: List of filesystem paths to remove.
        logger: Prefect logger for informational messages.
    """
    logger.info(f"Cleaning up temporary paths: {paths}")
    for path in paths:
        try:
            if not osp.exists(path):
                logger.warning(f"Autoclean: path does not exist {path}, skipping.")
                continue
            if osp.isdir(path):
                shutil.rmtree(path)
                logger.info(f"Autoclean: removed directory {path}")
            else:
                logger.warning(f"Autoclean: expected directory but found file {path}, skipping.")
        except Exception as e:  # pylint: disable=broad-exception-caught
            logger.warning(f"Autoclean failed for the shared path {path}: {e}")

compute_eopf_origin_datetime(env, input_products, *, items=None)

Compute the maximum eopf:origin_datetime across all input products.

For each input product, this function retrieves the corresponding item from the catalog using its item ID and collection ID, extracts the eopf:origin_datetime property, and returns the latest (maximum) datetime value found.

If an item cannot be retrieved from the catalog, processing fails. Already retrieved items can be supplied to avoid duplicate catalog requests.

Parameters

env : object Execution environment object used to serialize and pass context to the catalog flow. input_products : Iterable[FlowInputProduct] Input products with item IDs and collection names. items : list[Item] | None Previously retrieved input STAC items, if available.

Returns

str ISO 8601 string representing the maximum eopf:origin_datetime found among all retrieved items. Raises ValueError if no origin datetimes are found.

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

def compute_eopf_origin_datetime(env, input_products, *, items: list[Item] | None = None) -> str:
    """
    Compute the maximum ``eopf:origin_datetime`` across all input products.

    For each input product, this function retrieves the corresponding item
    from the catalog using its item ID and collection ID, extracts the
    ``eopf:origin_datetime`` property, and returns the latest (maximum)
    datetime value found.

    If an item cannot be retrieved from the catalog, processing fails.
    Already retrieved items can be supplied to avoid duplicate catalog requests.

    Parameters
    ----------
    env : object
        Execution environment object used to serialize and pass context
        to the catalog flow.
    input_products : Iterable[FlowInputProduct]
        Input products with item IDs and collection names.
    items : list[Item] | None
        Previously retrieved input STAC items, if available.

    Returns
    -------
    str
        ISO 8601 string representing the maximum ``eopf:origin_datetime``
        found among all retrieved items. Raises ValueError if no origin
        datetimes are found.
    """
    logger = get_logger()
    if items is None:
        items = get_input_product_items(env, input_products)

    dates = [
        datetime.datetime.fromisoformat(origin_dt.replace("Z", "+00:00"))
        for item in items
        if (origin_dt := item.to_dict().get("properties", {}).get("eopf:origin_datetime"))
    ]

    try:
        max_eopf_datetime = max(dates).isoformat()
    except ValueError as ve:
        logger.exception("Failed to compute maximum eopf:origin_datetime")
        raise ValueError("Maximum eopf datetime could not be computed") from ve

    logger.info(f"Maximum eopf datetime computed from all items is {max_eopf_datetime}")
    return max_eopf_datetime

create_stac_item(eopf_origin_datetime, eopf_feature, s3_data_location, product_name, dpr_processor, zarr_zip_href=None)

Create a list of STAC Items from EOPF features and processing payload metadata.

This function builds STAC Items compliant with EOPF constraints by: - Injecting EOPF-specific properties into each feature - Attaching output product assets - Propagating origin datetimes from input products

Parameters:

Name	Type	Description	Default
eopf_features	list[dict]	List of GeoJSON-like feature dictionaries.	required
s3_data_location	str	Base S3 path where output products are stored.	required
product_name	str	Product name	required
dpr_processor	str	DPR processor name	required
zarr_zip_href	str | None	S3 path of the associated .zarr.zip product, if any. When set, an additional asset is added to the item.	None

Returns:

Type	Description
Item	list[Item]: List of constructed STAC Item objects.

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

def create_stac_item(
    eopf_origin_datetime,
    eopf_feature,
    s3_data_location,
    product_name: str,
    dpr_processor: str,
    zarr_zip_href: str | None = None,
) -> Item:
    """
    Create a list of STAC Items from EOPF features and processing payload metadata.

    This function builds STAC Items compliant with EOPF constraints by:
    - Injecting EOPF-specific properties into each feature
    - Attaching output product assets
    - Propagating origin datetimes from input products

    Args:
        eopf_features (list[dict]): List of GeoJSON-like feature dictionaries.
        s3_data_location (str): Base S3 path where output products are stored.
        product_name (str): Product name
        dpr_processor (str): DPR processor name
        zarr_zip_href (str | None): S3 path of the associated .zarr.zip product, if any.
            When set, an additional asset is added to the item.

    Returns:
        list[Item]: List of constructed STAC Item objects.
    """

    def build_item(
        feature_dict: dict,
        eopf_origin_datetimes,
        product_name,
        dpr_processor: str,
        assets: dict[str, Asset],
    ) -> Item:
        """
        Build a STAC Item from a feature dictionary.

        This function mutates the feature dictionary by injecting mandatory
        EOPF and STAC properties before constructing the Item.

        Args:
            feature_dict (dict): GeoJSON-like feature dictionary.
            eopf_origin_datetimes (str | list[str]): Origin datetime(s) derived
                from input EOPF products.

        Returns:
            Item: A STAC Item populated with geometry, properties, and extensions.
        """
        if eopf_origin_datetimes:
            feature_dict["properties"]["eopf:origin_datetime"] = eopf_origin_datetimes

        # C1.2 Ensure that all EOPF items have stac_version property set to "1.1.0"
        feature_dict["properties"]["stac_version"] = "1.1.0"

        # C1.3 Add stac_extensions following the list from the PRIP ICD §3.3.4
        # TODO: According to the 821 story, we have to:
        # - do not set stac_extension SAR for Sentinel-2 products "with instrument different from SRAL"
        # - do not set stac_extension SAR for Sentinel-3 products "with instrument different from SRAL"
        # Get in line with the story once clarified !
        stac_extensions: list[str] = []
        if dpr_processor == DprProcessor.S1L0.value:
            stac_extensions = [
                # TODO: We don't include the full list for now to avoid issues with catalog ingestion
                # This is because some extensions may require specific properties that are not properly
                # set by the DPR processor at this time.
                # "https://stac-extensions.github.io/sat/v1.1.0/schema.json",
                # "https://stac-extensions.github.io/processing/v1.2.0/schema.json",
                # "https://stac-extensions.github.io/product/v1.0.0/schema.json",
                # "https://stac-extensions.github.io/scientific/v1.0.0/schema.json",
                # "https://stac-extensions.github.io/eo/v2.0.0/schema.json",
                # "https://stac-extensions.github.io/grid/v1.1.0/schema.json",
                # "https://stac-extensions.github.io/view/v1.1.0/schema.json",
                # "https://stac-extensions.github.io/sar/v1.3.0/schema.json",
                # "https://cs-si.github.io/eopf-stac-extension/v1.2.0/schema.json",
                # "https://stac-extensions.github.io/timestamps/v1.1.0/schema.json",
                # "https://stac-extensions.github.io/authentication/v1.1.0/schema.json",
            ]
        # tempfix for OL2 with 'datetime': null in the feature_dict, we set it to the 'created' property if available
        if feature_dict["properties"].get("datetime") is None:
            feature_dict["properties"]["datetime"] = feature_dict["properties"]["created"]

        # STAC 1.1 allows bands on assets, but forbids them in Item properties.
        if dpr_processor in (DprProcessor.S3L1OLCI.value, DprProcessor.S3L2OLCI.value):
            bands = feature_dict["properties"].pop("bands", None)
            if bands is not None:
                assets[product_name].extra_fields["bands"] = bands

        return Item(
            id=product_name,
            geometry=feature_dict.get("geometry"),
            bbox=feature_dict.get("bbox"),
            datetime=datetime.datetime.fromisoformat(feature_dict["properties"]["datetime"]),
            properties=feature_dict["properties"],
            stac_extensions=stac_extensions,
            assets=assets,
        )

    def build_asset(path: str, product_name: str) -> Asset:
        """
        Build a STAC Asset representing a Zarr output product.

        Args:
            path (str): Full path or URL to the asset.
            title (str): Human-readable asset title.

        Returns:
            Asset: A STAC Asset configured for EOPF output products.
        """
        return Asset(
            href=path.rsplit("/", maxsplit=1)[0] if path.endswith(("/.zattrs", "/zarr.json")) else path,
            title=product_name,
            media_type="application/vnd+zarr",
            roles=["data", "metadata"],
            # TODO: The story RSPY-280 is implemented in the catalog to fill the auth:ref field
            # extra_fields={
            #     "auth:ref": "should be filled thanks to story RSPY-280",
            # },
        )

    assets = {product_name: build_asset(s3_data_location, product_name)}
    if zarr_zip_href:
        assets[f"{product_name}.zip"] = Asset(
            href=zarr_zip_href,
            title=f"{product_name}.zip",
            media_type="application/zip",
            roles=["data"],
        )

    item = build_item(
        eopf_feature,
        eopf_origin_datetime,
        product_name,
        dpr_processor,
        assets=assets,
    )
    return item

extract_products_and_zattrs(files, base_path, *, zarr_v3=True)

Extract product names and their corresponding .zattrs file paths.

With zarr_v3=True, also discover zarr.json and prefer it per product.

Filters a list of file paths to find .zattrs files located under product directories within a base path, following one of these structures: base_path/.zattrs base_path/product_name/.zattrs base_path/products/product_name/.zattrs

Parameters:

Name	Type	Description	Default
files	list[str]	List of file paths to search through.	required
base_path	str	The base directory path to strip from file paths.	required

Returns:

Type	Description
	A list of tuples, where each tuple contains: - product_name (str): The name of the product directory. - file (str): The full path to the .zattrs file.
	If base_path/.zattrs exists, it is treated as the product metadata
	and nested group metadata is ignored. Otherwise, includes files that are
	directly under a product directory, with an optional products
	container directory under the base path.

Example

files = [ ... "/data/products/product_a/.zattrs", ... "/data/products/product_b/.zattrs", ... "/data/products/product_c/subdir/file.txt" ... ] extract_products_and_zattrs(files, "/data/products") [('product_a', '/data/products/product_a/.zattrs'), ('product_b', '/data/products/product_b/.zattrs')]

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

def extract_products_and_zattrs(files: list[str], base_path: str, *, zarr_v3: bool = True):
    """Extract product names and their corresponding .zattrs file paths.

    With ``zarr_v3=True``, also discover zarr.json and prefer it per product.

    Filters a list of file paths to find .zattrs files located under product
    directories within a base path, following one of these structures:
    base_path/.zattrs
    base_path/product_name/.zattrs
    base_path/products/product_name/.zattrs

    Args:
        files: List of file paths to search through.
        base_path: The base directory path to strip from file paths.

    Returns:
        A list of tuples, where each tuple contains:
            - product_name (str): The name of the product directory.
            - file (str): The full path to the .zattrs file.

        If ``base_path/.zattrs`` exists, it is treated as the product metadata
        and nested group metadata is ignored. Otherwise, includes files that are
        directly under a product directory, with an optional ``products``
        container directory under the base path.

    Example:
        >>> files = [
        ...     "/data/products/product_a/.zattrs",
        ...     "/data/products/product_b/.zattrs",
        ...     "/data/products/product_c/subdir/file.txt"
        ... ]
        >>> extract_products_and_zattrs(files, "/data/products")
        [('product_a', '/data/products/product_a/.zattrs'),
         ('product_b', '/data/products/product_b/.zattrs')]
    """
    # OLCI CPM 3 writes root attributes in zarr.json. Prefer it over legacy metadata.
    metadata_names = ("zarr.json", ".zattrs") if zarr_v3 else (".zattrs",)
    dirs_and_attrs: dict = {}

    normalized_base_path = base_path.rstrip("/")

    for metadata_name in metadata_names:
        root_metadata = f"{normalized_base_path}/{metadata_name}"
        if root_metadata in files:
            product_dirname = normalized_base_path.rsplit("/", maxsplit=1)[-1]
            return [(product_dirname, root_metadata)]

    for file in files:
        if not file.startswith(normalized_base_path + "/"):
            continue

        rest = file[len(normalized_base_path) :].lstrip("/")  # noqa: E203
        parts = rest.split("/")

        # 1: base_path/product/.zattrs
        if len(parts) == 2 and parts[1] in metadata_names:
            product_dirname = parts[0]
            previous = dirs_and_attrs.get(product_dirname)
            if previous is None or file.endswith("/zarr.json"):
                dirs_and_attrs[product_dirname] = file

        # 2: base_path/products/product/.zattrs
        elif len(parts) == 3 and parts[0] == "products" and parts[2] in metadata_names:
            product_dirname = parts[1]
            previous = dirs_and_attrs.get(product_dirname)
            if previous is None or file.endswith("/zarr.json"):
                dirs_and_attrs[product_dirname] = file

    return list(dirs_and_attrs.items())

find_zarr_zip(zattrs_s3_location)

Return the S3 path of the <PRODUCT_NAME>.zarr.zip stored next to a Zarr product, if it exists.

Parameters:

Name	Type	Description	Default
zattrs_s3_location	str	S3 path of the product root .zattrs file.	required

Returns:

Type	Description
str | None	The S3 path of the zip file, or None if the zip is not present.

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

def find_zarr_zip(zattrs_s3_location: str) -> str | None:
    """Return the S3 path of the ``<PRODUCT_NAME>.zarr.zip`` stored next to a Zarr product, if it exists.

    Args:
        zattrs_s3_location: S3 path of the product root ``.zattrs`` file.

    Returns:
        The S3 path of the zip file, or None if the zip is not present.
    """
    zip_s3_path = zattrs_s3_location.removesuffix("/.zattrs") + ".zip"
    if not zip_s3_path.endswith(".zarr.zip"):
        return None

    s3_bucket, key = prefect_utils.get_s3_bucket(zip_s3_path)
    objects = s3_bucket._get_bucket_resource().objects  # pylint: disable=protected-access
    if any(obj.key == key for obj in objects.filter(Prefix=key)):
        return zip_s3_path
    return None

get_input_product_items(env, input_products)

Retrieve source STAC items, excluding auxiliary inputs, once per update.

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

def get_input_product_items(env, input_products) -> list[Item]:
    """Retrieve source STAC items, excluding auxiliary inputs, once per update."""
    logger = get_logger()
    items: list[Item] = []
    if not input_products:
        logger.error("No valid input products found to compute eopf:origin_datetime. Exit")
        raise RuntimeError("No valid input products found to compute eopf:origin_datetime")

    for input_product in input_products:
        item_id = input_product.item_id
        collection_name = input_product.collection_name
        try:
            future = catalog_flow.get_item.submit(
                env.serialize(),
                collection_name,
                item_id,
            )
            # Prefect resolves the async task before returning its result.
            item = cast(Item | None, future.result())
            if not item:
                logger.error(
                    f"Expected valid input product item {item_id} was not found"
                    " to compute eopf:origin_datetime. Exit",
                )
                raise RuntimeError(
                    f"Expected valid input product item {item_id} was not found to compute eopf:origin_datetime",
                )
            items.append(item)
        except RuntimeError as rte:
            logger.exception(f"Failed to get item '{item_id}' from collection '{collection_name}'")
            raise RuntimeError("No valid items found to compute eopf:origin_datetime") from rte

    logger.info(f"Items matching input found in catalog: {len(items)}")

    return items

read_zattrs_sync(path)

Read a .zattrs file from S3 synchronously and return the parsed JSON, in memory.

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

def read_zattrs_sync(path: str):
    """Read a `.zattrs` file from S3 synchronously and return the parsed JSON, in memory."""
    return json.loads(prefect_utils.s3_read_bytes(path, _sync=True))  # type: ignore

run_processor(env, processor, payload, cluster_info, s3_payload_run, input_products) async

Run the DPR processor.

Parameters:

Name	Type	Description	Default
env	FlowEnvArgs	Prefect flow environment	required
processor	str	DPR processor name	required
s3_payload_run	str	S3 bucket location of the output final DPR payload file.	required

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

@task(name="run-processor")
async def run_processor(
    env: FlowEnvArgs,
    processor: str,
    payload: PayloadSchema,
    cluster_info: ClusterInfo,
    s3_payload_run: str,
    input_products: list[FlowInputProduct],
) -> list[DprProcessedItemMetadata]:
    """
    Run the DPR processor.

    Args:
        env: Prefect flow environment
        processor: DPR processor name
        s3_payload_run: S3 bucket location of the output final DPR payload file.
    """
    logger = get_logger()

    # Init flow environment and opentelemetry span
    flow_env = FlowEnv(env)
    with flow_env.start_span(__name__, "run-processor"):
        if payload.io is None:
            raise ValueError("Payload io configuration is missing.")
        # First, remove the output products that are not final products from
        # the payload to avoid triggering the catalog registration for them
        # Create a temporary list for keeping track of products to keep
        kept_products = []
        # List of paths to delete if autoclean is enabled
        paths_to_delete: list[str] = []

        # Iterate over the original products
        for prod in payload.io.output_products:
            if prod.final_product:
                kept_products.append(prod)
            else:
                logger.info(f"Output product {prod.id} is not marked as final_product, skipping catalog registration.")
            # Record autoclean path for any product with autoclean=True
            if prod.autoclean and prod.path not in paths_to_delete:
                paths_to_delete.append(prod.path)

        # Update the original output_products list with the kept products
        payload.io.output_products[:] = kept_products

        record_performance_indicators(  # type: ignore
            start_date=datetime.datetime.now(),
            status="OK",
            dpr_processing_input_stac_items=s3_payload_run,
            payload=payload,
            dpr_processor_name=processor,
        )
        # Trigger the processor run from the dpr service
        dpr_client: DprClient = flow_env.rs_client.get_dpr_client()
        start_time = time.time()
        s3_payload_dir = osp.dirname(s3_payload_run)
        s3_payload_filename = osp.basename(s3_payload_run)
        is_local_path = not s3_payload_run.startswith(("s3://", "S3://"))
        logger.info(
            f'Triggering DPR processor {processor!r} with payload file "'
            f"{s3_payload_filename!r} in {'local' if is_local_path else 'S3'} dir "
            f"{s3_payload_dir!r} from s3_payload_run {s3_payload_run!r}",
        )
        logger.info(f"Triggering DPR processor {processor!r}")
        job_status = dpr_client.run_process(
            process=processor,
            cluster_info=cluster_info,
            s3_config_dir=s3_payload_dir,
            payload_subpath=s3_payload_filename,
            s3_report_dir=s3_payload_dir,
        )
        try:
            dpr_client.wait_for_job(job_status, logger, f"{processor!r} processor")
        finally:
            logger.info(f"Processor execution time: {str(timedelta(seconds=time.time() - start_time))}")
            if is_local_path:
                # For local paths, the processor writes reports to the local directory directly
                local_report_dir = osp.join(s3_payload_dir, "reports")
                local_log_file = osp.join(
                    local_report_dir,
                    Path(s3_payload_filename).with_suffix(".processor.log").name,
                )
                logger.info(f"Reading processor log file from local path: {local_log_file!r}")
                try:
                    async with await anyio.open_file(local_log_file, encoding="utf-8") as opened:
                        s3_log_file = await opened.read()
                        for entry in parse_logs(s3_log_file):
                            level = entry["level"].strip().lower()
                            getattr(logger, level, logger.info)(entry["message"])
                except FileNotFoundError:
                    logger.info(f"No processor log file was found at: {local_log_file!r}")
            else:
                # Download reports folder from the s3 bucket
                with tempfile.TemporaryDirectory() as tmpdir:
                    await prefect_utils.s3_download_dir(s3_payload_dir, tmpdir)
                    local_log_file = osp.join(
                        tmpdir,
                        "reports",
                        Path(s3_payload_filename).with_suffix(".processor.log").name,
                    )
                    try:
                        async with await anyio.open_file(local_log_file, encoding="utf-8") as opened:
                            s3_log_file = await opened.read()
                            for entry in parse_logs(s3_log_file):
                                level = entry["level"].strip().lower()
                                getattr(logger, level, logger.info)(entry["message"])
                    except FileNotFoundError:
                        logger.info(f"No processor log file was uploaded under: {s3_payload_dir!r}")
            # After processing, clean up autoclean paths. IMPORTANT : the shared disk has to be mounted
            # in the current flow environment (prefect worker) for this to work ! So, the shared_disk has to be
            # mounted in both dask worker environment (where the processor runs) and in the prefect worker
            # environment (where this flow runs)
            clean_paths(paths_to_delete, logger)

        items_metadata = update_eopf_assets(flow_env, input_products, payload, processor)
        eopf_stac_items = [asset.stac_item for asset in items_metadata]
        eopf_types = [asset.product_type for asset in items_metadata]

        # Wait for the job to finish
        record_performance_indicators(  # type: ignore
            stop_date=datetime.datetime.now(),
            status="OK",
            stac_items=eopf_stac_items,
            payload=payload,
            dpr_processor_name=processor,
            eopf_types=eopf_types,
        )
        return items_metadata

s3_list(s3_prefix)

List all S3 objects under a prefix without downloading.

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

def s3_list(s3_prefix: str):
    """List all S3 objects under a prefix without downloading."""
    s3_bucket, prefix = prefect_utils.get_s3_bucket(s3_prefix)
    objects = s3_bucket._get_bucket_resource().objects  # pylint: disable=protected-access

    return [f"s3://{s3_bucket.bucket_name}/{obj.key}" for obj in objects.filter(Prefix=prefix.rstrip("/") + "/")]

update_eopf_assets(env, input_products, payload, dpr_processor)

Update EOPF assets by extracting metadata and creating STAC items.

This Prefect task processes output products from a DPR (Data Processing Request) workflow, extracts EOPF (Earth Observation Processing Framework) metadata from .zattrs files, and generates STAC (SpatioTemporal Asset Catalog) items for each discovered product.

Workflow

1. Lists all .zattrs files in the output product paths
2. Reads and validates EOPF discovery metadata from each .zattrs file
3. Extracts product type information
4. Creates corresponding STAC items for catalog registration

Parameters:

Name	Type	Description	Default
env		Environment configuration object containing runtime settings.	required
input_products	list[FlowInputProduct]	Source product references (excluding auxiliary inputs).	required
payload	PayloadSchema	PayloadSchema object containing I/O configuration, including output_product paths to scan for .zattrs files.	required
dpr_processor	str	str DPR processor name	required

Returns:

Type	Description
list[DprProcessedItemMetadata]	A list of DprProcessedItemMetadata objects, each containing: - stac_item (Item): STAC item created from EOPF metadata. - product_type (str): Product type extracted from the .zattrs file. - output_product_id (str): The ID of the original output product.

Raises:

Type	Description
RuntimeError	If any .zattrs file cannot be read or does not contain required EOPF discovery metadata (stac_discovery.properties).

Notes

• Requires .zattrs files to follow EOPF metadata conventions
• Each .zattrs file must contain stac_discovery.properties.product:type
• Uses S3 storage backend (via s3_list and read_zattrs_sync functions)

Source code in docs/rs-client-libraries/rs_workflows/dpr_flow.py

@task(name="Update eopf assets")
def update_eopf_assets(
    env,
    input_products: list[FlowInputProduct],
    payload: PayloadSchema,
    dpr_processor: str,
) -> list[DprProcessedItemMetadata]:
    """Update EOPF assets by extracting metadata and creating STAC items.

    This Prefect task processes output products from a DPR (Data Processing Request)
    workflow, extracts EOPF (Earth Observation Processing Framework) metadata from
    .zattrs files, and generates STAC (SpatioTemporal Asset Catalog) items for
    each discovered product.

    Workflow:
        1. Lists all .zattrs files in the output product paths
        2. Reads and validates EOPF discovery metadata from each .zattrs file
        3. Extracts product type information
        4. Creates corresponding STAC items for catalog registration

    Args:
        env: Environment configuration object containing runtime settings.
        input_products: Source product references (excluding auxiliary inputs).
        payload: PayloadSchema object containing I/O configuration, including
            output_product paths to scan for .zattrs files.
        dpr_processor: str
            DPR processor name

    Returns:
        A list of DprProcessedItemMetadata objects, each containing:
            - stac_item (Item): STAC item created from EOPF metadata.
            - product_type (str): Product type extracted from the .zattrs file.
            - output_product_id (str): The ID of the original output product.

    Raises:
        RuntimeError: If any .zattrs file cannot be read or does not contain
            required EOPF discovery metadata (stac_discovery.properties).

    Notes:
        - Requires .zattrs files to follow EOPF metadata conventions
        - Each .zattrs file must contain stac_discovery.properties.product:type
        - Uses S3 storage backend (via s3_list and read_zattrs_sync functions)
    """
    logger = get_logger()
    logger.info("Starting EOPF asset update.")
    logger.info(f"Payload received: {payload}")
    logger.info(f"Input products: {input_products}")

    if payload.io is None:
        raise RuntimeError("Payload io configuration is missing.")
    # Get all .zattrs files found in the output products paths
    zattrs_list = []
    for prod in payload.io.output_products:
        path = prod.path
        # Keep track of which output product ID each .zattrs belongs to
        new_zattrs = extract_products_and_zattrs(s3_list(path), path)
        for name, loc in new_zattrs:
            zattrs_list.append((name, loc, prod.id))
        logger.info(
            f"The output product section {prod.id} has been added to the list, "
            "and all product types found at this path will be published to the catalog. "
            f"The path is {path}",
        )

    # List & extract
    logger.info(
        f"Found {len(zattrs_list)} .zattrs files from {len(payload.io.output_products)} "
        f"output product sections from payload. The list with products to be published: {zattrs_list}",
    )

    input_items = get_input_product_items(env, input_products) if input_products and zattrs_list else []
    # Copy the first available station, preserving the input property value.
    acquisition_station_properties = {}
    for input_item in input_items:
        properties = input_item.to_dict().get("properties", {})
        if "sat:acquisition_station" in properties:
            acquisition_station_properties["sat:acquisition_station"] = properties["sat:acquisition_station"]
            logger.info(
                f"Copying sat:acquisition_station={properties['sat:acquisition_station']!r} "
                "from input products to output STAC items.",
            )
            break

    # C1.1 Add the property eopf:origin_datetime with value equal to the maximum
    # eopf:origin_datetime among all input products (excluding ADFS inputs)
    # Note: input_products != input_adfs
    # disabled for mockup
    if dpr_processor.lower() in ["mockup"]:
        eopf_origin_datetime = "2026-01-01T00:00:00Z"
    elif input_products and zattrs_list:
        eopf_origin_datetime = compute_eopf_origin_datetime(env, input_products, items=input_items)
    else:
        eopf_origin_datetime = None

    items_metadata = []
    for product_name, zattrs_s3_location, output_product_id in zattrs_list:
        logger.info(f"Product = {product_name} | zattrs = {zattrs_s3_location}")
        # Read metadata
        zattrs_data = read_zattrs_sync(zattrs_s3_location)
        if not zattrs_data:
            logger.error(f"Could not read .zattrs file {zattrs_s3_location}. Exiting.")
            raise RuntimeError(f"Could not read .zattrs file {zattrs_s3_location}. Exiting.")
        # Zarr v3 stores root attributes inside "attributes" for every processor.
        # Legacy .zattrs files contain those attributes directly.
        zattrs_data = zattrs_data.get("attributes", zattrs_data)
        logger.debug(f"DPR processor attributes {zattrs_data}")

        # Extract EOPF info
        if "stac_discovery" not in zattrs_data or "properties" not in zattrs_data["stac_discovery"]:
            logger.error(f".zattrs file {zattrs_s3_location} does not contain EOPF discovery metadata. Exiting.")
            raise RuntimeError(f".zattrs file {zattrs_s3_location} does not contain EOPF discovery metadata. Exiting.")

        product_type = zattrs_data["stac_discovery"]["properties"].get("product:type", None)
        logger.info(f"Extracted EOPF product type: {product_type}")

        eopf_item = zattrs_data["stac_discovery"]
        logger.debug(f"EOPF discovery metadata extracted: {eopf_item}")

        eopf_item["properties"].update(acquisition_station_properties)

        # Build STAC items
        zarr_zip_href = find_zarr_zip(zattrs_s3_location)
        if zarr_zip_href:
            logger.info(f"Found zarr.zip associated with product {product_name}: {zarr_zip_href}")
        stac_item = create_stac_item(
            eopf_origin_datetime,
            eopf_item,
            zattrs_s3_location,
            product_name,
            dpr_processor,
            zarr_zip_href,
        )

        items_metadata.append(
            DprProcessedItemMetadata(
                stac_item=stac_item,
                product_type=product_type,
                output_product_id=output_product_id,
            ),
        )
        logger.info(f"Added one stac item metadata to the already existing list. Length: {len(items_metadata)}.")

    logger.info(f"List with stac_items to be published: {items_metadata}.")
    return items_metadata