social-determinants-of-health-clbp/model-cnn-ner/postprocess.py at main · BCHSI/social-determinants-of-health-clbp · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import json
import logging

import spacy
from spacy.language import Language
from spacy.tokens import Span

Span.set_extension("cui", default=None, force=True)
Span.set_extension("snomed_ct", default=None, force=True)
Span.set_extension("canonical_text", default=None, force=True)

#import sys
#print("file:", __file__, file=sys.stderr)
#import ipdb; ipdb.set_trace()

@Language.factory("sdoh_cui")
class SDOH:
    def __init__(self, nlp, name="sdoh_cui"):
        self.data = {}

    def __call__(self, doc):
        for ent in doc.ents:
            entry = self.data.get(ent.label_, None)
            if entry is not None:
                cui = entry.get("CUI", None)
                if cui:
                    ent._.set("cui", cui)
                snomed_ct = entry.get("SNOMED_CT", None)
                if snomed_ct:
                    ent._.set("snomed_ct", snomed_ct)
                canonical_text = entry.get("canonical_text", None)
                if snomed_ct:
                    ent._.set("canonical_text", canonical_text)
            else:
                logging.warning(f"cannot find an entry for {ent} ({ent.label_}) in {len(self.data)} entries")
        return doc

    def add(self, key, value):
        # Add something to the component's data
        self.data[key] = value

    def to_disk(self, path, exclude=tuple()):
        # This will receive the directory path + /my_component
        data_path = path / "mapping-ontology-to-umls.json"
        with data_path.open("w", encoding="utf8") as f:
            f.write(json.dumps(self.data))

    def from_disk(self, path, exclude=tuple()):
        # This will receive the directory path + /my_component
        data_path = path / "mapping-ontology-to-umls.json"
        with data_path.open("r", encoding="utf8") as f:
            self.data = json.load(f)
        return self

    def initalize(self, get_examples=None, nlp=None, data={}):
        logging.info("initializing sdoh_cui")
        logging.info(f"data: {data}")
        self.data = data

# with open("./mapping-ontology-to-umls.json") as fh:
#     meta = json.load(fh)
#
#
# @Language.component("sdoh_cui")
# def add_sdoh_cui(doc):
#     for ent in doc.ents:
#         entry = meta.get(ent.label_, None)
#
#         if entry is not None:
#             cui = entry.get("CUI", None)
#             if cui:
#                 ent._.set("cui", cui)
#             snomed_ct = entry.get("SNOMED_CT", None)
#             if snomed_ct:
#                 ent._.set("snomed_ct", snomed_ct)
#             canonical_text = entry.get("canonical_text", None)
#             if snomed_ct:
#                 ent._.set("canonical_text", canonical_text)
#         else:
#             logging.warning(f"cannot find an entry for {ent} ({ent.label_})")
#     return doc
#