Add HITL

maintenance_flow.py

@@ -0,0 +1,53 @@
+from prefect import flow,get_run_logger, pause_flow_run, settings
+from prefect.blocks.notifications import SlackWebhook
+from prefect.context import get_run_context
+from DB_utils import set_maintenance_status,generate_empty_well,insert_maintenance_log
+from prefect.input import RunInput
+
+class UserInput(RunInput):
+    Maintenance_Completed : bool 
+    name: str
+    
+
+@flow
+def request_wells_maintenance(maintenance_type):
+    MESSAGE = "HITL request"
+    logger = get_run_logger()
+    slack_block = SlackWebhook.load("prefect-test")
+    message = str(MESSAGE)  
+    flow_run = get_run_context().flow_run
+
+    if flow_run and settings.PREFECT_UI_URL:
+        flow_run_url = (
+            f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
+        )
+        message += f"\n\nOT2-LCM requests a {maintenance_type}, please open the <{flow_run_url}|paused flow run>, complete with your user name and then click 'Resume'" 
+    slack_block.notify(message)
+    
+    description = f"""
+    OT2-LCM requests a {maintenance_type}!
+    Please finish the maintenance and enter your name:
+    """
+
+    user_input = pause_flow_run(
+        wait_for_input=UserInput.with_initial_data(
+            description=description
+        ),
+        timeout = 3000 #Should we have a timeout, or just let it waits until maintenance complete?
+    )
+    #Same as above, and even without this timeout handler, following database update codes will not run
+    if not user_input:
+        logger.warning("[DEBUG]Flow resumed automatically due to timeout.")
+        return "Maintenance flow timed out with no user interaction. " \
+        "The requested maintenance was not carried out."
+    
+    set_maintenance_status(maintenance_type,0)
+
+    if maintenance_type == "wellplate_maintenance":
+        generate_empty_well()
+        msg_out = f"Updating wells status on DB by {user_input.name}"
+        logger.info(msg_out)
+    
+    insert_maintenance_log(maintenance_type, user_input.name)
+
+    return msg_out

prefect.yaml

@@ -0,0 +1,34 @@
+# Welcome to your prefect.yaml file! You can use this file for storing and managing
+# configuration for deploying your flows. We recommend committing this file to source
+# control along with your flow code.
+
+# Generic metadata about this project
+name: OT-2-LCM-test
+prefect-version: 3.1.5
+
+# build section allows you to manage and build docker images
+build:
+
+# push section allows you to manage if and how this project is uploaded to remote locations
+push:
+
+# pull section allows you to provide instructions for cloning this project in remote locations
+pull:
+- prefect.deployments.steps.set_working_directory:
+    directory: /home/user/app
+
+# the deployments section allows you to provide configuration for deploying flows
+deployments:
+- name: wells-maintenance
+  version:
+  tags: []
+  concurrency_limit:
+  description:
+  entrypoint: maintenance_flow.py:request_wells_maintenance
+  parameters: {}
+  work_pool:
+    name: ot2-pool
+    work_queue_name:
+    job_variables: {}
+  enforce_parameter_schema: true
+  schedules: []

prefect_utils.py

@@ -0,0 +1,28 @@
+import subprocess
+import threading
+
+def start_prefect_worker(work_pool_name: str = "ot2-pool"):
+    def worker_runner():
+        print(f"Starting Prefect Worker for work pool: {work_pool_name}")
+        subprocess.run([
+            "prefect", "worker", "start",
+            "--pool", work_pool_name
+        ])
+    worker_thread = threading.Thread(target=worker_runner, daemon=True)
+    worker_thread.start()
+    print("Prefect Worker started in background thread.")
+
+
+from prefect.client import get_client
+
+async def trigger_maintenance_request(maintenance_type: str):
+    async with get_client() as client:
+        # find deployment id
+        deployment = await client.read_deployment_by_name(
+            name="request-wells-maintenance/wells-maintenance"
+        )
+
+        await client.create_flow_run_from_deployment(
+            deployment_id=deployment.id,
+            parameters={"maintenance_type": maintenance_type}
+        )

prefectignore

@@ -0,0 +1,41 @@
+# prefect artifacts
+.prefectignore
+
+# python artifacts
+__pycache__/
+*.py[cod]
+*$py.class
+*.egg-info/
+*.egg
+
+# Type checking artifacts
+.mypy_cache/
+.dmypy.json
+dmypy.json
+.pyre/
+
+# IPython
+profile_default/
+ipython_config.py
+*.ipynb_checkpoints/*
+
+# Environments
+.python-version
+.env
+.venv
+env/
+venv/
+
+# MacOS
+.DS_Store
+
+# Dask
+dask-worker-space/
+
+# Editors
+.idea/
+.vscode/
+
+# VCS
+.git/
+.hg/