diff --git a/src/recordlinker/hl7/fhir.py b/src/recordlinker/hl7/fhir.py index 1a1a61c9..75d823be 100644 --- a/src/recordlinker/hl7/fhir.py +++ b/src/recordlinker/hl7/fhir.py @@ -27,6 +27,7 @@ def fhir_record_to_pii_record(fhir_record: dict) -> schemas.PIIRecord: "race": None, "gender": None, "telecom": fhir_record.get("telecom", []), + "drivers_license": None, } for identifier in fhir_record.get("identifier", []): for coding in identifier.get("type", {}).get("coding", []): @@ -34,6 +35,13 @@ def fhir_record_to_pii_record(fhir_record: dict) -> schemas.PIIRecord: val["mrn"] = identifier.get("value") elif coding.get("code") == "SS": val["ssn"] = identifier.get("value") + elif coding.get("code") == "DL": + license_number = identifier.get("value") + authority = identifier.get("assigner", {}).get("identifier", {}).get("value") # Assuming `issuer` contains authority info + val["drivers_license"] = { + "value": license_number, + "authority": authority + } for address in val["address"]: address["county"] = address.get("district", "") for extension in address.get("extension", []): @@ -53,7 +61,7 @@ def fhir_record_to_pii_record(fhir_record: dict) -> schemas.PIIRecord: if ext.get("url") == "value": for coding in ext.get("valueCodeableConcept", {}).get("coding", []): val["gender"] = coding.get("display") - + return schemas.PIIRecord(**val) def add_person_resource( diff --git a/src/recordlinker/schemas/pii.py b/src/recordlinker/schemas/pii.py index 627ab627..1ce1c407 100644 --- a/src/recordlinker/schemas/pii.py +++ b/src/recordlinker/schemas/pii.py @@ -29,6 +29,7 @@ class Feature(enum.Enum): TELEPHONE = "TELEPHONE" SUFFIX = "SUFFIX" COUNTY = "COUNTY" + DRIVERS_LICENSE = "DRIVERS_LICENSE" def __str__(self): """ @@ -133,10 +134,19 @@ class Telecom(pydantic.BaseModel): model_config = pydantic.ConfigDict(extra="allow") - value: str # future use + value: str system: typing.Optional[str] = None use: typing.Optional[str] = None +class DriversLicense(pydantic.BaseModel): + """ + The schema for a Drivers License record + """ + + model_config = pydantic.ConfigDict(extra="allow") + + value: str + authority: str class PIIRecord(pydantic.BaseModel): """ @@ -157,6 +167,7 @@ class PIIRecord(pydantic.BaseModel): ssn: typing.Optional[str] = None race: typing.Optional[Race] = None gender: typing.Optional[Gender] = None + drivers_license: typing.Optional[DriversLicense] = None @classmethod def model_construct(cls, _fields_set: set[str] | None = None, **values: typing.Any) -> typing.Self: @@ -171,6 +182,7 @@ def model_construct(cls, _fields_set: set[str] | None = None, **values: typing.A obj.address = [Address.model_construct(**a) for a in values.get("address", [])] obj.name = [Name.model_construct(**n) for n in values.get("name", [])] obj.telecom = [Telecom.model_construct(**t) for t in values.get("telecom", [])] + obj.drivers_license = DriversLicense.model_construct(**values.get("drivers_license", {})) return obj @pydantic.field_validator("external_id", mode="before") @@ -263,7 +275,8 @@ def parse_gender(cls, value): return Gender.NON_BINARY elif "declined" in val or "asked" in val: return Gender.ASKED_DECLINED - return Gender.UNKNOWN + return Gender.UNKNOWN + def feature_iter(self, feature: Feature) -> typing.Iterator[str]: """ @@ -332,6 +345,10 @@ def feature_iter(self, feature: Feature) -> typing.Iterator[str]: for address in self.address: if address.county: yield address.county + elif feature == Feature.DRIVERS_LICENSE: + if self.drivers_license: + if self.drivers_license.value and self.drivers_license.authority: + yield f"{self.drivers_license.value}|{self.drivers_license.authority}" def blocking_keys(self, key: models.BlockingKey) -> set[str]: """ diff --git a/tests/unit/hl7/test_fhir.py b/tests/unit/hl7/test_fhir.py index a5622ed6..1aabd5c2 100644 --- a/tests/unit/hl7/test_fhir.py +++ b/tests/unit/hl7/test_fhir.py @@ -36,6 +36,28 @@ def test_fhir_record_to_pii_record(): "code" : "SS" }] }, + }, + { + "use": "official", + "type": { + "text": "Driver's License", + "coding": [ + { + "system": "http://terminology.hl7.org/CodeSystem/v2-0203", + "code": "DL", + "display": "Driver's License" + } + ], + }, + "system": "urn:oid:2.16.840.1.113883.3.19.3.1.8", + "value": "D1234567", + "assigner": { + "display": "State DMV", + "identifier": { + "system": "urn:oid:2.16.840.1.113883.19.3.1.1", + "value": "CA" + } + } } ], "name": [ @@ -117,7 +139,8 @@ def test_fhir_record_to_pii_record(): assert pii_record.telecom[0].system == "phone" assert str(pii_record.race) == "WHITE" assert str(pii_record.gender) == "FEMALE" - + assert pii_record.drivers_license.authority == "CA" + assert pii_record.drivers_license.value == "D1234567" def test_add_person_resource(): bundle = load_json_asset("patient_bundle.json") diff --git a/tests/unit/schemas/test_pii.py b/tests/unit/schemas/test_pii.py index 5f9f8abf..47e1ce79 100644 --- a/tests/unit/schemas/test_pii.py +++ b/tests/unit/schemas/test_pii.py @@ -44,6 +44,10 @@ def test_model_construct(self): }, ], "telecom": [{"value": "555-123-4567"}, {"value": "555-987-6543"}], + "drivers_license": { + "authority": "VA", + "value": "D1234567" + } } record = pii.PIIRecord.model_construct(**data) assert record.mrn == "99" @@ -62,6 +66,8 @@ def test_model_construct(self): assert record.address[1].state == "CA" assert record.address[1].postal_code == "98765-4321" assert record.address[1].county == "county2" + assert record.drivers_license.value == "D1234567" + assert record.drivers_license.authority == "VA" def test_parse_external_id(self): @@ -234,6 +240,7 @@ def test_feature_iter(self): pii.Telecom(value="555-123-4567"), pii.Telecom(value="555-987-6543"), ], + drivers_license=pii.DriversLicense(value="D1234567", authority="VA") ) with pytest.raises(ValueError): @@ -254,6 +261,7 @@ def test_feature_iter(self): assert list(record.feature_iter(pii.Feature.TELEPHONE)) == ["555-123-4567", "555-987-6543"] assert list(record.feature_iter(pii.Feature.SUFFIX)) == ["suffix", "suffix2"] assert list(record.feature_iter(pii.Feature.COUNTY)) == ["county"] + assert list(record.feature_iter(pii.Feature.DRIVERS_LICENSE)) == ["D1234567|VA"] def test_blocking_keys_invalid(self): rec = pii.PIIRecord()