Update memory: Camie Tagger v2 upgrade results
This commit is contained in:
root
25
skills/ml-pipeline/scripts/README.md
Normal file
25
skills/ml-pipeline/scripts/README.md
Normal file
@@ -0,0 +1,25 @@
|
||||
# Scripts -- ml-feature-engineering
|
||||
|
||||
Bundled utility scripts for the consolidated ML Feature Engineering skill.
|
||||
|
||||
## Pipeline Automation (from automl-pipeline-builder-2)
|
||||
|
||||
- **data_validation.py** -- Validate input data quality before pipeline execution.
|
||||
- **model_evaluation.py** -- Evaluate trained model performance and generate reports.
|
||||
- **pipeline_deployment.py** -- Deploy a trained pipeline with rollback support.
|
||||
|
||||
## Feature Engineering (from machine-learning-feature-engineering-toolkit)
|
||||
|
||||
- **feature_engineering_pipeline.py** -- End-to-end feature engineering process.
|
||||
- **feature_importance_analyzer.py** -- Analyse feature importance (permutation, SHAP, tree-based).
|
||||
- **data_visualizer.py** -- Visualise feature distributions and target relationships.
|
||||
- **feature_store_integration.py** -- Integrate with feature stores (Feast, Tecton).
|
||||
|
||||
## Usage
|
||||
|
||||
All scripts accept `--help` for argument documentation:
|
||||
|
||||
```bash
|
||||
python scripts/data_validation.py --help
|
||||
python scripts/feature_importance_analyzer.py --help
|
||||
```
|
||||
100
skills/ml-pipeline/scripts/data_validation.py
Normal file
100
skills/ml-pipeline/scripts/data_validation.py
Normal file
@@ -0,0 +1,100 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
automl-pipeline-builder - data_validation.py
|
||||
Script to validate input data for the AutoML pipeline, ensuring data quality and preventing errors.
|
||||
Generated: 2025-12-10 03:48:17
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
def process_file(file_path: Path) -> bool:
|
||||
"""Process individual file."""
|
||||
if not file_path.exists():
|
||||
print(f"❌ File not found: {file_path}")
|
||||
return False
|
||||
|
||||
print(f"📄 Processing: {file_path}")
|
||||
|
||||
# Add processing logic here based on skill requirements
|
||||
# This is a template that can be customized
|
||||
|
||||
try:
|
||||
if file_path.suffix == '.json':
|
||||
with open(file_path) as f:
|
||||
data = json.load(f)
|
||||
print(f" ✓ Valid JSON with {len(data)} keys")
|
||||
else:
|
||||
size = file_path.stat().st_size
|
||||
print(f" ✓ File size: {size:,} bytes")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Error: {e}")
|
||||
return False
|
||||
|
||||
def process_directory(dir_path: Path) -> int:
|
||||
"""Process all files in directory."""
|
||||
processed = 0
|
||||
failed = 0
|
||||
|
||||
for file_path in dir_path.rglob('*'):
|
||||
if file_path.is_file():
|
||||
if process_file(file_path):
|
||||
processed += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
return processed, failed
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Script to validate input data for the AutoML pipeline, ensuring data quality and preventing errors."
|
||||
)
|
||||
parser.add_argument('input', help='Input file or directory')
|
||||
parser.add_argument('--output', '-o', help='Output directory')
|
||||
parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
|
||||
parser.add_argument('--config', '-c', help='Configuration file')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
input_path = Path(args.input)
|
||||
|
||||
print(f"🚀 automl-pipeline-builder - data_validation.py")
|
||||
print(f" Category: ai-ml")
|
||||
print(f" Plugin: automl-pipeline-builder")
|
||||
print(f" Input: {input_path}")
|
||||
|
||||
if args.config:
|
||||
if Path(args.config).exists():
|
||||
with open(args.config) as f:
|
||||
config = json.load(f)
|
||||
print(f" Config: {args.config}")
|
||||
|
||||
# Process input
|
||||
if input_path.is_file():
|
||||
success = process_file(input_path)
|
||||
result = 0 if success else 1
|
||||
elif input_path.is_dir():
|
||||
processed, failed = process_directory(input_path)
|
||||
print(f"\n📊 SUMMARY")
|
||||
print(f" ✅ Processed: {processed}")
|
||||
print(f" ❌ Failed: {failed}")
|
||||
result = 0 if failed == 0 else 1
|
||||
else:
|
||||
print(f"❌ Invalid input: {input_path}")
|
||||
result = 1
|
||||
|
||||
if result == 0:
|
||||
print("\n✅ Completed successfully")
|
||||
else:
|
||||
print("\n❌ Completed with errors")
|
||||
|
||||
return result
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
100
skills/ml-pipeline/scripts/data_visualizer.py
Normal file
100
skills/ml-pipeline/scripts/data_visualizer.py
Normal file
@@ -0,0 +1,100 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
feature-engineering-toolkit - data_visualizer.py
|
||||
Generates visualizations of features and their relationships to the target variable, aiding in understanding data patterns.
|
||||
Generated: 2025-12-10 03:48:17
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
def process_file(file_path: Path) -> bool:
|
||||
"""Process individual file."""
|
||||
if not file_path.exists():
|
||||
print(f"❌ File not found: {file_path}")
|
||||
return False
|
||||
|
||||
print(f"📄 Processing: {file_path}")
|
||||
|
||||
# Add processing logic here based on skill requirements
|
||||
# This is a template that can be customized
|
||||
|
||||
try:
|
||||
if file_path.suffix == '.json':
|
||||
with open(file_path) as f:
|
||||
data = json.load(f)
|
||||
print(f" ✓ Valid JSON with {len(data)} keys")
|
||||
else:
|
||||
size = file_path.stat().st_size
|
||||
print(f" ✓ File size: {size:,} bytes")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Error: {e}")
|
||||
return False
|
||||
|
||||
def process_directory(dir_path: Path) -> int:
|
||||
"""Process all files in directory."""
|
||||
processed = 0
|
||||
failed = 0
|
||||
|
||||
for file_path in dir_path.rglob('*'):
|
||||
if file_path.is_file():
|
||||
if process_file(file_path):
|
||||
processed += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
return processed, failed
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Generates visualizations of features and their relationships to the target variable, aiding in understanding data patterns."
|
||||
)
|
||||
parser.add_argument('input', help='Input file or directory')
|
||||
parser.add_argument('--output', '-o', help='Output directory')
|
||||
parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
|
||||
parser.add_argument('--config', '-c', help='Configuration file')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
input_path = Path(args.input)
|
||||
|
||||
print(f"🚀 feature-engineering-toolkit - data_visualizer.py")
|
||||
print(f" Category: ai-ml")
|
||||
print(f" Plugin: feature-engineering-toolkit")
|
||||
print(f" Input: {input_path}")
|
||||
|
||||
if args.config:
|
||||
if Path(args.config).exists():
|
||||
with open(args.config) as f:
|
||||
config = json.load(f)
|
||||
print(f" Config: {args.config}")
|
||||
|
||||
# Process input
|
||||
if input_path.is_file():
|
||||
success = process_file(input_path)
|
||||
result = 0 if success else 1
|
||||
elif input_path.is_dir():
|
||||
processed, failed = process_directory(input_path)
|
||||
print(f"\n📊 SUMMARY")
|
||||
print(f" ✅ Processed: {processed}")
|
||||
print(f" ❌ Failed: {failed}")
|
||||
result = 0 if failed == 0 else 1
|
||||
else:
|
||||
print(f"❌ Invalid input: {input_path}")
|
||||
result = 1
|
||||
|
||||
if result == 0:
|
||||
print("\n✅ Completed successfully")
|
||||
else:
|
||||
print("\n❌ Completed with errors")
|
||||
|
||||
return result
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
100
skills/ml-pipeline/scripts/feature_engineering_pipeline.py
Normal file
100
skills/ml-pipeline/scripts/feature_engineering_pipeline.py
Normal file
@@ -0,0 +1,100 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
feature-engineering-toolkit - feature_engineering_pipeline.py
|
||||
Automates the entire feature engineering process, including data loading, cleaning, transformation, selection, and model training.
|
||||
Generated: 2025-12-10 03:48:17
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
def process_file(file_path: Path) -> bool:
|
||||
"""Process individual file."""
|
||||
if not file_path.exists():
|
||||
print(f"❌ File not found: {file_path}")
|
||||
return False
|
||||
|
||||
print(f"📄 Processing: {file_path}")
|
||||
|
||||
# Add processing logic here based on skill requirements
|
||||
# This is a template that can be customized
|
||||
|
||||
try:
|
||||
if file_path.suffix == '.json':
|
||||
with open(file_path) as f:
|
||||
data = json.load(f)
|
||||
print(f" ✓ Valid JSON with {len(data)} keys")
|
||||
else:
|
||||
size = file_path.stat().st_size
|
||||
print(f" ✓ File size: {size:,} bytes")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Error: {e}")
|
||||
return False
|
||||
|
||||
def process_directory(dir_path: Path) -> int:
|
||||
"""Process all files in directory."""
|
||||
processed = 0
|
||||
failed = 0
|
||||
|
||||
for file_path in dir_path.rglob('*'):
|
||||
if file_path.is_file():
|
||||
if process_file(file_path):
|
||||
processed += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
return processed, failed
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Automates the entire feature engineering process, including data loading, cleaning, transformation, selection, and model training."
|
||||
)
|
||||
parser.add_argument('input', help='Input file or directory')
|
||||
parser.add_argument('--output', '-o', help='Output directory')
|
||||
parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
|
||||
parser.add_argument('--config', '-c', help='Configuration file')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
input_path = Path(args.input)
|
||||
|
||||
print(f"🚀 feature-engineering-toolkit - feature_engineering_pipeline.py")
|
||||
print(f" Category: ai-ml")
|
||||
print(f" Plugin: feature-engineering-toolkit")
|
||||
print(f" Input: {input_path}")
|
||||
|
||||
if args.config:
|
||||
if Path(args.config).exists():
|
||||
with open(args.config) as f:
|
||||
config = json.load(f)
|
||||
print(f" Config: {args.config}")
|
||||
|
||||
# Process input
|
||||
if input_path.is_file():
|
||||
success = process_file(input_path)
|
||||
result = 0 if success else 1
|
||||
elif input_path.is_dir():
|
||||
processed, failed = process_directory(input_path)
|
||||
print(f"\n📊 SUMMARY")
|
||||
print(f" ✅ Processed: {processed}")
|
||||
print(f" ❌ Failed: {failed}")
|
||||
result = 0 if failed == 0 else 1
|
||||
else:
|
||||
print(f"❌ Invalid input: {input_path}")
|
||||
result = 1
|
||||
|
||||
if result == 0:
|
||||
print("\n✅ Completed successfully")
|
||||
else:
|
||||
print("\n❌ Completed with errors")
|
||||
|
||||
return result
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
134
skills/ml-pipeline/scripts/feature_importance_analyzer.py
Normal file
134
skills/ml-pipeline/scripts/feature_importance_analyzer.py
Normal file
@@ -0,0 +1,134 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
feature-engineering-toolkit - Analysis Script
|
||||
Analyzes feature importance using various techniques (e.g., permutation importance, SHAP values) and provides insights into which features are most influential.
|
||||
Generated: 2025-12-10 03:48:17
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from typing import Dict, List
|
||||
from datetime import datetime
|
||||
|
||||
class Analyzer:
|
||||
def __init__(self, target_path: str):
|
||||
self.target_path = Path(target_path)
|
||||
self.stats = {
|
||||
'total_files': 0,
|
||||
'total_size': 0,
|
||||
'file_types': {},
|
||||
'issues': [],
|
||||
'recommendations': []
|
||||
}
|
||||
|
||||
def analyze_directory(self) -> Dict:
|
||||
"""Analyze directory structure and contents."""
|
||||
if not self.target_path.exists():
|
||||
self.stats['issues'].append(f"Path does not exist: {self.target_path}")
|
||||
return self.stats
|
||||
|
||||
for file_path in self.target_path.rglob('*'):
|
||||
if file_path.is_file():
|
||||
self.analyze_file(file_path)
|
||||
|
||||
return self.stats
|
||||
|
||||
def analyze_file(self, file_path: Path):
|
||||
"""Analyze individual file."""
|
||||
self.stats['total_files'] += 1
|
||||
self.stats['total_size'] += file_path.stat().st_size
|
||||
|
||||
# Track file types
|
||||
ext = file_path.suffix.lower()
|
||||
if ext:
|
||||
self.stats['file_types'][ext] = self.stats['file_types'].get(ext, 0) + 1
|
||||
|
||||
# Check for potential issues
|
||||
if file_path.stat().st_size > 100 * 1024 * 1024: # 100MB
|
||||
self.stats['issues'].append(f"Large file: {file_path} ({file_path.stat().st_size // 1024 // 1024}MB)")
|
||||
|
||||
if file_path.stat().st_size == 0:
|
||||
self.stats['issues'].append(f"Empty file: {file_path}")
|
||||
|
||||
def generate_recommendations(self):
|
||||
"""Generate recommendations based on analysis."""
|
||||
if self.stats['total_files'] == 0:
|
||||
self.stats['recommendations'].append("No files found - check target path")
|
||||
|
||||
if len(self.stats['file_types']) > 20:
|
||||
self.stats['recommendations'].append("Many file types detected - consider organizing")
|
||||
|
||||
if self.stats['total_size'] > 1024 * 1024 * 1024: # 1GB
|
||||
self.stats['recommendations'].append("Large total size - consider archiving old data")
|
||||
|
||||
def generate_report(self) -> str:
|
||||
"""Generate analysis report."""
|
||||
report = []
|
||||
report.append("\n" + "="*60)
|
||||
report.append(f"ANALYSIS REPORT - feature-engineering-toolkit")
|
||||
report.append("="*60)
|
||||
report.append(f"Target: {self.target_path}")
|
||||
report.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
report.append("")
|
||||
|
||||
# Statistics
|
||||
report.append("📊 STATISTICS")
|
||||
report.append(f" Total Files: {self.stats['total_files']:,}")
|
||||
report.append(f" Total Size: {self.stats['total_size'] / 1024 / 1024:.2f} MB")
|
||||
report.append(f" File Types: {len(self.stats['file_types'])}")
|
||||
|
||||
# Top file types
|
||||
if self.stats['file_types']:
|
||||
report.append("\n📁 TOP FILE TYPES")
|
||||
sorted_types = sorted(self.stats['file_types'].items(), key=lambda x: x[1], reverse=True)[:5]
|
||||
for ext, count in sorted_types:
|
||||
report.append(f" {ext or 'no extension'}: {count} files")
|
||||
|
||||
# Issues
|
||||
if self.stats['issues']:
|
||||
report.append(f"\n⚠️ ISSUES ({len(self.stats['issues'])})")
|
||||
for issue in self.stats['issues'][:10]:
|
||||
report.append(f" - {issue}")
|
||||
if len(self.stats['issues']) > 10:
|
||||
report.append(f" ... and {len(self.stats['issues']) - 10} more")
|
||||
|
||||
# Recommendations
|
||||
if self.stats['recommendations']:
|
||||
report.append("\n💡 RECOMMENDATIONS")
|
||||
for rec in self.stats['recommendations']:
|
||||
report.append(f" - {rec}")
|
||||
|
||||
report.append("")
|
||||
return "\n".join(report)
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Analyzes feature importance using various techniques (e.g., permutation importance, SHAP values) and provides insights into which features are most influential.")
|
||||
parser.add_argument('target', help='Target directory to analyze')
|
||||
parser.add_argument('--output', '-o', help='Output report file')
|
||||
parser.add_argument('--json', action='store_true', help='Output as JSON')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
print(f"🔍 Analyzing {args.target}...")
|
||||
analyzer = Analyzer(args.target)
|
||||
stats = analyzer.analyze_directory()
|
||||
analyzer.generate_recommendations()
|
||||
|
||||
if args.json:
|
||||
output = json.dumps(stats, indent=2)
|
||||
else:
|
||||
output = analyzer.generate_report()
|
||||
|
||||
if args.output:
|
||||
Path(args.output).write_text(output)
|
||||
print(f"✓ Report saved to {args.output}")
|
||||
else:
|
||||
print(output)
|
||||
|
||||
return 0 if len(stats['issues']) == 0 else 1
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
sys.exit(main())
|
||||
100
skills/ml-pipeline/scripts/feature_store_integration.py
Normal file
100
skills/ml-pipeline/scripts/feature_store_integration.py
Normal file
@@ -0,0 +1,100 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
feature-engineering-toolkit - feature_store_integration.py
|
||||
Integrates with feature stores (e.g., Feast, Tecton) to manage and serve features for online and offline model deployment.
|
||||
Generated: 2025-12-10 03:48:17
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
def process_file(file_path: Path) -> bool:
|
||||
"""Process individual file."""
|
||||
if not file_path.exists():
|
||||
print(f"❌ File not found: {file_path}")
|
||||
return False
|
||||
|
||||
print(f"📄 Processing: {file_path}")
|
||||
|
||||
# Add processing logic here based on skill requirements
|
||||
# This is a template that can be customized
|
||||
|
||||
try:
|
||||
if file_path.suffix == '.json':
|
||||
with open(file_path) as f:
|
||||
data = json.load(f)
|
||||
print(f" ✓ Valid JSON with {len(data)} keys")
|
||||
else:
|
||||
size = file_path.stat().st_size
|
||||
print(f" ✓ File size: {size:,} bytes")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Error: {e}")
|
||||
return False
|
||||
|
||||
def process_directory(dir_path: Path) -> int:
|
||||
"""Process all files in directory."""
|
||||
processed = 0
|
||||
failed = 0
|
||||
|
||||
for file_path in dir_path.rglob('*'):
|
||||
if file_path.is_file():
|
||||
if process_file(file_path):
|
||||
processed += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
return processed, failed
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Integrates with feature stores (e.g., Feast, Tecton) to manage and serve features for online and offline model deployment."
|
||||
)
|
||||
parser.add_argument('input', help='Input file or directory')
|
||||
parser.add_argument('--output', '-o', help='Output directory')
|
||||
parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
|
||||
parser.add_argument('--config', '-c', help='Configuration file')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
input_path = Path(args.input)
|
||||
|
||||
print(f"🚀 feature-engineering-toolkit - feature_store_integration.py")
|
||||
print(f" Category: ai-ml")
|
||||
print(f" Plugin: feature-engineering-toolkit")
|
||||
print(f" Input: {input_path}")
|
||||
|
||||
if args.config:
|
||||
if Path(args.config).exists():
|
||||
with open(args.config) as f:
|
||||
config = json.load(f)
|
||||
print(f" Config: {args.config}")
|
||||
|
||||
# Process input
|
||||
if input_path.is_file():
|
||||
success = process_file(input_path)
|
||||
result = 0 if success else 1
|
||||
elif input_path.is_dir():
|
||||
processed, failed = process_directory(input_path)
|
||||
print(f"\n📊 SUMMARY")
|
||||
print(f" ✅ Processed: {processed}")
|
||||
print(f" ❌ Failed: {failed}")
|
||||
result = 0 if failed == 0 else 1
|
||||
else:
|
||||
print(f"❌ Invalid input: {input_path}")
|
||||
result = 1
|
||||
|
||||
if result == 0:
|
||||
print("\n✅ Completed successfully")
|
||||
else:
|
||||
print("\n❌ Completed with errors")
|
||||
|
||||
return result
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
100
skills/ml-pipeline/scripts/model_evaluation.py
Normal file
100
skills/ml-pipeline/scripts/model_evaluation.py
Normal file
@@ -0,0 +1,100 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
automl-pipeline-builder - model_evaluation.py
|
||||
Script to evaluate the performance of the trained AutoML model using various metrics and generate a report.
|
||||
Generated: 2025-12-10 03:48:17
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
def process_file(file_path: Path) -> bool:
|
||||
"""Process individual file."""
|
||||
if not file_path.exists():
|
||||
print(f"❌ File not found: {file_path}")
|
||||
return False
|
||||
|
||||
print(f"📄 Processing: {file_path}")
|
||||
|
||||
# Add processing logic here based on skill requirements
|
||||
# This is a template that can be customized
|
||||
|
||||
try:
|
||||
if file_path.suffix == '.json':
|
||||
with open(file_path) as f:
|
||||
data = json.load(f)
|
||||
print(f" ✓ Valid JSON with {len(data)} keys")
|
||||
else:
|
||||
size = file_path.stat().st_size
|
||||
print(f" ✓ File size: {size:,} bytes")
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f" ✗ Error: {e}")
|
||||
return False
|
||||
|
||||
def process_directory(dir_path: Path) -> int:
|
||||
"""Process all files in directory."""
|
||||
processed = 0
|
||||
failed = 0
|
||||
|
||||
for file_path in dir_path.rglob('*'):
|
||||
if file_path.is_file():
|
||||
if process_file(file_path):
|
||||
processed += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
return processed, failed
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Script to evaluate the performance of the trained AutoML model using various metrics and generate a report."
|
||||
)
|
||||
parser.add_argument('input', help='Input file or directory')
|
||||
parser.add_argument('--output', '-o', help='Output directory')
|
||||
parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
|
||||
parser.add_argument('--config', '-c', help='Configuration file')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
input_path = Path(args.input)
|
||||
|
||||
print(f"🚀 automl-pipeline-builder - model_evaluation.py")
|
||||
print(f" Category: ai-ml")
|
||||
print(f" Plugin: automl-pipeline-builder")
|
||||
print(f" Input: {input_path}")
|
||||
|
||||
if args.config:
|
||||
if Path(args.config).exists():
|
||||
with open(args.config) as f:
|
||||
config = json.load(f)
|
||||
print(f" Config: {args.config}")
|
||||
|
||||
# Process input
|
||||
if input_path.is_file():
|
||||
success = process_file(input_path)
|
||||
result = 0 if success else 1
|
||||
elif input_path.is_dir():
|
||||
processed, failed = process_directory(input_path)
|
||||
print(f"\n📊 SUMMARY")
|
||||
print(f" ✅ Processed: {processed}")
|
||||
print(f" ❌ Failed: {failed}")
|
||||
result = 0 if failed == 0 else 1
|
||||
else:
|
||||
print(f"❌ Invalid input: {input_path}")
|
||||
result = 1
|
||||
|
||||
if result == 0:
|
||||
print("\n✅ Completed successfully")
|
||||
else:
|
||||
print("\n❌ Completed with errors")
|
||||
|
||||
return result
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
170
skills/ml-pipeline/scripts/pipeline_deployment.py
Normal file
170
skills/ml-pipeline/scripts/pipeline_deployment.py
Normal file
@@ -0,0 +1,170 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
automl-pipeline-builder - Deployment Script
|
||||
Script to deploy the trained AutoML pipeline to a production environment.
|
||||
Generated: 2025-12-10 03:48:17
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import shutil
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
class Deployer:
|
||||
def __init__(self, source: str, target: str):
|
||||
self.source = Path(source)
|
||||
self.target = Path(target)
|
||||
self.deployed = []
|
||||
self.failed = []
|
||||
|
||||
def validate_source(self) -> bool:
|
||||
"""Validate source directory exists."""
|
||||
if not self.source.exists():
|
||||
print(f"❌ Source directory not found: {self.source}")
|
||||
return False
|
||||
|
||||
if not any(self.source.iterdir()):
|
||||
print(f"❌ Source directory is empty: {self.source}")
|
||||
return False
|
||||
|
||||
print(f"✓ Source validated: {self.source}")
|
||||
return True
|
||||
|
||||
def prepare_target(self) -> bool:
|
||||
"""Prepare target directory."""
|
||||
try:
|
||||
self.target.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create deployment metadata
|
||||
metadata = {
|
||||
"deployment_time": datetime.now().isoformat(),
|
||||
"source": str(self.source),
|
||||
"skill": "automl-pipeline-builder",
|
||||
"category": "ai-ml",
|
||||
"plugin": "automl-pipeline-builder"
|
||||
}
|
||||
|
||||
metadata_file = self.target / ".deployment.json"
|
||||
with open(metadata_file, 'w') as f:
|
||||
json.dump(metadata, f, indent=2)
|
||||
|
||||
print(f"✓ Target prepared: {self.target}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"❌ Failed to prepare target: {e}")
|
||||
return False
|
||||
|
||||
def deploy_files(self) -> bool:
|
||||
"""Deploy files from source to target."""
|
||||
success = True
|
||||
|
||||
for source_file in self.source.rglob('*'):
|
||||
if source_file.is_file():
|
||||
relative_path = source_file.relative_to(self.source)
|
||||
target_file = self.target / relative_path
|
||||
|
||||
try:
|
||||
target_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(source_file, target_file)
|
||||
self.deployed.append(str(relative_path))
|
||||
print(f" ✓ Deployed: {relative_path}")
|
||||
except Exception as e:
|
||||
self.failed.append({
|
||||
"file": str(relative_path),
|
||||
"error": str(e)
|
||||
})
|
||||
print(f" ✗ Failed: {relative_path} - {e}")
|
||||
success = False
|
||||
|
||||
return success
|
||||
|
||||
def generate_report(self) -> Dict:
|
||||
"""Generate deployment report."""
|
||||
report = {
|
||||
"deployment_time": datetime.now().isoformat(),
|
||||
"skill": "automl-pipeline-builder",
|
||||
"source": str(self.source),
|
||||
"target": str(self.target),
|
||||
"total_files": len(self.deployed) + len(self.failed),
|
||||
"deployed": len(self.deployed),
|
||||
"failed": len(self.failed),
|
||||
"deployed_files": self.deployed,
|
||||
"failed_files": self.failed
|
||||
}
|
||||
|
||||
# Save report
|
||||
report_file = self.target / "deployment_report.json"
|
||||
with open(report_file, 'w') as f:
|
||||
json.dump(report, f, indent=2)
|
||||
|
||||
return report
|
||||
|
||||
def rollback(self):
|
||||
"""Rollback deployment on failure."""
|
||||
print("⚠️ Rolling back deployment...")
|
||||
|
||||
for deployed_file in self.deployed:
|
||||
file_path = self.target / deployed_file
|
||||
if file_path.exists():
|
||||
file_path.unlink()
|
||||
print(f" ✓ Removed: {deployed_file}")
|
||||
|
||||
# Remove empty directories
|
||||
for dir_path in sorted(self.target.rglob('*'), reverse=True):
|
||||
if dir_path.is_dir() and not any(dir_path.iterdir()):
|
||||
dir_path.rmdir()
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Script to deploy the trained AutoML pipeline to a production environment.")
|
||||
parser.add_argument('source', help='Source directory')
|
||||
parser.add_argument('target', help='Target deployment directory')
|
||||
parser.add_argument('--dry-run', action='store_true', help='Simulate deployment')
|
||||
parser.add_argument('--force', action='store_true', help='Overwrite existing files')
|
||||
parser.add_argument('--rollback-on-error', action='store_true', help='Rollback on any error')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
deployer = Deployer(args.source, args.target)
|
||||
|
||||
print(f"🚀 Deploying automl-pipeline-builder...")
|
||||
print(f" Source: {args.source}")
|
||||
print(f" Target: {args.target}")
|
||||
|
||||
if args.dry_run:
|
||||
print("\n⚠️ DRY RUN MODE - No files will be deployed")
|
||||
return 0
|
||||
|
||||
# Validate and prepare
|
||||
if not deployer.validate_source():
|
||||
return 1
|
||||
|
||||
if not deployer.prepare_target():
|
||||
return 1
|
||||
|
||||
# Deploy
|
||||
success = deployer.deploy_files()
|
||||
|
||||
# Generate report
|
||||
report = deployer.generate_report()
|
||||
|
||||
print(f"\n📊 DEPLOYMENT SUMMARY")
|
||||
print(f" Total Files: {report['total_files']}")
|
||||
print(f" ✅ Deployed: {report['deployed']}")
|
||||
print(f" ❌ Failed: {report['failed']}")
|
||||
|
||||
if not success and args.rollback_on_error:
|
||||
deployer.rollback()
|
||||
return 1
|
||||
|
||||
if report['failed'] == 0:
|
||||
print(f"\n✅ Deployment completed successfully!")
|
||||
return 0
|
||||
else:
|
||||
print(f"\n⚠️ Deployment completed with errors")
|
||||
return 1
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user