wuenlp_tools.models.sentiment.HarryMotionsAPI 

 1from __future__ import annotations
 2
 3import tempfile
 4from enum import Enum
 5from pathlib import Path
 6from typing import Callable, Type
 7
 8import requests
 9from loguru import logger
10from wuenlp.impl.UIMANLPStructs import UIMADocument
11from wuenlp.impl.uima import UIMASpan, UIMAInteraction, UIMACharacter
12
13from wuenlp_tools.pipeline import PipelineStep, PipelineProcessor
14from wuenlp_tools.utils.api import query_wuenlp_api
15
16
17class HarryMotionsModelType(str, Enum):
18    BERT = "bert"
19    LLAMA = "llama"
20
21
22def harrymotions_api(doc: UIMADocument, model: HarryMotionsModelType = HarryMotionsModelType.LLAMA,
23                     overwrite: bool = False,
24                     base_url: str = "/harrymotions/"):
25    if doc.interactions and not overwrite:
26        logger.info(f"Document {doc} already has interactions. Skipping")
27        return doc
28
29    document_full_extension = ".xmi.zip"
30    with tempfile.NamedTemporaryFile(suffix=document_full_extension, delete=False) as temp_file:
31        doc.serialize(temp_file.name)
32
33    if not base_url.endswith("/"):
34        base_url += "/"
35    with open(temp_file.name, "rb") as file:
36        files = {"file": (file.name, file)}
37        logger.info(f"Sending file to {model.value} model")
38        response = query_wuenlp_api(url=f"{base_url}{model.value}", files=files)
39
40    with tempfile.NamedTemporaryFile(suffix=document_full_extension, delete=False) as temp_out_file:
41        with open(temp_out_file.name, "wb") as f:
42            f.write(response.content)
43
44    return UIMADocument.from_xmi(temp_out_file.name)
45
46
47class LlamaHMProcessor(PipelineProcessor):
48    def __call__(self, doc: UIMADocument, unit_type=None, overwrite: bool = False, **kwargs):
49        return harrymotions_api(doc, model=HarryMotionsModelType.LLAMA, overwrite=overwrite)
50
51
52LlamaHMAnnotator = PipelineStep("Llama HarryMotions", LlamaHMProcessor(), unit_type=None,
53                                modified_types=[UIMAInteraction, UIMACharacter], needs_manual_merge=True)
class HarryMotionsModelType(builtins.str, enum.Enum): 
18class HarryMotionsModelType(str, Enum):
19    BERT = "bert"
20    LLAMA = "llama"

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

BERT = <HarryMotionsModelType.BERT: 'bert'>
LLAMA = <HarryMotionsModelType.LLAMA: 'llama'>
def harrymotions_api( doc: wuenlp.impl.uima.UIMANLPStructs.UIMADocument, model: HarryMotionsModelType = <HarryMotionsModelType.LLAMA: 'llama'>, overwrite: bool = False, base_url: str = '/harrymotions/'): 
23def harrymotions_api(doc: UIMADocument, model: HarryMotionsModelType = HarryMotionsModelType.LLAMA,
24                     overwrite: bool = False,
25                     base_url: str = "/harrymotions/"):
26    if doc.interactions and not overwrite:
27        logger.info(f"Document {doc} already has interactions. Skipping")
28        return doc
29
30    document_full_extension = ".xmi.zip"
31    with tempfile.NamedTemporaryFile(suffix=document_full_extension, delete=False) as temp_file:
32        doc.serialize(temp_file.name)
33
34    if not base_url.endswith("/"):
35        base_url += "/"
36    with open(temp_file.name, "rb") as file:
37        files = {"file": (file.name, file)}
38        logger.info(f"Sending file to {model.value} model")
39        response = query_wuenlp_api(url=f"{base_url}{model.value}", files=files)
40
41    with tempfile.NamedTemporaryFile(suffix=document_full_extension, delete=False) as temp_out_file:
42        with open(temp_out_file.name, "wb") as f:
43            f.write(response.content)
44
45    return UIMADocument.from_xmi(temp_out_file.name)
class LlamaHMProcessor(wuenlp_tools.pipeline.AbstractPipelineProcessor[wuenlp.impl.uima.UIMANLPStructs.UIMADocument]): 
48class LlamaHMProcessor(PipelineProcessor):
49    def __call__(self, doc: UIMADocument, unit_type=None, overwrite: bool = False, **kwargs):
50        return harrymotions_api(doc, model=HarryMotionsModelType.LLAMA, overwrite=overwrite)

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:
        ...
LlamaHMAnnotator = PipelineStep('Llama HarryMotions', processor=LlamaHMProcessor)

Pipeline step Llama HarryMotions (LlamaHMProcessor).

modifies UIMAInteraction, UIMACharacter.