diff --git a/automation/canfar_polling.py b/automation/canfar_polling.py index b5ac308..6cb131c 100644 --- a/automation/canfar_polling.py +++ b/automation/canfar_polling.py @@ -16,6 +16,9 @@ async def submit_canfar_job(): # then submit job launch_session(...) + # using asyncio.run() to manage the event loop + asyncio.run(submit_canfar_job()) + Rules: - Only flow-runs in Prefect state RUNNING are checked. - Only flow-runs tagged with "canfar_session:" are checked. diff --git a/possum_pipeline_control/check_status_and_launch_3Dpipeline_v2.py b/possum_pipeline_control/check_status_and_launch_3Dpipeline_v2.py index 4f232ed..d324ea5 100644 --- a/possum_pipeline_control/check_status_and_launch_3Dpipeline_v2.py +++ b/possum_pipeline_control/check_status_and_launch_3Dpipeline_v2.py @@ -23,6 +23,7 @@ @author: Erik Osinga """ import argparse +import asyncio import os import subprocess from datetime import datetime, timedelta, timezone @@ -510,5 +511,6 @@ def run_prefect_db_backup(): help="Path to .env file with database connection parameters.", ) args = parser.parse_args() - launch_band1_3Dpipeline(args.database_config_path) + + asyncio.run(launch_band1_3Dpipeline(args.database_config_path)) diff --git a/possum_pipeline_control/launch_1Dpipeline_PartialTiles_band1.py b/possum_pipeline_control/launch_1Dpipeline_PartialTiles_band1.py index bb82d21..7bb1aba 100644 --- a/possum_pipeline_control/launch_1Dpipeline_PartialTiles_band1.py +++ b/possum_pipeline_control/launch_1Dpipeline_PartialTiles_band1.py @@ -1,5 +1,6 @@ import argparse import ast +import asyncio import os from datetime import datetime @@ -50,7 +51,7 @@ def launch_session(run_name, field_ID, tilenumbers, SBnumber, image, cores, ram) @flow(name="launch_1d_partialtiles", log_prints=True) -def main_flow(field_ID, tilenumbers, SBnumber): +async def main_flow(field_ID, tilenumbers, SBnumber): timestr = ((datetime.now().strftime("%d/%m/%Y %H:%M:%S"))[11:]).replace( ":", "-" ) # ":" is not allowed character @@ -66,12 +67,12 @@ def main_flow(field_ID, tilenumbers, SBnumber): ram = 20 * number_of_tiles # check if there are any stuck jobs before launching new ones - reconcile_summary = canfar_polling.reconcile_running_prefect_with_canfar_task() + reconcile_summary = await canfar_polling.reconcile_running_prefect_with_canfar_task() - # Check allowed values at canfar.net/science-portal, 10, 20, 30, 40 GB should be allowed - launch_session( - run_name, field_ID, tilenumbers, SBnumber, image, cores, ram - ) + # # Check allowed values at canfar.net/science-portal, 10, 20, 30, 40 GB should be allowed + # launch_session( + # run_name, field_ID, tilenumbers, SBnumber, image, cores, ram + # ) if __name__ == "__main__": parser = argparse.ArgumentParser( @@ -92,4 +93,4 @@ def main_flow(field_ID, tilenumbers, SBnumber): tilenumbers = args.tilenumbers SBnumber = args.SBnumber - main_flow(field_ID, tilenumbers, SBnumber) + res = asyncio.run(main_flow(field_ID, tilenumbers, SBnumber))