Upload folder using huggingface_hub

.gitattributes

@@ -1,35 +0,0 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
-*.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,6 @@
+models/ 
+data_storage/ 
+elephmind.db 
+.env 
+venv/ 
+__pycache__/

Dockerfile

@@ -1,38 +1,38 @@
-# Hugging Face Spaces Docker Configuration
-FROM python:3.10-slim
-
-# Create non-root user (required by HuggingFace)
-RUN useradd -m -u 1000 user
-ENV HOME=/home/user
-ENV PATH="/home/user/.local/bin:$PATH"
-
-WORKDIR /app
-
-# Install system dependencies as root
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    libgl1 \
-    libglib2.0-0 \
-    libsm6 \
-    libxext6 \
-    libxrender1 \
-    && rm -rf /var/lib/apt/lists/*
-
-# Create directories as root BEFORE switching user
-RUN mkdir -p /app/storage/uploads /app/storage/processed && \
-    chown -R user:user /app
-
-# Switch to non-root user
-USER user
-
-# Copy requirements and install
-COPY --chown=user requirements.txt .
-RUN pip install --no-cache-dir --user -r requirements.txt
-
-# Copy the rest of the application
-COPY --chown=user . /app
-
-# Expose port 7860 (required by Hugging Face Spaces)
-EXPOSE 7860
-
-# Run the application
-CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]
+# Hugging Face Spaces Docker Configuration
+FROM python:3.10-slim
+
+# Create non-root user (required by HuggingFace)
+RUN useradd -m -u 1000 user
+ENV HOME=/home/user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+# Install system dependencies as root
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libgl1 \
+    libglib2.0-0 \
+    libsm6 \
+    libxext6 \
+    libxrender1 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Create directories as root BEFORE switching user
+RUN mkdir -p /app/storage/uploads /app/storage/processed && \
+    chown -R user:user /app
+
+# Switch to non-root user
+USER user
+
+# Copy requirements and install
+COPY --chown=user requirements.txt .
+RUN pip install --no-cache-dir --user -r requirements.txt
+
+# Copy the rest of the application
+COPY --chown=user . /app
+
+# Expose port 7860 (required by Hugging Face Spaces)
+EXPOSE 7860
+
+# Run the application
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,16 +1,21 @@
----
-title: Elephmind Api
-emoji: 🏃
-colorFrom: indigo
-colorTo: red
-sdk: docker
-pinned: false
-license: gemma
-short_description: ' IA imagerie medical '
----
-
-# ElephMind API v5.1
-
-Backend for medical image analysis.
-
-**Last Update**: 2026-01-31 17:00 - Fixed KeyError issues and disabled faulty CTR morphology engine.
+---
+title: ElephMind Medical AI
+emoji: 🏥
+colorFrom: green
+colorTo: green
+sdk: docker
+app_port: 7860
+pinned: true
+license: apache-2.0
+---
+
+# ElephMind - Diagnostic IA Médical
+
+Application d'aide au diagnostic médical basée sur l'intelligence artificielle.
+
+## Fonctionnalités
+- Analyse de radiographies thoraciques
+- Analyse dermatologique
+- Analyse histologique
+- Analyse ophtalmologique
+- Analyse orthopédique

database.py

@@ -1,513 +1,464 @@
-import sqlite3
-import os
-import logging
-from typing import Optional, List, Dict, Any
-from enum import Enum
-
-class JobStatus(str, Enum):
-    PENDING = "pending"
-    PROCESSING = "processing"
-    COMPLETED = "completed"
-    FAILED = "failed"
-
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-# HUGGING FACE PERSISTENCE FIX: Use /data if available
-if os.path.exists('/data'):
-    DB_NAME = '/data/elephmind.db'
-    logging.info("Using PERSISTENT storage at /data/elephmind.db")
-else:
-    DB_NAME = os.path.join(BASE_DIR, "elephmind.db")
-    logging.info(f"Using LOCAL storage at {DB_NAME}")
-
-def get_db_connection():
-    conn = sqlite3.connect(DB_NAME)
-    conn.row_factory = sqlite3.Row
-    return conn
-
-def init_db():
-    conn = get_db_connection()
-    c = conn.cursor()
-    
-    # Create Users Table
-    c.execute('''
-        CREATE TABLE IF NOT EXISTS users (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            username TEXT UNIQUE NOT NULL,
-            hashed_password TEXT NOT NULL,
-            email TEXT,
-            security_question TEXT NOT NULL,
-            security_answer TEXT NOT NULL,
-            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-        )
-    ''')
-    
-    # Create Feedback Table
-    c.execute('''
-        CREATE TABLE IF NOT EXISTS feedback (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            username TEXT,
-            rating INTEGER,
-            comment TEXT,
-            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-        )
-    ''')
-    
-    # Create Audit Log Table (RGPD Compliance)
-    c.execute('''
-        CREATE TABLE IF NOT EXISTS audit_log (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            username TEXT,
-            action TEXT NOT NULL,
-            resource TEXT,
-            ip_address TEXT,
-            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-        )
-    ''')
-
-    # --- MIGRATIONS ---
-    # Ensure security columns exist (backward compatibility)
-    try:
-        c.execute("ALTER TABLE users ADD COLUMN security_question TEXT DEFAULT 'Question?'")
-    except sqlite3.OperationalError:
-        pass # Column exists
-
-    try:
-        c.execute("ALTER TABLE users ADD COLUMN security_answer TEXT DEFAULT 'answer'")
-    except sqlite3.OperationalError:
-        pass # Column exists
-    # ------------------
-
-    # Create Patients Table
-    c.execute('''
-        CREATE TABLE IF NOT EXISTS patients (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            patient_id TEXT UNIQUE NOT NULL, -- e.g. PAT-2026-1234
-            owner_username TEXT NOT NULL,
-            first_name TEXT,
-            last_name TEXT,
-            birth_date TEXT,
-            photo TEXT, -- Stores base64 or URL
-            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-            FOREIGN KEY(owner_username) REFERENCES users(username)
-        )
-    ''')
-
-    # Create Jobs Table (PERSISTENCE)
-    c.execute('''
-        CREATE TABLE IF NOT EXISTS jobs (
-            id TEXT PRIMARY KEY,
-            status TEXT NOT NULL,
-            result TEXT, -- JSON serialized
-            error TEXT,
-            created_at REAL,
-            storage_path TEXT,
-            username TEXT,
-            file_type TEXT,
-            FOREIGN KEY(username) REFERENCES users(username)
-        )
-    ''')
-    
-    conn.commit()
-    conn.close()
-    logging.info(f"Database {DB_NAME} initialized successfully.")
-
-# --- User Operations ---
-
-def create_user(user: Dict[str, Any]) -> bool:
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        c.execute('''
-            INSERT INTO users (username, hashed_password, email, security_question, security_answer)
-            VALUES (?, ?, ?, ?, ?)
-        ''', (
-            user['username'],
-            user['hashed_password'],
-            user.get('email', ''),
-            user['security_question'],
-            user['security_answer']
-        ))
-        conn.commit()
-        return True
-    except sqlite3.IntegrityError:
-        return False
-    except Exception as e:
-        logging.error(f"Error creating user: {e}")
-        return False
-    finally:
-        conn.close()
-
-def get_user_by_username(username: str) -> Optional[Dict[str, Any]]:
-    conn = get_db_connection()
-    c = conn.cursor()
-    c.execute('SELECT * FROM users WHERE username = ?', (username,))
-    row = c.fetchone()
-    conn.close()
-    if row:
-        return dict(row)
-    return None
-
-def update_password(username: str, new_hashed_password: str) -> bool:
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        c.execute('UPDATE users SET hashed_password = ? WHERE username = ?', (new_hashed_password, username))
-        conn.commit()
-        conn.close()
-        return True
-    except Exception as e:
-        logging.error(f"Error updating password: {e}")
-        return False
-
-# --- Feedback Operations ---
-
-def add_feedback(username: str, rating: int, comment: str):
-    conn = get_db_connection()
-    c = conn.cursor()
-    c.execute('INSERT INTO feedback (username, rating, comment) VALUES (?, ?, ?)', (username, rating, comment))
-    conn.commit()
-    conn.close()
-
-# --- Audit Log Operations (RGPD Compliance) ---
-
-def log_audit(username: str, action: str, resource: str = None, ip_address: str = None):
-    """Log user actions for RGPD compliance and security auditing."""
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        c.execute(
-            'INSERT INTO audit_log (username, action, resource, ip_address) VALUES (?, ?, ?, ?)',
-            (username, action, resource, ip_address)
-        )
-        conn.commit()
-        conn.close()
-    except Exception as e:
-        logging.error(f"Error logging audit: {e}")
-
-def get_user_audit_log(username: str, limit: int = 100) -> List[Dict[str, Any]]:
-    """Get audit log for a specific user."""
-    conn = get_db_connection()
-    c = conn.cursor()
-    c.execute(
-        'SELECT * FROM audit_log WHERE username = ? ORDER BY created_at DESC LIMIT ?',
-        (username, limit)
-    )
-    rows = c.fetchall()
-    conn.close()
-    return [dict(row) for row in rows]
-
-# --- Analysis Registry (REAL DATA ONLY) ---
-
-def init_analysis_registry():
-    """Create the analysis_registry table if it doesn't exist."""
-    conn = get_db_connection()
-    c = conn.cursor()
-    c.execute('''
-        CREATE TABLE IF NOT EXISTS analysis_registry (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            username TEXT NOT NULL,
-            domain TEXT NOT NULL,
-            top_diagnosis TEXT,
-            confidence REAL,
-            priority TEXT,
-            computation_time_ms INTEGER,
-            file_type TEXT,
-            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-        )
-    ''')
-    conn.commit()
-    conn.close()
-
-def log_analysis(
-    username: str,
-    domain: str,
-    top_diagnosis: str,
-    confidence: float,
-    priority: str,
-    computation_time_ms: int,
-    file_type: str
-) -> bool:
-    """Log a real analysis to the registry. NO FAKE DATA."""
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        c.execute('''
-            INSERT INTO analysis_registry 
-            (username, domain, top_diagnosis, confidence, priority, computation_time_ms, file_type)
-            VALUES (?, ?, ?, ?, ?, ?, ?)
-        ''', (username, domain, top_diagnosis, confidence, priority, computation_time_ms, file_type))
-        conn.commit()
-        conn.close()
-        return True
-    except Exception as e:
-        logging.error(f"Error logging analysis: {e}")
-        return False
-
-def get_dashboard_stats(username: str) -> Dict[str, Any]:
-    """Get real dashboard statistics for a user. Returns zeros if no data."""
-    conn = get_db_connection()
-    c = conn.cursor()
-    
-    # Total count
-    c.execute('SELECT COUNT(*) FROM analysis_registry WHERE username = ?', (username,))
-    total = c.fetchone()[0]
-    
-    # By domain
-    c.execute('''
-        SELECT domain, COUNT(*) as count 
-        FROM analysis_registry 
-        WHERE username = ? 
-        GROUP BY domain
-    ''', (username,))
-    by_domain = {row['domain']: row['count'] for row in c.fetchall()}
-    
-    # By priority
-    c.execute('''
-        SELECT priority, COUNT(*) as count 
-        FROM analysis_registry 
-        WHERE username = ? 
-        GROUP BY priority
-    ''', (username,))
-    by_priority = {row['priority']: row['count'] for row in c.fetchall()}
-    
-    # Average computation time
-    c.execute('''
-        SELECT AVG(computation_time_ms) 
-        FROM analysis_registry 
-        WHERE username = ?
-    ''', (username,))
-    avg_time = c.fetchone()[0] or 0
-    
-    conn.close()
-    
-    return {
-        "total_analyses": total,
-        "by_domain": by_domain,
-        "by_priority": by_priority,
-        "avg_computation_time_ms": round(avg_time, 0)
-    }
-
-def get_recent_analyses(username: str, limit: int = 10) -> List[Dict[str, Any]]:
-    """Get recent real analyses for a user. Returns empty list if none."""
-    conn = get_db_connection()
-    c = conn.cursor()
-    c.execute('''
-        SELECT id, domain, top_diagnosis, confidence, priority, computation_time_ms, file_type, created_at
-        FROM analysis_registry 
-        WHERE username = ? 
-        ORDER BY created_at DESC 
-        LIMIT ?
-    ''', (username, limit))
-    rows = c.fetchall()
-    conn.close()
-    return [dict(row) for row in rows]
-
-# --- Patient Operations (New for Migration) ---
-
-def create_patient(
-    owner_username: str,
-    patient_id: str,
-    first_name: str,
-    last_name: str,
-    birth_date: str,
-    photo: str
-) -> Optional[int]:
-    """Create a new patient record."""
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        c.execute('''
-            INSERT INTO patients (owner_username, patient_id, first_name, last_name, birth_date, photo)
-            VALUES (?, ?, ?, ?, ?, ?)
-        ''', (owner_username, patient_id, first_name, last_name, birth_date, photo))
-        patient_id_db = c.lastrowid
-        conn.commit()
-        conn.close()
-        return patient_id_db
-    except Exception as e:
-        logging.error(f"Error creating patient: {e}")
-        return None
-
-def get_patients_by_user(username: str) -> List[Dict[str, Any]]:
-    """Get all patients belonging to a user."""
-    conn = get_db_connection()
-    c = conn.cursor()
-    c.execute('SELECT * FROM patients WHERE owner_username = ? ORDER BY created_at DESC', (username,))
-    rows = c.fetchall()
-    conn.close()
-    return [dict(row) for row in rows]
-
-def delete_patient(username: str, patient_db_id: int) -> bool:
-    """Delete a patient record if owned by user."""
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        c.execute('DELETE FROM patients WHERE id = ? AND owner_username = ?', (patient_db_id, username))
-        count = c.rowcount
-        conn.commit()
-        conn.close()
-        return count > 0
-    except Exception as e:
-        logging.error(f"Error deleting patient: {e}")
-        return False
-
-def update_patient(username: str, patient_db_id: int, updates: Dict[str, Any]) -> bool:
-    """Update patient fields."""
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        
-        # Build query dynamically
-        fields = []
-        values = []
-        for k, v in updates.items():
-            if k in ['first_name', 'last_name', 'birth_date', 'photo']:
-                fields.append(f"{k} = ?")
-                values.append(v)
-        
-        if not fields:
-            return False
-            
-        values.extend([patient_db_id, username])
-        query = f"UPDATE patients SET {', '.join(fields)} WHERE id = ? AND owner_username = ?"
-        
-
-        c.execute(query, values)
-        count = c.rowcount
-        conn.commit()
-        conn.close()
-        return count > 0
-    except Exception as e:
-        logging.error(f"Error updating patient: {e}")
-        return False
-
-# --- Job Operations (Persistence) ---
-
-import json
-
-def create_job(job_data: Dict[str, Any]):
-    """Create a new job record."""
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        c.execute('''
-            INSERT INTO jobs (id, status, result, error, created_at, storage_path, username, file_type)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-        ''', (
-            job_data['id'],
-            job_data.get('status', 'pending'),
-            json.dumps(job_data.get('result')) if job_data.get('result') else None,
-            job_data.get('error'),
-            job_data['created_at'],
-            job_data.get('storage_path'),
-            job_data.get('username'),
-            job_data.get('file_type')
-        ))
-        conn.commit()
-        conn.close()
-        return True
-    except Exception as e:
-        logging.error(f"Error creating job: {e}")
-        return False
-
-def get_job(job_id: str, username: Optional[str] = None) -> Optional[Dict[str, Any]]:
-    """Retrieve job by ID, optionally enforcing ownership via SQL."""
-    conn = get_db_connection()
-    c = conn.cursor()
-    
-    if username:
-        c.execute('SELECT * FROM jobs WHERE id = ? AND username = ?', (job_id, username))
-    else:
-        c.execute('SELECT * FROM jobs WHERE id = ?', (job_id,))
-        
-    row = c.fetchone()
-    conn.close()
-    
-    if row:
-        job = dict(row)
-        if job['result']:
-            try:
-                job['result'] = json.loads(job['result'])
-            except:
-                job['result'] = None
-        return job
-    return None
-
-def update_job_status(job_id: str, status: str, result: Optional[Dict] = None, error: Optional[str] = None):
-    """Update job status and result."""
-    try:
-        conn = get_db_connection()
-        c = conn.cursor()
-        
-        updates = ["status = ?"]
-        params = [status]
-        
-        if result is not None:
-            updates.append("result = ?")
-            params.append(json.dumps(result))
-            
-        if error is not None:
-            updates.append("error = ?")
-            params.append(error)
-            
-        params.append(job_id)
-        
-        query = f"UPDATE jobs SET {', '.join(updates)} WHERE id = ?"
-        c.execute(query, params)
-        conn.commit()
-        conn.close()
-        return True
-    except Exception as e:
-        logging.error(f"Error updating job: {e}")
-        return False
-
-
-
-def get_latest_job(username: str) -> Optional[Dict[str, Any]]:
-    """Retrieve the most recent job for a user."""
-    conn = get_db_connection()
-    c = conn.cursor()
-    c.execute('''
-        SELECT * FROM jobs 
-        WHERE username = ? 
-        ORDER BY created_at DESC 
-        LIMIT 1
-    ''', (username,))
-    row = c.fetchone()
-    conn.close()
-    
-    if row:
-        job = dict(row)
-        if job['result']:
-            try:
-                job['result'] = json.loads(job['result'])
-            except:
-                job['result'] = None
-        return job
-    return None
-
-def get_active_job_by_image(username: str, image_id: str) -> Optional[Dict[str, Any]]:
-    """
-    Retrieve the most recent job for a specific image and user.
-    Used for Idempotence (Strict Lifecycle).
-    """
-    conn = get_db_connection()
-    c = conn.cursor()
-    c.execute('''
-        SELECT * FROM jobs 
-        WHERE username = ? AND storage_path = ?
-        ORDER BY created_at DESC 
-        LIMIT 1
-    ''', (username, image_id))
-    row = c.fetchone()
-    conn.close()
-    
-    if row:
-        job = dict(row)
-        if job['result']:
-            try:
-                job['result'] = json.loads(job['result'])
-            except:
-                job['result'] = None
-        return job
-    return None
+import sqlite3
+import os
+import logging
+from typing import Optional, List, Dict, Any
+from enum import Enum
+
+class JobStatus(str, Enum):
+    PENDING = "pending"
+    PROCESSING = "processing"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+# HUGGING FACE PERSISTENCE FIX: Use /data if available
+if os.path.exists('/data'):
+    DB_NAME = '/data/elephmind.db'
+    logging.info("Using PERSISTENT storage at /data/elephmind.db")
+else:
+    DB_NAME = os.path.join(BASE_DIR, "elephmind.db")
+    logging.info(f"Using LOCAL storage at {DB_NAME}")
+
+def get_db_connection():
+    conn = sqlite3.connect(DB_NAME)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+def init_db():
+    conn = get_db_connection()
+    c = conn.cursor()
+    
+    # Create Users Table
+    c.execute('''
+        CREATE TABLE IF NOT EXISTS users (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT UNIQUE NOT NULL,
+            hashed_password TEXT NOT NULL,
+            email TEXT,
+            security_question TEXT NOT NULL,
+            security_answer TEXT NOT NULL,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+    ''')
+    
+    # Create Feedback Table
+    c.execute('''
+        CREATE TABLE IF NOT EXISTS feedback (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT,
+            rating INTEGER,
+            comment TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+    ''')
+    
+    # Create Audit Log Table (RGPD Compliance)
+    c.execute('''
+        CREATE TABLE IF NOT EXISTS audit_log (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT,
+            action TEXT NOT NULL,
+            resource TEXT,
+            ip_address TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+    ''')
+
+    # --- MIGRATIONS ---
+    # Ensure security columns exist (backward compatibility)
+    try:
+        c.execute("ALTER TABLE users ADD COLUMN security_question TEXT DEFAULT 'Question?'")
+    except sqlite3.OperationalError:
+        pass # Column exists
+
+    try:
+        c.execute("ALTER TABLE users ADD COLUMN security_answer TEXT DEFAULT 'answer'")
+    except sqlite3.OperationalError:
+        pass # Column exists
+    # ------------------
+
+    # Create Patients Table
+    c.execute('''
+        CREATE TABLE IF NOT EXISTS patients (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            patient_id TEXT UNIQUE NOT NULL, -- e.g. PAT-2026-1234
+            owner_username TEXT NOT NULL,
+            first_name TEXT,
+            last_name TEXT,
+            birth_date TEXT,
+            photo TEXT, -- Stores base64 or URL
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+            FOREIGN KEY(owner_username) REFERENCES users(username)
+        )
+    ''')
+
+    # Create Jobs Table (PERSISTENCE)
+    c.execute('''
+        CREATE TABLE IF NOT EXISTS jobs (
+            id TEXT PRIMARY KEY,
+            status TEXT NOT NULL,
+            result TEXT, -- JSON serialized
+            error TEXT,
+            created_at REAL,
+            storage_path TEXT,
+            username TEXT,
+            file_type TEXT,
+            FOREIGN KEY(username) REFERENCES users(username)
+        )
+    ''')
+    
+    conn.commit()
+    conn.close()
+    logging.info(f"Database {DB_NAME} initialized successfully.")
+
+# --- User Operations ---
+
+def create_user(user: Dict[str, Any]) -> bool:
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        c.execute('''
+            INSERT INTO users (username, hashed_password, email, security_question, security_answer)
+            VALUES (?, ?, ?, ?, ?)
+        ''', (
+            user['username'],
+            user['hashed_password'],
+            user.get('email', ''),
+            user['security_question'],
+            user['security_answer']
+        ))
+        conn.commit()
+        return True
+    except sqlite3.IntegrityError:
+        return False
+    except Exception as e:
+        logging.error(f"Error creating user: {e}")
+        return False
+    finally:
+        conn.close()
+
+def get_user_by_username(username: str) -> Optional[Dict[str, Any]]:
+    conn = get_db_connection()
+    c = conn.cursor()
+    c.execute('SELECT * FROM users WHERE username = ?', (username,))
+    row = c.fetchone()
+    conn.close()
+    if row:
+        return dict(row)
+    return None
+
+def update_password(username: str, new_hashed_password: str) -> bool:
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        c.execute('UPDATE users SET hashed_password = ? WHERE username = ?', (new_hashed_password, username))
+        conn.commit()
+        conn.close()
+        return True
+    except Exception as e:
+        logging.error(f"Error updating password: {e}")
+        return False
+
+# --- Feedback Operations ---
+
+def add_feedback(username: str, rating: int, comment: str):
+    conn = get_db_connection()
+    c = conn.cursor()
+    c.execute('INSERT INTO feedback (username, rating, comment) VALUES (?, ?, ?)', (username, rating, comment))
+    conn.commit()
+    conn.close()
+
+# --- Audit Log Operations (RGPD Compliance) ---
+
+def log_audit(username: str, action: str, resource: str = None, ip_address: str = None):
+    """Log user actions for RGPD compliance and security auditing."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        c.execute(
+            'INSERT INTO audit_log (username, action, resource, ip_address) VALUES (?, ?, ?, ?)',
+            (username, action, resource, ip_address)
+        )
+        conn.commit()
+        conn.close()
+    except Exception as e:
+        logging.error(f"Error logging audit: {e}")
+
+def get_user_audit_log(username: str, limit: int = 100) -> List[Dict[str, Any]]:
+    """Get audit log for a specific user."""
+    conn = get_db_connection()
+    c = conn.cursor()
+    c.execute(
+        'SELECT * FROM audit_log WHERE username = ? ORDER BY created_at DESC LIMIT ?',
+        (username, limit)
+    )
+    rows = c.fetchall()
+    conn.close()
+    return [dict(row) for row in rows]
+
+# --- Analysis Registry (REAL DATA ONLY) ---
+
+def init_analysis_registry():
+    """Create the analysis_registry table if it doesn't exist."""
+    conn = get_db_connection()
+    c = conn.cursor()
+    c.execute('''
+        CREATE TABLE IF NOT EXISTS analysis_registry (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT NOT NULL,
+            domain TEXT NOT NULL,
+            top_diagnosis TEXT,
+            confidence REAL,
+            priority TEXT,
+            computation_time_ms INTEGER,
+            file_type TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+    ''')
+    conn.commit()
+    conn.close()
+
+def log_analysis(
+    username: str,
+    domain: str,
+    top_diagnosis: str,
+    confidence: float,
+    priority: str,
+    computation_time_ms: int,
+    file_type: str
+) -> bool:
+    """Log a real analysis to the registry. NO FAKE DATA."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        c.execute('''
+            INSERT INTO analysis_registry 
+            (username, domain, top_diagnosis, confidence, priority, computation_time_ms, file_type)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+        ''', (username, domain, top_diagnosis, confidence, priority, computation_time_ms, file_type))
+        conn.commit()
+        conn.close()
+        return True
+    except Exception as e:
+        logging.error(f"Error logging analysis: {e}")
+        return False
+
+def get_dashboard_stats(username: str) -> Dict[str, Any]:
+    """Get real dashboard statistics for a user. Returns zeros if no data."""
+    conn = get_db_connection()
+    c = conn.cursor()
+    
+    # Total count
+    c.execute('SELECT COUNT(*) FROM analysis_registry WHERE username = ?', (username,))
+    total = c.fetchone()[0]
+    
+    # By domain
+    c.execute('''
+        SELECT domain, COUNT(*) as count 
+        FROM analysis_registry 
+        WHERE username = ? 
+        GROUP BY domain
+    ''', (username,))
+    by_domain = {row['domain']: row['count'] for row in c.fetchall()}
+    
+    # By priority
+    c.execute('''
+        SELECT priority, COUNT(*) as count 
+        FROM analysis_registry 
+        WHERE username = ? 
+        GROUP BY priority
+    ''', (username,))
+    by_priority = {row['priority']: row['count'] for row in c.fetchall()}
+    
+    # Average computation time
+    c.execute('''
+        SELECT AVG(computation_time_ms) 
+        FROM analysis_registry 
+        WHERE username = ?
+    ''', (username,))
+    avg_time = c.fetchone()[0] or 0
+    
+    conn.close()
+    
+    return {
+        "total_analyses": total,
+        "by_domain": by_domain,
+        "by_priority": by_priority,
+        "avg_computation_time_ms": round(avg_time, 0)
+    }
+
+def get_recent_analyses(username: str, limit: int = 10) -> List[Dict[str, Any]]:
+    """Get recent real analyses for a user. Returns empty list if none."""
+    conn = get_db_connection()
+    c = conn.cursor()
+    c.execute('''
+        SELECT id, domain, top_diagnosis, confidence, priority, computation_time_ms, file_type, created_at
+        FROM analysis_registry 
+        WHERE username = ? 
+        ORDER BY created_at DESC 
+        LIMIT ?
+    ''', (username, limit))
+    rows = c.fetchall()
+    conn.close()
+    return [dict(row) for row in rows]
+
+# --- Patient Operations (New for Migration) ---
+
+def create_patient(
+    owner_username: str,
+    patient_id: str,
+    first_name: str,
+    last_name: str,
+    birth_date: str,
+    photo: str
+) -> Optional[int]:
+    """Create a new patient record."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        c.execute('''
+            INSERT INTO patients (owner_username, patient_id, first_name, last_name, birth_date, photo)
+            VALUES (?, ?, ?, ?, ?, ?)
+        ''', (owner_username, patient_id, first_name, last_name, birth_date, photo))
+        patient_id_db = c.lastrowid
+        conn.commit()
+        conn.close()
+        return patient_id_db
+    except Exception as e:
+        logging.error(f"Error creating patient: {e}")
+        return None
+
+def get_patients_by_user(username: str) -> List[Dict[str, Any]]:
+    """Get all patients belonging to a user."""
+    conn = get_db_connection()
+    c = conn.cursor()
+    c.execute('SELECT * FROM patients WHERE owner_username = ? ORDER BY created_at DESC', (username,))
+    rows = c.fetchall()
+    conn.close()
+    return [dict(row) for row in rows]
+
+def delete_patient(username: str, patient_db_id: int) -> bool:
+    """Delete a patient record if owned by user."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        c.execute('DELETE FROM patients WHERE id = ? AND owner_username = ?', (patient_db_id, username))
+        count = c.rowcount
+        conn.commit()
+        conn.close()
+        return count > 0
+    except Exception as e:
+        logging.error(f"Error deleting patient: {e}")
+        return False
+
+def update_patient(username: str, patient_db_id: int, updates: Dict[str, Any]) -> bool:
+    """Update patient fields."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        
+        # Build query dynamically
+        fields = []
+        values = []
+        for k, v in updates.items():
+            if k in ['first_name', 'last_name', 'birth_date', 'photo']:
+                fields.append(f"{k} = ?")
+                values.append(v)
+        
+        if not fields:
+            return False
+            
+        values.extend([patient_db_id, username])
+        query = f"UPDATE patients SET {', '.join(fields)} WHERE id = ? AND owner_username = ?"
+        
+
+        c.execute(query, values)
+        count = c.rowcount
+        conn.commit()
+        conn.close()
+        return count > 0
+    except Exception as e:
+        logging.error(f"Error updating patient: {e}")
+        return False
+
+# --- Job Operations (Persistence) ---
+
+import json
+
+def create_job(job_data: Dict[str, Any]):
+    """Create a new job record."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        c.execute('''
+            INSERT INTO jobs (id, status, result, error, created_at, storage_path, username, file_type)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        ''', (
+            job_data['id'],
+            job_data.get('status', 'pending'),
+            json.dumps(job_data.get('result')) if job_data.get('result') else None,
+            job_data.get('error'),
+            job_data['created_at'],
+            job_data.get('storage_path'),
+            job_data.get('username'),
+            job_data.get('file_type')
+        ))
+        conn.commit()
+        conn.close()
+        return True
+    except Exception as e:
+        logging.error(f"Error creating job: {e}")
+        return False
+
+def get_job(job_id: str, username: Optional[str] = None) -> Optional[Dict[str, Any]]:
+    """Retrieve job by ID, optionally enforcing ownership via SQL."""
+    conn = get_db_connection()
+    c = conn.cursor()
+    
+    if username:
+        c.execute('SELECT * FROM jobs WHERE id = ? AND username = ?', (job_id, username))
+    else:
+        c.execute('SELECT * FROM jobs WHERE id = ?', (job_id,))
+        
+    row = c.fetchone()
+    conn.close()
+    
+    if row:
+        job = dict(row)
+        if job['result']:
+            try:
+                job['result'] = json.loads(job['result'])
+            except:
+                job['result'] = None
+        return job
+    return None
+
+def update_job_status(job_id: str, status: str, result: Optional[Dict] = None, error: Optional[str] = None):
+    """Update job status and result."""
+    try:
+        conn = get_db_connection()
+        c = conn.cursor()
+        
+        updates = ["status = ?"]
+        params = [status]
+        
+        if result is not None:
+            updates.append("result = ?")
+            params.append(json.dumps(result))
+            
+        if error is not None:
+            updates.append("error = ?")
+            params.append(error)
+            
+        params.append(job_id)
+        
+        query = f"UPDATE jobs SET {', '.join(updates)} WHERE id = ?"
+        c.execute(query, params)
+        conn.commit()
+        conn.close()
+        return True
+    except Exception as e:
+        logging.error(f"Error updating job: {e}")
+        return False
+
+

dicom_processor.py

@@ -1,167 +1,167 @@
-import pydicom
-import logging
-import hashlib
-from typing import Tuple, Dict, Any, Optional
-from pathlib import Path
-import os
-import io
-
-logger = logging.getLogger(__name__)
-
-# Mandatory DICOM Tags for Medical Validity
-REQUIRED_TAGS = [
-    'PatientID',
-    'StudyInstanceUID',
-    'SeriesInstanceUID',
-    'Modality',
-    'PixelSpacing', # Crucial for measurements
-]
-
-# Tags to Anonymize (PHI)
-PHI_TAGS = [
-    'PatientName',
-    'PatientBirthDate',
-    'PatientAddress',
-    'InstitutionName',
-    'ReferringPhysicianName'
-]
-
-def validate_dicom(file_bytes: bytes) -> pydicom.dataset.FileDataset:
-    """
-    Strict validation of DICOM file.
-    Raises ValueError if invalid.
-    """
-    try:
-        # 1. Parse without loading pixel data first (speed)
-        ds = pydicom.dcmread(io.BytesIO(file_bytes), stop_before_pixels=False)
-    except Exception as e:
-        raise ValueError(f"Invalid DICOM format: {str(e)}")
-
-    # 2. Check Mandatory Tags
-    missing_tags = [tag for tag in REQUIRED_TAGS if tag not in ds]
-    if missing_tags:
-        raise ValueError(f"Missing critical DICOM tags: {missing_tags}")
-
-    # 3. Check Pixel Data presence
-    if 'PixelData' not in ds:
-         raise ValueError("DICOM file has no image data (PixelData missing).")
-
-    return ds
-
-def anonymize_dicom(ds: pydicom.dataset.FileDataset) -> pydicom.dataset.FileDataset:
-    """
-    Remove PHI from dataset.
-    Returns modified dataset.
-    """
-    # Hash PatientID to keep linkable anonymous ID
-    original_id = str(ds.get('PatientID', 'Unknown'))
-    hashed_id = hashlib.sha256(original_id.encode()).hexdigest()[:16].upper()
-    
-    ds.PatientID = f"ANON-{hashed_id}"
-    
-    # Wipe other fields
-    for tag in PHI_TAGS:
-        if tag in ds:
-            if 'Date' in tag: # VR DA requires YYYYMMDD
-                ds.data_element(tag).value = "19010101"
-            else:
-                ds.data_element(tag).value = "ANONYMIZED"
-            
-    return ds
-
-def process_dicom_upload(file_bytes: bytes, username: str) -> Tuple[bytes, Dict[str, Any]]:
-    """
-    Main Gateway Function: Validate -> Anonymize -> Return Bytes & Metadata
-    """
-    # 1. Validate
-    try:
-        ds = validate_dicom(file_bytes)
-    except Exception as e:
-        logger.error(f"DICOM Validation Failed: {e}")
-        raise ValueError(f"DICOM Rejected: {e}")
-
-    # 2. Anonymize
-    ds = anonymize_dicom(ds)
-    
-    # 3. Extract safe metadata
-    metadata = {
-        "modality": ds.get("Modality", "Unknown"),
-        "body_part": ds.get("BodyPartExamined", "Unknown"),
-        "study_uid": str(ds.get("StudyInstanceUID", "")),
-        "pixel_spacing": ds.get("PixelSpacing", [1.0, 1.0]),
-        "original_filename_hint": "dicom_file.dcm"
-    }
-    
-    # 4. Convert back to bytes for storage
-    with io.BytesIO() as buffer:
-        ds.save_as(buffer)
-        safe_bytes = buffer.getvalue()
-        
-    return safe_bytes, metadata
-
-def convert_dicom_to_image(ds: pydicom.dataset.FileDataset) -> Any:
-    """
-    Convert DICOM to PIL Image / Numpy array with Medical Physics awareness.
-    1. Check RAS Orientation (Basic Validation).
-    2. Apply Hounsfield Units (CT) or Intensity Normalization (MRI/XRay).
-    3. Windowing (Lung/Bone/Soft Tissue).
-    """
-    import numpy as np
-    from PIL import Image
-    
-    try:
-        # 1. Image Geometry & Orientation Check (RAS)
-        # We enforce that slices are roughly axial/standard for now, or at least valid.
-        orientation = ds.get("ImageOrientationPatient")
-        if orientation:
-            # Check for orthogonality (basic sanity)
-            row_cosine = np.array(orientation[:3])
-            col_cosine = np.array(orientation[3:])
-            if np.abs(np.dot(row_cosine, col_cosine)) > 1e-3:
-                logger.warning("DICOM Orientation vectors are not orthogonal. Image might be skewed.")
-        
-        # 2. Extract Raw Pixels
-        pixel_array = ds.pixel_array.astype(float)
-        
-        # 3. Apply Rescale Slope/Intercept (Physics -> HU)
-        slope = getattr(ds, 'RescaleSlope', 1)
-        intercept = getattr(ds, 'RescaleIntercept', 0)
-        pixel_array = (pixel_array * slope) + intercept
-
-        # 4. Modality-Specific Normalization
-        modality = ds.get("Modality", "Unknown")
-        
-        if modality == 'CT':
-            # Hounsfield Units: Air -1000, Bone +1000
-            # Robust Min-Max scaling for visualization feeding
-            # Clip outlier HU (metal artifacts > 3000, air < -1000)
-            pixel_array = np.clip(pixel_array, -1000, 3000)
-            
-        elif modality == 'MR':
-            # MRI is relative intensity. 
-            # Simple 1-99 percentile clipping removes spikes.
-            p1, p99 = np.percentile(pixel_array, [1, 99])
-            pixel_array = np.clip(pixel_array, p1, p99)
-            
-        # 5. Normalization to 0-255 (Display Space)
-        pixel_min = np.min(pixel_array)
-        pixel_max = np.max(pixel_array)
-        
-        if pixel_max - pixel_min != 0:
-            pixel_array = ((pixel_array - pixel_min) / (pixel_max - pixel_min)) * 255.0
-        else:
-            pixel_array = np.zeros_like(pixel_array)
-            
-        pixel_array = pixel_array.astype(np.uint8)
-        
-        # 6. Color Space
-        if len(pixel_array.shape) == 2:
-            image = Image.fromarray(pixel_array).convert("RGB")
-        else:
-            image = Image.fromarray(pixel_array) 
-            
-        return image
-        
-    except Exception as e:
-        logger.error(f"DICOM Conversion Error: {e}")
-        raise ValueError(f"Could not convert DICOM to image: {e}")
+import pydicom
+import logging
+import hashlib
+from typing import Tuple, Dict, Any, Optional
+from pathlib import Path
+import os
+import io
+
+logger = logging.getLogger(__name__)
+
+# Mandatory DICOM Tags for Medical Validity
+REQUIRED_TAGS = [
+    'PatientID',
+    'StudyInstanceUID',
+    'SeriesInstanceUID',
+    'Modality',
+    'PixelSpacing', # Crucial for measurements
+]
+
+# Tags to Anonymize (PHI)
+PHI_TAGS = [
+    'PatientName',
+    'PatientBirthDate',
+    'PatientAddress',
+    'InstitutionName',
+    'ReferringPhysicianName'
+]
+
+def validate_dicom(file_bytes: bytes) -> pydicom.dataset.FileDataset:
+    """
+    Strict validation of DICOM file.
+    Raises ValueError if invalid.
+    """
+    try:
+        # 1. Parse without loading pixel data first (speed)
+        ds = pydicom.dcmread(io.BytesIO(file_bytes), stop_before_pixels=False)
+    except Exception as e:
+        raise ValueError(f"Invalid DICOM format: {str(e)}")
+
+    # 2. Check Mandatory Tags
+    missing_tags = [tag for tag in REQUIRED_TAGS if tag not in ds]
+    if missing_tags:
+        raise ValueError(f"Missing critical DICOM tags: {missing_tags}")
+
+    # 3. Check Pixel Data presence
+    if 'PixelData' not in ds:
+         raise ValueError("DICOM file has no image data (PixelData missing).")
+
+    return ds
+
+def anonymize_dicom(ds: pydicom.dataset.FileDataset) -> pydicom.dataset.FileDataset:
+    """
+    Remove PHI from dataset.
+    Returns modified dataset.
+    """
+    # Hash PatientID to keep linkable anonymous ID
+    original_id = str(ds.get('PatientID', 'Unknown'))
+    hashed_id = hashlib.sha256(original_id.encode()).hexdigest()[:16].upper()
+    
+    ds.PatientID = f"ANON-{hashed_id}"
+    
+    # Wipe other fields
+    for tag in PHI_TAGS:
+        if tag in ds:
+            if 'Date' in tag: # VR DA requires YYYYMMDD
+                ds.data_element(tag).value = "19010101"
+            else:
+                ds.data_element(tag).value = "ANONYMIZED"
+            
+    return ds
+
+def process_dicom_upload(file_bytes: bytes, username: str) -> Tuple[bytes, Dict[str, Any]]:
+    """
+    Main Gateway Function: Validate -> Anonymize -> Return Bytes & Metadata
+    """
+    # 1. Validate
+    try:
+        ds = validate_dicom(file_bytes)
+    except Exception as e:
+        logger.error(f"DICOM Validation Failed: {e}")
+        raise ValueError(f"DICOM Rejected: {e}")
+
+    # 2. Anonymize
+    ds = anonymize_dicom(ds)
+    
+    # 3. Extract safe metadata
+    metadata = {
+        "modality": ds.get("Modality", "Unknown"),
+        "body_part": ds.get("BodyPartExamined", "Unknown"),
+        "study_uid": str(ds.get("StudyInstanceUID", "")),
+        "pixel_spacing": ds.get("PixelSpacing", [1.0, 1.0]),
+        "original_filename_hint": "dicom_file.dcm"
+    }
+    
+    # 4. Convert back to bytes for storage
+    with io.BytesIO() as buffer:
+        ds.save_as(buffer)
+        safe_bytes = buffer.getvalue()
+        
+    return safe_bytes, metadata
+
+def convert_dicom_to_image(ds: pydicom.dataset.FileDataset) -> Any:
+    """
+    Convert DICOM to PIL Image / Numpy array with Medical Physics awareness.
+    1. Check RAS Orientation (Basic Validation).
+    2. Apply Hounsfield Units (CT) or Intensity Normalization (MRI/XRay).
+    3. Windowing (Lung/Bone/Soft Tissue).
+    """
+    import numpy as np
+    from PIL import Image
+    
+    try:
+        # 1. Image Geometry & Orientation Check (RAS)
+        # We enforce that slices are roughly axial/standard for now, or at least valid.
+        orientation = ds.get("ImageOrientationPatient")
+        if orientation:
+            # Check for orthogonality (basic sanity)
+            row_cosine = np.array(orientation[:3])
+            col_cosine = np.array(orientation[3:])
+            if np.abs(np.dot(row_cosine, col_cosine)) > 1e-3:
+                logger.warning("DICOM Orientation vectors are not orthogonal. Image might be skewed.")
+        
+        # 2. Extract Raw Pixels
+        pixel_array = ds.pixel_array.astype(float)
+        
+        # 3. Apply Rescale Slope/Intercept (Physics -> HU)
+        slope = getattr(ds, 'RescaleSlope', 1)
+        intercept = getattr(ds, 'RescaleIntercept', 0)
+        pixel_array = (pixel_array * slope) + intercept
+
+        # 4. Modality-Specific Normalization
+        modality = ds.get("Modality", "Unknown")
+        
+        if modality == 'CT':
+            # Hounsfield Units: Air -1000, Bone +1000
+            # Robust Min-Max scaling for visualization feeding
+            # Clip outlier HU (metal artifacts > 3000, air < -1000)
+            pixel_array = np.clip(pixel_array, -1000, 3000)
+            
+        elif modality == 'MR':
+            # MRI is relative intensity. 
+            # Simple 1-99 percentile clipping removes spikes.
+            p1, p99 = np.percentile(pixel_array, [1, 99])
+            pixel_array = np.clip(pixel_array, p1, p99)
+            
+        # 5. Normalization to 0-255 (Display Space)
+        pixel_min = np.min(pixel_array)
+        pixel_max = np.max(pixel_array)
+        
+        if pixel_max - pixel_min != 0:
+            pixel_array = ((pixel_array - pixel_min) / (pixel_max - pixel_min)) * 255.0
+        else:
+            pixel_array = np.zeros_like(pixel_array)
+            
+        pixel_array = pixel_array.astype(np.uint8)
+        
+        # 6. Color Space
+        if len(pixel_array.shape) == 2:
+            image = Image.fromarray(pixel_array).convert("RGB")
+        else:
+            image = Image.fromarray(pixel_array) 
+            
+        return image
+        
+    except Exception as e:
+        logger.error(f"DICOM Conversion Error: {e}")
+        raise ValueError(f"Could not convert DICOM to image: {e}")

encryption.py

@@ -1,71 +1,71 @@
-from cryptography.fernet import Fernet
-import os
-import sys
-import logging
-from typing import Optional
-
-# -------------------------------------------------------------------------
-# ENCRYPTION CONFIGURATION - PRODUCTION READY
-# -------------------------------------------------------------------------
-
-# Environment detection
-ENVIRONMENT = os.getenv("ENVIRONMENT", "development")
-IS_PRODUCTION = ENVIRONMENT == "production"
-
-# Encryption Key - Load from environment variable
-ENCRYPTION_KEY = os.getenv("ENCRYPTION_KEY")
-
-if not ENCRYPTION_KEY:
-    if IS_PRODUCTION:
-        logging.critical("🔴 FATAL ERROR: ENCRYPTION_KEY must be set in production environment")
-        logging.critical("Generate one with: python -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())'")
-        sys.exit(1)  # Fail-fast in production
-    else:
-        # Development fallback with ephemeral key
-        ENCRYPTION_KEY = Fernet.generate_key().decode()
-        logging.warning("⚠️  WARNING: Using ephemeral encryption key (development only)")
-
-# Initialize cipher
-cipher_suite = Fernet(ENCRYPTION_KEY.encode() if isinstance(ENCRYPTION_KEY, str) else ENCRYPTION_KEY)
-
-def encrypt_data(data: str) -> str:
-    """
-    Encrypts a string and returns the encrypted token as a string.
-    """
-    if not data: return ""
-    encrypted_bytes = cipher_suite.encrypt(data.encode('utf-8'))
-    return encrypted_bytes.decode('utf-8')
-
-def decrypt_data(token: str) -> Optional[str]:
-    """
-    Decrypts a token and returns the original string.
-    """
-    if not token: return None
-    try:
-        decrypted_bytes = cipher_suite.decrypt(token.encode('utf-8'))
-        return decrypted_bytes.decode('utf-8')
-    except Exception as e:
-        print(f"Decryption failed: {e}")
-        return None
-
-def rotate_key():
-    """
-    Example function to rotate keys (advanced).
-    """
-    global key, cipher_suite
-    key = Fernet.generate_key()
-    cipher_suite = Fernet(key)
-    with open(ENCRYPTION_KEY_PATH, "wb") as key_file:
-        key_file.write(key)
-    print(f"New key generated and saved to {ENCRYPTION_KEY_PATH}")
-
-if __name__ == "__main__":
-    # Test
-    original = "Jean Dupont - Patient Zero"
-    encrypted = encrypt_data(original)
-    decrypted = decrypt_data(encrypted)
-    
-    print(f"Original: {original}")
-    print(f"Encrypted: {encrypted}")
-    print(f"Decrypted: {decrypted}")
-    assert original == decrypted
+from cryptography.fernet import Fernet
+import os
+import sys
+import logging
+from typing import Optional
+
+# -------------------------------------------------------------------------
+# ENCRYPTION CONFIGURATION - PRODUCTION READY
+# -------------------------------------------------------------------------
+
+# Environment detection
+ENVIRONMENT = os.getenv("ENVIRONMENT", "development")
+IS_PRODUCTION = ENVIRONMENT == "production"
+
+# Encryption Key - Load from environment variable
+ENCRYPTION_KEY = os.getenv("ENCRYPTION_KEY")
+
+if not ENCRYPTION_KEY:
+    if IS_PRODUCTION:
+        logging.critical("🔴 FATAL ERROR: ENCRYPTION_KEY must be set in production environment")
+        logging.critical("Generate one with: python -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())'")
+        sys.exit(1)  # Fail-fast in production
+    else:
+        # Development fallback with ephemeral key
+        ENCRYPTION_KEY = Fernet.generate_key().decode()
+        logging.warning("⚠️  WARNING: Using ephemeral encryption key (development only)")
+
+# Initialize cipher
+cipher_suite = Fernet(ENCRYPTION_KEY.encode() if isinstance(ENCRYPTION_KEY, str) else ENCRYPTION_KEY)
+
+def encrypt_data(data: str) -> str:
+    """
+    Encrypts a string and returns the encrypted token as a string.
+    """
+    if not data: return ""
+    encrypted_bytes = cipher_suite.encrypt(data.encode('utf-8'))
+    return encrypted_bytes.decode('utf-8')
+
+def decrypt_data(token: str) -> Optional[str]:
+    """
+    Decrypts a token and returns the original string.
+    """
+    if not token: return None
+    try:
+        decrypted_bytes = cipher_suite.decrypt(token.encode('utf-8'))
+        return decrypted_bytes.decode('utf-8')
+    except Exception as e:
+        print(f"Decryption failed: {e}")
+        return None
+
+def rotate_key():
+    """
+    Example function to rotate keys (advanced).
+    """
+    global key, cipher_suite
+    key = Fernet.generate_key()
+    cipher_suite = Fernet(key)
+    with open(ENCRYPTION_KEY_PATH, "wb") as key_file:
+        key_file.write(key)
+    print(f"New key generated and saved to {ENCRYPTION_KEY_PATH}")
+
+if __name__ == "__main__":
+    # Test
+    original = "Jean Dupont - Patient Zero"
+    encrypted = encrypt_data(original)
+    decrypted = decrypt_data(encrypted)
+    
+    print(f"Original: {original}")
+    print(f"Encrypted: {encrypted}")
+    print(f"Decrypted: {decrypted}")
+    assert original == decrypted

explainability.py

@@ -1,462 +1,378 @@
-
-import torch
-import torch.nn as nn
-import numpy as np
-import cv2
-from PIL import Image
-import logging
-from typing import List, Dict, Any, Optional, Tuple, Union
-from pytorch_grad_cam import GradCAMPlusPlus
-from pytorch_grad_cam.utils.image import show_cam_on_image
-from dataclasses import dataclass
-
-logger = logging.getLogger(__name__)
-
-# =========================================================================
-# CONFIGURATION & EXPERT KNOWLEDGE
-# =========================================================================
-
-@dataclass
-class ExpertSegConfig:
-    modality: str
-    target_organ: str
-    anatomical_prompts: List[str] # For Segmentation Mask
-    threshold_percentile: int # Top X% activation
-    min_area_ratio: float
-    max_area_ratio: float
-    morphology_kernel: int
-
-# Expert Knowledge Base
-EXPERT_KNOWLEDGE = {
-    "Thoracic": ExpertSegConfig(
-        modality="CXR/CT",
-        target_organ="Lung Parenchyma",
-        anatomical_prompts=[
-            "lung parenchyma",
-            "bilateral lungs",
-            "pulmonary fields",
-            "chest x-ray lungs excluding heart"
-        ],
-        threshold_percentile=75, # Top 25%
-        min_area_ratio=0.15,
-        max_area_ratio=0.60,
-        morphology_kernel=7
-    ),
-    "Orthopedics": ExpertSegConfig(
-        modality="X-Ray",
-        target_organ="Bone Structure",
-        anatomical_prompts=[
-            "bone structure",
-            "knee joint",
-            "cortical bone",
-            "skeletal anatomy"
-        ],
-        threshold_percentile=85, # Top 15%
-        min_area_ratio=0.05,
-        max_area_ratio=0.50,
-        morphology_kernel=5
-    ),
-    "Default": ExpertSegConfig(
-        modality="General",
-        target_organ="Body Part",
-        anatomical_prompts=["medical image body part"],
-        threshold_percentile=80,
-        min_area_ratio=0.05,
-        max_area_ratio=0.90,
-        morphology_kernel=5
-    )
-}
-
-# =========================================================================
-# WRAPPERS AND UTILS
-# =========================================================================
-
-class HuggingFaceWeirdCLIPWrapper(nn.Module):
-    """
-    Wraps SigLIP to act like a standard classifier for Grad-CAM.
-    Target: Cosine Similarity Score.
-    """
-    def __init__(self, model, text_input_ids, attention_mask):
-        super(HuggingFaceWeirdCLIPWrapper, self).__init__()
-        self.model = model
-        self.text_input_ids = text_input_ids
-        self.attention_mask = attention_mask
-
-    def forward(self, pixel_values):
-        outputs = self.model(
-            pixel_values=pixel_values, 
-            input_ids=self.text_input_ids,
-            attention_mask=self.attention_mask
-        )
-        # outputs.logits_per_image is (Batch, Num_Prompts)
-        # This IS the similarity score (scaled). 
-        # Grad-CAM++ will derive gradients relative to this score.
-        return outputs.logits_per_image
-
-def reshape_transform(tensor, width=32, height=32):
-    """Reshape Transformer attention/embeddings for Grad-CAM."""
-    # Squeeze CLS if present logic (usually SigLIP doesn't have it in last layers same way)
-    # Tensor: (Batch, Num_Tokens, Dim)
-    num_tokens = tensor.size(1)
-    side = int(np.sqrt(num_tokens))
-    result = tensor.reshape(tensor.size(0), side, side, tensor.size(2))
-    # Bring channels first: (B, C, H, W)
-    result = result.transpose(2, 3).transpose(1, 2)
-    return result
-
-# =========================================================================
-# EXPERT+ EXPLAINABILITY ENGINE
-# =========================================================================
-
-class ExplainabilityEngine:
-    def __init__(self, model_wrapper):
-        self.wrapper = model_wrapper
-        self.model = model_wrapper.model
-        self.processor = model_wrapper.processor
-        self.device = self.model.device
-
-    def _get_expert_config(self, anatomical_context: str) -> ExpertSegConfig:
-        if "lung" in anatomical_context.lower():
-            return EXPERT_KNOWLEDGE["Thoracic"]
-        elif "bone" in anatomical_context.lower() or "knee" in anatomical_context.lower():
-            return EXPERT_KNOWLEDGE["Orthopedics"]
-        else:
-            base = EXPERT_KNOWLEDGE["Default"]
-            base.anatomical_prompts = [anatomical_context]
-            return base
-
-    def generate_expert_mask(self, image: Image.Image, config: ExpertSegConfig) -> Dict[str, Any]:
-        """
-        Expert Segmentation: 
-        Multi-Prompt Ensembling -> Patch Similarity -> Adaptive Threshold -> Morphology -> Validation.
-        """
-        audit = {
-            "seg_prompts": config.anatomical_prompts,
-            "seg_status": "INIT"
-        }
-        try:
-            w, h = image.size
-            inputs = self.processor(text=config.anatomical_prompts, images=image, padding="max_length", return_tensors="pt")
-            inputs = {k: v.to(self.device) for k, v in inputs.items()}
-            
-            with torch.no_grad():
-                # Vision Features (1, Token, Dim)
-                vision_outputs = self.model.vision_model(
-                    pixel_values=inputs["pixel_values"],
-                    output_hidden_states=True
-                )
-                last_hidden_state = vision_outputs.last_hidden_state
-                
-                # Text Features (Prompts, Dim)
-                # Text Features (Prompts, Dim)
-                # FIX: Robustly handle attention_mask (some processors don't return it for text-only inputs if irrelevant)
-                text_inputs_ids = inputs["input_ids"]
-                text_attention_mask = inputs.get("attention_mask")
-                
-                if text_attention_mask is None:
-                    text_attention_mask = torch.ones_like(text_inputs_ids)
-
-                text_outputs = self.model.text_model(
-                    input_ids=text_inputs_ids,
-                    attention_mask=text_attention_mask
-                )
-                text_embeds = text_outputs.pooler_output
-                text_embeds = text_embeds / text_embeds.norm(p=2, dim=-1, keepdim=True)
-                
-                # Similarity: (1, T, D) @ (D, P) -> (1, T, P)
-                sim_map = torch.matmul(last_hidden_state, text_embeds.t())
-                # Mean across Prompts -> (1, T)
-                sim_map = sim_map.mean(dim=2)
-                
-                # Reshape & Upscale
-                num_tokens = sim_map.size(1)
-                side = int(np.sqrt(num_tokens))
-                sim_grid = sim_map.reshape(1, side, side)
-                
-                sim_grid = torch.nn.functional.interpolate(
-                    sim_grid.unsqueeze(0), 
-                    size=(h, w), 
-                    mode='bilinear', 
-                    align_corners=False
-                ).squeeze().cpu().numpy()
-                
-                # Adaptive Thresholding (Percentile)
-                thresh = np.percentile(sim_grid, config.threshold_percentile)
-                binary_mask = (sim_grid > thresh).astype(np.float32)
-                audit["seg_threshold"] = float(thresh)
-
-                # Morphological Cleaning
-                kernel = np.ones((config.morphology_kernel, config.morphology_kernel), np.uint8)
-                binary_mask = cv2.morphologyEx(binary_mask, cv2.MORPH_OPEN, kernel) # Remove noise
-                binary_mask = cv2.morphologyEx(binary_mask, cv2.MORPH_CLOSE, kernel) # Fill holes
-                binary_mask = cv2.GaussianBlur(binary_mask, (15, 15), 0) # Smooth contours
-                binary_mask = (binary_mask - binary_mask.min()) / (binary_mask.max() - binary_mask.min() + 1e-8)
-                
-                # Validation
-                val = self._validate_mask(binary_mask, config)
-                audit["seg_validation"] = val
-                
-                if not val["valid"]:
-                    logger.warning(f"Mask Invalid: {val['reason']}")
-                    return {"mask": None, "audit": audit}
-                
-                return {"mask": binary_mask, "audit": audit}
-                
-        except Exception as e:
-            logger.error(f"Segmentation Failed: {e}")
-            audit["seg_error"] = str(e)
-            return {"mask": None, "audit": audit}
-
-    def _validate_mask(self, mask: np.ndarray, config: ExpertSegConfig) -> Dict[str, Any]:
-        area_ratio = np.sum(mask > 0.5) / mask.size
-        
-        if area_ratio < config.min_area_ratio:
-            return {"valid": False, "reason": f"Small Area: {area_ratio:.2f} < {config.min_area_ratio}"}
-        if area_ratio > config.max_area_ratio:
-            return {"valid": False, "reason": f"Large Area: {area_ratio:.2f} > {config.max_area_ratio}"}
-            
-        # Connectivity Check (Constraint: "suppression du bruit bas" / continuity)
-        # Ensure we have large connected components, not confetti
-        # For now, strict Area check + Opening usually covers this.
-        return {"valid": True}
-
-    def generate_expert_gradcam(self, image: Image.Image, target_prompts: List[str]) -> Dict[str, Any]:
-        """
-        Expert Grad-CAM:
-        1. Multi-Prompt Ensembling (Averaging heatmaps).
-        2. Layer Selection: Encoder Layer -2.
-        3. Target: Cosine Score.
-        """
-        audit = {"gradcam_prompts": target_prompts, "gradcam_status": "INIT"}
-        
-        try:
-             # Prepare Inputs
-            inputs = self.processor(text=target_prompts, images=image, padding="max_length", return_tensors="pt")
-            inputs = {k: v.to(self.device) for k, v in inputs.items()}
-            
-            # Robust Mask handling
-            input_ids = inputs.get('input_ids')
-            attention_mask = inputs.get('attention_mask')
-            if attention_mask is None and input_ids is not None:
-                attention_mask = torch.ones_like(input_ids)
-
-            # Wrapper
-            model_wrapper_cam = HuggingFaceWeirdCLIPWrapper(self.model, input_ids, attention_mask)
-            
-            # Layer Selection: 2nd to last encoder layer (Better spatial features than last Norm)
-            # SigLIP structure: model.vision_model.encoder.layers
-            target_layers = [self.model.vision_model.encoder.layers[-2].layer_norm1] 
-            
-            cam = GradCAMPlusPlus(
-                model=model_wrapper_cam, 
-                target_layers=target_layers,
-                reshape_transform=reshape_transform # Needs to handle (B, T, D)
-            )
-            
-            pixel_values = inputs.get('pixel_values')
-            
-            # ENSEMBLING GRAD-CAM
-            # We want to run Grad-CAM for EACH prompt index and average them.
-            # Grayscale CAM output is (Batch, H, W)
-            # We assume Batch=1 here.
-            
-            maps = []
-            for i in range(len(target_prompts)):
-                # Target Class Index = i (The index of the prompt in the logits)
-                # GradCAMPlusPlus targets=[ClassifierOutputTarget(i)]
-                from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
-                
-                targets = [ClassifierOutputTarget(i)]
-                grayscale_cam = cam(input_tensor=pixel_values, targets=targets)
-                maps.append(grayscale_cam[0, :])
-            
-            # Average
-            avg_cam = np.mean(np.array(maps), axis=0)
-            
-            # Point 5: Smart Normalization & Thresholding
-            # "cam = normalize(cam)"
-            if avg_cam.max() > avg_cam.min():
-                avg_cam = (avg_cam - avg_cam.min()) / (avg_cam.max() - avg_cam.min())
-            
-            # "mask = cam > percentile(cam, 85)" - Removing low confidence noise
-            # We keep it continuous for heatmap but suppress low values
-            # Using 80th percentile as soft threshold (User said 85, let's use 80 to be safe but clean)
-            cam_threshold = np.percentile(avg_cam, 80) 
-            avg_cam[avg_cam < cam_threshold] = 0.0
-            
-            # Re-normalize the top 20% to spread 0-1 for visibility
-            if avg_cam.max() > 0:
-                avg_cam = avg_cam / avg_cam.max()
-            
-            # Smoothing after thresholding to remove jagged edges
-            avg_cam = cv2.GaussianBlur(avg_cam, (11, 11), 0)
-            
-            audit["gradcam_threshold_val"] = float(cam_threshold)
-            
-            return {"map": avg_cam, "audit": audit}
-            
-        except Exception as e:
-            logger.error(f"Grad-CAM Failed: {e}")
-            audit["gradcam_error"] = str(e)
-            return {"map": None, "audit": audit}
-
-    def explain(self, image: Image.Image, target_text: str, anatomical_context: str) -> Dict[str, Any]:
-        """
-        Final Expert Fusion Pipeline.
-        """
-        # 0. Setup
-        config = self._get_expert_config(anatomical_context)
-        
-        # 1. Anatomical Mask (Strict Constraint)
-        seg_res = self.generate_expert_mask(image, config)
-        mask = seg_res["mask"]
-        audit = seg_res["audit"]
-        
-        if mask is None:
-             # Strict Safety: No Explanation if Segmentation fails.
-            return {
-                "heatmap_array": None, 
-                "heatmap_raw": None, 
-                "reliability_score": 0.0, 
-                "confidence_label": "UNSAFE", # Point 8
-                "audit": audit,
-                "display_text": "Validation Anatomique Échouée"
-            }
-            
-        # 2. Attention Map (Multi-Prompt)
-        # Using list of prompts implies Multi-Prompt Grad-CAM (Point 4)
-        # We can auto-augment target_text if needed, but for now we trust the input.
-        gradcam_res = self.generate_expert_gradcam(image, [target_text])
-        heatmap = gradcam_res["map"]
-        audit.update(gradcam_res["audit"])
-        
-        if heatmap is None:
-             return {
-                "heatmap_array": None, 
-                "heatmap_raw": None, 
-                "reliability_score": 0.0, 
-                "confidence_label": "LOW",
-                "audit": audit,
-                "display_text": "Attention Insuffisante"
-            }
-             
-        # 3. Constraint Fusion (Point 7)
-        if mask.shape != heatmap.shape:
-             mask = cv2.resize(mask, (heatmap.shape[1], heatmap.shape[0]))
-             
-        final_map = heatmap * mask
-        
-        # 4. Reliability (Point 8)
-        total = np.sum(heatmap) + 1e-8
-        retained = np.sum(final_map)
-        reliability = retained / total
-        
-        # Point 9: Responsible Display
-        confidence = "HIGH" if reliability > 0.6 else "LOW"
-        # FIX: JSON Serialization Error (np.float32 -> float)
-        audit["reliability_score"] = round(float(reliability), 4)
-        
-        # 5. Visualize
-        img_np = np.array(image)
-        
-        # FIX: Ensure img_np is float32 [0,1]
-        img_np = img_np.astype(np.float32) / 255.0
-        
-        # FIX: Resize final_map (Heatmap) to match Original Image Size
-        # show_cam_on_image requires heatmap and image to be same shape
-        if final_map.shape != img_np.shape[:2]:
-            final_map = cv2.resize(final_map, (img_np.shape[1], img_np.shape[0]))
-            
-        visualization = show_cam_on_image(img_np, final_map, use_rgb=True)
-        
-        return {
-            "heatmap_array": visualization,
-            "heatmap_raw": final_map,
-            # FIX: Cast to float for JSON safety
-            "reliability_score": round(float(reliability), 2),
-            "confidence_label": confidence,
-            "display_text": "Zone d'attention du modèle (Grad-CAM++)"
-        }
-
-    def calculate_cardiothoracic_ratio(self, image: Image.Image) -> Dict[str, Any]:
-        """
-        Morphology Engine: Calculate Heart/Thorax Ratio (CTR).
-        
-        Algorithm:
-        1. Segment Heart (Prompt: 'heart silhouette')
-        2. Segment Lungs (Prompt: 'lungs thoracic cage')
-        3. Calculate Max Width of Heart Mask.
-        4. Calculate Max Width of Lung Mask (at Costophrenic angle ideally, but Max Width is proxy).
-        5. Ratio = Heart / Lungs.
-        """
-        audit = {"ctr_status": "INIT"}
-        
-        try:
-            # 1. Heart Segmentation
-            heart_config = ExpertSegConfig(
-                modality="CXR",
-                target_organ="Heart",
-                anatomical_prompts=["heart silhouette", "cardiac shadow", "mediastinum"],
-                threshold_percentile=85, # Heart is salient
-                min_area_ratio=0.05,
-                max_area_ratio=0.40,
-                morphology_kernel=5
-            )
-            heart_res = self.generate_expert_mask(image, heart_config)
-            heart_mask = heart_res["mask"]
-            
-            if heart_mask is None:
-                return {"ctr": 0.0, "valid": False, "reason": "Heart segmentation failed"}
-                
-            # 2. Lung/Thorax Segmentation
-            lung_config = ExpertSegConfig(
-                modality="CXR",
-                target_organ="Thorax",
-                anatomical_prompts=["lung fields", "thoracic cage", "rib cage", "diaphragm"],
-                threshold_percentile=75,
-                min_area_ratio=0.20,
-                max_area_ratio=0.85,
-                morphology_kernel=5
-            )
-            lung_res = self.generate_expert_mask(image, lung_config)
-            lung_mask = lung_res["mask"]
-            
-            if lung_mask is None:
-                 return {"ctr": 0.0, "valid": False, "reason": "Lung segmentation failed"}
-                 
-            # 3. Calculate Widths
-            # Sum along Vertical Axis (0) -> shape (Width,)
-            # Pixels > 0.5 count as "structure"
-            
-            # Heart Width
-            heart_proj = np.max(heart_mask, axis=0) # [0, 1] projection
-            heart_pixels = np.where(heart_proj > 0.5)[0]
-            if len(heart_pixels) == 0:
-                 return {"ctr": 0.0, "valid": False, "reason": "Empty heart mask"}
-            heart_width = heart_pixels.max() - heart_pixels.min()
-            
-            # Lung Width
-            lung_proj = np.max(lung_mask, axis=0)
-            lung_pixels = np.where(lung_proj > 0.5)[0]
-            if len(lung_pixels) == 0:
-                 return {"ctr": 0.0, "valid": False, "reason": "Empty lung mask"}
-            lung_width = lung_pixels.max() - lung_pixels.min()
-            
-            # 4. Compute Ratio
-            if lung_width == 0:
-                 return {"ctr": 0.0, "valid": False, "reason": "Zero lung width"}
-                 
-            ctr = heart_width / lung_width
-            logger.info(f"📐 Morphology Engine: Heart={heart_width}px, Lungs={lung_width}px, CTR={ctr:.2f}")
-            
-            return {
-                "ctr": round(float(ctr), 2),
-                "heart_width_px": int(heart_width),
-                "lung_width_px": int(lung_width),
-                "valid": True,
-                "reason": "Success"
-            }
-            
-        except Exception as e:
-            logger.error(f"CTR Calculation Failed: {e}")
-            return {"ctr": 0.0, "valid": False, "reason": str(e)}
+
+import torch
+import torch.nn as nn
+import numpy as np
+import cv2
+from PIL import Image
+import logging
+from typing import List, Dict, Any, Optional, Tuple, Union
+from pytorch_grad_cam import GradCAMPlusPlus
+from pytorch_grad_cam.utils.image import show_cam_on_image
+from dataclasses import dataclass
+
+logger = logging.getLogger(__name__)
+
+# =========================================================================
+# CONFIGURATION & EXPERT KNOWLEDGE
+# =========================================================================
+
+@dataclass
+class ExpertSegConfig:
+    modality: str
+    target_organ: str
+    anatomical_prompts: List[str] # For Segmentation Mask
+    threshold_percentile: int # Top X% activation
+    min_area_ratio: float
+    max_area_ratio: float
+    morphology_kernel: int
+
+# Expert Knowledge Base
+EXPERT_KNOWLEDGE = {
+    "Thoracic": ExpertSegConfig(
+        modality="CXR/CT",
+        target_organ="Lung Parenchyma",
+        anatomical_prompts=[
+            "lung parenchyma",
+            "bilateral lungs",
+            "pulmonary fields",
+            "chest x-ray lungs excluding heart"
+        ],
+        threshold_percentile=75, # Top 25%
+        min_area_ratio=0.15,
+        max_area_ratio=0.60,
+        morphology_kernel=7
+    ),
+    "Orthopedics": ExpertSegConfig(
+        modality="X-Ray",
+        target_organ="Bone Structure",
+        anatomical_prompts=[
+            "bone structure",
+            "knee joint",
+            "cortical bone",
+            "skeletal anatomy"
+        ],
+        threshold_percentile=85, # Top 15%
+        min_area_ratio=0.05,
+        max_area_ratio=0.50,
+        morphology_kernel=5
+    ),
+    "Default": ExpertSegConfig(
+        modality="General",
+        target_organ="Body Part",
+        anatomical_prompts=["medical image body part"],
+        threshold_percentile=80,
+        min_area_ratio=0.05,
+        max_area_ratio=0.90,
+        morphology_kernel=5
+    )
+}
+
+# =========================================================================
+# WRAPPERS AND UTILS
+# =========================================================================
+
+class HuggingFaceWeirdCLIPWrapper(nn.Module):
+    """
+    Wraps SigLIP to act like a standard classifier for Grad-CAM.
+    Target: Cosine Similarity Score.
+    """
+    def __init__(self, model, text_input_ids, attention_mask):
+        super(HuggingFaceWeirdCLIPWrapper, self).__init__()
+        self.model = model
+        self.text_input_ids = text_input_ids
+        self.attention_mask = attention_mask
+
+    def forward(self, pixel_values):
+        outputs = self.model(
+            pixel_values=pixel_values, 
+            input_ids=self.text_input_ids,
+            attention_mask=self.attention_mask
+        )
+        # outputs.logits_per_image is (Batch, Num_Prompts)
+        # This IS the similarity score (scaled). 
+        # Grad-CAM++ will derive gradients relative to this score.
+        return outputs.logits_per_image
+
+def reshape_transform(tensor, width=32, height=32):
+    """Reshape Transformer attention/embeddings for Grad-CAM."""
+    # Squeeze CLS if present logic (usually SigLIP doesn't have it in last layers same way)
+    # Tensor: (Batch, Num_Tokens, Dim)
+    num_tokens = tensor.size(1)
+    side = int(np.sqrt(num_tokens))
+    result = tensor.reshape(tensor.size(0), side, side, tensor.size(2))
+    # Bring channels first: (B, C, H, W)
+    result = result.transpose(2, 3).transpose(1, 2)
+    return result
+
+# =========================================================================
+# EXPERT+ EXPLAINABILITY ENGINE
+# =========================================================================
+
+class ExplainabilityEngine:
+    def __init__(self, model_wrapper):
+        self.wrapper = model_wrapper
+        self.model = model_wrapper.model
+        self.processor = model_wrapper.processor
+        self.device = self.model.device
+
+    def _get_expert_config(self, anatomical_context: str) -> ExpertSegConfig:
+        if "lung" in anatomical_context.lower():
+            return EXPERT_KNOWLEDGE["Thoracic"]
+        elif "bone" in anatomical_context.lower() or "knee" in anatomical_context.lower():
+            return EXPERT_KNOWLEDGE["Orthopedics"]
+        else:
+            base = EXPERT_KNOWLEDGE["Default"]
+            base.anatomical_prompts = [anatomical_context]
+            return base
+
+    def generate_expert_mask(self, image: Image.Image, config: ExpertSegConfig) -> Dict[str, Any]:
+        """
+        Expert Segmentation: 
+        Multi-Prompt Ensembling -> Patch Similarity -> Adaptive Threshold -> Morphology -> Validation.
+        """
+        audit = {
+            "seg_prompts": config.anatomical_prompts,
+            "seg_status": "INIT"
+        }
+        try:
+            w, h = image.size
+            inputs = self.processor(text=config.anatomical_prompts, images=image, padding="max_length", return_tensors="pt")
+            inputs = {k: v.to(self.device) for k, v in inputs.items()}
+            
+            with torch.no_grad():
+                # Vision Features (1, Token, Dim)
+                vision_outputs = self.model.vision_model(
+                    pixel_values=inputs["pixel_values"],
+                    output_hidden_states=True
+                )
+                last_hidden_state = vision_outputs.last_hidden_state
+                
+                # Text Features (Prompts, Dim)
+                # Text Features (Prompts, Dim)
+                # FIX: Robustly handle attention_mask (some processors don't return it for text-only inputs if irrelevant)
+                text_inputs_ids = inputs["input_ids"]
+                text_attention_mask = inputs.get("attention_mask")
+                
+                if text_attention_mask is None:
+                    text_attention_mask = torch.ones_like(text_inputs_ids)
+
+                text_outputs = self.model.text_model(
+                    input_ids=text_inputs_ids,
+                    attention_mask=text_attention_mask
+                )
+                text_embeds = text_outputs.pooler_output
+                text_embeds = text_embeds / text_embeds.norm(p=2, dim=-1, keepdim=True)
+                
+                # Similarity: (1, T, D) @ (D, P) -> (1, T, P)
+                sim_map = torch.matmul(last_hidden_state, text_embeds.t())
+                # Mean across Prompts -> (1, T)
+                sim_map = sim_map.mean(dim=2)
+                
+                # Reshape & Upscale
+                num_tokens = sim_map.size(1)
+                side = int(np.sqrt(num_tokens))
+                sim_grid = sim_map.reshape(1, side, side)
+                
+                sim_grid = torch.nn.functional.interpolate(
+                    sim_grid.unsqueeze(0), 
+                    size=(h, w), 
+                    mode='bilinear', 
+                    align_corners=False
+                ).squeeze().cpu().numpy()
+                
+                # Adaptive Thresholding (Percentile)
+                thresh = np.percentile(sim_grid, config.threshold_percentile)
+                binary_mask = (sim_grid > thresh).astype(np.float32)
+                audit["seg_threshold"] = float(thresh)
+
+                # Morphological Cleaning
+                kernel = np.ones((config.morphology_kernel, config.morphology_kernel), np.uint8)
+                binary_mask = cv2.morphologyEx(binary_mask, cv2.MORPH_OPEN, kernel) # Remove noise
+                binary_mask = cv2.morphologyEx(binary_mask, cv2.MORPH_CLOSE, kernel) # Fill holes
+                binary_mask = cv2.GaussianBlur(binary_mask, (15, 15), 0) # Smooth contours
+                binary_mask = (binary_mask - binary_mask.min()) / (binary_mask.max() - binary_mask.min() + 1e-8)
+                
+                # Validation
+                val = self._validate_mask(binary_mask, config)
+                audit["seg_validation"] = val
+                
+                if not val["valid"]:
+                    logger.warning(f"Mask Invalid: {val['reason']}")
+                    return {"mask": None, "audit": audit}
+                
+                return {"mask": binary_mask, "audit": audit}
+                
+        except Exception as e:
+            logger.error(f"Segmentation Failed: {e}")
+            audit["seg_error"] = str(e)
+            return {"mask": None, "audit": audit}
+
+    def _validate_mask(self, mask: np.ndarray, config: ExpertSegConfig) -> Dict[str, Any]:
+        area_ratio = np.sum(mask > 0.5) / mask.size
+        
+        if area_ratio < config.min_area_ratio:
+            return {"valid": False, "reason": f"Small Area: {area_ratio:.2f} < {config.min_area_ratio}"}
+        if area_ratio > config.max_area_ratio:
+            return {"valid": False, "reason": f"Large Area: {area_ratio:.2f} > {config.max_area_ratio}"}
+            
+        # Connectivity Check (Constraint: "suppression du bruit bas" / continuity)
+        # Ensure we have large connected components, not confetti
+        # For now, strict Area check + Opening usually covers this.
+        return {"valid": True}
+
+    def generate_expert_gradcam(self, image: Image.Image, target_prompts: List[str]) -> Dict[str, Any]:
+        """
+        Expert Grad-CAM:
+        1. Multi-Prompt Ensembling (Averaging heatmaps).
+        2. Layer Selection: Encoder Layer -2.
+        3. Target: Cosine Score.
+        """
+        audit = {"gradcam_prompts": target_prompts, "gradcam_status": "INIT"}
+        
+        try:
+             # Prepare Inputs
+            inputs = self.processor(text=target_prompts, images=image, padding="max_length", return_tensors="pt")
+            inputs = {k: v.to(self.device) for k, v in inputs.items()}
+            
+            # Robust Mask handling
+            input_ids = inputs.get('input_ids')
+            attention_mask = inputs.get('attention_mask')
+            if attention_mask is None and input_ids is not None:
+                attention_mask = torch.ones_like(input_ids)
+
+            # Wrapper
+            model_wrapper_cam = HuggingFaceWeirdCLIPWrapper(self.model, input_ids, attention_mask)
+            
+            # Layer Selection: 2nd to last encoder layer (Better spatial features than last Norm)
+            # SigLIP structure: model.vision_model.encoder.layers
+            target_layers = [self.model.vision_model.encoder.layers[-2].layer_norm1] 
+            
+            cam = GradCAMPlusPlus(
+                model=model_wrapper_cam, 
+                target_layers=target_layers,
+                reshape_transform=reshape_transform # Needs to handle (B, T, D)
+            )
+            
+            pixel_values = inputs.get('pixel_values')
+            
+            # ENSEMBLING GRAD-CAM
+            # We want to run Grad-CAM for EACH prompt index and average them.
+            # Grayscale CAM output is (Batch, H, W)
+            # We assume Batch=1 here.
+            
+            maps = []
+            for i in range(len(target_prompts)):
+                # Target Class Index = i (The index of the prompt in the logits)
+                # GradCAMPlusPlus targets=[ClassifierOutputTarget(i)]
+                from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
+                
+                targets = [ClassifierOutputTarget(i)]
+                grayscale_cam = cam(input_tensor=pixel_values, targets=targets)
+                maps.append(grayscale_cam[0, :])
+            
+            # Average
+            avg_cam = np.mean(np.array(maps), axis=0)
+            
+            # Point 5: Smart Normalization & Thresholding
+            # "cam = normalize(cam)"
+            if avg_cam.max() > avg_cam.min():
+                avg_cam = (avg_cam - avg_cam.min()) / (avg_cam.max() - avg_cam.min())
+            
+            # "mask = cam > percentile(cam, 85)" - Removing low confidence noise
+            # We keep it continuous for heatmap but suppress low values
+            # Using 80th percentile as soft threshold (User said 85, let's use 80 to be safe but clean)
+            cam_threshold = np.percentile(avg_cam, 80) 
+            avg_cam[avg_cam < cam_threshold] = 0.0
+            
+            # Re-normalize the top 20% to spread 0-1 for visibility
+            if avg_cam.max() > 0:
+                avg_cam = avg_cam / avg_cam.max()
+            
+            # Smoothing after thresholding to remove jagged edges
+            avg_cam = cv2.GaussianBlur(avg_cam, (11, 11), 0)
+            
+            audit["gradcam_threshold_val"] = float(cam_threshold)
+            
+            return {"map": avg_cam, "audit": audit}
+            
+        except Exception as e:
+            logger.error(f"Grad-CAM Failed: {e}")
+            audit["gradcam_error"] = str(e)
+            return {"map": None, "audit": audit}
+
+    def explain(self, image: Image.Image, target_text: str, anatomical_context: str) -> Dict[str, Any]:
+        """
+        Final Expert Fusion Pipeline.
+        """
+        # 0. Setup
+        config = self._get_expert_config(anatomical_context)
+        
+        # 1. Anatomical Mask (Strict Constraint)
+        seg_res = self.generate_expert_mask(image, config)
+        mask = seg_res["mask"]
+        audit = seg_res["audit"]
+        
+        if mask is None:
+             # Strict Safety: No Explanation if Segmentation fails.
+            return {
+                "heatmap_array": None, 
+                "heatmap_raw": None, 
+                "reliability_score": 0.0, 
+                "confidence_label": "UNSAFE", # Point 8
+                "audit": audit,
+                "display_text": "Validation Anatomique Échouée"
+            }
+            
+        # 2. Attention Map (Multi-Prompt)
+        # Using list of prompts implies Multi-Prompt Grad-CAM (Point 4)
+        # We can auto-augment target_text if needed, but for now we trust the input.
+        gradcam_res = self.generate_expert_gradcam(image, [target_text])
+        heatmap = gradcam_res["map"]
+        audit.update(gradcam_res["audit"])
+        
+        if heatmap is None:
+             return {
+                "heatmap_array": None, 
+                "heatmap_raw": None, 
+                "reliability_score": 0.0, 
+                "confidence_label": "LOW",
+                "audit": audit,
+                "display_text": "Attention Insuffisante"
+            }
+             
+        # 3. Constraint Fusion (Point 7)
+        if mask.shape != heatmap.shape:
+             mask = cv2.resize(mask, (heatmap.shape[1], heatmap.shape[0]))
+             
+        final_map = heatmap * mask
+        
+        # 4. Reliability (Point 8)
+        total = np.sum(heatmap) + 1e-8
+        retained = np.sum(final_map)
+        reliability = retained / total
+        
+        # Point 9: Responsible Display
+        confidence = "HIGH" if reliability > 0.6 else "LOW"
+        audit["reliability_score"] = round(reliability, 4)
+        
+        # 5. Visualize
+        img_np = np.array(image)
+        
+        # FIX: Ensure img_np is float32 [0,1]
+        img_np = img_np.astype(np.float32) / 255.0
+        
+        # FIX: Resize final_map (Heatmap) to match Original Image Size
+        # show_cam_on_image requires heatmap and image to be same shape
+        if final_map.shape != img_np.shape[:2]:
+            final_map = cv2.resize(final_map, (img_np.shape[1], img_np.shape[0]))
+            
+        visualization = show_cam_on_image(img_np, final_map, use_rgb=True)
+        
+        return {
+            "heatmap_array": visualization,
+            "heatmap_raw": final_map,
+            "reliability_score": round(reliability, 2),
+            "confidence_label": confidence,
+            "audit": audit,
+            "display_text": "Zone d'attention du modèle (Grad-CAM++)"
+        }

localization.py

@@ -1,83 +1,83 @@
-# Mappings de localisation (Anglais -> Français)
-# Ce fichier permet de traduire les résultats de l'IA sans modifier les prompts originaux
-# qui doivent rester en anglais pour la performance du modèle.
-
-DOMAIN_TRANSLATIONS = {
-    'Thoracic': {
-        'label': 'Thoracique',
-        'description': 'Analyse Radiographique du Thorax'
-    },
-    'Dermatology': {
-        'label': 'Dermatologie',
-        'description': 'Analyse Dermatoscope des Lésions Cutanées'
-    },
-    'Histology': {
-        'label': 'Histologie',
-        'description': 'Analyse Microscopique (H&E)'
-    },
-    'Ophthalmology': {
-        'label': 'Ophtalmologie',
-        'description': 'Fond d\'Oeil (Rétine)'
-    },
-    'Orthopedics': {
-        'label': 'Orthopédie',
-        'description': 'Radiographie Osseuse'
-    }
-}
-
-LABEL_TRANSLATIONS = {
-    # --- THORACIC ---
-    'Diffuse interstitial opacities or ground-glass pattern (Viral/Atypical Pneumonia)': 
-        'Opacités interstitielles diffuses ou aspect en verre dépoli (Pneumonie Virale/Atypique)',
-    
-    'Focal alveolar consolidation with air bronchograms (Bacterial Pneumonia)': 
-        'Condensation alvéolaire focale avec bronchogrammes aériens (Pneumonie Bactérienne)',
-    
-    'Perfectly clear lungs, sharp costophrenic angles, no pathology': 
-        'Poumons parfaitement clairs, angles costophréniques nets, aucune pathologie',
-    
-    'Pneumothorax (Lung collapse)': 'Pneumothorax (Décollement de la plèvre)',
-    'Pleural Effusion (Fluid)': 'Épanchement Pleural (Liquide)',
-    'Cardiomegaly (Enlarged heart)': 'Cardiomégalie (Cœur élargi)',
-    'Pulmonary Edema': 'Œdème Pulmonaire',
-    'Lung Nodule or Mass': 'Nodule ou Masse Pulmonaire',
-    'Atelectasis (Lung collapse)': 'Atélectasie (Affaissement pulmonaire)',
-
-    # --- DERMATOLOGY ---
-    'A healthy skin area without lesion': 'Zone de peau saine sans lésion',
-    'A benign nevus (mole) regular, symmetrical and homogeneous': 'Nævus bénin (grain de beauté) régulier, symétrique et homogène',
-    'A seborrheic keratosis (benign warty lesion)': 'Kératose séborrhéique (lésion verruqueuse bénigne)',
-    'A malignant melanoma with asymmetry, irregular borders and multiple colors': 'Mélanome malin (Asymétrie, Bords irréguliers, Couleurs multiples)',
-    'A basal cell carcinoma (pearly or ulcerated lesion)': 'Carcinome basocellulaire (lésion perlée ou ulcérée)',
-    'A squamous cell carcinoma (crusty or budding lesion)': 'Carcinome épidermoïde (lésion croûteuse ou bourgeonnante)',
-    'A non-specific inflammatory skin lesion': 'Lésion cutanée inflammatoire non spécifique',
-
-    # --- ORTHOPEDICS ---
-    'Severe osteoarthritis with bone-on-bone contact and large osteophytes (Grade 4)': 'Arthrose sévère avec contact os-contre-os et ostéophytes importants (Grade 4)',
-    'Moderate osteoarthritis with definite joint space narrowing (Grade 2-3)': 'Arthrose modérée avec pincement articulaire net (Grade 2-3)',
-    'Normal knee joint with preserved joint space and no osteophytes (Grade 0-1)': 'Genou normal, interligne articulaire préservé (Grade 0-1)',
-    'Total knee arthroplasty (TKA) with metallic implant': 'Prothèse totale de genou (implant métallique)',
-    'Acute knee fracture or dislocation': 'Fracture ou luxation aiguë du genou',
-    'Other x-ray view (Chest, Hand, Foot, Pediatric) - OUT OF DISTRIBUTION': 'Autre vue radiographique (Hors périmètre)',
-    'A knee x-ray view (Knee Joint)': 'Radiographie du Genou'
-}
-
-def localize_result(result_json):
-    """
-    Traduit les résultats bruts (Anglais) en Français
-    en utilisant les dictionnaires de mapping.
-    """
-    # 1. Localiser le Domaine
-    domain_key = result_json['domain']['label']
-    if domain_key in DOMAIN_TRANSLATIONS:
-        result_json['domain']['label'] = DOMAIN_TRANSLATIONS[domain_key]['label']
-        result_json['domain']['description'] = DOMAIN_TRANSLATIONS[domain_key]['description']
-    
-    # 2. Localiser les Résultats Spécifiques
-    for item in result_json['specific']:
-        original_label = item['label']
-        if original_label in LABEL_TRANSLATIONS:
-            item['label'] = LABEL_TRANSLATIONS[original_label]
-        # Si pas de traduction trouvée, on garde l'anglais (fallback)
-        
-    return result_json
+# Mappings de localisation (Anglais -> Français)
+# Ce fichier permet de traduire les résultats de l'IA sans modifier les prompts originaux
+# qui doivent rester en anglais pour la performance du modèle.
+
+DOMAIN_TRANSLATIONS = {
+    'Thoracic': {
+        'label': 'Thoracique',
+        'description': 'Analyse Radiographique du Thorax'
+    },
+    'Dermatology': {
+        'label': 'Dermatologie',
+        'description': 'Analyse Dermatoscope des Lésions Cutanées'
+    },
+    'Histology': {
+        'label': 'Histologie',
+        'description': 'Analyse Microscopique (H&E)'
+    },
+    'Ophthalmology': {
+        'label': 'Ophtalmologie',
+        'description': 'Fond d\'Oeil (Rétine)'
+    },
+    'Orthopedics': {
+        'label': 'Orthopédie',
+        'description': 'Radiographie Osseuse'
+    }
+}
+
+LABEL_TRANSLATIONS = {
+    # --- THORACIC ---
+    'Diffuse interstitial opacities or ground-glass pattern (Viral/Atypical Pneumonia)': 
+        'Opacités interstitielles diffuses ou aspect en verre dépoli (Pneumonie Virale/Atypique)',
+    
+    'Focal alveolar consolidation with air bronchograms (Bacterial Pneumonia)': 
+        'Condensation alvéolaire focale avec bronchogrammes aériens (Pneumonie Bactérienne)',
+    
+    'Perfectly clear lungs, sharp costophrenic angles, no pathology': 
+        'Poumons parfaitement clairs, angles costophréniques nets, aucune pathologie',
+    
+    'Pneumothorax (Lung collapse)': 'Pneumothorax (Décollement de la plèvre)',
+    'Pleural Effusion (Fluid)': 'Épanchement Pleural (Liquide)',
+    'Cardiomegaly (Enlarged heart)': 'Cardiomégalie (Cœur élargi)',
+    'Pulmonary Edema': 'Œdème Pulmonaire',
+    'Lung Nodule or Mass': 'Nodule ou Masse Pulmonaire',
+    'Atelectasis (Lung collapse)': 'Atélectasie (Affaissement pulmonaire)',
+
+    # --- DERMATOLOGY ---
+    'A healthy skin area without lesion': 'Zone de peau saine sans lésion',
+    'A benign nevus (mole) regular, symmetrical and homogeneous': 'Nævus bénin (grain de beauté) régulier, symétrique et homogène',
+    'A seborrheic keratosis (benign warty lesion)': 'Kératose séborrhéique (lésion verruqueuse bénigne)',
+    'A malignant melanoma with asymmetry, irregular borders and multiple colors': 'Mélanome malin (Asymétrie, Bords irréguliers, Couleurs multiples)',
+    'A basal cell carcinoma (pearly or ulcerated lesion)': 'Carcinome basocellulaire (lésion perlée ou ulcérée)',
+    'A squamous cell carcinoma (crusty or budding lesion)': 'Carcinome épidermoïde (lésion croûteuse ou bourgeonnante)',
+    'A non-specific inflammatory skin lesion': 'Lésion cutanée inflammatoire non spécifique',
+
+    # --- ORTHOPEDICS ---
+    'Severe osteoarthritis with bone-on-bone contact and large osteophytes (Grade 4)': 'Arthrose sévère avec contact os-contre-os et ostéophytes importants (Grade 4)',
+    'Moderate osteoarthritis with definite joint space narrowing (Grade 2-3)': 'Arthrose modérée avec pincement articulaire net (Grade 2-3)',
+    'Normal knee joint with preserved joint space and no osteophytes (Grade 0-1)': 'Genou normal, interligne articulaire préservé (Grade 0-1)',
+    'Total knee arthroplasty (TKA) with metallic implant': 'Prothèse totale de genou (implant métallique)',
+    'Acute knee fracture or dislocation': 'Fracture ou luxation aiguë du genou',
+    'Other x-ray view (Chest, Hand, Foot, Pediatric) - OUT OF DISTRIBUTION': 'Autre vue radiographique (Hors périmètre)',
+    'A knee x-ray view (Knee Joint)': 'Radiographie du Genou'
+}
+
+def localize_result(result_json):
+    """
+    Traduit les résultats bruts (Anglais) en Français
+    en utilisant les dictionnaires de mapping.
+    """
+    # 1. Localiser le Domaine
+    domain_key = result_json['domain']['label']
+    if domain_key in DOMAIN_TRANSLATIONS:
+        result_json['domain']['label'] = DOMAIN_TRANSLATIONS[domain_key]['label']
+        result_json['domain']['description'] = DOMAIN_TRANSLATIONS[domain_key]['description']
+    
+    # 2. Localiser les Résultats Spécifiques
+    for item in result_json['specific']:
+        original_label = item['label']
+        if original_label in LABEL_TRANSLATIONS:
+            item['label'] = LABEL_TRANSLATIONS[original_label]
+        # Si pas de traduction trouvée, on garde l'anglais (fallback)
+        
+    return result_json

medical_labels.py

@@ -1,307 +0,0 @@
-from typing import Dict, List, Any
-
-# =========================================================================
-# CANONICAL MEDICAL DOMAINS CONFIGURATION (MODEL SOURCE OF TRUTH)
-# =========================================================================
-# - Prompts must be in ENGLISH (Model Language).
-# - Labels must have a stable 'id'.
-# - Logic Gates define structural/quality constraints.
-
-MEDICAL_DOMAINS = {
-    'Thoracic': {
-        'id': 'DOM_THORACIC',
-        'domain_prompt': 'Chest X-Ray Analysis',
-        'specific_labels': [
-            {'id': 'TH_PNEUMONIA_VIRAL', 'label_en': 'Diffuse interstitial opacities or ground-glass pattern (Viral/Atypical Pneumonia)'},
-            {'id': 'TH_PNEUMONIA_BACT', 'label_en': 'Focal alveolar consolidation with air bronchograms (Bacterial Pneumonia)'},
-            {'id': 'TH_NORMAL', 'label_en': 'Normal chest radiograph: normal cardiothoracic ratio, clear lungs, no pleural abnormality'}, 
-            {'id': 'TH_PNEUMOTHORAX', 'label_en': 'Pneumothorax (Lung collapse)'},
-            {'id': 'TH_PLEURAL_EFFUSION', 'label_en': 'Pleural Effusion (Fluid)'},
-            {'id': 'TH_CARDIOMEGALY_CLEAR', 'label_en': 'Cardiomegaly with clear lung fields (no pulmonary edema)'}, 
-            {'id': 'TH_CARDIOMEGALY_EDEMA', 'label_en': 'Cardiomegaly with pulmonary congestion or edema'},
-            {'id': 'TH_EDEMA', 'label_en': 'Pulmonary Edema (without cardiomegaly)'},
-            {'id': 'TH_NODULE', 'label_en': 'Lung Nodule or Mass'},
-            {'id': 'TH_ATELECTASIS', 'label_en': 'Atelectasis (Lung collapse)'}
-        ],
-        'logic_gate': {
-            'prompt': 'Evaluate cardiac silhouette size',
-            'labels': ['Normal cardiac size (CTR < 0.5)', 'Enlarged cardiac silhouette (Cardiomegaly)'],
-            'penalty_target': 'TH_NORMAL', # Penalize the ID of the normal label
-            'abnormal_index': 1
-        }
-    },
-    'Dermatology': {
-        'id': 'DOM_DERMATOLOGY',
-        'domain_prompt': 'Dermatoscopic analysis of a pigmented or non-pigmented skin lesion',
-        'specific_labels': [
-            {'id': 'DERM_NORMAL', 'label_en': 'Normal skin without visible lesion or abnormal pigmentation'},
-            {'id': 'DERM_NEVUS', 'label_en': 'Benign melanocytic nevus with symmetry and uniform pigmentation'},
-            {'id': 'DERM_SEBORRHEIC', 'label_en': 'Seborrheic keratosis (benign warty lesion)'},
-            {'id': 'DERM_MELANOMA', 'label_en': 'Malignant melanoma with asymmetry, irregular borders, and color variegation'},
-            {'id': 'DERM_BCC', 'label_en': 'Basal cell carcinoma (pearly or ulcerated lesion)'},
-            {'id': 'DERM_SCC', 'label_en': 'Squamous cell carcinoma (crusty or budding lesion)'},
-            {'id': 'DERM_INFLAMMATORY', 'label_en': 'Inflammatory skin lesion (Eczema, Psoriasis)'}
-        ],
-        'logic_gate': {
-            'prompt': 'Is there a visible skin lesion?',
-            'labels': ['No visible skin lesion', 'Visible skin lesion (pigmented or non-pigmented)'],
-            'penalty_target': 'ALL_PATHOLOGY', 
-            'abnormal_index': 0 
-        }
-    },
-    'Histology': {
-        'id': 'DOM_HISTOLOGY',
-        'domain_prompt': 'Microscopic analysis of a histological section (H&E stain)',
-        'specific_labels': [
-            {'id': 'HIST_HEALTHY_BREAST', 'label_en': 'Healthy breast tissue with preserved lobular architecture'},
-            {'id': 'HIST_HEALTHY_PROSTATE', 'label_en': 'Healthy prostatic tissue with regular glands'},
-            {'id': 'HIST_IDC_BREAST', 'label_en': 'Invasive ductal carcinoma (Disorganized cells)'},
-            {'id': 'HIST_ADENO_PROSTATE', 'label_en': 'Prostate adenocarcinoma (Gland fusion)'},
-            {'id': 'HIST_DYSPLASIA', 'label_en': 'Cervical dysplasia or intraepithelial neoplasia'},
-            {'id': 'HIST_COLON_CA', 'label_en': 'Colon cancer tumor tissue'},
-            {'id': 'HIST_LUNG_CA', 'label_en': 'Lung cancer tumor tissue'},
-            {'id': 'HIST_ADIPOSE', 'label_en': 'Adipose tissue (Fat) or connective stroma'},
-            {'id': 'HIST_ARTIFACT', 'label_en': 'Preparation artifact, empty area, or blurred region'} 
-        ],
-        'logic_gate': {
-            'prompt': 'Assess histological validity of the image',
-            'labels': ['Adequate H&E tissue section', 'Artifact, empty area, or blurred region'],
-            'penalty_target': 'ALL_DIAGNOSIS',
-            'abnormal_index': 1
-        }
-    },
-    'Ophthalmology': {
-        'id': 'DOM_OPHTHALMOLOGY',
-        'domain_prompt': 'Fundus photography (Retina)',
-        'specific_labels': [
-            {'id': 'OPH_NORMAL', 'label_en': 'Normal retina with visible optic disc and macula'},
-            {'id': 'OPH_DIABETIC', 'label_en': 'Diabetic retinopathy (hemorrhages, exudates)'},
-            {'id': 'OPH_GLAUCOMA', 'label_en': 'Glaucoma (optic disc cupping)'},
-            {'id': 'OPH_AMD', 'label_en': 'Macular degeneration (drusen or atrophy)'}
-        ],
-        'logic_gate': {
-            'prompt': 'Is the fundus image clinically interpretable?',
-            'labels': ['Good quality fundus image', 'Poor quality, uninterpretable or partial view'],
-            'penalty_target': 'ALL_DIAGNOSIS',
-            'abnormal_index': 1
-        }
-    },
-    'Orthopedics': {
-        'id': 'DOM_ORTHOPEDICS',
-        'domain_prompt': 'Bone X-Ray (Musculoskeletal)',
-        'stage_1_triage': {
-            'prompt': 'Anatomical region identification',
-            'labels': [
-                'Other x-ray view (Chest, Hand, Foot, Pediatric) - OUT OF DISTRIBUTION',
-                'A knee x-ray view (Knee Joint)'
-            ]
-        },
-        'specific_labels': [ 
-             {'id': 'ORTH_OA_SEVERE', 'label_en': 'Severe osteoarthritis (Grade 4)'}, 
-             {'id': 'ORTH_OA_MODERATE', 'label_en': 'Moderate osteoarthritis (Grade 2-3)'}, 
-             {'id': 'ORTH_NORMAL', 'label_en': 'Normal knee'}, 
-             {'id': 'ORTH_IMPLANT', 'label_en': 'Implant'}
-        ],
-        'stage_2_diagnosis': {
-            'prompt': 'Knee Osteoarthritis Severity Assessment',
-            'labels': [
-                {'id': 'ORTH_OA_SEVERE', 'label_en': 'Severe osteoarthritis with bone-on-bone contact (Grade 4)'},
-                {'id': 'ORTH_OA_MODERATE', 'label_en': 'Moderate osteoarthritis with definite joint space narrowing (Grade 2-3)'},
-                {'id': 'ORTH_NORMAL', 'label_en': 'Normal knee joint with preserved joint space (Grade 0-1)'},
-                {'id': 'ORTH_IMPLANT', 'label_en': 'Total knee arthroplasty (TKA) with metallic implant'}, 
-                {'id': 'ORTH_FRACTURE', 'label_en': 'Acute knee fracture or dislocation'}
-            ]
-        },
-        'logic_gate': {
-            'prompt': 'Is there a metallic implant?',
-            'labels': ['Native knee joint', 'Knee with metallic implant (Arthroplasty)'],
-            'penalty_target': 'ORTH_OA', # Logic target string match (Prefix)
-            'abnormal_index': 1
-        }
-    }
-}
-
-# =========================================================================
-# FRENCH TRANSLATIONS (USER INTERFACE ONLY)
-# =========================================================================
-# - Strict Mapping: ID -> {title, description}
-# - No dynamic translation allowed.
-
-LABEL_TRANSLATIONS_FR = {
-    # --- THORACIC ---
-    'TH_NORMAL': {
-        'short': 'Thorax sans anomalie',
-        'long': 'Silhouette cardiaque normale, poumons clairs, pas d’épanchement.',
-        'severity': 'low'
-    },
-    'TH_PNEUMONIA_VIRAL': {
-        'short': 'Pneumonie Virale / Atypique',
-        'long': 'Opacités interstitielles diffuses ou verre dépoli.',
-        'severity': 'high'
-    },
-    'TH_PNEUMONIA_BACT': {
-        'short': 'Pneumonie Bactérienne',
-        'long': 'Consolidation alvéolaire focale avec bronchogramme aérien.',
-        'severity': 'high'
-    },
-    'TH_PNEUMOTHORAX': {
-        'short': 'Pneumothorax',
-        'long': 'Présence possible d’air dans la cavité pleurale (collapsus).',
-        'severity': 'emergency'
-    },
-    'TH_PLEURAL_EFFUSION': {
-        'short': 'Épanchement Pleural',
-        'long': 'Accumulation de liquide dans l’espace pleural.',
-        'severity': 'medium'
-    },
-    'TH_CARDIOMEGALY_CLEAR': {  # UPDATED ID
-        'short': 'Cardiomégalie (Poumons clairs)',
-        'long': 'Silhouette cardiaque augmentée de taille sans signe d’œdème pulmonaire.',
-        'severity': 'medium'
-    },
-    'TH_CARDIOMEGALY_EDEMA': {
-        'short': 'Cardiomégalie avec Stase',
-        'long': 'Cœur augmenté de taille associé à une congestion pulmonaire.',
-        'severity': 'high'
-    },
-    'TH_EDEMA': {
-        'short': 'Œdème Pulmonaire',
-        'long': 'Surcharge liquidienne pulmonaire (sans cardiomégalie évidente).',
-        'severity': 'high'
-    },
-    'TH_NODULE': {
-        'short': 'Nodule ou Masse Pulmonaire',
-        'long': 'Lésion focale suspecte nécessitant un scanner de contrôle.',
-        'severity': 'high'
-    },
-    'TH_ATELECTASIS': {
-        'short': 'Atélectasie',
-        'long': 'Affaissement d’une partie du poumon.',
-        'severity': 'medium'
-    },
-
-    # --- DERMATOLOGY ---
-    'DERM_NORMAL': {
-        'short': 'Peau saine / Pas de lésion',
-        'long': 'Aucune lésion dermatologique suspecte visible.',
-        'severity': 'low'
-    },
-    'DERM_NEVUS': {
-        'short': 'Nævus Bénin (Grain de beauté)',
-        'long': 'Lésion régulière, symétrique et homogène.',
-        'severity': 'low'
-    },
-    'DERM_SEBORRHEIC': {
-        'short': 'Kératose Séborrhéique',
-        'long': 'Lésion bénigne fréquente ("verrue de vieillesse").',
-        'severity': 'low'
-    },
-    'DERM_MELANOMA': {
-        'short': 'Suspicion de Mélanome',
-        'long': 'Lésion pigmentée asymétrique, bords irréguliers (critères ABCDE). Urgence.',
-        'severity': 'emergency'
-    },
-    'DERM_BCC': {
-        'short': 'Carcinome Basocellulaire',
-        'long': 'Lésion perlée ou ulcérée suggérant un carcinome non-mélanique.',
-        'severity': 'high'
-    },
-    'DERM_SCC': {
-        'short': 'Carcinome Épidermoïde',
-        'long': 'Lésion croûteuse ou bourgeonnante suspecte.',
-        'severity': 'high'
-    },
-    'DERM_INFLAMMATORY': {
-        'short': 'Lésion Inflammatoire',
-        'long': 'Aspect compatible avec eczéma, psoriasis ou dermatite.',
-        'severity': 'medium'
-    },
-
-    # --- HISTOLOGY ---
-    'HIST_ARTIFACT': {
-        'short': 'Qualité Insuffisante (Artefact)',
-        'long': 'Tissu non interprétable (section vide, floue ou artefact technique).',
-        'severity': 'none'
-    },
-    'HIST_HEALTHY_BREAST': {
-        'short': 'Tissu Mammaire Sain',
-        'long': 'Architecture lobulaire préservée.',
-        'severity': 'low'
-    },
-    'HIST_IDC_BREAST': {
-        'short': 'Carcinome Canalaire Infiltrant',
-        'long': 'Prolifération cellulaire désorganisée invasive (Sein).',
-        'severity': 'high'
-    },
-    'HIST_HEALTHY_PROSTATE': {
-        'short': 'Tissu Prostatique Sain',
-        'long': 'Glandes régulières, stroma normal.',
-        'severity': 'low'
-    },
-    'HIST_ADENO_PROSTATE': {
-        'short': 'Adénocarcinome Prostatique',
-        'long': 'Fusion glandulaire et atypies cytonucléaires.',
-        'severity': 'high'
-    },
-    'HIST_COLON_CA': {'short': 'Cancer Colorectal', 'long': 'Tissu tumoral colique.', 'severity': 'high'},
-    'HIST_LUNG_CA': {'short': 'Cancer Pulmonaire', 'long': 'Tissu tumoral pulmonaire.', 'severity': 'high'},
-    'HIST_DYSPLASIA': {'short': 'Dysplasie / CIN', 'long': 'Anomalies précancéreuses.', 'severity': 'medium'},
-    'HIST_ADIPOSE': {'short': 'Tissu Adipeux / Stroma', 'long': 'Tissu de soutien normal.', 'severity': 'low'},
-
-    # --- OPHTHALMOLOGY ---
-    'OPH_NORMAL': {
-        'short': 'Fond d’œil Normal',
-        'long': 'Rétine, macula et papille d’aspect sain.',
-        'severity': 'low'
-    },
-    'OPH_DIABETIC': {
-        'short': 'Rétinopathie Diabétique',
-        'long': 'Présence d’hémorragies, exsudats ou anévrismes.',
-        'severity': 'high'
-    },
-    'OPH_GLAUCOMA': {
-        'short': 'Suspicion de Glaucome',
-        'long': 'Excavation papillaire (cup/disc ratio) augmentée.',
-        'severity': 'high'
-    },
-    'OPH_AMD': {
-        'short': 'DMLA',
-        'long': 'Dégénérescence Maculaire (drusens ou atrophie).',
-        'severity': 'medium'
-    },
-
-    # --- ORTHOPEDICS ---
-    'ORTH_NORMAL': {
-        'short': 'Genou Normal',
-        'long': 'Interligne articulaire préservé, pas d’ostéophyte.',
-        'severity': 'low'
-    },
-    'ORTH_OA_MODERATE': {
-        'short': 'Arthrose Modérée (Grade 2-3)',
-        'long': 'Pincement articulaire visible et ostéophytes.',
-        'severity': 'medium'
-    },
-    'ORTH_OA_SEVERE': {
-        'short': 'Arthrose Sévère (Grade 4)',
-        'long': 'Disparition de l’interligne (os sur os), déformation.',
-        'severity': 'high'
-    },
-    'ORTH_IMPLANT': {
-        'short': 'Prothèse Totale (PTG)',
-        'long': 'Genou avec implant métallique (Arthroplastie).',
-        'severity': 'low'
-    },
-    'ORTH_FRACTURE': {
-        'short': 'Fracture Récente / Luxation',
-        'long': 'Solution de continuité osseuse ou perte de congruence.',
-        'severity': 'emergency'
-    }
-}
-
-DOMAIN_TRANSLATIONS_FR = {
-    'Thoracic': 'Radiographie Thoracique',
-    'Dermatology': 'Dermatoscopie',
-    'Histology': 'Histopathologie (H&E)',
-    'Ophthalmology': 'Fond d’Oeil (Rétine)',
-    'Orthopedics': 'Radiographie Osseuse'
-}

quality_control.py

@@ -1,235 +0,0 @@
-
-import numpy as np
-import cv2
-import pydicom
-import logging
-from typing import Dict, Any, List, Tuple, Union
-from PIL import Image
-
-logger = logging.getLogger("ElephMind-QC")
-
-class QualityControlEngine:
-    """
-    Advanced Quality Control Engine (Gatekeeper).
-    Implements the 9-Point QC Checklist.
-    
-    Metrics:
-    1. Structural (DICOM)
-    2. Intensity (Contrast)
-    3. Blur (Laplacian)
-    4. Noise (SNR)
-    5. Saturation (Clipping)
-    6. Spatial (Aspect Ratio)
-    
-    Decision:
-    QC Score = Weighted Sum
-    Threshold >= 0.75 -> PASS
-    """
-    
-    def __init__(self):
-        # Weights defined by user
-        self.weights = {
-            "structure": 0.30, # Weight 3 (Normalized approx)
-            "blur": 0.20,      # Weight 2
-            "contrast": 0.20,  # Weight 2
-            "noise": 0.10,     # Weight 1
-            "saturation": 0.10,
-            "spatial": 0.10
-        }
-        # Thresholds
-        self.thresholds = {
-            "blur_var": 100.0,      # Laplacian Variance < 100 -> Blurry
-            "contrast_std": 10.0,   # Std Dev < 10 -> Low Contrast
-            "entropy": 4.0,         # Entropy < 4.0 -> Low Info
-            "snr_min": 2.0,         # Signal-to-Noise Ratio < 2.0 -> Noisy
-            "saturation_max": 0.05, # >5% pixels at min/max -> Saturated
-            "aspect_min": 0.5,      # Too thin
-            "aspect_max": 2.0       # Too wide
-        }
-
-    def evaluate_dicom(self, dataset: pydicom.dataset.FileDataset) -> Dict[str, Any]:
-        """
-        Gate 1: Structural DICOM Check.
-        """
-        reasons = []
-        passed = True
-        
-        try:
-            # 1. Pixel Data Presence
-            if not hasattr(dataset, "PixelData") or dataset.PixelData is None:
-                return {"passed": False, "score": 0.0, "reasons": ["CRITICAL: Missing PixelData"]}
-            
-            # 2. Dimensions
-            rows = getattr(dataset, "Rows", 0)
-            cols = getattr(dataset, "Columns", 0)
-            if rows <= 0 or cols <= 0:
-                return {"passed": False, "score": 0.0, "reasons": ["CRITICAL: Invalid Dimensions (Rows/Cols <= 0)"]}
-                
-            # 3. Transfer Syntax (Compression check - basic)
-            # If we can read pixel_array, it's usually mostly fine, preventing crash is handled in processor.
-            # Here we just check logical validity.
-            
-            pass
-        except Exception as e:
-             return {"passed": False, "score": 0.0, "reasons": [f"CRITICAL: DICOM Corrupt ({str(e)})"]}
-
-        return {"passed": True, "score": 1.0, "reasons": []}
-
-    def compute_metrics(self, image: np.ndarray) -> Dict[str, float]:
-        """
-        Compute raw metrics for the image (H, W) or (H, W, C).
-        Image input should be uint8 0-255 or float.
-        """
-        metrics = {}
-        
-        # Ensure Grayscale for calculation
-        if len(image.shape) == 3:
-            gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
-        else:
-            gray = image
-            
-        # 1. Blur (Variance of Laplacian)
-        metrics['blur_var'] = cv2.Laplacian(gray, cv2.CV_64F).var()
-        
-        # 2. Intensity / Contrast
-        metrics['std_dev'] = np.std(gray)
-        # Entropy
-        hist, _ = np.histogram(gray, bins=256, range=(0, 256))
-        prob = hist / (np.sum(hist) + 1e-8)
-        prob = prob[prob > 0]
-        metrics['entropy'] = -np.sum(prob * np.log2(prob))
-        
-        # 3. Noise (Simple SNR estimate)
-        # Signal = Mean, Noise = Std(High Pass)
-        # Simple High Pass: Image - Blurred
-        blurred = cv2.GaussianBlur(gray, (5, 5), 0)
-        noise_img = gray.astype(float) - blurred.astype(float)
-        noise_std = np.std(noise_img) + 1e-8
-        signal_mean = np.mean(gray)
-        metrics['snr'] = signal_mean / noise_std
-        
-        # 4. Saturation
-        # % pixels at 0 or 255
-        n_pixels = gray.size
-        n_sat = np.sum(gray <= 5) + np.sum(gray >= 250)
-        metrics['saturation_pct'] = n_sat / n_pixels
-        
-        # 5. Spatial
-        h, w = gray.shape
-        metrics['aspect_ratio'] = w / h
-        
-        return metrics
-
-    def run_quality_check(self, image_input: Union[Image.Image, np.ndarray, pydicom.dataset.FileDataset]) -> Dict[str, Any]:
-        """
-        Main Entry Point.
-        Returns: {
-            "passed": bool,
-            "quality_score": float (0-1),
-            "reasons": List[str],
-            "metrics": Dict
-        }
-        """
-        reasons = []
-        scores = {}
-        
-        # --- PHASE 1: DICOM STRUCTURE (If DICOM) ---
-        dicom_score = 1.0
-        if isinstance(image_input, pydicom.dataset.FileDataset):
-            res_struct = self.evaluate_dicom(image_input)
-            if not res_struct['passed']:
-                return {
-                    "passed": False, 
-                    "quality_score": 0.0, 
-                    "reasons": res_struct['reasons'], 
-                    "metrics": {} 
-                }
-            # Convert to numpy for image analysis using standard processor logic (simplified here or assume pre-converted)
-            # ideally the caller passes the converted image. 
-            # If input is DICOM, we assume we can't analyze image metrics easily here without converting.
-            # To simplify integration: Check DICOM Structure, then rely on caller to pass Image object for Visual QC.
-            # For this implementation, we assume input is PIL Image or Numpy Array for Visual QC.
-            pass
-
-        # Prepare Image
-        if isinstance(image_input, Image.Image):
-             img_np = np.array(image_input)
-        elif isinstance(image_input, np.ndarray):
-             img_np = image_input
-        else:
-             # If strictly DICOM passed without conversion capability, we only did struct check
-             return {"passed": True, "quality_score": 1.0, "reasons": [], "metrics": {}}
-
-        # --- PHASE 2: VISUAL METRICS ---
-        m = self.compute_metrics(img_np)
-        
-        # 1. Blur Check
-        # Sigmoid-like soft score or Hard Threshold? User implies Hard Rules composed into Score.
-        # "Structure: weight 3, Blur: weight 2..."
-        # Let's assign 0 or 1 per category based on threshold, then weight.
-        
-        # Blur
-        if m['blur_var'] < self.thresholds['blur_var']:
-            scores['blur'] = 0.0
-            reasons.append("Image Floue (Netteté insuffisante)")
-        else:
-            scores['blur'] = 1.0
-            
-        # Contrast / Intensity
-        if m['std_dev'] < self.thresholds['contrast_std'] or m['entropy'] < self.thresholds['entropy']:
-            scores['contrast'] = 0.0
-            reasons.append("Contraste Insuffisant (Image plate/sombre)")
-        else:
-            scores['contrast'] = 1.0
-            
-        # Noise
-        if m['snr'] < self.thresholds['snr_min']:
-            scores['noise'] = 0.0
-            reasons.append("Bruit Excessif (SNR faible)")
-        else:
-            scores['noise'] = 1.0
-            
-        # Saturation
-        if m['saturation_pct'] > self.thresholds['saturation_max']:
-            scores['saturation'] = 0.0
-            reasons.append("Saturation Excessive (>5% clipping)")
-        else:
-            scores['saturation'] = 1.0
-            
-        # Spatial
-        if not (self.thresholds['aspect_min'] <= m['aspect_ratio'] <= self.thresholds['aspect_max']):
-            scores['spatial'] = 0.0
-            reasons.append(f"Format Anatomique Invalide (Ratio {m['aspect_ratio']:.2f})")
-        else:
-             scores['spatial'] = 1.0
-             
-        # Structural (Implicitly 1 if we got here with an image)
-        scores['structure'] = 1.0 
-        
-        # --- PHASE 3: GLOBAL SCORE ---
-        # QC_score = Sum(w * s)
-        final_score = (
-            self.weights['structure'] * scores.get('structure', 1.0) +
-            self.weights['blur'] * scores.get('blur', 1.0) +
-            self.weights['contrast'] * scores.get('contrast', 1.0) +
-            self.weights['noise'] * scores.get('noise', 1.0) +
-            self.weights['saturation'] * scores.get('saturation', 1.0) +
-            self.weights['spatial'] * scores.get('spatial', 1.0)
-        )
-        
-        # Normalize weights sum just in case
-        total_weight = sum(self.weights.values())
-        final_score = final_score / total_weight
-        
-        # DECISION
-        is_passed = final_score >= 0.75
-        
-        status = "PASSED" if is_passed else "REJECTED"
-        logger.info(f"QC Evaluation: {status} (Score: {final_score:.2f}) - Reasons: {reasons}")
-        
-        return {
-            "passed": is_passed,
-            "quality_score": round(final_score, 2),
-            "reasons": reasons,
-            "metrics": m
-        }

requirements.txt

@@ -1,21 +1,21 @@
-fastapi
-uvicorn
-python-multipart
-requests
-transformers
-torch
-Pillow
-sentencepiece
-pydicom
-numpy
-grad-cam
-python-jose[cryptography]
-passlib
-argon2-cffi
-bcrypt==4.0.1
-cryptography
-python-dotenv
-opencv-python
-python-swiftclient
-protobuf
-huggingface_hub
+fastapi
+uvicorn
+python-multipart
+requests
+transformers
+torch
+Pillow
+sentencepiece
+pydicom
+numpy
+grad-cam
+python-jose[cryptography]
+passlib
+argon2-cffi
+bcrypt==4.0.1
+cryptography
+python-dotenv
+opencv-python
+python-swiftclient
+protobuf
+huggingface_hub

scripts/README.md

@@ -0,0 +1,25 @@
+# ElephMind Utility Scripts
+
+This directory contains maintenance and debug scripts for the ElephMind backend.
+
+## How to Run
+
+Because these scripts import modules from the parent `server/` directory, you must run them with the parent directory in your `PYTHONPATH`.
+
+**Windows (PowerShell):**
+```powershell
+$env:PYTHONPATH=".."; python init_admin.py
+```
+
+**Linux/Mac:**
+```bash
+PYTHONPATH=.. python init_admin.py
+```
+
+## Available Scripts
+
+-   **`init_admin.py`**: Creates the initial 'admin' user with secure password hashing.
+-   **`verify_admin.py`**: Checks if the admin user exists in the database.
+-   **`test_auth.py`**: Unit tests for the authentication logic.
+-   **`debug_inference.py`**: Tests the ML model with a dummy image.
+-   **`inspect_model.py`**: Prints details about the loaded PyTorch model.

scripts/debug_inference.py

@@ -0,0 +1,71 @@
+import torch
+from transformers import AutoProcessor, AutoModel
+import numpy as np
+from PIL import Image, ImageDraw
+
+# Configuration
+MODEL_DIR = r"D:\oeil d'elephant"
+
+def test_inference():
+    print(f"Loading model from {MODEL_DIR}...")
+    try:
+        model = AutoModel.from_pretrained(MODEL_DIR, local_files_only=True)
+        processor = AutoProcessor.from_pretrained(MODEL_DIR, local_files_only=True)
+        model.eval()
+        
+        if hasattr(model, 'logit_scale'):
+            with torch.no_grad():
+                model.logit_scale.data.fill_(4.60517) # exp(4.6) = 100
+                
+        print("Model loaded.")
+    except Exception as e:
+        print(f"Failed to load model: {e}")
+        return
+
+    # Synthetic Chest X-ray
+    image = Image.new('RGB', (448, 448), color=(0, 0, 0))
+    draw = ImageDraw.Draw(image)
+    draw.ellipse([100, 100, 200, 350], fill=(200, 200, 200))
+    draw.ellipse([248, 100, 348, 350], fill=(200, 200, 200)) # Lungs
+    
+    # Simple Prompts Hypothesis
+    prompts = [
+        'Os', 
+        'Poumons', 
+        'Peau', 
+        'Oeil', 
+        'Sein', 
+        'Tissu'
+    ]
+    
+    # Also test slightly descriptive
+    prompts_v2 = [
+        'Radiographie Os',
+        'Radiographie Poumons',
+        'Photo Peau',
+        'Fond d\'oeil',
+        'Mammographie Sein',
+        'Microscope Tissu'
+    ]
+    
+    print("\nTesting Simple Prompts on Synthetic Chest X-ray:")
+    
+    for p_set in [prompts, prompts_v2]:
+        with torch.no_grad():
+            inputs = processor(text=p_set, images=image, padding="max_length", return_tensors="pt")
+            outputs = model(**inputs)
+            logits = outputs.logits_per_image
+            probs = torch.sigmoid(logits)[0]
+            
+            # Also calculate Softmax
+            probs_softmax = torch.softmax(logits, dim=1)[0]
+            
+            for i, prompt in enumerate(p_set):
+                l = logits[0][i].item()
+                p_sig = probs[i].item()
+                p_soft = probs_softmax[i].item()
+                print(f"Prompt: '{prompt:<20}' | Logit: {l:.4f} | Sigmoid: {p_sig*100:.6f}% | Softmax: {p_soft*100:.2f}%")
+        print("-" * 60)
+
+if __name__ == "__main__":
+    test_inference()

scripts/debug_pathology.py

@@ -0,0 +1,70 @@
+import torch
+from transformers import AutoProcessor, AutoModel
+import numpy as np
+from PIL import Image, ImageDraw
+
+# Configuration
+MODEL_DIR = r"D:\oeil d'elephant"
+
+def test_inference():
+    print(f"Loading model from {MODEL_DIR}...")
+    try:
+        model = AutoModel.from_pretrained(MODEL_DIR, local_files_only=True)
+        processor = AutoProcessor.from_pretrained(MODEL_DIR, local_files_only=True)
+        model.eval()
+        
+        # Apply fix
+        if hasattr(model, 'logit_scale'):
+            with torch.no_grad():
+                model.logit_scale.data.fill_(4.60517)
+                
+        print("Model loaded.")
+    except Exception as e:
+        print(f"Failed to load model: {e}")
+        return
+
+    # Synthetic Pneumonia X-ray
+    # Two lungs, one with a big white consolidation
+    image = Image.new('RGB', (448, 448), color=(0, 0, 0))
+    draw = ImageDraw.Draw(image)
+    draw.ellipse([100, 100, 200, 350], fill=(100, 100, 100)) # Left lung (clearer)
+    draw.ellipse([248, 100, 348, 350], fill=(200, 200, 200)) # Right lung (consolidated/white)
+    
+    # Check "Thoracic" specific labels
+    labels = [
+        'Cardiomédiastin élargi', 'Cardiomégalie', 'Opacité pulmonaire',
+        'Lésion pulmonaire', 'Consolidation', 'Œdème', 'Pneumonie',
+        'Atelectasis', 'Pneumothorax', 'Effusion pleurale', 'Pleural Autre'
+    ]
+    
+    # Try simplified versions too
+    simple_labels = [
+        'Coeur', 'Gros coeur', 'Opacité',
+        'Lésion', 'Blanc', 'Eau', 'Infection',
+        'Ecrasé', 'Air', 'Liquide', 'Autre'
+    ]
+    
+    print("\nTesting Pathology Prompts:")
+    
+    with torch.no_grad():
+        inputs = processor(text=labels, images=image, padding="max_length", return_tensors="pt")
+        outputs = model(**inputs)
+        logits = outputs.logits_per_image
+        probs = torch.sigmoid(logits)[0]
+        
+        print("\nOriginal Labels:")
+        for i, label in enumerate(labels):
+            print(f"'{label}': Logit {logits[0][i]:.4f} | Prob {probs[i]:.6f}")
+
+        # Test Simple
+        inputs_simple = processor(text=simple_labels, images=image, padding="max_length", return_tensors="pt")
+        outputs_simple = model(**inputs_simple)
+        logits_simple = outputs_simple.logits_per_image
+        probs_simple = torch.sigmoid(logits_simple)[0]
+
+        print("\nSimple Labels:")
+        for i, label in enumerate(simple_labels):
+            print(f"'{label}': Logit {logits_simple[0][i]:.4f} | Prob {probs_simple[0][i]:.6f}")
+
+if __name__ == "__main__":
+    test_inference()

scripts/init_admin.py

@@ -0,0 +1,27 @@
+import sys
+import os
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+import database
+from main import get_password_hash
+
+def create_admin():
+    database.init_db()
+    if database.get_user_by_username("admin"):
+        print("Admin already exists.")
+        return
+
+    admin_data = {
+        "username": "admin",
+        "hashed_password": get_password_hash("password123"),
+        "email": "admin@elephmind.com",
+        "security_question": "Quel est votre animal totem ?",
+        "security_answer": get_password_hash("elephant")
+    }
+    
+    if database.create_user(admin_data):
+        print("Admin user created successfully. (Login: admin / password123)")
+    else:
+        print("Failed to create admin user.")
+
+if __name__ == "__main__":
+    create_admin()

scripts/inspect_model.py

@@ -0,0 +1,29 @@
+import json
+import os
+
+MODEL_DIR = r"D:\oeil d'elephant"
+
+def inspect():
+    files = ["config.json", "preprocessor_config.json", "tokenizer_config.json"]
+    
+    for f in files:
+        path = os.path.join(MODEL_DIR, f)
+        print(f"\n--- {f} ---")
+        if os.path.exists(path):
+            try:
+                with open(path, 'r', encoding='utf-8') as file:
+                    content = json.load(file)
+                    # Print summary to avoid huge output
+                    if f == "config.json":
+                        print(json.dumps({k:v for k,v in content.items() if k in ['architectures', 'model_type', 'logit_scale_init_value', 'vision_config', 'text_config']}, indent=2))
+                    elif f == "preprocessor_config.json":
+                        print(json.dumps(content, indent=2))
+                    else:
+                        print(json.dumps(content, indent=2))
+            except Exception as e:
+                print(f"Error reading {f}: {e}")
+        else:
+            print("File not found.")
+
+if __name__ == "__main__":
+    inspect()

scripts/list_patients.py

@@ -0,0 +1,31 @@
+
+# Script to verify patients in DB
+
+import sqlite3
+import os
+
+DB_NAME = "elephmind.db"
+if os.path.exists('/data/elephmind.db'):
+    DB_NAME = '/data/elephmind.db'
+
+def list_patients():
+    if not os.path.exists(DB_NAME):
+        print(f"Database {DB_NAME} not found.")
+        return
+
+    conn = sqlite3.connect(DB_NAME)
+    conn.row_factory = sqlite3.Row
+    c = conn.cursor()
+    try:
+        c.execute("SELECT * FROM patients")
+        rows = c.fetchall()
+        print(f"Found {len(rows)} patients.")
+        for row in rows:
+            print(dict(row))
+    except Exception as e:
+        print(f"Error: {e}")
+    finally:
+        conn.close()
+
+if __name__ == "__main__":
+    list_patients()

scripts/test_auth.py

@@ -0,0 +1,63 @@
+import requests
+import sys
+
+BASE_URL = "http://127.0.0.1:8022"
+
+def test_health():
+    print(f"Testing Health Check at {BASE_URL}/health...")
+    try:
+        r = requests.get(f"{BASE_URL}/health")
+        if r.status_code == 200:
+            print("✅ Health Check Passed")
+            return True
+    except Exception as e:
+        print(f"❌ Health Check Failed: {e}")
+    return False
+
+def test_auth():
+    print("Testing Authentication...")
+    
+    # 1. Try to access protected route without token
+    try:
+        r = requests.post(f"{BASE_URL}/analyze")
+        if r.status_code == 401:
+            print("✅ Protected Endpoint correctly rejected unauthorized request (401)")
+        else:
+            print(f"❌ Protected Endpoint Failed: Expected 401, got {r.status_code}")
+            return False
+            
+        # 2. Login to get token
+        payload = {"username": "admin", "password": "secret"}
+        r = requests.post(f"{BASE_URL}/token", data=payload)
+        if r.status_code == 200:
+            token = r.json().get("access_token")
+            if token:
+                print("✅ Login Successful. Token received.")
+            else:
+                print("❌ Login Failed: No token in response")
+                return False
+        else:
+            print(f"❌ Login Failed: {r.status_code} - {r.text}")
+            return False
+            
+        # 3. Access protected route WITH token (Should fail on 422 Validation 'Field required' for file, NOT 401)
+        headers = {"Authorization": f"Bearer {token}"}
+        # We don't send file, expecting 422 Unprocessable Entity (Missing File), which means Auth passed!
+        r = requests.post(f"{BASE_URL}/analyze", headers=headers)
+        if r.status_code == 422:
+             print("✅ Protected Endpoint correctly accepted token (Got 422 for missing file, not 401)")
+             return True
+        elif r.status_code == 401:
+             print("❌ Protected Endpoint rejected valid token (401)")
+             return False
+        else:
+             print(f"⚠️ Unexpected status with token: {r.status_code}")
+             return True # Acceptable for now
+             
+    except Exception as e:
+        print(f"❌ Test Exception: {e}")
+        return False
+
+if __name__ == "__main__":
+    if test_health():
+        test_auth()

scripts/verify_admin.py

@@ -0,0 +1,30 @@
+
+
+import sys
+import os
+
+# Add server directory to path to import database
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+import database
+
+def check_admin():
+    print(f"Checking database: {database.DB_NAME}")
+    
+    # Initialize DB if tables missing (which seems to be the case in this context)
+    database.init_db()
+    
+    try:
+        user = database.get_user_by_username("admin")
+        if user:
+            print("USER 'admin' FOUND.")
+            print(f"   ID: {user['id']}")
+            print(f"   Email: {user['email']}")
+        else:
+            print("USER 'admin' NOT FOUND.")
+    except Exception as e:
+        print(f"Error querying database: {e}")
+
+if __name__ == "__main__":
+    check_admin()
+

secret.key

@@ -0,0 +1 @@
+6cfBgfzHb12RD2eW_9QxGrpoDdScGYoqpV3MYvz96LE=

storage.py

@@ -1,92 +1,92 @@
-import os
-import abc
-from datetime import datetime
-
-class StorageProvider(abc.ABC):
-    @abc.abstractmethod
-    def save_file(self, file_bytes: bytes, filename: str) -> str:
-        pass
-
-    @abc.abstractmethod
-    def get_file(self, filename: str) -> bytes:
-        pass
-
-class LocalStorage(StorageProvider):
-    def __init__(self, base_dir="data_storage"):
-        self.base_dir = base_dir
-        os.makedirs(base_dir, exist_ok=True)
-
-    def save_file(self, file_bytes: bytes, filename: str) -> str:
-        # Prepend timestamp to avoid collision
-        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
-        safe_name = f"{ts}_{filename}"
-        path = os.path.join(self.base_dir, safe_name)
-        with open(path, "wb") as f:
-            f.write(file_bytes)
-        return path
-
-    def get_file(self, filename: str) -> bytes:
-        path = os.path.join(self.base_dir, filename)
-        if not os.path.exists(path):
-            return None
-        with open(path, "rb") as f:
-            return f.read()
-
-class SwiftStorage(StorageProvider):
-    """
-    OpenStack Swift Storage Provider.
-    Requires python-swiftclient installed.
-    """
-    def __init__(self, auth_url, username, password, project_name, container_name="elephmind_images"):
-        # Import here to avoid error on Windows if not installed
-        try:
-             from swiftclient import Connection
-        except ImportError:
-             raise ImportError("python-swiftclient not installed!")
-             
-        self.container_name = container_name
-        self.conn = Connection(
-            authurl=auth_url,
-            user=username,
-            key=password,
-            tenant_name=project_name,
-            auth_version='3',
-            os_options={'user_domain_name': 'Default', 'project_domain_name': 'Default'}
-        )
-        # Ensure container exists
-        try:
-            self.conn.put_container(self.container_name)
-        except Exception as e:
-            print(f"Swift Connection Error: {e}")
-
-    def save_file(self, file_bytes: bytes, filename: str) -> str:
-        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
-        safe_name = f"{ts}_{filename}"
-        self.conn.put_object(
-            self.container_name, 
-            safe_name, 
-            contents=file_bytes, 
-            content_type='application/octet-stream'
-        )
-        return f"swift://{self.container_name}/{safe_name}"
-
-    def get_file(self, filename: str) -> bytes:
-        # filename could be safe_name
-        # logic to extract key if needed
-        try:
-             _, obj = self.conn.get_object(self.container_name, filename)
-             return obj
-        except Exception:
-             return None
-
-# Factory
-def get_storage_provider(config_mode="LOCAL"):
-    if config_mode == "OPENSTACK":
-        return SwiftStorage(
-            auth_url=os.getenv("OS_AUTH_URL"),
-            username=os.getenv("OS_USERNAME"),
-            password=os.getenv("OS_PASSWORD"),
-            project_name=os.getenv("OS_PROJECT_NAME")
-        )
-    else:
-        return LocalStorage()
+import os
+import abc
+from datetime import datetime
+
+class StorageProvider(abc.ABC):
+    @abc.abstractmethod
+    def save_file(self, file_bytes: bytes, filename: str) -> str:
+        pass
+
+    @abc.abstractmethod
+    def get_file(self, filename: str) -> bytes:
+        pass
+
+class LocalStorage(StorageProvider):
+    def __init__(self, base_dir="data_storage"):
+        self.base_dir = base_dir
+        os.makedirs(base_dir, exist_ok=True)
+
+    def save_file(self, file_bytes: bytes, filename: str) -> str:
+        # Prepend timestamp to avoid collision
+        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+        safe_name = f"{ts}_{filename}"
+        path = os.path.join(self.base_dir, safe_name)
+        with open(path, "wb") as f:
+            f.write(file_bytes)
+        return path
+
+    def get_file(self, filename: str) -> bytes:
+        path = os.path.join(self.base_dir, filename)
+        if not os.path.exists(path):
+            return None
+        with open(path, "rb") as f:
+            return f.read()
+
+class SwiftStorage(StorageProvider):
+    """
+    OpenStack Swift Storage Provider.
+    Requires python-swiftclient installed.
+    """
+    def __init__(self, auth_url, username, password, project_name, container_name="elephmind_images"):
+        # Import here to avoid error on Windows if not installed
+        try:
+             from swiftclient import Connection
+        except ImportError:
+             raise ImportError("python-swiftclient not installed!")
+             
+        self.container_name = container_name
+        self.conn = Connection(
+            authurl=auth_url,
+            user=username,
+            key=password,
+            tenant_name=project_name,
+            auth_version='3',
+            os_options={'user_domain_name': 'Default', 'project_domain_name': 'Default'}
+        )
+        # Ensure container exists
+        try:
+            self.conn.put_container(self.container_name)
+        except Exception as e:
+            print(f"Swift Connection Error: {e}")
+
+    def save_file(self, file_bytes: bytes, filename: str) -> str:
+        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+        safe_name = f"{ts}_{filename}"
+        self.conn.put_object(
+            self.container_name, 
+            safe_name, 
+            contents=file_bytes, 
+            content_type='application/octet-stream'
+        )
+        return f"swift://{self.container_name}/{safe_name}"
+
+    def get_file(self, filename: str) -> bytes:
+        # filename could be safe_name
+        # logic to extract key if needed
+        try:
+             _, obj = self.conn.get_object(self.container_name, filename)
+             return obj
+        except Exception:
+             return None
+
+# Factory
+def get_storage_provider(config_mode="LOCAL"):
+    if config_mode == "OPENSTACK":
+        return SwiftStorage(
+            auth_url=os.getenv("OS_AUTH_URL"),
+            username=os.getenv("OS_USERNAME"),
+            password=os.getenv("OS_PASSWORD"),
+            project_name=os.getenv("OS_PROJECT_NAME")
+        )
+    else:
+        return LocalStorage()

storage_manager.py

@@ -1,85 +1,85 @@
-import os
-import uuid
-import logging
-from pathlib import Path
-from typing import Tuple, Optional
-
-# Configure Logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-# Detect environment (Hugging Face Spaces vs Local)
-# HF Spaces with persistent storage usually mount at /data
-IS_HF_SPACE = os.path.exists('/data')
-if IS_HF_SPACE:
-    BASE_STORAGE_DIR = Path('/data/storage')
-    logger.info(f"Using PERSISTENT storage at {BASE_STORAGE_DIR}")
-else:
-    BASE_STORAGE_DIR = Path(os.path.dirname(os.path.abspath(__file__))) / "storage"
-    logger.info(f"Using LOCAL storage at {BASE_STORAGE_DIR}")
-
-def get_user_storage_path(username: str) -> Path:
-    """Get secure storage path for user, creating it if needed."""
-    # Sanitize username to prevent directory traversal
-    safe_username = "".join([c for c in username if c.isalnum() or c in ('-', '_')])
-    user_path = BASE_STORAGE_DIR / safe_username
-    user_path.mkdir(parents=True, exist_ok=True)
-    return user_path
-
-def save_image(username: str, file_bytes: bytes, filename_hint: str = "image.png") -> str:
-    """
-    Save image to disk and return a unique image_id.
-    Returns: image_id (e.g. IMG_ABC123)
-    """
-    # Generate ID
-    unique_suffix = uuid.uuid4().hex[:12].upper()
-    image_id = f"IMG_{unique_suffix}"
-    
-    # Determine extension
-    ext = os.path.splitext(filename_hint)[1].lower()
-    if not ext:
-        ext = ".png" # Default
-        
-    filename = f"{image_id}{ext}"
-    user_path = get_user_storage_path(username)
-    file_path = user_path / filename
-    
-    try:
-        with open(file_path, "wb") as f:
-            f.write(file_bytes)
-        logger.info(f"Saved image {image_id} for user {username} at {file_path}")
-        return image_id
-    except Exception as e:
-        logger.error(f"Failed to save image: {e}")
-        raise IOError(f"Storage Error: {e}")
-
-def load_image(username: str, image_id: str) -> Tuple[bytes, str]:
-    """
-    Load image bytes from disk.
-    Returns: (file_bytes, file_path_str)
-    """
-    # Security: Ensure ID format is valid
-    if not image_id.startswith("IMG_") or ".." in image_id or "/" in image_id:
-        raise ValueError("Invalid image_id format")
-        
-    user_path = get_user_storage_path(username)
-    
-    # We don't know the extension, so look for the file
-    # Or strict requirement: user must know? 
-    # Better: Search for matching file
-    for file in user_path.glob(f"{image_id}.*"):
-        try:
-            with open(file, "rb") as f:
-                return f.read(), str(file)
-        except Exception as e:
-            logger.error(f"Error reading file {file}: {e}")
-            raise IOError("Read error")
-            
-    raise FileNotFoundError(f"Image {image_id} not found for user {username}")
-
-def get_image_absolute_path(username: str, image_id: str) -> Optional[str]:
-    """Return absolute path if exists, else None."""
-    user_path = get_user_storage_path(username)
-    for file in user_path.glob(f"{image_id}.*"):
-        return str(file)
-    return None
+import os
+import uuid
+import logging
+from pathlib import Path
+from typing import Tuple, Optional
+
+# Configure Logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Detect environment (Hugging Face Spaces vs Local)
+# HF Spaces with persistent storage usually mount at /data
+IS_HF_SPACE = os.path.exists('/data')
+if IS_HF_SPACE:
+    BASE_STORAGE_DIR = Path('/data/storage')
+    logger.info(f"Using PERSISTENT storage at {BASE_STORAGE_DIR}")
+else:
+    BASE_STORAGE_DIR = Path(os.path.dirname(os.path.abspath(__file__))) / "storage"
+    logger.info(f"Using LOCAL storage at {BASE_STORAGE_DIR}")
+
+def get_user_storage_path(username: str) -> Path:
+    """Get secure storage path for user, creating it if needed."""
+    # Sanitize username to prevent directory traversal
+    safe_username = "".join([c for c in username if c.isalnum() or c in ('-', '_')])
+    user_path = BASE_STORAGE_DIR / safe_username
+    user_path.mkdir(parents=True, exist_ok=True)
+    return user_path
+
+def save_image(username: str, file_bytes: bytes, filename_hint: str = "image.png") -> str:
+    """
+    Save image to disk and return a unique image_id.
+    Returns: image_id (e.g. IMG_ABC123)
+    """
+    # Generate ID
+    unique_suffix = uuid.uuid4().hex[:12].upper()
+    image_id = f"IMG_{unique_suffix}"
+    
+    # Determine extension
+    ext = os.path.splitext(filename_hint)[1].lower()
+    if not ext:
+        ext = ".png" # Default
+        
+    filename = f"{image_id}{ext}"
+    user_path = get_user_storage_path(username)
+    file_path = user_path / filename
+    
+    try:
+        with open(file_path, "wb") as f:
+            f.write(file_bytes)
+        logger.info(f"Saved image {image_id} for user {username} at {file_path}")
+        return image_id
+    except Exception as e:
+        logger.error(f"Failed to save image: {e}")
+        raise IOError(f"Storage Error: {e}")
+
+def load_image(username: str, image_id: str) -> Tuple[bytes, str]:
+    """
+    Load image bytes from disk.
+    Returns: (file_bytes, file_path_str)
+    """
+    # Security: Ensure ID format is valid
+    if not image_id.startswith("IMG_") or ".." in image_id or "/" in image_id:
+        raise ValueError("Invalid image_id format")
+        
+    user_path = get_user_storage_path(username)
+    
+    # We don't know the extension, so look for the file
+    # Or strict requirement: user must know? 
+    # Better: Search for matching file
+    for file in user_path.glob(f"{image_id}.*"):
+        try:
+            with open(file, "rb") as f:
+                return f.read(), str(file)
+        except Exception as e:
+            logger.error(f"Error reading file {file}: {e}")
+            raise IOError("Read error")
+            
+    raise FileNotFoundError(f"Image {image_id} not found for user {username}")
+
+def get_image_absolute_path(username: str, image_id: str) -> Optional[str]:
+    """Return absolute path if exists, else None."""
+    user_path = get_user_storage_path(username)
+    for file in user_path.glob(f"{image_id}.*"):
+        return str(file)
+    return None

upload_model.py

@@ -0,0 +1,21 @@
+# upload_model.py - Upload model to Hugging Face Hub
+from huggingface_hub import upload_folder
+import os
+
+model_path = os.path.join("models", "oeil d'elephant")
+print(f"Uploading from: {model_path}")
+print(f"Path exists: {os.path.exists(model_path)}")
+
+if os.path.exists(model_path):
+    print("Starting upload... (this may take a while for 3.5GB)")
+    upload_folder(
+        folder_path=model_path,
+        repo_id="issoufzousko07/medsigclip-model",
+        repo_type="model"
+    )
+    print("Upload complete!")
+else:
+    print(f"ERROR: Path not found: {model_path}")
+    print("Available in models/:")
+    if os.path.exists("models"):
+        print(os.listdir("models"))

upload_space.py

@@ -0,0 +1,21 @@
+# upload_space.py - Upload code to HuggingFace Space (excluding large model files)
+from huggingface_hub import upload_folder
+import os
+
+print("Uploading ElephMind API to HuggingFace Space...")
+print("(Model will be downloaded from Hub at runtime)")
+
+# Determine the server directory (where this script lives)
+server_dir = os.path.dirname(os.path.abspath(__file__))
+
+print(f"Uploading directory: {server_dir}")
+
+upload_folder(
+    folder_path=server_dir,
+    repo_id="issoufzousko07/elephmind-api",
+    repo_type="space",
+    ignore_patterns=["models/*", "*.pyc", "__pycache__", "*.db", "storage/*", "data_storage/*", ".env", "venv", ".git", ".idea"]
+)
+
+print("[OK] Upload complete!")
+print("Your Space should start building at: https://huggingface.co/spaces/issoufzousko07/elephmind-api")