diff --git a/.gitignore b/.gitignore index e542d43a..458ae12f 100644 --- a/.gitignore +++ b/.gitignore @@ -233,4 +233,10 @@ sct2_Description_Full-en_INT_20250501.txt output.csv Disease Relationships Data.csv onlyCUIs.csv -MRCONSO.RRF \ No newline at end of file +MRCONSO.RRF + +# CV +emotion_dataset_1percent +emotion_dataset_5percent +emotion_dataset_1percent_yolo +full_emotion_dataset \ No newline at end of file diff --git a/3d/Frontend/src/app/consultations/page.tsx b/3d/Frontend/src/app/consultations/page.tsx index b700249f..2988154c 100644 --- a/3d/Frontend/src/app/consultations/page.tsx +++ b/3d/Frontend/src/app/consultations/page.tsx @@ -192,10 +192,7 @@ export default function ConsultationsPage() { Back

My Consultations

- + {/* Removed Book New button */}
@@ -240,8 +237,7 @@ export default function ConsultationsPage() { router.push('/book-consultation')} + /* Removed actionText and onAction for booking new consultation */ /> ) : ( <> diff --git a/computer-vision/claude/prompts/skin_prompt.py b/computer-vision/claude/prompts/skin_prompt.py new file mode 100644 index 00000000..7cae2e87 --- /dev/null +++ b/computer-vision/claude/prompts/skin_prompt.py @@ -0,0 +1,26 @@ +skin_prompt = """ +You are a medical assistant specialized in dermatology. Given the attached photo of a patientโ€™s skin condition, analyze the image in detail and provide a structured, comprehensive report that could be used by an AI doctor for diagnosis. + +Please follow this structured output format exactly: + +General Description +- Summarize what you see in 1โ€“2 sentences. +- Note the location on the body if it's clear (e.g. arm, leg, face). + +Lesion/Morphology Description +- Shape: (e.g. round, oval, irregular) +- Size estimate: (approximate in cm or mm if possible) +- Color(s): (e.g. erythematous, brown, black, hypopigmented) +- Borders: (well-defined, poorly defined, raised, flat) +- Surface texture: (smooth, scaly, crusted, ulcerated, verrucous) +- Number of lesions: (single, multiple, grouped) +- Distribution: (localized, widespread, linear, dermatomal, symmetrical) + +Associated Findings +- Signs of infection (pus, crusting) +- Signs of inflammation (redness, swelling) +- Signs of bleeding, ulceration +- Atrophy or scarring +- Other skin features nearby + +""" \ No newline at end of file diff --git a/computer-vision/claude/skin_conditions.py b/computer-vision/claude/skin_conditions.py new file mode 100644 index 00000000..ac6485cb --- /dev/null +++ b/computer-vision/claude/skin_conditions.py @@ -0,0 +1,100 @@ +import cv2 +import anthropic +import base64 +import os +from datetime import datetime +from prompts.skin_prompt import skin_prompt +import os + +# Initialize client +client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY")) + +def analyze_skin_condition(image_path): + # Read and encode image + with open(image_path, "rb") as image_file: + image_data = base64.b64encode(image_file.read()).decode('utf-8') + + # Send to Claude + message = client.messages.create( + model="claude-3-5-sonnet-20241022", + max_tokens=1000, + messages=[ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": "image/png", + "data": image_data + } + }, + { + "type": "text", + "text": skin_prompt + } + ] + } + ] + ) + + return message.content[0].text + +def main(): + # Initialize camera + cap = cv2.VideoCapture(0) + + if not cap.isOpened(): + print("Error: Could not open camera") + return + + # Create images directory if it doesn't exist + os.makedirs("computer-vision/claude/images", exist_ok=True) + + print("Camera started. Press 't' to take picture and analyze, 'q' to quit") + + while True: + # Read frame from camera + ret, frame = cap.read() + + if not ret: + print("Error: Could not read frame") + break + + # Display the frame + cv2.imshow('Camera Feed - Press "t" to capture', frame) + + # Check for key press + key = cv2.waitKey(1) & 0xFF + + if key == ord('t'): + # Generate timestamp for unique filename + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + image_path = f"computer-vision/claude/images/capture_{timestamp}.png" + + # Save the frame + cv2.imwrite(image_path, frame) + print(f"\nImage captured: {image_path}") + print("Analyzing skin condition...") + + try: + # Analyze the captured image + result = analyze_skin_condition(image_path) + print("\n" + "="*50) + print("SKIN ANALYSIS RESULT:") + print("="*50) + print(result) + print("="*50 + "\n") + except Exception as e: + print(f"Error analyzing image: {e}") + + elif key == ord('q'): + break + + # Clean up + cap.release() + cv2.destroyAllWindows() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/computer-vision/fullpicode/gpu_server_code.py b/computer-vision/fullpicode/gpu_server_code.py new file mode 100644 index 00000000..35492416 --- /dev/null +++ b/computer-vision/fullpicode/gpu_server_code.py @@ -0,0 +1,470 @@ +#!/usr/bin/env python3 +""" +Complete GPU Server for Medical Analysis +Fixed LLaVA output and added quick detection endpoint +""" + +import os +import json +import time +import torch +import cv2 +import numpy as np +from datetime import datetime +from PIL import Image +import io + +# Flask for API +from flask import Flask, request, jsonify + +# YOLO for object detection +import ultralytics +from ultralytics import YOLO + +# Transformers for LLaVA and Clinical AI +from transformers import ( + AutoTokenizer, AutoModelForCausalLM, + LlavaNextProcessor, LlavaNextForConditionalGeneration +) + +# ============================================================================ +# CONFIGURATION +# ============================================================================ +YOLO_MODEL_PATH = "yolov8n.pt" +VISION_MODEL_NAME = "llava-hf/llava-v1.6-mistral-7b-hf" +CLINICAL_MODEL_PATH = "CodCodingCode/llama-3.1-8b-clinical-V1.4" + +HF_TOKEN = "_" + +HAND_CONFIDENCE_THRESHOLD = 0.2 # Lowered for better detection +TARGET_CLASSES = ['person', 'hand'] + +app = Flask(__name__) + +# Global model storage +models = { + 'yolo': None, + 'vision_processor': None, + 'vision_model': None, + 'clinical_tokenizer': None, + 'clinical_model': None +} + +# ============================================================================ +# BASE PATIENT INFORMATION +# ============================================================================ +BASE_PATIENT_INFO = """ +Patient presents for medical evaluation with concerns about a visible condition. +Initial assessment indicates the need for further clinical questioning to establish +proper diagnosis and treatment plan. Patient appears cooperative and willing to +provide medical history. Vital signs stable at time of presentation. +""" + +# ============================================================================ +# MEDICAL PROMPTS +# ============================================================================ +MEDICAL_IMAGE_ANALYSIS_PROMPT = """ +You are an AI-powered medical assistant. Analyze this medical image and provide a detailed clinical description of what you observe. Focus on medically relevant findings that would assist in diagnosis. + +Describe in a single paragraph what is shown in the image, including: +- Body part/location visible +- Any abnormalities, lesions, discoloration, or concerning findings +- Size, shape, and characteristics of any findings +- Signs of inflammation, injury, or infection +- Any other clinically relevant observations + +Be specific and clinical in your description. +""" + +# ============================================================================ +# MODEL LOADING +# ============================================================================ +def load_yolo_model(): + try: + print("๐Ÿ”„ Loading YOLO model...") + model = YOLO(YOLO_MODEL_PATH) + print("โœ… YOLO model loaded") + return model + except Exception as e: + print(f"โŒ YOLO load failed: {e}") + return None + +def load_vision_models(): + try: + print("๐Ÿ”„ Loading LLaVA vision model...") + processor = LlavaNextProcessor.from_pretrained(VISION_MODEL_NAME, token=HF_TOKEN) + model = LlavaNextForConditionalGeneration.from_pretrained( + VISION_MODEL_NAME, + token=HF_TOKEN, + torch_dtype=torch.float16, + device_map="auto" + ) + processor.tokenizer.padding_side = "left" + print("โœ… LLaVA model loaded") + return processor, model + except Exception as e: + print(f"โŒ LLaVA load failed: {e}") + return None, None + +def load_clinical_model(): + try: + print("๐Ÿ”„ Loading clinical model...") + tokenizer = AutoTokenizer.from_pretrained(CLINICAL_MODEL_PATH, token=HF_TOKEN, trust_remote_code=True) + if tokenizer.pad_token is None: + tokenizer.pad_token = tokenizer.eos_token + model = AutoModelForCausalLM.from_pretrained( + CLINICAL_MODEL_PATH, + token=HF_TOKEN, + device_map="auto", + torch_dtype=torch.bfloat16, + trust_remote_code=True, + low_cpu_mem_usage=True + ) + model.eval() + print("โœ… Clinical model loaded") + return tokenizer, model + except Exception as e: + print(f"โŒ Clinical model load failed: {e}") + return None, None + +def initialize_models(): + print("๐Ÿš€ Initializing all models...") + models['yolo'] = load_yolo_model() + models['vision_processor'], models['vision_model'] = load_vision_models() + models['clinical_tokenizer'], models['clinical_model'] = load_clinical_model() + + success = all(model is not None for model in models.values()) + if success: + print("โœ… All models loaded successfully!") + else: + print("โŒ Some models failed to load") + return success + +# ============================================================================ +# ANALYSIS PIPELINE +# ============================================================================ +def detect_objects(image_bytes): + try: + nparr = np.frombuffer(image_bytes, np.uint8) + image = cv2.imdecode(nparr, cv2.IMREAD_COLOR) + results = models['yolo'](image, conf=HAND_CONFIDENCE_THRESHOLD) + + detections = [] + for result in results: + if result.boxes is not None: + for box in result.boxes: + class_id = int(box.cls) + confidence = float(box.conf) + class_name = models['yolo'].names[class_id] + if class_name in TARGET_CLASSES: + detections.append({ + 'class': class_name, + 'confidence': confidence + }) + + detections.sort(key=lambda x: x['confidence'], reverse=True) + return detections[0] if detections else None + + except Exception as e: + print(f"โŒ Detection error: {e}") + return None + +def analyze_with_llava(image_bytes): + try: + print("๐Ÿ”„ Starting LLaVA image analysis...") + image = Image.open(io.BytesIO(image_bytes)).convert("RGB") + + conversation = [{ + "role": "user", + "content": [ + {"type": "image"}, + {"type": "text", "text": MEDICAL_IMAGE_ANALYSIS_PROMPT}, + ], + }] + + text_prompt = models['vision_processor'].apply_chat_template(conversation, add_generation_prompt=True) + inputs = models['vision_processor'](image, text_prompt, return_tensors="pt").to(models['vision_model'].device) + + print("๐Ÿ”„ LLaVA generating analysis...") + with torch.no_grad(): + output = models['vision_model'].generate( + **inputs, + max_new_tokens=400, + do_sample=False, + temperature=0.1 + ) + + response = models['vision_processor'].decode(output[0], skip_special_tokens=True) + assistant_start = response.find("[/INST]") + len("[/INST]") + if assistant_start > len("[/INST]") - 1: + analysis = response[assistant_start:].strip() + else: + analysis = response + + print("โœ… LLaVA analysis completed!") + print("=" * 60) + print("๐Ÿ‘๏ธ LLAVA MEDICAL IMAGE ANALYSIS:") + print("=" * 60) + print(analysis) + print("=" * 60) + + return analysis + + except Exception as e: + print(f"โŒ LLaVA analysis error: {e}") + return None + +def generate_clinical_text(prompt, max_tokens=500): + try: + inputs = models['clinical_tokenizer']( + prompt, + return_tensors="pt", + truncation=True, + max_length=1500, + padding=False + ) + + device = next(models['clinical_model'].parameters()).device + inputs = {k: v.to(device) for k, v in inputs.items()} + + with torch.no_grad(): + outputs = models['clinical_model'].generate( + **inputs, + max_new_tokens=max_tokens, + temperature=0.2, + do_sample=True, + repetition_penalty=1.1, + pad_token_id=models['clinical_tokenizer'].pad_token_id, + eos_token_id=models['clinical_tokenizer'].eos_token_id + ) + + generated_text = models['clinical_tokenizer'].decode(outputs[0], skip_special_tokens=True) + return generated_text[len(prompt):].strip() + + except Exception as e: + print(f"โŒ Clinical generation error: {e}") + return None + +def run_medical_pipeline(image_bytes): + """ + COMPLETE PIPELINE: Image โ†’ YOLO โ†’ LLaVA โ†’ Combine โ†’ Summarizer โ†’ Diagnoser โ†’ Questioner + """ + + # Step 1: Object Detection + print("๐Ÿ” Step 1: Object Detection") + detection = detect_objects(image_bytes) + if not detection: + return {"error": "No relevant objects detected"} + print(f"โœ… Detected: {detection['class']} ({detection['confidence']:.1%})") + + # Step 2: LLaVA Analysis + print("๐Ÿ” Step 2: LLaVA Analysis") + llava_analysis = analyze_with_llava(image_bytes) + if not llava_analysis: + return {"error": "LLaVA analysis failed"} + + # Step 3: Combine with base patient info + print("๐Ÿ” Step 3: Combining patient information") + combined_patient_info = f"""{BASE_PATIENT_INFO} + +Medical Image Analysis: {llava_analysis}""" + + print("๐Ÿ“‹ Combined Patient Info:") + print("-" * 40) + print(combined_patient_info) + print("-" * 40) + + # Step 4: Clinical Summary Generation + print("๐Ÿ” Step 4: Clinical Summarizer") + summary_prompt = f"""You are a clinical summarizer. Given the following patient information, extract and summarize the key clinical findings into a structured clinical vignette. + +Patient Information: +{combined_patient_info} + +Clinical Summary:""" + + clinical_summary = generate_clinical_text(summary_prompt, max_tokens=400) + if not clinical_summary: + return {"error": "Clinical summary failed"} + + print("๐Ÿ“‹ Clinical Summary:") + print("-" * 40) + print(clinical_summary) + print("-" * 40) + + # Step 5: Diagnostic Reasoning + print("๐Ÿ” Step 5: Diagnostic Reasoning") + diagnostic_prompt = f"""You are a diagnostic reasoning model. Based on the clinical summary, generate a list of plausible diagnoses with reasoning. + +Clinical Summary: +{clinical_summary} + +Diagnostic Analysis:""" + + diagnostic_analysis = generate_clinical_text(diagnostic_prompt, max_tokens=400) + if not diagnostic_analysis: + return {"error": "Diagnostic analysis failed"} + + print("๐Ÿ”ฌ Diagnostic Analysis:") + print("-" * 40) + print(diagnostic_analysis) + print("-" * 40) + + # Step 6: Question Generation + print("๐Ÿ” Step 6: Question Generation") + question_prompt = f"""You are a medical questioning agent. Based on the clinical summary and diagnostic analysis, generate ONE specific, relevant medical question to gather more information for diagnosis. + +Clinical Summary: +{clinical_summary} + +Diagnostic Analysis: +{diagnostic_analysis} + +Generate ONE clear, specific medical question:""" + + medical_question = generate_clinical_text(question_prompt, max_tokens=200) + if not medical_question: + return {"error": "Question generation failed"} + + # Clean up the question + medical_question = medical_question.split('\n')[0].strip() + if not medical_question.endswith('?'): + medical_question += '?' + + print("โ“ Generated Medical Question:") + print("=" * 60) + print(f"๐Ÿฉบ {medical_question}") + print("=" * 60) + + return { + "success": True, + "detection": detection, + "llava_analysis": llava_analysis, + "combined_patient_info": combined_patient_info, + "clinical_summary": clinical_summary, + "diagnostic_analysis": diagnostic_analysis, + "medical_question": medical_question, + "timestamp": datetime.now().isoformat() + } + +# ============================================================================ +# FLASK ROUTES +# ============================================================================ +@app.route("/") +def home(): + return jsonify({ + "status": "online", + "message": "Complete Medical Analysis Server", + "models_loaded": all(model is not None for model in models.values()), + "gpu_available": torch.cuda.is_available() + }) + +@app.route("/quick_detect", methods=["POST"]) +def quick_detect(): + """Quick YOLO detection only - no LLaVA analysis""" + try: + if 'image' not in request.files: + return jsonify({"error": "No image provided"}), 400 + + image_file = request.files['image'] + image_bytes = image_file.read() + + print(f"๐Ÿ” Quick detection check for {len(image_bytes)} byte image") + + # Only run YOLO detection + detection = detect_objects(image_bytes) + + if detection: + print(f"โœ… Quick detect: {detection['class']} ({detection['confidence']:.1%})") + return jsonify({ + "detected": True, + "detection": detection + }) + else: + print("โŒ Quick detect: No objects found") + return jsonify({ + "detected": False + }) + + except Exception as e: + print(f"โŒ Quick detection error: {e}") + return jsonify({"error": str(e)}), 500 + +@app.route("/analyze_frame", methods=["POST"]) +def analyze_frame(): + """ + COMPLETE PIPELINE: Image โ†’ YOLO โ†’ LLaVA โ†’ Summary โ†’ Diagnosis โ†’ Question + """ + try: + if 'image' not in request.files: + return jsonify({"error": "No image provided"}), 400 + + image_file = request.files['image'] + image_bytes = image_file.read() + + print(f"\n๐Ÿš€ Starting COMPLETE medical pipeline for {len(image_bytes)} byte image") + print("=" * 80) + + # Run the complete pipeline + result = run_medical_pipeline(image_bytes) + + if "error" in result: + print(f"โŒ Pipeline failed: {result['error']}") + return jsonify(result), 500 + + print("โœ… Complete medical pipeline finished successfully!") + print("=" * 80) + return jsonify(result) + + except Exception as e: + print(f"โŒ Pipeline error: {e}") + import traceback + traceback.print_exc() + return jsonify({"error": str(e)}), 500 + +@app.route("/health", methods=["GET"]) +def health_check(): + return jsonify({ + "status": "healthy", + "models_loaded": { + "yolo": models['yolo'] is not None, + "llava": models['vision_model'] is not None, + "clinical": models['clinical_model'] is not None + }, + "gpu_available": torch.cuda.is_available(), + "timestamp": datetime.now().isoformat() + }) + +# ============================================================================ +# MAIN EXECUTION +# ============================================================================ +def main(): + print("๐Ÿฅ Complete Medical Analysis Server") + print("=" * 50) + print("๐Ÿ“‹ Pipeline: Image โ†’ YOLO โ†’ LLaVA โ†’ Summary โ†’ Diagnosis โ†’ Question") + print(f"๐Ÿ” YOLO Detection Threshold: {HAND_CONFIDENCE_THRESHOLD}") + print(f"๐ŸŽฏ Target Classes: {TARGET_CLASSES}") + print(f"๐ŸŽฎ GPU Available: {torch.cuda.is_available()}") + + if torch.cuda.is_available(): + print(f"๐Ÿ”ง GPU Count: {torch.cuda.device_count()}") + for i in range(torch.cuda.device_count()): + props = torch.cuda.get_device_properties(i) + print(f"๐Ÿ’พ GPU {i}: {props.name} ({props.total_memory / 1e9:.1f} GB)") + + if not initialize_models(): + print("โŒ Failed to initialize models - exiting") + return + + print("\n๐ŸŒ Starting Flask server...") + print("๐Ÿ“‹ Endpoints:") + print(" GET / - Server status") + print(" POST /quick_detect - Fast YOLO detection only") + print(" POST /analyze_frame - Complete medical pipeline") + print(" GET /health - Health check") + print("\n๐Ÿš€ Server ready!") + + app.run(host="0.0.0.0", port=7860, debug=False, threaded=True) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/computer-vision/fullpicode/no_touchscreen.py b/computer-vision/fullpicode/no_touchscreen.py new file mode 100644 index 00000000..2ab92d4e --- /dev/null +++ b/computer-vision/fullpicode/no_touchscreen.py @@ -0,0 +1,365 @@ +#!/usr/bin/env python3 +""" +Complete Live Medical Scanner for Raspberry Pi +Uses quick detection endpoint and complete analysis pipeline +""" + +import io +import requests +import time +import cv2 +import numpy as np +from datetime import datetime +from picamera2 import Picamera2 +import threading + +# Configuration +GPU_SERVER_URL = "http://192.222.58.119:7860" +DETECTION_INTERVAL = 3 # Check for detection every 3 seconds +ANALYSIS_COOLDOWN = 10 # Wait 10 seconds between full analyses +DETECTION_CONFIDENCE = 0.4 # Minimum confidence for triggering analysis + +class CompleteLiveScanner: + def __init__(self): + self.camera = None + self.running = False + self.last_analysis_time = 0 + self.processing = False + + self.initialize_camera() + + def initialize_camera(self): + """Initialize camera for live monitoring""" + try: + self.camera = Picamera2() + + # Configure for live preview + config = self.camera.create_preview_configuration( + main={"size": (640, 480)} + ) + self.camera.configure(config) + + print("โœ… Camera initialized for live detection") + except Exception as e: + print(f"โŒ Camera initialization failed: {e}") + self.camera = None + + def check_server(self): + """Check if GPU server is accessible""" + try: + response = requests.get(f"{GPU_SERVER_URL}/", timeout=10) + if response.status_code == 200: + data = response.json() + print(f"โœ… GPU server online - Models loaded: {data.get('models_loaded', 'Unknown')}") + return True + return False + except Exception as e: + print(f"โŒ GPU server check failed: {e}") + return False + + def capture_detection_frame(self): + """Capture low-res frame for detection check""" + try: + stream = io.BytesIO() + self.camera.capture_file(stream, format='jpeg') + stream.seek(0) + return stream.getvalue() + except Exception as e: + print(f"โŒ Detection frame capture failed: {e}") + return None + + def capture_analysis_frame(self): + """Capture high-res frame for full analysis""" + try: + print("๐Ÿ“ธ Capturing high-resolution image for complete analysis...") + + # Switch to still configuration temporarily + self.camera.stop() + + still_config = self.camera.create_still_configuration( + main={"size": (1024, 768)} + ) + self.camera.configure(still_config) + self.camera.start() + time.sleep(1) # Longer stabilization for better image + + # Capture high-res image + stream = io.BytesIO() + self.camera.capture_file(stream, format='jpeg') + stream.seek(0) + image_bytes = stream.getvalue() + + # Return to preview mode + self.camera.stop() + preview_config = self.camera.create_preview_configuration( + main={"size": (640, 480)} + ) + self.camera.configure(preview_config) + self.camera.start() + + print(f"โœ… Captured {len(image_bytes)} bytes for analysis") + return image_bytes + + except Exception as e: + print(f"โŒ Analysis frame capture failed: {e}") + return None + + def quick_detection_check(self, image_bytes): + """Send frame to GPU server for quick YOLO detection only""" + try: + files = {'image': ('check.jpg', image_bytes, 'image/jpeg')} + + # Use quick detection endpoint with short timeout + response = requests.post( + f"{GPU_SERVER_URL}/quick_detect", + files=files, + timeout=15 + ) + + if response.status_code == 200: + result = response.json() + if result.get('detected', False): + detection = result.get('detection', {}) + if detection.get('confidence', 0) >= DETECTION_CONFIDENCE: + return detection + return None + + except requests.exceptions.Timeout: + print("โฑ๏ธ Quick detection timed out") + return None + except Exception as e: + print(f"โŒ Quick detection failed: {e}") + return None + + def run_complete_analysis(self, image_bytes): + """Send high-res image for complete medical analysis pipeline""" + try: + print("๐Ÿš€ Running COMPLETE medical analysis pipeline...") + + files = {'image': ('analysis.jpg', image_bytes, 'image/jpeg')} + response = requests.post( + f"{GPU_SERVER_URL}/analyze_frame", + files=files, + timeout=300 # 5 minutes for complete pipeline + ) + + if response.status_code == 200: + result = response.json() + if result.get('success'): + return result + else: + print(f"โŒ Pipeline failed: {result.get('error')}") + return None + else: + print(f"โŒ Server error {response.status_code}: {response.text}") + return None + + except requests.exceptions.Timeout: + print("โฑ๏ธ Complete analysis timed out (this is normal for LLaVA)") + return None + except Exception as e: + print(f"โŒ Complete analysis failed: {e}") + return None + + def display_results(self, result): + """Display complete medical analysis results""" + print("\n" + "=" * 80) + print("๐ŸŽ‰ AUTOMATIC MEDICAL ANALYSIS COMPLETE!") + print("=" * 80) + + # Detection info + detection = result['detection'] + print(f"\n๐ŸŽฏ Auto-detected: {detection['class']} ({detection['confidence']:.1%} confidence)") + + # LLaVA analysis + print(f"\n๐Ÿ‘๏ธ LLaVA Visual Analysis:") + print("โ”€" * 50) + print(result['llava_analysis']) + + # Combined patient info + print(f"\n๐Ÿ“‹ Combined Patient Information:") + print("โ”€" * 50) + print(result['combined_patient_info']) + + # Clinical summary + print(f"\n๐Ÿฉบ Clinical Summary:") + print("โ”€" * 50) + print(result['clinical_summary']) + + # Diagnostic analysis + print(f"\n๐Ÿ”ฌ Diagnostic Analysis:") + print("โ”€" * 50) + print(result['diagnostic_analysis']) + + # Final medical question + print(f"\n" + "=" * 80) + print("โ“ FINAL MEDICAL QUESTION") + print("=" * 80) + print(f"๐Ÿฉบ {result['medical_question']}") + print("=" * 80) + + # Save timestamp + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"\nโฐ Analysis completed at: {timestamp}") + print(f"๐Ÿ‘๏ธ Resuming live detection in {ANALYSIS_COOLDOWN} seconds...\n") + + def process_detection(self, detection): + """Process detected object (runs in separate thread)""" + try: + print(f"\n๐ŸŽฏ PROCESSING: {detection['class']} ({detection['confidence']:.1%} confidence)") + print("๐Ÿ”„ Starting complete medical analysis pipeline...") + + # Capture high-resolution image + image_bytes = self.capture_analysis_frame() + if not image_bytes: + print("โŒ Failed to capture analysis image") + return + + # Run complete analysis pipeline + result = self.run_complete_analysis(image_bytes) + if result: + self.display_results(result) + else: + print(f"\nโŒ Complete analysis failed! Resuming detection...\n") + + except Exception as e: + print(f"โŒ Processing error: {e}") + import traceback + traceback.print_exc() + finally: + # Reset processing flag + self.processing = False + + def live_detection_loop(self): + """Main live detection loop""" + print("\n๐Ÿ”ด LIVE MEDICAL DETECTION ACTIVE") + print("=" * 50) + print("๐Ÿ‘๏ธ Camera continuously monitoring for person/hand...") + print("๐ŸŽฏ Will auto-capture and analyze when detected") + print(f"โšก Using quick detection checks every {DETECTION_INTERVAL}s") + print(f"๐Ÿ• Analysis cooldown: {ANALYSIS_COOLDOWN}s between analyses") + print("Press Ctrl+C to quit\n") + + detection_count = 0 + + while self.running: + try: + current_time = time.time() + detection_count += 1 + + # Status update + status = f"๐Ÿ‘€ Monitoring... ({detection_count}) - Press Ctrl+C to quit" + if self.processing: + status = "๐Ÿ”„ Processing analysis... - Press Ctrl+C to quit" + print(status, end='\r') + + # Skip detection if currently processing + if self.processing: + time.sleep(1) + continue + + # Capture frame for detection + frame_bytes = self.capture_detection_frame() + if not frame_bytes: + time.sleep(1) + continue + + # Quick detection check with GPU server + detection = self.quick_detection_check(frame_bytes) + + if detection: + # Check cooldown period + if current_time - self.last_analysis_time > ANALYSIS_COOLDOWN: + print(f"\n๐ŸŽฏ DETECTED: {detection['class']} ({detection['confidence']:.1%})") + print("๐Ÿš€ Triggering complete medical analysis...") + + # Set processing flag and timestamp + self.processing = True + self.last_analysis_time = current_time + + # Process in separate thread + analysis_thread = threading.Thread( + target=self.process_detection, + args=(detection,) + ) + analysis_thread.start() + else: + remaining = ANALYSIS_COOLDOWN - (current_time - self.last_analysis_time) + print(f"\n๐Ÿ• Detection on cooldown ({remaining:.1f}s remaining)") + + # Wait before next detection check + time.sleep(DETECTION_INTERVAL) + + except KeyboardInterrupt: + break + except Exception as e: + print(f"\nโŒ Detection loop error: {e}") + time.sleep(3) + + print("\n๐Ÿ›‘ Live detection stopped") + + def start_live_detection(self): + """Start the live detection system""" + if not self.camera: + print("โŒ No camera available") + return False + + if not self.check_server(): + print("โŒ GPU server not accessible") + return False + + try: + # Start camera + self.camera.start() + print("๐Ÿ“ท Camera warming up...") + time.sleep(3) + + self.running = True + + # Start detection loop + self.live_detection_loop() + + return True + + except Exception as e: + print(f"โŒ Live detection failed: {e}") + return False + finally: + self.cleanup() + + def cleanup(self): + """Clean up resources""" + try: + self.running = False + if self.camera: + self.camera.stop() + self.camera.close() + print("๐Ÿ“ท Camera closed") + except Exception as e: + print(f"โš ๏ธ Cleanup warning: {e}") + +def main(): + print("๐Ÿฅ Complete Live Medical Detection Scanner") + print("=" * 60) + print(f"๐ŸŒ GPU Server: {GPU_SERVER_URL}") + print("๐Ÿ“‹ Pipeline: Quick Detect โ†’ Capture โ†’ Complete Analysis โ†’ Question") + print(f"๐ŸŽฏ Detection Confidence: {DETECTION_CONFIDENCE}") + print(f"โฑ๏ธ Detection Interval: {DETECTION_INTERVAL}s") + print(f"๐Ÿ• Analysis Cooldown: {ANALYSIS_COOLDOWN}s") + + scanner = CompleteLiveScanner() + + try: + print("\n๐Ÿš€ Starting live medical detection...") + success = scanner.start_live_detection() + + if not success: + print("โŒ Failed to start live detection") + + except KeyboardInterrupt: + print("\n๐Ÿ‘‹ Goodbye!") + finally: + scanner.cleanup() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/computer-vision/fullpicode/pi_controlled_code.py b/computer-vision/fullpicode/pi_controlled_code.py new file mode 100644 index 00000000..9a057f23 --- /dev/null +++ b/computer-vision/fullpicode/pi_controlled_code.py @@ -0,0 +1,584 @@ +import tkinter as tk +from tkinter import font, messagebox +import threading +import time +import io +import requests +from PIL import Image, ImageTk +import json +from datetime import datetime +from picamera2 import Picamera2 +CAMERA_TYPE = "picamera2" + +# Server URL - UPDATE THIS TO YOUR GPU SERVER IP +UNIFIED_SERVER = "http://104.171.203.124:7860" + +class MedicalScannerApp: + def __init__(self, root): + self.root = root + self.root.title("Medical Skin Scanner") + self.root.attributes('-fullscreen', True) # Fullscreen for touchscreen + self.root.configure(bg='#2c3e50') + + # State variables + self.scanning = False + self.session_id = None + self.conversation_history = [] + self.camera = None + self.current_iteration = 0 + + # Setup fonts + self.title_font = font.Font(family="Arial", size=24, weight="bold") + self.instruction_font = font.Font(family="Arial", size=18) + self.button_font = font.Font(family="Arial", size=16, weight="bold") + self.question_font = font.Font(family="Arial", size=16) + self.status_font = font.Font(family="Arial", size=14) + + self.setup_ui() + self.initialize_camera() + + def setup_ui(self): + """Setup the touchscreen interface""" + # Main container + main_frame = tk.Frame(self.root, bg='#2c3e50') + main_frame.pack(fill=tk.BOTH, expand=True, padx=20, pady=20) + + # Title + self.title_label = tk.Label( + main_frame, + text="๐Ÿฅ Medical Skin Scanner", + font=self.title_font, + fg='white', + bg='#2c3e50' + ) + self.title_label.pack(pady=(0, 20)) + + # Status display + self.status_label = tk.Label( + main_frame, + text="Ready to scan", + font=self.status_font, + fg='#3498db', + bg='#2c3e50', + wraplength=600 + ) + self.status_label.pack(pady=(0, 10)) + + # Main instruction/question display + self.instruction_label = tk.Label( + main_frame, + text="Touch 'Start Scan' to begin analyzing your skin condition", + font=self.question_font, + fg='white', + bg='#2c3e50', + wraplength=700, + justify=tk.CENTER + ) + self.instruction_label.pack(pady=(0, 30)) + + # Progress indicator + self.progress_label = tk.Label( + main_frame, + text="", + font=self.status_font, + fg='#f39c12', + bg='#2c3e50' + ) + self.progress_label.pack(pady=(0, 20)) + + # Button frame + button_frame = tk.Frame(main_frame, bg='#2c3e50') + button_frame.pack(pady=20) + + # Start scan button + self.start_button = tk.Button( + button_frame, + text="๐Ÿ” Start Scan", + font=self.button_font, + bg='#27ae60', + fg='white', + activebackground='#2ecc71', + activeforeground='white', + padx=30, + pady=15, + command=self.start_scan + ) + self.start_button.pack(side=tk.LEFT, padx=10) + + # Reset button + self.reset_button = tk.Button( + button_frame, + text="๐Ÿ”„ New Scan", + font=self.button_font, + bg='#f39c12', + fg='white', + activebackground='#f4d03f', + activeforeground='white', + padx=20, + pady=15, + command=self.reset_conversation, + state=tk.DISABLED + ) + self.reset_button.pack(side=tk.LEFT, padx=10) + + # Response input frame (for patient responses) + self.response_frame = tk.Frame(main_frame, bg='#2c3e50') + self.response_frame.pack(pady=20, fill=tk.X) + + self.response_label = tk.Label( + self.response_frame, + text="Your response:", + font=self.instruction_font, + fg='white', + bg='#2c3e50' + ) + + # Text input with scrollbar + text_frame = tk.Frame(self.response_frame, bg='#2c3e50') + + self.response_entry = tk.Text( + text_frame, + height=4, + font=self.question_font, + wrap=tk.WORD, + bg='#ecf0f1', + fg='#2c3e50', + insertbackground='#2c3e50' + ) + + scrollbar = tk.Scrollbar(text_frame, orient="vertical", command=self.response_entry.yview) + self.response_entry.configure(yscrollcommand=scrollbar.set) + + self.response_entry.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) + scrollbar.pack(side=tk.RIGHT, fill=tk.Y) + + # Submit button + self.submit_button = tk.Button( + self.response_frame, + text="๐Ÿ“ค Submit Response", + font=self.button_font, + bg='#3498db', + fg='white', + activebackground='#5dade2', + activeforeground='white', + padx=20, + pady=10, + command=self.submit_response + ) + + # Initially hide response widgets + self.hide_response_input() + + # Bottom frame for exit and info + bottom_frame = tk.Frame(main_frame, bg='#2c3e50') + bottom_frame.pack(side=tk.BOTTOM, fill=tk.X, pady=10) + + # Connection status + self.connection_label = tk.Label( + bottom_frame, + text="Checking server connection...", + font=self.status_font, + fg='#95a5a6', + bg='#2c3e50' + ) + self.connection_label.pack(side=tk.LEFT) + + # Exit button + exit_button = tk.Button( + bottom_frame, + text="โŒ Exit", + font=self.button_font, + bg='#e74c3c', + fg='white', + activebackground='#ec7063', + activeforeground='white', + padx=20, + pady=10, + command=self.exit_app + ) + exit_button.pack(side=tk.RIGHT) + + # Check server connection on startup + threading.Thread(target=self.check_server_connection, daemon=True).start() + + def initialize_camera(self): + """Initialize the camera based on available library""" + try: + if CAMERA_TYPE == "picamera2": + self.camera = Picamera2() + # Configure camera for still capture + config = self.camera.create_still_configuration( + main={"size": (640, 480)} + ) + self.camera.configure(config) + print("โœ… Camera initialized with picamera2") + + elif CAMERA_TYPE == "picamera": + self.camera = picamera.PiCamera() + self.camera.resolution = (640, 480) + print("โœ… Camera initialized with picamera") + + else: + print("โŒ No camera available") + self.camera = None + self.update_status("โš ๏ธ No camera detected - Check camera connection") + + except Exception as e: + print(f"โŒ Camera initialization failed: {e}") + self.camera = None + self.update_status(f"โš ๏ธ Camera error: {str(e)}") + + def check_server_connection(self): + """Check if server is reachable""" + try: + response = requests.get(f"{UNIFIED_SERVER}/", timeout=5) + if response.status_code == 200: + self.root.after(0, lambda: self.connection_label.config( + text="๐ŸŸข Server connected", fg='#27ae60' + )) + else: + self.root.after(0, lambda: self.connection_label.config( + text="๐ŸŸก Server responding with errors", fg='#f39c12' + )) + except Exception as e: + self.root.after(0, lambda: self.connection_label.config( + text="๐Ÿ”ด Server disconnected", fg='#e74c3c' + )) + print(f"โŒ Cannot connect to server: {e}") + + def capture_image(self): + """Capture image using appropriate camera library""" + if not self.camera: + raise Exception("No camera available") + + try: + stream = io.BytesIO() + + if CAMERA_TYPE == "picamera2": + # Start camera if not already started + if not self.camera.started: + self.camera.start() + time.sleep(2) # Let camera warm up + + # Capture to stream + self.camera.capture_file(stream, format='jpeg') + + elif CAMERA_TYPE == "picamera": + # Start preview and capture + self.camera.start_preview() + time.sleep(2) # Camera warm-up time + self.camera.capture(stream, format='jpeg') + self.camera.stop_preview() + + stream.seek(0) + return stream + + except Exception as e: + print(f"โŒ Image capture failed: {e}") + raise + + def start_scan(self): + """Start the scanning process""" + if self.scanning: + return + + if not self.camera: + messagebox.showerror("Error", "No camera available. Please check camera connection and restart.") + return + + self.scanning = True + self.start_button.config(state=tk.DISABLED) + self.reset_button.config(state=tk.DISABLED) + self.update_status("๐Ÿ”„ Preparing camera...") + self.update_instruction("Please position your skin condition in front of the camera") + self.update_progress("") + + # Start scanning in separate thread + threading.Thread(target=self.scan_process, daemon=True).start() + + def scan_process(self): + """Main scanning process - runs in separate thread""" + try: + # Countdown + for i in range(3, 0, -1): + self.update_status(f"๐Ÿ“ธ Capturing in {i}...") + self.update_progress(f"{'โ—' * (4-i)}{'โ—‹' * (i-1)}") + time.sleep(1) + + self.update_status("๐Ÿ“ธ Capturing image...") + self.update_progress("โ—โ—โ—โ—") + + # Capture image + image_stream = self.capture_image() + + self.update_status("๐Ÿ” Analyzing image with AI...") + self.update_progress("Processing...") + + # Send to unified server for analysis + files = {'image': ('image.jpg', image_stream, 'image/jpeg')} + response = requests.post( + f"{UNIFIED_SERVER}/analyze_frame", + files=files, + timeout=60 # Increased timeout for AI processing + ) + + if response.status_code == 200: + result = response.json() + + if result['object_detected']: + confidence = result['detection_confidence'] + self.update_status(f"โœ… Object detected ({confidence:.1%} confidence)") + + if result['analysis']: + self.update_status("๐Ÿฉบ Starting medical consultation...") + # Start conversation with clinical AI + self.start_clinical_conversation(result['analysis']) + else: + self.update_status("โŒ Medical analysis failed") + self.reset_ui() + else: + self.update_status("โŒ No object detected. Please position yourself clearly in front of camera.") + self.update_instruction("Make sure your skin condition is clearly visible and try again.") + self.reset_ui() + else: + error_msg = response.json().get('error', f'Server error: {response.status_code}') + self.update_status(f"โŒ Server error: {error_msg}") + self.reset_ui() + + except requests.exceptions.Timeout: + self.update_status("โฑ๏ธ Request timed out. Server may be busy.") + self.reset_ui() + except Exception as e: + print(f"โŒ Scan process error: {e}") + self.update_status(f"โŒ Error: {str(e)}") + self.reset_ui() + + def start_clinical_conversation(self, image_analysis): + """Start conversation with clinical AI server""" + try: + self.update_status("๐Ÿฉบ Initializing medical consultation...") + + # Initialize conversation + payload = { + "session_id": f"scan_{int(time.time())}", + "image_analysis": image_analysis + } + + response = requests.post( + f"{UNIFIED_SERVER}/start_conversation", + json=payload, + timeout=60 + ) + + if response.status_code == 200: + result = response.json() + self.session_id = result['session_id'] + first_question = result['question'] + + # Update UI to show question and response input + self.update_status("โœ… Medical consultation started") + self.update_instruction(f"Doctor: {first_question}") + self.update_progress("Question 1/6") + self.current_iteration = 1 + self.show_response_input() + self.reset_button.config(state=tk.NORMAL) + + else: + error_msg = response.json().get('error', 'Unknown error') + self.update_status(f"โŒ Failed to start consultation: {error_msg}") + self.reset_ui() + + except Exception as e: + print(f"โŒ Clinical conversation error: {e}") + self.update_status(f"โŒ Consultation error: {str(e)}") + self.reset_ui() + + def submit_response(self): + """Submit patient response to clinical AI""" + response_text = self.response_entry.get("1.0", tk.END).strip() + + if not response_text: + messagebox.showwarning("Warning", "Please enter a response") + return + + if not self.session_id: + messagebox.showerror("Error", "No active session") + return + + try: + self.update_status("๐Ÿค” AI is thinking...") + self.submit_button.config(state=tk.DISABLED) + + payload = { + "session_id": self.session_id, + "patient_response": response_text + } + + # Continue conversation + response = requests.post( + f"{UNIFIED_SERVER}/continue_conversation", + json=payload, + timeout=60 + ) + + if response.status_code == 200: + result = response.json() + + if result.get('conversation_complete'): + # Show final diagnosis/summary + summary = result.get('summary', 'Consultation completed.') + self.update_status("โœ… Consultation completed") + self.update_instruction(f"๐Ÿ“‹ Final Assessment:\n\n{summary}") + self.update_progress("Complete!") + self.hide_response_input() + + # Auto-reset after showing results + self.root.after(15000, self.reset_conversation) # Reset after 15 seconds + + else: + # Show next question + next_question = result.get('question', 'Thank you for your response.') + iteration = result.get('iteration', self.current_iteration + 1) + self.current_iteration = iteration + + self.update_instruction(f"Doctor: {next_question}") + self.update_progress(f"Question {iteration}/6") + self.response_entry.delete("1.0", tk.END) + + self.update_status("๐Ÿ’ฌ Ready for your response") + self.submit_button.config(state=tk.NORMAL) + + else: + error_msg = response.json().get('error', 'Unknown error') + self.update_status(f"โŒ Failed to process response: {error_msg}") + self.submit_button.config(state=tk.NORMAL) + + except requests.exceptions.Timeout: + self.update_status("โฑ๏ธ Response timed out. AI may be busy.") + self.submit_button.config(state=tk.NORMAL) + except Exception as e: + print(f"โŒ Response submission error: {e}") + self.update_status(f"โŒ Error: {str(e)}") + self.submit_button.config(state=tk.NORMAL) + + def show_response_input(self): + """Show response input widgets""" + self.response_label.pack(pady=(20, 5)) + text_frame = self.response_entry.master + text_frame.pack(fill=tk.X, pady=5) + self.submit_button.pack(pady=10) + + # Focus on text input + self.response_entry.focus_set() + + def hide_response_input(self): + """Hide response input widgets""" + self.response_label.pack_forget() + text_frame = self.response_entry.master + text_frame.pack_forget() + self.submit_button.pack_forget() + + def reset_conversation(self): + """Reset to start a new conversation""" + self.session_id = None + self.current_iteration = 0 + self.conversation_history = [] + self.scanning = False + self.reset_ui(delay=0) + + def reset_ui(self, delay=3): + """Reset UI to initial state""" + def reset(): + if delay > 0: + time.sleep(delay) + self.scanning = False + self.root.after(0, self._reset_ui_main_thread) + + if delay > 0: + threading.Thread(target=reset, daemon=True).start() + else: + self._reset_ui_main_thread() + + def _reset_ui_main_thread(self): + """Reset UI components (must run in main thread)""" + self.start_button.config(state=tk.NORMAL) + self.reset_button.config(state=tk.DISABLED) + self.submit_button.config(state=tk.NORMAL) + self.update_status("Ready to scan") + self.update_instruction("Touch 'Start Scan' to begin analyzing your skin condition") + self.update_progress("") + self.hide_response_input() + + def update_status(self, text): + """Update status label (thread-safe)""" + self.root.after(0, lambda: self.status_label.config(text=text)) + + def update_instruction(self, text): + """Update instruction label (thread-safe)""" + self.root.after(0, lambda: self.instruction_label.config(text=text)) + + def update_progress(self, text): + """Update progress label (thread-safe)""" + self.root.after(0, lambda: self.progress_label.config(text=text)) + + def exit_app(self): + """Clean up and exit""" + try: + if self.camera and CAMERA_TYPE == "picamera2": + if self.camera.started: + self.camera.stop() + self.camera.close() + elif self.camera and CAMERA_TYPE == "picamera": + self.camera.close() + except: + pass + + self.root.quit() + self.root.destroy() + +def main(): + # Check server connectivity + print("๐Ÿ” Checking server connectivity...") + + try: + response = requests.get(f"{UNIFIED_SERVER}/", timeout=5) + print("โœ… Unified medical server connected") + except: + print(f"โŒ Cannot connect to unified server at {UNIFIED_SERVER}") + print("Please update UNIFIED_SERVER variable with correct IP") + + # Create and run app + root = tk.Tk() + app = MedicalScannerApp(root) + + # Bind escape key to exit (for testing without touchscreen) + root.bind('', lambda e: app.exit_app()) + + # Bind Enter key to submit response when text widget has focus + def on_enter(event): + if app.submit_button['state'] == tk.NORMAL and app.response_entry.winfo_viewable(): + app.submit_response() + return 'break' + + root.bind('', on_enter) # Ctrl+Enter to submit + + try: + root.mainloop() + except KeyboardInterrupt: + app.exit_app() + +if __name__ == "__main__": + # Configuration check + if "YOUR_GPU_SERVER_IP" in UNIFIED_SERVER: + print("โš ๏ธ WARNING: Please update UNIFIED_SERVER with your actual GPU server IP!") + print(f"Current setting: {UNIFIED_SERVER}") + print("Example: UNIFIED_SERVER = 'http://192.168.1.100:7860'") + print("") + + print("๐Ÿš€ Medical Skin Scanner - Raspberry Pi Client") + print(f"๐Ÿ“ก Server: {UNIFIED_SERVER}") + print(f"๐Ÿ“ท Camera: {CAMERA_TYPE if CAMERA_TYPE else 'Not available'}") + print("๐Ÿ–ฅ๏ธ Starting touchscreen interface...") + print("") + + main() \ No newline at end of file diff --git a/computer-vision/fullpicode/test1/gpu.py b/computer-vision/fullpicode/test1/gpu.py new file mode 100644 index 00000000..3dac87f6 --- /dev/null +++ b/computer-vision/fullpicode/test1/gpu.py @@ -0,0 +1,362 @@ +#!/usr/bin/env python3 +""" +Fixed GPU Medical Analysis Server - Complete Revamp +Forces GPU usage and fixes all LLaVA issues +""" + +import os +import json +import time +import torch +import cv2 +import numpy as np +from datetime import datetime +from PIL import Image +import io + +# Flask for API +from flask import Flask, request, jsonify + +# YOLO for object detection +from ultralytics import YOLO + +# Use BLIP instead of problematic LLaVA +from transformers import BlipProcessor, BlipForConditionalGeneration + +# ============================================================================ +# CONFIGURATION +# ============================================================================ +YOLO_MODEL_PATH = "yolov8n.pt" +VISION_MODEL_NAME = "Salesforce/blip-image-captioning-large" # More reliable than LLaVA + +HF_TOKEN = "_" + +HAND_CONFIDENCE_THRESHOLD = 0.2 +TARGET_CLASSES = ['person', 'hand'] + +app = Flask(__name__) + +# Global model storage +models = { + 'yolo': None, + 'vision_processor': None, + 'vision_model': None +} + +# ============================================================================ +# FORCE GPU SETUP +# ============================================================================ +def setup_gpu(): + """Force GPU setup and verification""" + print("๐Ÿ”ง Setting up GPU environment...") + + # Force CUDA if available + if torch.cuda.is_available(): + torch.cuda.empty_cache() + device = torch.device("cuda:0") + print(f"โœ… GPU setup complete: {torch.cuda.get_device_name(0)}") + print(f"๐Ÿ’พ GPU Memory: {torch.cuda.get_device_properties(0).total_memory / 1e9:.1f} GB") + return device + else: + print("โŒ No GPU detected - check nvidia-smi") + return torch.device("cpu") + +# ============================================================================ +# MODEL LOADING WITH FORCED GPU +# ============================================================================ +def load_yolo_model(): + try: + print("๐Ÿ”„ Loading YOLO model...") + model = YOLO(YOLO_MODEL_PATH) + print("โœ… YOLO model loaded") + return model + except Exception as e: + print(f"โŒ YOLO load failed: {e}") + return None + +def load_vision_models(): + try: + print("๐Ÿ”„ Loading BLIP vision model (GPU-optimized)...") + + # Force GPU device + device = setup_gpu() + + processor = BlipProcessor.from_pretrained(VISION_MODEL_NAME) + + if device.type == "cuda": + model = BlipForConditionalGeneration.from_pretrained( + VISION_MODEL_NAME, + torch_dtype=torch.float16, + device_map="auto" + ).to(device) + print(f"โœ… BLIP loaded on GPU: {model.device}") + else: + model = BlipForConditionalGeneration.from_pretrained( + VISION_MODEL_NAME, + torch_dtype=torch.float32 + ).to(device) + print(f"โš ๏ธ BLIP loaded on CPU: {model.device}") + + return processor, model + + except Exception as e: + print(f"โŒ BLIP load failed: {e}") + import traceback + traceback.print_exc() + return None, None + +def initialize_models(): + print("๐Ÿš€ Initializing all models with GPU optimization...") + + models['yolo'] = load_yolo_model() + models['vision_processor'], models['vision_model'] = load_vision_models() + + success = all([ + models['yolo'] is not None, + models['vision_processor'] is not None, + models['vision_model'] is not None + ]) + + if success: + print("โœ… All models loaded successfully!") + print(f"๐ŸŽฎ GPU Usage: {torch.cuda.is_available()}") + if torch.cuda.is_available(): + print(f"๐Ÿ“Š GPU Memory Used: {torch.cuda.memory_allocated(0) / 1e9:.1f} GB") + else: + print("โŒ Some models failed to load") + + return success + +# ============================================================================ +# ANALYSIS FUNCTIONS +# ============================================================================ +def detect_objects(image_bytes): + try: + nparr = np.frombuffer(image_bytes, np.uint8) + image = cv2.imdecode(nparr, cv2.IMREAD_COLOR) + results = models['yolo'](image, conf=HAND_CONFIDENCE_THRESHOLD, verbose=False) + + detections = [] + for result in results: + if result.boxes is not None: + for box in result.boxes: + class_id = int(box.cls) + confidence = float(box.conf) + class_name = models['yolo'].names[class_id] + if class_name in TARGET_CLASSES: + detections.append({ + 'class': class_name, + 'confidence': confidence + }) + + detections.sort(key=lambda x: x['confidence'], reverse=True) + return detections[0] if detections else None + + except Exception as e: + print(f"โŒ Detection error: {e}") + return None + +def analyze_with_blip(image_bytes): + """Use BLIP for reliable image analysis""" + try: + print("๐Ÿ”„ Starting BLIP image analysis...") + start_time = time.time() + + # Load image + image = Image.open(io.BytesIO(image_bytes)).convert("RGB") + print(f"๐Ÿ“ท Image loaded: {image.size}") + + # Create medical prompt + prompt = "a medical photograph showing" + + # Process with BLIP + device = models['vision_model'].device + inputs = models['vision_processor'](image, prompt, return_tensors="pt").to(device) + + print("๐Ÿ”„ BLIP generating analysis...") + with torch.no_grad(): + out = models['vision_model'].generate( + **inputs, + max_length=100, + num_beams=5, + early_stopping=True + ) + + # Decode result + analysis = models['vision_processor'].decode(out[0], skip_special_tokens=True) + + # Remove the prompt from the output + if prompt in analysis: + analysis = analysis.replace(prompt, "").strip() + + total_time = time.time() - start_time + + print(f"โœ… BLIP analysis completed in {total_time:.1f} seconds") + print("=" * 60) + print("๐Ÿ‘๏ธ BLIP MEDICAL IMAGE ANALYSIS:") + print("=" * 60) + print(analysis) + print("=" * 60) + + return analysis, total_time + + except Exception as e: + print(f"โŒ BLIP analysis error: {e}") + import traceback + traceback.print_exc() + return None, 0 + +def run_analysis_pipeline(image_bytes): + """Complete analysis pipeline""" + + # Step 1: Object Detection + print("๐Ÿ” Step 1: YOLO Object Detection") + start_total = time.time() + + detection = detect_objects(image_bytes) + if not detection: + return {"error": "No relevant objects detected"} + + print(f"โœ… Detected: {detection['class']} ({detection['confidence']:.1%})") + + # Step 2: BLIP Analysis + print("๐Ÿ” Step 2: BLIP Image Analysis") + analysis, analysis_time = analyze_with_blip(image_bytes) + + if not analysis: + return {"error": "Image analysis failed"} + + total_time = time.time() - start_total + + print(f"โœ… PIPELINE COMPLETE in {total_time:.1f} seconds!") + + return { + "success": True, + "detection": detection, + "image_analysis": analysis, + "analysis_time": analysis_time, + "total_time": total_time, + "gpu_used": torch.cuda.is_available(), + "timestamp": datetime.now().isoformat() + } + +# ============================================================================ +# FLASK ROUTES +# ============================================================================ +@app.route("/") +def home(): + gpu_info = {} + if torch.cuda.is_available(): + gpu_info = { + "gpu_name": torch.cuda.get_device_name(0), + "gpu_memory_gb": torch.cuda.get_device_properties(0).total_memory / 1e9, + "gpu_memory_used_gb": torch.cuda.memory_allocated(0) / 1e9 + } + + return jsonify({ + "status": "online", + "message": "Fixed GPU Medical Analysis Server", + "models_loaded": { + "yolo": models['yolo'] is not None, + "vision": models['vision_model'] is not None + }, + "gpu_available": torch.cuda.is_available(), + "gpu_info": gpu_info + }) + +@app.route("/quick_detect", methods=["POST"]) +def quick_detect(): + """Quick YOLO detection only""" + try: + if 'image' not in request.files: + return jsonify({"error": "No image provided"}), 400 + + image_file = request.files['image'] + image_bytes = image_file.read() + + detection = detect_objects(image_bytes) + + if detection: + return jsonify({ + "detected": True, + "detection": detection + }) + else: + return jsonify({ + "detected": False + }) + + except Exception as e: + print(f"โŒ Quick detection error: {e}") + return jsonify({"error": str(e)}), 500 + +@app.route("/analyze_frame", methods=["POST"]) +def analyze_frame(): + """Complete analysis pipeline""" + try: + if 'image' not in request.files: + return jsonify({"error": "No image provided"}), 400 + + image_file = request.files['image'] + image_bytes = image_file.read() + + print(f"๐Ÿš€ Starting analysis pipeline for {len(image_bytes)} byte image") + + result = run_analysis_pipeline(image_bytes) + + if "error" in result: + return jsonify(result), 500 + + return jsonify(result) + + except Exception as e: + print(f"โŒ Pipeline error: {e}") + import traceback + traceback.print_exc() + return jsonify({"error": str(e)}), 500 + +@app.route("/health", methods=["GET"]) +def health_check(): + return jsonify({ + "status": "healthy", + "models_loaded": { + "yolo": models['yolo'] is not None, + "vision": models['vision_model'] is not None + }, + "gpu_available": torch.cuda.is_available(), + "timestamp": datetime.now().isoformat() + }) + +# ============================================================================ +# MAIN EXECUTION +# ============================================================================ +def main(): + print("๐Ÿฅ Fixed GPU Medical Analysis Server") + print("=" * 50) + print("โšก Using BLIP for reliable image analysis") + print("๐Ÿš€ Optimized for GPU performance") + + # Check GPU immediately + if torch.cuda.is_available(): + print(f"โœ… GPU Detected: {torch.cuda.get_device_name(0)}") + print(f"๐Ÿ’พ GPU Memory: {torch.cuda.get_device_properties(0).total_memory / 1e9:.1f} GB") + else: + print("โš ๏ธ No GPU detected - will use CPU") + print("๐Ÿ’ก Check: nvidia-smi") + + if not initialize_models(): + print("โŒ Failed to initialize models - exiting") + return + + print("\n๐ŸŒ Starting Flask server...") + print("๐Ÿ“‹ Endpoints:") + print(" GET / - Server status with GPU info") + print(" POST /quick_detect - Fast YOLO detection") + print(" POST /analyze_frame - Complete analysis pipeline") + print(" GET /health - Health check") + print("\n๐Ÿš€ Server ready!") + + app.run(host="0.0.0.0", port=7860, debug=False, threaded=True) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/computer-vision/fullpicode/test1/pi.py b/computer-vision/fullpicode/test1/pi.py new file mode 100644 index 00000000..134ec049 --- /dev/null +++ b/computer-vision/fullpicode/test1/pi.py @@ -0,0 +1,280 @@ +#!/usr/bin/env python3 +""" +Simple Pi Client - Terminal Medical Scanner +Just handles camera and displays AI doctor conversation +""" + +import io +import requests +import time +from datetime import datetime +from picamera2 import Picamera2 + +# Configuration +GPU_SERVER_URL = "http://192.222.58.119:7860" + +class SimpleMedicalClient: + def __init__(self): + self.camera = None + self.init_camera() + + def init_camera(self): + """Initialize camera""" + try: + self.camera = Picamera2() + config = self.camera.create_preview_configuration(main={"size": (640, 480)}) + self.camera.configure(config) + print("โœ… Camera ready") + except Exception as e: + print(f"โŒ Camera failed: {e}") + self.camera = None + + def check_server(self): + """Check server status""" + try: + response = requests.get(f"{GPU_SERVER_URL}/", timeout=10) + if response.status_code == 200: + print("โœ… Medical AI server online") + return True + return False + except Exception as e: + print(f"โŒ Server check failed: {e}") + return False + + def capture_medical_image(self): + """Capture high-resolution medical image""" + try: + if not self.camera: + print("โŒ No camera available") + return None + + # Start camera + if not self.camera.started: + self.camera.start() + time.sleep(2) + + # Switch to high-res for medical analysis + self.camera.stop() + config = self.camera.create_still_configuration(main={"size": (1024, 768)}) + self.camera.configure(config) + self.camera.start() + time.sleep(1) + + print("\n๐Ÿ“ธ MEDICAL IMAGE CAPTURE") + print("=" * 40) + print("๐Ÿ“ท Position the area of concern clearly in view") + print("๐Ÿ’ก Ensure good lighting for best analysis") + + # Countdown + for i in range(3, 0, -1): + print(f"๐Ÿ“ธ Capturing in {i}...") + time.sleep(1) + + print("๐Ÿ“ธ *CLICK* - Image captured!") + + stream = io.BytesIO() + self.camera.capture_file(stream, format='jpeg') + stream.seek(0) + image_bytes = stream.getvalue() + + print(f"โœ… Image ready: {len(image_bytes)} bytes") + + # Return to preview mode + self.camera.stop() + config = self.camera.create_preview_configuration(main={"size": (640, 480)}) + self.camera.configure(config) + self.camera.start() + + return image_bytes + + except Exception as e: + print(f"โŒ Image capture failed: {e}") + return None + + def analyze_medical_image(self, image_bytes): + """Send image for complete medical analysis""" + try: + print("\n๐Ÿš€ MEDICAL ANALYSIS IN PROGRESS") + print("=" * 50) + print("๐Ÿ” Running object detection...") + print("๐Ÿ‘๏ธ Analyzing image with BLIP AI...") + print("๐Ÿฉบ Processing with medical AI...") + print("โณ This may take 30-90 seconds...") + + files = {'image': ('medical_scan.jpg', image_bytes, 'image/jpeg')} + + response = requests.post( + f"{GPU_SERVER_URL}/analyze_frame", + files=files, + timeout=120 # 2 minutes + ) + + if response.status_code == 200: + result = response.json() + + if result.get('success'): + print("\nโœ… MEDICAL ANALYSIS COMPLETE!") + return result + else: + print(f"โŒ Analysis failed: {result.get('error')}") + return None + else: + print(f"โŒ Server error: {response.status_code}") + return None + + except requests.exceptions.Timeout: + print("โฑ๏ธ Analysis timed out - please try again") + return None + except Exception as e: + print(f"โŒ Analysis error: {e}") + return None + + def display_analysis_results(self, result): + """Display complete medical analysis results""" + print("\n" + "=" * 70) + print("๐Ÿฅ COMPLETE MEDICAL ANALYSIS RESULTS") + print("=" * 70) + + # Detection info + detection = result['detection'] + print(f"\n๐ŸŽฏ OBJECT DETECTION:") + print(f" Detected: {detection['class']} ({detection['confidence']:.1%} confidence)") + + # BLIP analysis + print(f"\n๐Ÿ‘๏ธ VISUAL ANALYSIS (BLIP AI):") + print(" " + "โ”€" * 60) + print(f" {result['blip_analysis']}") + print(" " + "โ”€" * 60) + + # Combined patient info + print(f"\n๐Ÿ“‹ PATIENT INFORMATION:") + print(" " + "โ”€" * 60) + print(f" {result['combined_info']}") + print(" " + "โ”€" * 60) + + # Clinical summary + print(f"\n๐Ÿฉบ CLINICAL SUMMARY:") + print(" " + "โ”€" * 60) + print(f" {result['clinical_summary']}") + print(" " + "โ”€" * 60) + + # Diagnostic analysis + print(f"\n๐Ÿ”ฌ DIAGNOSTIC ANALYSIS:") + print(" " + "โ”€" * 60) + print(f" {result['diagnostic_analysis']}") + print(" " + "โ”€" * 60) + + # Final medical question + print(f"\nโ“ AI DOCTOR QUESTION:") + print("=" * 70) + print(f"๐Ÿฉบ {result['medical_question']}") + print("=" * 70) + + # Processing times + print(f"\nโฑ๏ธ ANALYSIS TIMING:") + print(f" BLIP Processing: {result['blip_time']:.1f} seconds") + print(f" Total Pipeline: {result.get('total_time', 0):.1f} seconds") + + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f" Completed: {timestamp}") + + print("\nโš•๏ธ MEDICAL DISCLAIMER:") + print(" This AI analysis is for informational purposes only.") + print(" Please consult a healthcare provider for proper medical evaluation.") + + def run_medical_scan(self): + """Run complete medical scan workflow""" + print("\n๐Ÿฅ MEDICAL SCAN WORKFLOW") + print("=" * 50) + + if not self.camera: + print("โŒ Camera not available") + return False + + if not self.check_server(): + print("โŒ Medical AI server not accessible") + return False + + try: + # Step 1: Capture medical image + image_bytes = self.capture_medical_image() + if not image_bytes: + print("โŒ Failed to capture image") + return False + + # Step 2: Analyze with complete AI pipeline + result = self.analyze_medical_image(image_bytes) + if not result: + print("โŒ Medical analysis failed") + return False + + # Step 3: Display comprehensive results + self.display_analysis_results(result) + + print("\nโœ… Medical scan completed successfully!") + return True + + except KeyboardInterrupt: + print("\n๐Ÿ›‘ Medical scan interrupted") + return False + except Exception as e: + print(f"โŒ Scan error: {e}") + return False + finally: + self.cleanup() + + def cleanup(self): + """Cleanup resources""" + try: + if self.camera: + if self.camera.started: + self.camera.stop() + self.camera.close() + print("๐Ÿ“ท Camera closed") + except: + pass + +def main(): + print("๐Ÿฅ AI Medical Scanner") + print("=" * 30) + print("๐Ÿ“‹ Complete Pipeline: Camera โ†’ BLIP โ†’ AI Doctor") + print(f"๐ŸŒ Server: {GPU_SERVER_URL}") + print("โš•๏ธ For informational purposes only") + print("\nPress Ctrl+C to exit\n") + + client = SimpleMedicalClient() + + try: + while True: + print("\n" + "=" * 50) + input("๐Ÿ“‹ Press Enter to start medical scan (or Ctrl+C to exit)...") + + success = client.run_medical_scan() + + if success: + print("\n๐ŸŽ‰ Scan completed!") + else: + print("\nโŒ Scan failed!") + + # Ask if user wants another scan + while True: + try: + again = input("\n๐Ÿ”„ Perform another scan? (y/n): ").strip().lower() + if again in ['y', 'yes']: + break + elif again in ['n', 'no']: + print("๐Ÿ‘‹ Thank you for using AI Medical Scanner!") + return + else: + print("Please enter 'y' or 'n'") + except KeyboardInterrupt: + print("\n๐Ÿ‘‹ Goodbye!") + return + + except KeyboardInterrupt: + print("\n๐Ÿ‘‹ Goodbye!") + finally: + client.cleanup() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/computer-vision/fullpicode/test1/pulse.py b/computer-vision/fullpicode/test1/pulse.py new file mode 100644 index 00000000..ae6019a1 --- /dev/null +++ b/computer-vision/fullpicode/test1/pulse.py @@ -0,0 +1,74 @@ +import time +import qwiic_max3010x +import numpy as np +import matplotlib.pyplot as plt +from scipy.signal import find_peaks + +# Initialize sensor +sensor = qwiic_max3010x.QwiicMax3010x() + +if not sensor.connected: + print("MAX30105 not connected. Check wiring.") + exit() + +sensor.setup() # Default configuration: 69 samples/sec, LED power medium +sensor.set_led_pulse_amplitude_red(0x1F) # Moderate LED brightness +sensor.set_led_pulse_amplitude_green(0) # Green off (not used for pulse) + +print("Starting pulse measurement. Press Ctrl+C to stop.") + +ir_values = [] +time_stamps = [] + +try: + start_time = time.time() + + while True: + if sensor.available(): + ir = sensor.get_ir() # IR light absorption reflects pulse + timestamp = time.time() - start_time + + ir_values.append(ir) + time_stamps.append(timestamp) + + print(f"IR: {ir}") + + # Keep only last 5 seconds of data for real-time plot + if len(ir_values) > 500: + ir_values = ir_values[-500:] + time_stamps = time_stamps[-500:] + + time.sleep(0.01) # 100 Hz sampling approx. + +except KeyboardInterrupt: + print("Stopped. Processing data...") + + # Basic peak detection for heart rate (offline example) + ir_array = np.array(ir_values) + time_array = np.array(time_stamps) + + # Normalize + ir_array = ir_array - np.mean(ir_array) + + # Find peaks + peaks, _ = find_peaks(ir_array, distance=30) # Adjust 'distance' for real data + + # Calculate heart rate + peak_times = time_array[peaks] + intervals = np.diff(peak_times) + + if len(intervals) > 0: + avg_interval = np.mean(intervals) + bpm = 60 / avg_interval + print(f"Estimated Heart Rate: {bpm:.1f} BPM") + else: + print("No peaks detected.") + + # Plot IR signal and peaks + plt.plot(time_array, ir_array, label='IR Signal') + plt.plot(peak_times, ir_array[peaks], 'rx', label='Peaks') + plt.xlabel('Time (s)') + plt.ylabel('IR Value') + plt.title('Pulse Signal') + plt.legend() + plt.show() diff --git a/computer-vision/llava/images/test.png b/computer-vision/llava/images/test.png new file mode 100644 index 00000000..38a0dc1a Binary files /dev/null and b/computer-vision/llava/images/test.png differ diff --git a/computer-vision/llava/pi_code.py b/computer-vision/llava/pi_code.py new file mode 100644 index 00000000..972d3122 --- /dev/null +++ b/computer-vision/llava/pi_code.py @@ -0,0 +1,35 @@ +import io +import requests +import picamera + +# Backend URL (Update this to your serverโ€™s IP or domain) +BACKEND_URL = "http://104.171.203.124:7860/analyze_image" + +# Optional: Custom prompt +payload = { + 'prompt': 'Describe this medical image clearly.' +} + +# Capture image to memory +stream = io.BytesIO() +with picamera.PiCamera() as camera: + camera.resolution = (640, 480) + camera.start_preview() + camera.capture(stream, format='jpeg') + camera.close() + +# Prepare stream for upload +stream.seek(0) +files = {'image': ('image.jpg', stream, 'image/jpeg')} + +# Send POST request +print("๐Ÿ“ค Sending image to backend...") +response = requests.post(BACKEND_URL, files=files, data=payload) + +# Output result +if response.status_code == 200: + print("โœ… Response from backend:") + print(response.json()) +else: + print(f"โŒ Error: {response.status_code}") + print(response.text) diff --git a/computer-vision/llava/server.py b/computer-vision/llava/server.py new file mode 100644 index 00000000..abdb4f8f --- /dev/null +++ b/computer-vision/llava/server.py @@ -0,0 +1,582 @@ +import json +import os +from datetime import datetime +import torch +from transformers import AutoTokenizer, AutoModelForCausalLM, LlavaNextProcessor, LlavaNextForConditionalGeneration +from PIL import Image +import requests +from flask import Flask, request, jsonify +import io + +# ============================================================================ +# CONFIGURATION - REPLACE WITH YOUR ACTUAL PATHS AND TOKENS +# ============================================================================ +CLINICAL_MODEL_PATH = "CodCodingCode/llama-3.1-8b-clinical-V1.4" +VISION_MODEL_NAME = "llava-hf/llava-v1.6-mistral-7b-hf" + +# Get HuggingFace token ONLY for the clinical model (if needed) +HF_TOKEN = os.getenv("HUGGINGFACE_TOKEN") + +app = Flask(__name__) + +# ============================================================================ +# VISION ANALYSIS SETUP (LLaVA is open source - no token needed) +# ============================================================================ +print("๐Ÿ”„ Loading LLaVA-NeXT vision model...") +try: + # LLaVA is public - load without token + vision_processor = LlavaNextProcessor.from_pretrained(VISION_MODEL_NAME) + vision_model = LlavaNextForConditionalGeneration.from_pretrained( + VISION_MODEL_NAME, + torch_dtype=torch.float16, + device_map="auto" + ) + + vision_processor.tokenizer.padding_side = "left" + print("โœ… Vision model loaded successfully!") + +except Exception as e: + print(f"โŒ Error loading vision model: {e}") + print("๐Ÿ’ก This shouldn't happen - LLaVA is open source!") + print(f" Error details: {e}") + exit(1) + +GENERAL_MEDICAL_IMAGE_PROMPT = """ +You are an AI-powered medical assistant. Given the attached photo of a patient's body part, object, or condition, analyze the image carefully and provide a detailed, medically relevant description that can assist a healthcare provider in understanding what is depicted. + +Present your findings as a well-written, concise paragraph, not as a list or bullet points. + +In your description, include the following where applicable: +- What is shown in the image (e.g. skin, joint, limb, face, wound, swelling, discoloration, visible abnormality, etc.) +- The body location if identifiable (e.g. arm, leg, hand, face, torso) +- Shape or structure (e.g. round, irregular, swollen, asymmetric, deformity present) +- Size estimate (approximate in cm, mm, or general terms) +- Color(s) and tone (e.g. normal skin tone, redness, bruising, pallor, abnormal colors) +- Texture or surface characteristics (e.g. smooth, rough, scaly, cracked, ulcerated) +- Edges or borders (e.g. well-defined, irregular, merging with surrounding tissue) +- Number or extent of findings (e.g. single abnormality, multiple areas, diffuse involvement) +- Symmetry or asymmetry +- Signs of inflammation (e.g. redness, heat, swelling) +- Signs of injury (e.g. cuts, bruises, bleeding, abrasions) +- Signs of infection (e.g. pus, discharge, crusting) +- Presence of medical devices, bandages, tattoos, or distinguishing marks +- Any other visible relevant information + +If the image content is unclear or ambiguous, clearly state the uncertainty but still describe any observable details to the best of your ability. +""" + +def analyze_image_with_llava(image, prompt=GENERAL_MEDICAL_IMAGE_PROMPT, max_tokens=500): + """Analyze medical image using LLaVA-NeXT""" + conversation = [ + { + "role": "user", + "content": [ + {"type": "image"}, + {"type": "text", "text": prompt}, + ], + }, + ] + text_prompt = vision_processor.apply_chat_template(conversation, add_generation_prompt=True) + inputs = vision_processor(image, text_prompt, return_tensors="pt").to(vision_model.device) + + with torch.no_grad(): + output = vision_model.generate( + **inputs, + max_new_tokens=max_tokens, + do_sample=False, + temperature=0.1 + ) + + response = vision_processor.decode(output[0], skip_special_tokens=True) + assistant_start = response.find("[/INST]") + len("[/INST]") + if assistant_start > len("[/INST]") - 1: + return response[assistant_start:].strip() + else: + return response + +# ============================================================================ +# CLINICAL AI SETUP +# ============================================================================ +class LocalModelInference: + def __init__(self, model_path, device="auto"): + self.model_path = model_path + self.device = device + + print(f"๐Ÿ”„ Loading clinical model from {model_path}...") + + # Load tokenizer + try: + if HF_TOKEN: + self.tokenizer = AutoTokenizer.from_pretrained( + model_path, token=HF_TOKEN, use_fast=True, trust_remote_code=True + ) + else: + self.tokenizer = AutoTokenizer.from_pretrained( + model_path, use_fast=True, trust_remote_code=True + ) + except Exception as e: + print(f"โŒ Error loading tokenizer: {e}") + print("๐Ÿ’ก Try setting HuggingFace token: export HUGGINGFACE_TOKEN=your_token") + raise + + # Ensure pad token is set + if self.tokenizer.pad_token is None: + self.tokenizer.pad_token = self.tokenizer.eos_token + print("๐Ÿ”ง Set pad_token to eos_token") + + # Load model + try: + if HF_TOKEN: + self.model = AutoModelForCausalLM.from_pretrained( + model_path, + token=HF_TOKEN, + device_map=device, + torch_dtype=torch.bfloat16, + trust_remote_code=True, + low_cpu_mem_usage=True, + ) + else: + self.model = AutoModelForCausalLM.from_pretrained( + model_path, + device_map=device, + torch_dtype=torch.bfloat16, + trust_remote_code=True, + low_cpu_mem_usage=True, + ) + except Exception as e: + print(f"โŒ Error loading model: {e}") + print("๐Ÿ’ก Try setting HuggingFace token: export HUGGINGFACE_TOKEN=your_token") + raise + + # Set model to evaluation mode + self.model.eval() + + print(f"โœ… Clinical model loaded successfully!") + print(f"๐Ÿ“Š Model device map: {getattr(self.model, 'hf_device_map', 'N/A')}") + + def generate(self, prompt, max_new_tokens=800): + """Generate text using the local clinical model""" + try: + # Tokenize input + inputs = self.tokenizer( + prompt, + return_tensors="pt", + truncation=True, + max_length=2048, + padding=False, + ) + + # Move inputs to the same device as model + if hasattr(self.model, "device"): + device = self.model.device + else: + device = next(self.model.parameters()).device + + inputs = {k: v.to(device) for k, v in inputs.items()} + + # Generate + with torch.no_grad(): + outputs = self.model.generate( + **inputs, + max_new_tokens=max_new_tokens, + temperature=0.2, + do_sample=True, + repetition_penalty=1.1, + no_repeat_ngram_size=3, + pad_token_id=self.tokenizer.pad_token_id, + eos_token_id=self.tokenizer.eos_token_id, + early_stopping=False, + ) + + # Decode the generated text + generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True) + + # Remove the input prompt from the generated text + response = generated_text[len(prompt):].strip() + + return response + + except Exception as e: + print(f"โŒ Generation failed: {e}") + raise + + def clear_cache(self): + """Clear GPU cache to free memory""" + if torch.cuda.is_available(): + torch.cuda.empty_cache() + +# Initialize the clinical inference client +print("๐Ÿš€ Initializing clinical model...") +clinical_model = LocalModelInference(CLINICAL_MODEL_PATH) + +# ============================================================================ +# UTILITY FUNCTIONS +# ============================================================================ +def save_conversation_state(conversation_log, filename="clinical_conversation_log.json"): + """Save current conversation state to JSON file after each turn""" + with open(filename, "w") as f: + json.dump(conversation_log, f, indent=2) + +def extract_thinking_and_answer(raw_output): + """Extract THINKING and ANSWER sections from raw output""" + thinking = "" + answer = "" + + if "THINKING:" in raw_output and "ANSWER:" in raw_output: + thinking_start = raw_output.find("THINKING:") + len("THINKING:") + answer_start = raw_output.find("ANSWER:") + len("ANSWER:") + thinking = raw_output[thinking_start:raw_output.find("ANSWER:")].strip() + answer = raw_output[answer_start:].strip() + answer = answer.replace("STOP HERE", "").replace("Do not add", "").strip() + elif "THINKING:" in raw_output: + thinking_start = raw_output.find("THINKING:") + len("THINKING:") + thinking = raw_output[thinking_start:].strip() + answer = "" + elif "ANSWER:" in raw_output: + answer_start = raw_output.find("ANSWER:") + len("ANSWER:") + answer = raw_output[answer_start:].strip() + answer = answer.replace("STOP HERE", "").replace("Do not add", "").strip() + thinking = "" + else: + answer = raw_output.strip() + thinking = "" + + return thinking, answer + +def clean_doctor_question(raw_question): + """Clean and extract just the question from doctor output""" + cleaned = raw_question.strip() + cleaned = cleaned.replace("THINKING:", "").replace("ANSWER:", "") + + lines = cleaned.split("\n") + question_lines = [] + + for line in lines: + line = line.strip() + if any(skip_phrase in line.lower() for skip_phrase in [ + "this question is different", "therefore, asking about", + "specifically, knowing what", "additionally, vital sign", + "whereas now i want", "the vignette indicates" + ]): + continue + if "?" in line and len(line) > 10: + question_lines.append(line) + + if question_lines: + return question_lines[0].strip() + + # Fallback logic + sentences = cleaned.split(".") + for sentence in sentences: + sentence = sentence.strip() + if "?" in sentence and len(sentence) > 10: + return sentence.strip() + ("?" if not sentence.endswith("?") else "") + + return "Can you tell me more about your symptoms?" + +def test_models(): + """Test both models""" + try: + print("๐Ÿ”„ Testing clinical model...") + result = clinical_model.generate("Hello, this is a test.", max_new_tokens=20) + print(f"โœ… Clinical model working! Test: {result[:50]}...") + + print("๐Ÿ”„ Testing vision model...") + # Create a simple test image + test_image = Image.new('RGB', (100, 100), color='red') + vision_result = analyze_image_with_llava(test_image, "What do you see?", max_tokens=50) + print(f"โœ… Vision model working! Test: {vision_result[:50]}...") + + return True + except Exception as e: + print(f"โŒ Model testing failed: {e}") + return False + +# ============================================================================ +# FLASK ROUTES +# ============================================================================ +@app.route("/") +def home(): + return "โœ… Combined Medical Vision + AI Doctor Server is Running!" + +@app.route("/analyze_image_url", methods=["POST"]) +def analyze_image_url(): + """Analyze image from URL""" + data = request.json + image_url = data.get("image_url") + prompt = data.get("prompt", GENERAL_MEDICAL_IMAGE_PROMPT) + + if not image_url: + return jsonify({"error": "Missing image_url"}), 400 + + try: + print(f"๐Ÿ“ฅ Fetching image from: {image_url}") + response = requests.get(image_url, stream=True, timeout=10) + response.raise_for_status() + image = Image.open(io.BytesIO(response.content)).convert("RGB") + except Exception as e: + return jsonify({"error": f"Failed to fetch/open image: {e}"}), 400 + + try: + print("๐Ÿค– Running vision analysis...") + result = analyze_image_with_llava(image, prompt) + return jsonify({"result": result}) + except Exception as e: + return jsonify({"error": f"Analysis error: {e}"}), 500 + +@app.route("/analyze_image_upload", methods=["POST"]) +def analyze_image_upload(): + """Analyze uploaded image""" + if 'image' not in request.files: + return jsonify({"error": "Missing image file"}), 400 + + image_file = request.files['image'] + prompt = request.form.get("prompt", GENERAL_MEDICAL_IMAGE_PROMPT) + + try: + image = Image.open(image_file).convert("RGB") + except Exception as e: + return jsonify({"error": f"Failed to open image: {e}"}), 400 + + try: + print("๐Ÿค– Running vision analysis...") + result = analyze_image_with_llava(image, prompt) + return jsonify({"result": result}) + except Exception as e: + return jsonify({"error": f"Analysis error: {e}"}), 500 + +@app.route("/start_diagnosis", methods=["POST"]) +def start_diagnosis(): + """Start diagnostic conversation with optional image analysis""" + data = request.json + initial_complaint = data.get("initial_complaint", "") + image_analysis = data.get("image_analysis", "") + + # Combine initial complaint with image analysis + full_patient_info = initial_complaint + if image_analysis: + full_patient_info += f"\n\nAdditional visual findings from medical image: {image_analysis}" + + # Initialize conversation + conversation_log = { + "timestamp": datetime.now().isoformat(), + "initial_complaint": initial_complaint, + "image_analysis": image_analysis, + "full_patient_info": full_patient_info, + "iterations": [] + } + + return jsonify({ + "message": "Diagnostic conversation started", + "full_patient_info": full_patient_info, + "conversation_id": conversation_log["timestamp"] + }) + +# ============================================================================ +# MAIN DIAGNOSTIC CONVERSATION FUNCTION +# ============================================================================ +def run_conversation_with_vision(): + """Run the complete diagnostic conversation with optional vision analysis""" + + # Test models first + if not test_models(): + print("โŒ Please check your model paths and tokens") + return + + print("\n" + "=" * 60) + print("๐Ÿฅ Medical Vision + AI Doctor System") + print("=" * 60) + + # Step 1: Optional image analysis + image_analysis = "" + use_image = input("๐Ÿ“ธ Do you have a medical image to analyze? (y/n): ").strip().lower() + + if use_image == 'y': + image_path = input("๐Ÿ“ Enter the path to your medical image: ").strip() + try: + image = Image.open(image_path).convert("RGB") + print("๐Ÿ”„ Analyzing medical image...") + image_analysis = analyze_image_with_llava(image, GENERAL_MEDICAL_IMAGE_PROMPT) + print(f"\n๐Ÿ“‹ Medical Image Analysis:") + print(f"๐Ÿ” {image_analysis}") + print("\n" + "-" * 50) + except Exception as e: + print(f"โŒ Error analyzing image: {e}") + image_analysis = "" + + # Step 2: Get initial complaint + print("\n๐Ÿ‘ฉโ€โš•๏ธ Doctor: What brings you in today?") + initial_complaint = input("๐Ÿ‘ค Patient: ").strip() + + # Step 3: Combine initial complaint with image analysis + full_patient_info = initial_complaint + if image_analysis: + full_patient_info += f"\n\nAdditional visual findings from medical image: {image_analysis}" + print(f"\n๐Ÿ“ Combined patient information:") + print(f"๐Ÿ’ฌ Initial complaint: {initial_complaint}") + print(f"๐Ÿ” Image analysis: {image_analysis}") + + # Initialize conversation log + conversation_log = { + "timestamp": datetime.now().isoformat(), + "initial_complaint": initial_complaint, + "image_analysis": image_analysis, + "full_patient_info": full_patient_info, + "iterations": [] + } + + # Step 4: Start diagnostic conversation + convo = [] + prev_questions = [] + convo.append("Doctor: What brings you in today?") + convo.append(f"Patient: {full_patient_info}") + + patient_response = full_patient_info + prev_vignette = "" + + for i in range(6): # Maximum 6 iterations + print(f"\n{'='*20} Iteration {i+1} {'='*20}") + + iteration_data = { + "iteration": i + 1, + "clinical_summary": {}, + "diagnostic_reasoning": {}, + "question_generation": {}, + "treatment_plan": {}, + "patient_response": patient_response, + "conversation_history": convo.copy(), + } + + # Clear GPU cache periodically + if i % 2 == 0: + clinical_model.clear_cache() + + # 1. Clinical Summary Generation + input_text = f"""Instruction: You are a clinical summarizer. Given a transcript of a doctorโ€“patient dialogue, extract a structured clinical vignette summarizing the key symptoms, relevant history, and any diagnostic clues. +Input: {patient_response} Previous Vignette: {prev_vignette} +Output: THINKING: """ + + print("๐Ÿ”„ Generating clinical summary...") + raw_output = clinical_model.generate(input_text, max_new_tokens=1000) + thinking, answer = extract_thinking_and_answer(raw_output) + + iteration_data["clinical_summary"] = { + "input": input_text, + "raw_output": raw_output, + "thinking": thinking, + "answer": answer, + } + + output = answer if answer else raw_output + print("๐Ÿ“‹ Clinical Summary:") + print(output[:300] + "..." if len(output) > 300 else output) + + # 2. Diagnostic Reasoning + input_text2 = f"""Instruction: You are a diagnostic reasoning model (Early Stage). Based on the patient vignette and early-stage observations, generate a list of plausible diagnoses with reasoning. Focus on broad differentials, considering common and uncommon conditions +Input: {output} +Output: THINKING:""" + + print("๐Ÿ”„ Generating diagnostic reasoning...") + raw_output2 = clinical_model.generate(input_text2, max_new_tokens=800) + thinking2, answer2 = extract_thinking_and_answer(raw_output2) + + iteration_data["diagnostic_reasoning"] = { + "input": input_text2, + "raw_output": raw_output2, + "thinking": thinking2, + "answer": answer2, + } + + output2 = answer2 if answer2 else raw_output2 + print("๐Ÿฉบ Diagnostic Reasoning:") + print(output2[:300] + "..." if len(output2) > 300 else output2) + + # Check if we've reached the final iteration + if i == 5: + print("\n๐Ÿ Reached maximum iterations. Final diagnosis:") + print(output2) + break + + # 3. Question Generation + input_text3 = f"""Instruction: You are a questioning agent (Early Stage). Your task is to propose highly relevant early-stage questions that can open the differential diagnosis widely. Use epidemiology, demographics, and vague presenting symptoms as guides. +Input: VIGNETTE: {output} DIAGNOSIS: {output2} PREVIOUS Questions: {prev_questions} Conversation History: {convo} +Output: THINKING:""" + + print("๐Ÿ”„ Generating next question...") + raw_output3 = clinical_model.generate(input_text3, max_new_tokens=1000) + thinking3, answer3 = extract_thinking_and_answer(raw_output3) + + iteration_data["question_generation"] = { + "input": input_text3, + "raw_output": raw_output3, + "thinking": thinking3, + "answer": answer3, + } + + # Clean the doctor output to get just the question + doctor_output = clean_doctor_question(answer3 if answer3 else raw_output3) + print(f"โ“ Next Question: {doctor_output}") + + # 4. Treatment Plan Generation + input_text4 = f"""Instruction: You are a board-certified clinician. Based on the provided diagnosis and patient vignette, propose a realistic, evidence-based treatment plan suitable for initiation by a primary care physician or psychiatrist. +Input: Diagnosis: {output2} Vignette: {output} +Output: THINKING:""" + + print("๐Ÿ”„ Generating treatment plan...") + raw_output4 = clinical_model.generate(input_text4, max_new_tokens=1000) + thinking4, answer4 = extract_thinking_and_answer(raw_output4) + + iteration_data["treatment_plan"] = { + "input": input_text4, + "raw_output": raw_output4, + "thinking": thinking4, + "answer": answer4, + } + + print("๐Ÿ’Š Treatment Plan:") + treatment_output = answer4 if answer4 else raw_output4 + print(treatment_output[:300] + "..." if len(treatment_output) > 300 else treatment_output) + + # Save conversation state + conversation_log["iterations"].append(iteration_data) + save_conversation_state(conversation_log) + + # Get patient response + print(f"\n๐Ÿฉบ Doctor: {doctor_output}") + convo.append(f"Doctor: {doctor_output}") + prev_questions.append(doctor_output) + + patient_response = input("๐Ÿ‘ค Your response as the patient: ").strip() + if not patient_response: + patient_response = "I'm not sure what else to tell you." + + convo.append(f"Patient: {patient_response}") + prev_vignette = output + + # Final cleanup and save + save_conversation_state(conversation_log) + print(f"\n๐Ÿ“„ Complete conversation log saved to clinical_conversation_log.json") + print("โœ… Diagnostic conversation completed!") + clinical_model.clear_cache() + +# ============================================================================ +# MAIN EXECUTION +# ============================================================================ +if __name__ == "__main__": + print("๐Ÿš€ Combined Medical Vision + AI Doctor System") + print(f"๐Ÿ” Vision Model: {VISION_MODEL_NAME}") + print(f"๐Ÿฉบ Clinical Model: {CLINICAL_MODEL_PATH}") + print(f"๐ŸŽฎ GPU Available: {torch.cuda.is_available()}") + + if torch.cuda.is_available(): + print(f"๐Ÿ”ง GPU Count: {torch.cuda.device_count()}") + print(f"๐Ÿ’พ GPU Memory: {torch.cuda.get_device_properties(0).total_memory / 1e9:.1f} GB") + + mode = input("\n๐ŸŽฏ Choose mode:\n1. Interactive Console (c)\n2. Web API Server (w)\nEnter choice: ").strip().lower() + + if mode == 'w': + print("\n๐ŸŒ Starting web server...") + app.run(host="0.0.0.0", port=7860, debug=True) + else: + print("\n๐Ÿ’ฌ Starting interactive console...") + run_conversation_with_vision() \ No newline at end of file diff --git a/computer-vision/position_detection/main.py b/computer-vision/position_detection/main.py new file mode 100644 index 00000000..0cb88db0 --- /dev/null +++ b/computer-vision/position_detection/main.py @@ -0,0 +1,78 @@ +import cv2 +import torch + +# Load pre-trained YOLOv5 model (person detection) +model = torch.hub.load('ultralytics/yolov5', 'yolov5s') + +# Store last detection results +last_detections = [] +frame_count = 0 + +def run_detection(frame): + """Run object detection and return results""" + results = model(frame) + persons = results.xyxy[0] # bounding boxes for detected persons + + detections = [] + for *box, conf, cls in persons: + if int(cls) == 0 and conf > 0.5: # class 0 = person in COCO + x1, y1, x2, y2 = map(int, box) + box_height = y2 - y1 + detections.append((x1, y1, x2, y2, box_height)) + + return detections + +def draw_results(frame, detections, min_box_height=400, max_box_height=800): + """Draw bounding boxes and distance feedback using cached detection results""" + for x1, y1, x2, y2, box_height in detections: + # Draw box + cv2.rectangle(frame, (x1, y1), (x2, y2), (0,255,0), 2) + + # Check distance based on box size + if box_height > max_box_height: + cv2.putText(frame, "TOO CLOSE - Step Back", (x1, y1-10), + cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,0,255), 2) + elif box_height < min_box_height: + cv2.putText(frame, "TOO FAR - Step Closer", (x1, y1-10), + cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,0,255), 2) + else: + cv2.putText(frame, "GOOD DISTANCE", (x1, y1-10), + cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,255,0), 2) + + return frame +# Initialize camera +cap = cv2.VideoCapture(0) + +print("Press 'q' to quit") + +while True: + ret, frame = cap.read() + if not ret: + print("Can't read camera") + break + + # Flip frame so it's like a mirror + frame = cv2.flip(frame, 1) + + frame_count += 1 + + # Only run detection every 10 frames (or whatever number works for you) + if frame_count % 10 == 0: + last_detections = run_detection(frame.copy()) + + # Always draw results using the most recent detection data + frame = draw_results(frame, last_detections) + + # Show the frame + cv2.imshow('Distance Check', frame) + + # Quit if 'q' is pressed + if cv2.waitKey(1) & 0xFF == ord('q'): + break + +# Clean up +cap.release() +cv2.destroyAllWindows() + + +# scp computer-vision/fullpicode/pi_controlled_code.py jerryhuang@192.168.50.249:/home/jerryhuang/ \ No newline at end of file diff --git a/rPPG/main.py b/rPPG/main.py new file mode 100644 index 00000000..05306c79 --- /dev/null +++ b/rPPG/main.py @@ -0,0 +1,105 @@ +import cv2 +import numpy as np +import matplotlib.pyplot as plt +import time +import os +from ultralytics import YOLO +import urllib.request +from pyVHR.signals import Signal +from pyVHR.vhr import HRVideo + +# ============================ +# Download YOLO Face Model if not exists +# ============================ +model_url = "https://github.com/lindevs/yolov8-face/releases/latest/download/yolov8n-face-lindevs.pt" +model_path = "yolov8n-face-lindevs.pt" + +if not os.path.isfile(model_path): + print("Downloading YOLO face model...") + urllib.request.urlretrieve(model_url, model_path) + print("Download complete.") + +# ============================ +# Initialize Video Capture and YOLO +# ============================ +cap = cv2.VideoCapture(0) + +if not cap.isOpened(): + print("Error: Could not open webcam.") + exit() + +model = YOLO(model_path) + +# ============================ +# Parameters +# ============================ +frame_rate = 30 +buffer_size = frame_rate * 10 # 10-second window +signal_data = [] + +plt.ion() +fig, ax = plt.subplots() +line, = ax.plot([], [], lw=2) +ax.set_ylim(-10, 10) +ax.set_xlim(0, buffer_size) +ax.set_title('Live Respiratory Signal (pyVHR)') +ax.set_xlabel('Frames') +ax.set_ylabel('Amplitude') + +# Initialize pyVHR signal processor +vhr = HRVideo() + +# ============================ +# Main Loop +# ============================ +try: + while True: + ret, frame = cap.read() + if not ret: + break + + small_frame = cv2.resize(frame, (640, 480)) + + results = model.predict(small_frame, verbose=False) + + if results and results[0].boxes: + box = results[0].boxes.xyxy[0].cpu().numpy().astype(int) + x1, y1, x2, y2 = box + + x1, y1 = max(0, x1), max(0, y1) + x2, y2 = min(small_frame.shape[1], x2), min(small_frame.shape[0], y2) + + roi = small_frame[y1:y2, x1:x2] + + if roi.size > 0: + mean_green = np.mean(roi[:, :, 1]) + signal_data.append(mean_green) + + if len(signal_data) > buffer_size: + signal_data = signal_data[-buffer_size:] + + # Use pyVHR signal processing (placeholder for actual RR estimation) + sig = Signal(signal_data, frame_rate) + processed_signal = sig.signal - np.mean(sig.signal) + + line.set_ydata(processed_signal) + line.set_xdata(np.arange(len(processed_signal))) + ax.set_xlim(0, len(processed_signal)) + ax.set_ylim(-np.std(processed_signal)*3, np.std(processed_signal)*3) + fig.canvas.draw() + fig.canvas.flush_events() + + cv2.rectangle(small_frame, (x1, y1), (x2, y2), (0, 255, 0), 2) + + cv2.imshow('rPPG Face Detection', small_frame) + + if cv2.waitKey(1) & 0xFF == ord('q'): + break + +except KeyboardInterrupt: + pass + +cap.release() +cv2.destroyAllWindows() +plt.ioff() +plt.show() diff --git a/rPPG/yolov8n-face-lindevs.pt b/rPPG/yolov8n-face-lindevs.pt new file mode 100644 index 00000000..4f52f700 Binary files /dev/null and b/rPPG/yolov8n-face-lindevs.pt differ diff --git a/training-cv/smaller_data.py b/training-cv/smaller_data.py new file mode 100644 index 00000000..59c63be6 --- /dev/null +++ b/training-cv/smaller_data.py @@ -0,0 +1,51 @@ +import os +import shutil +import random +from pathlib import Path + +def create_1percent_dataset(source_path="facial_dataset", output_path="emotion_dataset_1percent"): + """Create 1% subset of the emotion dataset""" + + source_path = Path(source_path) + output_path = Path(output_path) + + # Create output structure + for split in ['train', 'validation']: + for emotion in ['angry', 'disgust', 'fear', 'happy', 'neutral', 'sad', 'surprise']: + (output_path / split / emotion).mkdir(parents=True, exist_ok=True) + + print("Creating 1% subset...") + + for split in ['train', 'validation']: + for emotion in ['angry', 'disgust', 'fear', 'happy', 'neutral', 'sad', 'surprise']: + source_dir = source_path / split / emotion + output_dir = output_path / split / emotion + + if source_dir.exists(): + # Get all images with multiple extensions + images = [] + for ext in ['*.jpg', '*.jpeg', '*.png', '*.JPG', '*.JPEG', '*.PNG', '*.bmp', '*.BMP']: + images.extend(list(source_dir.glob(ext))) + + print(f"Found {len(images)} images in {source_dir}") + + if len(images) > 0: + # Take 1% + subset_count = max(1, int(len(images) * 0.01)) + selected = random.sample(images, min(subset_count, len(images))) + + # Copy images + for img in selected: + shutil.copy2(img, output_dir / img.name) + + print(f"{split}/{emotion}: {len(selected)}/{len(images)} images copied") + else: + print(f"โŒ No images found in {source_dir}") + else: + print(f"โŒ Directory doesn't exist: {source_dir}") + + print(f"\nโœ… 1% subset created at: {output_path}") + +if __name__ == "__main__": + random.seed(42) + create_1percent_dataset() \ No newline at end of file diff --git a/training-cv/train.py b/training-cv/train.py new file mode 100644 index 00000000..4f1f8f02 --- /dev/null +++ b/training-cv/train.py @@ -0,0 +1,176 @@ +from ultralytics import YOLO +import cv2 +import os +import yaml +from pathlib import Path +import shutil + +def convert_to_yolo_format(source_path="emotion_dataset_1percent"): + """Convert emotion dataset to YOLO format""" + + source_path = Path(source_path) + yolo_path = Path(f"{source_path.name}_yolo") + + # Create YOLO structure + for split in ['train', 'val']: + (yolo_path / split / 'images').mkdir(parents=True, exist_ok=True) + (yolo_path / split / 'labels').mkdir(parents=True, exist_ok=True) + + emotion_classes = ['angry', 'disgust', 'fear', 'happy', 'neutral', 'sad', 'surprise'] + + print("Converting to YOLO format...") + + for split in ['train', 'validation']: + yolo_split = 'train' if split == 'train' else 'val' + + for class_idx, emotion in enumerate(emotion_classes): + emotion_dir = source_path / split / emotion + + if emotion_dir.exists(): + for img_file in emotion_dir.glob("*.jpg"): + # Copy image + new_name = f"{emotion}_{img_file.stem}.jpg" + shutil.copy2(img_file, yolo_path / yolo_split / 'images' / new_name) + + # Create label (whole image is the face/emotion) + label_content = f"{class_idx} 0.5 0.5 1.0 1.0\n" + label_file = yolo_path / yolo_split / 'labels' / f"{emotion}_{img_file.stem}.txt" + + with open(label_file, 'w') as f: + f.write(label_content) + + print(f"Processed {emotion}: {len(list(emotion_dir.glob('*.jpg')))} images") + + # Create dataset.yaml + dataset_config = { + 'path': str(yolo_path.absolute()), + 'train': 'train/images', + 'val': 'val/images', + 'nc': len(emotion_classes), + 'names': emotion_classes + } + + with open(yolo_path / 'dataset.yaml', 'w') as f: + yaml.dump(dataset_config, f) + + print(f"โœ… YOLO dataset created: {yolo_path}") + return yolo_path / 'dataset.yaml' + +def train_emotion_model(): + """Train YOLOv8 for emotion detection""" + + # Convert dataset + dataset_yaml = convert_to_yolo_format() + + # Load YOLOv8 model (nano for speed) + model = YOLO('yolov8n.pt') + + print("Starting training...") + print("โฑ๏ธ Estimated time: 20-40 minutes on CPU, 5-15 minutes on GPU") + + # Train model + results = model.train( + data=str(dataset_yaml), + epochs=50, + imgsz=416, # Smaller for speed + batch=8, # Small batch for CPU + name='emotion_1percent', + patience=10, + device='cpu', # Change to 0 for GPU + + # Optimization + cache=False, # Don't cache on CPU + save=True, + + # Augmentation + flipud=0.0, # No vertical flip for faces + fliplr=0.5, # Horizontal flip only + mosaic=0.5, + + # Learning + lr0=0.01, + momentum=0.937, + weight_decay=0.0005, + ) + + model_path = "runs/detect/emotion_1percent/weights/best.pt" + print(f"โœ… Training complete! Model saved: {model_path}") + return model_path + +def test_trained_model(model_path="runs/detect/emotion_1percent/weights/best.pt"): + """Test the trained model on webcam""" + + model = YOLO(model_path) + cap = cv2.VideoCapture(0) + + emotion_colors = { + 0: (0, 0, 255), # angry - red + 1: (0, 255, 0), # disgust - green + 2: (128, 0, 128), # fear - purple + 3: (0, 255, 255), # happy - yellow + 4: (128, 128, 128), # neutral - gray + 5: (255, 0, 0), # sad - blue + 6: (255, 165, 0) # surprise - orange + } + + emotion_names = ['angry', 'disgust', 'fear', 'happy', 'neutral', 'sad', 'surprise'] + + print("Testing model on webcam. Press 'q' to quit.") + + while True: + ret, frame = cap.read() + if not ret: + break + + # Run detection + results = model(frame, conf=0.3) + + # Draw results + for result in results: + boxes = result.boxes + if boxes is not None: + for box in boxes: + x1, y1, x2, y2 = map(int, box.xyxy[0]) + conf = box.conf[0] + cls = int(box.cls[0]) + + color = emotion_colors.get(cls, (255, 255, 255)) + emotion = emotion_names[cls] + + # Draw box and label + cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2) + label = f"{emotion}: {conf:.2f}" + cv2.putText(frame, label, (x1, y1-10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2) + + cv2.imshow('Emotion Detection Test', frame) + + if cv2.waitKey(1) & 0xFF == ord('q'): + break + + cap.release() + cv2.destroyAllWindows() + +if __name__ == "__main__": + print("๐Ÿš€ YOLOv8 Emotion Detection Training") + print("=====================================") + + # Check if 1% dataset exists + if not Path("emotion_dataset_1percent").exists(): + print("โŒ Run the subset creation script first!") + print(" python create_subset.py") + exit() + + # Step 1: Train model + print("\n1. Training YOLOv8 model...") + model_path = train_emotion_model() + + # Step 2: Test model + print("\n2. Testing trained model...") + test_trained_model(model_path) + + print("\n๐ŸŽ‰ Training complete!") + print(f"๐Ÿ“ Model saved at: {model_path}") + print("\nNext steps:") + print("- If results look good, train on full dataset") + print("- Adjust hyperparameters if needed") + print("- Export model for deployment") \ No newline at end of file diff --git a/training-cv/yolov8n.pt b/training-cv/yolov8n.pt new file mode 100644 index 00000000..0db4ca4b Binary files /dev/null and b/training-cv/yolov8n.pt differ