5190565251

pending completion

Build # 5190565251

Build Type

push

github

Committed by

veghp

Commit Message

Bump to v3.2.11

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

2966 of 3299 relevant lines covered (89.91%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.33

/dnachisel/builtin_specifications/AvoidBlastMatches.py

"""Implementation of AvoidBlastMatches."""

from ..Specification import Specification, SpecEvaluation

# from .VoidSpecification import VoidSpecification
from ..biotools import blast_sequence
from ..Location import Location


class AvoidBlastMatches(Specification):
    """Enforce that the sequence has no BLAST matches with a given database.

    WARNING: try using AvoidMatches instead, it is much better!!

    Uses NCBI Blast+. Only local BLAST is supported/tested as for now

    Parameters
    ----------

    blast_db
      Path to a local BLAST database. These databases can be obtained with
      NCBI's `makeblastdb`. Omit the extension, e.g. `ecoli_db/ecoli_db`.

    word_size
      Word size used by the BLAST algorithm

    perc_identity
      Minimal percentage of identity for BLAST matches. 100 means that only
      perfect matches are considered.

    num_alignments
      Number alignments

    num_threads
      Number of threads/CPU cores to use for the BLAST algorithm.

    min_align_length
      Minimal length that an alignment should have to be considered.
    """

    priority = -2
    best_possible_score = 0
    blasts_paths = {}

    def __init__(
        self,
        blast_db=None,
        sequences=None,
        word_size=4,
        perc_identity=100,
        num_alignments=100000,
        num_threads=3,
        min_align_length=20,
        ungapped=True,
        e_value=1e80,
        culling_limit=1,
        location=None,
    ):
        """Initialize."""
        self.blast_db = blast_db
        self.sequences = sequences
        self.word_size = word_size
        self.perc_identity = perc_identity
        self.num_alignments = num_alignments
        self.num_threads = num_threads
        self.min_align_length = min_align_length
        self.location = Location.from_data(location)
        self.e_value = e_value
        self.ungapped = ungapped
        self.culling_limit = culling_limit

    def initialized_on_problem(self, problem, role=None):
        return self._copy_with_full_span_if_no_location(problem)

    def evaluate(self, problem):
        """Score as (-total number of blast identities in matches)."""
        location = self.location
        if location is None:
            location = Location(0, len(problem.sequence))
        sequence = location.extract_sequence(problem.sequence)

        blast_record = blast_sequence(
            sequence,
            blast_db=self.blast_db,
            subject_sequences=self.sequences,
            word_size=self.word_size,
            perc_identity=self.perc_identity,
            num_alignments=self.num_alignments,
            num_threads=self.num_threads,
            ungapped=self.ungapped,
            e_value=self.e_value,
            culling_limit=self.culling_limit,
            task="megablast"
        )

        if isinstance(blast_record, list):
            alignments = [
                alignment
                for rec in blast_record
                for alignment in rec.alignments
            ]
        else:
            alignments = blast_record.alignments

        query_hits = [
            (
                min(hit.query_start, hit.query_end) + location.start - 1,
                max(hit.query_start, hit.query_end) + location.start,
                1 - 2 * (hit.query_start > hit.query_end),
                hit.identities,
            )
            for alignment in alignments
            for hit in alignment.hsps
        ]

        locations = sorted(
            [
                (start, end, ids)
                for (start, end, strand, ids) in query_hits
                if (end - start) >= self.min_align_length
            ]
        )

        score = -sum([ids for start, end, ids in locations])
        locations = [Location(start, end) for start, end, ids in locations]

        if locations == []:
            return SpecEvaluation(
                self, problem, score=1, message="Passed: no BLAST match found"
            )

        return SpecEvaluation(
            self,
            problem,
            score=score,
            locations=locations,
            message="Failed - %s matches at %s" % (len(locations), locations),
        )

    def localized(self, location, problem=None, with_righthand=True):
        """Localize the evaluation."""
        new_location = self.location.overlap_region(location)
        if new_location is None:
            return None
        new_location = location.extended(
            self.min_align_length - 1, right=with_righthand
        )
        return self.copy_with_changes(location=new_location)

    def feature_label_parameters(self):
        return [self.blast_db]

1	"""Implementation of AvoidBlastMatches."""
2
3	from ..Specification import Specification, SpecEvaluation	1✔
4
5	# from .VoidSpecification import VoidSpecification
6	from ..biotools import blast_sequence	1✔
7	from ..Location import Location	1✔
8
9
10	class AvoidBlastMatches(Specification):	1✔
11	"""Enforce that the sequence has no BLAST matches with a given database.
12
13	WARNING: try using AvoidMatches instead, it is much better!!
14
15	Uses NCBI Blast+. Only local BLAST is supported/tested as for now
16
17	Parameters
18	----------
19
20	blast_db
21	Path to a local BLAST database. These databases can be obtained with
22	NCBI's `makeblastdb`. Omit the extension, e.g. `ecoli_db/ecoli_db`.
23
24	word_size
25	Word size used by the BLAST algorithm
26
27	perc_identity
28	Minimal percentage of identity for BLAST matches. 100 means that only
29	perfect matches are considered.
30
31	num_alignments
32	Number alignments
33
34	num_threads
35	Number of threads/CPU cores to use for the BLAST algorithm.
36
37	min_align_length
38	Minimal length that an alignment should have to be considered.
39	"""
40
41	priority = -2	1✔
42	best_possible_score = 0	1✔
43	blasts_paths = {}	1✔
44
45	def __init__(	1✔
46	self,
47	blast_db=None,
48	sequences=None,
49	word_size=4,
50	perc_identity=100,
51	num_alignments=100000,
52	num_threads=3,
53	min_align_length=20,
54	ungapped=True,
55	e_value=1e80,
56	culling_limit=1,
57	location=None,
58	):
59	"""Initialize."""
60	self.blast_db = blast_db	1✔
61	self.sequences = sequences	1✔
62	self.word_size = word_size	1✔
63	self.perc_identity = perc_identity	1✔
64	self.num_alignments = num_alignments	1✔
65	self.num_threads = num_threads	1✔
66	self.min_align_length = min_align_length	1✔
67	self.location = Location.from_data(location)	1✔
68	self.e_value = e_value	1✔
69	self.ungapped = ungapped	1✔
70	self.culling_limit = culling_limit	1✔
71
72	def initialized_on_problem(self, problem, role=None):	1✔
73	return self._copy_with_full_span_if_no_location(problem)	1✔
74
75	def evaluate(self, problem):	1✔
76	"""Score as (-total number of blast identities in matches)."""
77	location = self.location	1✔
78	if location is None:	1✔
79	location = Location(0, len(problem.sequence))	1✔
80	sequence = location.extract_sequence(problem.sequence)	1✔
81
82	blast_record = blast_sequence(	1✔
83	sequence,
84	blast_db=self.blast_db,
85	subject_sequences=self.sequences,
86	word_size=self.word_size,
87	perc_identity=self.perc_identity,
88	num_alignments=self.num_alignments,
89	num_threads=self.num_threads,
90	ungapped=self.ungapped,
91	e_value=self.e_value,
92	culling_limit=self.culling_limit,
93	task="megablast"
94	)
95
96	if isinstance(blast_record, list):	1✔
97	alignments = [	×
98	alignment
99	for rec in blast_record
100	for alignment in rec.alignments
101	]
102	else:
103	alignments = blast_record.alignments	1✔
104
105	query_hits = [	1✔
106	(
107	min(hit.query_start, hit.query_end) + location.start - 1,
108	max(hit.query_start, hit.query_end) + location.start,
109	1 - 2 * (hit.query_start > hit.query_end),
110	hit.identities,
111	)
112	for alignment in alignments
113	for hit in alignment.hsps
114	]
115
116	locations = sorted(	1✔
117	[
118	(start, end, ids)
119	for (start, end, strand, ids) in query_hits
120	if (end - start) >= self.min_align_length
121	]
122	)
123
124	score = -sum([ids for start, end, ids in locations])	1✔
125	locations = [Location(start, end) for start, end, ids in locations]	1✔
126
127	if locations == []:	1✔
128	return SpecEvaluation(	1✔
129	self, problem, score=1, message="Passed: no BLAST match found"
130	)
131
132	return SpecEvaluation(	1✔
133	self,
134	problem,
135	score=score,
136	locations=locations,
137	message="Failed - %s matches at %s" % (len(locations), locations),
138	)
139
140	def localized(self, location, problem=None, with_righthand=True):	1✔
141	"""Localize the evaluation."""
142	new_location = self.location.overlap_region(location)	1✔
143	if new_location is None:	1✔
144	return None	×
145	new_location = location.extended(	1✔
146	self.min_align_length - 1, right=with_righthand
147	)
148	return self.copy_with_changes(location=new_location)	1✔
149
150	def feature_label_parameters(self):	1✔
151	return [self.blast_db]	×

Edinburgh-Genome-Foundry / DnaChisel / 5190565251

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous