prefetch batch > 10000 size fix

Author: SamoraHunter

pat2vec/patvec_get_batch_methods/get_merged_batches.py

@@ -368,10 +368,6 @@ def get_merged_pat_batch_drugs(
         return pd.DataFrame()  # Return an empty DataFrame in case of error
 
 
-import os
-import pandas as pd
-
-
 def get_merged_pat_batch_diagnostics(
     client_idcode_list,
     config_obj=None,
@@ -571,11 +567,6 @@ def get_merged_pat_batch_mct_docs(
         # Apply data type filters for MCT documents
         batch_target = apply_data_type_mct_docs_filters(config_obj, batch_target)
 
-        # larger batches returned as lists...
-        if isinstance(batch_target, list) and len(batch_target) >= 1:
-
-            batch_target = batch_target[0]
-
         # Drop rows with NaN values in critical columns
         col_list_drop_nan = [
             "observation_valuetext_analysed",
@@ -717,10 +708,6 @@ def get_merged_pat_batch_epr_docs(
                     text_column="body_analysed",
                     debug=config_obj.verbosity > 5,
                 )
-        # larger batches returned as lists...
-        if isinstance(batch_target, list) and len(batch_target) >= 1:
-
-            batch_target = batch_target[0]
 
         # Drop rows with NaN values in critical columns
         col_list_drop_nan = ["body_analysed", "updatetime", "client_idcode"]
@@ -850,11 +837,6 @@ def get_merged_pat_batch_textual_obs_docs(
             search_string=f"{bloods_time_field}:[{global_start_year}-{global_start_month}-{global_start_day} TO {global_end_year}-{global_end_month}-{global_end_day}]",
         )
 
-        # larger batches returned as lists...
-        if isinstance(batch_target, list) and len(batch_target) >= 1:
-
-            batch_target = batch_target[0]
-
         # Drop rows with no textualObs
         batch_target = batch_target.dropna(subset=["textualObs"])
         # Drop rows with empty string in textualObs
@@ -1429,9 +1411,6 @@ def get_merged_pat_batch_reports(
             f"updatetime:[{global_start_year}-{global_start_month}-{global_start_day} TO {global_end_year}-{global_end_month}-{global_end_day}]",
         )
 
-        if isinstance(batch_target, list) and len(batch_target) >= 1:
-
-            batch_target = batch_target[0]
         # Combine textualObs and basicobs_value_analysed into body_analysed
         batch_target["body_analysed"] = (
             batch_target["textualObs"].astype(str)

pat2vec/util/cogstack_v8_lite.py

@@ -96,7 +96,7 @@ def create_credentials_file():
         hosts = ["https://your-actual-elasticsearch-host:9200"]
 
 
-print(f"Imported cogstack_v8_lite from pat2vec.util")
+print(f"Imported cogstack_v8_lite from pat2vec.util .")
 print(f"Username: %s" % username)
 
 
@@ -258,10 +258,17 @@ def list_chunker(entered_list):
     return chunks
 
 
+# Use a generator to yield DataFrames one by one
+def dataframe_generator(list_of_dfs):
+    for df in list_of_dfs:
+        yield df
+
+
 def cohort_searcher_with_terms_and_search(
     index_name, fields_list, term_name, entered_list, search_string
 ):
     if len(entered_list) >= 10000:
+        print("cohort_searcher_with_terms_and_search list chunking")
         results = []
         chunked_list = list_chunker(entered_list)
         for mini_list in chunked_list:
@@ -284,9 +291,18 @@ def cohort_searcher_with_terms_and_search(
             merged_df = [df.set_index("_id") for df in results]
 
         except Exception as e:
+            print(e)
+            raise e
             print(e)
             return results
 
+        try:
+            # Concatenate DataFrames using the generator
+            merged_df = pd.concat(dataframe_generator(results), ignore_index=True)
+            merged_df = merged_df.set_index("_id")
+        except Exception as e:
+            raise e
+
         return merged_df
     else:
         query = {