from pathlib import Path import numpy as np import gradio as gr import requests from typing import List # Define the COLUMN_MIN_MAX as provided COLUMN_MIN_MAX = { "Age": (18, 100), "Blood_Glucose_Level": (0, 300), "Blood_Pressure_Systolic": (80, 200), "Blood_Pressure_Diastolic": (40, 120), "BMI": (10, 50), "Condition_Severity": (1, 10), "Gender": (0, 2), "Ethnicity": (0, 5), "Geographic_Location": (0, 6), "Smoking_Status": (0, 2), "Diagnoses_ICD10": (0, 5), "Medications": (0, 7), "Allergies": (0, 5), "Previous_Treatments": (0, 5), "Alcohol_Consumption": (0, 3), "Exercise_Habits": (0, 4), "Diet": (0, 5), "Functional_Status": (0, 2), "Previous_Trial_Participation": (0, 1), } # Define possible categories for fields without predefined categories additional_categories = { "Gender": ["Male", "Female", "Other"], "Ethnicity": ["White", "Black or African American", "Asian", "American Indian or Alaska Native", "Native Hawaiian or Other Pacific Islander", "Other"], "Geographic_Location": ["North America", "South America", "Europe", "Asia", "Africa", "Australia", "Antarctica"], "Smoking_Status": ["Never", "Former", "Current"], "Diagnoses_ICD10": ["Actinic keratosis", "Melanoma", "Dermatofibroma", "Vascular lesion","None"], "Medications": ["Metformin", "Lisinopril", "Atorvastatin", "Amlodipine", "Omeprazole", "Simvastatin", "Levothyroxine", "None"], "Allergies": ["Penicillin", "Peanuts", "Shellfish", "Latex", "Bee stings", "None"], "Previous_Treatments": ["Chemotherapy", "Radiation Therapy", "Surgery", "Physical Therapy", "Immunotherapy", "None"], "Alcohol_Consumption": ["None", "Occasionally", "Regularly", "Heavy"], "Exercise_Habits": ["Sedentary", "Light", "Moderate", "Active", "Very Active"], "Diet": ["Omnivore", "Vegetarian", "Vegan", "Pescatarian", "Keto", "Mediterranean"], "Functional_Status": ["Independent", "Assisted", "Dependent"], "Previous_Trial_Participation": ["Yes", "No"] } # Define the input components for the researcher form with constraints min_age_input = gr.Number(label="Minimum Age", value=18, minimum=COLUMN_MIN_MAX["Age"][0], maximum=COLUMN_MIN_MAX["Age"][1]) max_age_input = gr.Number(label="Maximum Age", value=100, minimum=COLUMN_MIN_MAX["Age"][0], maximum=COLUMN_MIN_MAX["Age"][1]) gender_input = gr.CheckboxGroup(choices=additional_categories["Gender"], label="Gender", value=["Male"]) ethnicity_input = gr.CheckboxGroup(choices=additional_categories["Ethnicity"], label="Ethnicity", value=["White"]) geographic_location_input = gr.CheckboxGroup(choices=additional_categories["Geographic_Location"], label="Geographic Location", value=["North America"]) diagnoses_icd10_input = gr.CheckboxGroup(choices=additional_categories["Diagnoses_ICD10"], label="Skin Diagnosis", value=["Actinic keratosis"]) medications_input = gr.CheckboxGroup(choices=additional_categories["Medications"], label="Medications", value=["Metformin"]) allergies_input = gr.CheckboxGroup(choices=additional_categories["Allergies"], label="Allergies", value=["Peanuts"]) previous_treatments_input = gr.CheckboxGroup(choices=additional_categories["Previous_Treatments"], label="Previous Treatments", value=["None"]) min_blood_glucose_level_input = gr.Number(label="Minimum Blood Glucose Level", value=0, minimum=COLUMN_MIN_MAX["Blood_Glucose_Level"][0], maximum=COLUMN_MIN_MAX["Blood_Glucose_Level"][1]) max_blood_glucose_level_input = gr.Number(label="Maximum Blood Glucose Level", value=3, minimum=COLUMN_MIN_MAX["Blood_Glucose_Level"][0], maximum=COLUMN_MIN_MAX["Blood_Glucose_Level"][1]) blood_glucose_level_input = gr.Number(label="Blood Glucose Level", value=100, minimum=COLUMN_MIN_MAX["Blood_Glucose_Level"][0], maximum=COLUMN_MIN_MAX["Blood_Glucose_Level"][1]) min_blood_pressure_systolic_input = gr.Number(label="Minimum Blood Pressure (Systolic)", value=0, minimum=COLUMN_MIN_MAX["Blood_Pressure_Systolic"][0], maximum=COLUMN_MIN_MAX["Blood_Pressure_Systolic"][1]) max_blood_pressure_systolic_input = gr.Number(label="Maximum Blood Pressure (Systolic)", value=3, minimum=COLUMN_MIN_MAX["Blood_Pressure_Systolic"][0], maximum=COLUMN_MIN_MAX["Blood_Pressure_Systolic"][1]) blood_pressure_systolic_input = gr.Number(label="Blood Pressure (Systolic)", value=120, minimum=COLUMN_MIN_MAX["Blood_Pressure_Systolic"][0], maximum=COLUMN_MIN_MAX["Blood_Pressure_Systolic"][1]) min_blood_pressure_diastolic_input = gr.Number(label="Minimum Blood Pressure (Diastolic)", value=0, minimum=COLUMN_MIN_MAX["Blood_Pressure_Diastolic"][0], maximum=COLUMN_MIN_MAX["Blood_Pressure_Diastolic"][1]) max_blood_pressure_diastolic_input = gr.Number(label="Maximum Blood Pressure (Diastolic)", value=3, minimum=COLUMN_MIN_MAX["Blood_Pressure_Diastolic"][0], maximum=COLUMN_MIN_MAX["Blood_Pressure_Diastolic"][1]) blood_pressure_diastolic_input = gr.Number(label="Blood Pressure (Diastolic)", value=80, minimum=COLUMN_MIN_MAX["Blood_Pressure_Diastolic"][0], maximum=COLUMN_MIN_MAX["Blood_Pressure_Diastolic"][1]) min_bmi_input = gr.Number(label="Minimum BMI", value=0, minimum=COLUMN_MIN_MAX["BMI"][0], maximum=COLUMN_MIN_MAX["BMI"][1]) max_bmi_input = gr.Number(label="Maximum BMI", value=3, minimum=COLUMN_MIN_MAX["BMI"][0], maximum=COLUMN_MIN_MAX["BMI"][1]) bmi_input = gr.Number(label="BMI", value=20, minimum=COLUMN_MIN_MAX["BMI"][0], maximum=COLUMN_MIN_MAX["BMI"][1]) min_condition_severity_input = gr.Number(label="Condition Severity", value=5, minimum=COLUMN_MIN_MAX["Condition_Severity"][0], maximum=COLUMN_MIN_MAX["Condition_Severity"][1]) max_condition_severity_input = gr.Number(label="Condition Severity", value=5, minimum=COLUMN_MIN_MAX["Condition_Severity"][0], maximum=COLUMN_MIN_MAX["Condition_Severity"][1]) condition_severity_input = gr.Number(label="Condition Severity", value=5, minimum=0, maximum=10) smoking_status_input = gr.CheckboxGroup(choices=additional_categories["Smoking_Status"], label="Smoking Status", value=["Never"]) alcohol_consumption_input = gr.CheckboxGroup(choices=additional_categories["Alcohol_Consumption"], label="Alcohol Consumption", value=["None"]) exercise_habits_input = gr.CheckboxGroup(choices=additional_categories["Exercise_Habits"], label="Exercise Habits", value=["Sedentary"]) diet_input = gr.CheckboxGroup(choices=additional_categories["Diet"], label="Diet", value=["Omnivore"]) functional_status_input = gr.CheckboxGroup(choices=additional_categories["Functional_Status"], label="Functional Status", value=["Independent"]) previous_trial_participation_input = gr.CheckboxGroup(choices=additional_categories["Previous_Trial_Participation"], label="Previous Trial Participation", value=["No"]) # Define the server's URL SERVER_URL = "https://affordable-prot-bind-clarke.trycloudflare.com/requirements/create" # Ensure this is the correct endpoint def encode_categorical_data(data: List[str], category_name: str) -> List[int]: """Encodes a list of categorical values into their corresponding indices based on additional_categories.""" sub_cats = additional_categories.get(category_name, []) encoded_data = [] for value in data: if value in sub_cats: encoded_index = sub_cats.index(value) # Validate that the encoded index is within the specified range min_val, max_val = COLUMN_MIN_MAX.get(category_name, (0, len(sub_cats)-1)) if min_val <= encoded_index <= max_val: encoded_data.append(encoded_index) else: print(f"Encoded value for {category_name}='{value}' is out of range. Setting to 0.") encoded_data.append(0) else: print(f"Value '{value}' not recognized in category '{category_name}'. Setting to 0.") encoded_data.append(0) return encoded_data def process_researcher_data( min_age, max_age, gender, ethnicity, geographic_location, diagnoses_icd10, medications, allergies, previous_treatments, min_blood_glucose_level, max_blood_glucose_level, min_blood_pressure_systolic, max_blood_pressure_systolic, min_blood_pressure_diastolic, max_blood_pressure_diastolic, min_bmi, max_bmi, smoking_status, alcohol_consumption, exercise_habits, diet, min_condition_severity, max_condition_severity, functional_status, previous_trial_participation ): # Encode categorical data encoded_gender = encode_categorical_data(gender, "Gender") encoded_ethnicity = encode_categorical_data(ethnicity, "Ethnicity") encoded_geographic_location = encode_categorical_data(geographic_location, "Geographic_Location") encoded_diagnoses_icd10 = encode_categorical_data(diagnoses_icd10, "Diagnoses_ICD10") encoded_medications = encode_categorical_data(medications, "Medications") encoded_allergies = encode_categorical_data(allergies, "Allergies") encoded_previous_treatments = encode_categorical_data(previous_treatments, "Previous_Treatments") encoded_smoking_status = encode_categorical_data(smoking_status, "Smoking_Status") encoded_alcohol_consumption = encode_categorical_data(alcohol_consumption, "Alcohol_Consumption") encoded_exercise_habits = encode_categorical_data(exercise_habits, "Exercise_Habits") encoded_diet = encode_categorical_data(diet, "Diet") encoded_functional_status = encode_categorical_data(functional_status, "Functional_Status") encoded_previous_trial_participation = encode_categorical_data(previous_trial_participation, "Previous_Trial_Participation") # Create a list of requirements requirements = [] # Add numerical requirements numerical_fields = [ ("Age", min_age, "greater_than"), ("Age", max_age, "less_than"), ("Blood_Glucose_Level", min_blood_glucose_level, "greater_than"), ("Blood_Glucose_Level", max_blood_glucose_level, "less_than"), ("Blood_Pressure_Systolic", min_blood_pressure_systolic, "greater_than"), ("Blood_Pressure_Systolic", max_blood_pressure_systolic, "less_than"), ("Blood_Pressure_Diastolic", min_blood_pressure_diastolic, "greater_than"), ("Blood_Pressure_Diastolic", max_blood_pressure_diastolic, "less_than"), ("BMI", min_bmi, "greater_than"), ("BMI", max_bmi, "less_than"), ("Condition_Severity", min_condition_severity, "greater_than"), ("Condition_Severity", max_condition_severity, "less_than"), ] for field, value, comparison in numerical_fields: if value is not None: # Ensure the value is within the specified range min_val, max_val = COLUMN_MIN_MAX.get(field, (None, None)) if min_val is not None and max_val is not None: if not (min_val <= value <= max_val): print(f"Value for {field}={value} is out of range ({min_val}, {max_val}). Adjusting to fit within range.") value = max(min(value, max_val), min_val) requirements.append({ "column_name": field, "value": value, "comparison_type": comparison }) # Add categorical requirements categorical_fields = [ ("Gender", encoded_gender, "equal"), ("Ethnicity", encoded_ethnicity, "equal"), ("Geographic_Location", encoded_geographic_location, "equal"), ("Diagnoses_ICD10", encoded_diagnoses_icd10, "equal"), ("Medications", encoded_medications, "equal"), ("Allergies", encoded_allergies, "equal"), ("Previous_Treatments", encoded_previous_treatments, "equal"), ("Smoking_Status", encoded_smoking_status, "equal"), ("Alcohol_Consumption", encoded_alcohol_consumption, "equal"), ("Exercise_Habits", encoded_exercise_habits, "equal"), ("Diet", encoded_diet, "equal"), ("Functional_Status", encoded_functional_status, "equal"), ("Previous_Trial_Participation", encoded_previous_trial_participation, "equal"), ] for field, encoded_values, comparison in categorical_fields: min_val, max_val = COLUMN_MIN_MAX.get(field, (0, len(additional_categories[field])-1)) for encoded in encoded_values: if min_val <= encoded <= max_val: requirements.append({ "column_name": field, "value": encoded, "comparison_type": comparison }) else: print(f"Encoded value {encoded} for {field} is out of range ({min_val}, {max_val}). Skipping.") # Encode and add non-categorical fields like medications, allergies, previous treatments # Already handled above in categorical_fields # Construct the payload as a regular dictionary payload = { "model_name": "second_model", "requirements": requirements } print("Payload to send:", payload) # For debugging # Make the request to the server try: res = requests.post(SERVER_URL, json=payload) res.raise_for_status() # Raise an error for bad status codes except requests.exceptions.HTTPError as http_err: print(f"HTTP error occurred: {http_err}") # For debugging return f"HTTP error occurred: {http_err}" except Exception as err: print(f"Other error occurred: {err}") # For debugging return f"Other error occurred: {err}" # Get the response from the server try: response = res.json() print("Server response:", response) except ValueError: print("Response is not in JSON format.") return "Response is not in JSON format." return response.get("message", "No message received from server") # Create the Gradio interface for researchers researcher_demo = gr.Interface( fn=process_researcher_data, inputs=[ min_age_input, max_age_input, gender_input, ethnicity_input, geographic_location_input, diagnoses_icd10_input, medications_input, allergies_input, previous_treatments_input, min_blood_glucose_level_input, max_blood_glucose_level_input, min_blood_pressure_systolic_input, max_blood_pressure_systolic_input, min_blood_pressure_diastolic_input, max_blood_pressure_diastolic_input, min_bmi_input, max_bmi_input, smoking_status_input, alcohol_consumption_input, exercise_habits_input, diet_input, min_condition_severity_input, max_condition_severity_input, functional_status_input, previous_trial_participation_input ], outputs="text", title="Clinical Researcher Criteria Form", description="Please enter the criteria for the type of patients you are looking for." ) # Launch the researcher interface with a public link if __name__ == "__main__": researcher_demo.launch(share=False)