From 11ca0cab61ee818e252c971d310ea0c382653b34 Mon Sep 17 00:00:00 2001 From: Wu-Jung Lee Date: Sun, 23 Aug 2026 13:39:16 -0700 Subject: [PATCH] remove use of deployment_already_running --- src/echodataflow/flows/flows_acoustics.py | 17 +---------------- ...{test_flow_raw2sv.py => test_flow_raw2Sv.py} | 16 ---------------- 2 files changed, 1 insertion(+), 32 deletions(-) rename tests/{test_flow_raw2sv.py => test_flow_raw2Sv.py} (92%) diff --git a/src/echodataflow/flows/flows_acoustics.py b/src/echodataflow/flows/flows_acoustics.py index 0f18a06f..fc5f9b4f 100644 --- a/src/echodataflow/flows/flows_acoustics.py +++ b/src/echodataflow/flows/flows_acoustics.py @@ -10,11 +10,10 @@ from prefect import flow, get_run_logger, get_client from prefect.futures import as_completed -from prefect.states import Cancelled, Failed +from prefect.states import Failed from prefect import runtime from prefect.events import emit_event -from echodataflow.flows.flows_helper import deployment_already_running from echodataflow.deployment.task_runners import dask_task_runner_from_environment from echodataflow.utils.manifests import ( MVBS_COLUMNS_POSTPROCESSING, @@ -77,20 +76,6 @@ def flow_raw2Sv( add_splitbeam_angle: bool = False, ): - # Check if the deployment is already running - already_running = asyncio.run(deployment_already_running()) - if already_running: - - async def cancel_run(): - async with get_client() as client: - await client.set_flow_run_state( - flow_run_id=runtime.flow_run.id, - state=Cancelled(message="Another instance of this flow is already running"), - ) - - asyncio.run(cancel_run()) - return # exit the flow early - # Assemble paths path_Sv_zarr = Path(path_main) / "Sv" file_Sv_csv = Path(path_main) / file_Sv_csv diff --git a/tests/test_flow_raw2sv.py b/tests/test_flow_raw2Sv.py similarity index 92% rename from tests/test_flow_raw2sv.py rename to tests/test_flow_raw2Sv.py index 714a34be..3298596d 100644 --- a/tests/test_flow_raw2sv.py +++ b/tests/test_flow_raw2Sv.py @@ -8,10 +8,6 @@ from echodataflow.flows import flows_acoustics -async def _false_async(): - return False - - def test_flow_raw2sv_uses_processing_ledger(monkeypatch, tmp_path): raw_file = tmp_path / "example.raw" @@ -20,12 +16,6 @@ def test_flow_raw2sv_uses_processing_ledger(monkeypatch, tmp_path): "completed": [], } - monkeypatch.setattr( - flows_acoustics, - "deployment_already_running", - lambda: _false_async(), - ) - monkeypatch.setattr( flows_acoustics, "initialize_ledger", @@ -98,12 +88,6 @@ def test_flow_raw2sv_marks_failed(monkeypatch, tmp_path): "failed": [], } - monkeypatch.setattr( - flows_acoustics, - "deployment_already_running", - lambda: _false_async(), - ) - monkeypatch.setattr( flows_acoustics, "initialize_ledger",