Spaces:
Running
Running
| import requests | |
| import gradio as gr | |
| import logging | |
| import json | |
| import tensorflow as tf | |
| import tensorflow_hub as hub | |
| import numpy as np | |
| from PIL import Image | |
| import io | |
| # Set up logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # API key and user ID for on-demand | |
| api_key = 'KGSjxB1uptfSk8I8A7ciCuNT9Xa3qWC3' | |
| external_user_id = 'plugin-1717464304' | |
| # Load the keras model | |
| def load_model(): | |
| try: | |
| # Define custom objects dictionary | |
| custom_objects = { | |
| 'KerasLayer': hub.KerasLayer, | |
| # Add any other custom layers your model might use | |
| } | |
| # Load model with custom object scope | |
| with tf.keras.utils.custom_object_scope(custom_objects): | |
| model = tf.keras.models.load_model('model_epoch_01.h5.keras') | |
| logger.info("Model loaded successfully") | |
| return model | |
| except Exception as e: | |
| logger.error(f"Error loading model: {str(e)}") | |
| raise | |
| # Preprocess image for model | |
| def preprocess_image(image): | |
| try: | |
| # Convert to numpy array if needed | |
| if isinstance(image, Image.Image): | |
| image = np.array(image) | |
| # Ensure image has 3 channels (RGB) | |
| if len(image.shape) == 2: # Grayscale image | |
| image = np.stack((image,) * 3, axis=-1) | |
| elif len(image.shape) == 3 and image.shape[2] == 4: # RGBA image | |
| image = image[:, :, :3] | |
| # Resize image to match model's expected input shape | |
| target_size = (224, 224) # Change this to match your model's input size | |
| image = tf.image.resize(image, target_size) | |
| # Normalize pixel values | |
| image = image / 255.0 | |
| # Add batch dimension | |
| image = np.expand_dims(image, axis=0) | |
| return image | |
| except Exception as e: | |
| logger.error(f"Error preprocessing image: {str(e)}") | |
| raise | |
| def create_chat_session(): | |
| try: | |
| create_session_url = 'https://api.on-demand.io/chat/v1/sessions' | |
| create_session_headers = { | |
| 'apikey': api_key, | |
| 'Content-Type': 'application/json' | |
| } | |
| create_session_body = { | |
| "pluginIds": [], | |
| "externalUserId": external_user_id | |
| } | |
| response = requests.post(create_session_url, headers=create_session_headers, json=create_session_body) | |
| response.raise_for_status() | |
| return response.json()['data']['id'] | |
| except requests.exceptions.RequestException as e: | |
| logger.error(f"Error creating chat session: {str(e)}") | |
| raise | |
| def submit_query(session_id, query, image_analysis=None): | |
| try: | |
| submit_query_url = f'https://api.on-demand.io/chat/v1/sessions/{session_id}/query' | |
| submit_query_headers = { | |
| 'apikey': api_key, | |
| 'Content-Type': 'application/json' | |
| } | |
| # Include image analysis in the query if available | |
| query_with_image = query | |
| if image_analysis: | |
| query_with_image += f"\n\nImage Analysis Results: {image_analysis}" | |
| structured_query = f""" | |
| Based on the following patient information and image analysis, provide a detailed medical analysis in JSON format: | |
| {query_with_image} | |
| Return only valid JSON with these fields: | |
| - diagnosis_details | |
| - probable_diagnoses (array) | |
| - treatment_plans (array) | |
| - lifestyle_modifications (array) | |
| - medications (array of objects with name and dosage) | |
| - additional_tests (array) | |
| - precautions (array) | |
| - follow_up (string) | |
| - image_findings (object with prediction and confidence) | |
| """ | |
| submit_query_body = { | |
| "endpointId": "predefined-openai-gpt4o", | |
| "query": structured_query, | |
| "pluginIds": ["plugin-1712327325", "plugin-1713962163"], | |
| "responseMode": "sync" | |
| } | |
| response = requests.post(submit_query_url, headers=submit_query_headers, json=submit_query_body) | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.exceptions.RequestException as e: | |
| logger.error(f"Error submitting query: {str(e)}") | |
| raise | |
| def extract_json_from_answer(answer): | |
| """Extract and clean JSON from the LLM response""" | |
| try: | |
| return json.loads(answer) | |
| except json.JSONDecodeError: | |
| try: | |
| # Find the first occurrence of '{' and last occurrence of '}' | |
| start_idx = answer.find('{') | |
| end_idx = answer.rfind('}') + 1 | |
| if start_idx != -1 and end_idx != 0: | |
| json_str = answer[start_idx:end_idx] | |
| return json.loads(json_str) | |
| except (json.JSONDecodeError, ValueError): | |
| logger.error("Failed to parse JSON from response") | |
| raise | |
| def format_prediction(prediction): | |
| """Format model prediction into a standardized structure""" | |
| try: | |
| # Adjust this based on your model's output format | |
| confidence = float(prediction[0][0]) | |
| return { | |
| "prediction": "abnormal" if confidence > 0.5 else "normal", | |
| "confidence": round(confidence * 100, 2) | |
| } | |
| except Exception as e: | |
| logger.error(f"Error formatting prediction: {str(e)}") | |
| raise | |
| # Initialize the model | |
| try: | |
| model = load_model() | |
| except Exception as e: | |
| logger.error(f"Failed to initialize model: {str(e)}") | |
| model = None | |
| def gradio_interface(patient_info, image): | |
| try: | |
| if model is None: | |
| raise ValueError("Model not properly initialized") | |
| # Process image if provided | |
| image_analysis = None | |
| if image is not None: | |
| # Preprocess image | |
| processed_image = preprocess_image(image) | |
| # Get model prediction | |
| prediction = model.predict(processed_image) | |
| # Format prediction results | |
| image_analysis = format_prediction(prediction) | |
| # Create chat session and submit query | |
| session_id = create_chat_session() | |
| llm_response = submit_query(session_id, patient_info, | |
| json.dumps(image_analysis) if image_analysis else None) | |
| if not llm_response or 'data' not in llm_response or 'answer' not in llm_response['data']: | |
| raise ValueError("Invalid response structure from LLM") | |
| # Extract and clean JSON from the response | |
| json_data = extract_json_from_answer(llm_response['data']['answer']) | |
| # Format output for better readability | |
| return json.dumps(json_data, indent=2) | |
| except Exception as e: | |
| logger.error(f"Error in gradio_interface: {str(e)}") | |
| return json.dumps({"error": str(e)}, indent=2) | |
| # Gradio interface | |
| iface = gr.Interface( | |
| fn=gradio_interface, | |
| inputs=[ | |
| gr.Textbox( | |
| label="Patient Information", | |
| placeholder="Enter patient details including: symptoms, medical history, current medications, age, gender, and any relevant test results...", | |
| lines=5, | |
| max_lines=10 | |
| ), | |
| gr.Image( | |
| label="Medical Image", | |
| type="numpy", | |
| optional=True | |
| ) | |
| ], | |
| outputs=gr.Textbox( | |
| label="Medical Analysis", | |
| placeholder="JSON analysis will appear here...", | |
| lines=15 | |
| ), | |
| title="Medical Diagnosis Assistant", | |
| description="Enter patient information and optionally upload a medical image for analysis." | |
| ) | |
| if __name__ == "__main__": | |
| iface.launch() | |
| # import requests | |
| # import gradio as gr | |
| # import logging | |
| # import json | |
| # # Set up logging | |
| # logging.basicConfig(level=logging.INFO) | |
| # logger = logging.getLogger(__name__) | |
| # # API key and user ID for on-demand | |
| # api_key = 'KGSjxB1uptfSk8I8A7ciCuNT9Xa3qWC3' | |
| # external_user_id = 'plugin-1717464304' | |
| # def create_chat_session(): | |
| # try: | |
| # create_session_url = 'https://api.on-demand.io/chat/v1/sessions' | |
| # create_session_headers = { | |
| # 'apikey': api_key, | |
| # 'Content-Type': 'application/json' | |
| # } | |
| # create_session_body = { | |
| # "pluginIds": [], | |
| # "externalUserId": external_user_id | |
| # } | |
| # response = requests.post(create_session_url, headers=create_session_headers, json=create_session_body) | |
| # response.raise_for_status() | |
| # return response.json()['data']['id'] | |
| # except requests.exceptions.RequestException as e: | |
| # logger.error(f"Error creating chat session: {str(e)}") | |
| # raise | |
| # def submit_query(session_id, query): | |
| # try: | |
| # submit_query_url = f'https://api.on-demand.io/chat/v1/sessions/{session_id}/query' | |
| # submit_query_headers = { | |
| # 'apikey': api_key, | |
| # 'Content-Type': 'application/json' | |
| # } | |
| # structured_query = f""" | |
| # Based on the following patient information, provide a detailed medical analysis in JSON format: | |
| # {query} | |
| # Return only valid JSON with these fields: | |
| # - diagnosis_details | |
| # - probable_diagnoses (array) | |
| # - treatment_plans (array) | |
| # - lifestyle_modifications (array) | |
| # - medications (array of objects with name and dosage) | |
| # - additional_tests (array) | |
| # - precautions (array) | |
| # - follow_up (string) | |
| # """ | |
| # submit_query_body = { | |
| # "endpointId": "predefined-openai-gpt4o", | |
| # "query": structured_query, | |
| # "pluginIds": ["plugin-1712327325", "plugin-1713962163"], | |
| # "responseMode": "sync" | |
| # } | |
| # response = requests.post(submit_query_url, headers=submit_query_headers, json=submit_query_body) | |
| # response.raise_for_status() | |
| # return response.json() | |
| # except requests.exceptions.RequestException as e: | |
| # logger.error(f"Error submitting query: {str(e)}") | |
| # raise | |
| # def extract_json_from_answer(answer): | |
| # """Extract and clean JSON from the LLM response""" | |
| # try: | |
| # # First try to parse the answer directly | |
| # return json.loads(answer) | |
| # except json.JSONDecodeError: | |
| # try: | |
| # # If that fails, try to find JSON content and parse it | |
| # start_idx = answer.find('{') | |
| # end_idx = answer.rfind('}') + 1 | |
| # if start_idx != -1 and end_idx != 0: | |
| # json_str = answer[start_idx:end_idx] | |
| # return json.loads(json_str) | |
| # except (json.JSONDecodeError, ValueError): | |
| # logger.error("Failed to parse JSON from response") | |
| # raise | |
| # def gradio_interface(patient_info): | |
| # try: | |
| # session_id = create_chat_session() | |
| # llm_response = submit_query(session_id, patient_info) | |
| # if not llm_response or 'data' not in llm_response or 'answer' not in llm_response['data']: | |
| # raise ValueError("Invalid response structure") | |
| # # Extract and clean JSON from the response | |
| # json_data = extract_json_from_answer(llm_response['data']['answer']) | |
| # # Return clean JSON string without extra formatting | |
| # return json.dumps(json_data) | |
| # except Exception as e: | |
| # logger.error(f"Error in gradio_interface: {str(e)}") | |
| # return json.dumps({"error": str(e)}) | |
| # # Gradio interface | |
| # iface = gr.Interface( | |
| # fn=gradio_interface, | |
| # inputs=[ | |
| # gr.Textbox( | |
| # label="Patient Information", | |
| # placeholder="Enter patient details including: symptoms, medical history, current medications, age, gender, and any relevant test results...", | |
| # lines=5, | |
| # max_lines=10 | |
| # ) | |
| # ], | |
| # outputs=gr.Textbox( | |
| # label="Medical Analysis", | |
| # placeholder="JSON analysis will appear here...", | |
| # lines=15 | |
| # ), | |
| # title="Medical Diagnosis Assistant", | |
| # description="Enter detailed patient information to receive a structured medical analysis in JSON format." | |
| # ) | |
| # if __name__ == "__main__": | |
| # iface.launch() |