import sys
try:
import pyrustkmer
print("✅ 成功导入 pyrustkmer 模块")
except ImportError as e:
print(f"❌ 导入失败: {e}")
sys.exit(1)
def test_api_availability():
print("\n📋 测试 API 可用性:")
classes_to_test = [
"PyPrefixQuery",
"PyExtendedPrefixQuery",
"PyPrefixQueryMetrics",
"PyDatabase",
]
for cls_name in classes_to_test:
if hasattr(pyrustkmer, cls_name):
cls = getattr(pyrustkmer, cls_name)
print(f" ✅ {cls_name}: 可用")
methods = [
m
for m in dir(cls)
if not m.startswith("_") and callable(getattr(cls, m))
]
print(f" 方法数量: {len(methods)}")
main_methods = methods[:5] print(f" 主要方法: {', '.join(main_methods)}")
if len(methods) > 5:
print(f" ... 还有 {len(methods) - 5} 个方法")
else:
print(f" ❌ {cls_name}: 不可用")
def test_constructor_validation():
print("\n🧪 测试构造函数验证:")
try:
engine = pyrustkmer.PyPrefixQuery("/nonexistent/database.rkdb")
print(" ❌ 应该抛出异常但没有")
except Exception as e:
print(f" ✅ 正确处理不存在的文件: {type(e).__name__}")
try:
extended_engine = pyrustkmer.PyExtendedPrefixQuery(
"/nonexistent/database.rkdb"
)
print(" ❌ 应该抛出异常但没有")
except Exception as e:
print(f" ✅ 扩展引擎也正确处理错误: {type(e).__name__}")
def test_pattern_parsing():
print("\n🔍 测试模式解析功能:")
test_patterns = ["AAAAAAAA{N5}AAAAAA", "ATCG{N3}GCTA", "AAAAAAA", "N{N10}N"]
for pattern in test_patterns:
print(f"\n 测试模式: {pattern}")
try:
engine = pyrustkmer.PyPrefixQuery("/tmp/test.db")
info = engine.parse_pattern(pattern)
print(f" ✅ 解析成功:")
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" 📝 预期的解析错误: {type(e).__name__}")
def test_method_signatures():
print("\n🔧 测试方法签名:")
if hasattr(pyrustkmer, "PyPrefixQuery"):
cls = pyrustkmer.PyPrefixQuery
methods_to_check = [
"query_prefix_string",
"query_hybrid",
"parse_pattern",
"database_info",
]
for method_name in methods_to_check:
if hasattr(cls, method_name):
method = getattr(cls, method_name)
print(f" ✅ {method_name}: 方法存在")
else:
print(f" ❌ {method_name}: 方法不存在")
if hasattr(pyrustkmer, "PyExtendedPrefixQuery"):
cls = pyrustkmer.PyExtendedPrefixQuery
methods_to_check = [
"query_prefix_metrics",
"query_hybrid_metrics",
"query_prefix_batch_metrics",
]
for method_name in methods_to_check:
if hasattr(cls, method_name):
method = getattr(cls, method_name)
print(f" ✅ {method_name}: 方法存在")
else:
print(f" ❌ {method_name}: 方法不存在")
def show_usage_examples():
print("\n📖 使用示例:")
examples = [
{
"title": "基础用法",
"code": '''
import pyrustkmer
# 创建查询引擎
engine = pyrustkmer.PyPrefixQuery("path/to/database.rkdb")
# 前缀查询
results = engine.query_prefix_string("AAAAAAAA")
# 混合搜索
results = engine.query_hybrid("AAAAAAAA{N5}AAAAAA")
# 显示结果
for kmer, count in results.items():
print(f"{kmer}: {count}")
""",
},
{
"title": "带指标的查询",
"code": '''
import pyrustkmer
extended_engine = pyrustkmer.PyExtendedPrefixQuery("path/to/database.rkdb")
metrics = extended_engine.query_prefix_metrics("AAAAAAAA")
print(f"执行时间: {metrics.execution_time_ms} ms")
print(f"总匹配数: {metrics.total_matches}")
results = dict(metrics.results)
""",
},
{
"title": "批量查询",
"code": '''
import pyrustkmer
engine = pyrustkmer.PyExtendedPrefixQuery("path/to/database.rkdb")
# 批量查询多个模式
patterns = ["AAAAAAAA", "AAAAAAA{N5}AAAAAA", "ATCG{N3}GCTA"]
results = engine.query_prefix_batch_metrics(patterns)
for pattern, metrics in results.items():
print(f"模式 '{pattern}': {metrics.total_matches} 个结果")
""",
},
]
for i, example in enumerate(examples, 1):
print(f"\n 示例 {i}: {example['title']}")
print(" " + "-" * 40)
for line in example["code"].strip().split("\n"):
print(f" {line}")
def main():
print("🚀 PyO3 Python Binding API 测试")
print("=" * 50)
test_api_availability()
test_constructor_validation()
test_pattern_parsing()
test_method_signatures()
show_usage_examples()
print("\n" + "=" * 50)
print("✅ PyO3 API 测试完成!")
print(f"\n💡 下一步:")
print(f" 1. 使用真实数据库文件测试:")
print(
f" python3 simple_prefix_query.py ~/Data/data/kmer/K19/R1_001.rkdb 'AAAAAAA{{N5}}AAAAAA'"
)
print(f" 2. 查看完整文档:")
print(f" PYO3_PYTHON_BINDING_GUIDE.md")
print(f" 3. 运行完整演示:")
print(f" python3 demo_pyo3_binding.py")
if __name__ == "__main__":
main()