File size: 5,678 Bytes
85020ae 793d027 | 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 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 | """Antibiotic query tools for AMR-Guard workflow."""
from typing import Optional
from src.db.database import execute_query
def query_antibiotic_info(
antibiotic_name: str,
include_category: bool = True,
include_formulations: bool = True
) -> list[dict]:
"""
Query EML antibiotic database for classification and details.
Args:
antibiotic_name: Name of the antibiotic (partial match supported)
include_category: Include WHO stewardship category
include_formulations: Include available formulations
Returns:
List of matching antibiotics with details
Used by: Agent 1, Agent 4
"""
query = """
SELECT
medicine_name,
who_category,
eml_section,
formulations,
indication,
atc_codes,
combined_with,
status
FROM eml_antibiotics
WHERE LOWER(medicine_name) LIKE LOWER(?)
ORDER BY
CASE who_category
WHEN 'ACCESS' THEN 1
WHEN 'WATCH' THEN 2
WHEN 'RESERVE' THEN 3
ELSE 4
END
"""
results = execute_query(query, (f"%{antibiotic_name}%",))
# Filter columns based on parameters
if not include_category or not include_formulations:
filtered_results = []
for r in results:
filtered = dict(r)
if not include_category:
filtered.pop('who_category', None)
if not include_formulations:
filtered.pop('formulations', None)
filtered_results.append(filtered)
return filtered_results
return results
def get_antibiotics_by_category(category: str) -> list[dict]:
"""
Get all antibiotics in a specific WHO category.
Args:
category: WHO category ('ACCESS', 'WATCH', 'RESERVE')
Returns:
List of antibiotics in that category
"""
query = """
SELECT medicine_name, indication, formulations, atc_codes
FROM eml_antibiotics
WHERE UPPER(who_category) = UPPER(?)
ORDER BY medicine_name
"""
return execute_query(query, (category,))
def get_antibiotic_for_indication(indication_keyword: str) -> list[dict]:
"""
Find antibiotics based on indication keywords.
Args:
indication_keyword: Keyword to search in indications
Returns:
List of matching antibiotics with indications
"""
query = """
SELECT
medicine_name,
who_category,
indication,
formulations
FROM eml_antibiotics
WHERE LOWER(indication) LIKE LOWER(?)
ORDER BY
CASE who_category
WHEN 'ACCESS' THEN 1
WHEN 'WATCH' THEN 2
WHEN 'RESERVE' THEN 3
ELSE 4
END
"""
return execute_query(query, (f"%{indication_keyword}%",))
def interpret_mic_value(
pathogen: str,
antibiotic: str,
mic_value: float,
route: str = None
) -> dict:
"""
Interpret MIC value against EUCAST breakpoints.
Args:
pathogen: Pathogen name or group
antibiotic: Antibiotic name
mic_value: MIC value in mg/L
route: Administration route (IV, Oral)
Returns:
Dict with interpretation (S/I/R), breakpoint values, clinical notes
Used by: Agent 2, Agent 3
"""
query = """
SELECT
pathogen_group,
antibiotic,
mic_susceptible,
mic_resistant,
notes,
route
FROM mic_breakpoints
WHERE LOWER(pathogen_group) LIKE LOWER(?)
AND LOWER(antibiotic) LIKE LOWER(?)
LIMIT 1
"""
results = execute_query(query, (f"%{pathogen}%", f"%{antibiotic}%"))
if not results:
return {
"interpretation": "UNKNOWN",
"message": f"No breakpoint found for {antibiotic} against {pathogen}",
"mic_value": mic_value,
"breakpoints": None
}
bp = results[0]
mic_s = bp.get('mic_susceptible')
mic_r = bp.get('mic_resistant')
# Determine interpretation
if mic_s is not None and mic_value <= mic_s:
interpretation = "SUSCEPTIBLE"
message = f"MIC ({mic_value} mg/L) ≤ S breakpoint ({mic_s} mg/L)"
elif mic_r is not None and mic_value > mic_r:
interpretation = "RESISTANT"
message = f"MIC ({mic_value} mg/L) > R breakpoint ({mic_r} mg/L)"
elif mic_s is not None and mic_r is not None:
interpretation = "INTERMEDIATE"
message = f"MIC ({mic_value} mg/L) between S ({mic_s}) and R ({mic_r}) breakpoints"
else:
interpretation = "UNKNOWN"
message = "Incomplete breakpoint data"
return {
"interpretation": interpretation,
"message": message,
"mic_value": mic_value,
"breakpoints": {
"susceptible": mic_s,
"resistant": mic_r
},
"pathogen_group": bp.get('pathogen_group'),
"notes": bp.get('notes')
}
def get_breakpoints_for_pathogen(pathogen: str) -> list[dict]:
"""
Get all available breakpoints for a pathogen.
Args:
pathogen: Pathogen name or group
Returns:
List of antibiotic breakpoints for the pathogen
"""
query = """
SELECT
antibiotic,
mic_susceptible,
mic_resistant,
route,
notes
FROM mic_breakpoints
WHERE LOWER(pathogen_group) LIKE LOWER(?)
ORDER BY antibiotic
"""
return execute_query(query, (f"%{pathogen}%",))
|