from concrete.ml.deployment import FHEModelClient from pathlib import Path import numpy as np import gradio as gr import requests import json from typing import List # 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": ["E11.9", "I10", "J45.909", "M54.5", "F32.9", "K21.9"], "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 min_age_input = gr.Number(label="Minimum Age", value=18) max_age_input = gr.Number(label="Maximum Age", value=100) gender_input = gr.CheckboxGroup(choices=additional_categories["Gender"], label="Gender") ethnicity_input = gr.CheckboxGroup(choices=additional_categories["Ethnicity"], label="Ethnicity") geographic_location_input = gr.CheckboxGroup(choices=additional_categories["Geographic_Location"], label="Geographic Location") diagnoses_icd10_input = gr.CheckboxGroup(choices=additional_categories["Diagnoses_ICD10"], label="Diagnoses (ICD-10)") medications_input = gr.CheckboxGroup(choices=additional_categories["Medications"], label="Medications") allergies_input = gr.CheckboxGroup(choices=additional_categories["Allergies"], label="Allergies") previous_treatments_input = gr.CheckboxGroup(choices=additional_categories["Previous_Treatments"], label="Previous Treatments") min_blood_glucose_level_input = gr.Number(label="Minimum Blood Glucose Level", value=0) max_blood_glucose_level_input = gr.Number(label="Maximum Blood Glucose Level", value=300) min_blood_pressure_systolic_input = gr.Number(label="Minimum Blood Pressure (Systolic)", value=80) max_blood_pressure_systolic_input = gr.Number(label="Maximum Blood Pressure (Systolic)", value=200) min_blood_pressure_diastolic_input = gr.Number(label="Minimum Blood Pressure (Diastolic)", value=40) max_blood_pressure_diastolic_input = gr.Number(label="Maximum Blood Pressure (Diastolic)", value=120) min_bmi_input = gr.Number(label="Minimum BMI", value=10) max_bmi_input = gr.Number(label="Maximum BMI", value=50) smoking_status_input = gr.CheckboxGroup(choices=additional_categories["Smoking_Status"], label="Smoking Status") alcohol_consumption_input = gr.CheckboxGroup(choices=additional_categories["Alcohol_Consumption"], label="Alcohol Consumption") exercise_habits_input = gr.CheckboxGroup(choices=additional_categories["Exercise_Habits"], label="Exercise Habits") diet_input = gr.CheckboxGroup(choices=additional_categories["Diet"], label="Diet") min_condition_severity_input = gr.Number(label="Minimum Condition Severity", value=1) max_condition_severity_input = gr.Number(label="Maximum Condition Severity", value=10) functional_status_input = gr.CheckboxGroup(choices=additional_categories["Functional_Status"], label="Functional Status") previous_trial_participation_input = gr.CheckboxGroup(choices=additional_categories["Previous_Trial_Participation"], label="Previous Trial Participation") 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_data.append(sub_cats.index(value) + 1) # Adding 1 to avoid index 0 for valid entries else: encoded_data.append(0) # Encode unmatched as 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_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 if min_age is not None: requirements.append({ "column_name": "Age", "value": int(min_age), "comparison_type": "greater_than" }) if max_age is not None: requirements.append({ "column_name": "Age", "value": int(max_age), "comparison_type": "less_than" }) if min_blood_glucose_level is not None: requirements.append({ "column_name": "Blood_Glucose_Level", "value": int(min_blood_glucose_level), "comparison_type": "greater_than" }) if max_blood_glucose_level is not None: requirements.append({ "column_name": "Blood_Glucose_Level", "value": int(max_blood_glucose_level), "comparison_type": "less_than" }) if min_blood_pressure_systolic is not None: requirements.append({ "column_name": "Blood_Pressure_Systolic", "value": int(min_blood_pressure_systolic), "comparison_type": "greater_than" }) if max_blood_pressure_systolic is not None: requirements.append({ "column_name": "Blood_Pressure_Systolic", "value": int(max_blood_pressure_systolic), "comparison_type": "less_than" }) if min_blood_pressure_diastolic is not None: requirements.append({ "column_name": "Blood_Pressure_Diastolic", "value": int(min_blood_pressure_diastolic), "comparison_type": "greater_than" }) if max_blood_pressure_diastolic is not None: requirements.append({ "column_name": "Blood_Pressure_Diastolic", "value": int(max_blood_pressure_diastolic), "comparison_type": "less_than" }) if min_bmi is not None: requirements.append({ "column_name": "BMI", "value": float(min_bmi), "comparison_type": "greater_than" }) if max_bmi is not None: requirements.append({ "column_name": "BMI", "value": float(max_bmi), "comparison_type": "less_than" }) if min_condition_severity is not None: requirements.append({ "column_name": "Condition_Severity", "value": int(min_condition_severity), "comparison_type": "greater_than" }) if max_condition_severity is not None: requirements.append({ "column_name": "Condition_Severity", "value": int(max_condition_severity), "comparison_type": "less_than" }) # Add categorical requirements for gender_value in encoded_gender: if gender_value > 0: requirements.append({ "column_name": "Gender", "value": gender_value, "comparison_type": "equal" }) for ethnicity_value in encoded_ethnicity: if ethnicity_value > 0: requirements.append({ "column_name": "Ethnicity", "value": ethnicity_value, "comparison_type": "equal" }) for location_value in encoded_geographic_location: if location_value > 0: requirements.append({ "column_name": "Geographic_Location", "value": location_value, "comparison_type": "equal" }) for diagnosis_value in encoded_diagnoses_icd10: if diagnosis_value > 0: requirements.append({ "column_name": "Diagnoses_ICD10", "value": diagnosis_value, "comparison_type": "equal" }) for smoking_status_value in encoded_smoking_status: if smoking_status_value > 0: requirements.append({ "column_name": "Smoking_Status", "value": smoking_status_value, "comparison_type": "equal" }) for alcohol_value in encoded_alcohol_consumption: if alcohol_value > 0: requirements.append({ "column_name": "Alcohol_Consumption", "value": alcohol_value, "comparison_type": "equal" }) for exercise_value in encoded_exercise_habits: if exercise_value > 0: requirements.append({ "column_name": "Exercise_Habits", "value": exercise_value, "comparison_type": "equal" }) for diet_value in encoded_diet: if diet_value > 0: requirements.append({ "column_name": "Diet", "value": diet_value, "comparison_type": "equal" }) for status in encoded_functional_status: if status > 0: requirements.append({ "column_name": "Functional_Status", "value": status, "comparison_type": "equal" }) for participation in encoded_previous_trial_participation: if participation > 0: requirements.append({ "column_name": "Previous_Trial_Participation", "value": participation, "comparison_type": "equal" }) # Encode and add non-categorical fields like medications, allergies, previous treatments for medication in medications: encoded_medications = encode_categorical_data([medication], "Medications") for med_value in encoded_medications: if med_value > 0: requirements.append({ "column_name": "Medications", "value": med_value, "comparison_type": "equal" }) for allergy in allergies: encoded_allergies = encode_categorical_data([allergy], "Allergies") for allergy_value in encoded_allergies: if allergy_value > 0: requirements.append({ "column_name": "Allergies", "value": allergy_value, "comparison_type": "equal" }) for treatment in previous_treatments: encoded_treatments = encode_categorical_data([treatment], "Previous_Treatments") for treatment_value in encoded_treatments: if treatment_value > 0: requirements.append({ "column_name": "Previous_Treatments", "value": treatment_value, "comparison_type": "equal" }) # Construct the payload as a regular dictionary payload = { "model_name": "fhe_model_v1", "requirements": requirements } # turn the payload into a JSON object payload = json.dumps(payload) print("Payload:", payload) # Store the server's URL SERVER_URL = "https://ppaihack-match.azurewebsites.net/requirements/create" # 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=True)