Skip to content

Commit

Permalink
Drivers License Feature (#113)
Browse files Browse the repository at this point in the history
## Description
Add a new Feature to run comparisons on driver's license numbers.

## Related Issues
closes #95 

## Additional Notes
  • Loading branch information
cbrinson-rise8 authored Nov 1, 2024
1 parent 1eb17ad commit e87247a
Show file tree
Hide file tree
Showing 4 changed files with 60 additions and 4 deletions.
10 changes: 9 additions & 1 deletion src/recordlinker/hl7/fhir.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,21 @@ 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", []):
if coding.get("code") == "MR":
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", []):
Expand All @@ -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(
Expand Down
21 changes: 19 additions & 2 deletions src/recordlinker/schemas/pii.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ class Feature(enum.Enum):
TELEPHONE = "TELEPHONE"
SUFFIX = "SUFFIX"
COUNTY = "COUNTY"
DRIVERS_LICENSE = "DRIVERS_LICENSE"

def __str__(self):
"""
Expand Down Expand Up @@ -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):
"""
Expand All @@ -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:
Expand All @@ -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")
Expand Down Expand Up @@ -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]:
"""
Expand Down Expand Up @@ -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]:
"""
Expand Down
25 changes: 24 additions & 1 deletion tests/unit/hl7/test_fhir.py
Original file line number Diff line number Diff line change
Expand Up @@ -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": [
Expand Down Expand Up @@ -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")
Expand Down
8 changes: 8 additions & 0 deletions tests/unit/schemas/test_pii.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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):
Expand Down Expand Up @@ -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):
Expand All @@ -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()
Expand Down

0 comments on commit e87247a

Please sign in to comment.