"""
Analyze FHIR DeviceRequest to extract FDA-relevant device information.
This is the entry point for FDA verification workflows.
"""
from typing import Dict, List, Any, Optional
from pydantic import BaseModel, Field
from core.app import mcp
from fastmcp import Context
import re
class DeviceIdentifiers(BaseModel):
"""FDA-relevant device identifiers."""
fda_k_number: Optional[str] = Field(default=None, description="FDA 510(k) clearance number")
fda_pma_number: Optional[str] = Field(default=None, description="FDA PMA approval number")
udi: Optional[str] = Field(default=None, description="Unique Device Identifier")
product_code: Optional[str] = Field(default=None, description="FDA product classification code")
class DeviceInfo(BaseModel):
"""Complete device information extracted from FHIR DeviceRequest."""
device_name: str = Field(description="Name/description of the device")
manufacturer: Optional[str] = Field(default=None, description="Device manufacturer")
model_number: Optional[str] = Field(default=None, description="Model or catalog number")
identifiers: DeviceIdentifiers = Field(
default_factory=DeviceIdentifiers,
description="FDA-relevant identifiers"
)
device_type: Optional[str] = Field(default=None, description="Device type/category")
is_otc: bool = Field(default=False, description="Whether device is over-the-counter")
codes: List[Dict[str, str]] = Field(
default_factory=list,
description="Medical coding (SNOMED, CPT, etc.)"
)
extracted_from: Dict[str, Any] = Field(
default_factory=dict,
description="Source locations in DeviceRequest"
)
confidence: str = Field(
default="low",
description="Confidence in extraction: high/medium/low"
)
@mcp.tool
async def analyze_device_request(
device_request: Dict[str, Any],
ctx: Context = None
) -> Dict[str, Any]:
"""
Extract FDA-relevant device information from a FHIR DeviceRequest.
This tool is the starting point for FDA verification. It parses a FHIR
DeviceRequest to identify all device details needed for FDA database
searches, including names, manufacturers, FDA numbers, and codes.
Args:
device_request: FHIR DeviceRequest resource as dictionary
ctx: FastMCP context for logging
Returns:
Dictionary with all extracted device details, identifiers,
and confidence level in the extraction
"""
if ctx:
await ctx.info("Analyzing DeviceRequest for FDA information")
# Initialize device info with default name
device_info = DeviceInfo(device_name="Unknown Device")
# Extract device name from codeCodeableConcept
if "codeCodeableConcept" in device_request:
code = device_request["codeCodeableConcept"]
device_info.device_name = code.get("text", "Unknown Device")
device_info.extracted_from["device_name"] = "codeCodeableConcept.text"
# Extract medical codes
for coding in code.get("coding", []):
device_info.codes.append({
"system": coding.get("system", ""),
"code": coding.get("code", ""),
"display": coding.get("display", "")
})
# Process parameters for FDA information
for param in device_request.get("parameter", []):
param_code = _extract_parameter_code(param)
if param_code == "fda-510k":
k_number = _extract_parameter_value(param, "text")
if k_number:
device_info.identifiers.fda_k_number = _normalize_k_number(k_number)
device_info.extracted_from["k_number"] = "parameter[fda-510k]"
device_info.confidence = "high"
if ctx:
await ctx.info(f"Found K-number: {device_info.identifiers.fda_k_number}")
elif param_code == "fda-pma":
pma_number = _extract_parameter_value(param, "text")
if pma_number:
device_info.identifiers.fda_pma_number = _normalize_pma_number(pma_number)
device_info.extracted_from["pma_number"] = "parameter[fda-pma]"
device_info.confidence = "high"
elif param_code == "manufacturer":
manufacturer = _extract_parameter_value(param, "string")
if manufacturer:
device_info.manufacturer = manufacturer
device_info.extracted_from["manufacturer"] = "parameter[manufacturer]"
elif param_code == "model":
model = _extract_parameter_value(param, "string")
if model:
device_info.model_number = model
device_info.extracted_from["model"] = "parameter[model]"
elif param_code == "otc-status":
is_otc = param.get("valueBoolean", False)
device_info.is_otc = is_otc
device_info.extracted_from["otc_status"] = "parameter[otc-status]"
elif param_code == "product-code":
product_code = _extract_parameter_value(param, "text")
if product_code:
device_info.identifiers.product_code = product_code
device_info.extracted_from["product_code"] = "parameter[product-code]"
# Extract manufacturer from performer if not in parameters
if not device_info.manufacturer and "performer" in device_request:
performer = device_request["performer"]
if isinstance(performer, dict):
device_info.manufacturer = performer.get("display", "")
device_info.extracted_from["manufacturer"] = "performer.display"
# Parse notes for additional FDA information
notes_text = _combine_notes(device_request.get("note", []))
if notes_text:
# Look for K-numbers in notes
if not device_info.identifiers.fda_k_number:
k_matches = re.findall(r'\bK\d{6}\b', notes_text, re.IGNORECASE)
if k_matches:
device_info.identifiers.fda_k_number = k_matches[0].upper()
device_info.extracted_from["k_number"] = "note (regex extraction)"
device_info.confidence = "medium" if device_info.confidence == "low" else device_info.confidence
if ctx:
await ctx.info(f"Extracted K-number from notes: {device_info.identifiers.fda_k_number}")
# Look for PMA numbers in notes
if not device_info.identifiers.fda_pma_number:
pma_matches = re.findall(r'\bP\d{6}\b', notes_text, re.IGNORECASE)
if pma_matches:
device_info.identifiers.fda_pma_number = pma_matches[0].upper()
device_info.extracted_from["pma_number"] = "note (regex extraction)"
device_info.confidence = "medium" if device_info.confidence == "low" else device_info.confidence
# Look for product codes (3-letter FDA codes)
if not device_info.identifiers.product_code:
# Common hearing aid product codes
for code in ["QUH", "OSM", "DQA", "ESD"]:
if code in notes_text.upper():
device_info.identifiers.product_code = code
device_info.extracted_from["product_code"] = "note (known code match)"
break
# Check for OTC mentions
if "otc" in notes_text.lower() or "over-the-counter" in notes_text.lower():
device_info.is_otc = True
if "otc_status" not in device_info.extracted_from:
device_info.extracted_from["otc_status"] = "note (keyword match)"
# Extract manufacturer from notes if not found
if not device_info.manufacturer:
manufacturer = _extract_manufacturer_from_text(notes_text)
if manufacturer:
device_info.manufacturer = manufacturer
device_info.extracted_from["manufacturer"] = "note (pattern match)"
# Determine device type from name and context
device_info.device_type = _classify_device_type(device_info.device_name, device_info.is_otc)
# Set final confidence based on what we found
if device_info.identifiers.fda_k_number or device_info.identifiers.fda_pma_number:
device_info.confidence = "high"
elif device_info.manufacturer and device_info.model_number:
device_info.confidence = "medium"
elif device_info.device_name != "Unknown Device":
device_info.confidence = "low"
if ctx:
await ctx.info(f"Extraction complete: {device_info.device_name} ({device_info.confidence} confidence)")
return device_info.model_dump()
def _extract_parameter_code(param: Dict[str, Any]) -> Optional[str]:
"""Extract parameter code from FHIR parameter."""
code = param.get("code", {})
codings = code.get("coding", [])
if codings:
return codings[0].get("code")
return None
def _extract_parameter_value(param: Dict[str, Any], value_type: str) -> Optional[str]:
"""Extract value from FHIR parameter based on type."""
if value_type == "text":
# For CodeableConcept values
value_cc = param.get("valueCodeableConcept", {})
return value_cc.get("text")
elif value_type == "string":
return param.get("valueString")
elif value_type == "boolean":
return param.get("valueBoolean")
return None
def _normalize_k_number(k_number: str) -> str:
"""Normalize FDA K-number format."""
# Remove any non-alphanumeric characters
cleaned = re.sub(r'[^A-Z0-9]', '', k_number.upper())
# Ensure it starts with K and has 6 digits
if cleaned.startswith("K") and len(cleaned) >= 7:
return cleaned[:7]
# Try to extract just the numeric part and add K
numbers = re.findall(r'\d{6}', cleaned)
if numbers:
return f"K{numbers[0]}"
return k_number.upper()
def _normalize_pma_number(pma_number: str) -> str:
"""Normalize FDA PMA number format."""
cleaned = re.sub(r'[^A-Z0-9]', '', pma_number.upper())
if cleaned.startswith("P") and len(cleaned) >= 7:
return cleaned[:7]
numbers = re.findall(r'\d{6}', cleaned)
if numbers:
return f"P{numbers[0]}"
return pma_number.upper()
def _combine_notes(notes: List[Dict[str, Any]]) -> str:
"""Combine all notes into single text."""
texts = []
for note in notes:
if "text" in note:
texts.append(note["text"])
return " ".join(texts)
def _extract_manufacturer_from_text(text: str) -> Optional[str]:
"""Try to extract manufacturer name from text."""
# Common hearing aid manufacturers
manufacturers = [
"GN Hearing", "Bose", "Sony", "Eargo", "Audicus", "Lively",
"Phonak", "Oticon", "Widex", "Signia", "Starkey", "ReSound",
"hearX", "Lexie", "Jabra", "WS Audiology", "Sonova", "Demant"
]
text_lower = text.lower()
for manufacturer in manufacturers:
if manufacturer.lower() in text_lower:
return manufacturer
return None
def _classify_device_type(device_name: str, is_otc: bool) -> str:
"""Classify device type based on name and OTC status."""
name_lower = device_name.lower()
if "hearing aid" in name_lower:
if is_otc:
return "OTC Hearing Aid"
else:
return "Prescription Hearing Aid"
elif "cochlear" in name_lower:
return "Cochlear Implant"
elif "amplifier" in name_lower or "psap" in name_lower:
return "Personal Sound Amplifier"
elif "processor" in name_lower:
return "Sound Processor"
else:
return "Medical Device"
