diff --git a/src/biotite/application/sra/app.py b/src/biotite/application/sra/app.py index e55d96274..a5a56fe6b 100644 --- a/src/biotite/application/sra/app.py +++ b/src/biotite/application/sra/app.py @@ -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, @@ -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): @@ -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__( @@ -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) @@ -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 @@ -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 ------- diff --git a/src/biotite/sequence/io/fastq/file.py b/src/biotite/sequence/io/fastq/file.py index be0b2a209..ece1f1231 100644 --- a/src/biotite/sequence/io/fastq/file.py +++ b/src/biotite/sequence/io/fastq/file.py @@ -9,9 +9,9 @@ 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 @@ -19,15 +19,6 @@ __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. @@ -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: @@ -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. @@ -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) @@ -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: """ @@ -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 @@ -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) @@ -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 @@ -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] = [] @@ -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: """ @@ -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 @@ -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: @@ -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__}" - ) diff --git a/src/biotite/sequence/io/general.py b/src/biotite/sequence/io/general.py index c455a27a1..24413fe36 100644 --- a/src/biotite/sequence/io/general.py +++ b/src/biotite/sequence/io/general.py @@ -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())) @@ -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 @@ -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() @@ -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)) diff --git a/tests/sequence/test_fastq.py b/tests/sequence/test_fastq.py index 6cc41e8d5..b45dd5c2e 100644 --- a/tests/sequence/test_fastq.py +++ b/tests/sequence/test_fastq.py @@ -55,10 +55,32 @@ def test_conversion(chars_per_line): assert np.array_equal(test_scores, ref_scores) +@pytest.mark.parametrize( + ("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" @@ -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() @@ -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):