Compare commits
No commits in common. "2a80c2aea239be31f646360c558170ff391cc499" and "ecd488b8514d14eeea69dc5d8669797e2a2ce662" have entirely different histories.
2a80c2aea2
...
ecd488b851
36
Jenkinsfile
vendored
36
Jenkinsfile
vendored
@ -1,36 +0,0 @@
|
|||||||
pipeline {
|
|
||||||
agent any
|
|
||||||
stages {
|
|
||||||
stage("clean") {
|
|
||||||
steps {
|
|
||||||
sh 'rm -rf ./dist/*'
|
|
||||||
}
|
|
||||||
}
|
|
||||||
stage("install") {
|
|
||||||
steps {
|
|
||||||
sh 'mamba env update --file environment.yml'
|
|
||||||
sh 'echo "mamba activate bmlsa" >> ~/.bashrc'
|
|
||||||
}
|
|
||||||
}
|
|
||||||
stage("build") {
|
|
||||||
steps {
|
|
||||||
sh "python -m build"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
stage("test") {
|
|
||||||
steps {
|
|
||||||
sh "pip install dist/*.whl"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
stage("publish") {
|
|
||||||
when {
|
|
||||||
branch '**/master'
|
|
||||||
}
|
|
||||||
steps {
|
|
||||||
withCredentials([usernamePassword(credentialsId: 'rs-git-package-registry-ydeng', passwordVariable: 'PASS', usernameVariable: 'USER')]) {
|
|
||||||
sh "python -m twine upload --repository-url https://git.reslate.systems/api/packages/${USER}/pypi -u ${USER} -p ${PASS} --non-interactive --disable-progress-bar --verbose dist/*"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
@ -1,43 +0,0 @@
|
|||||||
from Bio.Align import PairwiseAligner, substitution_matrices
|
|
||||||
from exceptions import UnexpectedAlignmentResult
|
|
||||||
|
|
||||||
from datatypes import AlignedSequence
|
|
||||||
|
|
||||||
|
|
||||||
def protein_align_many_to_one_ssw(sequence: str, queries: dict[str, AlignedSequence]):
|
|
||||||
annotation_pairs = {}
|
|
||||||
aligner = PairwiseAligner()
|
|
||||||
aligner.mode = "local"
|
|
||||||
aligner.substitution_matrix = substitution_matrices.load("BLOSUM62")
|
|
||||||
aligner.extend_gap_score = -1
|
|
||||||
aligner.open_gap_score = -11
|
|
||||||
for id, query in queries.items():
|
|
||||||
try:
|
|
||||||
alignments = aligner.align(sequence, query.sequence)
|
|
||||||
except ValueError:
|
|
||||||
continue
|
|
||||||
if len(alignments) > 1:
|
|
||||||
raise UnexpectedAlignmentResult(
|
|
||||||
"More than one alignment resulted from a single query."
|
|
||||||
)
|
|
||||||
for alignment in alignments:
|
|
||||||
score, query_aligned = (alignment.score, alignment.aligned[0][0])
|
|
||||||
aligned_start, aligned_end = query_aligned
|
|
||||||
annotation_pairs[id] = AlignedSequence(
|
|
||||||
id,
|
|
||||||
query.sequence,
|
|
||||||
query.name,
|
|
||||||
query.description,
|
|
||||||
query.start,
|
|
||||||
query.end,
|
|
||||||
query.score,
|
|
||||||
), AlignedSequence(
|
|
||||||
id,
|
|
||||||
alignment.query,
|
|
||||||
query.name,
|
|
||||||
query.description,
|
|
||||||
aligned_start,
|
|
||||||
aligned_end,
|
|
||||||
score,
|
|
||||||
)
|
|
||||||
return annotation_pairs
|
|
94
bmlsa/cli.py
94
bmlsa/cli.py
@ -1,94 +0,0 @@
|
|||||||
import os
|
|
||||||
import argparse
|
|
||||||
from Bio import SeqIO
|
|
||||||
from aligner import protein_align_many_to_one_ssw
|
|
||||||
|
|
||||||
from persistence import read_annotations_from_csv, save_alignments_to_csv
|
|
||||||
|
|
||||||
|
|
||||||
def main():
|
|
||||||
argparser = argparse.ArgumentParser("blmsa")
|
|
||||||
argparser.add_argument(
|
|
||||||
"annotations",
|
|
||||||
type=str,
|
|
||||||
help=(
|
|
||||||
"Path to CSV containing the sequences to align as well as the "
|
|
||||||
"annotations for the respective sequences"
|
|
||||||
),
|
|
||||||
metavar="a",
|
|
||||||
)
|
|
||||||
argparser.add_argument(
|
|
||||||
"sequence",
|
|
||||||
type=str,
|
|
||||||
help=(
|
|
||||||
"Path to the sequence to annotate in FASTA format. "
|
|
||||||
"If multiple sequences are present, annotations will be run on each"
|
|
||||||
),
|
|
||||||
metavar="s",
|
|
||||||
)
|
|
||||||
argparser.add_argument(
|
|
||||||
"output", type=str, help="Path to output location", metavar="o"
|
|
||||||
)
|
|
||||||
argparser.add_argument(
|
|
||||||
"-I", "--id-header", type=str, help="The header for the ID of the annotation"
|
|
||||||
)
|
|
||||||
argparser.add_argument(
|
|
||||||
"-N",
|
|
||||||
"--name-header",
|
|
||||||
type=str,
|
|
||||||
help="The header for the name of the annotation",
|
|
||||||
required=False,
|
|
||||||
)
|
|
||||||
argparser.add_argument(
|
|
||||||
"-D",
|
|
||||||
"--desc-header",
|
|
||||||
type=str,
|
|
||||||
help="The header for the description of the annotation",
|
|
||||||
required=False,
|
|
||||||
)
|
|
||||||
argparser.add_argument(
|
|
||||||
"-T",
|
|
||||||
"--start-header",
|
|
||||||
type=str,
|
|
||||||
help="The header for the start of the annotation",
|
|
||||||
required=False,
|
|
||||||
)
|
|
||||||
argparser.add_argument(
|
|
||||||
"-E",
|
|
||||||
"--end-header",
|
|
||||||
type=str,
|
|
||||||
help="The header for the end of the annotation",
|
|
||||||
required=False,
|
|
||||||
)
|
|
||||||
argparser.add_argument(
|
|
||||||
"-S",
|
|
||||||
"--seq-header",
|
|
||||||
type=str,
|
|
||||||
help="The header for the sequence of the annotation",
|
|
||||||
)
|
|
||||||
args = argparser.parse_args()
|
|
||||||
given_annotations = read_annotations_from_csv(
|
|
||||||
args.annotations,
|
|
||||||
args.id_header,
|
|
||||||
args.name_header,
|
|
||||||
args.desc_header,
|
|
||||||
args.start_header,
|
|
||||||
args.end_header,
|
|
||||||
args.seq_header,
|
|
||||||
)
|
|
||||||
with open(args.sequence, "r") as sequence_fd:
|
|
||||||
for sequence in SeqIO.parse(sequence_fd, "fasta"):
|
|
||||||
aligned_annotations = protein_align_many_to_one_ssw(
|
|
||||||
str(sequence.seq), given_annotations
|
|
||||||
)
|
|
||||||
save_alignments_to_csv(
|
|
||||||
aligned_annotations,
|
|
||||||
os.path.join(
|
|
||||||
args.output,
|
|
||||||
sequence.id.replace("|", "+").replace(".", "_") + ".csv",
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
main()
|
|
@ -1,2 +0,0 @@
|
|||||||
class UnexpectedAlignmentResult(Exception):
|
|
||||||
pass
|
|
@ -1,70 +0,0 @@
|
|||||||
import csv
|
|
||||||
from datatypes import AlignedSequence
|
|
||||||
|
|
||||||
|
|
||||||
def read_annotations_from_csv(
|
|
||||||
csv_path: str,
|
|
||||||
id_header: str,
|
|
||||||
name_header: str,
|
|
||||||
desc_header: str,
|
|
||||||
start_header: str,
|
|
||||||
end_header: str,
|
|
||||||
sequence_header: str,
|
|
||||||
):
|
|
||||||
annotations = {}
|
|
||||||
with open(csv_path, "r") as csv_fd:
|
|
||||||
reader = csv.reader(csv_fd)
|
|
||||||
id_ind = None
|
|
||||||
name_ind = None
|
|
||||||
desc_ind = None
|
|
||||||
start_ind = None
|
|
||||||
end_ind = None
|
|
||||||
sequence_ind = None
|
|
||||||
headers_parsed = False
|
|
||||||
for row in reader:
|
|
||||||
if not headers_parsed:
|
|
||||||
id_ind = row.index(id_header)
|
|
||||||
name_ind = row.index(name_header) if name_header else None
|
|
||||||
desc_ind = row.index(desc_header) if desc_header else None
|
|
||||||
start_ind = row.index(start_header) if start_header else None
|
|
||||||
end_ind = row.index(end_header) if end_header else None
|
|
||||||
sequence_ind = row.index(sequence_header)
|
|
||||||
headers_parsed = True
|
|
||||||
continue
|
|
||||||
id = row[id_ind]
|
|
||||||
name = row[name_ind] if name_header else None
|
|
||||||
desc = row[desc_ind] if desc_header else None
|
|
||||||
start = row[start_ind] if start_header else None
|
|
||||||
end = row[end_ind] if end_header else None
|
|
||||||
sequence = row[sequence_ind]
|
|
||||||
annotations[id] = AlignedSequence(
|
|
||||||
id,
|
|
||||||
sequence,
|
|
||||||
name,
|
|
||||||
desc,
|
|
||||||
int(start) if start else None,
|
|
||||||
int(end) if end else None,
|
|
||||||
)
|
|
||||||
return annotations
|
|
||||||
|
|
||||||
|
|
||||||
def save_alignments_to_csv(
|
|
||||||
aligned_pairs: dict[str, tuple[AlignedSequence, AlignedSequence]], output_path: str
|
|
||||||
):
|
|
||||||
with open(output_path, "w") as output_fd:
|
|
||||||
writer = csv.writer(output_fd)
|
|
||||||
header_wrote = False
|
|
||||||
header_order = None
|
|
||||||
for id, annotations in aligned_pairs.items():
|
|
||||||
original, aligned = annotations
|
|
||||||
original_vars = vars(original)
|
|
||||||
aligned_vars = vars(aligned)
|
|
||||||
if not header_wrote:
|
|
||||||
header_order = list(original_vars.keys())
|
|
||||||
header = ["original" + key for key in header_order]
|
|
||||||
header.extend(["aligned" + key for key in header_order])
|
|
||||||
writer.writerow(header)
|
|
||||||
header_wrote = True
|
|
||||||
row_data = [original_vars[header] for header in header_order]
|
|
||||||
row_data.extend([aligned_vars[header] for header in header_order])
|
|
||||||
writer.writerow(row_data)
|
|
@ -1,8 +0,0 @@
|
|||||||
name: bmlsa
|
|
||||||
channels:
|
|
||||||
- conda-forge
|
|
||||||
dependencies:
|
|
||||||
- biopython=1.81
|
|
||||||
- build=0.7
|
|
||||||
- pytest=7
|
|
||||||
- twine=4
|
|
@ -1,4 +1,4 @@
|
|||||||
class AlignedSequence:
|
class Annotation:
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
id: str,
|
id: str,
|
193
protannot/protannot.py
Normal file
193
protannot/protannot.py
Normal file
@ -0,0 +1,193 @@
|
|||||||
|
import csv
|
||||||
|
import os
|
||||||
|
from skbio.alignment import StripedSmithWaterman
|
||||||
|
import argparse
|
||||||
|
from Bio import SeqIO
|
||||||
|
|
||||||
|
from Datatypes import Annotation
|
||||||
|
|
||||||
|
|
||||||
|
def read_annotations_from_csv(
|
||||||
|
csv_path: str,
|
||||||
|
id_header: str,
|
||||||
|
name_header: str,
|
||||||
|
desc_header: str,
|
||||||
|
start_header: str,
|
||||||
|
end_header: str,
|
||||||
|
sequence_header: str,
|
||||||
|
):
|
||||||
|
annotations = {}
|
||||||
|
with open(csv_path, "r") as csv_fd:
|
||||||
|
reader = csv.reader(csv_fd)
|
||||||
|
id_ind = None
|
||||||
|
name_ind = None
|
||||||
|
desc_ind = None
|
||||||
|
start_ind = None
|
||||||
|
end_ind = None
|
||||||
|
sequence_ind = None
|
||||||
|
headers_parsed = False
|
||||||
|
for row in reader:
|
||||||
|
if not headers_parsed:
|
||||||
|
id_ind = row.index(id_header)
|
||||||
|
name_ind = row.index(name_header) if name_header else None
|
||||||
|
desc_ind = row.index(desc_header) if desc_header else None
|
||||||
|
start_ind = row.index(start_header) if start_header else None
|
||||||
|
end_ind = row.index(end_header) if end_header else None
|
||||||
|
sequence_ind = row.index(sequence_header)
|
||||||
|
headers_parsed = True
|
||||||
|
continue
|
||||||
|
id = row[id_ind]
|
||||||
|
name = row[name_ind] if name_header else None
|
||||||
|
desc = row[desc_ind] if desc_header else None
|
||||||
|
start = row[start_ind] if start_header else None
|
||||||
|
end = row[end_ind] if end_header else None
|
||||||
|
sequence = row[sequence_ind]
|
||||||
|
annotations[id] = Annotation(
|
||||||
|
id,
|
||||||
|
sequence,
|
||||||
|
name,
|
||||||
|
desc,
|
||||||
|
int(start) if start else None,
|
||||||
|
int(end) if end else None,
|
||||||
|
)
|
||||||
|
return annotations
|
||||||
|
|
||||||
|
|
||||||
|
def generate_annotated_positions(sequence: str, annotations: dict[str, Annotation]):
|
||||||
|
annotation_pairs = {}
|
||||||
|
align = StripedSmithWaterman(sequence)
|
||||||
|
for id, annot in annotations.items():
|
||||||
|
alignment = align(annot.sequence)
|
||||||
|
score, aligned_start, aligned_end = (
|
||||||
|
alignment.optimal_alignment_score,
|
||||||
|
alignment.query_begin,
|
||||||
|
alignment.query_end,
|
||||||
|
)
|
||||||
|
annotation_pairs[id] = Annotation(
|
||||||
|
id,
|
||||||
|
annot.sequence,
|
||||||
|
annot.name,
|
||||||
|
annot.description,
|
||||||
|
annot.start,
|
||||||
|
annot.end,
|
||||||
|
annot.score,
|
||||||
|
), Annotation(
|
||||||
|
id,
|
||||||
|
alignment.aligned_target_sequence,
|
||||||
|
annot.name,
|
||||||
|
annot.description,
|
||||||
|
aligned_start,
|
||||||
|
aligned_end,
|
||||||
|
score,
|
||||||
|
)
|
||||||
|
return annotation_pairs
|
||||||
|
|
||||||
|
|
||||||
|
def save_alignments_to_csv(
|
||||||
|
aligned_pairs: dict[str, tuple[Annotation, Annotation]], output_path: str
|
||||||
|
):
|
||||||
|
with open(output_path, "w") as output_fd:
|
||||||
|
writer = csv.writer(output_fd)
|
||||||
|
header_wrote = False
|
||||||
|
header_order = None
|
||||||
|
for id, annotations in aligned_pairs.items():
|
||||||
|
original, aligned = annotations
|
||||||
|
original_vars = vars(original)
|
||||||
|
aligned_vars = vars(aligned)
|
||||||
|
if not header_wrote:
|
||||||
|
header_order = list(original_vars.keys())
|
||||||
|
header = ["original" + key for key in header_order]
|
||||||
|
header.extend(["aligned" + key for key in header_order])
|
||||||
|
writer.writerow(header)
|
||||||
|
header_wrote = True
|
||||||
|
row_data = [original_vars[header] for header in header_order]
|
||||||
|
row_data.extend([aligned_vars[header] for header in header_order])
|
||||||
|
writer.writerow(row_data)
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
argparser = argparse.ArgumentParser("protannot")
|
||||||
|
argparser.add_argument(
|
||||||
|
"annotations",
|
||||||
|
type=str,
|
||||||
|
help=(
|
||||||
|
"Path to CSV containing the sequences to align as well as the "
|
||||||
|
"annotations for the respective sequences"
|
||||||
|
),
|
||||||
|
metavar="a",
|
||||||
|
)
|
||||||
|
argparser.add_argument(
|
||||||
|
"sequence",
|
||||||
|
type=str,
|
||||||
|
help=(
|
||||||
|
"Path to the sequence to annotate in FASTA format. "
|
||||||
|
"If multiple sequences are present, annotations will be run on each"
|
||||||
|
),
|
||||||
|
metavar="s",
|
||||||
|
)
|
||||||
|
argparser.add_argument(
|
||||||
|
"output", type=str, help="Path to output location", metavar="o"
|
||||||
|
)
|
||||||
|
argparser.add_argument(
|
||||||
|
"-I", "--id-header", type=str, help="The header for the ID of the annotation"
|
||||||
|
)
|
||||||
|
argparser.add_argument(
|
||||||
|
"-N",
|
||||||
|
"--name-header",
|
||||||
|
type=str,
|
||||||
|
help="The header for the name of the annotation",
|
||||||
|
required=False,
|
||||||
|
)
|
||||||
|
argparser.add_argument(
|
||||||
|
"-D",
|
||||||
|
"--desc-header",
|
||||||
|
type=str,
|
||||||
|
help="The header for the description of the annotation",
|
||||||
|
required=False,
|
||||||
|
)
|
||||||
|
argparser.add_argument(
|
||||||
|
"-T",
|
||||||
|
"--start-header",
|
||||||
|
type=str,
|
||||||
|
help="The header for the start of the annotation",
|
||||||
|
required=False,
|
||||||
|
)
|
||||||
|
argparser.add_argument(
|
||||||
|
"-E",
|
||||||
|
"--end-header",
|
||||||
|
type=str,
|
||||||
|
help="The header for the end of the annotation",
|
||||||
|
required=False,
|
||||||
|
)
|
||||||
|
argparser.add_argument(
|
||||||
|
"-S",
|
||||||
|
"--seq-header",
|
||||||
|
type=str,
|
||||||
|
help="The header for the sequence of the annotation",
|
||||||
|
)
|
||||||
|
args = argparser.parse_args()
|
||||||
|
given_annotations = read_annotations_from_csv(
|
||||||
|
args.annotations,
|
||||||
|
args.id_header,
|
||||||
|
args.name_header,
|
||||||
|
args.desc_header,
|
||||||
|
args.start_header,
|
||||||
|
args.end_header,
|
||||||
|
args.seq_header,
|
||||||
|
)
|
||||||
|
with open(args.sequence, "r") as sequence_fd:
|
||||||
|
for sequence in SeqIO.parse(sequence_fd, "fasta"):
|
||||||
|
aligned_annotations = generate_annotated_positions(
|
||||||
|
str(sequence.seq), given_annotations
|
||||||
|
)
|
||||||
|
save_alignments_to_csv(
|
||||||
|
aligned_annotations,
|
||||||
|
os.path.join(
|
||||||
|
args.output,
|
||||||
|
sequence.id.replace("|", "+").replace(".", "_") + ".csv",
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
@ -1,3 +0,0 @@
|
|||||||
[build-system]
|
|
||||||
build-backend = "setuptools.build_meta"
|
|
||||||
requires = ["setuptools", "wheel"]
|
|
12
setup.cfg
12
setup.cfg
@ -1,12 +0,0 @@
|
|||||||
[metadata]
|
|
||||||
name = bmlsa
|
|
||||||
version = 0.0.1
|
|
||||||
|
|
||||||
[options]
|
|
||||||
packages = bmlsa
|
|
||||||
install_requires =
|
|
||||||
biopython ==1.81
|
|
||||||
|
|
||||||
[options.entry_points]
|
|
||||||
console_scripts =
|
|
||||||
bmlsa = bmlsa.cli:main
|
|
Loading…
x
Reference in New Issue
Block a user