<commit_before>"""Read an Association File and store the data in a Python object."""

import sys
import timeit
import datetime
import collections as cx
from goatools.evidence_codes import EvidenceCodes
from goatools.anno.opts import AnnoOptions
from goatools.godag.consts import NAMESPACE2NS
from goatools.gosubdag.go_tasks import get_go2parents_go2obj

__copyright__ = "Copyright (C) 2016-2019, <NAME>, <NAME>. All rights reserved."
__author__ = "<NAME>"


# pylint: disable=too-many-instance-attributes
class AnnoReaderBase(object):
    """Reads a Gene Association File. Returns a Python object."""
    # pylint: disable=broad-except,line-too-long,too-many-instance-attributes

    tic = timeit.default_timer()

    # Expected values for a Qualifier
    exp_qualifiers = set([
        # Seen in both GAF and gene2go
        'not', 'contributes_to', 'colocalizes_with',
    ])

    valid_formats = {'gpad', 'gaf', 'gene2go', 'id2gos'}

    exp_nss = set(['BP', 'MF', 'CC'])

    def __init__(self, name, filename=None, **kws):
        # kws: allow_missing_symbol
        self.name = name  # name is one of valid_formats
        self.filename = filename
        self.godag = kws.get('godag')
        self.namespaces = kws.get('namespaces')
        self.evobj = EvidenceCodes()
        # Read anotation file, store namedtuples:
        #     Gene2GoReader(filename=None, taxids=None):
        #     GafReader(filename=None, hdr_only=False, prt=sys.stdout, allow_missing_symbol=False):
        #     GpadReader(filename=None, hdr_only=False):
        self.hdr = None
        self.datobj = None
        # pylint: disable=no-member
        self.associations = self._init_associations(filename, **kws)
        # assert self.associations, 'NO ANNOTATIONS FOUND: {ANNO}'.format(ANNO=filename)
        assert self.namespaces is None or isinstance(self.namespaces, set)

    def get_desc(self):
        """Get description"""
        return '{NAME} {NSs} {GODAG}'.format(
            NAME=self.name,
            NSs='' if self.namespaces is None else ','.join(self.namespaces),
            GODAG='' if self.godag is None else 'godag')

    # pylint: disable=unused-argument
    def get_associations(self, taxid=None):
        """Get associations"""
        # taxid is for NCBI's gene2gos
        return self.associations

    def prt_summary_anno2ev(self, prt=sys.stdout):
        """Print annotation/evidence code summary."""
        self.evobj.prt_summary_anno2ev(self.associations, prt)

    def get_name(self):
        """Return type of annotation"""
        return self.name

    # pylint: disable=no-self-use
    def get_taxid(self):
        """Return taxid, if one was provided, otherwise return -1"""
        return -1

    def get_ns2assc(self, **kws):
        """Return given associations into 3 (BP, MF, CC) dicts, id2gos"""
        return {ns:self._get_id2gos(nts, **kws) for ns, nts in self.get_ns2ntsanno(kws.get('taxid')).items()}

    # pylint: disable=unused-argument
    def get_ns2ntsanno(self, taxid=None):
        """Split list of annotations into 3 lists: BP, MF, CC"""
        return self._get_ns2ntsanno(self.associations)

    def _get_ns2ntsanno(self, annotations):
        """Split list of annotations into 3 lists: BP, MF, CC"""
        if self.name in {'gpad', 'id2gos'}:
            assert self.godag is not None, "{T}: LOAD godag TO USE {C}::ns2ntsanno".format(
                C=self.__class__.__name__, T=self.name)
        ns2nts = cx.defaultdict(list)
        for nta in annotations:
            ns2nts[nta.NS].append(nta)
        return {ns:ns2nts[ns] for ns in self.exp_nss.intersection(ns2nts)}

    def get_id2gos_nss(self, **kws):
        """Return all associations in a dict, id2gos, regardless of namespace"""
        return self._get_id2gos(self.associations, **kws)

    def get_id2gos(self, namespace=None, **kws):
        """Return associations from specified namespace in a dict, id2gos"""
        # pylint: disable=superfluous-parens
        if self.has_ns():  # Anno namedtuple has NS field
            nspc, assoc = self._get_1ns_assn(namespace)
            id2gos = self._get_id2gos(assoc, **kws)
            print('{N} IDs in loaded association branch, {NS}'.format(N=len(id2gos), NS=nspc))
            return id2gos
        if namespace is not None:
            print('**ERROR get_id2gos: GODAG NOT LOADED. IGNORING namespace({NS})'.format(NS=namespace))
        id2gos = self._get_id2gos(self.associations, **kws)
        print('{N} IDs in all associations'.format(N=len(id2gos)))
        return id2gos

    def _get_1ns_assn(self, namespace_usr):
        """Get one namespace, given a user-provided namespace or a default"""
        # If all namespaces were loaded
        if self.namespaces is None:
            # Return user-specified namespace, if provided. Otherwise BP
            nspc = 'BP' if namespace_usr is None else namespace_usr
            # Return one namespace
            if nspc in set(NAMESPACE2NS.values()):
                return nspc, [nt for nt in self.associations if nt.NS == nspc]
            # Return all namespaces
            return nspc, self.associations
        # If one namespace was loaded, use that regardless of what user specfies
        if len(self.namespaces) == 1:
            nspc = next(iter(self.namespaces))
            if namespace_usr is not None and nspc != namespace_usr:<commit_msg>Improve error message if GODag not provided for gpad<commit_after>"""Read an Association File and store the data in a Python object."""

import sys
import timeit
import datetime
import collections as cx
from goatools.evidence_codes import EvidenceCodes
from goatools.anno.opts import AnnoOptions
from goatools.godag.consts import NAMESPACE2NS
from goatools.gosubdag.go_tasks import get_go2parents_go2obj

__copyright__ = "Copyright (C) 2016-2019, <NAME>, <NAME>. All rights reserved."
__author__ = "<NAME>"


# pylint: disable=too-many-instance-attributes
class AnnoReaderBase(object):
    """Reads a Gene Association File. Returns a Python object."""
    # pylint: disable=broad-except,line-too-long,too-many-instance-attributes

    tic = timeit.default_timer()

    # Expected values for a Qualifier
    exp_qualifiers = set([
        # Seen in both GAF and gene2go
        'not', 'contributes_to', 'colocalizes_with',
    ])

    valid_formats = {'gpad', 'gaf', 'gene2go', 'id2gos'}

    exp_nss = set(['BP', 'MF', 'CC'])

    def __init__(self, name, filename=None, **kws):
        # kws: allow_missing_symbol
        self.name = name  # name is one of valid_formats
        self.filename = filename
        self.godag = kws.get('godag')
        self.namespaces = kws.get('namespaces')
        self.evobj = EvidenceCodes()
        # Read anotation file, store namedtuples:
        #     Gene2GoReader(filename=None, taxids=None):
        #     GafReader(filename=None, hdr_only=False, prt=sys.stdout, allow_missing_symbol=False):
        #     GpadReader(filename=None, hdr_only=False):
        self.hdr = None
        self.datobj = None
        # pylint: disable=no-member
        self.associations = self._init_associations(filename, **kws)
        # assert self.associations, 'NO ANNOTATIONS FOUND: {ANNO}'.format(ANNO=filename)
        assert self.namespaces is None or isinstance(self.namespaces, set)

    def get_desc(self):
        """Get description"""
        return '{NAME} {NSs} {GODAG}'.format(
            NAME=self.name,
            NSs='' if self.namespaces is None else ','.join(self.namespaces),
            GODAG='' if self.godag is None else 'godag')

    # pylint: disable=unused-argument
    def get_associations(self, taxid=None):
        """Get associations"""
        # taxid is for NCBI's gene2gos
        return self.associations

    def prt_summary_anno2ev(self, prt=sys.stdout):
        """Print annotation/evidence code summary."""
        self.evobj.prt_summary_anno2ev(self.associations, prt)

    def get_name(self):
        """Return type of annotation"""
        return self.name

    # pylint: disable=no-self-use
    def get_taxid(self):
        """Return taxid, if one was provided, otherwise return -1"""
        return -1

    def get_ns2assc(self, **kws):
        """Return given associations into 3 (BP, MF, CC) dicts, id2gos"""
        return {ns:self._get_id2gos(nts, **kws) for ns, nts in self.get_ns2ntsanno(kws.get('taxid')).items()}

    # pylint: disable=unused-argument
    def get_ns2ntsanno(self, taxid=None):
        """Split list of annotations into 3 lists: BP, MF, CC"""
        return self._get_ns2ntsanno(self.associations)

    def _get_ns2ntsanno(self, annotations):
        """Split list of annotations into 3 lists: BP, MF, CC"""
        if self.name in {'gpad', 'id2gos'}:
            assert self.godag is not None, "{T}: LOAD godag TO USE {C}::ns2ntsanno".format(
                C=self.__class__.__name__, T=self.name)
        ns2nts = cx.defaultdict(list)
        for nta in annotations:
            ns2nts[nta.NS].append(nta)
        return {ns:ns2nts[ns] for ns in self.exp_nss.intersection(ns2nts)}

    def get_id2gos_nss(self, **kws):
        """Return all associations in a dict, id2gos, regardless of namespace"""
        return self._get_id2gos(self.associations, **kws)

    def get_id2gos(self, namespace=None, **kws):
        """Return associations from specified namespace in a dict, id2gos"""
        # pylint: disable=superfluous-parens
        if self.has_ns():  # Anno namedtuple has NS field
            nspc, assoc = self._get_1ns_assn(namespace)
            id2gos = self._get_id2gos(assoc, **kws)
            print('{N} IDs in loaded association branch, {NS}'.format(N=len(id2gos), NS=nspc))
            return id2gos
        if namespace is not None:
            print('**ERROR {CLS}(..., godag=None).get_id2gos: GODAG is None. IGNORING namespace({NS})'.format(
                NS=namespace, CLS=type(self).__name__))
        id2gos = self._get_id2gos(self.associations, **kws)
        print('{N} IDs in all associations'.format(N=len(id2gos)))
        return id2gos

    def _get_1ns_assn(self, namespace_usr):
        """Get one namespace, given a user-provided namespace or a default"""
        # If all namespaces were loaded
        if self.namespaces is None:
            # Return user-specified namespace, if provided. Otherwise BP
            nspc = 'BP' if namespace_usr is None else namespace_usr
            # Return one namespace
            if nspc in set(NAMESPACE2NS.values()):
                return nspc, [nt for nt in self.associations if nt.NS == nspc]
            # Return all namespaces
            return nspc, self.associations
        # If one namespace was loaded, use that regardless of what user specfies
        if len(self.namespaces) == 1:
            nspc = next(iter(self.namespaces))
            if namespace_usr is not None and nspc != namespace_usr:
{'fraction_non_alphanumeric': 0.11250345526582511, 'fraction_numerical': 0.00995116557633834, 'mean_word_length': 2.844845908607864, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2341264', 'n_tokens_mistral': 3504, 'n_tokens_neox': 3376, 'n_words': 930}