""" Test script for AbMelt pipeline validation """ import sys import os from pathlib import Path import logging import tempfile import time # Add src to path sys.path.insert(0, str(Path(__file__).parent / "src")) from structure_generator import StructureGenerator from gromacs_pipeline import GromacsPipeline, GromacsError from descriptor_calculator import DescriptorCalculator from ml_predictor import ThermostabilityPredictor from mdp_manager import MDPManager # Setup logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def test_structure_generation(force_fallback=False): """ Test antibody structure generation. Args: force_fallback (bool): If True, forces the use of the basic fallback generator. """ if force_fallback: logger.info("Testing structure generation (FORCING FALLBACK)...") else: logger.info("Testing structure generation (ImmuneBuilder if available)...") # Test sequences (example antibody variable regions) heavy_chain = "QVQLVQSGAEVKKPGASVKVSCKASGYTFTSYYMHWVRQAPGQGLEWMGIINPSGGSTNYAQKFQGRVTMTRDTSASTAYMELSSLRSEDTAVYYCARSTYYGGDWYFDVWGQGTLVTVSS" light_chain = "DIQMTQSPSSLSASVGDRVTITCRASQSISSYLNWYQQKPGKAPKLLIYAASSLQSGVPSRFSGSGSGTDFTLTISSLQPEDFATYYCQQSYSTPLTFGGGTKVEIK" try: generator = StructureGenerator() # Generate structure structure_path = generator.generate_structure(heavy_chain, light_chain, force_fallback=force_fallback) # Verify structure file exists if structure_path and os.path.exists(structure_path): logger.info(f"✓ Structure generated successfully: {structure_path}") # Check file size file_size = os.path.getsize(structure_path) if file_size > 1000: # Should be at least 1KB logger.info(f"✓ Structure file size reasonable: {file_size} bytes") # Add validation for NaN coordinates with open(structure_path, 'r') as f: for line in f: if "NaN" in line: logger.error(f"✗ NaN detected in generated structure file: {line.strip()}") return False, None logger.info("✓ No NaN coordinates found in structure file.") return True, structure_path else: logger.error(f"✗ Structure file too small: {file_size} bytes") return False, None else: logger.error("✗ Structure file not generated or path is None") return False, None except Exception as e: logger.error(f"✗ Structure generation failed: {e}") import traceback traceback.print_exc() return False, None finally: try: generator.cleanup() except: pass def test_gromacs_installation(): """Test if GROMACS is properly installed""" logger.info("Testing GROMACS installation...") try: pipeline = GromacsPipeline() logger.info("✓ GROMACS installation verified") return True except GromacsError as e: logger.error(f"✗ GROMACS test failed: {e}") return False except Exception as e: logger.error(f"✗ Unexpected error testing GROMACS: {e}") return False def test_ml_models(): """Test ML model loading""" logger.info("Testing ML model loading...") try: models_dir = Path(__file__).parent / "models" predictor = ThermostabilityPredictor(models_dir) model_info = predictor.get_model_info() logger.info(f"Models loaded: {model_info['models_loaded']}") logger.info(f"Available targets: {model_info['available_targets']}") if model_info['models_loaded'] > 0: logger.info("✓ ML models loaded successfully") # Test with dummy descriptors dummy_descriptors = { 'sasa_mean_300K': 120.5, 'hbonds_mean_300K': 25.3, 'rmsf_mean_300K': 0.15, 'rg_mean_300K': 2.1 } predictions = predictor.predict_thermostability(dummy_descriptors) logger.info(f"Test predictions: {predictions}") if any(pred.get('value') is not None for pred in predictions.values()): logger.info("✓ ML prediction test successful") return True else: logger.warning("⚠ ML models loaded but predictions failed") return False else: logger.error("✗ No ML models loaded") return False except Exception as e: logger.error(f"✗ ML model test failed: {e}") return False def test_quick_pipeline(): """Test a minimal pipeline run""" logger.info("Testing quick pipeline run (structure + system prep only)...") # Use shorter sequences for faster testing heavy_chain = "QVQLVQSGAEVKKPGASVKVSCKASGYTFTSYYMHWVRQAPGQGLEWMGIINPSGGSTNYAQKFQGRVTMTRDTSASTAYMELSSLRSEDTAVYYCAR" light_chain = "DIQMTQSPSSLSASVGDRVTITCRASQSISSYLNWYQQKPGKAPKLLIYAASSLQSGVPSRFSGSGSGTDFTLTISSLQPEDFATYYCQQSYST" try: # Test structure generation generator = StructureGenerator() structure_path = generator.generate_structure(heavy_chain, light_chain) if not os.path.exists(structure_path): logger.error("✗ Structure generation failed in quick test") return False # Test GROMACS system preparation (without running MD) md_pipeline = GromacsPipeline() try: # Just test the first step of system preparation prepared_system = md_pipeline.prepare_system(structure_path) if os.path.exists(prepared_system): logger.info("✓ Quick pipeline test successful") return True else: logger.error("✗ System preparation failed") return False except Exception as e: logger.error(f"✗ GROMACS pipeline failed: {e}") return False finally: md_pipeline.cleanup() except Exception as e: logger.error(f"✗ Quick pipeline test failed: {e}") return False finally: try: generator.cleanup() except: pass def test_mdp_templates(): """Test MDP template system""" logger.info("Testing MDP template system...") try: mdp_manager = MDPManager() # Check available templates templates = mdp_manager.get_available_templates() logger.info(f"Available templates: {templates}") required_templates = ['em.mdp', 'ions.mdp', 'nvt.mdp', 'npt.mdp', 'md.mdp'] missing = [t for t in required_templates if t not in templates] if missing: logger.error(f"✗ Missing required templates: {missing}") return False else: logger.info("✓ All required MDP templates found") # Test template modification test_output = tempfile.NamedTemporaryFile(suffix='.mdp', delete=False) test_output.close() try: mdp_manager.create_temperature_mdp('nvt.mdp', test_output.name, 350) # Verify temperature was changed with open(test_output.name, 'r') as f: content = f.read() if '350' in content: logger.info("✓ Template modification test successful") return True else: logger.error("✗ Template modification failed") return False finally: os.unlink(test_output.name) except Exception as e: logger.error(f"✗ MDP template test failed: {e}") return False def run_all_tests(): """Run all validation tests""" logger.info("Starting AbMelt pipeline validation tests...") results = {} # Test 1: MDP templates results['mdp_templates'] = test_mdp_templates() # Test 2: Structure generation with ImmuneBuilder (if available) results['structure_generation_immune_builder'] = test_structure_generation(force_fallback=False)[0] # Test 2.1: Structure generation with fallback generator results['structure_generation_fallback'] = test_structure_generation(force_fallback=True)[0] # Test 3: GROMACS installation results['gromacs_installation'] = test_gromacs_installation() # Test 4: ML models results['ml_models'] = test_ml_models() # Test 5: Quick pipeline if all([results['mdp_templates'], results['structure_generation_fallback'], results['gromacs_installation']]): results['quick_pipeline'] = test_quick_pipeline() else: results['quick_pipeline'] = False logger.info("Skipping quick pipeline test due to prerequisite failures") # Summary logger.info("\\n" + "="*50) logger.info("VALIDATION SUMMARY") logger.info("="*50) passed = 0 total = len(results) for test_name, result in results.items(): status = "✓ PASS" if result else "✗ FAIL" logger.info(f"{test_name:<25}: {status}") if result: passed += 1 logger.info(f"\\nOverall: {passed}/{total} tests passed") if passed == total: logger.info("🎉 All tests passed! Pipeline is ready for deployment.") return True else: logger.warning(f"⚠ {total - passed} test(s) failed. Review issues before deployment.") return False if __name__ == "__main__": success = run_all_tests() sys.exit(0 if success else 1)