ydeng/autoBIGS.engine
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Writing now uses named MLST profile
All checks were successful
automlst.engine/pipeline/head This commit looks good
Details
automlst.engine/pipeline/tag This commit looks good
Details

Browse Source
This commit is contained in:
Harrison Deng 2025-02-18 16:03:17 +00:00
parent 5a03c7e8d8
commit 7384895578
2 changed files with 12 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/autobigs/engine/writing.py
View File

@ -3,7 +3,7 @@ import csv
from os import PathLike from os import PathLike
from typing import AsyncIterable, Collection, Mapping, Sequence, Union from typing import AsyncIterable, Collection, Mapping, Sequence, Union
from autobigs.engine.structures.mlst import Allele, MLSTProfile from autobigs.engine.structures.mlst import Allele, MLSTProfile, NamedMLSTProfile
def alleles_to_text_map(alleles: Collection[Allele]) -> Mapping[str, Union[Sequence[str], str]]: def alleles_to_text_map(alleles: Collection[Allele]) -> Mapping[str, Union[Sequence[str], str]]:
@ -17,12 +17,14 @@ def alleles_to_text_map(alleles: Collection[Allele]) -> Mapping[str, Union[Seque
result[locus] = tuple(result[locus]) # type: ignore result[locus] = tuple(result[locus]) # type: ignore
return dict(result) return dict(result)
async def write_mlst_profiles_as_csv(mlst_profiles_iterable: AsyncIterable[tuple[str, Union[MLSTProfile, None]]], handle: Union[str, bytes, PathLike[str], PathLike[bytes]]) -> Sequence[str]: async def write_mlst_profiles_as_csv(mlst_profiles_iterable: AsyncIterable[NamedMLSTProfile], handle: Union[str, bytes, PathLike[str], PathLike[bytes]]) -> Sequence[str]:
failed = list() failed = list()
with open(handle, "w", newline='') as filehandle: with open(handle, "w", newline='') as filehandle:
header = None header = None
writer: Union[csv.DictWriter, None] = None writer: Union[csv.DictWriter, None] = None
async for name, mlst_profile in mlst_profiles_iterable: async for named_mlst_profile in mlst_profiles_iterable:
name = named_mlst_profile.name
mlst_profile = named_mlst_profile.mlst_profile
if mlst_profile is None: if mlst_profile is None:
failed.append(name) failed.append(name)
continue continue

12
tests/autobigs/engine/test_writing.py
View File

@ -3,7 +3,7 @@ from typing import AsyncIterable, Iterable
import pytest import pytest
from autobigs.engine.structures.alignment import AlignmentStats from autobigs.engine.structures.alignment import AlignmentStats
from autobigs.engine.writing import alleles_to_text_map, write_mlst_profiles_as_csv from autobigs.engine.writing import alleles_to_text_map, write_mlst_profiles_as_csv
from autobigs.engine.structures.mlst import Allele, MLSTProfile from autobigs.engine.structures.mlst import Allele, MLSTProfile, NamedMLSTProfile
import tempfile import tempfile
from csv import reader from csv import reader
from os import path from os import path
@ -11,20 +11,20 @@ from os import path
@pytest.fixture @pytest.fixture
def dummy_alphabet_mlst_profile(): def dummy_alphabet_mlst_profile():
return MLSTProfile(( return NamedMLSTProfile("name", MLSTProfile((
Allele("A", "1", None), Allele("A", "1", None),
Allele("D", "1", None), Allele("D", "1", None),
Allele("B", "1", None), Allele("B", "1", None),
Allele("C", "1", None), Allele("C", "1", None),
Allele("C", "2", AlignmentStats(90, 10, 0, 90)) Allele("C", "2", AlignmentStats(90, 10, 0, 90))
), "mysterious", "very mysterious") ), "mysterious", "very mysterious"))
async def iterable_to_asynciterable(iterable: Iterable): async def iterable_to_asynciterable(iterable: Iterable):
for iterated in iterable: for iterated in iterable:
yield iterated yield iterated
async def test_column_order_is_same_as_expected_file(dummy_alphabet_mlst_profile: MLSTProfile): async def test_column_order_is_same_as_expected_file(dummy_alphabet_mlst_profile: MLSTProfile):
dummy_profiles = [("test_1", dummy_alphabet_mlst_profile)] dummy_profiles = [dummy_alphabet_mlst_profile]
with tempfile.TemporaryDirectory() as temp_dir: with tempfile.TemporaryDirectory() as temp_dir:
output_path = path.join(temp_dir, "out.csv") output_path = path.join(temp_dir, "out.csv")
await write_mlst_profiles_as_csv(iterable_to_asynciterable(dummy_profiles), output_path) await write_mlst_profiles_as_csv(iterable_to_asynciterable(dummy_profiles), output_path)
@ -34,8 +34,8 @@ async def test_column_order_is_same_as_expected_file(dummy_alphabet_mlst_profile
target_columns = lines[4:] target_columns = lines[4:]
assert target_columns == sorted(target_columns) assert target_columns == sorted(target_columns)
async def test_alleles_to_text_map_mapping_is_correct(dummy_alphabet_mlst_profile: MLSTProfile): async def test_alleles_to_text_map_mapping_is_correct(dummy_alphabet_mlst_profile: NamedMLSTProfile):
mapping = alleles_to_text_map(dummy_alphabet_mlst_profile.alleles) mapping = alleles_to_text_map(dummy_alphabet_mlst_profile.mlst_profile.alleles) # type: ignore
expected_mapping = { expected_mapping = {
"A": "1", "A": "1",
"B": "1", "B": "1",