Spaces:
Build error
Build error
File size: 4,317 Bytes
8ef403e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 |
#!/usr/bin/env python3
"""
Standalone integration test for timing module.
Demonstrates the full timing flow without heavy pipeline dependencies.
"""
import sys
import time
import json
import tempfile
import os
from pathlib import Path
# Add src to path
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from timing import (
TimingReport,
get_timing_report,
reset_timing_report,
time_step
)
def simulate_pipeline():
"""Simulate the AbMelt pipeline with timing."""
print("=" * 60)
print("SIMULATED PIPELINE WITH TIMING")
print("=" * 60)
# Initialize timing (like in infer.py main())
reset_timing_report()
timing_report = get_timing_report()
timing_report.start()
# Simulate Step 1: Structure Preparation
with time_step("Structure Preparation"):
print("Step 1: Simulating structure preparation...")
time.sleep(0.2)
# Simulate Step 2: MD Simulation with sub-steps
with time_step("MD Simulation"):
print("Step 2: Simulating MD simulation...")
with time_step("GROMACS Preprocessing", parent="MD Simulation"):
time.sleep(0.1)
with time_step("System Setup", parent="MD Simulation"):
time.sleep(0.1)
with time_step("Multi-Temp Simulations", parent="MD Simulation"):
time.sleep(0.3)
with time_step("Trajectory Processing", parent="MD Simulation"):
time.sleep(0.1)
# Simulate Step 3: Descriptor Computation with sub-steps
with time_step("Descriptor Computation"):
print("Step 3: Simulating descriptor computation...")
with time_step("GROMACS Descriptors", parent="Descriptor Computation"):
time.sleep(0.1)
with time_step("Order Parameters", parent="Descriptor Computation"):
time.sleep(0.1)
with time_step("Core/Surface SASA", parent="Descriptor Computation"):
time.sleep(0.05)
with time_step("Lambda Features", parent="Descriptor Computation"):
time.sleep(0.05)
with time_step("Aggregate to DataFrame", parent="Descriptor Computation"):
time.sleep(0.05)
# Simulate Step 4: Model Inference
with time_step("Model Inference"):
print("Step 4: Simulating model inference...")
time.sleep(0.1)
# Stop timing
timing_report.stop()
# Print summary (like at end of infer.py main())
print(timing_report.format_summary())
# Save to JSON
with tempfile.TemporaryDirectory() as tmpdir:
json_path = os.path.join(tmpdir, "timing_test.json")
timing_report.save_json(json_path)
# Verify JSON was created correctly
with open(json_path, 'r') as f:
data = json.load(f)
print(f"\nJSON report saved successfully!")
print(f" Total entries: {len(data['entries'])}")
print(f" Total time: {data['total_duration_formatted']}")
return timing_report
def main():
"""Run the integration test."""
print("\n" + "=" * 60)
print("TIMING MODULE INTEGRATION TEST")
print("=" * 60 + "\n")
report = simulate_pipeline()
# Verify structure
expected_steps = [
"Structure Preparation",
"MD Simulation",
"GROMACS Preprocessing",
"System Setup",
"Multi-Temp Simulations",
"Trajectory Processing",
"Descriptor Computation",
"GROMACS Descriptors",
"Order Parameters",
"Core/Surface SASA",
"Lambda Features",
"Aggregate to DataFrame",
"Model Inference"
]
all_found = True
for step in expected_steps:
if step not in report.entries:
print(f"✗ Missing step: {step}")
all_found = False
if all_found:
print(f"\n✓ All {len(expected_steps)} timing entries present!")
print("✓ Integration test PASSED!")
return 0
else:
print("\n✗ Integration test FAILED!")
return 1
if __name__ == "__main__":
sys.exit(main())
|