Alterting compartment 2 to obtain kits from the DB instead of a CSV file in compartment 1

Author: adrianoaru-nhs

requirements.txt

@@ -6,4 +6,4 @@ oracledb>=2.5.1
 pandas>=2.2.3
 python-dotenv~=1.0.1
 Flask~=3.0.3
-fastparquet>=2024.11.0
+sqlalchemy>=2.0.38

tests/Smokescreen/test_compartment_2.py

@@ -4,7 +4,7 @@
 from utils.batch_processing import batch_processing
 from datetime import datetime
 from utils.screening_subject_page_searcher import verify_subject_event_status_by_nhs_no
-import pandas as pd
+from utils.fit_kit_generation import create_fit_id_df
 
 @pytest.mark.wip1
 def test_compartment_2(page: Page) -> None:
@@ -13,23 +13,22 @@ def test_compartment_2(page: Page) -> None:
 
     MainMenu(page).go_to_fit_test_kits_page()
     FITTestKits(page).go_to_log_devices_page()
-    subjectdf = pd.read_parquet('subject_kit_number.parquet', engine='fastparquet')
-    os.remove('subject_kit_number.parquet')
+    subjectdf = create_fit_id_df()
 
     for subject in range(4):
-        fit_device_id = subjectdf["FIT_Device_ID"].iloc[subject-1]
+        fit_device_id = subjectdf["fit_device_id"].iloc[subject]
         LogDevices(page).fill_fit_device_id_field(fit_device_id)
         sample_date = datetime.now().strftime("%#d %b %Y")
         LogDevices(page).fill_sample_date_field(sample_date)
         LogDevices(page).verify_successfully_logged_device_text()
 
-    nhs_no = subjectdf["NHS_Number"].iloc[0]
+    nhs_no = subjectdf["subject_nhs_number"].iloc[0]
     verify_subject_event_status_by_nhs_no(page, nhs_no, "S43 - Kit Returned and Logged (Initial Test)")
 
     NavigationBar(page).click_main_menu_link()
     MainMenu(page).go_to_fit_test_kits_page()
     FITTestKits(page).go_to_log_devices_page()
-    spoilt_fit_device_id = subjectdf["FIT_Device_ID"].iloc[4]
+    spoilt_fit_device_id = subjectdf["fit_device_id"].iloc[-1]
     LogDevices(page).fill_fit_device_id_field(spoilt_fit_device_id)
     LogDevices(page).click_device_spoilt_button()
     LogDevices(page).select_spoilt_device_dropdown_option()

utils/batch_processing.py

@@ -5,7 +5,6 @@
 from pages.manage_active_batch_page import ManageActiveBatch
 from pages.archived_batch_list_page import ArchivedBatchList
 from utils.pdf_reader import extract_nhs_no_from_pdf
-from utils.csv_reader import convert_csv_to_df
 from utils.screening_subject_page_searcher import verify_subject_event_status_by_nhs_no
 import os
 import pytest
@@ -59,9 +58,6 @@ def batch_processing(page: Page, batch_type: str, batch_description: str, latest
             first_nhs_no = nhs_numbers[0]
             os.remove(file) # Deletes the file after extracting the necessary data
         elif file.endswith(".csv"):
-            if batch_type == "S9":
-                csv_df = convert_csv_to_df(file) # Currently no use in compartment 1, will be necessary for future compartments
-                csv_df.to_parquet('subject_kit_number.parquet', engine='fastparquet')
             os.remove(file) # Deletes the file after extracting the necessary data
 
     # This loops through each Confirm printed button and clicks each one

utils/csv_reader.py

@@ -1,16 +1,17 @@
 import pandas as pd
 from datetime import datetime
 
-def convert_csv_to_df(file: str):
-    csv_df = pd.read_csv(file)
-    csv_df.drop(csv_df.columns[[0,2,3,4,5,6,7,8,9,10,12]], axis = 1, inplace = True) # Removing unnecessary columns
-    df = csv_df.rename(columns={csv_df.columns[0]: "NHS_Number",csv_df.columns[1]: "Kit_ID"}) # Renaming the columns to something more meaningful
-    df.dropna(inplace = True) # Deleting any Null records
-    df["NHS_Number"] = df["NHS_Number"].str.replace(" ", "") # Removing the space from the nhs number
-    df["FIT_Device_ID"] = df["Kit_ID"].apply(convert_kit_id_to_fit_device_id)
-    return df
+# Commented out as no longer required, need to see if it will be used at all
+# def convert_csv_to_df(file: str):
+#     csv_df = pd.read_csv(file)
+#     csv_df.drop(csv_df.columns[[0,2,3,4,5,6,7,8,9,10,12]], axis = 1, inplace = True) # Removing unnecessary columns
+#     df = csv_df.rename(columns={csv_df.columns[0]: "NHS_Number",csv_df.columns[1]: "Kit_ID"}) # Renaming the columns to something more meaningful
+#     df.dropna(inplace = True) # Deleting any Null records
+#     df["NHS_Number"] = df["NHS_Number"].str.replace(" ", "") # Removing the space from the nhs number
+#     df["FIT_Device_ID"] = df["Kit_ID"].apply(convert_kit_id_to_fit_device_id)
+#     return df
 
-def convert_kit_id_to_fit_device_id(kit_id):
-        today = datetime.now()
-        year = today.strftime("%y") # Get the year from todays date in YY format
-        return f"{kit_id}12{int(year)+1}12345/KD00001" # Creating the fit device id with an expiry date set to December next year
+# def convert_kit_id_to_fit_device_id(kit_id):
+#         today = datetime.now()
+#         year = today.strftime("%y") # Get the year from todays date in YY format
+#         return f"{kit_id}12{int(year)+1}12345/KD00001" # Creating the fit device id with an expiry date set to December next year

utils/fit_kit_generation.py

@@ -0,0 +1,50 @@
+from oracle import OracleDB
+import pandas as pd
+from datetime import datetime
+
+def create_fit_id_df():
+    df = get_kit_id_from_db()
+    df["fit_device_id"] = df["kitid"].apply(calculate_check_digit)
+    df["fit_device_id"] = df["fit_device_id"].apply(convert_kit_id_to_fit_device_id)
+
+    temp_string = df["screening_subject_id"].apply(str).iloc[0]
+    for subject in range(df.shape[0]):
+        subject_id = df["screening_subject_id"].apply(str).iloc[subject]
+        temp_string += f" or SCREENING_SUBJECT_ID = {subject_id}"
+
+    df = get_nhs_number_from_subject_id(temp_string, df)
+    return df
+
+def get_kit_id_from_db():
+    kit_id_df = OracleDB().execute_query("""select tk.kitid, tk.screening_subject_id
+    from tk_items_t tk
+    inner join ep_subject_episode_t se on se.screening_subject_id = tk.screening_subject_id
+    inner join screening_subject_t sst on (sst.screening_subject_id = tk.screening_subject_id)
+    inner join sd_contact_t sdc on (sdc.nhs_number = sst.subject_nhs_number)
+    where tk.tk_type_id = 3
+    and tk.logged_in_flag = 'N'
+    and sdc.hub_id = 23159
+    and device_id is null
+    and tk.invalidated_date is null
+    and se.latest_event_status_id in (11198, 11213)
+    fetch first 5 rows only""")
+
+    return kit_id_df
+
+def calculate_check_digit(kitID: str):
+    total = 0
+    charString = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ-. $/+%"
+    for i in range(len(kitID)):
+        total += charString.index(kitID[i-1])
+    check_digit = charString[total % 43]
+    return f"{kitID}-{check_digit}"
+
+def convert_kit_id_to_fit_device_id(kit_id: str):
+    today = datetime.now()
+    year = today.strftime("%y") # Get the year from todays date in YY format
+    return f"{kit_id}12{int(year)+1}12345/KD00001"
+
+def get_nhs_number_from_subject_id(subject_ids, df):
+    temp_df = OracleDB().execute_query(f"SELECT SCREENING_SUBJECT_ID, SUBJECT_NHS_NUMBER FROM SCREENING_SUBJECT_T WHERE SCREENING_SUBJECT_ID = {subject_ids}")
+    df = df.merge(temp_df[["screening_subject_id","subject_nhs_number"]], on="screening_subject_id", how="left")
+    return df

utils/oracle.py

@@ -1,7 +1,8 @@
 import oracledb
 import os
 from dotenv import load_dotenv
-
+from sqlalchemy import create_engine
+import pandas as pd
 class OracleDB:
     def __init__(self):
         load_dotenv()
@@ -74,7 +75,26 @@ def delete_all_users_from_approved_users_table(self):  # To remove all users fro
         except Exception as connectionError:
             print(f"Failed to connect to the DB! with connection error {connectionError}")
 
-    # The following two functions are commented out as they are not used currently, but may be needed in future compartments
+    def execute_query(self, query: str,): # To use when "select xxxx" (stored procedures)
+        try:
+            print("Attempting DB connection...")
+            conn = oracledb.connect(user=self.user, password=self.password, dsn=self.dns)
+            engine = create_engine('oracle+oracledb://', creator=lambda: conn)
+            print(conn.version, "DB connection successful!")
+            try:
+                print(f"Attempting to execute query")
+                df = pd.read_sql(query, engine)
+                print(conn.version, "query execution successful!")
+            except Exception as executionError:
+                print(f"Failed to execute query with execution error {executionError}")
+            finally:
+                if conn is not None:
+                    conn.close()
+                    return df
+        except Exception as connectionError:
+                print(f"Failed to connect to the DB! with connection error {connectionError}")
+
+    # The following function is commented out as it is not used currently, but may be needed in future compartments
 
     # def execute_stored_procedure(self, procedure: str): # To use when "exec xxxx" (stored procedures)
     #     try:
@@ -93,23 +113,3 @@ def delete_all_users_from_approved_users_table(self):  # To remove all users fro
     #                 conn.close()
     #     except Exception as connectionError:
     #             print(f"Failed to connect to the DB! with connection error: {connectionError}")
-
-    # def execute_query(self, query: str,): # To use when "select xxxx" (stored procedures)
-    #     try:
-    #         print("Attempting DB connection...")
-    #         conn = oracledb.connect(user=self.user, password=self.password, dsn=self.dns)
-    #         print(conn.version, "DB connection successful!")
-    #         try:
-    #             print(f"Attempting to execute query: {query}")
-    #             cursor = conn.cursor()
-    #             cursor.execute(query)
-    #             result = cursor.fetchall()
-    #             print(conn.version, "query execution successful!")
-    #         except Exception as executionError:
-    #             print(f"Failed to execute query with execution error {executionError}")
-    #         finally:
-    #             if conn is not None:
-    #                 conn.close()
-    #                 return result
-    #     except Exception as connectionError:
-    #             print(f"Failed to connect to the DB! with connection error {connectionError}")