seqpro

seqpro

用于快速处理生物序列的 Python/Rust 软件包。使用 Python+NumPy+Numba 处理热点循环，使用小型 Rust 扩展（src/kshuffle.rs）执行图算法操作，并使用原生 Rust 的 Ragged 数组（_core.Ragged）处理可变长度批次。导入方式：import seqpro as sp。

适用场景

• 编码/解码 DNA、RNA 或蛋白质序列（OHE、整数标记、填充）。
• 序列增强：反向互补、k-mer 打乱、抖动、随机生成。
• 序列统计：GC 含量、核苷酸组成、长度。
• 可变长度批次（例如不同大小的峰、转录本）→ sp.Ragged。
• 基因组区间 I/O：sp.bed、sp.gtf。

约定（请将以下内容载入工作记忆）

• 公共 API：完整导出列表见 python/seqpro/__init__.py。在假定某个符号存在之前，请重新查看该文件。
• 输入类型（python/seqpro/_utils.py 中的 SeqType）：str、bytes、嵌套的 str 列表，或 dtype 为 str_/object_/bytes_/uint8 的 ndarray。sp.cast_seqs(...) 会将类字符串输入规范化为 |S1 字节数组；uint8（OHE）保持不变。
• 规范的内存中 dtype：字符串序列使用 |S1，单热编码使用 uint8。
• 轴参数必须显式传入：大多数函数将 length_axis 和（对于 OHE）ohe_axis 作为位置或关键字 int 参数。允许使用负索引。check_axes() 会进行验证并尽早报错——不要捕获错误并试图掩盖。
• 库代码不得使用 Python 循环遍历序列。 热点路径使用 NumPy、_numba.py 中的 Numba 内核或 Rust 扩展。如果你想编写遍历残基的 for 循环，请先寻找已有的向量化操作或 Numba 内核。
• 字母表为单例：sp.DNA、sp.RNA、sp.AA。通过 sp.NucleotideAlphabet / sp.AminoAlphabet（python/seqpro/alphabets/_alphabets.py）构造自定义字母表。
• AminoAlphabet.translate(seqs, ..., validate=False, unknown="X")：将核苷酸翻译为氨基酸。区分大小写：小写/软屏蔽的 acgt 始终会被翻译。unknown= 控制非规范密码子（任何不属于 {A,C,G,T} 的内容）：单个字符（默认 "X"）会为每个异常密码子填充一个标记；字面值 "drop" 会移除异常密码子并返回 Ragged（即使输入为稠密格式，因为此时长度会发生变化）。validate=True 是唯一的快速失败路径——遇到 N/IUPAC/非单热编码输入时会（不区分大小写地）报错；若未报错，则保证翻译结果精确。没有单独的 error 模式。
• 变换（python/seqpro/transforms/）会将函数式操作封装为可调用对象——在数据流水线中请使用这些对象，而不是内联 lambda。

快速参考

任务	调用	说明
规范化输入	sp.cast_seqs(x)	→ `
单热编码	sp.ohe(x, alphabet, length_axis=-1)	为 OHE 维度添加末尾轴
解码 OHE	sp.decode_ohe(x, alphabet, ohe_axis=-1)
标记化 / 反标记化	sp.tokenize / sp.decode_tokens	整数 ID；parallel=True/False 强制启用/禁用并行内核（默认 None = 按大小启发式选择）
填充	sp.pad_seqs(x, pad_val, length=...)
反向互补	sp.reverse_complement(x, alphabet, length_axis=-1)	适用于 str/bytes/OHE
K-mer 打乱	sp.k_shuffle(x, k, length_axis=-1, seed=...)	调用 Rust _k_shuffle
抖动	sp.jitter(x, max_jitter, length_axis=-1)
随机序列	sp.random_seqs(shape, alphabet, seed=...)
GC 含量	sp.gc_content(x, length_axis=-1)
核苷酸含量	sp.nucleotide_content(x, alphabet, length_axis=-1)
覆盖度分箱	sp.bin_coverage(arr, bin_width, length_axis)
BED / GTF I/O	sp.bed.read_bedlike(...)、sp.gtf.read_gtf(...)	基于 polars/pyranges
哈希不规则字符串	rag.hash("sha256"|"md5"|"rapidhash")	调用 Rust 内核；仅 rapidhash 使用 seed

如需查看准确的签名和 kwargs，请直接阅读文档字符串（在 REPL 中运行 sp.<fn>?，或打开源文件——文件都很短）。

Ragged — 可变长度序列批次

sp.Ragged（由 python/seqpro/rag/_core.py 支持）是用于存储长度不同的序列批次的标准容器。它是一个原生 Rust 类，实现了 NDArrayOperatorsMixin（不是 ak.Array 的子类），支持一个不规则维度，或两个相邻的不规则维度（嵌套，例如 (batch, ploidy, None, None)），并可零拷贝访问底层扁平 NumPy 缓冲区和偏移量。

概念模型

Ragged 包含三部分：

• data：形状为 (total_elements, *fixed_trailing_dims) 的扁平连续 NDArray。可通过 rag.data 零拷贝访问。
• offsets：一个 int64 数组。形状为 (N+1,)（连续，这是常见情况）或 (2, N)（经过某些切片操作后，存储起始位置和结束位置）。可通过 rag.offsets 访问。嵌套数组的每个不规则轴都有一个这样的数组，顺序从最外层开始。
• shape：类似 (batch, None, ohe_dim) 的元组，其中一个元素为 None——这就是不规则轴；嵌套数组则有两个相邻的 None。rag.rag_dim 返回第一个不规则轴的索引。

rag.lengths 根据偏移量计算各段的长度（开销很小，返回 ndarray）。

构造

import numpy as np, seqpro as sp

# From lengths (most common — you have a flat numeric data buffer and per-segment lengths)
data = np.frombuffer(b"ACGTACGTACG", dtype=np.uint8)  # uint8 for char-as-number
lengths = np.array([4, 3, 4])
rag = sp.rag.Ragged.from_lengths(data, lengths)   # shape (3, None)

# From explicit offsets (also accepts S1 char arrays when shape has a None)
offsets = np.array([0, 4, 7, 11], dtype=np.int64)
char_data = np.frombuffer(b"ACGTACGTACG", dtype="S1")
rag = sp.rag.Ragged.from_offsets(char_data, shape=(3, None), offsets=offsets)  # shape (3, None)

# Empty with known shape
rag = sp.rag.Ragged.empty((10, None, 4), dtype=np.uint8)   # batch of 10 OHE seqs

Ragged.empty(shape, dtype) 要求 shape 中恰好有一个 None。尾部固定维度（例如 OHE 轴）应放在 None 之后。

使用 Ragged —— 应这样做，而不是那样做

任务	应做	不应做
对扁平数据执行批量数值运算	rag.data[:] = ... 或 rag.data.view(...) —— 零拷贝	迭代 for seq in rag:
应用 np.ufunc	直接调用：np.exp(rag) —— 通过 __array_ufunc__（NDArrayOperatorsMixin）分派，返回 Ragged	手动拆包并重建
统计顶层行数	len(rag) —— 返回 shape[0]（若 shape[0] 是不规则轴则会引发错误）	使用 rag.shape[0] 并手动转换为整数
索引不透明字符串 Ragged 中的一个组	rag[i] → 一个由 bytes 构成的 Ragged，每个字符串对应一个 bytes（len(rag[i]) == rag.lengths[i]）；rag[i][j] 是一个 bytes	期待得到一个拼接后的 bytes —— 这是 0.22 之前的行为，会静默丢弃每个字符串的边界
索引具有固定前导轴的嵌套数组，例如 (b, p, ~v, ~w)	对固定轴使用 NumPy 风格的键：rag[i] → (p, ~v, ~w)，rag[:, j] → (b, ~v, ~w)；rag[i, j] == rag[i][j] 会逐层剥离至 (n_v, ~w)	手动读取 rag.offsets 来查找某行的倍性
沿不规则轴或尾随轴索引	使用 NumPy/awkward 元组键，每个键对应下一个输出轴：rag[:, 0] → 每行的第一个元素（稠密；若某行过短则引发 IndexError），rag[:, 1:] / rag[:, -2:] → 每行切片（零拷贝），rag[:, :, k] 对 (b, ~, K) → 尾随列。嵌套数组也适用（对 (b, ~v, ~w) 使用 rag[:, :, 0] → 每个窗口的第一个元素）	循环遍历各行；不要期待步进切片、不规则轴上的花式键/掩码键（两个不规则轴中的第一个除外）以及记录上的尾随键可用
插入一个大小为 1 的前导轴	rag[np.newaxis] —— 返回形状为 (1, *old_shape) 的 Ragged	手动使用 from_offsets 重建
重新解释字节/数据类型	rag.view(np.uint8)	np.asarray(rag).view(...)（会丢失不规则结构）
重塑非不规则轴	rag.reshape(batch, None, k, 4)	直接修改 rag.data.shape
删除大小为 1 的轴	rag.squeeze(axis)（若压缩后变为 1D，则返回 ndarray）
转为稠密 NumPy 数组	rag.to_numpy()（根据 allow_missing 填充或引发错误）	循环并堆叠
打包到连续缓冲区	rag.to_packed() 或 sp.rag.to_packed(rag) —— 经 Numba 并行化，可安全用于 np.memmap；若已打包，使用 copy=False 可零拷贝直通	ak.to_packed(rag)
转为稠密数组并向右填充至固定长度	sp.rag.to_padded(rag, pad_value, *, length=None) —— 扁平缓冲区 numba 内核；length=None 填充至批次最大长度，显式指定 length 则填充/截断；仅支持不规则轴位于最后且非记录类型	使用 rag.to_numpy() 并手动切片，或使用 ak_str.rpad（慢约 3 倍；awkward 路径会分配额外的中间对象）
沿不规则轴拼接	sp.rag.concatenate(rags, axis) —— 沿不规则轴拼接 Ragged 数组列表（axis 必须是 None 维度；允许负值）；通过 Rust/rayon 内核进行偏移量算术和缓冲复制；支持数值数据类型（int32、float32，等等）	ak.concatenate(rags, axis=…)
转为普通 awkward	rag.to_ak()

记录布局 Ragged（多字段）

调用 sp.rag.zip（或等效的 Ragged.from_fields），并传入一个由单字段 Ragged 组成的字典来构建，这些 Ragged 必须共享同一个 offsets 对象。结果是具有记录布局的 Ragged：

import numpy as np, seqpro as sp
from seqpro.rag._utils import lengths_to_offsets

lengths = np.array([4, 3])
shared_offsets = lengths_to_offsets(lengths)

seq_rag   = sp.rag.Ragged.from_offsets(seq_flat,   shape=(2, None), offsets=shared_offsets)  # |S1
score_rag = sp.rag.Ragged.from_offsets(score_flat, shape=(2, None), offsets=shared_offsets)  # f4

batch = sp.rag.zip({"seq": seq_rag, "score": score_rag})   # → Ragged (record layout)
# equivalently: batch = sp.rag.Ragged.from_fields({"seq": seq_rag, "score": score_rag})
assert isinstance(batch, sp.rag.Ragged)

batch["score"].data[:] *= 2.0   # zero-copy mutation of the flat score buffer

输入必须共享同一个 offsets 对象（向每次 from_offsets 调用传入同一个 shared_offsets 数组）——这正是让结果成为单个 ragged 维度记录的关键。传入独立构造的 from_lengths 结果会引发 ValueError，因为它们的 offsets 对象不同，即使长度相同也一样。

• rag.dtype 返回 NumPy 结构化 dtype（例如 [("seq","S1"),("score","f4")]），仅用作描述符——内存采用 SoA，而非 AoS。
• rag.data 返回一个以字段名为键的字典，而不是单个数组。索引前务必先进行类型检查。
• rag["field"] 提供零拷贝的单字段访问，并与父对象共享 offsets 对象。它的 .data 是该字段的扁平 NumPy 缓冲区。
• 对记录布局调用 rag.to_numpy() 会返回一个 {field: dense ndarray} 字典（如果任何字段仍然是 jagged，则会引发异常——转换为稠密数组要求长度一致）。
• 记录布局不支持 view 和 apply——请逐字段操作。
• batch.to_ak() 会像 ak.zip 的默认行为一样，将记录放在叶节点：(2, None) → 2 * var * {seq, score}。在 0.23 之前，只有一个前导轴的记录会按行进行 zip（depth_limit=1）。
• 剥离一行（使用整数 i 调用 rag[i]）会返回一个字典，其中所有条目长度相同：数值/字符字段为 ndarray，不透明字符串字段为 bytes 类型的 Ragged。因此 zip(row["start"], row["alt"]) 才是正确的。

字符串哈希

使用并行 Rust 内核对 Ragged 中的每个字符串进行哈希（不透明字符串或 S1 字符叶节点，任意深度）：

digests = rag.hash("sha256")        # (N, 32) uint8, one digest per string
md5s    = rag.hash("md5")           # (N, 16) uint8
fast    = rag.hash("rapidhash")     # (N,) uint64
seeded  = rag.hash("rapidhash", seed=42)   # seed valid for rapidhash only
# equivalently: sp.rag.hash(rag, "sha256")

输出会保留字符串层级之上的结构：字符串未分组时返回常规 NumPy 数组（扁平 (N, …) / 前导固定维度）；字符串已分组时返回复用外层 offsets 的 Ragged（md5/sha256 为 (G, None, 16/32)，rapidhash 为 (G, None) uint64）。数值型和记录布局的 Ragged 会被拒绝。

NumPy 互操作——可依赖的功能

_core.Ragged 直接实现了 NDArrayOperatorsMixin 和 __array_ufunc__（不依赖 awkward）：

• 对非记录布局的 Ragged 使用 NumPy ufunc（np.add、np.exp 等）会返回 Ragged。记录布局会引发 NotImplementedError——请逐字段操作。
• rag.to_packed() / sp.rag.to_packed(rag) 是生成连续、从零开始的缓冲区的标准方式——由 Numba 并行化，在 np.memmap 上使用也安全。如果数组已打包，可使用 copy=False 进行零拷贝直通（若未打包则引发 ValueError）。
• 不要依赖 _core.Ragged 上的 awkward（ak.*）API——后端不再注册 ak.behavior。如果需要与 awkward 互操作，请使用 rag.to_ak() 获取 ak.Array，但优先使用原生 API。

如有疑问，请阅读 python/seqpro/rag/_core.py——这是当前实际使用的后端，其文档字符串是权威说明。同一目录下的 _layout.py、_ops.py 和 _utils.py 包含辅助内部实现。

常见问题

• 切片后 offsets 布局会变化。 rag.offsets 可能会变成 (2, N) 起始/结束位置，而不是 (N+1,)。对任何假定为 (N+1,) 的代码，先检查 rag.is_contiguous / 调用 rag.to_packed()。
• 记录布局上的 rag.data 是字典。 rag.data.shape 这样的代码会失败；请根据 isinstance(rag.data, dict) 分支处理，或使用 rag.parts 并检查内容。
• Ragged 的 shape 中必须有一个 None，或有两个相邻的 None（嵌套结构）。 如果从不匹配的 ragged 结构数据构造，会在 __init__ 中引发异常。如有疑问，请使用 from_lengths / from_offsets。
• Rust k-shuffle 要求数据为连续的 uint8，且最后一个轴表示序列长度。 sp.k_shuffle 会为你处理这些要求；如果直接调用 seqpro._k_shuffle，请确保布局符合要求。

查看位置（无需记忆——阅读源代码）

需求	文件
公共接口	python/seqpro/__init__.py
输入转换 / 轴辅助函数	python/seqpro/_utils.py
OHE / tokens / 填充	python/seqpro/_encoders.py
增强	python/seqpro/_modifiers.py
统计	python/seqpro/_analyzers.py
字母表	python/seqpro/alphabets/_alphabets.py
不规则数组	python/seqpro/rag/_core.py
转换（管道对象）	python/seqpro/transforms/
BED/GTF	python/seqpro/bed.py, gtf.py
Rust k-shuffle	src/kshuffle.rs
测试（用法示例）	tests/
渲染后的文档	site/（由 docs/ 构建）

禁止事项

• 不要在库代码中编写遍历残基或位置的 Python for 循环。应查找向量化操作、_numba.py 中的 Numba 内核，或扩展现有内核。
• 不要假定轴——始终显式传入 length_axis（以及相关情况下的 ohe_axis）。
• 不要在用户代码中访问 Ragged 的内部实现（_layout、_rl、__init__ 快捷方式）；请使用 data、offsets、fields、from_lengths、from_offsets、from_fields、empty。
• 不要在 Ragged 中引入字符串。仅使用 ASCII 字节（|S1）。
• 添加功能或更改公共签名时，必须更新此技能——参见 CLAUDE.md。