Aggregator and launcher for Flexpart

[ninaburg]

Purpose:

Trigger both preprocessing and Flexpart from Aviso

Changes:

• Extracted Aviso Files: Moved Aviso and service settings from FlexPrep (see PR #18) into this new orchestrator repo.
• Flexpart Aggregator: Added an aggregator for Flexpart that triggers after the Flexprep preprocessing step.
• Flexpart Launch: Implemented the launch of Flexpart by calling the containerize branch from the Flexpart repo (currently in development).

[victoria-cherkas]

I think it would be nicer if all reading and writing to the db was handled in this repo, so flexprep does not need to know about the DB at all. Then the workflow logic is separated from the processing.

[victoria-cherkas]

Can we remove the comment here? Also fine to remove the whole of pre-commit since linting/mypy is checked in CICD ?

[ninaburg]

Good idea to align how the linting is done with how it's done in CI/CD. I removed the dependencies for pre-commit

[victoria-cherkas]

Suggested change

	The flex-container-orchestrator manages Aviso notifications and triggers the flexprep and flexpart-ifs containers,
	as well as the file aggregation script for Flexpart.
	The flex-container-orchestrator manages the event driven workflow Flexpart IFS workflow, based on events from Aviso. The repo coordinates both flexprep (pre-processing of raw IFS data) and flexpart-ifs containers, ensuring all required lead time data is processed before launching Flexpart.

[victoria-cherkas]

Cool didnt know about this!

[victoria-cherkas]

how come we need uvicorn?

[ninaburg]

No, we don't. This came with the mch python template and I updated it now with:
$ poetry run python3 flex_container_orchestrator/main.py --date {date} --time {time} --step {step} --location {location}

[victoria-cherkas]

remove comment? or add to dev dependencies

[ninaburg]

The comment is removed. mchbuild doesn't work outside of meteoswiss network, so I don't see the value of adding it here. For testing, one can install it with pip.

[victoria-cherkas]

We should describe here what is the difference between lead times and all steps, to me it is not immediately clear

[ninaburg]

Refactored the description of the returns:

    Returns:
        tuple[list[list[str]], list[list[datetime]], set[str]]:
            - all_input_forecasts: A nested list where each sublist contains forecast labels
              for each Flexpart run in the format "{reference_time}_{step}".
            - all_flexpart_leadtimes: A nested list where each sublist contains datetime objects
              representing the leadtimes (reference_time + step) for each Flexpart run.
            - all_input_forecasts_set: A set of unique forecasts in the format "{reference_time}_{step}"
              required for Flexpart simulations.

[victoria-cherkas]

all_list_ltf, all_list_lt <- i think the naming of these could be improved, the similarity makes it hard to read

[ninaburg]

Changed it to: input_forecasts, flexpart_leadtimes

[victoria-cherkas]

Suggested change

	for item in items_f:
	if item[0]: # processed == True
	frt_str = (
	frt.strftime("%Y%m%d%H%M")
	if isinstance(frt, datetime.datetime)
	else str(frt)
	)
	processed_items.add(frt_str + f"{int(item[1]):02}")
	for processed, step in items_f:
	if processed:
	frt_str = (
	frt.strftime("%Y%m%d%H%M")
	if isinstance(frt, datetime.datetime)
	else str(frt)
	)
	processed_items.add(frt_str + f"{int(step):02}")

[victoria-cherkas]

The naming of these variables is a bit confusing

[ninaburg]

Changed it to:
all_input_forecasts_set
all_input_forecasts
all_flexpart_leadtimes

[victoria-cherkas]

I like how here (within main()) in the flow is clear and it exits if the lead time is not processed. But in the rest of this function, it is not clear where in the logic we discover that the forecast is not fully processed / ready to run Flexart,

[ninaburg]

I agree and have refactored the module. The main() function is now run_aggregator, and I've included comments for better clarity.