update maintenance code

.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/

DB_utls.py

@@ -1,6 +1,7 @@
-#from lambda_mongo_utils import send_data_to_mongodb
 from pymongo import MongoClient
 from datetime import datetime
+import pytz
+from prefect_utils import trigger_maintenance_request
 from prefect import task
 import pandas as pd
 import os
@@ -56,7 +57,6 @@ def update_used_wells(used_wells):
     # close connection
     dbclient.close()
 
-@task
 def find_unused_wells():
     dbclient = MongoClient(connection_string)
     db = dbclient["LCM-OT-2-SLD"]  
@@ -77,9 +77,13 @@ def find_unused_wells():
         empty_wells = sorted(df["well"].tolist(), key=well_sort_key)
     else:
         empty_wells = []
-    #print(empty_wells)
-    
-    # close connection
+
+    if len(empty_wells) == 1:
+        maintenance_type =  "wellplate_maintenance"
+        if check_maintenance_status(maintenance_type) == 0:
+            trigger_maintenance_request(maintenance_type)
+            set_maintenance_status(maintenance_type,1)
+
     dbclient.close()
 
     # Check if there are any empty wells
@@ -94,7 +98,7 @@ def save_result(result_data):
     db = dbclient["LCM-OT-2-SLD"]  
     collection = db["MSE403_result"] 
     #collection = db["test_result"] 
-    result_data["timestamp"] = datetime.utcnow()  # UTC time
+    result_data["timestamp"] = datetime.now(pytz.timezone("America/Toronto"))  # UTC time
     insert_result = collection.insert_one(result_data)
     inserted_id = insert_result.inserted_id
     # close connection
@@ -144,13 +148,52 @@ def add_student_quota(student_id, quota):
     student_data = {"student_id": student_id, "quota": quota}
     collection.update_one({"student_id": student_id}, {"$set": student_data}, upsert=True)
     dbclient.close()
-    
+
+def check_maintenance_status(maintenance_type):
+    dbclient = MongoClient(connection_string)
+    db = dbclient["LCM-OT-2-SLD"]
+    collection = db["HITL_status"]
+    status_doc = collection.find_one({"name": maintenance_type})
+    dbclient.close()
+
+    if status_doc is None:
+        return 0
+    return status_doc.get("flag", 0)
+
+def set_maintenance_status(maintenance_type,new_status):
+    #flag == 1, request sent; flag == 0, no request undergoing
+    dbclient = MongoClient(connection_string)
+    db = dbclient["LCM-OT-2-SLD"]
+    collection = db["HITL_status"]
+
+    collection.update_one(
+        {"name": maintenance_type},
+        {"$set": {"flag": new_status,
+                   "timestamp": datetime.now(pytz.timezone("America/Toronto"))}
+         },
+        upsert=True
+    )
+    dbclient.close()
+
+def insert_maintenance_log(maintenance_type, operator="Unknown"):
+    dbclient = MongoClient(connection_string)
+    db = dbclient["LCM-OT-2-SLD"]
+    collection = db["maintenance_logs"]
+
+    log_data = {
+        "maintenance_type": maintenance_type,
+        "timestamp": datetime.now(pytz.timezone("America/Toronto")),
+        "operator": operator
+    }
+    collection.insert_one(log_data)
+    dbclient.close()
+
 
 if __name__ == "__main__":
-    generate_empty_well()
-    #find_unused_wells()
+    #generate_empty_well()
+    
+    #set_maintenance_status(maintenance_type="wellplate_maintenance",new_status=0)
+    find_unused_wells()
     #test_id = "test"
     #quota = 999
     #add_student_quota(test_id, quota)
-       
-    

app.py

@@ -308,7 +308,6 @@ def add_to_queue(student_id, R, Y, B):
             "Message": "Debug ID cannot submit to real experiment queue. Please use your student id to submit experiment."
         }
         return
-
     
     # Validate RYB inputs
     validation_result = validate_ryb_input(R, Y, B)

maintenance_flow.py

@@ -0,0 +1,33 @@
+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_utls import set_maintenance_status,generate_empty_well,insert_maintenance_log
+
+@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\nOpen the <{flow_run_url}|paused flow run>, OT2-LCM requests a {maintenance_type}, please complete with your user name and then click 'Resume'" 
+
+    slack_block.notify(message)
+    
+    user = pause_flow_run(wait_for_input=str, timeout=600)
+    
+    set_maintenance_status(maintenance_type,0)
+
+    if maintenance_type == "wellplate_maintenance":
+        generate_empty_well()
+        msg_out = f"Updating wells status on DB by {user}"
+        logger.info(msg_out)
+    
+    insert_maintenance_log(maintenance_type, user)
+
+    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: null
+
+# push section allows you to manage if and how this project is uploaded to remote locations
+push: null
+
+# pull section allows you to provide instructions for cloning this project in remote locations
+pull:
+- prefect.deployments.steps.set_working_directory:
+    directory: /Users/yh/Documents/AC work file/work SDL/OT-2-LCM-test
+
+# the deployments section allows you to provide configuration for deploying flows
+deployments:
+- name: wells-maintenance
+  version: null
+  tags: []
+  concurrency_limit: null
+  description: null
+  entrypoint: maintenance_flow.py:request_wells_maintenance
+  parameters: {}
+  work_pool:
+    name: ot2-pool
+    work_queue_name: null
+    job_variables: {}
+  enforce_parameter_schema: true
+  schedules: []

prefect_utils.py

@@ -12,14 +12,11 @@ def start_prefect_worker(work_pool_name: str = "ot2-pool"):
     worker_thread.start()
     print("Prefect Worker started in background thread.")
 
+from prefect.deployments import run_deployment
 
-
-from prefect.blocks.notifications import SlackWebhook
-from prefect.input import RunInput
-
-slack_block = SlackWebhook.load("prefect-test")
-class UserInput(RunInput):
-    github_username: str
-    comments: str = ""
-    flag_for_review: bool
-
+def trigger_maintenance_request(maintenance_type: str):
+    deployment_name = "request-wells-maintenance/wells-maintenance"
+    run_deployment(
+        name=deployment_name,
+        parameters={"maintenance_type": maintenance_type}
+    )