From 6549f1f879781f66ef378f9f9950d61afca0eb87 Mon Sep 17 00:00:00 2001 From: Dylan Costo <99520325+DylanCosto@users.noreply.github.com> Date: Sat, 10 Oct 2026 23:43:01 -0400 Subject: [PATCH 01/11] Remove all-gap alignment columns without repeated list deletion --- lib/bx/align/core.py | 41 +++++++++++---------------------------- lib/bx/align/maf_tests.py | 9 ++++++++- 2 files changed, 19 insertions(+), 31 deletions(-) diff --git a/lib/bx/align/core.py b/lib/bx/align/core.py index 379736e..6864525 100644 --- a/lib/bx/align/core.py +++ b/lib/bx/align/core.py @@ -166,36 +166,17 @@ def remove_all_gap_columns(self): Remove any columns containing only gaps from alignment components, text of components is modified IN PLACE. """ - seqs = [] - for c in self.components: - if c.empty: - seqs.append(None) - try: - seqs.append(list(c.text)) - except TypeError: - seqs.append(None) - i = 0 - text_size = self.text_size - while i < text_size: - all_gap = True - for seq in seqs: - if seq is None: - continue - if seq[i] != "-": - all_gap = False - if all_gap: - for seq in seqs: - if seq is None: - continue - del seq[i] - text_size -= 1 - else: - i += 1 - for i in range(len(self.components)): - if seqs[i] is None: - continue - self.components[i].text = "".join(seqs[i]) - self.text_size = text_size + components = [c for c in self.components if not c.empty and c.text is not None] + if any("-" not in c.text for c in components): + return + # Decide which columns to retain before modifying any component. Deleting + # individual gaps from lists repeatedly shifts the remaining alignment. + keep = [i for i in range(self.text_size) if any(c.text[i] != "-" for c in components)] + if len(keep) == self.text_size: + return + for component in components: + component.text = "".join(component.text[i] for i in keep) + self.text_size = len(keep) def __eq__(self, other): if other is None or not isinstance(other, type(self)): diff --git a/lib/bx/align/maf_tests.py b/lib/bx/align/maf_tests.py index d343f0f..d82b1e1 100644 --- a/lib/bx/align/maf_tests.py +++ b/lib/bx/align/maf_tests.py @@ -4,6 +4,8 @@ from io import StringIO +import pytest + from bx import align from bx.align import maf @@ -211,7 +213,8 @@ def test_column_iter(): assert c == expected[i] -def test_remove_all_gap_column(): +@pytest.mark.parametrize("empty_index", [0, 1, 3]) +def test_remove_all_gap_column(empty_index): complex_maf_gap = align.Alignment() complex_maf_gap.score = "7009" complex_maf_gap.components.append( @@ -231,7 +234,11 @@ def test_remove_all_gap_column(): complex_maf_gap.components[-1].empty = True complex_maf_gap.components[-1].synteny_empty = maf.MAF_INSERT_STATUS complex_maf_gap.text_size = 11 + empty = complex_maf_gap.components.pop() + complex_maf_gap.components.insert(empty_index, empty) complex_maf_gap.remove_all_gap_columns() + # Return the empty row to its fixture position for the equality check. + complex_maf_gap.components.append(complex_maf_gap.components.pop(empty_index)) assert complex_maf_gap == complex_maf From 5637291dd37a2842785d834dae4c31912ac687a7 Mon Sep 17 00:00:00 2001 From: Dylan Costo <99520325+DylanCosto@users.noreply.github.com> Date: Sun, 11 Oct 2026 00:05:19 -0400 Subject: [PATCH 02/11] Compact shared alignment components only once --- lib/bx/align/core.py | 2 ++ lib/bx/align/maf_tests.py | 7 ++++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/lib/bx/align/core.py b/lib/bx/align/core.py index 6864525..3772191 100644 --- a/lib/bx/align/core.py +++ b/lib/bx/align/core.py @@ -167,6 +167,8 @@ def remove_all_gap_columns(self): text of components is modified IN PLACE. """ components = [c for c in self.components if not c.empty and c.text is not None] + # A component can occur more than once; rebuild each object only once. + components = list({id(c): c for c in components}.values()) if any("-" not in c.text for c in components): return # Decide which columns to retain before modifying any component. Deleting diff --git a/lib/bx/align/maf_tests.py b/lib/bx/align/maf_tests.py index d82b1e1..cf0710c 100644 --- a/lib/bx/align/maf_tests.py +++ b/lib/bx/align/maf_tests.py @@ -214,7 +214,8 @@ def test_column_iter(): @pytest.mark.parametrize("empty_index", [0, 1, 3]) -def test_remove_all_gap_column(empty_index): +@pytest.mark.parametrize("repeat_component", [False, True]) +def test_remove_all_gap_column(empty_index, repeat_component): complex_maf_gap = align.Alignment() complex_maf_gap.score = "7009" complex_maf_gap.components.append( @@ -236,7 +237,11 @@ def test_remove_all_gap_column(empty_index): complex_maf_gap.text_size = 11 empty = complex_maf_gap.components.pop() complex_maf_gap.components.insert(empty_index, empty) + if repeat_component: + complex_maf_gap.add_component(next(c for c in complex_maf_gap.components if not c.empty)) complex_maf_gap.remove_all_gap_columns() + if repeat_component: + complex_maf_gap.components.pop() # Return the empty row to its fixture position for the equality check. complex_maf_gap.components.append(complex_maf_gap.components.pop(empty_index)) assert complex_maf_gap == complex_maf From 91b676f201dd56bf859f01ae7eb7b01b7aa56e6d Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Thu, 15 Oct 2026 16:11:19 +0100 Subject: [PATCH 03/11] Fix `revcomp` handling in `SeqFile` Commit fd08da3b49e89e0accd2c372d4a99427628ce432 replaced `if (revcomp == True):` with `if revcomp:` in `SeqFile.__init__()`. Since any non-empty string is truthy, every string value of `revcomp` was then normalised to `"-5'"`, so with `"+5'"` or `"-3'"` `SeqFile.get()` silently returned the reverse complement of the wrong interval (`"+3'"` and `"maf"` were unaffected, as they map to `"-5'"` anyway). Check for `revcomp is True` instead, and add a test covering all the supported `revcomp` values. Also simplify `SeqFile.get()` to fetch the raw sequence only once, and add type annotations for `revcomp`. `SeqFile.get()` now only accepts `False`, `"-3'"` or `"-5'"` as the normalised scheme, so other falsy values (e.g. `0`, `None` or `""`) now raise `AssertionError` instead of being treated as no reverse complement. Co-Authored-By: Claude Opus 5.5 --- lib/bx/seq/seq.py | 26 +++++++++++++++----------- lib/bx/seq/seq_tests.py | 21 +++++++++++++++++++++ 2 files changed, 36 insertions(+), 11 deletions(-) diff --git a/lib/bx/seq/seq.py b/lib/bx/seq/seq.py index c63fe96..a53e901 100644 --- a/lib/bx/seq/seq.py +++ b/lib/bx/seq/seq.py @@ -4,6 +4,8 @@ :Author: Bob Harris (rsharris@bx.psu.edu) """ +from typing import Literal + # DNA reverse complement table DNA_COMP = ( @@ -36,10 +38,12 @@ class attributes: gap: gap character that aligners should use for gaps in this sequence """ - def __init__(self, file=None, revcomp=False, name="", gap=None): + def __init__( + self, file=None, revcomp: bool | Literal["maf", "+5'", "+3'", "-5'", "-3'"] = False, name="", gap=None + ) -> None: self.file = file - if revcomp or revcomp == "+3'": - self.revcomp = "-5'" + if revcomp is True or revcomp == "+3'": + self.revcomp: Literal[False, "-5'", "-3'"] = "-5'" elif revcomp == "+5'": self.revcomp = "-3'" elif revcomp == "maf": @@ -91,14 +95,14 @@ def get(self, start, length): assert ( start + length <= self.length ), f"Interval beyond end of sequence ({start}..{start + length} > {self.length})" - # Fetch sequence and reverse complement if necesary - if not self.revcomp: - return self.raw_fetch(start, length) - if self.revcomp == "-3'": - return self.reverse_complement(self.raw_fetch(start, length)) - assert self.revcomp == "-5'", "unrecognized reverse complement scheme" - start = self.length - (start + length) - return self.reverse_complement(self.raw_fetch(start, length)) + # Fetch sequence and reverse complement if necessary + assert self.revcomp is False or self.revcomp in ("-3'", "-5'"), "unrecognized reverse complement scheme" + if self.revcomp == "-5'": + start = self.length - (start + length) + raw_seq = self.raw_fetch(start, length) + if self.revcomp is False: + return raw_seq + return self.reverse_complement(raw_seq) def raw_fetch(self, start, length): return self.text[start : start + length] diff --git a/lib/bx/seq/seq_tests.py b/lib/bx/seq/seq_tests.py index 2499cbd..b2166bc 100644 --- a/lib/bx/seq/seq_tests.py +++ b/lib/bx/seq/seq_tests.py @@ -4,6 +4,8 @@ import unittest +import pytest + import bx.seq from bx.seq import ( fasta_tests, @@ -59,3 +61,22 @@ def test_get_reader(self): def check_get(seqfile, valid_seq, start, len): assert seqfile.get(start, len) == valid_seq[start : start + len] + + +@pytest.mark.parametrize( + "revcomp, expected_revcomp, expected_get", + [ + (False, False, "AAC"), + (True, "-5'", "GTA"), + ("+3'", "-5'", "GTA"), + ("+5'", "-3'", "GTT"), + ("maf", "-5'", "GTA"), + ("-5'", "-5'", "GTA"), + ("-3'", "-3'", "GTT"), + ], +) +def test_revcomp(revcomp, expected_revcomp, expected_get): + seqfile = bx.seq.seq.SeqFile(revcomp=revcomp) + seqfile.set_text("AACCGGTTAC") + assert seqfile.revcomp == expected_revcomp + assert seqfile.get(0, 3) == expected_get From ba66f8156fbc837d122681ebfea50b19009e9c22 Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Thu, 15 Oct 2026 16:37:12 +0100 Subject: [PATCH 04/11] Fix `species_to_lengths` passed to wrong parameter by `get_reader()` and `get_indexed()` `get_reader()` and `get_indexed()` passed `species_to_lengths` as a positional argument, but in the reader/indexed-access classes it isn't the next positional parameter: - `get_reader("maf", ...)` failed because `maf.Reader.__init__()` only accepts keyword arguments after `file`: ``` TypeError: Reader.__init__() takes 2 positional arguments but 3 were given ``` - `get_reader("axt", ...)` passed it as `species1`, so reading a block failed with: ``` TypeError: unsupported operand type(s) for +: 'dict' and 'str' ``` - `get_indexed()` passed it as `use_cache` for MAF and as `species1` for AXT, so it was never used as the species-to-lengths mapping. Pass it as a keyword argument instead, and add tests. Co-Authored-By: Claude Opus 5.5 --- lib/bx/align/core.py | 8 ++++---- lib/bx/align/core_tests.py | 42 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 4 deletions(-) create mode 100644 lib/bx/align/core_tests.py diff --git a/lib/bx/align/core.py b/lib/bx/align/core.py index 379736e..f936e1b 100644 --- a/lib/bx/align/core.py +++ b/lib/bx/align/core.py @@ -432,9 +432,9 @@ def get_reader(format, infile, species_to_lengths=None): import bx.align.maf if format == "maf": - return bx.align.maf.Reader(infile, species_to_lengths) + return bx.align.maf.Reader(infile, species_to_lengths=species_to_lengths) elif format == "axt": - return bx.align.axt.Reader(infile, species_to_lengths) + return bx.align.axt.Reader(infile, species_to_lengths=species_to_lengths) elif format == "lav": return bx.align.lav.Reader(infile) else: @@ -464,9 +464,9 @@ def get_indexed(format, filename, index_filename=None, keep_open=False, species_ import bx.align.maf if format == "maf": - return bx.align.maf.Indexed(filename, index_filename, keep_open, species_to_lengths) + return bx.align.maf.Indexed(filename, index_filename, keep_open, species_to_lengths=species_to_lengths) elif format == "axt": - return bx.align.axt.Indexed(filename, index_filename, keep_open, species_to_lengths) + return bx.align.axt.Indexed(filename, index_filename, keep_open, species_to_lengths=species_to_lengths) elif format == "lav": raise Exception("LAV support for Indexed has not been implemented") else: diff --git a/lib/bx/align/core_tests.py b/lib/bx/align/core_tests.py new file mode 100644 index 0000000..bf4c572 --- /dev/null +++ b/lib/bx/align/core_tests.py @@ -0,0 +1,42 @@ +""" +Tests for `bx.align.core`. +""" + +from io import StringIO + +from bx.align.core import ( + get_indexed, + get_reader, +) + +test_maf = "test_data/maf_tests/mm10_chr12_lessspe.maf" + +axt_text = """0 chr1 11 13 chr2 21 23 + 3500 +ACG +ACT +""" + + +def test_get_reader_maf_species_to_lengths(): + species_to_lengths = {"mm10": {"chr12": 120129022}} + with open(test_maf) as f: + reader = get_reader("maf", f, species_to_lengths=species_to_lengths) + alignment = next(reader) + assert alignment.species_to_lengths is species_to_lengths + + +def test_get_reader_axt_species_to_lengths(): + species_to_lengths = {"species1": {"chr1": 100}, "species2": {"chr2": 200}} + reader = get_reader("axt", StringIO(axt_text), species_to_lengths=species_to_lengths) + alignment = next(reader) + assert alignment.species_to_lengths is species_to_lengths + + +def test_get_indexed_maf_species_to_lengths(): + species_to_lengths = {"mm10": {"chr12": 120129022}} + indexed = get_indexed("maf", test_maf, species_to_lengths=species_to_lengths) + try: + alignment = indexed.get_at_offset(0) + finally: + indexed.close() + assert alignment.species_to_lengths is species_to_lengths From 3ba2cc9b6daf7cc64d535f4bfadaadd287dc464b Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Thu, 15 Oct 2026 16:37:34 +0100 Subject: [PATCH 05/11] Fix `remove_all_gap_columns()` for alignments with empty components For an empty component, `remove_all_gap_columns()` appended `None` to `seqs`, then also tried `list(c.text)`, which raised `TypeError` (`text` is `None`) and appended a second `None`. `seqs` therefore got out of step with `self.components`: gap columns were checked against, and the new texts were assigned from, the wrong sequences, so all-gap columns could be left in the components after an empty one. E.g. for components with texts `"A-C"`, `None` (empty) and `"G-T"`, the result was `"AC"`, `None` and `"G-T"` instead of `"GT"` for the last one. Append exactly one entry to `seqs` per component, and add a test. Co-Authored-By: Claude Opus 5.5 --- lib/bx/align/core.py | 6 ++---- lib/bx/align/core_tests.py | 14 ++++++++++++++ 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/lib/bx/align/core.py b/lib/bx/align/core.py index f936e1b..b4a35ef 100644 --- a/lib/bx/align/core.py +++ b/lib/bx/align/core.py @@ -168,12 +168,10 @@ def remove_all_gap_columns(self): """ seqs = [] for c in self.components: - if c.empty: + if c.empty or c.text is None: seqs.append(None) - try: + else: seqs.append(list(c.text)) - except TypeError: - seqs.append(None) i = 0 text_size = self.text_size while i < text_size: diff --git a/lib/bx/align/core_tests.py b/lib/bx/align/core_tests.py index bf4c572..5576e5c 100644 --- a/lib/bx/align/core_tests.py +++ b/lib/bx/align/core_tests.py @@ -5,6 +5,8 @@ from io import StringIO from bx.align.core import ( + Alignment, + Component, get_indexed, get_reader, ) @@ -40,3 +42,15 @@ def test_get_indexed_maf_species_to_lengths(): finally: indexed.close() assert alignment.species_to_lengths is species_to_lengths + + +def test_remove_all_gap_columns_with_empty_component(): + alignment = Alignment() + alignment.add_component(Component("hg.chr1", 0, 2, "+", 100, "A-C")) + empty = Component("mm.chr2", 0, 0, "+", 200, None) + empty.empty = True + alignment.add_component(empty) + alignment.add_component(Component("rn.chr3", 0, 2, "+", 300, "G-T")) + alignment.remove_all_gap_columns() + assert [c.text for c in alignment.components] == ["AC", None, "GT"] + assert alignment.text_size == 2 From 47756bc5c95190f3839ed32d184beac173dc8c61 Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Thu, 15 Oct 2026 16:37:47 +0100 Subject: [PATCH 06/11] Close the lengths file in `read_lengths_file()` on errors `read_lengths_file()` only closed the file after reading it successfully, so a malformed line, which raises `ValueError`, left the file handle open. Use a `with` statement instead. Co-Authored-By: Claude Opus 5.5 --- lib/bx/misc/readlengths.py | 33 ++++++++++++++++----------------- 1 file changed, 16 insertions(+), 17 deletions(-) diff --git a/lib/bx/misc/readlengths.py b/lib/bx/misc/readlengths.py index aae4cd9..9831288 100644 --- a/lib/bx/misc/readlengths.py +++ b/lib/bx/misc/readlengths.py @@ -11,21 +11,20 @@ def read_lengths_file(name): """ chrom_to_length = {} - f = open(name) - for line in f: - line = line.strip() - if line == "" or line[0] == "#": - continue - try: - fields = line.split() - if len(fields) != 2: - raise - chrom = fields[0] - length = int(fields[1]) - except Exception: - raise ValueError(f"bad length file line: {line}") - if chrom in chrom_to_length and length != chrom_to_length[chrom]: - raise ValueError(f"{chrom} has more than one length!") - chrom_to_length[chrom] = length - f.close() + with open(name) as f: + for line in f: + line = line.strip() + if line == "" or line[0] == "#": + continue + try: + fields = line.split() + if len(fields) != 2: + raise + chrom = fields[0] + length = int(fields[1]) + except Exception: + raise ValueError(f"bad length file line: {line}") + if chrom in chrom_to_length and length != chrom_to_length[chrom]: + raise ValueError(f"{chrom} has more than one length!") + chrom_to_length[chrom] = length return chrom_to_length From 60caafb51a7e453bd1866b967b165ae6d7a49a4f Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Thu, 15 Oct 2026 16:38:45 +0100 Subject: [PATCH 07/11] Pass `memo` to the `deepcopy()` calls in `Alignment.__deepcopy__()` `Alignment.__deepcopy__()` called `deepcopy()` on its attributes and components without passing `memo` on, so each call started a new memo and objects shared within an outer deep copy were duplicated. E.g. after `deepcopy([alignment1, alignment2])`, where both alignments share the same `species_to_lengths` dict, each copy got its own copy of the dict. Pass `memo` on, as the `copy` module documentation requires, and add a test. Co-Authored-By: Claude Opus 5.5 --- lib/bx/align/core.py | 6 ++++-- lib/bx/align/core_tests.py | 13 +++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/lib/bx/align/core.py b/lib/bx/align/core.py index b4a35ef..68eb61e 100644 --- a/lib/bx/align/core.py +++ b/lib/bx/align/core.py @@ -216,10 +216,12 @@ def __deepcopy__(self, memo): from copy import deepcopy new = Alignment( - score=self.score, attributes=deepcopy(self.attributes), species_to_lengths=deepcopy(self.species_to_lengths) + score=self.score, + attributes=deepcopy(self.attributes, memo), + species_to_lengths=deepcopy(self.species_to_lengths, memo), ) for component in self.components: - new.add_component(deepcopy(component)) + new.add_component(deepcopy(component, memo)) return new diff --git a/lib/bx/align/core_tests.py b/lib/bx/align/core_tests.py index 5576e5c..325544f 100644 --- a/lib/bx/align/core_tests.py +++ b/lib/bx/align/core_tests.py @@ -2,6 +2,7 @@ Tests for `bx.align.core`. """ +from copy import deepcopy from io import StringIO from bx.align.core import ( @@ -54,3 +55,15 @@ def test_remove_all_gap_columns_with_empty_component(): alignment.remove_all_gap_columns() assert [c.text for c in alignment.components] == ["AC", None, "GT"] assert alignment.text_size == 2 + + +def test_deepcopy_preserves_shared_objects(): + species_to_lengths = {"hg": {"chr1": 100}} + alignment1 = Alignment(species_to_lengths=species_to_lengths) + alignment1.add_component(Component("hg.chr1", 0, 3, "+", None, "ACG")) + alignment2 = Alignment(species_to_lengths=species_to_lengths) + copy1, copy2 = deepcopy([alignment1, alignment2]) + assert copy1 == alignment1 + assert copy1.species_to_lengths is not species_to_lengths + assert copy1.species_to_lengths is copy2.species_to_lengths + assert copy1.components[0].src_size == 100 From 7a02fcd95b9ea65ca9e2a3cb78563bf25eeef462 Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Thu, 15 Oct 2026 16:39:16 +0100 Subject: [PATCH 08/11] Add type annotations to `bx.align.core` Annotate all of `lib/bx/align/core.py`, adding the `SpeciesToLengths` and `AlignmentFormat` type aliases, and annotate the functions in other modules that it relies on or that use it: `read_lengths_file()`, `Reader.open_seqs()` in `lav.py`, `Writer.write()` in `axt.py` and `maf.py`, and `readline()` in `maf.py`. Small runtime changes needed to make the code type-check, or found while annotating it: - `Alignment.slice_by_component()` raises `ValueError` when no component has the given `src`, instead of `AttributeError` on `None`. - `Alignment.column_iter()` and `Component.coord_to_col()` raise `ValueError` for a non-empty component without text, instead of `TypeError`. - `Component.get_src_size()` raises its usual "component has no src_size" exception when the alignment it belongs to has been garbage-collected, instead of `AttributeError` on `None`. - `maf.Writer.write()` asserts that `strand`, `synteny_empty` (for empty components) and `text` are set, instead of failing with `TypeError` in `format_tabular()`. - `get_indexed()` uses `maf.MAFIndexedAccess` instead of its deprecated `maf.Indexed` alias. - `shuffle_columns()` reads `text` once per component instead of once per column. - The error raised by `lav.Reader.open_seqs()` when it can't open the seq2 file names `seq2_filename` instead of `seq1_filename`. Also: - Add `__all__` to `bx.align.core`. As a result `from bx.align import *` (and `bx.align`, which star-imports `bx.align.core`) no longer exposes `random`, `weakref`, `read_lengths_file` or `coord_to_col`, nor the new typing imports. `coord_to_col` is still available from `bx.align.core`. - Drop the `__ne__()` methods of `Alignment` and `Component`, which were equivalent to the default implementation. - Build `species_to_lengths` directly in `lav.Reader.open_seqs()`. Co-Authored-By: Claude Opus 5.5 --- lib/bx/align/axt.py | 2 +- lib/bx/align/core.py | 204 ++++++++++++++++++++++++------------- lib/bx/align/lav.py | 21 ++-- lib/bx/align/maf.py | 9 +- lib/bx/misc/readlengths.py | 4 +- 5 files changed, 155 insertions(+), 85 deletions(-) diff --git a/lib/bx/align/axt.py b/lib/bx/align/axt.py index c0e180d..44a5347 100644 --- a/lib/bx/align/axt.py +++ b/lib/bx/align/axt.py @@ -126,7 +126,7 @@ def __init__(self, file, attributes=None): if "src_split" in attributes: self.src_split = attributes["src_split"] - def write(self, alignment): + def write(self, alignment: Alignment) -> None: if len(alignment.components) != 2: raise ValueError(f"{len(alignment.components)}-component alignment is not compatible with axt") c1 = alignment.components[0] diff --git a/lib/bx/align/core.py b/lib/bx/align/core.py index 68eb61e..4edaa59 100644 --- a/lib/bx/align/core.py +++ b/lib/bx/align/core.py @@ -4,9 +4,40 @@ import random import weakref +from collections.abc import ( + Container, + Iterator, +) +from typing import ( + Any, + cast, + IO, + Literal, + TYPE_CHECKING, + TypeAlias, +) from bx.misc.readlengths import read_lengths_file +if TYPE_CHECKING: + import bx.align.axt + import bx.align.lav + import bx.align.maf + +__all__ = [ + "Alignment", + "AlignmentFormat", + "Component", + "DNA_COMP", + "get_indexed", + "get_reader", + "get_writer", + "shuffle_columns", + "SpeciesToLengths", + "src_merge", + "src_split", +] + # DNA reverse complement table # DNA_COMP = " - " \ # " TVGH CD M KN YSA BWXR tvgh cd m kn ysa bwxr " \ @@ -15,9 +46,21 @@ DNA_COMP = str.maketrans("ACGTacgt", "TGCAtgca") +# Maps a species name to a chromosome lengths file name, to a dict mapping +# chromosome names to their length, or to a single length value +SpeciesToLengths: TypeAlias = dict[str | None, str | dict[str, int] | int] + +# Alignment file formats supported by `get_reader()` and `get_writer()` +AlignmentFormat: TypeAlias = Literal["maf", "axt", "lav"] + class Alignment: - def __init__(self, score=0, attributes=None, species_to_lengths=None): + def __init__( + self, + score: int | float | str = 0, + attributes: dict[str, Any] | None = None, + species_to_lengths: SpeciesToLengths | None = None, + ) -> None: # species_to_lengths is needed only for file formats that don't provide # chromosome lengths; it maps each species name to one of these: # - the name of a file that contains a list of chromosome length pairs @@ -30,13 +73,14 @@ def __init__(self, score=0, attributes=None, species_to_lengths=None): self.score = score self.text_size = 0 self.attributes = attributes + self.species_to_lengths: SpeciesToLengths if species_to_lengths is None: self.species_to_lengths = {} else: self.species_to_lengths = species_to_lengths - self.components = [] + self.components: list[Component] = [] - def add_component(self, component): + def add_component(self, component: "Component") -> None: component._alignment = weakref.ref(self) self.components.append(component) if component.text is not None: @@ -45,10 +89,10 @@ def add_component(self, component): elif self.text_size != len(component.text): raise Exception("Components must have same text length") - def get_score(self): + def get_score(self) -> int | float | str: return self.__score - def set_score(self, score): + def set_score(self, score: int | float | str) -> None: if isinstance(score, str): try: score = int(score) @@ -61,7 +105,7 @@ def set_score(self, score): score = property(fget=get_score, fset=set_score) - def __str__(self): + def __str__(self) -> str: s = "a score=" + str(self.score) for key in self.attributes: s += f" {key}={self.attributes[key]}" @@ -72,12 +116,12 @@ def __str__(self): s += "\n" return s - def src_size(self, src): + def src_size(self, src: str) -> int: species, chrom = src_split(src) if species in self.species_to_lengths: chrom_to_length = self.species_to_lengths[species] elif chrom in self.species_to_lengths: - chrom_to_length = self.species_to_lengths + chrom_to_length = cast(dict[str, int], self.species_to_lengths) else: raise ValueError(f"no src_size (no length file for {species})") if isinstance(chrom_to_length, int): # (if it's a single length) @@ -89,38 +133,38 @@ def src_size(self, src): raise ValueError(f"no src_size ({species} has no length for {chrom})") return chrom_to_length[chrom] - def get_component_by_src(self, src): + def get_component_by_src(self, src: str) -> "Component | None": for c in self.components: if c.src == src: return c return None - def get_components_by_src(self, src): + def get_components_by_src(self, src: str) -> Iterator["Component"]: for c in self.components: if c.src == src: yield c - def get_component_by_src_start(self, src): + def get_component_by_src_start(self, src: str) -> "Component | None": for c in self.components: if c.src.startswith(src): return c return None - def slice(self, start, end): + def slice(self, start: int, end: int) -> "Alignment": new = Alignment(score=self.score, attributes=self.attributes) for component in self.components: new.components.append(component.slice(start, end)) new.text_size = end - start return new - def reverse_complement(self): + def reverse_complement(self) -> "Alignment": new = Alignment(score=self.score, attributes=self.attributes) for component in self.components: new.components.append(component.reverse_complement()) new.text_size = self.text_size return new - def slice_by_component(self, component_index, start, end): + def slice_by_component(self, component_index: "int | str | Component", start: int, end: int) -> "Alignment": """ Return a slice of the alignment, corresponding to an coordinate interval in a specific component. @@ -133,27 +177,36 @@ def slice_by_component(self, component_index, start, end): """ if isinstance(component_index, int): - ref = self.components[component_index] + ref: Component | None = self.components[component_index] elif isinstance(component_index, str): ref = self.get_component_by_src(component_index) elif isinstance(component_index, Component): ref = component_index else: raise ValueError("can't figure out what to do") + if ref is None: + raise ValueError(f"no component with src {component_index}") start_col = ref.coord_to_col(start) end_col = ref.coord_to_col(end) if ref.strand == "-": start_col, end_col = (end_col, start_col) return self.slice(start_col, end_col) - def column_iter(self): + def column_iter(self) -> Iterator[list[str]]: # FIXME: The empty component are not present # in column_iter. # Maybe it would be good to use - and = for i in range(self.text_size): - yield [c.text[i] for c in self.components if not c.empty] + col = [] + for c in self.components: + if c.empty: + continue + if c.text is None: + raise ValueError(f"Non-empty component {c.src} has no text") + col.append(c.text[i]) + yield col - def limit_to_species(self, species): + def limit_to_species(self, species: Container[str]) -> "Alignment": new = Alignment(score=self.score, attributes=self.attributes) new.text_size = self.text_size for component in self.components: @@ -161,12 +214,12 @@ def limit_to_species(self, species): new.add_component(component) return new - def remove_all_gap_columns(self): + def remove_all_gap_columns(self) -> None: """ Remove any columns containing only gaps from alignment components, text of components is modified IN PLACE. """ - seqs = [] + seqs: list[list[str] | None] = [] for c in self.components: if c.empty or c.text is None: seqs.append(None) @@ -189,13 +242,13 @@ def remove_all_gap_columns(self): text_size -= 1 else: i += 1 - for i in range(len(self.components)): - if seqs[i] is None: + for component, seq in zip(self.components, seqs): + if seq is None: continue - self.components[i].text = "".join(seqs[i]) + component.text = "".join(seq) self.text_size = text_size - def __eq__(self, other): + def __eq__(self, other: object) -> bool: if other is None or not isinstance(other, type(self)): return False if self.score != other.score: @@ -209,10 +262,7 @@ def __eq__(self, other): return False return True - def __ne__(self, other): - return not (self.__eq__(other)) - - def __deepcopy__(self, memo): + def __deepcopy__(self, memo: dict[int, Any]) -> "Alignment": from copy import deepcopy new = Alignment( @@ -226,8 +276,16 @@ def __deepcopy__(self, memo): class Component: - def __init__(self, src="", start=0, size=0, strand=None, src_size=None, text=""): - self._alignment = None + def __init__( + self, + src: str = "", + start: int = 0, + size: int = 0, + strand: Literal["+", "-", "?"] | None = None, + src_size: int | None = None, + text: str | None = "", + ) -> None: + self._alignment: weakref.ref[Alignment] | None = None self.src = src # Nota Bene: start, size, strand are as they appear in a MAF file: # origin-zero, end excluded, and minus strand counts from end of sequence @@ -236,19 +294,19 @@ def __init__(self, src="", start=0, size=0, strand=None, src_size=None, text="") self.strand = strand self._src_size = src_size self.text = text - self.quality = None + self.quality: str | None = None # Optional fields to keep track of synteny status (only makes sense # when the alignment is part of an ordered set) - self.synteny_left = None - self.synteny_right = None - self.synteny_empty = None + self.synteny_left: tuple[str, int] | None = None + self.synteny_right: tuple[str, int] | None = None + self.synteny_empty: str | None = None # If true, this component actually represents a non-aligning region, # and text is None. self.empty = False # Index maps a coordinate (distance along + strand from + start) to alignment column - self.index = None + self.index: list[int] | None = None - def __str__(self): + def __str__(self) -> str: if self.empty: rval = f"e {self.src} {self.start} {self.size} {self.strand} {self.src_size} {self.synteny_empty}" else: @@ -257,24 +315,25 @@ def __str__(self): rval += f"\ni {self.src} {self.synteny_left[0]} {self.synteny_left[1]} {self.synteny_right[0]} {self.synteny_right[1]}" return rval - def get_end(self): + def get_end(self) -> int: return self.start + self.size end = property(fget=get_end) - def get_src_size(self): + def get_src_size(self) -> int: if self._src_size is None: - if self._alignment is None: + alignment = self._alignment() if self._alignment is not None else None + if alignment is None: raise Exception("component has no src_size") - self._src_size = self._alignment().src_size(self.src) + self._src_size = alignment.src_size(self.src) return self._src_size - def set_src_size(self, src_size): + def set_src_size(self, src_size: int | None) -> None: self._src_size = src_size src_size = property(fget=get_src_size, fset=set_src_size) - def get_forward_strand_start(self): + def get_forward_strand_start(self) -> int: if self.strand == "-": return self.src_size - self.end else: @@ -282,7 +341,7 @@ def get_forward_strand_start(self): forward_strand_start = property(fget=get_forward_strand_start) - def get_forward_strand_end(self): + def get_forward_strand_end(self) -> int: if self.strand == "-": return self.src_size - self.start else: @@ -290,15 +349,13 @@ def get_forward_strand_end(self): forward_strand_end = property(fget=get_forward_strand_end) - def reverse_complement(self): + def reverse_complement(self) -> "Component": start = self.src_size - self.end - if self.strand == "+": - strand = "-" - else: - strand = "+" + strand: Literal["+", "-"] = "-" if self.strand == "+" else "+" if self.empty: text = None else: + assert self.text is not None text = self.text.translate(DNA_COMP)[::-1] new = Component(self.src, start, self.size, strand, self._src_size, text) if self.empty: @@ -312,7 +369,7 @@ def reverse_complement(self): new._alignment = self._alignment return new - def slice(self, start, end): + def slice(self, start: int, end: int) -> "Component": new = Component(src=self.src, start=self.start, strand=self.strand, src_size=self._src_size) new._alignment = self._alignment if self.empty: @@ -321,6 +378,7 @@ def slice(self, start, end): new.text = None new.synteny_empty = self.synteny_empty return new + assert self.text is not None new.text = self.text[start:end] # for i in range( 0, start ): @@ -345,7 +403,7 @@ def slice(self, start, end): return new - def slice_by_coord(self, start, end): + def slice_by_coord(self, start: int, end: int) -> "Component": """ Return the slice of the component corresponding to a coordinate interval. @@ -358,14 +416,14 @@ def slice_by_coord(self, start, end): start_col, end_col = (end_col, start_col) return self.slice(start_col, end_col) - def coord_to_col(self, pos): + def coord_to_col(self, pos: int) -> int: """ Return the alignment column index corresponding to coordinate pos. pos is relative to the + strand, regardless of the component's strand. """ - if self.empty: + if self.empty or self.text is None: raise ValueError("There is no column index. It is empty.") start, end = self.get_forward_strand_start(), self.get_forward_strand_end() if pos < start or pos > end: @@ -386,14 +444,12 @@ def coord_to_col(self, pos): if self.text[x] != "-": self.index.append(x) self.index.append(len(self.text)) - x = None try: - x = self.index[pos - start] + return self.index[pos - start] except IndexError: raise Exception("Error in index.") - return x - def __eq__(self, other): + def __eq__(self, other: object) -> bool: if other is None or not isinstance(other, type(self)): return False return ( @@ -409,10 +465,7 @@ def __eq__(self, other): and self.empty == other.empty ) - def __ne__(self, other): - return not (self.__eq__(other)) - - def __deepcopy__(self, memo): + def __deepcopy__(self, memo: dict[int, Any]) -> "Component": new = Component( src=self.src, start=self.start, size=self.size, strand=self.strand, src_size=self._src_size, text=self.text ) @@ -426,7 +479,9 @@ def __deepcopy__(self, memo): return new -def get_reader(format, infile, species_to_lengths=None): +def get_reader( + format: AlignmentFormat, infile: IO[str], species_to_lengths: SpeciesToLengths | None = None +) -> "bx.align.maf.Reader | bx.align.axt.Reader | bx.align.lav.Reader": import bx.align.axt import bx.align.lav import bx.align.maf @@ -441,7 +496,9 @@ def get_reader(format, infile, species_to_lengths=None): raise ValueError(f"Unknown alignment format {format}") -def get_writer(format, outfile, attributes=None): +def get_writer( + format: AlignmentFormat, outfile: IO[str], attributes: dict[str, Any] | None = None +) -> "bx.align.maf.Writer | bx.align.axt.Writer | bx.align.lav.Writer": import bx.align.axt import bx.align.lav import bx.align.maf @@ -458,13 +515,19 @@ def get_writer(format, outfile, attributes=None): raise ValueError(f"Unknown alignment format {format}") -def get_indexed(format, filename, index_filename=None, keep_open=False, species_to_lengths=None): +def get_indexed( + format: AlignmentFormat, + filename: str, + index_filename: str | None = None, + keep_open: bool = False, + species_to_lengths: SpeciesToLengths | None = None, +) -> "bx.align.maf.MAFIndexedAccess | bx.align.axt.Indexed": import bx.align.axt import bx.align.lav import bx.align.maf if format == "maf": - return bx.align.maf.Indexed(filename, index_filename, keep_open, species_to_lengths=species_to_lengths) + return bx.align.maf.MAFIndexedAccess(filename, index_filename, keep_open, species_to_lengths=species_to_lengths) elif format == "axt": return bx.align.axt.Indexed(filename, index_filename, keep_open, species_to_lengths=species_to_lengths) elif format == "lav": @@ -473,16 +536,17 @@ def get_indexed(format, filename, index_filename=None, keep_open=False, species_ raise ValueError(f"Unknown alignment format {format}") -def shuffle_columns(a): +def shuffle_columns(a: Alignment) -> None: """Randomize the columns of an alignment""" mask = list(range(a.text_size)) random.shuffle(mask) for c in a.components: - if not c.empty: - c.text = "".join(c.text[i] for i in mask) + if not c.empty and c.text is not None: + text = c.text + c.text = "".join(text[i] for i in mask) -def src_split(src): # splits src into species,chrom +def src_split(src: str) -> tuple[str | None, str]: # splits src into species,chrom dot = src.rfind(".") if dot == -1: return None, src @@ -490,7 +554,7 @@ def src_split(src): # splits src into species,chrom return src[:dot], src[dot + 1 :] -def src_merge(species, chrom, contig=None): # creates src (inverse of src_split) +def src_merge(species: str | None, chrom: str, contig: int | None = None) -> str: # creates src (inverse of src_split) if species is None: src = chrom else: @@ -507,7 +571,7 @@ def src_merge(species, chrom, contig=None): # creates src (inverse of src_split from ._core import coord_to_col except ImportError: - def coord_to_col(start, text, pos): + def coord_to_col(start: int, text: str, pos: int) -> int: col = 0 while start < pos: if text[col] != "-": diff --git a/lib/bx/align/lav.py b/lib/bx/align/lav.py index 2165415..e8f8000 100644 --- a/lib/bx/align/lav.py +++ b/lib/bx/align/lav.py @@ -7,6 +7,10 @@ import sys from io import StringIO +from typing import ( + IO, + Literal, +) import bx.seq from bx.align import ( @@ -89,13 +93,13 @@ def __iter__(self): def close(self): self.file.close() - def open_seqs(self): + def open_seqs(self) -> None: if self.seq1_file is not None and self.seq2_file is not None: return if self.seq1_file is None: if self.seq1_strand == "+": - revcomp = False + revcomp: Literal[False, "-5'"] = False else: revcomp = "-5'" if self.seq1_contig == 1: @@ -103,7 +107,7 @@ def open_seqs(self): else: contig = self.seq1_contig try: - f = open(self.seq1_filename, "rb") + f: IO = open(self.seq1_filename, "rb") except Exception: if self.fail_to_ns: f = StringIO(">seq1\n" + ("n" * (self.seq1_end - self.seq1_start))) @@ -142,7 +146,7 @@ def open_seqs(self): revcomp = False contig = 1 else: - raise Exception(f"failed to open {self.seq1_filename}") + raise Exception(f"failed to open {self.seq2_filename}") self.seq2_file = bx.seq.seq_file(f, revcomp=revcomp, contig=contig) self.seq2_gap = self.seq2_file.gap try: @@ -163,11 +167,10 @@ def open_seqs(self): (species1 != species2) or (chrom1 != chrom2) or (length1 == length2) ), f"conflicting lengths for {self.seq1_src} ({length1} and {length2})" - self.species_to_lengths = {} - self.species_to_lengths[species1] = {} - self.species_to_lengths[species2] = {} # (OK if it clobbers line above) - self.species_to_lengths[species1][chrom1] = self.seq1_file.length - self.species_to_lengths[species2][chrom2] = self.seq2_file.length + self.species_to_lengths = {species1: {chrom1: length1}} + self.species_to_lengths.setdefault(species2, {})[ + chrom2 + ] = length2 # reuses the dict above if species2 == species1 def close_seqs(self): if self.seq1_file is not None: diff --git a/lib/bx/align/maf.py b/lib/bx/align/maf.py index e5478ee..5f18ee2 100644 --- a/lib/bx/align/maf.py +++ b/lib/bx/align/maf.py @@ -116,19 +116,22 @@ def __init__(self, file, attributes=None): self.file.writelines(f" {key}={attributes[key]}") self.file.write("\n") - def write(self, alignment): + def write(self, alignment: Alignment) -> None: self.file.write("a score=" + str(alignment.score)) for key in alignment.attributes: self.file.write(f" {key}={alignment.attributes[key]}") self.file.write("\n") # Components - rows = [] + rows: list[tuple[str, str, str, str, str, str, str]] = [] for c in alignment.components: # "Empty component" generates an 'e' row + assert c.strand is not None if c.empty: + assert c.synteny_empty is not None rows.append(("e", c.src, str(c.start), str(c.size), c.strand, str(c.src_size), c.synteny_empty)) continue # Regular component + assert c.text is not None rows.append(("s", c.src, str(c.start), str(c.size), c.strand, str(c.src_size), c.text)) # If component has quality, write a q row if c.quality is not None: @@ -225,7 +228,7 @@ def read_next_maf(file, species_to_lengths=None, parse_e_rows=False): return alignment -def readline(file, skip_blank=False): +def readline(file, skip_blank: bool = False) -> str | None: """Read a line from provided file, skipping any blank or comment lines""" while True: line = file.readline() diff --git a/lib/bx/misc/readlengths.py b/lib/bx/misc/readlengths.py index 9831288..9b52984 100644 --- a/lib/bx/misc/readlengths.py +++ b/lib/bx/misc/readlengths.py @@ -5,12 +5,12 @@ """ -def read_lengths_file(name): +def read_lengths_file(name: str) -> dict[str, int]: """ Returns a hash from sequence name to length. """ - chrom_to_length = {} + chrom_to_length: dict[str, int] = {} with open(name) as f: for line in f: line = line.strip() From 63502aa63a7cf8d69cd11d1419167c176cebcefb Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Thu, 15 Oct 2026 17:36:13 +0100 Subject: [PATCH 09/11] Drop support for Python 3.10 which reached end of life on 2026-10-01. --- .github/workflows/test.yaml | 4 ++-- pyproject.toml | 3 +-- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index cd862da..309118f 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -9,7 +9,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ['3.10', '3.15'] + python-version: ['3.11', '3.15'] steps: - uses: actions/checkout@v7.0.1 with: @@ -28,7 +28,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ['3.10', '3.11', '3.12', '3.13', '3.14', '3.15'] + python-version: ['3.11', '3.12', '3.13', '3.14', '3.15'] steps: - uses: actions/checkout@v7.0.1 with: diff --git a/pyproject.toml b/pyproject.toml index 7bf6ef0..baf1685 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,6 @@ classifiers = [ "Intended Audience :: Science/Research", "Operating System :: POSIX", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", @@ -38,7 +37,7 @@ classifiers = [ "Topic :: Scientific/Engineering :: Bio-Informatics", "Topic :: Software Development :: Libraries :: Python Modules", ] -requires-python = ">=3.10" +requires-python = ">=3.11" dependencies = [ "numpy", "pyparsing", From 8cfd4f2b10289f476b888de007318ac609c34d87 Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Thu, 15 Oct 2026 18:05:21 +0100 Subject: [PATCH 10/11] Temporarily drop support for PyPy3.12 NumPy 2.5.0 removed support for PyPy3.12, can be restored when https://github.com/numpy/numpy/pull/32332 is merged. --- .github/workflows/deploy.yaml | 1 - pyproject.toml | 4 ++++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/.github/workflows/deploy.yaml b/.github/workflows/deploy.yaml index 6474de8..1ca4669 100644 --- a/.github/workflows/deploy.yaml +++ b/.github/workflows/deploy.yaml @@ -33,7 +33,6 @@ jobs: run: python -m cibuildwheel --output-dir dist env: CIBW_ARCHS: ${{ matrix.archs }} - CIBW_ENABLE: "pypy" - name: Check packages run: twine check dist/* - uses: actions/upload-artifact@v7 diff --git a/pyproject.toml b/pyproject.toml index baf1685..77a4640 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,10 @@ include = '\.pyi?$' line-length = 120 [tool.cibuildwheel] +enable = ["pypy"] +# NumPy 2.5.0 removed support for PyPy3.12, can be restored when +# https://github.com/numpy/numpy/pull/32332 is merged +skip = ["pp312-*"] test-command = """ python -c 'import bx, bx.align, bx.align.axt, bx.align.core, bx.align.epo, \ bx.align.lav, bx.align.maf, bx.align.score, bx.align.sitemask, \ From 251e1e7f59591199a69c903cdc9087f416b1d25b Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Tue, 20 Oct 2026 09:50:29 +0100 Subject: [PATCH 11/11] Also test `text_size` --- lib/bx/align/maf_tests.py | 1 + 1 file changed, 1 insertion(+) diff --git a/lib/bx/align/maf_tests.py b/lib/bx/align/maf_tests.py index cf0710c..b9ec01e 100644 --- a/lib/bx/align/maf_tests.py +++ b/lib/bx/align/maf_tests.py @@ -245,6 +245,7 @@ def test_remove_all_gap_column(empty_index, repeat_component): # Return the empty row to its fixture position for the equality check. complex_maf_gap.components.append(complex_maf_gap.components.pop(empty_index)) assert complex_maf_gap == complex_maf + assert complex_maf_gap.text_size == complex_maf.text_size def test_read_with_synteny():