First commit

.gitignore

@@ -0,0 +1,53 @@
+# Python
+__pycache__/
+
+# Distribution / packaging
+init
+.Python
+env/
+venv/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*,cover
+.hypothesis/
+.pytest_cache/
+
+# Notebooks
+.ipynb_checkpoints/
+*.ipynb
+
+# Data
+*.csv
+*.pickle
+*.txt
+*.xls
+*.xlsx
+*.tar.gz
+
+# Editors
+.idea
+.vscode
+
+# Files
+.DS_Store

LICENSE

@@ -0,0 +1,11 @@
+Copyright 2021 Assistance Publique - Hôpitaux de Paris
+
+Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

README.md

@@ -0,0 +1,3 @@
+# NLPTools
+
+A simple library to group together the different pre-processing pipelines that are used at AP-HP, as Spacy component.

nlptools/rules/__init__.py

@@ -0,0 +1,2 @@
+import nlptools.rules.sections
+import nlptools.rules.pollution

nlptools/rules/base.py

@@ -0,0 +1,95 @@
+from itertools import chain
+from typing import List, Optional, Tuple
+
+from spacy.tokens import Doc, Span
+
+if not Doc.has_extension('note_id'):
+    Doc.set_extension('note_id', default=None)
+
+
+class BaseComponent(object):
+    """
+    Base component that contains the logic for :
+    - boundaries selections
+    - match filtering
+    """
+
+    split_on_punct = True
+
+    @staticmethod
+    def _filter_matches(matches: List[Span]) -> List[Span]:
+        """
+        Filter matches to remove duplicates and inclusions.
+
+        Arguments
+        ---------
+        matches: List of matches (spans).
+
+        Returns
+        -------
+        filtered_matches: List of filtered matches.
+        """
+
+        filtered_matches = []
+
+        for span in matches:
+
+            if not set(range(span.start, span.end)).intersection(
+                    chain(*[set(range(s.start, s.end)) for s in filtered_matches])
+            ):
+                filtered_matches.append(span)
+
+            else:
+                s = set(range(span.start, span.end))
+
+                for match in filtered_matches[:]:
+                    m = set(range(match.start, match.end))
+
+                    if m & s:
+                        tokens = sorted(list(s | m))
+                        start, end = tokens[0], tokens[-1]
+
+                        new_span = Span(span.doc, start, end + 1, label=span.label_)
+
+                        filtered_matches.remove(match)
+                        filtered_matches.append(new_span)
+                        break
+
+        return filtered_matches
+
+    def _boundaries(self, doc: Doc, terminations: Optional[List[Span]] = None) -> List[Tuple[int, int]]:
+        """
+        Create sub sentences based sentences and terminations found in text.
+        
+        Parameters
+        ----------
+        doc: spaCy Doc object
+        terminations: List of tuples with (match_id, start, end)
+            
+        Returns
+        -------
+        boundaries: List of tuples with (start, end) of spans
+        """
+
+        if terminations is None:
+            terminations = []
+
+        sent_starts = [sent.start for sent in doc.sents]
+        termination_starts = [t.start for t in terminations]
+
+        if self.split_on_punct:
+            punctuations = [t.i for t in doc if t.is_punct and '-' not in t.text]
+        else:
+            punctuations = []
+
+        starts = sent_starts + termination_starts + punctuations + [len(doc)]
+
+        # Remove duplicates
+        starts = list(set(starts))
+
+        # Sort starts
+        starts.sort()
+
+        boundaries = [(start, end) for start, end in zip(starts[:-1], starts[1:])]
+
+        return boundaries

nlptools/rules/pollution/__init__.py

@@ -0,0 +1,2 @@
+from .pollution import Pollution
+from .terms import pollutions

nlptools/rules/pollution/pollution.py

@@ -0,0 +1,227 @@
+from typing import List, Tuple, Dict, Optional
+
+from spacy.language import Language
+from spacy.tokens import Token, Span, Doc
+from spaczz.matcher import RegexMatcher
+
+from nlptools.rules.base import BaseComponent
+from nlptools.rules.pollution import terms
+
+import numpy as np
+
+
+if not Doc.has_extension('note_id'):
+    Doc.set_extension('note_id', default=None)
+
+
+def clean_getter(doc: Doc) -> List[Token]:
+    """
+    Gets a list of tokens with pollution removed.
+    
+    Arguments
+    ---------
+    doc: Spacy Doc object.
+    
+    Returns
+    -------
+    tokens: List of clean tokens.
+    """
+
+    tokens = []
+
+    for token in doc:
+        if not token._.pollution:
+            tokens.append(token)
+
+    return tokens
+
+
+def clean2original(doc: Doc) -> np.ndarray:
+    """
+    Creates an alignment array to convert spans from the cleaned 
+    textual representation to the original text object.
+    
+    Arguments
+    ---------
+    doc: Spacy Doc object.
+    
+    Returns
+    -------
+    alignement: Alignment array.
+    """
+
+    lengths = np.array([len(token.text_with_ws) for token in doc])
+    pollution = np.array([token._.pollution for token in doc])
+
+    current = 0
+
+    clean = []
+
+    for l, p in zip(lengths, pollution):
+        if not p:
+            current += l
+        clean.append(current)
+    clean = np.array(clean)
+
+    alignment = np.stack([lengths.cumsum(), clean])
+
+    return alignment
+
+
+def align(doc: Doc, index: int) -> int:
+    """
+    Aligns a character found in the clean text with 
+    its index in the original text.
+    
+    Arguments
+    ---------
+    doc: Spacy Doc object.
+    index: Character index in the clean text.
+    
+    Returns
+    -------
+    index: Character index in the original text.
+    """
+
+    if index < 0:
+        index = len(doc._.clean_) - index
+
+    alignment = clean2original(doc)
+    offset = alignment[0] - alignment[1]
+
+    return index + offset[alignment[1] < index][-1]
+
+def char_clean2original(
+    doc: Doc, 
+    start: int, 
+    end: int, 
+    alignment_mode: Optional[str] = 'strict',
+) -> Span:
+    """
+    Returns a Spacy Span object from character span computed on the clean text.
+    
+    Arguments
+    ---------
+    doc: Spacy Doc object
+    start: Character index of the beginning of the expression in the clean text.
+    end: Character index of the end of the expression in the clean text.
+    alignment_mode: Alignment mode. See https://spacy.io/api/doc#char_span.
+    
+    Returns
+    -------
+    span: Span in the original text.
+    """
+
+    start, end = align(doc, start), align(doc, end)
+    return doc.char_span(start, end, alignment_mode=alignment_mode)
+
+
+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.
+    """
+
+    def __init__(
+            self,
+            nlp: Language,
+            pollution: Dict[str, str],
+    ):
+
+        self.nlp = nlp
+
+        self.pollution = pollution
+
+        if not Token.has_extension('pollution'):
+            Token.set_extension('pollution', default=False)
+
+        if not Doc.has_extension('clean'):
+            Doc.set_extension('clean', getter=clean_getter)
+
+        if not Doc.has_extension('clean_'):
+            Doc.set_extension('clean_', getter=lambda doc: ''.join([t.text_with_ws for t in doc._.clean]))
+
+        if not Doc.has_extension('char_clean_span'):
+            Doc.set_extension('char_clean_span', method=char_clean2original)
+
+        self.build_patterns()
+
+    def build_patterns(self) -> None:
+        """
+        Builds the patterns for phrase matching.
+        """
+
+        # efficiently build spaCy matcher patterns
+        self.matcher = RegexMatcher(self.nlp.vocab)
+
+        for term in self.pollution.values():
+            self.matcher.add("pollution", [term])
+
+    def process_pollutions(self, doc: Doc) -> Tuple[List[Span], List[Span], 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
+        """
+
+        matches = self.matcher(doc)
+        pollutions = [
+            Span(doc, start, end, label='neg_pseudo')
+            for match_id, start, end, ratio in matches
+            if match_id == "pollution"
+        ]
+
+        pollutions = self._filter_matches(pollutions)
+
+        return pollutions
+
+    def __call__(self, doc: Doc) -> Doc:
+        """
+        Tags pollutions.
+
+        Parameters
+        ----------
+        doc: spaCy Doc object
+        
+        Returns
+        -------
+        doc: spaCy Doc object, annotated for negation
+        """
+        pollutions = self.process_pollutions(doc)
+
+        for pollution in pollutions:
+
+            for token in pollution:
+                token._.pollution = True
+
+        # for token in doc:
+        #     if token.is_space:
+        #         token._.pollution = True
+
+        return doc
+
+
+default_config = dict(
+    pollution=terms.pollution,
+)
+
+
+@Language.factory("pollution", default_config=default_config)
+def create_pollution_component(
+        nlp: Language,
+        name: str,
+        pollution: Dict[str, str],
+):
+    return Pollution(nlp, pollution=pollution)