Skip to content

rs_workflows/on_demand/common/l0_last_steps.md

<< Back to index

common Level-0 processing.

get_session_collections(settings)

Return the catalog collections of the staged sessions, mandatory in the settings of a Level-0 processing.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/common/l0_last_steps.py 
44
45
46
47
48
def get_session_collections(settings: ProcessorSettings) -> list[str]:
    """Return the catalog collections of the staged sessions, mandatory in the settings of a Level-0 processing."""
    if not settings.input_collections:
        raise ValueError("❌ The settings of the Level-0 processing must define the 'input_collections'")
    return settings.input_collections

process_l0_last_steps(mission, session, env_args, settings, workflow, verbose) async

Final processing steps that are common to all missions.

Returns:

Type Description
list[dict[str, Any]] | None

The STAC items of the generated products, or None if the session is not in the catalog.

Raises:

Type Description
ValueError

If the session has no valid 'datetime' property.
Source code in docs/rs-client-libraries/rs_workflows/on_demand/common/l0_last_steps.py 
 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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def process_l0_last_steps(
    mission: str,
    session: str,
    env_args: FlowEnvArgs,
    settings: ProcessorSettings,
    workflow: WorkflowType,
    verbose: bool,
) -> list[dict[str, Any]] | None:
    """
    Final processing steps that are common to all missions.

    Returns:
        The STAC items of the generated products, or None if the session is not in the catalog.

    Raises:
        ValueError: If the session has no valid 'datetime' property.
    """
    logger = get_logger()
    logger.info(f"Starting L0 last steps: mission={mission!r}, session={session!r}, verbose={verbose!r}")
    session_collections = get_session_collections(settings)

    flow_env = FlowEnv(env_args)
    with flow_env.start_span(__name__, f"sentinel{mission}-level0-processing"):
        logger.info(f"Looking up catalog session: session={session!r}, collections={session_collections!r}")
        item_session: Item | None = await get_single_catalog_item(flow_env, session, session_collections)
        if not item_session:
            logger.error("❌ Session %r was not found; DPR processing cannot be launched.", session)
            return None
        logger.info(f"✅ The session {session} has been found in the catalog.")
        logger.info("Catalog item id=%r, properties=%r", item_session.id, item_session.properties)

        # Session datetime date
        published = item_session.properties.get("datetime")
        if not isinstance(published, str):
            raise ValueError("Missing or invalid 'datetime' property in item_session")
        session_datetime = datetime.fromisoformat(published)

        input_products = [
            FlowInputProduct(
                name=L0_PROCESSINGS[mission][1],
                item_id=session,
                collection_name=str(item_session.collection_id),
            ),
        ]
        l0_result = await call_dpr_flow(
            env_args,
            settings=settings,
            input_products=input_products,
            external_variables={
                "start_datetime": session_datetime,
                "end_datetime": session_datetime,
                "satellite": f"sentinel-{mission}{session[2].lower()}",
            },
            workflow=workflow,
        )
        logger.info(f"L0 processing completed for mission={mission!r}, session={session!r}: {len(l0_result)} products")
        return l0_result

process_staged_session(mission, session, env_args, force_systematic, verbose) async

Level-0 processing of a session already staged in the catalog: read the settings of the processing of the mission, then run the final processing steps.

Returns:

Type Description
list[dict[str, Any]] | None

The STAC items of the generated products, or None if the session is not in the catalog.
Source code in docs/rs-client-libraries/rs_workflows/on_demand/common/l0_last_steps.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def process_staged_session(
    mission: str,
    session: str,
    env_args: FlowEnvArgs,
    force_systematic: bool,
    verbose: bool,
) -> list[dict[str, Any]] | None:
    """
    Level-0 processing of a session already staged in the catalog: read the settings of the processing of the
    mission, then run the final processing steps.

    Returns:
        The STAC items of the generated products, or None if the session is not in the catalog.
    """
    settings = await load_processor_settings(L0_PROCESSINGS[mission][0])
    return await process_l0_last_steps(
        mission=mission,
        session=session,
        env_args=env_args,
        settings=settings,
        workflow=WorkflowType.of_run(force_systematic),
        verbose=verbose,
    )