e2edutch.stanza

Complexity

Total Complexity

10

Size/Duplication

Total Lines	135
Duplicated Lines	0 %

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	CorefProcessor._set_up_model()	0	3	1
A	CorefProcessor.__init__()	0	22	1
B	CorefProcessor.process()	0	64	6

import tensorflow.compat.v1 as tf

Missing module docstring

Unable to import 'tensorflow.compat.v1'

Unused tensorflow.compat.v1 imported as tf

import os

The import os seems to be unused.

standard import "import os" should be placed before "import tensorflow.compat.v1 as tf"

import stanza
import logging

standard import "import logging" should be placed before "import tensorflow.compat.v1 as tf"

from pathlib import Path

standard import "from pathlib import Path" should be placed before "import tensorflow.compat.v1 as tf"

from e2edutch import util
from e2edutch import coref_model as cm

Unused coref_model imported from e2edutch as cm

from e2edutch.download import download_data
from e2edutch.predict import Predictor

from stanza.pipeline.processor import Processor, register_processor

third party import "from stanza.pipeline.processor import Processor, register_processor" should be placed before "from e2edutch import util"

from stanza.models.common.doc import Document, Span

Unused Document imported from stanza.models.common.doc

third party import "from stanza.models.common.doc import Document, Span" should be placed before "from e2edutch import util"

# Add a Clusters property to documents as a List of List of Span:
# Clusters is a List of cluster, cluster is a List of Span
def clusterSetter(self, value):

Function name "clusterSetter" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Missing function or method docstring

    if isinstance(value, type([])):
        self._clusters = value

It seems like _clusters was declared protected and should not be accessed from this context.

    else:
        logger.error('Clusters must be a List')


stanza.models.common.doc.Document.add_property('clusters', default='[]', setter=clusterSetter)


logger = logging.getLogger('e2edutch')
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())


@register_processor('coref')
class CorefProcessor(Processor):
    ''' Processor that appends coreference information
    Coreferences are added similarly to Stanza's entities:
    * a Document has an attribute clusters that is a List of coreference clusters;
    * a coreference cluster is a List of Stanza Spans.
    '''
    _requires = set(['tokenize'])
    _provides = set(['coref'])

    def __init__(self, config, pipeline, use_gpu):

The __init__ method of the super-class Processor is not called.

        # Make e2edutch follow Stanza's GPU settings:
        # set the environment value for GPU, so that initialize_from_env picks it up.
        # if use_gpu:
        #    os.environ['GPU'] = ' '.join(tf.config.experimental.list_physical_devices('GPU'))
        # else:
        #    if 'GPU' in os.environ['GPU'] :
        #        os.environ.pop('GPU')

        self.e2econfig = util.initialize_from_env(model_name='final')

        # Override datapath and log_root:
        # store e2edata with the Stanza resources, ie. a 'stanza_resources/nl/coref' directory
        self.e2econfig['datapath'] = Path(config['model_path']).parent
        self.e2econfig['log_root'] = Path(config['model_path']).parent

        # Download data files if not present
        download_data(self.e2econfig)

        # Start and stop a session to cache all models
        predictor = Predictor(config=self.e2econfig)
        predictor.end_session()

    def _set_up_model(self, *args):

The argument args seems to be unused.

This method could be written as a function/class method.

        print('_set_up_model')
        pass

Unnecessary pass statement

    def process(self, doc):

This function exceeds the maximum number of variables (18/15).

        predictor = Predictor(config=self.e2econfig)

        # build the example argument for predict:
        #   example (dict): dict with the following fields:
        #                     sentences ([[str]])
        #                     doc_id (str)
        #                     clusters ([[(int, int)]]) (optional)
        example = {}
        example['sentences'] = []
        example['doc_id'] = 'document_from_stanza'  # TODO check what this should be

TODO and FIXME comments should generally be avoided.

        example['doc_key'] = 'undocumented'  # TODO check what this should be

TODO and FIXME comments should generally be avoided.

        for sentence in doc.sentences:
            s = []

Variable name "s" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

            for word in sentence.words:
                s.append(word.text)
            example['sentences'].append(s)

        predicted_clusters = predictor.predict(example)  # a list of tuples

        # Add the predicted clusters back to the Stanza document

        clusters = []
        for predicted_cluster in predicted_clusters:  # a tuple of entities
            cluster = []
            for predicted_reference in predicted_cluster:  # a tuple of (start, end) word
                start, end = predicted_reference

                # find the sentence_id of the sentence containing this reference
                sentence_id = 0
                sentence = doc.sentences[0]
                sentence_start_word = 0
                sentence_end_word = len(sentence.words) - 1

                while sentence_end_word < start:
                    sentence_start_word = sentence_end_word + 1

                    # move to the next sentence
                    sentence_id += 1
                    sentence = doc.sentences[sentence_id]

                    sentence_end_word = sentence_start_word + len(sentence.words) - 1

                # start counting words from the start of this sentence
                start -= sentence_start_word
                end -= sentence_start_word

                span = Span(  # a list of Tokens
                    tokens=[word.parent for word in sentence.words[start:end + 1]],
                    doc=doc,
                    type='COREF',
                    sent=doc.sentences[sentence_id]
                )
                cluster.append(span)

            clusters.append(cluster)

        doc.clusters = clusters

        predictor.end_session()

        return doc