radiology-api / app /post_processor.py
MakPr016
Updated response and returning proper summary
ed38d82
NEGATIVE_TERMS = ['clear', 'normal', 'unremarkable', 'no', 'negative', 'absent', 'within normal limits', 'stable']
POSITIVE_TERMS = ['opacity', 'infiltrate', 'consolidation', 'fracture', 'abnormal', 'effusion', 'edema', 'pneumothorax', 'cardiomegaly', 'mass', 'nodule', 'atelectasis']
CRITICAL_TERMS = ['acute', 'severe', 'large', 'extensive', 'pneumothorax', 'fracture', 'mass', 'suspicious']
MEDICAL_GLOSSARY = {
'pulmonary edema': 'Fluid buildup in the lungs, making breathing difficult',
'focal consolidation': 'Dense area in lung tissue, possibly indicating infection',
'pleural effusion': 'Fluid accumulation around the lungs',
'pneumothorax': 'Collapsed lung due to air leak',
'cardiomegaly': 'Enlarged heart',
'opacity': 'Cloudy area on X-ray, may indicate infection or fluid',
'infiltrate': 'Abnormal substance in lung tissue',
'atelectasis': 'Partial lung collapse',
'nodule': 'Small round growth in the lung',
'mass': 'Larger abnormal tissue growth requiring investigation',
'consolidation': 'Area where air spaces are filled with fluid or tissue',
'effusion': 'Abnormal fluid collection'
}
def structure_entities(entities):
anatomy_entities = []
observation_entities = []
positive_findings = []
negative_findings = []
critical_findings = []
for entity in entities:
if entity['label'] == 'ANATOMY':
anatomy_entities.append(entity['text'])
elif entity['label'] == 'OBSERVATION':
observation_entities.append(entity['text'])
obs_lower = entity['text'].lower()
is_critical = any(term in obs_lower for term in CRITICAL_TERMS)
is_negative = any(term in obs_lower for term in NEGATIVE_TERMS)
is_positive = any(term in obs_lower for term in POSITIVE_TERMS)
if is_critical:
critical_findings.append(entity['text'])
elif is_negative:
negative_findings.append(entity['text'])
elif is_positive:
positive_findings.append(entity['text'])
return {
'anatomy': list(set(anatomy_entities)),
'all_observations': observation_entities,
'positive_findings': list(set(positive_findings)),
'negative_findings': list(set(negative_findings)),
'critical_findings': list(set(critical_findings))
}
def generate_summary(structured):
total_entities = len(structured['anatomy']) + len(structured['all_observations'])
anatomy_count = len(structured['anatomy'])
observations_count = len(structured['all_observations'])
has_critical = len(structured['critical_findings']) > 0
has_abnormalities = len(structured['positive_findings']) > 0
return {
'total_entities': total_entities,
'anatomy_count': anatomy_count,
'observations_count': observations_count,
'has_critical_findings': has_critical,
'has_abnormalities': has_abnormalities
}
def generate_patient_friendly_summary(structured):
summary = {
'overall_status': '',
'key_findings': [],
'areas_of_concern': [],
'next_steps': [],
'explanation': ''
}
total_abnormalities = len(structured['positive_findings'])
total_critical = len(structured['critical_findings'])
if total_critical > 0:
summary['overall_status'] = 'URGENT ATTENTION REQUIRED'
summary['explanation'] = 'Your X-ray shows findings that need immediate medical attention. Please contact your doctor right away.'
elif total_abnormalities > 0:
summary['overall_status'] = 'ABNORMALITIES DETECTED'
summary['explanation'] = 'Your X-ray shows some areas of concern that should be discussed with your doctor. This does not necessarily mean a serious problem, but follow-up is recommended.'
else:
summary['overall_status'] = 'NO MAJOR CONCERNS'
summary['explanation'] = 'Your chest X-ray appears normal with no significant abnormalities detected. Continue regular health maintenance.'
for finding in structured['positive_findings']:
finding_lower = finding.lower()
explanation = None
for term, desc in MEDICAL_GLOSSARY.items():
if term in finding_lower:
explanation = desc
break
summary['key_findings'].append({
'finding': finding,
'explanation': explanation if explanation else 'Please consult your doctor for clarification on this finding'
})
for critical in structured['critical_findings']:
critical_lower = critical.lower()
explanation = None
for term, desc in MEDICAL_GLOSSARY.items():
if term in critical_lower:
explanation = desc
break
summary['areas_of_concern'].append({
'finding': critical,
'explanation': explanation if explanation else 'This requires immediate medical evaluation',
'severity': 'HIGH'
})
if total_critical > 0:
summary['next_steps'] = [
'Contact your doctor immediately',
'Do not delay seeking medical attention',
'Bring this report to your appointment',
'Follow all recommended treatments'
]
elif total_abnormalities > 0:
summary['next_steps'] = [
'Schedule a follow-up with your doctor within 1-2 weeks',
'Additional tests or imaging may be recommended',
'Discuss treatment options if needed',
'Monitor for any new or worsening symptoms'
]
else:
summary['next_steps'] = [
'Maintain regular health check-ups',
'Continue healthy lifestyle practices',
'Report any new respiratory symptoms to your doctor',
'Follow recommended screening schedules'
]
return summary
def generate_recommendations(structured):
recommendations = []
if len(structured['critical_findings']) > 0:
recommendations.append("Immediate medical consultation required")
recommendations.append("Consider emergency department visit if symptomatic")
if len(structured['positive_findings']) > 0:
recommendations.append("Follow-up imaging may be needed")
recommendations.append("Consult with a pulmonologist or radiologist")
if 'pneumothorax' in ' '.join(structured['positive_findings']).lower():
recommendations.append("Urgent evaluation for possible chest tube placement")
if 'mass' in ' '.join(structured['positive_findings']).lower() or 'nodule' in ' '.join(structured['positive_findings']).lower():
recommendations.append("CT scan recommended for further evaluation")
recommendations.append("Consider biopsy if clinically indicated")
if 'effusion' in ' '.join(structured['positive_findings']).lower():
recommendations.append("Consider thoracentesis if symptomatic")
if len(structured['critical_findings']) == 0 and len(structured['positive_findings']) == 0:
recommendations.append("Continue routine health maintenance")
recommendations.append("Repeat imaging only if clinically indicated")
return recommendations