import sys
import os
import tempfile
import subprocess
sys.path.insert(0, "/Users/forrest/Github/rustkmer/examples/python")
try:
import pyrustkmer
PYO3_AVAILABLE = True
except ImportError:
PYO3_AVAILABLE = False
print("⚠️ PyO3 扩展未加载，请先构建扩展")
def create_test_database():
print("🔧 创建测试数据库...")
fasta_content = """>test_sequence_1
AAAAAAAAAAATTTTTTTTTTCCCCCCCCCCGGGGGGGGGG
>test_sequence_2
ATCGATCGATCGATCGATCGATCGATCGATCGATCGATCG
>test_sequence_3
AAAAAAAAAAACCCCCCCCGGGGGGGGGGTTTTTTTTTT
"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".fasta", delete=False) as f:
f.write(fasta_content)
fasta_file = f.name
db_file = tempfile.NamedTemporaryFile(suffix=".rkdb", delete=False)
db_file.close()
try:
cmd = [
"/Users/forrest/Github/rustkmer/target/release/rustkmer",
"build-kmer-db",
fasta_file,
db_file.name,
"--kmer-size",
"19",
"--load-mode",
"memory",
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
print(f"✅ 测试数据库创建成功: {db_file.name}")
return db_file.name
else:
print(f"❌ 数据库创建失败: {result.stderr}")
return None
except Exception as e:
print(f"❌ 创建数据库时出错: {e}")
return None
finally:
os.unlink(fasta_file)
def demo_basic_usage():
print("\n" + "=" * 60)
print("🎯 PyO3 Binding 基础使用演示")
print("=" * 60)
if not PYO3_AVAILABLE:
print("❌ PyO3 扩展不可用，跳过演示")
return
print("\n📋 可用的 PyO3 类:")
classes = [x for x in dir(pyrustkmer) if not x.startswith("_") and x[0].isupper()]
for cls in classes:
print(f" • {cls}")
db_path = create_test_database()
if not db_path:
print("❌ 无法创建测试数据库，跳过实际演示")
return
try:
print(f"\n🔍 初始化查询引擎...")
engine = pyrustkmer.PyPrefixQuery(db_path)
print("✅ 查询引擎创建成功!")
print(f"\n📊 数据库信息:")
db_info = engine.database_info()
for key, value in db_info.items():
print(f" {key}: {value}")
print(f"\n🔤 前缀查询演示:")
prefix_patterns = ["AAAAAAAAA", "ATCGATCGAT"]
for pattern in prefix_patterns:
try:
results = engine.query_prefix_string(pattern)
print(f" 模式 '{pattern}': 找到 {len(results)} 个结果")
if results:
sample = list(results.items())[:3]
for kmer, count in sample:
print(f" {kmer}: {count}")
if len(results) > 3:
print(f" ... 还有 {len(results) - 3} 个结果")
except Exception as e:
print(f" 模式 '{pattern}': 查询失败 - {e}")
print(f"\n🎯 混合搜索演示:")
hybrid_patterns = ["AAAAA{N5}GGGGG", "ATCG{N3}GCTA"]
for pattern in hybrid_patterns:
try:
pattern_info = engine.parse_pattern(pattern)
print(f" 模式解析 '{pattern}':")
print(f" 前缀: '{pattern_info['prefix']}'")
print(f" 后缀: '{pattern_info['suffix']}'")
print(f" N数量: {pattern_info['n_count']}")
results = engine.query_hybrid(pattern)
print(f" 查询结果: 找到 {len(results)} 个结果")
if results:
sample = list(results.items())[:2]
for kmer, count in sample:
print(f" {kmer}: {count}")
except Exception as e:
print(f" 模式 '{pattern}': 查询失败 - {e}")
print(f"\n📈 扩展查询引擎演示:")
try:
extended_engine = pyrustkmer.PyExtendedPrefixQuery(db_path)
metrics = extended_engine.query_prefix_metrics("AAAAAAAAA")
print(f" 前缀查询指标:")
print(f" 执行时间: {metrics.execution_time_ms} ms")
print(f" 总匹配数: {metrics.total_matches}")
print(f" 内存块大小: {metrics.block_size}")
except Exception as e:
print(f" 扩展引擎演示失败: {e}")
finally:
try:
os.unlink(db_path)
print(f"\n🧹 清理测试数据库: {db_path}")
except:
pass
def demo_command_equivalence():
print("\n" + "=" * 60)
print("🔄 命令等价性演示")
print("=" * 60)
print("\n📝 Rust CLI 命令:")
print(
" ./target/release/rustkmer prefix-query ~/Data/data/kmer/K19/R1_001.rkdb AAAAAAAA{N5}AAAAAA"
)
print("\n🐍 Python PyO3 等价代码:")
print("""
import pyrustkmer
# 创建查询引擎
engine = pyrustkmer.PyPrefixQuery("~/Data/data/kmer/K19/R1_001.rkdb")
# 执行混合搜索
results = engine.query_hybrid("AAAAAAAA{N5}AAAAAA")
# 处理结果
for kmer, count in results.items():
print(f"{kmer}: {count}")
""")
print("\n📋 简化脚本使用:")
print(
" python3 simple_prefix_query.py ~/Data/data/kmer/K19/R1_001.rkdb 'AAAAAAAA{N5}AAAAAA'"
)
print("\n🎛️ 完整脚本使用:")
print(
" python3 prefix_query_pyo3_binding.py ~/Data/data/kmer/K19/R1_001.rkdb 'AAAAAAAA{N5}AAAAAA' --with-metrics"
)
def demo_pattern_syntax():
print("\n" + "=" * 60)
print("📝 模式语法演示")
print("=" * 60)
patterns = {
"纯前缀": "AAAAAAAA",
"中等前缀": "ATCGATCGATCG",
"短前缀+混合": "A{N5}T",
"长前缀+混合": "ATCGATCG{N3}GCTA",
"复杂混合": "AAAAA{N2}TTTTT{N2}GGGGG",
"全N": "NNNNNNNNNNNNNNNNNNN",
}
print("\n🔍 支持的模式类型:")
for desc, pattern in patterns.items():
print(f" {desc}: {pattern}")
if PYO3_AVAILABLE:
db_path = create_test_database()
if db_path:
try:
engine = pyrustkmer.PyPrefixQuery(db_path)
print(f"\n🧪 模式解析演示:")
for desc, pattern in list(patterns.items())[:3]: try:
info = engine.parse_pattern(pattern)
print(f" {desc} ({pattern}):")
print(f" 前缀: '{info['prefix']}'")
print(f" 后缀: '{info['suffix']}'")
print(f" N数量: {info['n_count']}")
print(f" 总长度: {info['total_length']}")
except Exception as e:
print(f" {desc} ({pattern}): 解析失败 - {e}")
finally:
try:
os.unlink(db_path)
except:
pass
def main():
print("🚀 PyO3 Python Binding 完整演示")
print("=" * 60)
if not PYO3_AVAILABLE:
print("⚠️ PyO3 扩展未加载")
print("💡 请先运行以下命令构建扩展:")
print(" cd rustkmer/pyo3")
print(" export RUSTFLAGS='-C link-arg=-undefined -C link-arg=dynamic_lookup'")
print(" export PYO3_PYTHON=/usr/bin/python3")
print(" cargo build")
print(" export PYTHONPATH='$PWD/target/debug:$PYTHONPATH'")
return
demo_basic_usage()
demo_command_equivalence()
demo_pattern_syntax()
print("\n" + "=" * 60)
print("✅ PyO3 Binding 演示完成!")
print("=" * 60)
print(f"\n📚 更多信息请参考:")
print(f" • PYO3_PYTHON_BINDING_GUIDE.md - 完整使用指南")
print(f" • README.md - 项目总览")
print(f" • PREFIX_QUERY_USAGE_GUIDE.md - 前缀查询指南")
if __name__ == "__main__":
main()