Update responseparser.py (#5)

- Update responseparser.py (c55e3b0fd1ba42e21dce494341108786bcffdb1a)

responseparser.py

@@ -1,21 +1,23 @@
-import json
+import lxml.etree as etree
 from datetime import datetime
 from typing import List, Dict, Optional, Union
 
 class PatientDataExtractor:
-    """Class to extract all fields from a FHIR Patient resource in a Bundle response."""
+    """Class to extract all fields from a FHIR Patient resource in a Bundle response (XML format)."""
     
     def __init__(self, patient_data: str):
-        """Initialize with patient data in JSON string format."""
-        self.data = json.loads(patient_data) if isinstance(patient_data, str) else patient_data
+        """Initialize with patient data in XML string format."""
+        # Parse XML string or use pre-parsed data
+        self.data = etree.fromstring(patient_data) if isinstance(patient_data, str) else patient_data
+        # Define FHIR namespace for XPath queries
+        self.ns = {'fhir': 'http://hl7.org/fhir'}
         self.patients = self._extract_patients()
         self.current_patient_idx = 0  # Default to first patient
         
-    def _extract_patients(self) -> List[Dict]:
+    def _extract_patients(self) -> List[etree._Element]:
         """Extract all patient entries from the Bundle."""
-        if self.data.get("resourceType") != "Bundle" or "entry" not in self.data:
-            raise ValueError("Invalid FHIR Bundle format")
-        return [entry["resource"] for entry in self.data["entry"] if entry["resource"]["resourceType"] == "Patient"]
+        # Use XPath to find all Patient elements in the Bundle
+        return self.data.xpath("//fhir:entry/fhir:resource/fhir:Patient", namespaces=self.ns)
 
     def set_patient_by_index(self, index: int) -> bool:
         """Set the current patient by index. Returns True if successful."""
@@ -27,86 +29,101 @@ class PatientDataExtractor:
     def set_patient_by_id(self, patient_id: str) -> bool:
         """Set the current patient by FHIR Patient ID. Returns True if successful."""
         for i, patient in enumerate(self.patients):
-            if patient["id"] == patient_id:
+            if patient.attrib.get("id") == patient_id:
                 self.current_patient_idx = i
                 return True
         return False
 
-    def _get_current_patient(self) -> Dict:
+    def _get_current_patient(self) -> etree._Element:
         """Get the currently selected patient resource."""
         return self.patients[self.current_patient_idx]
 
     # Basic Identification Fields
     def get_id(self) -> str:
         """Extract FHIR Patient ID."""
-        return self._get_current_patient().get("id", "")
+        return self._get_current_patient().attrib.get("id", "")
 
     def get_resource_type(self) -> str:
         """Extract resource type (should always be 'Patient')."""
-        return self._get_current_patient().get("resourceType", "")
+        return etree.QName(self._get_current_patient().tag).localname
 
     def get_meta_last_updated(self) -> str:
         """Extract last updated timestamp from meta."""
-        return self._get_current_patient().get("meta", {}).get("lastUpdated", "")
+        patient = self._get_current_patient()
+        last_updated = patient.xpath("fhir:meta/fhir:lastUpdated/@value", namespaces=self.ns)
+        return last_updated[0] if last_updated else ""
 
     def get_meta_profile(self) -> List[str]:
         """Extract profile URIs from meta."""
-        return self._get_current_patient().get("meta", {}).get("profile", [])
+        patient = self._get_current_patient()
+        profiles = patient.xpath("fhir:meta/fhir:profile/@value", namespaces=self.ns)
+        return profiles
 
     def get_text_div(self) -> str:
         """Extract generated text narrative (div content)."""
-        return self._get_current_patient().get("text", {}).get("div", "")
+        patient = self._get_current_patient()
+        div = patient.xpath("fhir:text/fhir:div", namespaces=self.ns)
+        if div:
+            return etree.tostring(div[0], encoding="unicode")
+        return ""
 
     # Name Fields
     def get_first_name(self) -> str:
         """Extract patient's first name."""
         patient = self._get_current_patient()
-        names = patient.get("name", [])
-        for name in names:
-            if name.get("use") == "official" and "given" in name:
-                return name["given"][0]
+        official_names = patient.xpath("fhir:name[fhir:use/@value='official']", namespaces=self.ns)
+        if official_names:
+            given = official_names[0].xpath("fhir:given/@value", namespaces=self.ns)
+            if given:
+                return given[0]
         return ""
 
     def get_last_name(self) -> str:
         """Extract patient's last name."""
         patient = self._get_current_patient()
-        names = patient.get("name", [])
-        for name in names:
-            if name.get("use") == "official" and "family" in name:
-                return name["family"]
+        official_names = patient.xpath("fhir:name[fhir:use/@value='official']", namespaces=self.ns)
+        if official_names:
+            family = official_names[0].xpath("fhir:family/@value", namespaces=self.ns)
+            if family:
+                return family[0]
         return ""
 
     def get_middle_initial(self) -> str:
         """Extract patient's middle initial (second given name initial if present)."""
         patient = self._get_current_patient()
-        names = patient.get("name", [])
-        for name in names:
-            if name.get("use") == "official" and "given" in name and len(name["given"]) > 1:
-                return name["given"][1][0]
+        official_names = patient.xpath("fhir:name[fhir:use/@value='official']", namespaces=self.ns)
+        if official_names:
+            given = official_names[0].xpath("fhir:given/@value", namespaces=self.ns)
+            if len(given) > 1:
+                return given[1][0]
         return ""
 
     def get_name_prefix(self) -> str:
         """Extract patient's name prefix (e.g., Mr., Mrs.)."""
         patient = self._get_current_patient()
-        names = patient.get("name", [])
-        for name in names:
-            if name.get("use") == "official" and "prefix" in name:
-                return name["prefix"][0]
+        official_names = patient.xpath("fhir:name[fhir:use/@value='official']", namespaces=self.ns)
+        if official_names:
+            prefix = official_names[0].xpath("fhir:prefix/@value", namespaces=self.ns)
+            if prefix:
+                return prefix[0]
         return ""
 
     def get_maiden_name(self) -> str:
         """Extract patient's maiden name if available."""
         patient = self._get_current_patient()
-        names = patient.get("name", [])
-        for name in names:
-            if name.get("use") == "maiden" and "family" in name:
-                return name["family"]
+        maiden_names = patient.xpath("fhir:name[fhir:use/@value='maiden']", namespaces=self.ns)
+        if maiden_names:
+            family = maiden_names[0].xpath("fhir:family/@value", namespaces=self.ns)
+            if family:
+                return family[0]
         return ""
 
     # Demographic Fields
     def get_dob(self) -> str:
         """Extract patient's date of birth."""
-        return self._get_current_patient().get("birthDate", "")
+        patient = self._get_current_patient()
+        dob = patient.xpath("fhir:birthDate/@value", namespaces=self.ns)
+        return dob[0] if dob else ""
 
     def get_age(self) -> str:
         """Calculate patient's age based on birth date."""
@@ -120,160 +137,145 @@ class PatientDataExtractor:
 
     def get_gender(self) -> str:
         """Extract patient's gender."""
-        return self._get_current_patient().get("gender", "").capitalize()
+        patient = self._get_current_patient()
+        gender = patient.xpath("fhir:gender/@value", namespaces=self.ns)
+        return gender[0].capitalize() if gender else ""
 
     def get_birth_sex(self) -> str:
         """Extract patient's birth sex from extensions."""
         patient = self._get_current_patient()
-        for ext in patient.get("extension", []):
-            if ext.get("url") == "http://hl7.org/fhir/us/core/StructureDefinition/us-core-birthsex":
-                return ext.get("valueCode", "")
-        return ""
+        birth_sex = patient.xpath("fhir:extension[@url='http://hl7.org/fhir/us/core/StructureDefinition/us-core-birthsex']/fhir:valueCode/@value", namespaces=self.ns)
+        return birth_sex[0] if birth_sex else ""
 
     def get_multiple_birth(self) -> Union[bool, None]:
         """Extract multiple birth status."""
-        return self._get_current_patient().get("multipleBirthBoolean", None)
+        patient = self._get_current_patient()
+        multiple_birth = patient.xpath("fhir:multipleBirthBoolean/@value", namespaces=self.ns)
+        return multiple_birth[0] == "true" if multiple_birth else None
 
     # Address Fields
     def get_address_line(self) -> str:
         """Extract patient's street address."""
         patient = self._get_current_patient()
-        addresses = patient.get("address", [])
-        return addresses[0]["line"][0] if addresses and "line" in addresses[0] else ""
+        line = patient.xpath("fhir:address/fhir:line/@value", namespaces=self.ns)
+        return line[0] if line else ""
 
     def get_city(self) -> str:
         """Extract patient's city."""
         patient = self._get_current_patient()
-        addresses = patient.get("address", [])
-        return addresses[0]["city"] if addresses and "city" in addresses[0] else ""
+        city = patient.xpath("fhir:address/fhir:city/@value", namespaces=self.ns)
+        return city[0] if city else ""
 
     def get_state(self) -> str:
         """Extract patient's state."""
         patient = self._get_current_patient()
-        addresses = patient.get("address", [])
-        return addresses[0]["state"] if addresses and "state" in addresses[0] else ""
+        state = patient.xpath("fhir:address/fhir:state/@value", namespaces=self.ns)
+        return state[0] if state else ""
 
     def get_zip_code(self) -> str:
         """Extract patient's postal code."""
         patient = self._get_current_patient()
-        addresses = patient.get("address", [])
-        return addresses[0]["postalCode"] if addresses and "postalCode" in addresses[0] else ""
+        postal_code = patient.xpath("fhir:address/fhir:postalCode/@value", namespaces=self.ns)
+        return postal_code[0] if postal_code else ""
 
     def get_country(self) -> str:
         """Extract patient's country."""
         patient = self._get_current_patient()
-        addresses = patient.get("address", [])
-        return addresses[0]["country"] if addresses and "country" in addresses[0] else ""
+        country = patient.xpath("fhir:address/fhir:country/@value", namespaces=self.ns)
+        return country[0] if country else ""
 
     def get_geolocation(self) -> Dict[str, float]:
         """Extract geolocation (latitude and longitude) from address extension."""
         patient = self._get_current_patient()
-        addresses = patient.get("address", [])
-        if not addresses:
-            return {"latitude": None, "longitude": None}
-        for ext in addresses[0].get("extension", []):
-            if ext.get("url") == "http://hl7.org/fhir/StructureDefinition/geolocation":
-                geo = {}
-                for sub_ext in ext.get("extension", []):
-                    if sub_ext.get("url") == "latitude":
-                        geo["latitude"] = sub_ext.get("valueDecimal")
-                    elif sub_ext.get("url") == "longitude":
-                        geo["longitude"] = sub_ext.get("valueDecimal")
-                return geo
-        return {"latitude": None, "longitude": None}
+        lat = patient.xpath("fhir:address/fhir:extension[@url='http://hl7.org/fhir/StructureDefinition/geolocation']/fhir:extension[@url='latitude']/fhir:valueDecimal/@value", namespaces=self.ns)
+        lon = patient.xpath("fhir:address/fhir:extension[@url='http://hl7.org/fhir/StructureDefinition/geolocation']/fhir:extension[@url='longitude']/fhir:valueDecimal/@value", namespaces=self.ns)
+        return {
+            "latitude": float(lat[0]) if lat else None,
+            "longitude": float(lon[0]) if lon else None
+        }
 
     # Contact Fields
     def get_phone(self) -> str:
         """Extract patient's phone number."""
         patient = self._get_current_patient()
-        telecoms = patient.get("telecom", [])
-        for telecom in telecoms:
-            if telecom.get("system") == "phone" and telecom.get("use") == "home":
-                return telecom.get("value", "")
-        return ""
+        phone = patient.xpath("fhir:telecom[fhir:system/@value='phone' and fhir:use/@value='home']/fhir:value/@value", namespaces=self.ns)
+        return phone[0] if phone else ""
 
     # Identifiers
     def get_identifiers(self) -> Dict[str, str]:
         """Extract all identifiers (e.g., SSN, MRN, Driver's License)."""
         patient = self._get_current_patient()
-        identifiers = patient.get("identifier", [])
         id_dict = {}
+        identifiers = patient.xpath("fhir:identifier", namespaces=self.ns)
         for id_entry in identifiers:
-            id_type = id_entry.get("type", {}).get("text", "Unknown")
-            id_dict[id_type] = id_entry.get("value", "")
+            id_type = id_entry.xpath("fhir:type/fhir:text/@value", namespaces=self.ns)
+            id_value = id_entry.xpath("fhir:value/@value", namespaces=self.ns)
+            if id_type and id_value:
+                id_dict[id_type[0]] = id_value[0]
         return id_dict
 
     # Extensions
     def get_race(self) -> str:
         """Extract patient's race from extensions."""
         patient = self._get_current_patient()
-        for ext in patient.get("extension", []):
-            if ext.get("url") == "http://hl7.org/fhir/us/core/StructureDefinition/us-core-race":
-                for sub_ext in ext.get("extension", []):
-                    if sub_ext.get("url") == "text":
-                        return sub_ext.get("valueString", "")
-        return ""
+        race = patient.xpath("fhir:extension[@url='http://hl7.org/fhir/us/core/StructureDefinition/us-core-race']/fhir:extension[@url='text']/fhir:valueString/@value", namespaces=self.ns)
+        return race[0] if race else ""
 
     def get_ethnicity(self) -> str:
         """Extract patient's ethnicity from extensions."""
         patient = self._get_current_patient()
-        for ext in patient.get("extension", []):
-            if ext.get("url") == "http://hl7.org/fhir/us/core/StructureDefinition/us-core-ethnicity":
-                for sub_ext in ext.get("extension", []):
-                    if sub_ext.get("url") == "text":
-                        return sub_ext.get("valueString", "")
-        return ""
+        ethnicity = patient.xpath("fhir:extension[@url='http://hl7.org/fhir/us/core/StructureDefinition/us-core-ethnicity']/fhir:extension[@url='text']/fhir:valueString/@value", namespaces=self.ns)
+        return ethnicity[0] if ethnicity else ""
 
     def get_mothers_maiden_name(self) -> str:
         """Extract patient's mother's maiden name from extensions."""
         patient = self._get_current_patient()
-        for ext in patient.get("extension", []):
-            if ext.get("url") == "http://hl7.org/fhir/StructureDefinition/patient-mothersMaidenName":
-                return ext.get("valueString", "")
-        return ""
+        mothers_maiden = patient.xpath("fhir:extension[@url='http://hl7.org/fhir/StructureDefinition/patient-mothersMaidenName']/fhir:valueString/@value", namespaces=self.ns)
+        return mothers_maiden[0] if mothers_maiden else ""
 
     def get_birth_place(self) -> Dict[str, str]:
         """Extract patient's birth place from extensions."""
         patient = self._get_current_patient()
-        for ext in patient.get("extension", []):
-            if ext.get("url") == "http://hl7.org/fhir/StructureDefinition/patient-birthPlace":
-                addr = ext.get("valueAddress", {})
-                return {
-                    "city": addr.get("city", ""),
-                    "state": addr.get("state", ""),
-                    "country": addr.get("country", "")
-                }
+        birth_place = patient.xpath("fhir:extension[@url='http://hl7.org/fhir/StructureDefinition/patient-birthPlace']/fhir:valueAddress", namespaces=self.ns)
+        if birth_place:
+            city = birth_place[0].xpath("fhir:city/@value", namespaces=self.ns)
+            state = birth_place[0].xpath("fhir:state/@value", namespaces=self.ns)
+            country = birth_place[0].xpath("fhir:country/@value", namespaces=self.ns)
+            return {
+                "city": city[0] if city else "",
+                "state": state[0] if state else "",
+                "country": country[0] if country else ""
+            }
         return {"city": "", "state": "", "country": ""}
 
     def get_disability_adjusted_life_years(self) -> Optional[float]:
         """Extract disability-adjusted life years from extensions."""
         patient = self._get_current_patient()
-        for ext in patient.get("extension", []):
-            if ext.get("url") == "http://synthetichealth.github.io/synthea/disability-adjusted-life-years":
-                return ext.get("valueDecimal")
-        return None
+        daly = patient.xpath("fhir:extension[@url='http://synthetichealth.github.io/synthea/disability-adjusted-life-years']/fhir:valueDecimal/@value", namespaces=self.ns)
+        return float(daly[0]) if daly else None
 
     def get_quality_adjusted_life_years(self) -> Optional[float]:
         """Extract quality-adjusted life years from extensions."""
         patient = self._get_current_patient()
-        for ext in patient.get("extension", []):
-            if ext.get("url") == "http://synthetichealth.github.io/synthea/quality-adjusted-life-years":
-                return ext.get("valueDecimal")
-        return None
+        qaly = patient.xpath("fhir:extension[@url='http://synthetichealth.github.io/synthea/quality-adjusted-life-years']/fhir:valueDecimal/@value", namespaces=self.ns)
+        return float(qaly[0]) if qaly else None
 
     # Marital Status
     def get_marital_status(self) -> str:
         """Extract patient's marital status."""
         patient = self._get_current_patient()
-        status = patient.get("maritalStatus", {}).get("text", "")
-        return status if status else patient.get("maritalStatus", {}).get("coding", [{}])[0].get("display", "")
+        status = patient.xpath("fhir:maritalStatus/fhir:text/@value", namespaces=self.ns)
+        if status:
+            return status[0]
+        coding = patient.xpath("fhir:maritalStatus/fhir:coding/fhir:display/@value", namespaces=self.ns)
+        return coding[0] if coding else ""
 
     # Communication
     def get_language(self) -> str:
         """Extract patient's preferred language."""
         patient = self._get_current_patient()
-        comms = patient.get("communication", [])
-        return comms[0]["language"]["text"] if comms and "language" in comms[0] else ""
+        language = patient.xpath("fhir:communication/fhir:language/fhir:text/@value", namespaces=self.ns)
+        return language[0] if language else ""
 
     # Comprehensive Extraction
     def get_all_patient_data(self) -> Dict[str, Union[str, Dict, List, float, bool, None]]:
@@ -359,7 +361,7 @@ class PatientDataExtractor:
 
     def get_patient_ids(self) -> List[str]:
         """Return a list of all patient IDs in the Bundle."""
-        return [patient["id"] for patient in self.patients]
+        return [patient.attrib.get("id", "") for patient in self.patients]
 
 # # Example usage with integration into app.py
 # def integrate_with_app(patient_data: str):