diff --git a/config.json b/config.json index 1e7d4db..689e9ff 100644 --- a/config.json +++ b/config.json @@ -4,6 +4,7 @@ "attribute": "length", "ascending": false }, + "lookup_lists_path": "./data/lookup_lists", "redactor_open_char": "<", "redactor_close_char": ">", "annotators": { diff --git a/deduce/deduce.py b/deduce/deduce.py index d04ed14..01b55ca 100644 --- a/deduce/deduce.py +++ b/deduce/deduce.py @@ -10,7 +10,6 @@ import deduce.backwards_compat from deduce import utils -from deduce.lookup_sets import get_lookup_sets from deduce.process.annotation_processing import ( DeduceMergeAdjacentAnnotations, PersonAnnotationConverter, @@ -18,6 +17,7 @@ from deduce.process.annotator import AnnotationContextPatternAnnotator from deduce.process.redact import DeduceRedactor from deduce.tokenize import DeduceTokenizer +from deduce.lookup_sets import LookupSetLoader warnings.simplefilter(action="once") @@ -32,13 +32,17 @@ class Deduce(dd.DocDeid): def __init__(self, config_file: Optional[str] = None) -> None: super().__init__() - self.config = self._initialize_config(config_file) - self.lookup_sets = get_lookup_sets() + self.config = self.initialize_config(config_file) + self.lookup_sets = self.initialize_lookupsets(self.config) self.tokenizers = self._initialize_tokenizers() self.initialize_doc_processors() @staticmethod - def _initialize_config(config_file: Optional[str] = None) -> dict: + def initialize_lookupsets(config) -> dd.ds.DsCollection: + return LookupSetLoader(config).build_lookup_sets() + + @staticmethod + def initialize_config(config_file: Optional[str] = None) -> dict: """ Initialize the config file. @@ -66,7 +70,7 @@ def _initialize_tokenizers(self) -> dict: @staticmethod def _initialize_annotators( - annotator_cnfg: dict, lookup_sets: dd.ds.DsCollection, tokenizer: dd.tokenize.Tokenizer + annotator_cnfg: dict, lookup_sets, tokenizer: dd.tokenize.Tokenizer ) -> dd.process.DocProcessorGroup: """Initializes annotators.""" diff --git a/deduce/lookup_sets.py b/deduce/lookup_sets.py index dbc922f..e6cd7e7 100644 --- a/deduce/lookup_sets.py +++ b/deduce/lookup_sets.py @@ -10,186 +10,178 @@ TakeLastToken, ) -data_path = Path(os.path.dirname(__file__)).parent / "data" / "lookup_lists" +class LookupSetLoader: -def _get_first_names_lookup_set() -> dd.ds.LookupSet: - """Get first names LookupSet.""" + def __init__(self, config_dict): + data_path_str = os.path.abspath(config_dict["lookup_lists_path"]) + self.data_path = Path(data_path_str) - first_names = dd.ds.LookupSet() + # body of the constructor + def _load_first_names(self) -> dd.ds.LookupSet: + """Get first names LookupSet.""" - first_names.add_items_from_file( - os.path.join(data_path, "first_names.txt"), - cleaning_pipeline=[dd.str.FilterByLength(min_len=2)], - ) + first_names = dd.ds.LookupSet() - return first_names + first_names.add_items_from_file( + os.path.join(self.data_path, "first_names.txt"), + cleaning_pipeline=[dd.str.FilterByLength(min_len=2)], + ) + return first_names -def _get_surnames_lookup_set() -> dd.ds.LookupSet: - """Get surnames LookupSet.""" + def _load_surnames(self) -> dd.ds.LookupSet: + """Get surnames LookupSet.""" - surnames = dd.ds.LookupSet() + surnames = dd.ds.LookupSet() - surnames.add_items_from_file( - os.path.join(data_path, "surnames.txt"), - cleaning_pipeline=[dd.str.FilterByLength(min_len=2)], - ) + surnames.add_items_from_file( + os.path.join(self.data_path, "surnames.txt"), + cleaning_pipeline=[dd.str.FilterByLength(min_len=2)], + ) - return surnames + return surnames + def _load_interfixes(self) -> dd.ds.LookupSet: + """Get interfixes LookupSet ('van der', etc.)""" -def _get_interfixes_lookup_set() -> dd.ds.LookupSet: - """Get interfixes LookupSet ('van der', etc.)""" + interfixes = dd.ds.LookupSet() - interfixes = dd.ds.LookupSet() + interfixes.add_items_from_file(os.path.join(self.data_path, "interfixes.txt")) - interfixes.add_items_from_file(os.path.join(data_path, "interfixes.txt")) + return interfixes - return interfixes + def _load_interfix_surnames(self) -> dd.ds.LookupSet: + """Get interfix surnames LookupSet (e.g. 'Jong' for 'de Jong')""" + interfix_surnames = dd.ds.LookupSet() -def _get_interfix_surnames_lookup_set() -> dd.ds.LookupSet: - """Get interfix surnames LookupSet (e.g. 'Jong' for 'de Jong')""" + interfix_surnames.add_items_from_file( + os.path.join(self.data_path, "interfix_surnames.txt"), + cleaning_pipeline=[TakeLastToken()], + ) - interfix_surnames = dd.ds.LookupSet() + return interfix_surnames - interfix_surnames.add_items_from_file( - os.path.join(data_path, "interfix_surnames.txt"), - cleaning_pipeline=[TakeLastToken()], - ) + def _load_prefixes(self) -> dd.ds.LookupSet: + """Get prefixes LookupSet (e.g. 'dr', 'mw')""" - return interfix_surnames + prefixes = dd.ds.LookupSet() + prefixes.add_items_from_file(os.path.join(self.data_path, "prefixes.txt")) -def _get_prefixes_lookup_set() -> dd.ds.LookupSet: - """Get prefixes LookupSet (e.g. 'dr', 'mw')""" + return prefixes - prefixes = dd.ds.LookupSet() + def _load_whitelist(self) -> dd.ds.LookupSet: + """ + Get whitelist LookupSet. - prefixes.add_items_from_file(os.path.join(data_path, "prefixes.txt")) + Composed of medical terms, top 1000 frequent words (except surnames), and stopwords. + Returns: + """ + med_terms = dd.ds.LookupSet() + med_terms.add_items_from_file( + os.path.join(self.data_path, "medical_terms.txt"), + ) - return prefixes + top1000 = dd.ds.LookupSet() + top1000.add_items_from_file( + os.path.join(self.data_path, "top_1000_terms.txt"), + ) + surnames_lowercase = dd.ds.LookupSet() + surnames_lowercase.add_items_from_file( + os.path.join(self.data_path, "surnames.txt"), + cleaning_pipeline=[ + dd.str.LowercaseString(), + dd.str.FilterByLength(min_len=2), + ], + ) -def _get_whitelist_lookup_set() -> dd.ds.LookupSet: - """ - Get whitelist LookupSet. + top1000 = top1000 - surnames_lowercase - Composed of medical terms, top 1000 frequent words (except surnames), and stopwords. - Returns: - """ - med_terms = dd.ds.LookupSet() - med_terms.add_items_from_file( - os.path.join(data_path, "medical_terms.txt"), - ) + stopwords = dd.ds.LookupSet() + stopwords.add_items_from_file(os.path.join(self.data_path, "stop_words.txt")) - top1000 = dd.ds.LookupSet() - top1000.add_items_from_file( - os.path.join(data_path, "top_1000_terms.txt"), - ) + whitelist = dd.ds.LookupSet(matching_pipeline=[dd.str.LowercaseString()]) + whitelist.add_items_from_iterable( + med_terms + top1000 + stopwords, + cleaning_pipeline=[dd.str.FilterByLength(min_len=2)], + ) - surnames_lowercase = dd.ds.LookupSet() - surnames_lowercase.add_items_from_file( - os.path.join(data_path, "surnames.txt"), - cleaning_pipeline=[ - dd.str.LowercaseString(), - dd.str.FilterByLength(min_len=2), - ], - ) + return whitelist - top1000 = top1000 - surnames_lowercase + def _load_institutions(self) -> dd.ds.LookupSet: + """Get institutions LookupSet.""" - stopwords = dd.ds.LookupSet() - stopwords.add_items_from_file(os.path.join(data_path, "stop_words.txt")) + institutions_raw = dd.ds.LookupSet() + institutions_raw.add_items_from_file( + os.path.join(self.data_path, "institutions.txt"), + cleaning_pipeline=[dd.str.FilterByLength(min_len=3), dd.str.LowercaseString()], + ) - whitelist = dd.ds.LookupSet(matching_pipeline=[dd.str.LowercaseString()]) - whitelist.add_items_from_iterable( - med_terms + top1000 + stopwords, - cleaning_pipeline=[dd.str.FilterByLength(min_len=2)], - ) + institutions = dd.ds.LookupSet(matching_pipeline=[dd.str.LowercaseString()]) + institutions.add_items_from_iterable(institutions_raw, cleaning_pipeline=[dd.str.StripString()]) - return whitelist + institutions.add_items_from_iterable( + institutions_raw, + cleaning_pipeline=[ + RemoveValues(filter_values=["dr.", "der", "van", "de", "het", "'t", "in", "d'"]), + dd.str.StripString(), + ], + ) + institutions.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue(".", ""), dd.str.StripString()]) -def _get_institutions_lookup_set() -> dd.ds.LookupSet: - """Get institutions LookupSet.""" + institutions.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue("st ", "sint ")]) - institutions_raw = dd.ds.LookupSet() - institutions_raw.add_items_from_file( - os.path.join(data_path, "institutions.txt"), - cleaning_pipeline=[dd.str.FilterByLength(min_len=3), dd.str.LowercaseString()], - ) + institutions.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue("st. ", "sint ")]) - institutions = dd.ds.LookupSet(matching_pipeline=[dd.str.LowercaseString()]) - institutions.add_items_from_iterable(institutions_raw, cleaning_pipeline=[dd.str.StripString()]) + institutions.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue("ziekenhuis", "zkh")]) - institutions.add_items_from_iterable( - institutions_raw, - cleaning_pipeline=[ - RemoveValues(filter_values=["dr.", "der", "van", "de", "het", "'t", "in", "d'"]), - dd.str.StripString(), - ], - ) + institutions.add_items_from_self( + cleaning_pipeline=[dd.str.LowercaseString(), Acronimify(), dd.str.FilterByLength(min_len=3)] + ) - institutions.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue(".", ""), dd.str.StripString()]) + institutions = institutions - self._load_whitelist() - institutions.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue("st ", "sint ")]) + return institutions - institutions.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue("st. ", "sint ")]) + def _load_residences(self) -> dd.ds.LookupSet: + """Get residences LookupSet.""" - institutions.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue("ziekenhuis", "zkh")]) + residences = dd.ds.LookupSet() + residences.add_items_from_file( + file_path=os.path.join(self.data_path, "residences.txt"), + cleaning_pipeline=[dd.str.ReplaceValueRegexp(r"\(.+\)", ""), dd.str.StripString()], + ) - institutions.add_items_from_self( - cleaning_pipeline=[dd.str.LowercaseString(), Acronimify(), dd.str.FilterByLength(min_len=3)] - ) + residences.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue("-", " ")]) - institutions = institutions - _get_whitelist_lookup_set() + residences.add_items_from_self( + cleaning_pipeline=[FilterBasedOnLookupSet(filter_set=self._load_whitelist(), case_sensitive=False)], + replace=True, + ) - return institutions + return residences + def build_lookup_sets(self) -> dd.ds.DsCollection: + """ + Get all lookupsets. -def _get_residences_lookup_set() -> dd.ds.LookupSet: - """Get residences LookupSet.""" + Returns: + A DsCollection with all lookup sets. + """ - residences = dd.ds.LookupSet() - residences.add_items_from_file( - file_path=os.path.join(data_path, "residences.txt"), - cleaning_pipeline=[dd.str.ReplaceValueRegexp(r"\(.+\)", ""), dd.str.StripString()], - ) + lookup_sets = dd.ds.DsCollection() - residences.add_items_from_self(cleaning_pipeline=[dd.str.ReplaceValue("-", " ")]) + lookup_sets["first_names"] = self._load_first_names() + lookup_sets["surnames"] = self._load_surnames() + lookup_sets["interfixes"] = self._load_interfixes() + lookup_sets["interfix_surnames"] = self._load_interfix_surnames() + lookup_sets["prefixes"] = self._load_prefixes() + lookup_sets["whitelist"] = self._load_whitelist() + lookup_sets["institutions"] = self._load_institutions() + lookup_sets["residences"] = self._load_residences() - residences.add_items_from_self( - cleaning_pipeline=[FilterBasedOnLookupSet(filter_set=_get_whitelist_lookup_set(), case_sensitive=False)], - replace=True, - ) - - return residences - - -def get_lookup_sets() -> dd.ds.DsCollection: - """ - Get all lookupsets. - - Returns: - A DsCollection with all lookup sets. - """ - - lookup_sets = dd.ds.DsCollection() - - lookup_set_mapping = { - "first_names": _get_first_names_lookup_set, - "surnames": _get_surnames_lookup_set, - "interfixes": _get_interfixes_lookup_set, - "interfix_surnames": _get_interfix_surnames_lookup_set, - "prefixes": _get_prefixes_lookup_set, - "whitelist": _get_whitelist_lookup_set, - "institutions": _get_institutions_lookup_set, - "residences": _get_residences_lookup_set, - } - - for name, init_function in lookup_set_mapping.items(): - lookup_sets[name] = init_function() - - return lookup_sets + return lookup_sets diff --git a/tests/unit/pattern/test_name.py b/tests/unit/pattern/test_name.py index c5ab9b1..626a52d 100644 --- a/tests/unit/pattern/test_name.py +++ b/tests/unit/pattern/test_name.py @@ -1,7 +1,6 @@ import docdeid as dd import pytest - -from deduce.lookup_sets import get_lookup_sets +from deduce.deduce import Deduce from deduce.pattern.name import ( FirstNameLookupPattern, InitiaalInterfixCapitalPattern, @@ -12,7 +11,8 @@ ) from tests.helpers import linked_tokens -lookup_sets = get_lookup_sets() +config = Deduce.initialize_config() +lookup_sets = Deduce.initialize_lookupsets(config) @pytest.fixture diff --git a/tests/unit/pattern/test_name_context.py b/tests/unit/pattern/test_name_context.py index 9c53da1..f5e58c7 100644 --- a/tests/unit/pattern/test_name_context.py +++ b/tests/unit/pattern/test_name_context.py @@ -1,6 +1,6 @@ import docdeid as dd -from deduce.lookup_sets import get_lookup_sets +from deduce.deduce import Deduce from deduce.pattern.name_context import ( InitialNameContextPattern, InitialsContextPattern, @@ -9,7 +9,8 @@ ) from tests.helpers import linked_tokens -lookup_sets = get_lookup_sets() +config = Deduce.initialize_config() +lookup_sets = Deduce.initialize_lookupsets(config) class TestInterfixContextPattern: diff --git a/tests/unit/test_deduce_processors.py b/tests/unit/test_deduce_processors.py index 6ea92f8..89c9c24 100644 --- a/tests/unit/test_deduce_processors.py +++ b/tests/unit/test_deduce_processors.py @@ -3,11 +3,10 @@ import docdeid as dd from deduce.deduce import Deduce -from deduce.lookup_sets import get_lookup_sets from deduce.tokenize import DeduceTokenizer -config = Deduce._initialize_config() -lookup_sets = get_lookup_sets() +config = Deduce.initialize_config() +lookup_sets = Deduce.initialize_lookupsets(config) tokenizer = DeduceTokenizer() deduce_processors = Deduce._initialize_annotators(config["annotators"].copy(), lookup_sets, tokenizer)