wuenlp_tools.models.preprocess 

 1from __future__ import annotations
 2
 3import tempfile
 4from pathlib import Path
 5from typing import Type, Set
 6
 7from wuenlp.impl.UIMANLPStructs import UIMADocument, UIMASpan
 8from wuenlp.utils.import_mikalli import from_mikalli
 9
10from wuenlp_tools.pipeline import PipelineStep, PipelineProcessor, PipelinePreprocessor
11from wuenlp_tools.pipeline import PipelineCapability
12from wuenlp_tools.utils import resource_dir
13from wuenlp_tools.utils.api import query_wuenlp_api
14
15
16def annotate_text(text: str,
17                  url: str = '/kallimachosPipeline',
18                  ignore_types: Set[str] = set(), log_level: str = "ERROR") -> UIMADocument:
19    myobj = {'text': text}
20    headers = {'Content-type': 'application/json; charset=utf-8'}
21
22    response = query_wuenlp_api(url, headers, myobj)
23
24    with tempfile.NamedTemporaryFile(mode="w") as tf:
25        tf.write(response.text.strip("\x00"))
26        tf.flush()
27
28        # logger.debug(Path(tf.name).read_text())
29        doc = from_mikalli(Path(tf.name), mikalli_typesystem_path=resource_dir / "MiKalliTypesystem.xml",
30                           use_system_coref_id=False,
31                           ignore_types=ignore_types,
32                           log_level=log_level)
33
34    return doc
35
36
37class PreprocessPipelineProcessor(PipelinePreprocessor):
38    def __call__(self, doc: UIMADocument | str, unit_type: Type[UIMASpan], **kwargs) -> UIMADocument:  # type: ignore
39        if isinstance(doc, UIMADocument):
40            text = doc.text
41        else:
42            text = doc
43        doc = annotate_text(text)
44        return doc
45
46
47Preprocessor = PipelineStep(
48    "Preprocess",
49    PreprocessPipelineProcessor(),
50    unit_type=None,
51    provides=[PipelineCapability.PREPROCESS, PipelineCapability.COREF, PipelineCapability.CHARACTERS],
52)
def annotate_text( text: str, url: str = '/kallimachosPipeline', ignore_types: Set[str] = set(), log_level: str = 'ERROR') -> wuenlp.impl.uima.UIMANLPStructs.UIMADocument: 
17def annotate_text(text: str,
18                  url: str = '/kallimachosPipeline',
19                  ignore_types: Set[str] = set(), log_level: str = "ERROR") -> UIMADocument:
20    myobj = {'text': text}
21    headers = {'Content-type': 'application/json; charset=utf-8'}
22
23    response = query_wuenlp_api(url, headers, myobj)
24
25    with tempfile.NamedTemporaryFile(mode="w") as tf:
26        tf.write(response.text.strip("\x00"))
27        tf.flush()
28
29        # logger.debug(Path(tf.name).read_text())
30        doc = from_mikalli(Path(tf.name), mikalli_typesystem_path=resource_dir / "MiKalliTypesystem.xml",
31                           use_system_coref_id=False,
32                           ignore_types=ignore_types,
33                           log_level=log_level)
34
35    return doc
class PreprocessPipelineProcessor(wuenlp_tools.pipeline.AbstractPipelineProcessor[-InDocT]): 
38class PreprocessPipelineProcessor(PipelinePreprocessor):
39    def __call__(self, doc: UIMADocument | str, unit_type: Type[UIMASpan], **kwargs) -> UIMADocument:  # type: ignore
40        if isinstance(doc, UIMADocument):
41            text = doc.text
42        else:
43            text = doc
44        doc = annotate_text(text)
45        return doc

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...
Preprocessor = PipelineStep('Preprocess', processor=PreprocessPipelineProcessor, provides=['preprocess', 'coref', 'characters'])

Pipeline step Preprocess (PreprocessPipelineProcessor).

Provides: preprocess, coref, characters

  • preprocess: Provides a fully initialized UIMADocument after preprocessing (tokenization, NER, rule-based setup).
  • coref: Provides coreference annotations (UIMAEntity, UIMAEntityReference) needed by character-related steps.
  • characters: Provides UIMACharacter annotations derived from entity/coreference information.