Skip to content

edsnlp.pipelines.core

sentences

terms

punctuation = ['!', '.', '?', '։', '؟', '۔', '܀', '܁', '܂', '߹', '।', '॥', '၊', '။', '።', '፧', '፨', '᙮', '᜵', '᜶', '᠃', '᠉', '᥄', '᥅', '᪨', '᪩', '᪪', '᪫', '᭚', '᭛', '᭞', '᭟', '᰻', '᰼', '᱾', '᱿', '‼', '‽', '⁇', '⁈', '⁉', '⸮', '⸼', '꓿', '꘎', '꘏', '꛳', '꛷', '꡶', '꡷', '꣎', '꣏', '꤯', '꧈', '꧉', '꩝', '꩞', '꩟', '꫰', '꫱', '꯫', '﹒', '﹖', '﹗', '！', '．', '？', '𐩖', '𐩗', '𑁇', '𑁈', '𑂾', '𑂿', '𑃀', '𑃁', '𑅁', '𑅂', '𑅃', '𑇅', '𑇆', '𑇍', '𑇞', '𑇟', '𑈸', '𑈹', '𑈻', '𑈼', '𑊩', '𑑋', '𑑌', '𑗂', '𑗃', '𑗉', '𑗊', '𑗋', '𑗌', '𑗍', '𑗎', '𑗏', '𑗐', '𑗑', '𑗒', '𑗓', '𑗔', '𑗕', '𑗖', '𑗗', '𑙁', '𑙂', '𑜼', '𑜽', '𑜾', '𑩂', '𑩃', '𑪛', '𑪜', '𑱁', '𑱂', '𖩮', '𖩯', '𖫵', '𖬷', '𖬸', '𖭄', '𛲟', '𝪈', '｡', '。'] module-attribute

sentences

SentenceSegmenter

Bases: object

Segments the Doc into sentences using a rule-based strategy, specific to AP-HP documents.

Applies the same rule-based pipeline as spaCy's sentencizer, and adds a simple rule on the new lines : if a new line is followed by a capitalised word, then it is also an end of sentence.

DOCS: https://spacy.io/api/sentencizer

Arguments

punct_chars : Optional[List[str]] Punctuation characters. use_endlines : bool Whether to use endlines prediction.

Source code in edsnlp/pipelines/core/sentences/sentences.py 
 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
83
84
85
86
87
88
89
class SentenceSegmenter(object):
    """
    Segments the Doc into sentences using a rule-based strategy,
    specific to AP-HP documents.

    Applies the same rule-based pipeline as spaCy's sentencizer,
    and adds a simple rule on the new lines : if a new line is followed by a
    capitalised word, then it is also an end of sentence.

    DOCS: https://spacy.io/api/sentencizer

    Arguments
    ---------
    punct_chars : Optional[List[str]]
        Punctuation characters.
    use_endlines : bool
        Whether to use endlines prediction.
    """

    def __init__(
        self,
        punct_chars: Optional[List[str]],
        use_endlines: bool,
    ):

        if punct_chars is None:
            punct_chars = punctuation

        self.punct_chars = set(punct_chars)
        self.use_endlines = use_endlines

    def __call__(self, doc: Doc) -> Doc:
        """
        Segments the document in sentences.

        Arguments
        ---------
        doc:
            A spacy Doc object.

        Returns
        -------
        doc:
            A spaCy Doc object, annotated for sentences.
        """

        if not doc:
            return doc

        doc[0].sent_start = True

        seen_period = False
        seen_newline = False

        for i, token in enumerate(doc):
            is_in_punct_chars = token.text in self.punct_chars
            is_newline = token.is_space and "\n" in token.text

            if self.use_endlines:
                end_line = getattr(token._, "end_line", None)
                is_newline = is_newline and (end_line or end_line is None)

            token.sent_start = (
                i == 0
            )  # To set the attributes at False by default for the other tokens
            if seen_period or seen_newline:
                if token.is_punct or is_in_punct_chars or is_newline:
                    continue
                if seen_period:
                    token.sent_start = True
                    seen_newline = False
                    seen_period = False
                else:
                    token.sent_start = token.shape_.startswith("Xx")
                    seen_newline = False
                    seen_period = False
            elif is_in_punct_chars:
                seen_period = True
            elif is_newline:
                seen_newline = True

        return doc
punct_chars = set(punct_chars) instance-attribute
use_endlines = use_endlines instance-attribute
__init__(punct_chars, use_endlines)
Source code in edsnlp/pipelines/core/sentences/sentences.py 
27
28
29
30
31
32
33
34
35
36
37
def __init__(
    self,
    punct_chars: Optional[List[str]],
    use_endlines: bool,
):

    if punct_chars is None:
        punct_chars = punctuation

    self.punct_chars = set(punct_chars)
    self.use_endlines = use_endlines
__call__(doc)

Segments the document in sentences.

Arguments

doc: A spacy Doc object.

RETURNS DESCRIPTION
doc

A spaCy Doc object, annotated for sentences.
Source code in edsnlp/pipelines/core/sentences/sentences.py 
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
83
84
85
86
87
88
89
def __call__(self, doc: Doc) -> Doc:
    """
    Segments the document in sentences.

    Arguments
    ---------
    doc:
        A spacy Doc object.

    Returns
    -------
    doc:
        A spaCy Doc object, annotated for sentences.
    """

    if not doc:
        return doc

    doc[0].sent_start = True

    seen_period = False
    seen_newline = False

    for i, token in enumerate(doc):
        is_in_punct_chars = token.text in self.punct_chars
        is_newline = token.is_space and "\n" in token.text

        if self.use_endlines:
            end_line = getattr(token._, "end_line", None)
            is_newline = is_newline and (end_line or end_line is None)

        token.sent_start = (
            i == 0
        )  # To set the attributes at False by default for the other tokens
        if seen_period or seen_newline:
            if token.is_punct or is_in_punct_chars or is_newline:
                continue
            if seen_period:
                token.sent_start = True
                seen_newline = False
                seen_period = False
            else:
                token.sent_start = token.shape_.startswith("Xx")
                seen_newline = False
                seen_period = False
        elif is_in_punct_chars:
            seen_period = True
        elif is_newline:
            seen_newline = True

    return doc

factory

DEFAULT_CONFIG = dict(punct_chars=None, use_endlines=True) module-attribute
create_component(nlp, name, punct_chars, use_endlines)
Source code in edsnlp/pipelines/core/sentences/factory.py 
15
16
17
18
19
20
21
22
23
24
25
26
@deprecated_factory("sentences", "eds.sentences", default_config=DEFAULT_CONFIG)
@Language.factory("eds.sentences", default_config=DEFAULT_CONFIG)
def create_component(
    nlp: Language,
    name: str,
    punct_chars: Optional[List[str]],
    use_endlines: bool,
):
    return SentenceSegmenter(
        punct_chars=punct_chars,
        use_endlines=use_endlines,
    )

matcher

matcher

GenericMatcher

Bases: BaseComponent

Provides a generic matcher component.

PARAMETER DESCRIPTION
nlp

The spaCy object.

TYPE: Language

terms

A dictionary of terms.

TYPE: Optional[Patterns]

regex

A dictionary of regular expressions.

TYPE: Optional[Patterns]

attr

The default attribute to use for matching. Can be overiden using the terms and regex configurations.

TYPE: str

filter_matches

Whether to filter out matches.

TYPE: bool

on_ents_only

Whether to to look for matches around pre-extracted entities only.

TYPE: bool

ignore_excluded

Whether to skip excluded tokens (requires an upstream pipeline to mark excluded tokens).

TYPE: bool

Source code in edsnlp/pipelines/core/matcher/matcher.py 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class GenericMatcher(BaseComponent):
    """
    Provides a generic matcher component.

    Parameters
    ----------
    nlp : Language
        The spaCy object.
    terms : Optional[Patterns]
        A dictionary of terms.
    regex : Optional[Patterns]
        A dictionary of regular expressions.
    attr : str
        The default attribute to use for matching.
        Can be overiden using the `terms` and `regex` configurations.
    filter_matches : bool
        Whether to filter out matches.
    on_ents_only : bool
        Whether to to look for matches around pre-extracted entities only.
    ignore_excluded : bool
        Whether to skip excluded tokens (requires an upstream
        pipeline to mark excluded tokens).
    """

    def __init__(
        self,
        nlp: Language,
        terms: Optional[Patterns],
        regex: Optional[Patterns],
        attr: str,
        ignore_excluded: bool,
    ):

        self.nlp = nlp

        self.attr = attr

        self.phrase_matcher = EDSPhraseMatcher(
            self.nlp.vocab,
            attr=attr,
            ignore_excluded=ignore_excluded,
        )
        self.regex_matcher = RegexMatcher(
            attr=attr,
            ignore_excluded=ignore_excluded,
        )

        self.phrase_matcher.build_patterns(nlp=nlp, terms=terms)
        self.regex_matcher.build_patterns(regex=regex)

        self.set_extensions()

    def process(self, doc: Doc) -> List[Span]:
        """
        Find matching spans in doc.

        Parameters
        ----------
        doc:
            spaCy Doc object.

        Returns
        -------
        spans:
            List of Spans returned by the matchers.
        """

        matches = self.phrase_matcher(doc, as_spans=True)
        regex_matches = self.regex_matcher(doc, as_spans=True)

        spans = list(matches) + list(regex_matches)

        return spans

    def __call__(self, doc: Doc) -> Doc:
        """
        Adds spans to document.

        Parameters
        ----------
        doc:
            spaCy Doc object

        Returns
        -------
        doc:
            spaCy Doc object, annotated for extracted terms.
        """
        matches = self.process(doc)

        for span in matches:
            if span.label_ not in doc.spans:
                doc.spans[span.label_] = []
            doc.spans[span.label_].append(span)

        ents, discarded = filter_spans(list(doc.ents) + matches, return_discarded=True)

        doc.ents = ents

        if "discarded" not in doc.spans:
            doc.spans["discarded"] = []
        doc.spans["discarded"].extend(discarded)

        return doc
nlp = nlp instance-attribute
attr = attr instance-attribute
phrase_matcher = EDSPhraseMatcher(self.nlp.vocab, attr=attr, ignore_excluded=ignore_excluded) instance-attribute
regex_matcher = RegexMatcher(attr=attr, ignore_excluded=ignore_excluded) instance-attribute
__init__(nlp, terms, regex, attr, ignore_excluded)
Source code in edsnlp/pipelines/core/matcher/matcher.py 
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
def __init__(
    self,
    nlp: Language,
    terms: Optional[Patterns],
    regex: Optional[Patterns],
    attr: str,
    ignore_excluded: bool,
):

    self.nlp = nlp

    self.attr = attr

    self.phrase_matcher = EDSPhraseMatcher(
        self.nlp.vocab,
        attr=attr,
        ignore_excluded=ignore_excluded,
    )
    self.regex_matcher = RegexMatcher(
        attr=attr,
        ignore_excluded=ignore_excluded,
    )

    self.phrase_matcher.build_patterns(nlp=nlp, terms=terms)
    self.regex_matcher.build_patterns(regex=regex)

    self.set_extensions()
process(doc)

Find matching spans in doc.

PARAMETER DESCRIPTION
doc

spaCy Doc object.

TYPE: Doc

RETURNS DESCRIPTION
spans

List of Spans returned by the matchers.
Source code in edsnlp/pipelines/core/matcher/matcher.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def process(self, doc: Doc) -> List[Span]:
    """
    Find matching spans in doc.

    Parameters
    ----------
    doc:
        spaCy Doc object.

    Returns
    -------
    spans:
        List of Spans returned by the matchers.
    """

    matches = self.phrase_matcher(doc, as_spans=True)
    regex_matches = self.regex_matcher(doc, as_spans=True)

    spans = list(matches) + list(regex_matches)

    return spans
__call__(doc)

Adds spans to document.

PARAMETER DESCRIPTION
doc

spaCy Doc object

TYPE: Doc

RETURNS DESCRIPTION
doc

spaCy Doc object, annotated for extracted terms.
Source code in edsnlp/pipelines/core/matcher/matcher.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def __call__(self, doc: Doc) -> Doc:
    """
    Adds spans to document.

    Parameters
    ----------
    doc:
        spaCy Doc object

    Returns
    -------
    doc:
        spaCy Doc object, annotated for extracted terms.
    """
    matches = self.process(doc)

    for span in matches:
        if span.label_ not in doc.spans:
            doc.spans[span.label_] = []
        doc.spans[span.label_].append(span)

    ents, discarded = filter_spans(list(doc.ents) + matches, return_discarded=True)

    doc.ents = ents

    if "discarded" not in doc.spans:
        doc.spans["discarded"] = []
    doc.spans["discarded"].extend(discarded)

    return doc

factory

DEFAULT_CONFIG = dict(terms=None, regex=None, attr='TEXT', ignore_excluded=False) module-attribute
create_component(nlp, name, terms, attr, regex, ignore_excluded)
Source code in edsnlp/pipelines/core/matcher/factory.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@deprecated_factory("matcher", "eds.matcher", default_config=DEFAULT_CONFIG)
@Language.factory("eds.matcher", default_config=DEFAULT_CONFIG)
def create_component(
    nlp: Language,
    name: str,
    terms: Optional[Dict[str, Union[str, List[str]]]],
    attr: Union[str, Dict[str, str]],
    regex: Optional[Dict[str, Union[str, List[str]]]],
    ignore_excluded: bool,
):
    assert not (terms is None and regex is None)

    if terms is None:
        terms = dict()
    if regex is None:
        regex = dict()

    return GenericMatcher(
        nlp,
        terms=terms,
        attr=attr,
        regex=regex,
        ignore_excluded=ignore_excluded,
    )

endlines

functional

_get_label(prediction)

Returns the label for the prediction PREDICTED_END_LINE

PARAMETER DESCRIPTION
prediction

value of PREDICTED_END_LINE

TYPE: bool

RETURNS DESCRIPTION
str

Label for PREDICTED_END_LINE
Source code in edsnlp/pipelines/core/endlines/functional.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def _get_label(prediction: bool) -> str:
    """Returns the label for the prediction `PREDICTED_END_LINE`

    Parameters
    ----------
    prediction : bool
        value of `PREDICTED_END_LINE`

    Returns
    -------
    str
        Label for `PREDICTED_END_LINE`
    """
    if prediction:
        return "end_line"
    else:
        return "space"
get_dir_path(file)
Source code in edsnlp/pipelines/core/endlines/functional.py 
26
27
28
def get_dir_path(file):
    path_file = os.path.dirname(os.path.realpath(file))
    return path_file
build_path(file, relative_path)

Function to build an absolut path.

PARAMETER DESCRIPTION
file

relative_path

relative path from the main file to the desired output
RETURNS DESCRIPTION
path
Source code in edsnlp/pipelines/core/endlines/functional.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def build_path(file, relative_path):
    """
    Function to build an absolut path.

    Parameters
    ----------
    file: main file from where we are calling. It could be __file__
    relative_path: str,
        relative path from the main file to the desired output

    Returns
    -------
    path: absolute path
    """
    dir_path = get_dir_path(file)
    path = os.path.abspath(os.path.join(dir_path, relative_path))
    return path
_convert_series_to_array(s)

Converts pandas series of n elements to an array of shape (n,1).

PARAMETER DESCRIPTION
s

TYPE: pd.Series

RETURNS DESCRIPTION
np.ndarray
Source code in edsnlp/pipelines/core/endlines/functional.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
def _convert_series_to_array(s: pd.Series) -> np.ndarray:
    """Converts pandas series of n elements to an array of shape (n,1).

    Parameters
    ----------
    s : pd.Series

    Returns
    -------
    np.ndarray
    """
    X = s.to_numpy().reshape(-1, 1).astype("O")  # .astype(np.int64)
    return X

endlines

EndLines

Bases: GenericMatcher

spaCy Pipeline to detect whether a newline character should be considered a space (ie introduced by the PDF).

The pipeline will add the extension end_line to spans and tokens. The end_line attribute is a boolean or None, set to True if the pipeline predicts that the new line is an end line character. Otherwise, it is set to False if the new line is classified as a space. If no classification has been done over that token, it will remain None.

PARAMETER DESCRIPTION
nlp

spaCy nlp pipeline to use for matching.

TYPE: Language

end_lines_model : Optional[Union[str, EndLinesModel]], by default None path to trained model. If None, it will use a default model

Source code in edsnlp/pipelines/core/endlines/endlines.py 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
class EndLines(GenericMatcher):
    """
    spaCy Pipeline to detect whether a newline character should
    be considered a space (ie introduced by the PDF).

    The pipeline will add the extension `end_line` to spans
    and tokens. The `end_line` attribute is a boolean or `None`,
    set to `True` if the pipeline predicts that the new line
    is an end line character. Otherwise, it is  set to `False`
    if the new line is classified as a space. If no classification
    has been done over that token, it will remain `None`.

    Parameters
    ----------
    nlp : Language
        spaCy nlp pipeline to use for matching.

    end_lines_model : Optional[Union[str, EndLinesModel]], by default None
        path to trained model. If None, it will use a default model
    """

    def __init__(
        self,
        nlp: Language,
        end_lines_model: Optional[Union[str, EndLinesModel]],
        **kwargs,
    ):

        super().__init__(
            nlp,
            terms=None,
            attr="TEXT",
            regex=dict(
                new_line=r"\n+",
            ),
            ignore_excluded=False,
            **kwargs,
        )

        if not Token.has_extension("end_line"):
            Token.set_extension("end_line", default=None)

        if not Span.has_extension("end_line"):
            Span.set_extension("end_line", default=None)

        self._read_model(end_lines_model)

    def _read_model(self, end_lines_model: Optional[Union[str, EndLinesModel]]):
        """
        Parameters
        ----------
        end_lines_model : Optional[Union[str, EndLinesModel]]

        Raises
        ------
        TypeError
        """
        if end_lines_model is None:
            path = build_path(__file__, "base_model.pkl")

            with open(path, "rb") as inp:
                self.model = pickle.load(inp)
        elif type(end_lines_model) == str:
            with open(end_lines_model, "rb") as inp:
                self.model = pickle.load(inp)
        elif type(end_lines_model) == EndLinesModel:
            self.model = end_lines_model
        else:
            raise TypeError(
                "type(`end_lines_model`) should be one of {None, str, EndLinesModel}"
            )

    @staticmethod
    def _spacy_compute_a3a4(token: Token) -> str:
        """Function to compute A3 and A4

        Parameters
        ----------
        token : Token

        Returns
        -------
        str
        """

        if token.is_upper:
            return "UPPER"

        elif token.shape_.startswith("Xx"):
            return "S_UPPER"

        elif token.shape_.startswith("x"):
            return "LOWER"

        elif (token.is_digit) & (
            (token.doc[max(token.i - 1, 0)].is_punct)
            | (token.doc[min(token.i + 1, len(token.doc) - 1)].is_punct)
        ):
            return "ENUMERATION"

        elif token.is_digit:
            return "DIGIT"

        elif (token.is_punct) & (token.text in [".", ";", "..", "..."]):
            return "STRONG_PUNCT"

        elif (token.is_punct) & (token.text not in [".", ";", "..", "..."]):
            return "SOFT_PUNCT"

        else:
            return "OTHER"

    @staticmethod
    def _compute_length(doc: Doc, start: int, end: int) -> int:
        """Compute length without spaces

        Parameters
        ----------
        doc : Doc
        start : int
        end : int

        Returns
        -------
        int
        """
        length = 0
        for t in doc[start:end]:
            length += len(t.text)

        return length

    def _get_df(self, doc: Doc, new_lines: List[Span]) -> pd.DataFrame:
        """Get a pandas DataFrame to call the classifier

        Parameters
        ----------
        doc : Doc
        new_lines : List[Span]

        Returns
        -------
        pd.DataFrame
        """

        data = []
        for i, span in enumerate(new_lines):
            start = span.start
            end = span.end

            max_index = len(doc) - 1
            a1_token = doc[max(start - 1, 0)]
            a2_token = doc[min(start + 1, max_index)]
            a1 = a1_token.orth
            a2 = a2_token.orth
            a3 = self._spacy_compute_a3a4(a1_token)
            a4 = self._spacy_compute_a3a4(a2_token)
            blank_line = "\n\n" in span.text

            if i > 0:
                start_previous = new_lines[i - 1].start + 1
            else:
                start_previous = 0

            length = self._compute_length(
                doc, start=start_previous, end=start
            )  # It's ok cause i count the total length from the previous up to this one

            data_dict = dict(
                span_start=start,
                span_end=end,
                A1=a1,
                A2=a2,
                A3=a3,
                A4=a4,
                BLANK_LINE=blank_line,
                length=length,
            )
            data.append(data_dict)

        df = pd.DataFrame(data)

        mu = df["length"].mean()
        sigma = df["length"].std()
        if np.isnan(sigma):
            sigma = 1

        cv = sigma / mu
        df["B1"] = (df["length"] - mu) / sigma
        df["B2"] = cv

        return df

    def __call__(self, doc: Doc) -> Doc:
        """
        Predict for each new line if it's an end of line or a space.

        Parameters
        ----------
        doc: spaCy Doc object

        Returns
        -------
        doc: spaCy Doc object, with each new line annotated
        """

        matches = self.process(doc)
        new_lines = get_spans(matches, "new_line")

        if len(new_lines) > 0:
            df = self._get_df(doc=doc, new_lines=new_lines)
            df = self.model.predict(df)

            spans = []
            for span, prediction in zip(new_lines, df.PREDICTED_END_LINE):

                span.label_ = _get_label(prediction)
                span._.end_line = prediction

                spans.append(span)
                for t in span:
                    t._.end_line = prediction
                    if not prediction:
                        t._.excluded = True

            doc.spans["new_lines"] = spans
        return doc
__init__(nlp, end_lines_model, **kwargs)
Source code in edsnlp/pipelines/core/endlines/endlines.py 
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
def __init__(
    self,
    nlp: Language,
    end_lines_model: Optional[Union[str, EndLinesModel]],
    **kwargs,
):

    super().__init__(
        nlp,
        terms=None,
        attr="TEXT",
        regex=dict(
            new_line=r"\n+",
        ),
        ignore_excluded=False,
        **kwargs,
    )

    if not Token.has_extension("end_line"):
        Token.set_extension("end_line", default=None)

    if not Span.has_extension("end_line"):
        Span.set_extension("end_line", default=None)

    self._read_model(end_lines_model)
_read_model(end_lines_model)
PARAMETER DESCRIPTION
end_lines_model

TYPE: Optional[Union[str, EndLinesModel]]

RAISES DESCRIPTION
TypeError
Source code in edsnlp/pipelines/core/endlines/endlines.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def _read_model(self, end_lines_model: Optional[Union[str, EndLinesModel]]):
    """
    Parameters
    ----------
    end_lines_model : Optional[Union[str, EndLinesModel]]

    Raises
    ------
    TypeError
    """
    if end_lines_model is None:
        path = build_path(__file__, "base_model.pkl")

        with open(path, "rb") as inp:
            self.model = pickle.load(inp)
    elif type(end_lines_model) == str:
        with open(end_lines_model, "rb") as inp:
            self.model = pickle.load(inp)
    elif type(end_lines_model) == EndLinesModel:
        self.model = end_lines_model
    else:
        raise TypeError(
            "type(`end_lines_model`) should be one of {None, str, EndLinesModel}"
        )
_spacy_compute_a3a4(token)

Function to compute A3 and A4

PARAMETER DESCRIPTION
token

TYPE: Token

RETURNS DESCRIPTION
str
Source code in edsnlp/pipelines/core/endlines/endlines.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@staticmethod
def _spacy_compute_a3a4(token: Token) -> str:
    """Function to compute A3 and A4

    Parameters
    ----------
    token : Token

    Returns
    -------
    str
    """

    if token.is_upper:
        return "UPPER"

    elif token.shape_.startswith("Xx"):
        return "S_UPPER"

    elif token.shape_.startswith("x"):
        return "LOWER"

    elif (token.is_digit) & (
        (token.doc[max(token.i - 1, 0)].is_punct)
        | (token.doc[min(token.i + 1, len(token.doc) - 1)].is_punct)
    ):
        return "ENUMERATION"

    elif token.is_digit:
        return "DIGIT"

    elif (token.is_punct) & (token.text in [".", ";", "..", "..."]):
        return "STRONG_PUNCT"

    elif (token.is_punct) & (token.text not in [".", ";", "..", "..."]):
        return "SOFT_PUNCT"

    else:
        return "OTHER"
_compute_length(doc, start, end)

Compute length without spaces

PARAMETER DESCRIPTION
doc

TYPE: Doc

start

TYPE: int

end

TYPE: int

RETURNS DESCRIPTION
int
Source code in edsnlp/pipelines/core/endlines/endlines.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@staticmethod
def _compute_length(doc: Doc, start: int, end: int) -> int:
    """Compute length without spaces

    Parameters
    ----------
    doc : Doc
    start : int
    end : int

    Returns
    -------
    int
    """
    length = 0
    for t in doc[start:end]:
        length += len(t.text)

    return length
_get_df(doc, new_lines)

Get a pandas DataFrame to call the classifier

PARAMETER DESCRIPTION
doc

TYPE: Doc

new_lines

TYPE: List[Span]

RETURNS DESCRIPTION
pd.DataFrame
Source code in edsnlp/pipelines/core/endlines/endlines.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def _get_df(self, doc: Doc, new_lines: List[Span]) -> pd.DataFrame:
    """Get a pandas DataFrame to call the classifier

    Parameters
    ----------
    doc : Doc
    new_lines : List[Span]

    Returns
    -------
    pd.DataFrame
    """

    data = []
    for i, span in enumerate(new_lines):
        start = span.start
        end = span.end

        max_index = len(doc) - 1
        a1_token = doc[max(start - 1, 0)]
        a2_token = doc[min(start + 1, max_index)]
        a1 = a1_token.orth
        a2 = a2_token.orth
        a3 = self._spacy_compute_a3a4(a1_token)
        a4 = self._spacy_compute_a3a4(a2_token)
        blank_line = "\n\n" in span.text

        if i > 0:
            start_previous = new_lines[i - 1].start + 1
        else:
            start_previous = 0

        length = self._compute_length(
            doc, start=start_previous, end=start
        )  # It's ok cause i count the total length from the previous up to this one

        data_dict = dict(
            span_start=start,
            span_end=end,
            A1=a1,
            A2=a2,
            A3=a3,
            A4=a4,
            BLANK_LINE=blank_line,
            length=length,
        )
        data.append(data_dict)

    df = pd.DataFrame(data)

    mu = df["length"].mean()
    sigma = df["length"].std()
    if np.isnan(sigma):
        sigma = 1

    cv = sigma / mu
    df["B1"] = (df["length"] - mu) / sigma
    df["B2"] = cv

    return df
__call__(doc)

Predict for each new line if it's an end of line or a space.

PARAMETER DESCRIPTION
doc

TYPE: Doc

RETURNS DESCRIPTION
doc
Source code in edsnlp/pipelines/core/endlines/endlines.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def __call__(self, doc: Doc) -> Doc:
    """
    Predict for each new line if it's an end of line or a space.

    Parameters
    ----------
    doc: spaCy Doc object

    Returns
    -------
    doc: spaCy Doc object, with each new line annotated
    """

    matches = self.process(doc)
    new_lines = get_spans(matches, "new_line")

    if len(new_lines) > 0:
        df = self._get_df(doc=doc, new_lines=new_lines)
        df = self.model.predict(df)

        spans = []
        for span, prediction in zip(new_lines, df.PREDICTED_END_LINE):

            span.label_ = _get_label(prediction)
            span._.end_line = prediction

            spans.append(span)
            for t in span:
                t._.end_line = prediction
                if not prediction:
                    t._.excluded = True

        doc.spans["new_lines"] = spans
    return doc

factory

create_component(nlp, name, model_path)
Source code in edsnlp/pipelines/core/endlines/factory.py 
10
11
12
13
14
15
16
17
@deprecated_factory("endlines", "eds.endlines")
@Language.factory("eds.endlines")
def create_component(
    nlp: Language,
    name: str,
    model_path: Optional[str],
):
    return EndLines(nlp, end_lines_model=model_path)

endlinesmodel

EndLinesModel

Model to classify if an end line is a real one or it should be a space.

PARAMETER DESCRIPTION
nlp

spaCy nlp pipeline to use for matching.

TYPE: Language

Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
class EndLinesModel:
    """Model to classify if an end line is a real one or it should be a space.

    Parameters
    ----------
    nlp : Language
        spaCy nlp pipeline to use for matching.
    """

    def __init__(self, nlp: Language):
        self.nlp = nlp

    def _preprocess_data(self, corpus: Iterable[Doc]) -> pd.DataFrame:
        """
        Parameters
        ----------
        corpus : Iterable[Doc]
            Corpus of documents

        Returns
        -------
        pd.DataFrame
            Preprocessed data
        """
        # Extract the vocabulary
        string_store = self.nlp.vocab.strings

        # Iterate in the corpus and construct a dataframe
        train_data_list = []
        for i, doc in enumerate(corpus):
            train_data_list.append(self._get_attributes(doc, i))

        df = pd.concat(train_data_list)
        df.reset_index(inplace=True, drop=False)
        df.rename(columns={"ORTH": "A1", "index": "original_token_index"}, inplace=True)

        # Retrieve string representation of token_id and shape
        df["TEXT"] = df.A1.apply(self._get_string, string_store=string_store)
        df["SHAPE_"] = df.SHAPE.apply(self._get_string, string_store=string_store)

        # Convert new lines as an attribute instead of a row
        df = self._convert_line_to_attribute(df, expr="\n", col="END_LINE")
        df = self._convert_line_to_attribute(df, expr="\n\n", col="BLANK_LINE")
        df = df.loc[~(df.END_LINE | df.BLANK_LINE)]
        df = df.drop(columns="END_LINE")
        df = df.drop(columns="BLANK_LINE")
        df.rename(
            columns={"TEMP_END_LINE": "END_LINE", "TEMP_BLANK_LINE": "BLANK_LINE"},
            inplace=True,
        )

        # Construct A2 by shifting
        df = self._shift_col(df, "A1", "A2", direction="backward")

        # Compute A3 and A4
        df = self._compute_a3(df)
        df = self._shift_col(df, "A3", "A4", direction="backward")

        # SPACE is the class to predict. Set 1 if not an END_LINE
        df["SPACE"] = np.logical_not(df["END_LINE"]).astype("int")

        df[["END_LINE", "BLANK_LINE"]] = df[["END_LINE", "BLANK_LINE"]].fillna(
            True, inplace=False
        )

        # Assign a sentence id to each token
        df = df.groupby("DOC_ID").apply(self._retrieve_lines)
        df["SENTENCE_ID"] = df["SENTENCE_ID"].astype("int")

        # Compute B1 and B2
        df = self._compute_B(df)

        # Drop Tokens without info (last token of doc)
        df.dropna(subset=["A1", "A2", "A3", "A4"], inplace=True)

        # Export the vocabularies to be able to use the model with another corpus
        voc_a3a4 = self._create_vocabulary(df.A3_.cat.categories)
        voc_B2 = self._create_vocabulary(df.cv_bin.cat.categories)
        voc_B1 = self._create_vocabulary(df.l_norm_bin.cat.categories)

        vocabulary = {"A3A4": voc_a3a4, "B1": voc_B1, "B2": voc_B2}

        self.vocabulary = vocabulary

        return df

    def fit_and_predict(self, corpus: Iterable[Doc]) -> pd.DataFrame:
        """Fit the model and predict for the training data

        Parameters
        ----------
        corpus : Iterable[Doc]
            An iterable of Documents

        Returns
        -------
        pd.DataFrame
            one line by end_line prediction
        """

        # Preprocess data to have a pd DF
        df = self._preprocess_data(corpus)

        # Train and predict M1
        self._fit_M1(df.A1, df.A2, df.A3, df.A4, df.SPACE)
        outputs_M1 = self._predict_M1(
            df.A1,
            df.A2,
            df.A3,
            df.A4,
        )
        df["M1"] = outputs_M1["predictions"]
        df["M1_proba"] = outputs_M1["predictions_proba"]

        # Force Blank lines to 0
        df.loc[df.BLANK_LINE, "M1"] = 0

        # Train and predict M2
        df_endlines = df.loc[df.END_LINE]
        self._fit_M2(B1=df_endlines.B1, B2=df_endlines.B2, label=df_endlines.M1)
        outputs_M2 = self._predict_M2(B1=df_endlines.B1, B2=df_endlines.B2)

        df.loc[df.END_LINE, "M2"] = outputs_M2["predictions"]
        df.loc[df.END_LINE, "M2_proba"] = outputs_M2["predictions_proba"]

        df["M2"] = df["M2"].astype(
            pd.Int64Dtype()
        )  # cast to pd.Int64Dtype cause there are None values

        # M1M2
        df = df.loc[df.END_LINE]
        df["M1M2_lr"] = (df["M2_proba"] / (1 - df["M2_proba"])) * (
            df["M1_proba"] / (1 - df["M1_proba"])
        )
        df["M1M2"] = (df["M1M2_lr"] > 1).astype("int")

        # Force Blank lines to 0
        df.loc[df.BLANK_LINE, ["M2", "M1M2"]] = 0

        # Make binary col
        df["PREDICTED_END_LINE"] = np.logical_not(df["M1M2"].astype(bool))

        return df

    def predict(self, df: pd.DataFrame) -> pd.DataFrame:
        """Use the model for inference

        The df should have the following columns:
        `["A1","A2","A3","A4","B1","B2","BLANK_LINE"]`

        Parameters
        ----------
        df : pd.DataFrame
            The df should have the following columns:
            `["A1","A2","A3","A4","B1","B2","BLANK_LINE"]`

        Returns
        -------
        pd.DataFrame
            The result is added to the column `PREDICTED_END_LINE`
        """

        df = self._convert_raw_data_to_codes(df)

        outputs_M1 = self._predict_M1(df.A1, df.A2, df._A3, df._A4)
        df["M1"] = outputs_M1["predictions"]
        df["M1_proba"] = outputs_M1["predictions_proba"]

        outputs_M2 = self._predict_M2(B1=df._B1, B2=df._B2)
        df["M2"] = outputs_M2["predictions"]
        df["M2_proba"] = outputs_M2["predictions_proba"]
        df["M2"] = df["M2"].astype(
            pd.Int64Dtype()
        )  # cast to pd.Int64Dtype cause there are None values

        # M1M2
        df["M1M2_lr"] = (df["M2_proba"] / (1 - df["M2_proba"])) * (
            df["M1_proba"] / (1 - df["M1_proba"])
        )
        df["M1M2"] = (df["M1M2_lr"] > 1).astype("int")

        # Force Blank lines to 0
        df.loc[
            df.BLANK_LINE,
            [
                "M1M2",
            ],
        ] = 0

        # Make binary col
        df["PREDICTED_END_LINE"] = np.logical_not(df["M1M2"].astype(bool))

        return df

    def save(self, path="base_model.pkl"):
        """Save a pickle of the model. It could be read by the pipeline later.

        Parameters
        ----------
        path : str, optional
            path to file .pkl, by default `base_model.pkl`
        """
        with open(path, "wb") as outp:
            del self.nlp
            pickle.dump(self, outp, pickle.HIGHEST_PROTOCOL)

    def _convert_A(self, df: pd.DataFrame, col: str) -> pd.DataFrame:
        """
        Parameters
        ----------
        df : pd.DataFrame
        col : str
            column to translate

        Returns
        -------
        pd.DataFrame
        """
        cat_type_A = CategoricalDtype(
            categories=self.vocabulary["A3A4"].keys(), ordered=True
        )
        new_col = "_" + col
        df[new_col] = df[col].astype(cat_type_A)
        df[new_col] = df[new_col].cat.codes
        # Ensure that not known values are coded as OTHER
        df.loc[
            ~df[col].isin(self.vocabulary["A3A4"].keys()), new_col
        ] = self.vocabulary["A3A4"]["OTHER"]
        return df

    def _convert_B(self, df: pd.DataFrame, col: str) -> pd.DataFrame:
        """
        Parameters
        ----------
        df : pd.DataFrame
            [description]
        col : str
            column to translate

        Returns
        -------
        pd.DataFrame
            [description]
        """
        # Translate B1
        index_B = pd.IntervalIndex(list(self.vocabulary[col].keys()))
        new_col = "_" + col
        df[new_col] = pd.cut(df[col], index_B)
        df[new_col] = df[new_col].cat.codes
        df.loc[df[col] >= index_B.right.max(), new_col] = max(
            self.vocabulary[col].values()
        )
        df.loc[df[col] <= index_B.left.min(), new_col] = min(
            self.vocabulary[col].values()
        )

        return df

    def _convert_raw_data_to_codes(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Function to translate data as extracted from spacy to the model codes.
        `A1` and `A2` are not translated cause are supposed to be already
        in good encoding.

        Parameters
        ----------
        df : pd.DataFrame
            It should have columns `['A3','A4','B1','B2']`

        Returns
        -------
        pd.DataFrame
        """
        df = self._convert_A(df, "A3")
        df = self._convert_A(df, "A4")
        df = self._convert_B(df, "B1")
        df = self._convert_B(df, "B2")
        return df

    def _convert_line_to_attribute(
        self, df: pd.DataFrame, expr: str, col: str
    ) -> pd.DataFrame:
        """
        Function to convert a line into an attribute (column) of the
        previous row. Particularly we use it to identify "\\n" and "\\n\\n"
        that are considered tokens, express this information as an attribute
        of the previous token.

        Parameters
        ----------
        df : pd.DataFrame
        expr : str
            pattern to search in the text. Ex.: "\\n"
        col : str
            name of the new column

        Returns
        -------
        pd.DataFrame
        """
        idx = df.TEXT.str.contains(expr)
        df.loc[idx, col] = True
        df[col] = df[col].fillna(False)
        df = self._shift_col(df, col, "TEMP_" + col, direction="backward")

        return df

    def _compute_a3(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        A3 (A4 respectively): typographic form  of left word (or right) :

        - All in capital letter
        - It starts with a capital letter
        - Starts by lowercase
        - It's a number
        - Strong punctuation
        - Soft punctuation
        - A number followed or preced by a punctuation (it's the case of enumerations)

        Parameters
        ----------
        df: pd.DataFrame

        Returns
        -------
        df: pd.DataFrame with the columns `A3` and `A3_`

        """
        df = self._shift_col(
            df, "IS_PUNCT", "IS_PUNCT_+1", direction="backward", fill=False
        )
        df = self._shift_col(
            df, "IS_PUNCT", "IS_PUNCT_-1", direction="forward", fill=False
        )

        CONDITION1 = df.IS_UPPER
        CONDITION2 = df.SHAPE_.str.startswith("Xx", na=False)
        CONDITION3 = df.SHAPE_.str.startswith("x", na=False)
        CONDITION4 = df.IS_DIGIT
        STRONG_PUNCT = [".", ";", "..", "..."]
        CONDITION5 = (df.IS_PUNCT) & (df.TEXT.isin(STRONG_PUNCT))
        CONDITION6 = (df.IS_PUNCT) & (~df.TEXT.isin(STRONG_PUNCT))
        CONDITION7 = (df.IS_DIGIT) & (df["IS_PUNCT_+1"] | df["IS_PUNCT_-1"])  # discuss

        df["A3_"] = None
        df.loc[CONDITION1, "A3_"] = "UPPER"
        df.loc[CONDITION2, "A3_"] = "S_UPPER"
        df.loc[CONDITION3, "A3_"] = "LOWER"
        df.loc[CONDITION4, "A3_"] = "DIGIT"
        df.loc[CONDITION5, "A3_"] = "STRONG_PUNCT"
        df.loc[CONDITION6, "A3_"] = "SOFT_PUNCT"
        df.loc[CONDITION7, "A3_"] = "ENUMERATION"

        df = df.drop(columns=["IS_PUNCT_+1", "IS_PUNCT_-1"])
        df["A3_"] = df["A3_"].astype("category")

        df["A3_"] = df["A3_"].cat.add_categories("OTHER")
        df["A3_"].fillna("OTHER", inplace=True)

        df["A3"] = df["A3_"].cat.codes

        return df

    def _fit_M1(
        self,
        A1: pd.Series,
        A2: pd.Series,
        A3: pd.Series,
        A4: pd.Series,
        label: pd.Series,
    ):
        """Function to train M1 classifier (Naive Bayes)

        Parameters
        ----------
        A1 : pd.Series
            [description]
        A2 : pd.Series
            [description]
        A3 : pd.Series
            [description]
        A4 : pd.Series
            [description]
        label : pd.Series
            [description]

        """
        # Encode classes to OneHotEncoder representation
        encoder_A1_A2 = self._fit_encoder_2S(A1, A2)
        self.encoder_A1_A2 = encoder_A1_A2

        encoder_A3_A4 = self._fit_encoder_2S(A3, A4)
        self.encoder_A3_A4 = encoder_A3_A4

        # M1
        m1 = MultinomialNB(alpha=1)

        X = self._get_X_for_M1(A1, A2, A3, A4)
        m1.fit(X, label)
        self.m1 = m1

    def _fit_M2(self, B1: pd.Series, B2: pd.Series, label: pd.Series):
        """Function to train M2 classifier (Naive Bayes)

        Parameters
        ----------
        B1 : pd.Series
        B2 : pd.Series
        label : pd.Series
        """

        # Encode classes to OneHotEncoder representation
        encoder_B1 = self._fit_encoder_1S(B1)
        self.encoder_B1 = encoder_B1
        encoder_B2 = self._fit_encoder_1S(B2)
        self.encoder_B2 = encoder_B2

        # Multinomial Naive Bayes
        m2 = MultinomialNB(alpha=1)
        X = self._get_X_for_M2(B1, B2)
        m2.fit(X, label)
        self.m2 = m2

    def _get_X_for_M1(
        self, A1: pd.Series, A2: pd.Series, A3: pd.Series, A4: pd.Series
    ) -> np.ndarray:
        """Get X matrix for classifier

        Parameters
        ----------
        A1 : pd.Series
        A2 : pd.Series
        A3 : pd.Series
        A4 : pd.Series

        Returns
        -------
        np.ndarray
        """
        A1_enc = self._encode_series(self.encoder_A1_A2, A1)
        A2_enc = self._encode_series(self.encoder_A1_A2, A2)
        A3_enc = self._encode_series(self.encoder_A3_A4, A3)
        A4_enc = self._encode_series(self.encoder_A3_A4, A4)
        X = hstack([A1_enc, A2_enc, A3_enc, A4_enc])
        return X

    def _get_X_for_M2(self, B1: pd.Series, B2: pd.Series) -> np.ndarray:
        """Get X matrix for classifier

        Parameters
        ----------
        B1 : pd.Series
        B2 : pd.Series

        Returns
        -------
        np.ndarray
        """
        B1_enc = self._encode_series(self.encoder_B1, B1)
        B2_enc = self._encode_series(self.encoder_B2, B2)
        X = hstack([B1_enc, B2_enc])
        return X

    def _predict_M1(
        self, A1: pd.Series, A2: pd.Series, A3: pd.Series, A4: pd.Series
    ) -> Dict[str, Any]:
        """Use M1 for prediction

        Parameters
        ----------
        A1 : pd.Series
        A2 : pd.Series
        A3 : pd.Series
        A4 : pd.Series

        Returns
        -------
        Dict[str, Any]
        """
        X = self._get_X_for_M1(A1, A2, A3, A4)
        predictions = self.m1.predict(X)
        predictions_proba = self.m1.predict_proba(X)[:, 1]
        outputs = {"predictions": predictions, "predictions_proba": predictions_proba}
        return outputs

    def _predict_M2(self, B1: pd.Series, B2: pd.Series) -> Dict[str, Any]:
        """Use M2 for prediction

        Parameters
        ----------
        B1 : pd.Series
        B2 : pd.Series

        Returns
        -------
        Dict[str, Any]
        """
        X = self._get_X_for_M2(B1, B2)
        predictions = self.m2.predict(X)
        predictions_proba = self.m2.predict_proba(X)[:, 1]
        outputs = {"predictions": predictions, "predictions_proba": predictions_proba}
        return outputs

    def _fit_encoder_2S(self, S1: pd.Series, S2: pd.Series) -> OneHotEncoder:
        """Fit a one hot encoder with 2 Series. It concatenates the series and after it fits.

        Parameters
        ----------
        S1 : pd.Series
        S2 : pd.Series

        Returns
        -------
        OneHotEncoder
        """
        _S1 = _convert_series_to_array(S1)
        _S2 = _convert_series_to_array(S2)
        S = np.concatenate([_S1, _S2])
        encoder = self._fit_one_hot_encoder(S)
        return encoder

    def _fit_encoder_1S(self, S1: pd.Series) -> OneHotEncoder:
        """Fit a one hot encoder with 1 Series.

        Parameters
        ----------
        S1 : pd.Series

        Returns
        -------
        OneHotEncoder
        """
        _S1 = _convert_series_to_array(S1)
        encoder = self._fit_one_hot_encoder(_S1)
        return encoder

    def _encode_series(self, encoder: OneHotEncoder, S: pd.Series) -> np.ndarray:
        """Use the one hot encoder to transform a series.

        Parameters
        ----------
        encoder : OneHotEncoder
        S : pd.Series
            a series to encode (transform)

        Returns
        -------
        np.ndarray
        """
        _S = _convert_series_to_array(S)
        S_enc = encoder.transform(_S)
        return S_enc

    def set_spans(self, corpus: Iterable[Doc], df: pd.DataFrame):
        """
        Function to set the results of the algorithm (pd.DataFrame)
        as spans of the spaCy document.

        Parameters
        ----------
        corpus : Iterable[Doc]
            Iterable of spaCy Documents
        df : pd.DataFrame
            It should have the columns:
            ["DOC_ID","original_token_index","PREDICTED_END_LINE"]
        """

        for doc_id, doc in enumerate(corpus):
            spans = []
            for token_i, pred in df.loc[
                df.DOC_ID == doc_id, ["original_token_index", "PREDICTED_END_LINE"]
            ].values:
                s = Span(doc, start=token_i, end=token_i + 1, label=_get_label(pred))

                spans.append(s)

            doc.spans["new_lines"] = spans

    @staticmethod
    def _retrieve_lines(dfg: DataFrameGroupBy) -> DataFrameGroupBy:
        """Function to give a sentence_id to each token.

        Parameters
        ----------
        dfg : DataFrameGroupBy

        Returns
        -------
        DataFrameGroupBy
            Same DataFrameGroupBy with the column `SENTENCE_ID`
        """
        sentences_ids = np.arange(dfg.END_LINE.sum())
        dfg.loc[dfg.END_LINE, "SENTENCE_ID"] = sentences_ids
        dfg["SENTENCE_ID"] = dfg["SENTENCE_ID"].fillna(method="bfill")
        return dfg

    @staticmethod
    def _create_vocabulary(x: iterable) -> dict:
        """Function to create a vocabulary for attributes in the training set.

        Parameters
        ----------
        x : iterable

        Returns
        -------
        dict
        """
        v = {}

        for i, key in enumerate(x):
            v[key] = i

        return v

    @staticmethod
    def _compute_B(df: pd.DataFrame) -> pd.DataFrame:
        """Function to compute B1 and B2

        Parameters
        ----------
        df : pd.DataFrame

        Returns
        -------
        pd.DataFrame
        """

        data = df.groupby(["DOC_ID", "SENTENCE_ID"]).agg(l=("LENGTH", "sum"))
        df_t = df.loc[df.END_LINE, ["DOC_ID", "SENTENCE_ID"]].merge(
            data, left_on=["DOC_ID", "SENTENCE_ID"], right_index=True, how="left"
        )

        stats_doc = df_t.groupby("DOC_ID").agg(mu=("l", "mean"), sigma=("l", "std"))
        stats_doc["sigma"].replace(
            0.0, 1.0, inplace=True
        )  # Replace the 0 std by unit std, otherwise it breaks the code.
        stats_doc["cv"] = stats_doc["sigma"] / stats_doc["mu"]

        df_t = df_t.drop(columns=["DOC_ID", "SENTENCE_ID"])
        df2 = df.merge(df_t, left_index=True, right_index=True, how="left")

        df2 = df2.merge(stats_doc, on=["DOC_ID"], how="left")
        df2["l_norm"] = (df2["l"] - df2["mu"]) / df2["sigma"]

        df2["cv_bin"] = pd.cut(df2["cv"], bins=10)
        df2["B2"] = df2["cv_bin"].cat.codes

        df2["l_norm_bin"] = pd.cut(df2["l_norm"], bins=10)
        df2["B1"] = df2["l_norm_bin"].cat.codes

        return df2

    @staticmethod
    def _shift_col(
        df: pd.DataFrame, col: str, new_col: str, direction="backward", fill=None
    ) -> pd.DataFrame:
        """Shifts a column one position into backward / forward direction.

        Parameters
        ----------
        df : pd.DataFrame
        col : str
            column to shift
        new_col : str
            column name to save the results
        direction : str, optional
            one of {"backward", "forward"}, by default "backward"
        fill : [type], optional
            , by default None

        Returns
        -------
        pd.DataFrame
            same df with `new_col` added.
        """
        df[new_col] = fill

        if direction == "backward":
            df.loc[df.index[:-1], new_col] = df[col].values[1:]

            different_doc_id = df["DOC_ID"].values[:-1] != df["DOC_ID"].values[1:]
            different_doc_id = np.append(different_doc_id, True)

        if direction == "forward":
            df.loc[df.index[1:], new_col] = df[col].values[:-1]
            different_doc_id = df["DOC_ID"].values[1:] != df["DOC_ID"].values[:-1]
            different_doc_id = np.append(True, different_doc_id)

        df.loc[different_doc_id, new_col] = fill
        return df

    @staticmethod
    def _get_attributes(doc: Doc, i=0):
        """Function to get the attributes of tokens of a spacy doc in a pd.DataFrame format.

        Parameters
        ----------
        doc : Doc
            spacy Doc
        i : int, optional
            document id, by default 0

        Returns
        -------
        pd.DataFrame
            Returns a dataframe with one line per token. It has the following columns :
            `[
            "ORTH",
            "LOWER",
            "SHAPE",
            "IS_DIGIT",
            "IS_SPACE",
            "IS_UPPER",
            "IS_PUNCT",
            "LENGTH",
            ]`
        """
        attributes = [
            "ORTH",
            "LOWER",
            "SHAPE",
            "IS_DIGIT",
            "IS_SPACE",
            "IS_UPPER",
            "IS_PUNCT",
            "LENGTH",
        ]
        attributes_array = doc.to_array(attributes)
        attributes_df = pd.DataFrame(attributes_array, columns=attributes)
        attributes_df["DOC_ID"] = i
        boolean_attr = []
        for a in attributes:
            if a[:3] == "IS_":
                boolean_attr.append(a)
        attributes_df[boolean_attr] = attributes_df[boolean_attr].astype("boolean")
        return attributes_df

    @staticmethod
    def _get_string(_id: int, string_store: StringStore) -> str:
        """Returns the string corresponding to the token_id

        Parameters
        ----------
        _id : int
            token id
        string_store : StringStore
            spaCy Language String Store

        Returns
        -------
        str
            string representation of the token.
        """
        return string_store[_id]

    @staticmethod
    def _fit_one_hot_encoder(X: np.ndarray) -> OneHotEncoder:
        """Fit a one hot encoder.

        Parameters
        ----------
        X : np.ndarray
            of shape (n,1)

        Returns
        -------
        OneHotEncoder
        """
        encoder = OneHotEncoder(handle_unknown="ignore")
        encoder.fit(X)
        return encoder
nlp = nlp instance-attribute
__init__(nlp)
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
28
29
def __init__(self, nlp: Language):
    self.nlp = nlp
_preprocess_data(corpus)
PARAMETER DESCRIPTION
corpus

Corpus of documents

TYPE: Iterable[Doc]

RETURNS DESCRIPTION
pd.DataFrame

Preprocessed data
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def _preprocess_data(self, corpus: Iterable[Doc]) -> pd.DataFrame:
    """
    Parameters
    ----------
    corpus : Iterable[Doc]
        Corpus of documents

    Returns
    -------
    pd.DataFrame
        Preprocessed data
    """
    # Extract the vocabulary
    string_store = self.nlp.vocab.strings

    # Iterate in the corpus and construct a dataframe
    train_data_list = []
    for i, doc in enumerate(corpus):
        train_data_list.append(self._get_attributes(doc, i))

    df = pd.concat(train_data_list)
    df.reset_index(inplace=True, drop=False)
    df.rename(columns={"ORTH": "A1", "index": "original_token_index"}, inplace=True)

    # Retrieve string representation of token_id and shape
    df["TEXT"] = df.A1.apply(self._get_string, string_store=string_store)
    df["SHAPE_"] = df.SHAPE.apply(self._get_string, string_store=string_store)

    # Convert new lines as an attribute instead of a row
    df = self._convert_line_to_attribute(df, expr="\n", col="END_LINE")
    df = self._convert_line_to_attribute(df, expr="\n\n", col="BLANK_LINE")
    df = df.loc[~(df.END_LINE | df.BLANK_LINE)]
    df = df.drop(columns="END_LINE")
    df = df.drop(columns="BLANK_LINE")
    df.rename(
        columns={"TEMP_END_LINE": "END_LINE", "TEMP_BLANK_LINE": "BLANK_LINE"},
        inplace=True,
    )

    # Construct A2 by shifting
    df = self._shift_col(df, "A1", "A2", direction="backward")

    # Compute A3 and A4
    df = self._compute_a3(df)
    df = self._shift_col(df, "A3", "A4", direction="backward")

    # SPACE is the class to predict. Set 1 if not an END_LINE
    df["SPACE"] = np.logical_not(df["END_LINE"]).astype("int")

    df[["END_LINE", "BLANK_LINE"]] = df[["END_LINE", "BLANK_LINE"]].fillna(
        True, inplace=False
    )

    # Assign a sentence id to each token
    df = df.groupby("DOC_ID").apply(self._retrieve_lines)
    df["SENTENCE_ID"] = df["SENTENCE_ID"].astype("int")

    # Compute B1 and B2
    df = self._compute_B(df)

    # Drop Tokens without info (last token of doc)
    df.dropna(subset=["A1", "A2", "A3", "A4"], inplace=True)

    # Export the vocabularies to be able to use the model with another corpus
    voc_a3a4 = self._create_vocabulary(df.A3_.cat.categories)
    voc_B2 = self._create_vocabulary(df.cv_bin.cat.categories)
    voc_B1 = self._create_vocabulary(df.l_norm_bin.cat.categories)

    vocabulary = {"A3A4": voc_a3a4, "B1": voc_B1, "B2": voc_B2}

    self.vocabulary = vocabulary

    return df
fit_and_predict(corpus)

Fit the model and predict for the training data

PARAMETER DESCRIPTION
corpus

An iterable of Documents

TYPE: Iterable[Doc]

RETURNS DESCRIPTION
pd.DataFrame

one line by end_line prediction
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def fit_and_predict(self, corpus: Iterable[Doc]) -> pd.DataFrame:
    """Fit the model and predict for the training data

    Parameters
    ----------
    corpus : Iterable[Doc]
        An iterable of Documents

    Returns
    -------
    pd.DataFrame
        one line by end_line prediction
    """

    # Preprocess data to have a pd DF
    df = self._preprocess_data(corpus)

    # Train and predict M1
    self._fit_M1(df.A1, df.A2, df.A3, df.A4, df.SPACE)
    outputs_M1 = self._predict_M1(
        df.A1,
        df.A2,
        df.A3,
        df.A4,
    )
    df["M1"] = outputs_M1["predictions"]
    df["M1_proba"] = outputs_M1["predictions_proba"]

    # Force Blank lines to 0
    df.loc[df.BLANK_LINE, "M1"] = 0

    # Train and predict M2
    df_endlines = df.loc[df.END_LINE]
    self._fit_M2(B1=df_endlines.B1, B2=df_endlines.B2, label=df_endlines.M1)
    outputs_M2 = self._predict_M2(B1=df_endlines.B1, B2=df_endlines.B2)

    df.loc[df.END_LINE, "M2"] = outputs_M2["predictions"]
    df.loc[df.END_LINE, "M2_proba"] = outputs_M2["predictions_proba"]

    df["M2"] = df["M2"].astype(
        pd.Int64Dtype()
    )  # cast to pd.Int64Dtype cause there are None values

    # M1M2
    df = df.loc[df.END_LINE]
    df["M1M2_lr"] = (df["M2_proba"] / (1 - df["M2_proba"])) * (
        df["M1_proba"] / (1 - df["M1_proba"])
    )
    df["M1M2"] = (df["M1M2_lr"] > 1).astype("int")

    # Force Blank lines to 0
    df.loc[df.BLANK_LINE, ["M2", "M1M2"]] = 0

    # Make binary col
    df["PREDICTED_END_LINE"] = np.logical_not(df["M1M2"].astype(bool))

    return df
predict(df)

Use the model for inference

The df should have the following columns: ["A1","A2","A3","A4","B1","B2","BLANK_LINE"]

PARAMETER DESCRIPTION
df

The df should have the following columns: ["A1","A2","A3","A4","B1","B2","BLANK_LINE"]

TYPE: pd.DataFrame

RETURNS DESCRIPTION
pd.DataFrame

The result is added to the column PREDICTED_END_LINE
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def predict(self, df: pd.DataFrame) -> pd.DataFrame:
    """Use the model for inference

    The df should have the following columns:
    `["A1","A2","A3","A4","B1","B2","BLANK_LINE"]`

    Parameters
    ----------
    df : pd.DataFrame
        The df should have the following columns:
        `["A1","A2","A3","A4","B1","B2","BLANK_LINE"]`

    Returns
    -------
    pd.DataFrame
        The result is added to the column `PREDICTED_END_LINE`
    """

    df = self._convert_raw_data_to_codes(df)

    outputs_M1 = self._predict_M1(df.A1, df.A2, df._A3, df._A4)
    df["M1"] = outputs_M1["predictions"]
    df["M1_proba"] = outputs_M1["predictions_proba"]

    outputs_M2 = self._predict_M2(B1=df._B1, B2=df._B2)
    df["M2"] = outputs_M2["predictions"]
    df["M2_proba"] = outputs_M2["predictions_proba"]
    df["M2"] = df["M2"].astype(
        pd.Int64Dtype()
    )  # cast to pd.Int64Dtype cause there are None values

    # M1M2
    df["M1M2_lr"] = (df["M2_proba"] / (1 - df["M2_proba"])) * (
        df["M1_proba"] / (1 - df["M1_proba"])
    )
    df["M1M2"] = (df["M1M2_lr"] > 1).astype("int")

    # Force Blank lines to 0
    df.loc[
        df.BLANK_LINE,
        [
            "M1M2",
        ],
    ] = 0

    # Make binary col
    df["PREDICTED_END_LINE"] = np.logical_not(df["M1M2"].astype(bool))

    return df
save(path='base_model.pkl')

Save a pickle of the model. It could be read by the pipeline later.

PARAMETER DESCRIPTION
path

path to file .pkl, by default base_model.pkl

TYPE: str, optional DEFAULT: 'base_model.pkl'

Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
213
214
215
216
217
218
219
220
221
222
223
def save(self, path="base_model.pkl"):
    """Save a pickle of the model. It could be read by the pipeline later.

    Parameters
    ----------
    path : str, optional
        path to file .pkl, by default `base_model.pkl`
    """
    with open(path, "wb") as outp:
        del self.nlp
        pickle.dump(self, outp, pickle.HIGHEST_PROTOCOL)
_convert_A(df, col)
PARAMETER DESCRIPTION
df

TYPE: pd.DataFrame

col

column to translate

TYPE: str

RETURNS DESCRIPTION
pd.DataFrame
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def _convert_A(self, df: pd.DataFrame, col: str) -> pd.DataFrame:
    """
    Parameters
    ----------
    df : pd.DataFrame
    col : str
        column to translate

    Returns
    -------
    pd.DataFrame
    """
    cat_type_A = CategoricalDtype(
        categories=self.vocabulary["A3A4"].keys(), ordered=True
    )
    new_col = "_" + col
    df[new_col] = df[col].astype(cat_type_A)
    df[new_col] = df[new_col].cat.codes
    # Ensure that not known values are coded as OTHER
    df.loc[
        ~df[col].isin(self.vocabulary["A3A4"].keys()), new_col
    ] = self.vocabulary["A3A4"]["OTHER"]
    return df
_convert_B(df, col)
PARAMETER DESCRIPTION
df

[description]

TYPE: pd.DataFrame

col

column to translate

TYPE: str

RETURNS DESCRIPTION
pd.DataFrame

[description]
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def _convert_B(self, df: pd.DataFrame, col: str) -> pd.DataFrame:
    """
    Parameters
    ----------
    df : pd.DataFrame
        [description]
    col : str
        column to translate

    Returns
    -------
    pd.DataFrame
        [description]
    """
    # Translate B1
    index_B = pd.IntervalIndex(list(self.vocabulary[col].keys()))
    new_col = "_" + col
    df[new_col] = pd.cut(df[col], index_B)
    df[new_col] = df[new_col].cat.codes
    df.loc[df[col] >= index_B.right.max(), new_col] = max(
        self.vocabulary[col].values()
    )
    df.loc[df[col] <= index_B.left.min(), new_col] = min(
        self.vocabulary[col].values()
    )

    return df
_convert_raw_data_to_codes(df)

Function to translate data as extracted from spacy to the model codes. A1 and A2 are not translated cause are supposed to be already in good encoding.

PARAMETER DESCRIPTION
df

It should have columns ['A3','A4','B1','B2']

TYPE: pd.DataFrame

RETURNS DESCRIPTION
pd.DataFrame
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def _convert_raw_data_to_codes(self, df: pd.DataFrame) -> pd.DataFrame:
    """
    Function to translate data as extracted from spacy to the model codes.
    `A1` and `A2` are not translated cause are supposed to be already
    in good encoding.

    Parameters
    ----------
    df : pd.DataFrame
        It should have columns `['A3','A4','B1','B2']`

    Returns
    -------
    pd.DataFrame
    """
    df = self._convert_A(df, "A3")
    df = self._convert_A(df, "A4")
    df = self._convert_B(df, "B1")
    df = self._convert_B(df, "B2")
    return df
_convert_line_to_attribute(df, expr, col)

Function to convert a line into an attribute (column) of the previous row. Particularly we use it to identify "\n" and "\n\n" that are considered tokens, express this information as an attribute of the previous token.

PARAMETER DESCRIPTION
df

TYPE: pd.DataFrame

expr

pattern to search in the text. Ex.: "\n"

TYPE: str

col

name of the new column

TYPE: str

RETURNS DESCRIPTION
pd.DataFrame
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def _convert_line_to_attribute(
    self, df: pd.DataFrame, expr: str, col: str
) -> pd.DataFrame:
    """
    Function to convert a line into an attribute (column) of the
    previous row. Particularly we use it to identify "\\n" and "\\n\\n"
    that are considered tokens, express this information as an attribute
    of the previous token.

    Parameters
    ----------
    df : pd.DataFrame
    expr : str
        pattern to search in the text. Ex.: "\\n"
    col : str
        name of the new column

    Returns
    -------
    pd.DataFrame
    """
    idx = df.TEXT.str.contains(expr)
    df.loc[idx, col] = True
    df[col] = df[col].fillna(False)
    df = self._shift_col(df, col, "TEMP_" + col, direction="backward")

    return df
_compute_a3(df)

A3 (A4 respectively): typographic form of left word (or right) :

  • All in capital letter
  • It starts with a capital letter
  • Starts by lowercase
  • It's a number
  • Strong punctuation
  • Soft punctuation
  • A number followed or preced by a punctuation (it's the case of enumerations)
PARAMETER DESCRIPTION
df

TYPE: pd.DataFrame

RETURNS DESCRIPTION
df
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
def _compute_a3(self, df: pd.DataFrame) -> pd.DataFrame:
    """
    A3 (A4 respectively): typographic form  of left word (or right) :

    - All in capital letter
    - It starts with a capital letter
    - Starts by lowercase
    - It's a number
    - Strong punctuation
    - Soft punctuation
    - A number followed or preced by a punctuation (it's the case of enumerations)

    Parameters
    ----------
    df: pd.DataFrame

    Returns
    -------
    df: pd.DataFrame with the columns `A3` and `A3_`

    """
    df = self._shift_col(
        df, "IS_PUNCT", "IS_PUNCT_+1", direction="backward", fill=False
    )
    df = self._shift_col(
        df, "IS_PUNCT", "IS_PUNCT_-1", direction="forward", fill=False
    )

    CONDITION1 = df.IS_UPPER
    CONDITION2 = df.SHAPE_.str.startswith("Xx", na=False)
    CONDITION3 = df.SHAPE_.str.startswith("x", na=False)
    CONDITION4 = df.IS_DIGIT
    STRONG_PUNCT = [".", ";", "..", "..."]
    CONDITION5 = (df.IS_PUNCT) & (df.TEXT.isin(STRONG_PUNCT))
    CONDITION6 = (df.IS_PUNCT) & (~df.TEXT.isin(STRONG_PUNCT))
    CONDITION7 = (df.IS_DIGIT) & (df["IS_PUNCT_+1"] | df["IS_PUNCT_-1"])  # discuss

    df["A3_"] = None
    df.loc[CONDITION1, "A3_"] = "UPPER"
    df.loc[CONDITION2, "A3_"] = "S_UPPER"
    df.loc[CONDITION3, "A3_"] = "LOWER"
    df.loc[CONDITION4, "A3_"] = "DIGIT"
    df.loc[CONDITION5, "A3_"] = "STRONG_PUNCT"
    df.loc[CONDITION6, "A3_"] = "SOFT_PUNCT"
    df.loc[CONDITION7, "A3_"] = "ENUMERATION"

    df = df.drop(columns=["IS_PUNCT_+1", "IS_PUNCT_-1"])
    df["A3_"] = df["A3_"].astype("category")

    df["A3_"] = df["A3_"].cat.add_categories("OTHER")
    df["A3_"].fillna("OTHER", inplace=True)

    df["A3"] = df["A3_"].cat.codes

    return df
_fit_M1(A1, A2, A3, A4, label)

Function to train M1 classifier (Naive Bayes)

PARAMETER DESCRIPTION
A1

[description]

TYPE: pd.Series

A2

[description]

TYPE: pd.Series

A3

[description]

TYPE: pd.Series

A4

[description]

TYPE: pd.Series

label

[description]

TYPE: pd.Series

Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
def _fit_M1(
    self,
    A1: pd.Series,
    A2: pd.Series,
    A3: pd.Series,
    A4: pd.Series,
    label: pd.Series,
):
    """Function to train M1 classifier (Naive Bayes)

    Parameters
    ----------
    A1 : pd.Series
        [description]
    A2 : pd.Series
        [description]
    A3 : pd.Series
        [description]
    A4 : pd.Series
        [description]
    label : pd.Series
        [description]

    """
    # Encode classes to OneHotEncoder representation
    encoder_A1_A2 = self._fit_encoder_2S(A1, A2)
    self.encoder_A1_A2 = encoder_A1_A2

    encoder_A3_A4 = self._fit_encoder_2S(A3, A4)
    self.encoder_A3_A4 = encoder_A3_A4

    # M1
    m1 = MultinomialNB(alpha=1)

    X = self._get_X_for_M1(A1, A2, A3, A4)
    m1.fit(X, label)
    self.m1 = m1
_fit_M2(B1, B2, label)

Function to train M2 classifier (Naive Bayes)

PARAMETER DESCRIPTION
B1

TYPE: pd.Series

B2

TYPE: pd.Series

label

TYPE: pd.Series

Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
def _fit_M2(self, B1: pd.Series, B2: pd.Series, label: pd.Series):
    """Function to train M2 classifier (Naive Bayes)

    Parameters
    ----------
    B1 : pd.Series
    B2 : pd.Series
    label : pd.Series
    """

    # Encode classes to OneHotEncoder representation
    encoder_B1 = self._fit_encoder_1S(B1)
    self.encoder_B1 = encoder_B1
    encoder_B2 = self._fit_encoder_1S(B2)
    self.encoder_B2 = encoder_B2

    # Multinomial Naive Bayes
    m2 = MultinomialNB(alpha=1)
    X = self._get_X_for_M2(B1, B2)
    m2.fit(X, label)
    self.m2 = m2
_get_X_for_M1(A1, A2, A3, A4)

Get X matrix for classifier

PARAMETER DESCRIPTION
A1

TYPE: pd.Series

A2

TYPE: pd.Series

A3

TYPE: pd.Series

A4

TYPE: pd.Series

RETURNS DESCRIPTION
np.ndarray
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
def _get_X_for_M1(
    self, A1: pd.Series, A2: pd.Series, A3: pd.Series, A4: pd.Series
) -> np.ndarray:
    """Get X matrix for classifier

    Parameters
    ----------
    A1 : pd.Series
    A2 : pd.Series
    A3 : pd.Series
    A4 : pd.Series

    Returns
    -------
    np.ndarray
    """
    A1_enc = self._encode_series(self.encoder_A1_A2, A1)
    A2_enc = self._encode_series(self.encoder_A1_A2, A2)
    A3_enc = self._encode_series(self.encoder_A3_A4, A3)
    A4_enc = self._encode_series(self.encoder_A3_A4, A4)
    X = hstack([A1_enc, A2_enc, A3_enc, A4_enc])
    return X
_get_X_for_M2(B1, B2)

Get X matrix for classifier

PARAMETER DESCRIPTION
B1

TYPE: pd.Series

B2

TYPE: pd.Series

RETURNS DESCRIPTION
np.ndarray
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
def _get_X_for_M2(self, B1: pd.Series, B2: pd.Series) -> np.ndarray:
    """Get X matrix for classifier

    Parameters
    ----------
    B1 : pd.Series
    B2 : pd.Series

    Returns
    -------
    np.ndarray
    """
    B1_enc = self._encode_series(self.encoder_B1, B1)
    B2_enc = self._encode_series(self.encoder_B2, B2)
    X = hstack([B1_enc, B2_enc])
    return X
_predict_M1(A1, A2, A3, A4)

Use M1 for prediction

PARAMETER DESCRIPTION
A1

TYPE: pd.Series

A2

TYPE: pd.Series

A3

TYPE: pd.Series

A4

TYPE: pd.Series

RETURNS DESCRIPTION
Dict[str, Any]
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
def _predict_M1(
    self, A1: pd.Series, A2: pd.Series, A3: pd.Series, A4: pd.Series
) -> Dict[str, Any]:
    """Use M1 for prediction

    Parameters
    ----------
    A1 : pd.Series
    A2 : pd.Series
    A3 : pd.Series
    A4 : pd.Series

    Returns
    -------
    Dict[str, Any]
    """
    X = self._get_X_for_M1(A1, A2, A3, A4)
    predictions = self.m1.predict(X)
    predictions_proba = self.m1.predict_proba(X)[:, 1]
    outputs = {"predictions": predictions, "predictions_proba": predictions_proba}
    return outputs
_predict_M2(B1, B2)

Use M2 for prediction

PARAMETER DESCRIPTION
B1

TYPE: pd.Series

B2

TYPE: pd.Series

RETURNS DESCRIPTION
Dict[str, Any]
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
def _predict_M2(self, B1: pd.Series, B2: pd.Series) -> Dict[str, Any]:
    """Use M2 for prediction

    Parameters
    ----------
    B1 : pd.Series
    B2 : pd.Series

    Returns
    -------
    Dict[str, Any]
    """
    X = self._get_X_for_M2(B1, B2)
    predictions = self.m2.predict(X)
    predictions_proba = self.m2.predict_proba(X)[:, 1]
    outputs = {"predictions": predictions, "predictions_proba": predictions_proba}
    return outputs
_fit_encoder_2S(S1, S2)

Fit a one hot encoder with 2 Series. It concatenates the series and after it fits.

PARAMETER DESCRIPTION
S1

TYPE: pd.Series

S2

TYPE: pd.Series

RETURNS DESCRIPTION
OneHotEncoder
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
def _fit_encoder_2S(self, S1: pd.Series, S2: pd.Series) -> OneHotEncoder:
    """Fit a one hot encoder with 2 Series. It concatenates the series and after it fits.

    Parameters
    ----------
    S1 : pd.Series
    S2 : pd.Series

    Returns
    -------
    OneHotEncoder
    """
    _S1 = _convert_series_to_array(S1)
    _S2 = _convert_series_to_array(S2)
    S = np.concatenate([_S1, _S2])
    encoder = self._fit_one_hot_encoder(S)
    return encoder
_fit_encoder_1S(S1)

Fit a one hot encoder with 1 Series.

PARAMETER DESCRIPTION
S1

TYPE: pd.Series

RETURNS DESCRIPTION
OneHotEncoder
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
540
541
542
543
544
545
546
547
548
549
550
551
552
553
def _fit_encoder_1S(self, S1: pd.Series) -> OneHotEncoder:
    """Fit a one hot encoder with 1 Series.

    Parameters
    ----------
    S1 : pd.Series

    Returns
    -------
    OneHotEncoder
    """
    _S1 = _convert_series_to_array(S1)
    encoder = self._fit_one_hot_encoder(_S1)
    return encoder
_encode_series(encoder, S)

Use the one hot encoder to transform a series.

PARAMETER DESCRIPTION
encoder

TYPE: OneHotEncoder

S

a series to encode (transform)

TYPE: pd.Series

RETURNS DESCRIPTION
np.ndarray
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
def _encode_series(self, encoder: OneHotEncoder, S: pd.Series) -> np.ndarray:
    """Use the one hot encoder to transform a series.

    Parameters
    ----------
    encoder : OneHotEncoder
    S : pd.Series
        a series to encode (transform)

    Returns
    -------
    np.ndarray
    """
    _S = _convert_series_to_array(S)
    S_enc = encoder.transform(_S)
    return S_enc
set_spans(corpus, df)

Function to set the results of the algorithm (pd.DataFrame) as spans of the spaCy document.

PARAMETER DESCRIPTION
corpus

Iterable of spaCy Documents

TYPE: Iterable[Doc]

df

It should have the columns: ["DOC_ID","original_token_index","PREDICTED_END_LINE"]

TYPE: pd.DataFrame

Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
def set_spans(self, corpus: Iterable[Doc], df: pd.DataFrame):
    """
    Function to set the results of the algorithm (pd.DataFrame)
    as spans of the spaCy document.

    Parameters
    ----------
    corpus : Iterable[Doc]
        Iterable of spaCy Documents
    df : pd.DataFrame
        It should have the columns:
        ["DOC_ID","original_token_index","PREDICTED_END_LINE"]
    """

    for doc_id, doc in enumerate(corpus):
        spans = []
        for token_i, pred in df.loc[
            df.DOC_ID == doc_id, ["original_token_index", "PREDICTED_END_LINE"]
        ].values:
            s = Span(doc, start=token_i, end=token_i + 1, label=_get_label(pred))

            spans.append(s)

        doc.spans["new_lines"] = spans
_retrieve_lines(dfg)

Function to give a sentence_id to each token.

PARAMETER DESCRIPTION
dfg

TYPE: DataFrameGroupBy

RETURNS DESCRIPTION
DataFrameGroupBy

Same DataFrameGroupBy with the column SENTENCE_ID
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
@staticmethod
def _retrieve_lines(dfg: DataFrameGroupBy) -> DataFrameGroupBy:
    """Function to give a sentence_id to each token.

    Parameters
    ----------
    dfg : DataFrameGroupBy

    Returns
    -------
    DataFrameGroupBy
        Same DataFrameGroupBy with the column `SENTENCE_ID`
    """
    sentences_ids = np.arange(dfg.END_LINE.sum())
    dfg.loc[dfg.END_LINE, "SENTENCE_ID"] = sentences_ids
    dfg["SENTENCE_ID"] = dfg["SENTENCE_ID"].fillna(method="bfill")
    return dfg
_create_vocabulary(x)

Function to create a vocabulary for attributes in the training set.

PARAMETER DESCRIPTION
x

TYPE: iterable

RETURNS DESCRIPTION
dict
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
@staticmethod
def _create_vocabulary(x: iterable) -> dict:
    """Function to create a vocabulary for attributes in the training set.

    Parameters
    ----------
    x : iterable

    Returns
    -------
    dict
    """
    v = {}

    for i, key in enumerate(x):
        v[key] = i

    return v
_compute_B(df)

Function to compute B1 and B2

PARAMETER DESCRIPTION
df

TYPE: pd.DataFrame

RETURNS DESCRIPTION
pd.DataFrame
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
@staticmethod
def _compute_B(df: pd.DataFrame) -> pd.DataFrame:
    """Function to compute B1 and B2

    Parameters
    ----------
    df : pd.DataFrame

    Returns
    -------
    pd.DataFrame
    """

    data = df.groupby(["DOC_ID", "SENTENCE_ID"]).agg(l=("LENGTH", "sum"))
    df_t = df.loc[df.END_LINE, ["DOC_ID", "SENTENCE_ID"]].merge(
        data, left_on=["DOC_ID", "SENTENCE_ID"], right_index=True, how="left"
    )

    stats_doc = df_t.groupby("DOC_ID").agg(mu=("l", "mean"), sigma=("l", "std"))
    stats_doc["sigma"].replace(
        0.0, 1.0, inplace=True
    )  # Replace the 0 std by unit std, otherwise it breaks the code.
    stats_doc["cv"] = stats_doc["sigma"] / stats_doc["mu"]

    df_t = df_t.drop(columns=["DOC_ID", "SENTENCE_ID"])
    df2 = df.merge(df_t, left_index=True, right_index=True, how="left")

    df2 = df2.merge(stats_doc, on=["DOC_ID"], how="left")
    df2["l_norm"] = (df2["l"] - df2["mu"]) / df2["sigma"]

    df2["cv_bin"] = pd.cut(df2["cv"], bins=10)
    df2["B2"] = df2["cv_bin"].cat.codes

    df2["l_norm_bin"] = pd.cut(df2["l_norm"], bins=10)
    df2["B1"] = df2["l_norm_bin"].cat.codes

    return df2
_shift_col(df, col, new_col, direction='backward', fill=None)

Shifts a column one position into backward / forward direction.

PARAMETER DESCRIPTION
df

TYPE: pd.DataFrame

col

column to shift

TYPE: str

new_col

column name to save the results

TYPE: str

direction

one of {"backward", "forward"}, by default "backward"

TYPE: str, optional DEFAULT: 'backward'

fill

, by default None

TYPE: [type], optional DEFAULT: None

RETURNS DESCRIPTION
pd.DataFrame

same df with new_col added.
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
@staticmethod
def _shift_col(
    df: pd.DataFrame, col: str, new_col: str, direction="backward", fill=None
) -> pd.DataFrame:
    """Shifts a column one position into backward / forward direction.

    Parameters
    ----------
    df : pd.DataFrame
    col : str
        column to shift
    new_col : str
        column name to save the results
    direction : str, optional
        one of {"backward", "forward"}, by default "backward"
    fill : [type], optional
        , by default None

    Returns
    -------
    pd.DataFrame
        same df with `new_col` added.
    """
    df[new_col] = fill

    if direction == "backward":
        df.loc[df.index[:-1], new_col] = df[col].values[1:]

        different_doc_id = df["DOC_ID"].values[:-1] != df["DOC_ID"].values[1:]
        different_doc_id = np.append(different_doc_id, True)

    if direction == "forward":
        df.loc[df.index[1:], new_col] = df[col].values[:-1]
        different_doc_id = df["DOC_ID"].values[1:] != df["DOC_ID"].values[:-1]
        different_doc_id = np.append(True, different_doc_id)

    df.loc[different_doc_id, new_col] = fill
    return df
_get_attributes(doc, i=0)

Function to get the attributes of tokens of a spacy doc in a pd.DataFrame format.

PARAMETER DESCRIPTION
doc

spacy Doc

TYPE: Doc

i

document id, by default 0

TYPE: int, optional DEFAULT: 0

RETURNS DESCRIPTION
pd.DataFrame

Returns a dataframe with one line per token. It has the following columns : [ "ORTH", "LOWER", "SHAPE", "IS_DIGIT", "IS_SPACE", "IS_UPPER", "IS_PUNCT", "LENGTH", ]
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
@staticmethod
def _get_attributes(doc: Doc, i=0):
    """Function to get the attributes of tokens of a spacy doc in a pd.DataFrame format.

    Parameters
    ----------
    doc : Doc
        spacy Doc
    i : int, optional
        document id, by default 0

    Returns
    -------
    pd.DataFrame
        Returns a dataframe with one line per token. It has the following columns :
        `[
        "ORTH",
        "LOWER",
        "SHAPE",
        "IS_DIGIT",
        "IS_SPACE",
        "IS_UPPER",
        "IS_PUNCT",
        "LENGTH",
        ]`
    """
    attributes = [
        "ORTH",
        "LOWER",
        "SHAPE",
        "IS_DIGIT",
        "IS_SPACE",
        "IS_UPPER",
        "IS_PUNCT",
        "LENGTH",
    ]
    attributes_array = doc.to_array(attributes)
    attributes_df = pd.DataFrame(attributes_array, columns=attributes)
    attributes_df["DOC_ID"] = i
    boolean_attr = []
    for a in attributes:
        if a[:3] == "IS_":
            boolean_attr.append(a)
    attributes_df[boolean_attr] = attributes_df[boolean_attr].astype("boolean")
    return attributes_df
_get_string(_id, string_store)

Returns the string corresponding to the token_id

PARAMETER DESCRIPTION
_id

token id

TYPE: int

string_store

spaCy Language String Store

TYPE: StringStore

RETURNS DESCRIPTION
str

string representation of the token.
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
@staticmethod
def _get_string(_id: int, string_store: StringStore) -> str:
    """Returns the string corresponding to the token_id

    Parameters
    ----------
    _id : int
        token id
    string_store : StringStore
        spaCy Language String Store

    Returns
    -------
    str
        string representation of the token.
    """
    return string_store[_id]
_fit_one_hot_encoder(X)

Fit a one hot encoder.

PARAMETER DESCRIPTION
X

of shape (n,1)

TYPE: np.ndarray

RETURNS DESCRIPTION
OneHotEncoder
Source code in edsnlp/pipelines/core/endlines/endlinesmodel.py 
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
@staticmethod
def _fit_one_hot_encoder(X: np.ndarray) -> OneHotEncoder:
    """Fit a one hot encoder.

    Parameters
    ----------
    X : np.ndarray
        of shape (n,1)

    Returns
    -------
    OneHotEncoder
    """
    encoder = OneHotEncoder(handle_unknown="ignore")
    encoder.fit(X)
    return encoder

context

context

ContextAdder

Bases: BaseComponent

Provides a generic context adder component.

PARAMETER DESCRIPTION
nlp

The spaCy object.

TYPE: Language

context

The list of extensions to add to the Doc

TYPE: List[str]

Source code in edsnlp/pipelines/core/context/context.py 
 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
class ContextAdder(BaseComponent):
    """
    Provides a generic context adder component.

    Parameters
    ----------
    nlp : Language
        The spaCy object.
    context : List[str]
        The list of extensions to add to the `Doc`
    """

    def __init__(
        self,
        nlp: Language,
        context: List[str],
    ):

        self.nlp = nlp
        self.context = context
        self.set_extensions()

    def set_extensions(self):
        for col in self.context:
            if not Doc.has_extension(col):
                Doc.set_extension(col, default=None)

    def __call__(self, doc: Doc) -> Doc:
        return doc
nlp = nlp instance-attribute
context = context instance-attribute
__init__(nlp, context)
Source code in edsnlp/pipelines/core/context/context.py 
21
22
23
24
25
26
27
28
29
def __init__(
    self,
    nlp: Language,
    context: List[str],
):

    self.nlp = nlp
    self.context = context
    self.set_extensions()
set_extensions()
Source code in edsnlp/pipelines/core/context/context.py 
31
32
33
34
def set_extensions(self):
    for col in self.context:
        if not Doc.has_extension(col):
            Doc.set_extension(col, default=None)
__call__(doc)
Source code in edsnlp/pipelines/core/context/context.py 
36
37
def __call__(self, doc: Doc) -> Doc:
    return doc

factory

DEFAULT_CONFIG = dict(context=['note_id']) module-attribute
create_component(nlp, name, context)
Source code in edsnlp/pipelines/core/context/factory.py 
12
13
14
15
16
17
18
19
20
21
22
@Language.factory("eds.context", default_config=DEFAULT_CONFIG)
def create_component(
    nlp: Language,
    name: str,
    context: List[str],
):

    return ContextAdder(
        nlp,
        context=context,
    )

normalizer

normalizer

Normalizer

Bases: object

Normalisation pipeline. Modifies the NORM attribute, acting on four dimensions :

  • lowercase: using the default NORM
  • accents: deterministic and fixed-length normalisation of accents.
  • quotes: deterministic and fixed-length normalisation of quotation marks.
  • pollution: removal of pollutions.
PARAMETER DESCRIPTION
lowercase

Whether to remove case.

TYPE: bool

accents

Optional Accents object.

TYPE: Optional[Accents]

quotes

Optional Quotes object.

TYPE: Optional[Quotes]

pollution

Optional Pollution object.

TYPE: Optional[Pollution]

Source code in edsnlp/pipelines/core/normalizer/normalizer.py 
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
class Normalizer(object):
    """
    Normalisation pipeline. Modifies the `NORM` attribute,
    acting on four dimensions :

    - `lowercase`: using the default `NORM`
    - `accents`: deterministic and fixed-length normalisation of accents.
    - `quotes`: deterministic and fixed-length normalisation of quotation marks.
    - `pollution`: removal of pollutions.

    Parameters
    ----------
    lowercase : bool
        Whether to remove case.
    accents : Optional[Accents]
        Optional `Accents` object.
    quotes : Optional[Quotes]
        Optional `Quotes` object.
    pollution : Optional[Pollution]
        Optional `Pollution` object.
    """

    def __init__(
        self,
        lowercase: bool,
        accents: Optional[Accents],
        quotes: Optional[Quotes],
        pollution: Optional[Pollution],
    ):
        self.lowercase = lowercase
        self.accents = accents
        self.quotes = quotes
        self.pollution = pollution

    def __call__(self, doc: Doc) -> Doc:
        """
        Apply the normalisation pipeline, one component at a time.

        Parameters
        ----------
        doc : Doc
            spaCy `Doc` object

        Returns
        -------
        Doc
            Doc object with `NORM` attribute modified
        """
        if not self.lowercase:
            remove_lowercase(doc)
        if self.accents is not None:
            self.accents(doc)
        if self.quotes is not None:
            self.quotes(doc)
        if self.pollution is not None:
            self.pollution(doc)

        return doc
lowercase = lowercase instance-attribute
accents = accents instance-attribute
quotes = quotes instance-attribute
pollution = pollution instance-attribute
__init__(lowercase, accents, quotes, pollution)
Source code in edsnlp/pipelines/core/normalizer/normalizer.py 
33
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    lowercase: bool,
    accents: Optional[Accents],
    quotes: Optional[Quotes],
    pollution: Optional[Pollution],
):
    self.lowercase = lowercase
    self.accents = accents
    self.quotes = quotes
    self.pollution = pollution
__call__(doc)

Apply the normalisation pipeline, one component at a time.

PARAMETER DESCRIPTION
doc

spaCy Doc object

TYPE: Doc

RETURNS DESCRIPTION
Doc

Doc object with NORM attribute modified
Source code in edsnlp/pipelines/core/normalizer/normalizer.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __call__(self, doc: Doc) -> Doc:
    """
    Apply the normalisation pipeline, one component at a time.

    Parameters
    ----------
    doc : Doc
        spaCy `Doc` object

    Returns
    -------
    Doc
        Doc object with `NORM` attribute modified
    """
    if not self.lowercase:
        remove_lowercase(doc)
    if self.accents is not None:
        self.accents(doc)
    if self.quotes is not None:
        self.quotes(doc)
    if self.pollution is not None:
        self.pollution(doc)

    return doc

factory

DEFAULT_CONFIG = dict(accents=True, lowercase=True, quotes=True, pollution=True) module-attribute
create_component(nlp, name, accents, lowercase, quotes, pollution)
Source code in edsnlp/pipelines/core/normalizer/factory.py 
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
@deprecated_factory("normalizer", "eds.normalizer", default_config=DEFAULT_CONFIG)
@Language.factory("eds.normalizer", default_config=DEFAULT_CONFIG)
def create_component(
    nlp: Language,
    name: str,
    accents: Union[bool, Dict[str, Any]],
    lowercase: Union[bool, Dict[str, Any]],
    quotes: Union[bool, Dict[str, Any]],
    pollution: Union[bool, Dict[str, Any]],
):

    if accents:
        config = dict(**accents_config)
        if isinstance(accents, dict):
            config.update(accents)
        accents = registry.get("factories", "eds.accents")(nlp, "eds.accents", **config)

    if quotes:
        config = dict(**quotes_config)
        if isinstance(quotes, dict):
            config.update(quotes)
        quotes = registry.get("factories", "eds.quotes")(nlp, "eds.quotes", **config)

    if pollution:
        config = dict(**pollution_config)
        if isinstance(pollution, dict):
            config.update(pollution)
        pollution = registry.get("factories", "eds.pollution")(
            nlp, "eds.pollution", **config
        )

    normalizer = Normalizer(
        lowercase=lowercase,
        accents=accents or None,
        quotes=quotes or None,
        pollution=pollution or None,
    )

    return normalizer

pollution

patterns
information = "(?s)(=====+\\s*)?(L\\s*e\\s*s\\sdonnées\\s*administratives,\\s*sociales\\s*|I?nfo\\s*rmation\\s*aux?\\s*patients?|L[’']AP-HP\\s*collecte\\s*vos\\s*données\\s*administratives|L[’']Assistance\\s*Publique\\s*-\\s*Hôpitaux\\s*de\\s*Paris\\s*\\(?AP-HP\\)?\\s*a\\s*créé\\s*une\\s*base\\s*de\\s*données).{,2000}https?:\\/\\/recherche\\.aphp\\.fr\\/eds\\/droit-opposition[\\s\\.]*" module-attribute
bars = '(?i)([nbw]|_|-|=){5,}' module-attribute
pollution = dict(information=information, bars=bars) module-attribute
pollution
Pollution

Bases: BaseComponent

Tags pollution tokens.

Populates a number of spaCy extensions :

  • Token._.pollution : indicates whether the token is a pollution
  • Doc._.clean : lists non-pollution tokens
  • Doc._.clean_ : original text with pollutions removed.
  • Doc._.char_clean_span : method to create a Span using character indices extracted using the cleaned text.
PARAMETER DESCRIPTION
nlp

Language pipeline object

TYPE: Language

pollution

Dictionary containing regular expressions of pollution.

TYPE: Dict[str, Union[str, List[str]]]

Source code in edsnlp/pipelines/core/normalizer/pollution/pollution.py 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class Pollution(BaseComponent):
    """
    Tags pollution tokens.

    Populates a number of spaCy extensions :

    - `Token._.pollution` : indicates whether the token is a pollution
    - `Doc._.clean` : lists non-pollution tokens
    - `Doc._.clean_` : original text with pollutions removed.
    - `Doc._.char_clean_span` : method to create a Span using character
      indices extracted using the cleaned text.

    Parameters
    ----------
    nlp : Language
        Language pipeline object
    pollution : Dict[str, Union[str, List[str]]]
        Dictionary containing regular expressions of pollution.
    """

    # noinspection PyProtectedMember
    def __init__(
        self,
        nlp: Language,
        pollution: Optional[Dict[str, Union[str, List[str]]]],
    ):

        self.nlp = nlp

        if pollution is None:
            pollution = patterns.pollution

        self.pollution = pollution

        for k, v in self.pollution.items():
            if isinstance(v, str):
                self.pollution[k] = [v]

        self.regex_matcher = RegexMatcher()
        self.build_patterns()

    def build_patterns(self) -> None:
        """
        Builds the patterns for phrase matching.
        """

        # efficiently build spaCy matcher patterns
        for k, v in self.pollution.items():
            self.regex_matcher.add(k, v)

    def process(self, doc: Doc) -> List[Span]:
        """
        Find pollutions in doc and clean candidate negations to remove pseudo negations

        Parameters
        ----------
        doc:
            spaCy Doc object

        Returns
        -------
        pollution:
            list of pollution spans
        """

        pollutions = self.regex_matcher(doc, as_spans=True)
        pollutions = filter_spans(pollutions)

        return pollutions

    def __call__(self, doc: Doc) -> Doc:
        """
        Tags pollutions.

        Parameters
        ----------
        doc:
            spaCy Doc object

        Returns
        -------
        doc:
            spaCy Doc object, annotated for pollutions.
        """
        pollutions = self.process(doc)

        for pollution in pollutions:

            for token in pollution:
                token._.excluded = True

        doc.spans["pollutions"] = pollutions

        return doc
nlp = nlp instance-attribute
pollution = pollution instance-attribute
regex_matcher = RegexMatcher() instance-attribute
__init__(nlp, pollution)
Source code in edsnlp/pipelines/core/normalizer/pollution/pollution.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    nlp: Language,
    pollution: Optional[Dict[str, Union[str, List[str]]]],
):

    self.nlp = nlp

    if pollution is None:
        pollution = patterns.pollution

    self.pollution = pollution

    for k, v in self.pollution.items():
        if isinstance(v, str):
            self.pollution[k] = [v]

    self.regex_matcher = RegexMatcher()
    self.build_patterns()
build_patterns()

Builds the patterns for phrase matching.

Source code in edsnlp/pipelines/core/normalizer/pollution/pollution.py 
54
55
56
57
58
59
60
61
def build_patterns(self) -> None:
    """
    Builds the patterns for phrase matching.
    """

    # efficiently build spaCy matcher patterns
    for k, v in self.pollution.items():
        self.regex_matcher.add(k, v)
process(doc)

Find pollutions in doc and clean candidate negations to remove pseudo negations

PARAMETER DESCRIPTION
doc

spaCy Doc object

TYPE: Doc

RETURNS DESCRIPTION
pollution

list of pollution spans
Source code in edsnlp/pipelines/core/normalizer/pollution/pollution.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def process(self, doc: Doc) -> List[Span]:
    """
    Find pollutions in doc and clean candidate negations to remove pseudo negations

    Parameters
    ----------
    doc:
        spaCy Doc object

    Returns
    -------
    pollution:
        list of pollution spans
    """

    pollutions = self.regex_matcher(doc, as_spans=True)
    pollutions = filter_spans(pollutions)

    return pollutions
__call__(doc)

Tags pollutions.

PARAMETER DESCRIPTION
doc

spaCy Doc object

TYPE: Doc

RETURNS DESCRIPTION
doc

spaCy Doc object, annotated for pollutions.
Source code in edsnlp/pipelines/core/normalizer/pollution/pollution.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def __call__(self, doc: Doc) -> Doc:
    """
    Tags pollutions.

    Parameters
    ----------
    doc:
        spaCy Doc object

    Returns
    -------
    doc:
        spaCy Doc object, annotated for pollutions.
    """
    pollutions = self.process(doc)

    for pollution in pollutions:

        for token in pollution:
            token._.excluded = True

    doc.spans["pollutions"] = pollutions

    return doc
factory
DEFAULT_CONFIG = dict(pollution=None) module-attribute
create_component(nlp, name, pollution)
Source code in edsnlp/pipelines/core/normalizer/pollution/factory.py 
14
15
16
17
18
19
20
21
22
23
24
@deprecated_factory("pollution", "eds.pollution", default_config=DEFAULT_CONFIG)
@Language.factory("eds.pollution", default_config=DEFAULT_CONFIG)
def create_component(
    nlp: Language,
    name: str,
    pollution: Optional[Dict[str, Union[str, List[str]]]],
):
    return Pollution(
        nlp,
        pollution=pollution,
    )

accents

patterns
accents: List[Tuple[str, str]] = [('ç', 'c'), ('àáâä', 'a'), ('èéêë', 'e'), ('ìíîï', 'i'), ('òóôö', 'o'), ('ùúûü', 'u')] module-attribute
accents
Accents

Bases: object

Normalises accents, using a same-length strategy.

PARAMETER DESCRIPTION
accents

List of accentuated characters and their transcription.

TYPE: List[Tuple[str, str]]

Source code in edsnlp/pipelines/core/normalizer/accents/accents.py 
 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
class Accents(object):
    """
    Normalises accents, using a same-length strategy.

    Parameters
    ----------
    accents : List[Tuple[str, str]]
        List of accentuated characters and their transcription.
    """

    def __init__(self, accents: Optional[List[Tuple[str, str]]]) -> None:
        if accents is None:
            accents = patterns.accents

        self.translation_table = str.maketrans(
            "".join(accent_group for accent_group, _ in accents),
            "".join(rep * len(accent_group) for accent_group, rep in accents),
        )

    def __call__(self, doc: Doc) -> Doc:
        """
        Remove accents from spacy `NORM` attribute.

        Parameters
        ----------
        doc : Doc
            The spaCy `Doc` object.

        Returns
        -------
        Doc
            The document, with accents removed in `Token.norm_`.
        """

        for token in doc:
            token.norm_ = token.norm_.translate(self.translation_table)

        return doc
translation_table = str.maketrans(''.join(accent_group for (accent_group, _) in accents), ''.join(rep * len(accent_group) for (accent_group, rep) in accents)) instance-attribute
__init__(accents)
Source code in edsnlp/pipelines/core/normalizer/accents/accents.py 
18
19
20
21
22
23
24
25
def __init__(self, accents: Optional[List[Tuple[str, str]]]) -> None:
    if accents is None:
        accents = patterns.accents

    self.translation_table = str.maketrans(
        "".join(accent_group for accent_group, _ in accents),
        "".join(rep * len(accent_group) for accent_group, rep in accents),
    )
__call__(doc)

Remove accents from spacy NORM attribute.

PARAMETER DESCRIPTION
doc

The spaCy Doc object.

TYPE: Doc

RETURNS DESCRIPTION
Doc

The document, with accents removed in Token.norm_.
Source code in edsnlp/pipelines/core/normalizer/accents/accents.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def __call__(self, doc: Doc) -> Doc:
    """
    Remove accents from spacy `NORM` attribute.

    Parameters
    ----------
    doc : Doc
        The spaCy `Doc` object.

    Returns
    -------
    Doc
        The document, with accents removed in `Token.norm_`.
    """

    for token in doc:
        token.norm_ = token.norm_.translate(self.translation_table)

    return doc
factory
DEFAULT_CONFIG = dict(accents=None) module-attribute
create_component(nlp, name, accents)
Source code in edsnlp/pipelines/core/normalizer/accents/factory.py 
14
15
16
17
18
19
20
21
22
23
@deprecated_factory("accents", "eds.accents", default_config=DEFAULT_CONFIG)
@Language.factory("eds.accents", default_config=DEFAULT_CONFIG)
def create_component(
    nlp: Language,
    name: str,
    accents: Optional[List[Tuple[str, str]]],
):
    return Accents(
        accents=accents,
    )

lowercase

factory
remove_lowercase(doc)

Add case on the NORM custom attribute. Should always be applied first.

PARAMETER DESCRIPTION
doc

The spaCy Doc object.

TYPE: Doc

RETURNS DESCRIPTION
Doc

The document, with case put back in NORM.
Source code in edsnlp/pipelines/core/normalizer/lowercase/factory.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@Language.component("remove-lowercase")
@Language.component("eds.remove-lowercase")
def remove_lowercase(doc: Doc):
    """
    Add case on the `NORM` custom attribute. Should always be applied first.

    Parameters
    ----------
    doc : Doc
        The spaCy `Doc` object.

    Returns
    -------
    Doc
        The document, with case put back in `NORM`.
    """

    for token in doc:
        token.norm_ = token.text

    return doc

quotes

quotes
Quotes

Bases: object

We normalise quotes, following this source <https://www.cl.cam.ac.uk/~mgk25/ucs/quotes.html>_.

PARAMETER DESCRIPTION
quotes

List of quotation characters and their transcription.

TYPE: List[Tuple[str, str]]

Source code in edsnlp/pipelines/core/normalizer/quotes/quotes.py 
 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
class Quotes(object):
    """
    We normalise quotes, following this
    `source <https://www.cl.cam.ac.uk/~mgk25/ucs/quotes.html>`_.

    Parameters
    ----------
    quotes : List[Tuple[str, str]]
        List of quotation characters and their transcription.
    """

    def __init__(self, quotes: Optional[List[Tuple[str, str]]]) -> None:
        if quotes is None:
            quotes = quotes_and_apostrophes

        self.translation_table = str.maketrans(
            "".join(quote_group for quote_group, _ in quotes),
            "".join(rep * len(quote_group) for quote_group, rep in quotes),
        )

    def __call__(self, doc: Doc) -> Doc:
        """
        Normalises quotes.

        Parameters
        ----------
        doc : Doc
            Document to process.

        Returns
        -------
        Doc
            Same document, with quotes normalised.
        """

        for token in doc:
            token.norm_ = token.norm_.translate(self.translation_table)

        return doc
translation_table = str.maketrans(''.join(quote_group for (quote_group, _) in quotes), ''.join(rep * len(quote_group) for (quote_group, rep) in quotes)) instance-attribute
__init__(quotes)
Source code in edsnlp/pipelines/core/normalizer/quotes/quotes.py 
19
20
21
22
23
24
25
26
def __init__(self, quotes: Optional[List[Tuple[str, str]]]) -> None:
    if quotes is None:
        quotes = quotes_and_apostrophes

    self.translation_table = str.maketrans(
        "".join(quote_group for quote_group, _ in quotes),
        "".join(rep * len(quote_group) for quote_group, rep in quotes),
    )
__call__(doc)

Normalises quotes.

PARAMETER DESCRIPTION
doc

Document to process.

TYPE: Doc

RETURNS DESCRIPTION
Doc

Same document, with quotes normalised.
Source code in edsnlp/pipelines/core/normalizer/quotes/quotes.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __call__(self, doc: Doc) -> Doc:
    """
    Normalises quotes.

    Parameters
    ----------
    doc : Doc
        Document to process.

    Returns
    -------
    Doc
        Same document, with quotes normalised.
    """

    for token in doc:
        token.norm_ = token.norm_.translate(self.translation_table)

    return doc
patterns
quotes: List[str] = ['＂', '〃', 'ײ', '᳓', '″', '״', '‶', '˶', 'ʺ', '“', '”', '˝', '‟'] module-attribute
apostrophes: List[str] = ['｀', '΄', '＇', 'ˈ', 'ˊ', 'ᑊ', 'ˋ', 'ꞌ', 'ᛌ', '𖽒', '𖽑', '‘', '’', 'י', '՚', '‛', '՝', '`', '`', '′', '׳', '´', 'ʹ', '˴', 'ߴ', '‵', 'ߵ', 'ʹ', 'ʻ', 'ʼ', '´', '᾽', 'ʽ', '῾', 'ʾ', '᾿'] module-attribute
quotes_and_apostrophes: List[Tuple[str, str]] = [(''.join(quotes), '"'), (''.join(apostrophes), "'")] module-attribute
factory
DEFAULT_CONFIG = dict(quotes=None) module-attribute
create_component(nlp, name, quotes)
Source code in edsnlp/pipelines/core/normalizer/quotes/factory.py 
14
15
16
17
18
19
20
21
22
23
@deprecated_factory("quotes", "eds.quotes", default_config=DEFAULT_CONFIG)
@Language.factory("eds.quotes", default_config=DEFAULT_CONFIG)
def create_component(
    nlp: Language,
    name: str,
    quotes: Optional[List[Tuple[str, str]]],
):
    return Quotes(
        quotes=quotes,
    )

advanced

factory

DEFAULT_CONFIG = dict(window=10, verbose=0, ignore_excluded=False, attr='NORM') module-attribute
create_component(nlp, name, regex_config, window, verbose, ignore_excluded, attr)
Source code in edsnlp/pipelines/core/advanced/factory.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@deprecated_factory(
    "advanced-regex", "eds.advanced-regex", default_config=DEFAULT_CONFIG
)
@Language.factory("eds.advanced-regex", default_config=DEFAULT_CONFIG)
def create_component(
    nlp: Language,
    name: str,
    regex_config: Dict[str, Any],
    window: int,
    verbose: int,
    ignore_excluded: bool,
    attr: str,
):

    return AdvancedRegex(
        nlp,
        regex_config=regex_config,
        window=window,
        verbose=verbose,
        ignore_excluded=ignore_excluded,
        attr=attr,
    )

advanced

AdvancedRegex

Bases: GenericMatcher

Allows additional matching in the surrounding context of the main match group, for qualification/filtering.

PARAMETER DESCRIPTION
nlp

spaCy Language object.

TYPE: Language

regex_config

Configuration for the main expression.

TYPE: Dict[str, Any]

window

Number of tokens to consider before and after the main expression.

TYPE: int

attr

Attribute to match on, eg TEXT, NORM, etc.

TYPE: str

verbose

Verbosity level, useful for debugging.

TYPE: int

ignore_excluded

Whether to skip excluded tokens.

TYPE: bool

Source code in edsnlp/pipelines/core/advanced/advanced.py 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
class AdvancedRegex(GenericMatcher):
    """
    Allows additional matching in the surrounding context of the main match group,
    for qualification/filtering.

    Parameters
    ----------
    nlp : Language
        spaCy `Language` object.
    regex_config : Dict[str, Any]
        Configuration for the main expression.
    window : int
        Number of tokens to consider before and after the main expression.
    attr : str
        Attribute to match on, eg `TEXT`, `NORM`, etc.
    verbose : int
        Verbosity level, useful for debugging.
    ignore_excluded : bool
        Whether to skip excluded tokens.
    """

    def __init__(
        self,
        nlp: Language,
        regex_config: Dict[str, Any],
        window: int,
        attr: str,
        verbose: int,
        ignore_excluded: bool,
    ):
        self.regex_config = _check_regex_config(regex_config)
        self.window = window
        regex = regex_config

        self.verbose = verbose

        super().__init__(
            nlp=nlp,
            terms=dict(),
            regex=regex,
            attr=attr,
            ignore_excluded=ignore_excluded,
        )

        self.ignore_excluded = ignore_excluded

        self.set_extensions()

    @staticmethod
    def set_extensions() -> None:
        if not Doc.has_extension("my_ents"):
            Doc.set_extension("my_ents", default=[])

        if not Span.has_extension("matcher_name"):
            Span.set_extension("matcher_name", default=None)

        if not Span.has_extension("before_extract"):
            Span.set_extension("before_extract", default=None)
        if not Span.has_extension("after_extract"):
            Span.set_extension("after_extract", default=None)

        if not Span.has_extension("window"):
            Span.set_extension("window", default=None)

        if not Span.has_extension("before_snippet"):
            Span.set_extension("before_snippet", default=None)
        if not Span.has_extension("after_snippet"):
            Span.set_extension("after_snippet", default=None)

    def process(self, doc: Doc) -> List[Span]:
        """
        Process the document, looking for named entities.

        Parameters
        ----------
        doc : Doc
            spaCy Doc object

        Returns
        -------
        List[Span]
            List of detected spans.
        """

        ents = super().process(doc)
        ents = self._postprocessing_pipeline(ents)

        return ents

    def __call__(self, doc: Doc) -> Doc:
        """
        Adds spans to document.

        Parameters
        ----------
        doc:
            spaCy Doc object

        Returns
        -------
        doc:
            spaCy Doc object, annotated for extracted terms.
        """

        ents = self.process(doc)

        ents, discarded = filter_spans(list(doc.ents) + ents, return_discarded=True)

        doc.ents = ents

        if "discarded" not in doc.spans:
            doc.spans["discarded"] = []
        doc.spans["discarded"].extend(discarded)

        return doc

    def _postprocessing_pipeline(self, ents: List[Span]):
        # add a window within the sentence around entities
        ents = [self._add_window(ent) for ent in ents]

        # Remove entities based on the snippet located just before and after the entity
        ents = filter(self._exclude_filter, ents)

        # Extract informations from the entity's context via regex
        ents = [self._snippet_extraction(ent) for ent in ents]

        return ents

    def _add_window(self, ent: Span) -> Span:
        ent._.window = ent.doc[
            max(ent.start - self.window, ent.sent.start) : min(
                ent.end + self.window, ent.sent.end
            )
        ]

        # include the entity in the snippets so that we can extract
        # the number when it is attached to the word, e.g. "3PA"
        ent._.before_snippet = ent.doc[
            max(ent.start - self.window, ent.sent.start) : ent.end
        ]
        ent._.after_snippet = ent.doc[
            ent.start : min(ent.end + self.window, ent.sent.end)
        ]
        return ent

    def get_text(self, span: Span, label) -> str:
        attr = self.regex_config[label].get("attr", self.attr)

        return get_text(
            doclike=span,
            attr=attr,
            ignore_excluded=self.ignore_excluded,
        )

    def _exclude_filter(self, ent: Span) -> Span:
        label = ent.label_

        before_exclude = self.regex_config[label].get("before_exclude", None)
        after_exclude = self.regex_config[label].get("after_exclude", None)

        if before_exclude is not None:
            t = ent._.before_snippet
            t = self.get_text(t, label)
            if re.compile(before_exclude).search(t) is not None:
                if self.verbose:
                    logger.info(
                        f"excluded (before) string: {t} - pattern {before_exclude}"
                    )
                return False

        if after_exclude is not None:
            t = ent._.after_snippet
            t = self.get_text(t, label)
            if re.compile(after_exclude).search(t) is not None:
                if self.verbose:
                    logger.info(
                        f"excluded (after) string: {t} - pattern {after_exclude}"
                    )
                return False

        return True

    def _snippet_extraction(self, ent: Span) -> Span:
        label = ent.label_

        before_extract = self.regex_config[label].get("before_extract", [])
        after_extract = self.regex_config[label].get("after_extract", [])

        if type(before_extract) == str:
            before_extract = [before_extract]
        if type(after_extract) == str:
            after_extract = [after_extract]

        t = ent._.before_snippet
        t = self.get_text(t, label)
        ent._.before_extract = []
        for pattern in before_extract:
            pattern = re.compile(pattern)
            match = pattern.search(t)
            ent._.before_extract.append(match.groups()[0] if match else None)

        t = ent._.after_snippet
        t = self.get_text(t, label)
        ent._.after_extract = []
        for pattern in after_extract:
            pattern = re.compile(pattern)
            match = pattern.search(t)
            ent._.after_extract.append(match.groups()[0] if match else None)

        return ent
regex_config = _check_regex_config(regex_config) instance-attribute
window = window instance-attribute
verbose = verbose instance-attribute
ignore_excluded = ignore_excluded instance-attribute
__init__(nlp, regex_config, window, attr, verbose, ignore_excluded)
Source code in edsnlp/pipelines/core/advanced/advanced.py 
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
def __init__(
    self,
    nlp: Language,
    regex_config: Dict[str, Any],
    window: int,
    attr: str,
    verbose: int,
    ignore_excluded: bool,
):
    self.regex_config = _check_regex_config(regex_config)
    self.window = window
    regex = regex_config

    self.verbose = verbose

    super().__init__(
        nlp=nlp,
        terms=dict(),
        regex=regex,
        attr=attr,
        ignore_excluded=ignore_excluded,
    )

    self.ignore_excluded = ignore_excluded

    self.set_extensions()
set_extensions()
Source code in edsnlp/pipelines/core/advanced/advanced.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@staticmethod
def set_extensions() -> None:
    if not Doc.has_extension("my_ents"):
        Doc.set_extension("my_ents", default=[])

    if not Span.has_extension("matcher_name"):
        Span.set_extension("matcher_name", default=None)

    if not Span.has_extension("before_extract"):
        Span.set_extension("before_extract", default=None)
    if not Span.has_extension("after_extract"):
        Span.set_extension("after_extract", default=None)

    if not Span.has_extension("window"):
        Span.set_extension("window", default=None)

    if not Span.has_extension("before_snippet"):
        Span.set_extension("before_snippet", default=None)
    if not Span.has_extension("after_snippet"):
        Span.set_extension("after_snippet", default=None)
process(doc)

Process the document, looking for named entities.

PARAMETER DESCRIPTION
doc

spaCy Doc object

TYPE: Doc

RETURNS DESCRIPTION
List[Span]

List of detected spans.
Source code in edsnlp/pipelines/core/advanced/advanced.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def process(self, doc: Doc) -> List[Span]:
    """
    Process the document, looking for named entities.

    Parameters
    ----------
    doc : Doc
        spaCy Doc object

    Returns
    -------
    List[Span]
        List of detected spans.
    """

    ents = super().process(doc)
    ents = self._postprocessing_pipeline(ents)

    return ents
__call__(doc)

Adds spans to document.

PARAMETER DESCRIPTION
doc

spaCy Doc object

TYPE: Doc

RETURNS DESCRIPTION
doc

spaCy Doc object, annotated for extracted terms.
Source code in edsnlp/pipelines/core/advanced/advanced.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def __call__(self, doc: Doc) -> Doc:
    """
    Adds spans to document.

    Parameters
    ----------
    doc:
        spaCy Doc object

    Returns
    -------
    doc:
        spaCy Doc object, annotated for extracted terms.
    """

    ents = self.process(doc)

    ents, discarded = filter_spans(list(doc.ents) + ents, return_discarded=True)

    doc.ents = ents

    if "discarded" not in doc.spans:
        doc.spans["discarded"] = []
    doc.spans["discarded"].extend(discarded)

    return doc
_postprocessing_pipeline(ents)
Source code in edsnlp/pipelines/core/advanced/advanced.py 
129
130
131
132
133
134
135
136
137
138
139
def _postprocessing_pipeline(self, ents: List[Span]):
    # add a window within the sentence around entities
    ents = [self._add_window(ent) for ent in ents]

    # Remove entities based on the snippet located just before and after the entity
    ents = filter(self._exclude_filter, ents)

    # Extract informations from the entity's context via regex
    ents = [self._snippet_extraction(ent) for ent in ents]

    return ents
_add_window(ent)
Source code in edsnlp/pipelines/core/advanced/advanced.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def _add_window(self, ent: Span) -> Span:
    ent._.window = ent.doc[
        max(ent.start - self.window, ent.sent.start) : min(
            ent.end + self.window, ent.sent.end
        )
    ]

    # include the entity in the snippets so that we can extract
    # the number when it is attached to the word, e.g. "3PA"
    ent._.before_snippet = ent.doc[
        max(ent.start - self.window, ent.sent.start) : ent.end
    ]
    ent._.after_snippet = ent.doc[
        ent.start : min(ent.end + self.window, ent.sent.end)
    ]
    return ent
get_text(span, label)
Source code in edsnlp/pipelines/core/advanced/advanced.py 
158
159
160
161
162
163
164
165
def get_text(self, span: Span, label) -> str:
    attr = self.regex_config[label].get("attr", self.attr)

    return get_text(
        doclike=span,
        attr=attr,
        ignore_excluded=self.ignore_excluded,
    )
_exclude_filter(ent)
Source code in edsnlp/pipelines/core/advanced/advanced.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def _exclude_filter(self, ent: Span) -> Span:
    label = ent.label_

    before_exclude = self.regex_config[label].get("before_exclude", None)
    after_exclude = self.regex_config[label].get("after_exclude", None)

    if before_exclude is not None:
        t = ent._.before_snippet
        t = self.get_text(t, label)
        if re.compile(before_exclude).search(t) is not None:
            if self.verbose:
                logger.info(
                    f"excluded (before) string: {t} - pattern {before_exclude}"
                )
            return False

    if after_exclude is not None:
        t = ent._.after_snippet
        t = self.get_text(t, label)
        if re.compile(after_exclude).search(t) is not None:
            if self.verbose:
                logger.info(
                    f"excluded (after) string: {t} - pattern {after_exclude}"
                )
            return False

    return True
_snippet_extraction(ent)
Source code in edsnlp/pipelines/core/advanced/advanced.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def _snippet_extraction(self, ent: Span) -> Span:
    label = ent.label_

    before_extract = self.regex_config[label].get("before_extract", [])
    after_extract = self.regex_config[label].get("after_extract", [])

    if type(before_extract) == str:
        before_extract = [before_extract]
    if type(after_extract) == str:
        after_extract = [after_extract]

    t = ent._.before_snippet
    t = self.get_text(t, label)
    ent._.before_extract = []
    for pattern in before_extract:
        pattern = re.compile(pattern)
        match = pattern.search(t)
        ent._.before_extract.append(match.groups()[0] if match else None)

    t = ent._.after_snippet
    t = self.get_text(t, label)
    ent._.after_extract = []
    for pattern in after_extract:
        pattern = re.compile(pattern)
        match = pattern.search(t)
        ent._.after_extract.append(match.groups()[0] if match else None)

    return ent
_check_regex_config(regex_config)
Source code in edsnlp/pipelines/core/advanced/advanced.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def _check_regex_config(regex_config):
    for k, v in regex_config.items():
        if type(v) is not dict:
            raise TypeError(
                f"The value of the key {k} is of type {type(v)}, but a dict is expected"
            )

        single_group_regex_keys = ["before_extract", "after_extract"]

        for single_group_regex_key in single_group_regex_keys:
            if single_group_regex_key in v:
                # ensure it is a list
                if type(v[single_group_regex_key]) is not list:
                    v[single_group_regex_key] = [v[single_group_regex_key]]

                for i, regex in enumerate(v[single_group_regex_key]):
                    n_groups = re.compile(regex).groups

                    if n_groups == 0:
                        # Adding grouping parenthesis
                        v[single_group_regex_key][i] = r"(" + regex + r")"
                    elif n_groups != 1:
                        # Accepting only 1 group per regex
                        raise ValueError(
                            f"The RegEx for {repr(k)} ({repr(regex)}) "
                            f"stored in {repr(single_group_regex_key)} "
                            f"contains {n_groups} capturing groups, 1 expected"
                        )

    return regex_config
Back to top