annif.corpus.subject

Complexity

Total Complexity

48

Size/Duplication

Total Lines	207
Duplicated Lines	93.24 %

Importance

Changes

0

25 Methods

Rating	Name	Duplication	Size	Complexity
A	SubjectFileTSV.__init__()	2	2	1
A	SubjectFileTSV._parse_line()	6	6	3
A	SubjectIndex.contains_uri()	2	2	1
A	SubjectIndex.__len__()	2	2	1
A	SubjectIndex.__init__()	6	6	1
A	SubjectFileTSV.languages()	4	4	1
A	SubjectIndex.save()	11	11	5
A	SubjectIndex.deprecated_ids()	5	5	1
A	SubjectIndex.by_uri()	9	9	3
A	SubjectIndex.by_label()	7	7	2
A	SubjectIndex.__getitem__()	3	3	1
A	SubjectFileTSV.subjects()	4	4	3
A	SubjectIndex.uris_to_labels()	7	7	1
A	SubjectFileTSV.save_skos()	4	4	1
B	SubjectSet.as_vector()	22	22	7
A	SubjectIndex.labels_to_uris()	7	7	1
A	SubjectSet.from_string()	6	6	2
A	SubjectIndex.load()	8	8	1
A	SubjectIndex._append()	6	6	1
A	SubjectSet.has_uris()	3	3	1
A	SubjectSet.__init__()	7	7	1
A	SubjectSet._parse_line()	11	11	5
A	SubjectIndex.load_subjects()	7	7	2
A	SubjectIndex.append()	3	3	1
A	SubjectIndex.active()	9	9	1

"""Classes for supporting subject corpora expressed as directories or files"""

import annif.util
import numpy as np
from annif import logger
from .types import Subject
from .skos import serialize_subjects_to_skos


class SubjectFileTSV:

This code seems to be duplicated in your project.

    """A subject vocabulary stored in a TSV file."""

    def __init__(self, path):
        self.path = path

    def _parse_line(self, line):
        vals = line.strip().split('\t', 2)
        clean_uri = annif.util.cleanup_uri(vals[0])
        label = vals[1] if len(vals) >= 2 else None
        notation = vals[2] if len(vals) >= 3 else None
        yield Subject(uri=clean_uri, label=label, notation=notation, text=None)

    @property
    def languages(self):
        # we don't have information about the language(s) of labels
        return None

    def subjects(self, language):
        with open(self.path, encoding='utf-8-sig') as subjfile:
            for line in subjfile:
                yield from self._parse_line(line)

    def save_skos(self, path, language):
        """Save the contents of the subject vocabulary into a SKOS/Turtle
        file with the given path name."""
        serialize_subjects_to_skos(self.subjects(language), language, path)


class SubjectIndex:

This code seems to be duplicated in your project.

    """An index that remembers the associations between integers subject IDs
    and their URIs and labels."""

    def __init__(self):
        self._uris = []
        self._labels = []
        self._notations = []
        self._uri_idx = {}
        self._label_idx = {}

    def load_subjects(self, corpus, language):
        """Initialize the subject index from a subject corpus using labels
        in the given language."""

        for subject_id, subject in enumerate(corpus.subjects(language)):
            self._append(subject_id, subject.uri, subject.label,
                         subject.notation)

    def __len__(self):
        return len(self._uris)

    def __getitem__(self, subject_id):
        return (self._uris[subject_id], self._labels[subject_id],
                self._notations[subject_id])

    def _append(self, subject_id, uri, label, notation):
        self._uris.append(uri)
        self._labels.append(label)
        self._notations.append(notation)
        self._uri_idx[uri] = subject_id
        self._label_idx[label] = subject_id

    def append(self, uri, label, notation):
        subject_id = len(self._uris)
        self._append(subject_id, uri, label, notation)

    def contains_uri(self, uri):
        return uri in self._uri_idx

    def by_uri(self, uri, warnings=True):
        """return the subject index of a subject by its URI, or None if not found.
        If warnings=True, log a warning message if the URI cannot be found."""
        try:
            return self._uri_idx[uri]
        except KeyError:
            if warnings:
                logger.warning('Unknown subject URI <%s>', uri)
            return None

    def by_label(self, label):
        """return the subject index of a subject by its label"""
        try:
            return self._label_idx[label]
        except KeyError:
            logger.warning('Unknown subject label "%s"', label)
            return None

    def uris_to_labels(self, uris):
        """return a list of labels corresponding to the given URIs; unknown
        URIs are ignored"""

        return [self[subject_id][1]
                for subject_id in (self.by_uri(uri) for uri in uris)
                if subject_id is not None]

    def labels_to_uris(self, labels):
        """return a list of URIs corresponding to the given labels; unknown
        labels are ignored"""

        return [self[subject_id][0]
                for subject_id in (self.by_label(label) for label in labels)
                if subject_id is not None]

    def deprecated_ids(self):
        """return indices of deprecated subjects"""

        return [subject_id for subject_id, label in enumerate(self._labels)
                if label is None]

    @property
    def active(self):
        """return a list of (subject_id, uri, label, notation) tuples of all
        subjects that are not deprecated"""

        return [(subj_id, uri, label, notation)
                for subj_id, (uri, label, notation)
                in enumerate(zip(self._uris, self._labels, self._notations))
                if label is not None]

    def save(self, path):
        """Save this subject index into a file."""

        with open(path, 'w', encoding='utf-8') as subjfile:
            for uri, label, notation in self:
                line = "<{}>".format(uri)
                if label is not None:
                    line += ('\t' + label)
                    if notation is not None:
                        line += ('\t' + notation)
                print(line, file=subjfile)

    @classmethod
    def load(cls, path):
        """Load a subject index from a TSV file and return it."""

        corpus = SubjectFileTSV(path)
        subject_index = cls()
        subject_index.load_subjects(corpus, None)
        return subject_index


class SubjectSet:

This code seems to be duplicated in your project.

    """Represents a set of subjects for a document."""

    def __init__(self, subj_data=None):
        """Create a SubjectSet and optionally initialize it from a tuple
        (URIs, labels)"""

        uris, labels = subj_data or ([], [])
        self.subject_uris = set(uris)
        self.subject_labels = set(labels)

    @classmethod
    def from_string(cls, subj_data):
        sset = cls()
        for line in subj_data.splitlines():
            sset._parse_line(line)
        return sset

    def _parse_line(self, line):
        vals = line.split("\t")
        for val in vals:
            val = val.strip()
            if val == '':
                continue
            if val.startswith('<') and val.endswith('>'):  # URI
                self.subject_uris.add(val[1:-1])
                continue
            self.subject_labels.add(val)
            return

    def has_uris(self):
        """returns True if the URIs for all subjects are known"""
        return len(self.subject_uris) >= len(self.subject_labels)

    def as_vector(self, subject_index, destination=None, warnings=True):
        """Return the hits as a one-dimensional NumPy array in sklearn
           multilabel indicator format, using a subject index as the source
           of subjects. Use destination array if given (not None), otherwise
           create and return a new one. If warnings=True, log warnings for
           unknown URIs."""

        if destination is None:
            destination = np.zeros(len(subject_index), dtype=bool)

        if self.has_uris():
            for uri in self.subject_uris:
                subject_id = subject_index.by_uri(
                    uri, warnings=warnings)
                if subject_id is not None:
                    destination[subject_id] = True
        else:
            for label in self.subject_labels:
                subject_id = subject_index.by_label(label)
                if subject_id is not None:
                    destination[subject_id] = True
        return destination