Antoine/Atomizer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Migrate drone_gimbal_arm_optimization to use structured logging system (Phase 1.3.1)

Browse Source 
Migrate drone_gimbal_arm study as reference implementation for Phase 1.3 logging system.

Changes:
- Replace all print() statements with logger calls throughout run_optimization.py
- Add logger.trial_start() and logger.trial_complete() for structured trial logging
- Use logger.trial_failed() for error handling with full tracebacks
- Add logger.study_start() and logger.study_complete() for lifecycle logging
- Replace constraint violation prints with logger.warning()
- Create comprehensive LOGGING_MIGRATION_GUIDE.md with before/after examples

Benefits:
- Color-coded console output (green INFO, yellow WARNING, red ERROR)
- Automatic file logging to 2_results/optimization.log with rotation (50MB, 3 backups)
- Structured format with timestamps for dashboard integration
- Professional error handling with exc_info=True
- Reference implementation for migrating remaining studies

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Anto01
2025-11-24 09:39:56 -05:00
parent 3bff7cf6b3
commit d2c18bb7db
2 changed files with 439 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

347
docs/LOGGING_MIGRATION_GUIDE.md Normal file
View File

@@ -0,0 +1,347 @@
# Logging Migration Guide
**How to migrate existing studies to use the new structured logging system**
## Overview
The new `optimization_engine.logger` module provides production-ready logging with:
- Color-coded console output
- Automatic file logging with rotation
- Structured trial logging for dashboard integration
- Zero external dependencies
## Migration Steps
### Step 1: Import the Logger
**Before:**
```python
import sys
import json
from pathlib import Path
```
**After:**
```python
import sys
import json
from pathlib import Path
from optimization_engine.logger import get_logger
```
### Step 2: Initialize Logger in main()
**Before:**
```python
def main():
study_dir = Path(__file__).parent
results_dir = study_dir / "2_results"
results_dir.mkdir(exist_ok=True)
print("=" * 80)
print("MY OPTIMIZATION STUDY")
print("=" * 80)
```
**After:**
```python
def main():
study_dir = Path(__file__).parent
results_dir = study_dir / "2_results"
results_dir.mkdir(exist_ok=True)
# Initialize logger with file logging
logger = get_logger(
"my_study_name",
study_dir=results_dir
)
logger.info("=" * 80)
logger.info("MY OPTIMIZATION STUDY")
logger.info("=" * 80)
```
### Step 3: Replace print() with logger calls
**Basic Replacements:**
```python
# Before
print("Starting optimization...")
print(f"[ERROR] Simulation failed")
print(f"[WARNING] Constraint violated")
# After
logger.info("Starting optimization...")
logger.error("Simulation failed")
logger.warning("Constraint violated")
```
### Step 4: Use Structured Trial Logging
**Trial Start - Before:**
```python
print(f"\n{'='*60}")
print(f"Trial #{trial.number}")
print(f"{'='*60}")
print(f"Design Variables:")
for name, value in design_vars.items():
print(f" {name}: {value:.3f}")
```
**Trial Start - After:**
```python
logger.trial_start(trial.number, design_vars)
```
**Trial Complete - Before:**
```python
print(f"\nTrial #{trial.number} COMPLETE")
print("Objectives:")
for name, value in objectives.items():
print(f" {name}: {value:.4f}")
print("Constraints:")
for name, value in constraints.items():
print(f" {name}: {value:.4f}")
print("[OK] Feasible" if feasible else "[WARNING] Infeasible")
```
**Trial Complete - After:**
```python
logger.trial_complete(
trial.number,
objectives=objectives,
constraints=constraints,
feasible=feasible
)
```
**Trial Failed - Before:**
```python
print(f"\n[ERROR] Trial #{trial.number} FAILED")
print(f"Error: {error_message}")
import traceback
traceback.print_exc()
```
**Trial Failed - After:**
```python
logger.trial_failed(trial.number, error_message)
logger.error("Full traceback:", exc_info=True)
```
### Step 5: Use Study Lifecycle Methods
**Study Start - Before:**
```python
print("=" * 80)
print(f"OPTIMIZATION STUDY: {study_name}")
print("=" * 80)
print(f"Trials: {n_trials}")
print(f"Sampler: {sampler}")
print("=" * 80)
```
**Study Start - After:**
```python
logger.study_start(study_name, n_trials=n_trials, sampler=sampler)
```
**Study Complete - Before:**
```python
print("=" * 80)
print(f"STUDY COMPLETE: {study_name}")
print("=" * 80)
print(f"Total trials: {n_trials}")
print(f"Successful: {n_successful}")
print(f"Failed/Pruned: {n_trials - n_successful}")
print("=" * 80)
```
**Study Complete - After:**
```python
logger.study_complete(study_name, n_trials=n_trials, n_successful=n_successful)
```
## Complete Example
### Before (Old Style)
```python
import sys
from pathlib import Path
import optuna
def main():
study_dir = Path(__file__).parent
results_dir = study_dir / "2_results"
results_dir.mkdir(exist_ok=True)
print("=" * 80)
print("MY OPTIMIZATION STUDY")
print("=" * 80)
def objective(trial):
x = trial.suggest_float("x", -10, 10)
print(f"\nTrial #{trial.number}")
print(f"x = {x:.4f}")
try:
result = x ** 2
print(f"Result: {result:.4f}")
return result
except Exception as e:
print(f"[ERROR] Trial failed: {e}")
raise
study = optuna.create_study()
study.optimize(objective, n_trials=10)
print("\nOptimization complete!")
print(f"Best value: {study.best_value:.4f}")
if __name__ == "__main__":
main()
```
### After (New Style with Logger)
```python
import sys
from pathlib import Path
import optuna
from optimization_engine.logger import get_logger
def main():
study_dir = Path(__file__).parent
results_dir = study_dir / "2_results"
results_dir.mkdir(exist_ok=True)
# Initialize logger with file logging
logger = get_logger("my_study", study_dir=results_dir)
logger.study_start("my_study", n_trials=10, sampler="TPESampler")
def objective(trial):
x = trial.suggest_float("x", -10, 10)
logger.trial_start(trial.number, {"x": x})
try:
result = x ** 2
logger.trial_complete(
trial.number,
objectives={"f(x)": result},
feasible=True
)
return result
except Exception as e:
logger.trial_failed(trial.number, str(e))
logger.error("Full traceback:", exc_info=True)
raise
study = optuna.create_study()
study.optimize(objective, n_trials=10)
logger.study_complete("my_study", n_trials=10, n_successful=len(study.trials))
logger.info(f"Best value: {study.best_value:.4f}")
if __name__ == "__main__":
main()
```
## Benefits
After migration, you'll get:
1. **Color-coded console output** - Green for INFO, Yellow for WARNING, Red for ERROR
2. **Automatic file logging** - All output saved to `2_results/optimization.log`
3. **Log rotation** - Automatic rotation at 50MB with 3 backups
4. **Structured format** - Timestamps and module names in file logs
5. **Dashboard integration** - Trial logs in structured format for future parsing
## Log File Location
After migration, logs will be automatically saved to:
```
studies/your_study/
└── 2_results/
├── optimization.log # Current log file
├── optimization.log.1 # Backup 1 (most recent)
├── optimization.log.2 # Backup 2
└── optimization.log.3 # Backup 3 (oldest)
```
## Testing Your Migration
Run your migrated study with a single trial:
```bash
cd studies/your_study
python run_optimization.py --trials 1
```
Check:
- ✅ Console output is color-coded
- ✅ File created at `2_results/optimization.log`
- ✅ Trial start/complete messages format correctly
- ✅ No errors about missing imports
## Common Patterns
### Error Handling with Context
**Before:**
```python
try:
result = run_simulation()
except Exception as e:
print(f"[ERROR] Simulation failed: {e}")
import traceback
traceback.print_exc()
raise
```
**After:**
```python
try:
result = run_simulation()
except Exception as e:
logger.error(f"Simulation failed: {e}", exc_info=True)
raise
```
### Conditional Logging
```python
# Use log levels appropriately
logger.debug("Detailed debugging info") # Only when debugging
logger.info("Starting optimization...") # General progress
logger.warning("Design var out of bounds") # Potential issues
logger.error("Simulation failed") # Actual errors
```
### Progress Messages
```python
# Before
print(f"Processing trial {i}/{total}...")
# After
logger.info(f"Processing trial {i}/{total}...")
```
## Next Steps
After migrating your study:
1. Test with a few trials
2. Check the log file in `2_results/optimization.log`
3. Verify color output in console
4. Update study documentation if needed
## Questions?
See:
- [Phase 1.3 Implementation Plan](docs/07_DEVELOPMENT/Phase_1_3_Implementation_Plan.md)
- [optimization_engine/logger.py](optimization_engine/logger.py) - Full API documentation
- [drone_gimbal_arm_optimization](studies/drone_gimbal_arm_optimization/) - Reference implementation

178
studies/drone_gimbal_arm_optimization/run_optimization.py
View File

@@ -35,6 +35,7 @@ from optimization_engine.extractors.extract_von_mises_stress import extract_soli
from optimization_engine.extractors.op2_extractor import OP2Extractor from optimization_engine.extractors.op2_extractor import OP2Extractor
from optimization_engine.extractors.extract_frequency import extract_frequency from optimization_engine.extractors.extract_frequency import extract_frequency
from optimization_engine.extractors.extract_mass_from_bdf import extract_mass_from_bdf from optimization_engine.extractors.extract_mass_from_bdf import extract_mass_from_bdf
from optimization_engine.logger import get_logger
# Import central configuration # Import central configuration
try: try:
@@ -57,52 +58,58 @@ def main():
# Get study directory # Get study directory
study_dir = Path(__file__).parent study_dir = Path(__file__).parent
results_dir = study_dir / "2_results"
results_dir.mkdir(exist_ok=True)
print("=" * 80) # Initialize logger with file logging
print("DRONE GIMBAL ARM OPTIMIZATION - PROTOCOL 11 (NSGA-II)") logger = get_logger(
print("=" * 80) "drone_gimbal_arm",
print() study_dir=results_dir
print("Engineering Scenario:") )
print(" Professional aerial cinematography drone camera gimbal support arm")
print() logger.info("=" * 80)
print("Objectives:") logger.info("DRONE GIMBAL ARM OPTIMIZATION - PROTOCOL 11 (NSGA-II)")
print(" 1. MINIMIZE mass (target < 4000g, baseline = 4500g)") logger.info("=" * 80)
print(" 2. MAXIMIZE fundamental frequency (target > 150 Hz)") logger.info("")
print() logger.info("Engineering Scenario:")
print("Constraints:") logger.info(" Professional aerial cinematography drone camera gimbal support arm")
print(" - Max displacement < 1.5mm (850g camera payload)") logger.info("")
print(" - Max von Mises stress < 120 MPa (Al 6061-T6, SF=2.3)") logger.info("Objectives:")
print(" - Natural frequency > 150 Hz (avoid rotor resonance 80-120 Hz)") logger.info(" 1. MINIMIZE mass (target < 4000g, baseline = 4500g)")
print() logger.info(" 2. MAXIMIZE fundamental frequency (target > 150 Hz)")
print("Design Variables:") logger.info("")
print(" - beam_half_core_thickness: 5-10 mm") logger.info("Constraints:")
print(" - beam_face_thickness: 1-3 mm") logger.info(" - Max displacement < 1.5mm (850g camera payload)")
print(" - holes_diameter: 10-50 mm") logger.info(" - Max von Mises stress < 120 MPa (Al 6061-T6, SF=2.3)")
print(" - hole_count: 8-14") logger.info(" - Natural frequency > 150 Hz (avoid rotor resonance 80-120 Hz)")
print() logger.info("")
print(f"Running {args.trials} trials with NSGA-II sampler...") logger.info("Design Variables:")
print("=" * 80) logger.info(" - beam_half_core_thickness: 5-10 mm")
print() logger.info(" - beam_face_thickness: 1-3 mm")
logger.info(" - holes_diameter: 10-50 mm")
logger.info(" - hole_count: 8-14")
logger.info("")
logger.info(f"Running {args.trials} trials with NSGA-II sampler...")
logger.info("=" * 80)
logger.info("")
# Load configuration # Load configuration
opt_config_file = study_dir / "1_setup" / "optimization_config.json" opt_config_file = study_dir / "1_setup" / "optimization_config.json"
if not opt_config_file.exists(): if not opt_config_file.exists():
print(f"[ERROR] Optimization config not found: {opt_config_file}") logger.error(f"Optimization config not found: {opt_config_file}")
sys.exit(1) sys.exit(1)
opt_config = load_config(opt_config_file) opt_config = load_config(opt_config_file)
print(f"Loaded optimization config: {opt_config['study_name']}") logger.info(f"Loaded optimization config: {opt_config['study_name']}")
print(f"Protocol: {opt_config['optimization_settings']['protocol']}") logger.info(f"Protocol: {opt_config['optimization_settings']['protocol']}")
print() logger.info("")
# Setup paths # Setup paths
model_dir = study_dir / "1_setup" / "model" model_dir = study_dir / "1_setup" / "model"
model_file = model_dir / "Beam.prt" model_file = model_dir / "Beam.prt"
sim_file = model_dir / "Beam_sim1.sim" sim_file = model_dir / "Beam_sim1.sim"
results_dir = study_dir / "2_results"
results_dir.mkdir(exist_ok=True)
# Initialize NX solver # Initialize NX solver
nx_solver = NXSolver( nx_solver = NXSolver(
@@ -129,15 +136,10 @@ def main():
dv['bounds'][1] dv['bounds'][1]
) )
print(f"\n{'='*60}") logger.trial_start(trial.number, design_vars)
print(f"Trial #{trial.number}")
print(f"{'='*60}")
print(f"Design Variables:")
for name, value in design_vars.items():
print(f" {name}: {value:.3f}")
# Run simulation # Run simulation
print(f"\nRunning simulation...") logger.info("Running simulation...")
try: try:
result = nx_solver.run_simulation( result = nx_solver.run_simulation(
sim_file=sim_file, sim_file=sim_file,
@@ -147,17 +149,18 @@ def main():
) )
if not result['success']: if not result['success']:
print(f"[ERROR] Simulation failed: {result.get('error', 'Unknown error')}") error_msg = result.get('error', 'Unknown error')
logger.trial_failed(trial.number, error_msg)
trial.set_user_attr("feasible", False) trial.set_user_attr("feasible", False)
trial.set_user_attr("error", result.get('error', 'Unknown')) trial.set_user_attr("error", error_msg)
# Prune failed simulations instead of returning penalty values # Prune failed simulations instead of returning penalty values
raise optuna.TrialPruned(f"Simulation failed: {result.get('error', 'Unknown error')}") raise optuna.TrialPruned(f"Simulation failed: {error_msg}")
op2_file = result['op2_file'] op2_file = result['op2_file']
print(f"Simulation successful: {op2_file}") logger.info(f"Simulation successful: {op2_file}")
# Extract all objectives and constraints # Extract all objectives and constraints
print(f"\nExtracting results...") logger.info("Extracting results...")
# Extract mass (grams) from CAD expression p173 # Extract mass (grams) from CAD expression p173
# This expression measures the CAD mass directly # This expression measures the CAD mass directly
@@ -166,28 +169,28 @@ def main():
prt_file = model_file # Beam.prt prt_file = model_file # Beam.prt
mass_kg = extract_mass_from_expression(prt_file, expression_name="p173") mass_kg = extract_mass_from_expression(prt_file, expression_name="p173")
mass = mass_kg * 1000.0 # Convert to grams mass = mass_kg * 1000.0 # Convert to grams
print(f" mass: {mass:.3f} g (from CAD expression p173)") logger.info(f" mass: {mass:.3f} g (from CAD expression p173)")
# Extract frequency (Hz) - from modal analysis (solution 2) # Extract frequency (Hz) - from modal analysis (solution 2)
# The drone gimbal has TWO solutions: solution_1 (static) and solution_2 (modal) # The drone gimbal has TWO solutions: solution_1 (static) and solution_2 (modal)
op2_modal = str(op2_file).replace("solution_1", "solution_2") op2_modal = str(op2_file).replace("solution_1", "solution_2")
freq_result = extract_frequency(op2_modal, subcase=1, mode_number=1) freq_result = extract_frequency(op2_modal, subcase=1, mode_number=1)
frequency = freq_result['frequency'] frequency = freq_result['frequency']
print(f" fundamental_frequency: {frequency:.3f} Hz") logger.info(f" fundamental_frequency: {frequency:.3f} Hz")
# Extract displacement (mm) - from static analysis (subcase 1) # Extract displacement (mm) - from static analysis (subcase 1)
disp_result = extract_displacement(op2_file, subcase=1) disp_result = extract_displacement(op2_file, subcase=1)
max_disp = disp_result['max_displacement'] max_disp = disp_result['max_displacement']
print(f" max_displacement_limit: {max_disp:.3f} mm") logger.info(f" max_displacement_limit: {max_disp:.3f} mm")
# Extract stress (MPa) - from static analysis (subcase 1) # Extract stress (MPa) - from static analysis (subcase 1)
stress_result = extract_solid_stress(op2_file, subcase=1, element_type='cquad4') stress_result = extract_solid_stress(op2_file, subcase=1, element_type='cquad4')
max_stress = stress_result['max_von_mises'] max_stress = stress_result['max_von_mises']
print(f" max_stress_limit: {max_stress:.3f} MPa") logger.info(f" max_stress_limit: {max_stress:.3f} MPa")
# Frequency constraint uses same value as objective # Frequency constraint uses same value as objective
min_freq = frequency min_freq = frequency
print(f" min_frequency_limit: {min_freq:.3f} Hz") logger.info(f" min_frequency_limit: {min_freq:.3f} Hz")
# Check constraints # Check constraints
constraint_values = { constraint_values = {
@@ -211,14 +214,14 @@ def main():
constraint_violations.append(f"{name}: {value:.2f} < {threshold} (violation: {violation:.1%})") constraint_violations.append(f"{name}: {value:.2f} < {threshold} (violation: {violation:.1%})")
if constraint_violations: if constraint_violations:
print(f"\n[WARNING] Constraint violations:") logger.warning("Constraint violations:")
for v in constraint_violations: for v in constraint_violations:
print(f" - {v}") logger.warning(f" - {v}")
trial.set_user_attr("constraint_violations", constraint_violations) trial.set_user_attr("constraint_violations", constraint_violations)
trial.set_user_attr("feasible", False) trial.set_user_attr("feasible", False)
# NSGA-II handles constraints through constraint_satisfied flag - no penalty needed # NSGA-II handles constraints through constraint_satisfied flag - no penalty needed
else: else:
print(f"\n[OK] All constraints satisfied") logger.info("All constraints satisfied")
trial.set_user_attr("feasible", True) trial.set_user_attr("feasible", True)
# Store all results as trial attributes for dashboard # Store all results as trial attributes for dashboard
@@ -228,9 +231,15 @@ def main():
trial.set_user_attr("max_stress", max_stress) trial.set_user_attr("max_stress", max_stress)
trial.set_user_attr("design_vars", design_vars) trial.set_user_attr("design_vars", design_vars)
print(f"\nObjectives:") # Log successful trial completion
print(f" mass: {mass:.2f} g (minimize)") objectives = {"mass": mass, "frequency": frequency}
print(f" frequency: {frequency:.2f} Hz (maximize)") constraints_dict = {
"max_displacement_limit": max_disp,
"max_stress_limit": max_stress,
"min_frequency_limit": min_freq
}
feasible = len(constraint_violations) == 0
logger.trial_complete(trial.number, objectives, constraints_dict, feasible)
# Return tuple for NSGA-II: (minimize mass, maximize frequency) # Return tuple for NSGA-II: (minimize mass, maximize frequency)
# Using proper semantic directions in study creation # Using proper semantic directions in study creation
@@ -240,9 +249,8 @@ def main():
# Re-raise pruned exceptions (don't catch them) # Re-raise pruned exceptions (don't catch them)
raise raise
except Exception as e: except Exception as e:
print(f"\n[ERROR] Trial failed with exception: {e}") logger.trial_failed(trial.number, str(e))
import traceback logger.error("Full traceback:", exc_info=True)
traceback.print_exc()
trial.set_user_attr("error", str(e)) trial.set_user_attr("error", str(e))
trial.set_user_attr("feasible", False) trial.set_user_attr("feasible", False)
# Prune corrupted trials instead of returning penalty values # Prune corrupted trials instead of returning penalty values
@@ -253,15 +261,15 @@ def main():
storage = f"sqlite:///{results_dir / 'study.db'}" storage = f"sqlite:///{results_dir / 'study.db'}"
if args.resume: if args.resume:
print(f"[INFO] Resuming existing study: {study_name}") logger.info(f"Resuming existing study: {study_name}")
study = optuna.load_study( study = optuna.load_study(
study_name=study_name, study_name=study_name,
storage=storage, storage=storage,
sampler=NSGAIISampler() sampler=NSGAIISampler()
) )
print(f"[INFO] Loaded study with {len(study.trials)} existing trials") logger.info(f"Loaded study with {len(study.trials)} existing trials")
else: else:
print(f"[INFO] Creating new study: {study_name}") logger.info(f"Creating new study: {study_name}")
study = optuna.create_study( study = optuna.create_study(
study_name=study_name, study_name=study_name,
storage=storage, storage=storage,
@@ -270,9 +278,9 @@ def main():
load_if_exists=True # Always allow resuming existing study load_if_exists=True # Always allow resuming existing study
) )
# Run optimization # Log study start
print(f"\nStarting optimization with {args.trials} trials...") logger.study_start(study_name, n_trials=args.trials, sampler="NSGAIISampler")
print() logger.info("")
study.optimize( study.optimize(
objective, objective,
@@ -280,36 +288,34 @@ def main():
show_progress_bar=True show_progress_bar=True
) )
# Save final results # Log study completion
print() n_successful = len([t for t in study.trials if t.state == optuna.trial.TrialState.COMPLETE])
print("=" * 80) logger.study_complete(study_name, n_trials=len(study.trials), n_successful=n_successful)
print("Optimization Complete!")
print("=" * 80) logger.info("")
print() logger.info(f"Pareto front solutions: {len(study.best_trials)}")
print(f"Total trials: {len(study.trials)}") logger.info("")
print(f"Pareto front solutions: {len(study.best_trials)}")
print()
# Show Pareto front # Show Pareto front
print("Pareto Front (non-dominated solutions):") logger.info("Pareto Front (non-dominated solutions):")
print() logger.info("")
for i, trial in enumerate(study.best_trials): for i, trial in enumerate(study.best_trials):
mass = trial.values[0] mass = trial.values[0]
freq = trial.values[1] # Frequency is stored as positive now freq = trial.values[1] # Frequency is stored as positive now
feasible = trial.user_attrs.get('feasible', False) feasible = trial.user_attrs.get('feasible', False)
print(f" Solution #{i+1} (Trial {trial.number}):") logger.info(f" Solution #{i+1} (Trial {trial.number}):")
print(f" Mass: {mass:.2f} g") logger.info(f" Mass: {mass:.2f} g")
print(f" Frequency: {freq:.2f} Hz") logger.info(f" Frequency: {freq:.2f} Hz")
print(f" Feasible: {feasible}") logger.info(f" Feasible: {feasible}")
print() logger.info("")
print("Results available in: studies/drone_gimbal_arm_optimization/2_results/") logger.info("Results available in: studies/drone_gimbal_arm_optimization/2_results/")
print() logger.info("")
print("View in Dashboard:") logger.info("View in Dashboard:")
print(" 1. Ensure backend is running: cd atomizer-dashboard/backend && python -m uvicorn api.main:app --reload") logger.info(" 1. Ensure backend is running: cd atomizer-dashboard/backend && python -m uvicorn api.main:app --reload")
print(" 2. Open dashboard: http://localhost:3003") logger.info(" 2. Open dashboard: http://localhost:3003")
print(" 3. Select study: drone_gimbal_arm_optimization") logger.info(" 3. Select study: drone_gimbal_arm_optimization")
print() logger.info("")
if __name__ == "__main__": if __name__ == "__main__":