Skip to content

rs_workflows/on_demand/common/l0.md

<< Back to index

common Level-0 processing.

process_l0(session, env_args, force_systematic=False, verbose=False) async

This is the generic l0 processing flow, started by the systematic processing after the staging of a session. It performs common L0 task like retrieving session from catalog an staging it from cadip if needed. Then it runs the Level-0 flow of the mission (process-s1-l0 or process-s3-l0) as a sub-flow, with the settings of the s<mission>l0-default-setting Prefect variable.

Parameters:

Name Type Description Default
session str

Identifier of the session.

 required
env_args FlowEnvArgs

Flow environment arguments.

 required
force_systematic bool

Run the processing as a systematic one, so that the automations start the next processing of the chain from the events of the generated products.

 False
verbose bool

not used yet.

 False
Source code in docs/rs-client-libraries/rs_workflows/on_demand/common/l0.py 
 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
 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
@flow(name="process-l0")
async def process_l0(
    session: str,
    env_args: FlowEnvArgs,
    force_systematic: bool = False,
    verbose: bool = False,
) -> list[dict[str, Any]] | None:
    """
    This is the generic l0 processing flow, started by the systematic processing after the staging of a session.
    It performs common L0 task like retrieving session from catalog an staging it from cadip if needed.
    Then it runs the Level-0 flow of the mission (`process-s1-l0` or `process-s3-l0`) as a sub-flow, with the
    settings of the `s<mission>l0-default-setting` Prefect variable.

    Args:
        session: Identifier of the session.
        env_args: Flow environment arguments.
        force_systematic: Run the processing as a systematic one, so that the automations start the next processing
            of the chain from the events of the generated products.
        verbose: not used yet.
    """
    logger = get_logger()
    logger.info(f"Mode verbose is set to {verbose}")

    # Check session name format
    pattern = re.compile(r"^S[123]._")
    if not pattern.match(session):
        logger.error("❌ Bad Sentinel-1,2,3 session name.")
        raise ValueError(f"Invalid session name: '{session}'")

    # We detect the mission
    mission: str = session[1]
    if mission not in L0_FLOWS:
        raise ValueError(f"❌ No Level-0 processing for Sentinel-{mission}")
    logger.info(f"✅ Sentinel-{mission} session name is correct.")

    # Read the settings and check the Dask clusters before searching the session
    settings = await load_processor_settings(L0_PROCESSINGS[mission][0])
    session_collections = get_session_collections(settings)

    flow_env = FlowEnv(env_args)
    with flow_env.start_span(__name__, "level0-processing"):
        found = False

        # Try to retrieve the session on the collection
        item_session: Item | None = await get_single_catalog_item(flow_env, session, session_collections)

        # If the session is not on the rs-catalog, we will try to stage it
        if item_session:
            found = True
            unpublished, unpublished_date = is_unpublished(item_session)
            if unpublished:
                logger.error(f"❌ The session '{session}' has been unpublished on {unpublished_date}")
                raise ValueError(f"'{session}' has been unpublished on {unpublished_date}")
            published, _ = is_published(item_session)
            if not published:
                logger.error(f"❌ The session '{session}' has not been published yet")
                raise ValueError(f"'{session}' has not been published")
        else:
            logger.info(f"📥 Try to stage session {session} from {mission} stations :{settings.cadip_collections}")
            station = await get_cadip_station(
                flow_env,
                session,
                settings.cadip_collections,
            )
            if station is not None:
                found = await stage_session_common(flow_env, station, session)

        # The session is staged at this step.
        # We can run the flow of the mission, that reads its settings again as it can also be run alone
        logger.info(f"🚧 We start Sentinel-{mission} processing.")
        if not found:
            return None
        return await L0_FLOWS[mission](
            session=session,
            env_args=env_args,
            force_systematic=force_systematic,
            verbose=verbose,
        )