Example: Data Scrubbing |
One very useful application of scripted transformation is for building a utility to remove confidential data from a set of HL7 messages.
This sample VMD file is an example of how you can remove confidential data from an HL7 message. The following Python scripts are defined in this VMD file:
- A Transformation Script that specifies a list of segments or fields in an incoming message. This script also specifies whether the list is of the fields to be scrubbed or the fields that are to be excluded from scrubbing. If the list is of the fields to be scrubbed, the script is in white list mode. If the list is of the fields not to be scrubbed, the script is in black list mode.
- A Global Inbound Script that defines the data scrubbing functions that the Transformation Script uses.
Here is the Transformation Script:
# This list contains fields that are passed as a white list/black list for scrubbing.
# Note that it's possible to be very specific or general depending
# on how much needs to be scrubbed
# Format of the specifier:
# 'PID' - the whole PID segment
# 'PID.1.1' - PID field 1 subfield 1
# 'PID.3' - All of PID field 3
scrub_list = \
['PID.5',
'PID.11',
'PID.7',
'PV1.7',
'PV1.8',
'OBX.5']
#convert list to dictionary
#so lookups are faster
scrub_list = to_dict(scrub_list)
#In white list mode, only fields specified in scrub_list are scrubbed
#In black list mode, fields specified in scrub_list are NOT scrubbed, everything else is
# Set to 0 for black list mode.
scrub_white_list_mode = 1
# This loop iterates through every segment, and every field in the segment
# scrubbing each one if necessary.
output = ''
current_segment = environment.input_segment_iterator()
while 1:
for field_index in range(1,current_segment.count_of_field()):
#skip field 1 and 2 for msh
if not (current_segment.segment_id() == 'MSH' and (field_index == 1)):
scrub_field(current_segment.segment_id(),
current_segment.field(field_index),
field_index,
scrub_list,
scrub_white_list_mode )
output = output + current_segment.output()
if not current_segment.move_one():
break
#We replace value with the output to yield the scrubbed data.
value = output
|
To use this Python script to scrub data, you must:
- Set the scrub_white_list_mode variable to indicate whether the script is in white list mode or black list mode. If the script is in white list mode, set scrub_white_list_mode to 1. If the script is in black list mode, set scrub_white_list_mode to 0.
- Create a list of segments and fields to be scrubbed or excluded from scrubbing, and assign it to the scrub_list variable:
scrub_list = \ ['PID.5', 'PID.11', 'PID.7', 'PV1.7', 'PV1.8', 'OBX.5'] |
Each item in the list can be an entire segment (for example, PID), a field in a segment (for example, PID.3) or a subfield (for example, PID.1.1). The value of scrub_white_list_mode indicates whether this list is to be scrubbed or excluded from scrubbing.
The Python code contained in the Global Inbound Script performs the actual work of data scrubbing. Alphanumeric values are replaced with X characters, and numbers are replaced with N characters.
import re
import string
# Use some regular expressions to do the scrubbing
letters_scrub_re = re.compile('([A-Z]|[a-z])')
numbers_scrub_re = re.compile('[0-9]')
# This function takes a string, scrubs it and returns it
# In this case, all letters are replaced with X,
# And all numbers are replaced with N.
# Punctuation is left alone.
def scrub( data ):
out = letters_scrub_re.sub( 'X', data)
out = numbers_scrub_re.sub( 'N', out)
return out
# A utility function to convert a list to a
# dictionary.
def to_dict(in_list):
out = {}
for obj in in_list: out[obj] = None
return out
# A helper function to determine if a field should be scrubbed or
# not.
def do_scrub(field, id_list, scrub_ids, white_list_mode):
# This line creates a list of possible strings
# that we can match against to see if we should scrub or not
possible_ids = ['.'.join(id_list[:i]) for i in range(1, len(id_list)+1)]
# Determine if any item in possible ids match the scrub_ids list
# Then based on white list/black list mode, and if it's in the list
# we scrub the field value.
matched = 0
if (1 in [(id in scrub_ids) for id in possible_ids]): matched = 1
if (matched and white_list_mode) or ((not matched) and (not white_list_mode)):
field.value = scrub(field.value)
# This iterates through every repeat/sub/subsub field in a given field
# and applies do_scrub
def scrub_field(segment_id, field, field_index, scrub_ids, white_list_mode ):
for repeat_index in range(0,field.count_of_repeat()):
repeat = field.repeat_field(repeat_index);
# Note that the 'or 1' is required to handle single entry fields
# e.g. |something|
for sub_field_index in range(0,repeat.count_of_subfield() or 1):
sub = repeat.subfield(sub_field_index);
for sub_sub_field_index in range(0, sub.count_of_subfield() or 1):
sub_sub = sub.subfield(sub_sub_field_index)
id_list = map(str, [ segment_id, field_index, sub_field_index, sub_sub_field_index ])
do_scrub(sub_sub, id_list, scrub_ids, white_list_mode)
|
Note that you will never need to edit the Global Inbound Script. Its operation is entirely controlled by the values of scrub_list and scrub_white_list_mode that are defined in the Transformation Script.
 | ||
|
To test your data scrubbing operation, you can use the Message Diff tool to display what fields you have altered. | ||
 |