a b/scripts/serve.py
1
from typing import List
2
3
import spacy
4
from fastapi import FastAPI
5
from pydantic import BaseModel
6
7
import edsnlp
8
9
app = FastAPI(title="EDS-NLP", version=edsnlp.__version__)
10
11
nlp = spacy.blank("eds")
12
13
nlp.add_pipe("eds.sentences")
14
15
config = dict(
16
    regex=dict(
17
        covid=[
18
            "covid",
19
            r"covid[-\s]?19",
20
            r"sars[-\s]?cov[-\s]?2",
21
            r"corona[-\s]?virus",
22
        ],
23
    ),
24
    attr="LOWER",
25
)
26
nlp.add_pipe("eds.matcher", config=config)
27
28
nlp.add_pipe("eds.negation")
29
nlp.add_pipe("eds.family")
30
nlp.add_pipe("eds.hypothesis")
31
nlp.add_pipe("eds.reported_speech")
32
33
34
class Entity(BaseModel):  # (2)
35
36
    # OMOP-style attributes
37
    start: int
38
    end: int
39
    label: str
40
    lexical_variant: str
41
    normalized_variant: str
42
43
    # Qualifiers
44
    negated: bool
45
    hypothesis: bool
46
    family: bool
47
    reported_speech: bool
48
49
50
class Document(BaseModel):  # (1)
51
    text: str
52
    ents: List[Entity]
53
54
55
@app.post("/process", response_model=List[Document])  # (1)
56
async def process(
57
    notes: List[str],  # (2)
58
):
59
60
    documents = []
61
62
    for doc in nlp.pipe(notes):
63
        entities = []
64
65
        for ent in doc.ents:
66
            entity = Entity(
67
                start=ent.start_char,
68
                end=ent.end_char,
69
                label=ent.label_,
70
                lexical_variant=ent.text,
71
                normalized_variant=ent._.normalized_variant,
72
                negated=ent._.negation,
73
                hypothesis=ent._.hypothesis,
74
                family=ent._.family,
75
                reported_speech=ent._.reported_speech,
76
            )
77
            entities.append(entity)
78
79
        documents.append(
80
            Document(
81
                text=doc.text,
82
                ents=entities,
83
            )
84
        )
85
86
    return documents