Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 9 additions & 14 deletions src/biotite/application/sra/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
from os.path import join
from subprocess import PIPE, Popen, SubprocessError, TimeoutExpired
from tempfile import TemporaryDirectory
from typing import Literal, TypeAlias
import numpy as np
from biotite.application.application import (
Application,
Expand All @@ -28,10 +27,6 @@
from biotite.sequence.io.fastq.file import FastqFile
from biotite.sequence.seqtypes import NucleotideSequence

_OffsetFormat: TypeAlias = Literal[
"Sanger", "Solexa", "Illumina-1.3", "Illumina-1.5", "Illumina-1.8"
]


# Do not use LocalApp, as two programs are executed
class _DumpApp(Application, metaclass=abc.ABCMeta):
Expand Down Expand Up @@ -237,11 +232,11 @@ class FastqDumpApp(_DumpApp):
prefetch_path, fasterq_dump_path : str, optional
Path to the ``prefetch_path`` and ``fasterq-dump`` binary,
respectively.
offset : int or {'Sanger', 'Solexa', 'Illumina-1.3', 'Illumina-1.5', 'Illumina-1.8'}, optional
offset : int or FastqFile.Offset, optional
This value is subtracted from the FASTQ ASCII code to obtain the
quality score.
Can either be directly the value, or a string that indicates
the score format.
Can be provided directly as integer or as a member of
:class:`FastqFile.Offset`.
"""

def __init__(
Expand All @@ -250,10 +245,10 @@ def __init__(
output_path_prefix: PathLike[str] | str | None = None,
prefetch_path: PathLike[str] | str = "prefetch",
fasterq_dump_path: PathLike[str] | str = "fasterq-dump",
offset: int | _OffsetFormat = "Sanger",
offset: int | FastqFile.Offset = FastqFile.Offset.SANGER,
) -> None:
super().__init__(uid, output_path_prefix, prefetch_path, fasterq_dump_path)
self._offset: int | _OffsetFormat = offset
self._offset: int | FastqFile.Offset = offset
self._fastq_files: list[FastqFile] | None = None

@requires_state(AppState.JOINED)
Expand Down Expand Up @@ -312,7 +307,7 @@ def fetch(
output_path_prefix: PathLike[str] | str | None = None,
prefetch_path: PathLike[str] | str = "prefetch",
fasterq_dump_path: PathLike[str] | str = "fasterq-dump",
offset: int | _OffsetFormat = "Sanger",
offset: int | FastqFile.Offset = FastqFile.Offset.SANGER,
) -> list[dict[str, NucleotideSequence]]:
"""
Get the sequences belonging to the UID from the
Expand All @@ -333,11 +328,11 @@ def fetch(
prefetch_path, fasterq_dump_path : str, optional
Path to the ``prefetch_path`` and ``fasterq-dump`` binary,
respectively.
offset : int or {'Sanger', 'Solexa', 'Illumina-1.3', 'Illumina-1.5', 'Illumina-1.8'}, optional
offset : int or FastqFile.Offset, optional
This value is subtracted from the FASTQ ASCII code to obtain the
quality score.
Can either be directly the value, or a string that indicates
the score format.
Can be provided directly as integer or as a member of
:class:`FastqFile.Offset`.

Returns
-------
Expand Down
87 changes: 36 additions & 51 deletions src/biotite/sequence/io/fastq/file.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,25 +9,16 @@

from collections import OrderedDict
from collections.abc import Iterable, Iterator, MutableMapping
from numbers import Integral
from enum import IntEnum
from os import PathLike
from typing import IO, Literal, Self
from typing import IO, Self
import numpy as np
from biotite.file import InvalidFileError, TextFile, wrap_string
from biotite.typing import K, NDArray1

__all__ = ["FastqFile"]


_OFFSETS = {
"Sanger": 33,
"Solexa": 64,
"Illumina-1.3": 64,
"Illumina-1.5": 64,
"Illumina-1.8": 33,
}


class FastqFile(TextFile, MutableMapping[str, tuple[str, np.ndarray]]):
"""
This class represents a file in FASTQ format.
Expand All @@ -42,8 +33,8 @@ class FastqFile(TextFile, MutableMapping[str, tuple[str, np.ndarray]]):
`offset` value.
The offset is format dependent.
As the offset is not reliably deducible from the file contets, it
must be provided explicitly, either as number or format
(e.g. ``'Illumina-1.8'``).
must be provided explicitly, either as number or as a
:class:`FastqFile.Offset` member.

Similar to the :class:`FastaFile` class, this class implements the
:class:`MutableMapping` interface:
Expand All @@ -53,11 +44,11 @@ class FastqFile(TextFile, MutableMapping[str, tuple[str, np.ndarray]]):

Parameters
----------
offset : int or {'Sanger', 'Solexa', 'Illumina-1.3', 'Illumina-1.5', 'Illumina-1.8'}
offset : int or FastqFile.Offset
This value is added to the quality score to obtain the
ASCII code.
Can either be directly the value, or a string that indicates
the score format.
Can be provided directly as integer or as a member of
:class:`FastqFile.Offset`.
chars_per_line : int, optional
The number characters in a line containing sequence data
after which a line break is inserted.
Expand All @@ -69,7 +60,7 @@ class FastqFile(TextFile, MutableMapping[str, tuple[str, np.ndarray]]):
--------

>>> import os.path
>>> file = FastqFile(offset="Sanger")
>>> file = FastqFile(offset=FastqFile.Offset.SANGER)
>>> file["seq1"] = str(NucleotideSequence("ATACT")), [0,3,10,7,12]
>>> file["seq2"] = str(NucleotideSequence("TTGTAGG")), [15,13,24,21,28,38,35]
>>> print(file)
Expand All @@ -95,23 +86,32 @@ class FastqFile(TextFile, MutableMapping[str, tuple[str, np.ndarray]]):
>>> file.write(os.path.join(path_to_directory, "test.fastq"))
"""

class Offset(IntEnum):
"""
Quality score ASCII offsets for FASTQ format variants.
"""

SANGER = 33
SOLEXA = 64
ILLUMINA_1_3 = 64
ILLUMINA_1_5 = 64
ILLUMINA_1_8 = 33

def __init__(
self,
offset: int
| Literal["Sanger", "Solexa", "Illumina-1.3", "Illumina-1.5", "Illumina-1.8"],
offset: int | FastqFile.Offset,
chars_per_line: int | None = None,
) -> None:
super().__init__()
self._chars_per_line = chars_per_line
self._entries: OrderedDict[str, tuple[int, int, int, int]] = OrderedDict()
self._offset = _convert_offset(offset)
self._offset = int(offset)

@classmethod
def read(
cls,
file: PathLike[str] | str | IO[str],
offset: int
| Literal["Sanger", "Solexa", "Illumina-1.3", "Illumina-1.5", "Illumina-1.8"],
offset: int | FastqFile.Offset,
chars_per_line: int | None = None,
) -> Self:
"""
Expand All @@ -122,10 +122,11 @@ def read(
file : file-like object or str
The file to be read.
Alternatively a file path can be supplied.
offset : int or {'Sanger', 'Solexa', 'Illumina-1.3', 'Illumina-1.5', 'Illumina-1.8'}
offset : int or FastqFile.Offset
This value is added to the quality score to obtain the
ASCII code.
Can either be directly the value, or a string that indicates
Can be provided directly as integer, as a member of
:class:`FastqFile.Offset`, or as a string that indicates
the score format.
chars_per_line : int, optional
The number characters in a line containing sequence data
Expand Down Expand Up @@ -330,8 +331,7 @@ def _find_entries(self) -> None:
@staticmethod
def read_iter(
file: PathLike[str] | str | IO[str],
offset: int
| Literal["Sanger", "Solexa", "Illumina-1.3", "Illumina-1.5", "Illumina-1.8"],
offset: int | FastqFile.Offset,
) -> Iterator[tuple[str, tuple[str, np.ndarray]]]:
"""
Create an iterator over each sequence (and corresponding scores)
Expand All @@ -342,10 +342,11 @@ def read_iter(
file : file-like object or str
The file to be read.
Alternatively a file path can be supplied.
offset : int or {'Sanger', 'Solexa', 'Illumina-1.3', 'Illumina-1.5', 'Illumina-1.8'}
This value that is added to the quality score to obtain the
offset : int or FastqFile.Offset
This value is added to the quality score to obtain the
ASCII code.
Can either be directly the value, or a string that indicates
Can be provided directly as integer, as a member of
:class:`FastqFile.Offset`, or as a string that indicates
the score format.

Yields
Expand All @@ -362,7 +363,7 @@ def read_iter(
`FastqFile.read(file, offset).items()`, but is slightly faster
and much more memory efficient.
"""
offset = _convert_offset(offset)
offset = int(offset)

identifier: str = "None"
seq_str_list: list[str] = []
Expand Down Expand Up @@ -422,8 +423,7 @@ def read_iter(
def write_iter(
file: PathLike[str] | str | IO[str],
items: Iterable[tuple[str, tuple[str, np.ndarray]]],
offset: int
| Literal["Sanger", "Solexa", "Illumina-1.3", "Illumina-1.5", "Illumina-1.8"],
offset: int | FastqFile.Offset,
chars_per_line: int | None = None,
) -> None:
"""
Expand All @@ -446,10 +446,11 @@ def write_iter(
The entries to be written into the file.
Each entry consists of an identifier string and a tuple
containing a sequence (as string) and a score array.
offset : int or {'Sanger', 'Solexa', 'Illumina-1.3', 'Illumina-1.5', 'Illumina-1.8'}
offset : int or FastqFile.Offset
This value is added to the quality score to obtain the
ASCII code.
Can either be directly the value, or a string that indicates
Can be provided directly as integer, as a member of
:class:`FastqFile.Offset`, or as a string that indicates
the score format.
chars_per_line : int, optional
The number characters in a line containing sequence data
Expand All @@ -463,7 +464,7 @@ def write_iter(
This method does not test, whether the given identifiers are
unambiguous.
"""
offset = _convert_offset(offset)
offset = int(offset)

def line_generator() -> Iterator[str]:
for item in items:
Expand Down Expand Up @@ -515,19 +516,3 @@ def _scores_to_score_str(scores: NDArray1[K, np.integer], offset: int) -> str:
"""
scores = np.asarray(scores) + offset
return scores.astype(np.int8, copy=False).tobytes().decode("ascii")


def _convert_offset(offset_val_or_string: int | str) -> int:
"""
If the given offset is a string return the corresponding numerical
value.
"""
if isinstance(offset_val_or_string, (int, Integral)):
return int(offset_val_or_string)
elif isinstance(offset_val_or_string, str):
return _OFFSETS[offset_val_or_string]
else:
raise TypeError(
f"The offset must be either an integer or a string "
f"indicating the format, not {type(offset_val_or_string).__name__}"
)
8 changes: 4 additions & 4 deletions src/biotite/sequence/io/general.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ def load_sequence(file_path: PathLike[str] | str) -> Sequence:

# Quality scores are irrelevant for this function
# -> Offset is irrelevant
file = FastqFile.read(file_path, offset="Sanger")
file = FastqFile.read(file_path, offset=FastqFile.Offset.SANGER)
# Get first sequence
try:
seq_str, _ = next(iter(file.values()))
Expand Down Expand Up @@ -98,7 +98,7 @@ def save_sequence(file_path: PathLike[str] | str, sequence: Sequence) -> None:

# Quality scores are irrelevant for this function
# -> Offset is irrelevant
file = FastqFile(offset="Sanger")
file = FastqFile(offset=FastqFile.Offset.SANGER)
# Scores are set to 0 since no score information is supplied
scores = np.zeros(len(sequence))
file["sequence"] = str(sequence), scores
Expand Down Expand Up @@ -150,7 +150,7 @@ def load_sequences(file_path: PathLike[str] | str) -> dict[str, Sequence]:

# Quality scores are irrelevant for this function
# -> Offset is irrelevant
file = FastqFile.read(file_path, offset="Sanger")
file = FastqFile.read(file_path, offset=FastqFile.Offset.SANGER)
return {
identifier: NucleotideSequence(seq_str)
for identifier, (seq_str, scores) in file.items()
Expand Down Expand Up @@ -199,7 +199,7 @@ def save_sequences(

# Quality scores are irrelevant for this function
# -> Offset is irrelevant
file = FastqFile(offset="Sanger")
file = FastqFile(offset=FastqFile.Offset.SANGER)
for identifier, sequence in sequences.items():
# Scores are set to 0 since no score information is supplied
scores = np.zeros(len(sequence))
Expand Down
34 changes: 30 additions & 4 deletions tests/sequence/test_fastq.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,10 +55,32 @@ def test_conversion(chars_per_line):
assert np.array_equal(test_scores, ref_scores)


@pytest.mark.parametrize(

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Without the need of the legacy names I think this test can be omitted.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in eb4753b7 — I removed the obsolete legacy-string test. The broader update also applies FastqFile.Offset throughout biotite.application.sra. Validation: 23 FASTQ tests passed and pixi run lint completed successfully.

("member_name", "expected_value"),
[
("SANGER", 33),
("SOLEXA", 64),
("ILLUMINA_1_3", 64),
("ILLUMINA_1_5", 64),
("ILLUMINA_1_8", 33),
],
)
def test_offset_enum(member_name, expected_value):
offset = fastq.FastqFile.Offset[member_name]
assert int(offset) == expected_value

scores = np.array([0, 1, 2])
enum_file = fastq.FastqFile(offset)
enum_file["seq"] = "ACG", scores
actual_sequence, actual_scores = enum_file["seq"]
assert actual_sequence == "ACG"
assert np.array_equal(actual_scores, scores)


def test_rna_conversion():
sequence = seq.NucleotideSequence("ACGT")
scores = np.array([0, 0, 0, 0])
fastq_file = fastq.FastqFile(offset="Sanger")
fastq_file = fastq.FastqFile(offset=fastq.FastqFile.Offset.SANGER)
fastq.set_sequence(fastq_file, sequence, scores, "seq1", as_rna=False)
fastq.set_sequence(fastq_file, sequence, scores, "seq2", as_rna=True)
assert fastq_file["seq1"][0] == "ACGT"
Expand All @@ -71,9 +93,13 @@ def test_rna_conversion():
ids=lambda path: path.name,
)
def test_read_iter(file_name):
ref_dict = dict(fastq.FastqFile.read(file_name, offset="Sanger").items())
ref_dict = dict(
fastq.FastqFile.read(file_name, offset=fastq.FastqFile.Offset.SANGER).items()
)

test_dict = dict(fastq.FastqFile.read_iter(file_name, offset="Sanger"))
test_dict = dict(
fastq.FastqFile.read_iter(file_name, offset=fastq.FastqFile.Offset.SANGER)
)

for (test_id, (test_seq, test_sc)), (ref_id, (ref_seq, ref_sc)) in zip(
test_dict.items(), ref_dict.items()
Expand All @@ -83,7 +109,7 @@ def test_read_iter(file_name):
assert (test_sc == ref_sc).all()


@pytest.mark.parametrize("offset", [33, 42, "Solexa"])
@pytest.mark.parametrize("offset", [33, 42, fastq.FastqFile.Offset.SOLEXA])
@pytest.mark.parametrize("chars_per_line", [None, 80])
@pytest.mark.parametrize("n_sequences", [1, 10])
def test_write_iter(offset, chars_per_line, n_sequences):
Expand Down
Loading