device token

api/routes/txagent.py

@@ -43,7 +43,7 @@ except ImportError as e:
     TXAGENT_AVAILABLE = False
 
 try:
-    from utils import clean_text_response
+    from utils import clean_text_response, format_risk_level, create_notification
 except ImportError:
     # Fallback: define the function locally if import fails
     def clean_text_response(text: str) -> str:
@@ -51,6 +51,21 @@ except ImportError:
         text = re.sub(r'\n\s*\n', '\n\n', text)
         text = re.sub(r'[ ]+', ' ', text)
         return text.replace("**", "").replace("__", "").strip()
+    
+    def format_risk_level(risk_level: str) -> str:
+        risk_level_mapping = {
+            'low': 'low', 'medium': 'moderate', 'moderate': 'moderate', 
+            'high': 'high', 'severe': 'severe', 'critical': 'severe',
+            'none': 'none', 'unknown': 'none'
+        }
+        return risk_level_mapping.get(risk_level.lower(), 'none')
+    
+    def create_notification(user_id: str, title: str, message: str, notification_type: str = "info", patient_id: str = None) -> dict:
+        return {
+            "user_id": user_id, "title": title, "message": message,
+            "type": notification_type, "read": False,
+            "timestamp": datetime.utcnow(), "patient_id": patient_id
+        }
 
 try:
     from analysis import analyze_patient_report
@@ -84,17 +99,7 @@ txagent_instance = None
 
 def _normalize_risk_level(risk_level):
     """Normalize risk level names to match expected format"""
-    risk_level_mapping = {
-        'low': 'low',
-        'medium': 'moderate',
-        'moderate': 'moderate', 
-        'high': 'high',
-        'severe': 'severe',
-        'critical': 'severe',
-        'none': 'none',
-        'unknown': 'none'
-    }
-    return risk_level_mapping.get(risk_level.lower(), 'none')
+    return format_risk_level(risk_level)
 
 def get_txagent():
     """Get or create TxAgent instance"""
@@ -777,4 +782,285 @@ async def synthesize_voice(
         logger.error(f"Error in voice synthesis: {e}")
         raise HTTPException(status_code=500, detail="Error generating voice output")
 
+# Notifications endpoints
+@router.get("/notifications")
+async def get_notifications(current_user: dict = Depends(get_current_user)):
+    """Get notifications for the current user"""
+    try:
+        if not any(role in current_user.get('roles', []) for role in ['doctor', 'admin']):
+            raise HTTPException(status_code=403, detail="Only doctors and admins can access notifications")
+        
+        logger.info(f"Fetching notifications for {current_user['email']}")
+        
+        # Import database collections
+        from db.mongo import db
+        notifications_collection = db.notifications
+        
+        # Get notifications for the current user
+        notifications = await notifications_collection.find({
+            "user_id": current_user.get('_id')
+        }).sort("timestamp", -1).limit(50).to_list(length=50)
+        
+        return [
+            {
+                "id": str(notification["_id"]),
+                "title": notification.get("title", ""),
+                "message": notification.get("message", ""),
+                "type": notification.get("type", "info"),
+                "read": notification.get("read", False),
+                "timestamp": notification.get("timestamp"),
+                "patient_id": notification.get("patient_id")
+            }
+            for notification in notifications
+        ]
+        
+    except Exception as e:
+        logger.error(f"Error getting notifications: {e}")
+        raise HTTPException(status_code=500, detail="Failed to get notifications")
+
+@router.post("/notifications/{notification_id}/read")
+async def mark_notification_read(
+    notification_id: str,
+    current_user: dict = Depends(get_current_user)
+):
+    """Mark a notification as read"""
+    try:
+        if not any(role in current_user.get('roles', []) for role in ['doctor', 'admin']):
+            raise HTTPException(status_code=403, detail="Only doctors and admins can mark notifications as read")
+        
+        logger.info(f"Marking notification {notification_id} as read by {current_user['email']}")
+        
+        # Import database collections
+        from db.mongo import db
+        notifications_collection = db.notifications
+        
+        # Update the notification
+        result = await notifications_collection.update_one(
+            {
+                "_id": ObjectId(notification_id),
+                "user_id": current_user.get('_id')
+            },
+            {"$set": {"read": True, "read_at": datetime.utcnow()}}
+        )
+        
+        if result.matched_count == 0:
+            raise HTTPException(status_code=404, detail="Notification not found")
+        
+        return {"message": "Notification marked as read"}
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error marking notification as read: {e}")
+        raise HTTPException(status_code=500, detail="Failed to mark notification as read")
+
+@router.post("/notifications/read-all")
+async def mark_all_notifications_read(current_user: dict = Depends(get_current_user)):
+    """Mark all notifications as read for the current user"""
+    try:
+        if not any(role in current_user.get('roles', []) for role in ['doctor', 'admin']):
+            raise HTTPException(status_code=403, detail="Only doctors and admins can mark notifications as read")
+        
+        logger.info(f"Marking all notifications as read for {current_user['email']}")
+        
+        # Import database collections
+        from db.mongo import db
+        notifications_collection = db.notifications
+        
+        # Update all unread notifications for the user
+        result = await notifications_collection.update_many(
+            {
+                "user_id": current_user.get('_id'),
+                "read": False
+            },
+            {"$set": {"read": True, "read_at": datetime.utcnow()}}
+        )
+        
+        return {
+            "message": f"Marked {result.modified_count} notifications as read",
+            "modified_count": result.modified_count
+        }
+        
+    except Exception as e:
+        logger.error(f"Error marking all notifications as read: {e}")
+        raise HTTPException(status_code=500, detail="Failed to mark notifications as read")
+
+# Voice chat endpoint
+@router.post("/voice/chat")
+async def voice_chat(
+    audio: UploadFile = File(...),
+    current_user: dict = Depends(get_current_user)
+):
+    """Voice chat with TxAgent"""
+    try:
+        if not any(role in current_user.get('roles', []) for role in ['doctor', 'admin']):
+            raise HTTPException(status_code=403, detail="Only doctors and admins can use voice features")
+        
+        logger.info(f"Voice chat initiated by {current_user['email']}")
+        
+        # Read audio file
+        audio_data = await audio.read()
+        
+        # Transcribe audio to text
+        try:
+            transcription = recognize_speech(audio_data)
+            if isinstance(transcription, dict):
+                transcription_text = transcription.get("transcription", "")
+            else:
+                transcription_text = str(transcription)
+        except Exception as e:
+            logger.error(f"Speech recognition failed: {e}")
+            transcription_text = "Sorry, I couldn't understand the audio."
+        
+        # Generate response (for now, a simple response)
+        response_text = f"I heard you say: '{transcription_text}'. How can I help you with patient care today?"
+        
+        # Store voice chat in the database
+        try:
+            from db.mongo import db
+            chats_collection = db.chats
+            
+            chat_entry = {
+                "message": transcription_text,
+                "response": response_text,
+                "user_id": current_user.get('_id'),
+                "user_email": current_user.get('email'),
+                "timestamp": datetime.utcnow(),
+                "chat_type": "voice_chat"
+            }
+            
+            await chats_collection.insert_one(chat_entry)
+            logger.info(f"Voice chat stored in database for user {current_user['email']}")
+            
+        except Exception as db_error:
+            logger.error(f"Failed to store voice chat in database: {str(db_error)}")
+        
+        # Convert response to speech
+        try:
+            audio_response = text_to_speech(response_text, language="en")
+        except Exception as e:
+            logger.error(f"Text-to-speech failed: {e}")
+            audio_response = b"Sorry, I couldn't generate audio response."
+        
+        return StreamingResponse(
+            io.BytesIO(audio_response),
+            media_type="audio/mpeg",
+            headers={"Content-Disposition": "attachment; filename=voice_response.mp3"}
+        )
+        
+    except Exception as e:
+        logger.error(f"Error in voice chat: {e}")
+        raise HTTPException(status_code=500, detail="Error processing voice chat")
+
+# Report analysis endpoint
+@router.post("/analyze-report")
+async def analyze_report(
+    file: UploadFile = File(...),
+    current_user: dict = Depends(get_current_user)
+):
+    """Analyze uploaded report (PDF, DOCX, etc.)"""
+    try:
+        if not any(role in current_user.get('roles', []) for role in ['doctor', 'admin']):
+            raise HTTPException(status_code=403, detail="Only doctors and admins can analyze reports")
+        
+        logger.info(f"Report analysis initiated by {current_user['email']}")
+        
+        # Read file content
+        file_content = await file.read()
+        file_extension = file.filename.split('.')[-1].lower()
+        
+        # Extract text based on file type
+        if file_extension == 'pdf':
+            try:
+                text_content = extract_text_from_pdf(file_content)
+            except Exception as e:
+                logger.error(f"PDF text extraction failed: {e}")
+                text_content = "Failed to extract text from PDF"
+        elif file_extension in ['docx', 'doc']:
+            try:
+                if Document:
+                    doc = Document(io.BytesIO(file_content))
+                    text_content = '\n'.join([paragraph.text for paragraph in doc.paragraphs])
+                else:
+                    text_content = "Document processing not available"
+            except Exception as e:
+                logger.error(f"DOCX text extraction failed: {e}")
+                text_content = "Failed to extract text from document"
+        else:
+            text_content = "Unsupported file format"
+        
+        # Analyze the content (for now, return a simple analysis)
+        analysis_result = {
+            "file_name": file.filename,
+            "file_type": file_extension,
+            "extracted_text": text_content[:500] + "..." if len(text_content) > 500 else text_content,
+            "analysis": {
+                "summary": f"Analyzed {file.filename} containing {len(text_content)} characters",
+                "key_findings": ["Sample finding 1", "Sample finding 2"],
+                "recommendations": ["Sample recommendation 1", "Sample recommendation 2"]
+            },
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+        return analysis_result
+        
+    except Exception as e:
+        logger.error(f"Error analyzing report: {e}")
+        raise HTTPException(status_code=500, detail="Error analyzing report")
 
+# Patient deletion endpoint
+@router.delete("/patients/{patient_id}")
+async def delete_patient(
+    patient_id: str,
+    current_user: dict = Depends(get_current_user)
+):
+    """Delete a patient and all associated data"""
+    try:
+        if not any(role in current_user.get('roles', []) for role in ['admin']):
+            raise HTTPException(status_code=403, detail="Only administrators can delete patients")
+        
+        logger.info(f"Patient deletion initiated by {current_user['email']} for patient {patient_id}")
+        
+        # Import database collections
+        from db.mongo import db
+        patients_collection = db.patients
+        analysis_collection = db.patient_analysis_results
+        chats_collection = db.chats
+        notifications_collection = db.notifications
+        
+        # Find the patient first
+        patient = await patients_collection.find_one({"fhir_id": patient_id})
+        if not patient:
+            raise HTTPException(status_code=404, detail="Patient not found")
+        
+        # Delete all associated data
+        try:
+            # Delete patient
+            await patients_collection.delete_one({"fhir_id": patient_id})
+            
+            # Delete analysis results
+            await analysis_collection.delete_many({"patient_id": patient_id})
+            
+            # Delete chats related to this patient
+            await chats_collection.delete_many({"patient_id": patient_id})
+            
+            # Delete notifications related to this patient
+            await notifications_collection.delete_many({"patient_id": patient_id})
+            
+            logger.info(f"Successfully deleted patient {patient_id} and all associated data")
+            
+            return {
+                "message": f"Patient {patient.get('full_name', patient_id)} and all associated data deleted successfully",
+                "patient_id": patient_id,
+                "deleted_at": datetime.utcnow().isoformat()
+            }
+            
+        except Exception as e:
+            logger.error(f"Error during patient deletion: {e}")
+            raise HTTPException(status_code=500, detail="Error deleting patient data")
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error deleting patient {patient_id}: {e}")
+        raise HTTPException(status_code=500, detail="Failed to delete patient")

utils.py

@@ -6,60 +6,17 @@ from datetime import datetime
 from typing import Dict, List, Tuple
 from bson import ObjectId
 import logging
-from config import logger
-# Add to your utils.py
-from fastapi import WebSocket
-import asyncio
-
-class NotificationManager:
-    def __init__(self):
-        self.active_connections = {}
-        self.notification_queue = asyncio.Queue()
-
-    async def connect(self, websocket: WebSocket, user_id: str):
-        await websocket.accept()
-        self.active_connections[user_id] = websocket
-
-    def disconnect(self, user_id: str):
-        if user_id in self.active_connections:
-            del self.active_connections[user_id]
-
-    async def broadcast_notification(self, notification: dict):
-        """Broadcast to all connected clients"""
-        for connection in self.active_connections.values():
-            try:
-                await connection.send_json({
-                    "type": "notification",
-                    "data": notification
-                })
-            except Exception as e:
-                logger.error(f"Error sending notification: {e}")
-
-notification_manager = NotificationManager()
-
-async def broadcast_notification(notification: dict):
-    """Broadcast notification to relevant users"""
-    # Determine recipients based on notification type/priority
-    recipients = []
-    if notification["priority"] == "high":
-        recipients = ["psychiatrist", "emergency_team", "primary_care"]
-    else:
-        recipients = ["primary_care", "case_manager"]
-    
-    # Add to each recipient's notification queue
-    await notification_manager.notification_queue.put({
-        "recipients": recipients,
-        "notification": notification
-    })
-
 
+logger = logging.getLogger(__name__)
 
 def clean_text_response(text: str) -> str:
+    """Clean and format text response"""
     text = re.sub(r'\n\s*\n', '\n\n', text)
     text = re.sub(r'[ ]+', ' ', text)
     return text.replace("**", "").replace("__", "").strip()
 
 def extract_section(text: str, heading: str) -> str:
+    """Extract a section from text based on heading"""
     try:
         pattern = rf"{re.escape(heading)}:\s*\n(.*?)(?=\n[A-Z][^\n]*:|\Z)"
         match = re.search(pattern, text, re.DOTALL | re.IGNORECASE)
@@ -69,6 +26,7 @@ def extract_section(text: str, heading: str) -> str:
         return ""
 
 def structure_medical_response(text: str) -> Dict:
+    """Structure medical response into sections"""
     def extract_improved(text: str, heading: str) -> str:
         patterns = [
             rf"{re.escape(heading)}:\s*\n(.*?)(?=\n\s*\n|\Z)",
@@ -97,14 +55,43 @@ def structure_medical_response(text: str) -> Dict:
     }
 
 def serialize_patient(patient: dict) -> dict:
+    """Serialize patient data for JSON response"""
     patient_copy = patient.copy()
     if "_id" in patient_copy:
         patient_copy["_id"] = str(patient_copy["_id"])
     return patient_copy
 
 def compute_patient_data_hash(data: dict) -> str:
+    """Compute hash of patient data for change detection"""
     serialized = json.dumps(data, sort_keys=True)
     return hashlib.sha256(serialized.encode()).hexdigest()
 
 def compute_file_content_hash(file_content: bytes) -> str:
-    return hashlib.sha256(file_content).hexdigest()
+    """Compute hash of file content"""
+    return hashlib.sha256(file_content).hexdigest()
+
+def create_notification(user_id: str, title: str, message: str, notification_type: str = "info", patient_id: str = None) -> dict:
+    """Create a notification object"""
+    return {
+        "user_id": user_id,
+        "title": title,
+        "message": message,
+        "type": notification_type,
+        "read": False,
+        "timestamp": datetime.utcnow(),
+        "patient_id": patient_id
+    }
+
+def format_risk_level(risk_level: str) -> str:
+    """Normalize risk level names"""
+    risk_level_mapping = {
+        'low': 'low',
+        'medium': 'moderate',
+        'moderate': 'moderate', 
+        'high': 'high',
+        'severe': 'severe',
+        'critical': 'severe',
+        'none': 'none',
+        'unknown': 'none'
+    }
+    return risk_level_mapping.get(risk_level.lower(), 'none')