Spaces:
Running
Running
File size: 5,299 Bytes
8f682c2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
# api_clients/rxnorm_client.py
"""
Client for the NLM RxNorm API.
This module is essential for the Drug Interaction & Safety Analyzer. It provides
functionality to standardize drug names to RxCUIs and then checks for
potential interactions between a list of those drugs.
"""
import asyncio
import aiohttp
from .config import RXNORM_BASE_URL, REQUEST_HEADERS
async def get_rxcui(session: aiohttp.ClientSession, drug_name: str) -> str | None:
"""
Converts a drug name string into its corresponding RxNorm Concept Unique Identifier (RxCUI).
An RxCUI is a standardized code necessary for accurately checking interactions.
Args:
session (aiohttp.ClientSession): The active HTTP session.
drug_name (str): The name of the drug (e.g., "Lipitor", "atorvastatin").
Returns:
str | None: The RxCUI as a string if found, otherwise None.
"""
if not drug_name:
return None
url = f"{RXNORM_BASE_URL}/rxcui.json"
params = {'name': drug_name, 'search': 1} # search=1 for exact match first
try:
async with session.get(url, params=params, headers=REQUEST_HEADERS, timeout=10) as resp:
resp.raise_for_status()
data = await resp.json()
# The RxCUI is nested within the 'idGroup'
id_group = data.get('idGroup', {})
if id_group and 'rxnormId' in id_group:
return id_group['rxnormId'][0] # Return the first and most likely CUI
return None
except aiohttp.ClientError as e:
print(f"Error fetching RxCUI for '{drug_name}': {e}")
return None
async def get_interactions(session: aiohttp.ClientSession, list_of_rxcuis: list[str]) -> list[dict]:
"""
Checks for interactions among a list of RxCUIs.
Args:
session (aiohttp.ClientSession): The active HTTP session.
list_of_rxcuis (list[str]): A list of standardized drug RxCUI strings.
Returns:
list[dict]: A list of interaction pairs, each with a description and severity.
Returns an empty list if no interactions are found or an error occurs.
"""
if not list_of_rxcuis or len(list_of_rxcuis) < 2:
return [] # Interactions require at least two drugs
# The API expects a space-separated string of RxCUIs
rxcuis_str = " ".join(list_of_rxcuis)
url = f"{RXNORM_BASE_URL}/interaction/list.json"
params = {'rxcuis': rxcuis_str}
try:
async with session.get(url, params=params, headers=REQUEST_HEADERS, timeout=15) as resp:
resp.raise_for_status()
data = await resp.json()
interaction_groups = data.get('fullInteractionTypeGroup', [])
parsed_interactions = []
# The API response is deeply nested, so we parse it carefully
for group in interaction_groups:
for interaction_type in group.get('fullInteractionType', []):
for pair in interaction_type.get('interactionPair', []):
# Extract the critical information into a clean format
description = pair.get('description', 'No description available.')
severity = pair.get('severity', 'Severity unknown')
# Identify the two drugs involved in this specific interaction
drug1_name = pair['interactionConcept'][0]['minConceptItem']['name']
drug2_name = pair['interactionConcept'][1]['minConceptItem']['name']
parsed_interactions.append({
"pair": f"{drug1_name} / {drug2_name}",
"severity": severity,
"description": description
})
return parsed_interactions
except aiohttp.ClientError as e:
print(f"Error fetching interactions for RxCUIs '{rxcuis_str}': {e}")
return []
async def run_interaction_check(drug_names: list[str]) -> list[dict]:
"""
High-level orchestrator for a complete drug interaction check.
This function handles the full workflow:
1. Takes a list of human-readable drug names.
2. Concurrently converts them all to RxCUIs.
3. Feeds the valid RxCUIs into the interaction checker.
Args:
drug_names (list[str]): A list of drug names from user input.
Returns:
list[dict]: A list of found interactions, ready for display or AI synthesis.
"""
async with aiohttp.ClientSession() as session:
# Step 1: Concurrently get RxCUIs for all provided drug names
rxcui_tasks = [get_rxcui(session, name) for name in drug_names]
resolved_rxcuis = await asyncio.gather(*rxcui_tasks)
# Step 2: Filter out any drugs that were not found (returned None)
valid_rxcuis = [rxcui for rxcui in resolved_rxcuis if rxcui]
if len(valid_rxcuis) < 2:
print("Fewer than two valid drugs found, cannot check for interactions.")
return []
# Step 3: Check for interactions using the list of valid RxCUIs
interactions = await get_interactions(session, valid_rxcuis)
return interactions |