Anto01
a26914bbe8
Dashboard: - Add Studio page with drag-drop model upload and Claude chat - Add intake system for study creation workflow - Improve session manager and context builder - Add intake API routes and frontend components Optimization Engine: - Add CLI module for command-line operations - Add intake module for study preprocessing - Add validation module with gate checks - Improve Zernike extractor documentation - Update spec models with better validation - Enhance solve_simulation robustness Documentation: - Add ATOMIZER_STUDIO.md planning doc - Add ATOMIZER_UX_SYSTEM.md for UX patterns - Update extractor library docs - Add study-readme-generator skill Tools: - Add test scripts for extraction validation - Add Zernike recentering test Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
384 lines
12 KiB
Python
384 lines
12 KiB
Python
"""
|
|
Atomizer CLI Main Entry Point
|
|
=============================
|
|
|
|
Provides the `atomizer` command with subcommands:
|
|
- intake: Process an intake folder
|
|
- validate: Validate a study
|
|
- finalize: Generate final report
|
|
- list: List studies
|
|
|
|
Usage:
|
|
atomizer intake bracket_project
|
|
atomizer validate bracket_mass_opt
|
|
atomizer finalize bracket_mass_opt --format html
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import argparse
|
|
import logging
|
|
|
|
|
|
def setup_logging(verbose: bool = False):
|
|
"""Setup logging configuration."""
|
|
level = logging.DEBUG if verbose else logging.INFO
|
|
logging.basicConfig(
|
|
level=level,
|
|
format="%(message)s",
|
|
)
|
|
|
|
|
|
def find_project_root() -> Path:
|
|
"""Find the Atomizer project root."""
|
|
current = Path(__file__).parent
|
|
while current != current.parent:
|
|
if (current / "CLAUDE.md").exists():
|
|
return current
|
|
current = current.parent
|
|
return Path.cwd()
|
|
|
|
|
|
def cmd_intake(args):
|
|
"""Process an intake folder."""
|
|
from optimization_engine.intake import IntakeProcessor
|
|
|
|
# Determine inbox folder
|
|
inbox_path = Path(args.folder)
|
|
|
|
if not inbox_path.is_absolute():
|
|
# Check if it's in _inbox
|
|
project_root = find_project_root()
|
|
inbox_dir = project_root / "studies" / "_inbox"
|
|
|
|
if (inbox_dir / args.folder).exists():
|
|
inbox_path = inbox_dir / args.folder
|
|
elif (project_root / "studies" / args.folder).exists():
|
|
inbox_path = project_root / "studies" / args.folder
|
|
|
|
if not inbox_path.exists():
|
|
print(f"Error: Folder not found: {inbox_path}")
|
|
return 1
|
|
|
|
print(f"Processing intake: {inbox_path}")
|
|
print("=" * 60)
|
|
|
|
# Progress callback
|
|
def progress(message: str, percent: float):
|
|
bar_width = 30
|
|
filled = int(bar_width * percent)
|
|
bar = "=" * filled + "-" * (bar_width - filled)
|
|
print(f"\r[{bar}] {percent * 100:5.1f}% {message}", end="", flush=True)
|
|
if percent >= 1.0:
|
|
print() # Newline at end
|
|
|
|
try:
|
|
processor = IntakeProcessor(
|
|
inbox_path,
|
|
progress_callback=progress if not args.quiet else None,
|
|
)
|
|
|
|
context = processor.process(
|
|
run_baseline=not args.skip_baseline,
|
|
copy_files=True,
|
|
run_introspection=True,
|
|
)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("INTAKE COMPLETE")
|
|
print("=" * 60)
|
|
|
|
# Show summary
|
|
summary = context.get_context_summary()
|
|
print(f"\nStudy: {context.study_name}")
|
|
print(f"Location: {processor.study_dir}")
|
|
print(f"\nContext loaded:")
|
|
print(f" Model: {'Yes' if summary['has_model'] else 'No'}")
|
|
print(f" Introspection: {'Yes' if summary['has_introspection'] else 'No'}")
|
|
print(f" Baseline: {'Yes' if summary['has_baseline'] else 'No'}")
|
|
print(f" Goals: {'Yes' if summary['has_goals'] else 'No'}")
|
|
print(f" Pre-config: {'Yes' if summary['has_preconfig'] else 'No'}")
|
|
print(
|
|
f" Expressions: {summary['num_expressions']} ({summary['num_dv_candidates']} candidates)"
|
|
)
|
|
|
|
if context.has_baseline:
|
|
print(f"\nBaseline: {context.get_baseline_summary()}")
|
|
|
|
if summary["warnings"]:
|
|
print(f"\nWarnings:")
|
|
for w in summary["warnings"]:
|
|
print(f" - {w}")
|
|
|
|
if args.interview:
|
|
print(f"\nTo continue with interview: atomizer interview {context.study_name}")
|
|
elif args.canvas:
|
|
print(f"\nOpen dashboard to configure in Canvas mode")
|
|
else:
|
|
print(f"\nNext steps:")
|
|
print(f" 1. Review context in {processor.study_dir / '0_intake'}")
|
|
print(f" 2. Configure study via interview or canvas")
|
|
print(f" 3. Run: atomizer validate {context.study_name}")
|
|
|
|
return 0
|
|
|
|
except Exception as e:
|
|
print(f"\nError: {e}")
|
|
if args.verbose:
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
|
|
def cmd_validate(args):
|
|
"""Validate a study before running."""
|
|
from optimization_engine.validation import ValidationGate
|
|
|
|
# Find study directory
|
|
study_path = Path(args.study)
|
|
|
|
if not study_path.is_absolute():
|
|
project_root = find_project_root()
|
|
study_path = project_root / "studies" / args.study
|
|
|
|
if not study_path.exists():
|
|
print(f"Error: Study not found: {study_path}")
|
|
return 1
|
|
|
|
print(f"Validating study: {study_path.name}")
|
|
print("=" * 60)
|
|
|
|
# Progress callback
|
|
def progress(message: str, percent: float):
|
|
bar_width = 30
|
|
filled = int(bar_width * percent)
|
|
bar = "=" * filled + "-" * (bar_width - filled)
|
|
print(f"\r[{bar}] {percent * 100:5.1f}% {message}", end="", flush=True)
|
|
if percent >= 1.0:
|
|
print()
|
|
|
|
try:
|
|
gate = ValidationGate(
|
|
study_path,
|
|
progress_callback=progress if not args.quiet else None,
|
|
)
|
|
|
|
result = gate.validate(
|
|
run_test_trials=not args.skip_trials,
|
|
n_test_trials=args.trials,
|
|
)
|
|
|
|
print("\n" + "=" * 60)
|
|
if result.passed:
|
|
print("VALIDATION PASSED")
|
|
else:
|
|
print("VALIDATION FAILED")
|
|
print("=" * 60)
|
|
|
|
# Show spec validation
|
|
if result.spec_check:
|
|
print(f"\nSpec Validation:")
|
|
print(f" Errors: {len(result.spec_check.errors)}")
|
|
print(f" Warnings: {len(result.spec_check.warnings)}")
|
|
|
|
for issue in result.spec_check.errors:
|
|
print(f" [ERROR] {issue.message}")
|
|
for issue in result.spec_check.warnings[:5]: # Limit warnings shown
|
|
print(f" [WARN] {issue.message}")
|
|
|
|
# Show test trials
|
|
if result.test_trials:
|
|
print(f"\nTest Trials:")
|
|
successful = [t for t in result.test_trials if t.success]
|
|
print(f" Completed: {len(successful)}/{len(result.test_trials)}")
|
|
|
|
if result.results_vary:
|
|
print(f" Results vary: Yes (good!)")
|
|
else:
|
|
print(f" Results vary: NO - MESH MAY NOT BE UPDATING!")
|
|
|
|
# Show trial results table
|
|
print(f"\n {'Trial':<8} {'Status':<10} {'Time (s)':<10}", end="")
|
|
if successful and successful[0].objectives:
|
|
for obj in list(successful[0].objectives.keys())[:3]:
|
|
print(f" {obj:<12}", end="")
|
|
print()
|
|
print(" " + "-" * 50)
|
|
|
|
for trial in result.test_trials:
|
|
status = "OK" if trial.success else "FAIL"
|
|
print(
|
|
f" {trial.trial_number:<8} {status:<10} {trial.solve_time_seconds:<10.1f}",
|
|
end="",
|
|
)
|
|
for val in list(trial.objectives.values())[:3]:
|
|
print(f" {val:<12.4f}", end="")
|
|
print()
|
|
|
|
# Show estimates
|
|
if result.avg_solve_time:
|
|
print(f"\nRuntime Estimate:")
|
|
print(f" Avg solve time: {result.avg_solve_time:.1f}s")
|
|
if result.estimated_total_runtime:
|
|
hours = result.estimated_total_runtime / 3600
|
|
print(f" Est. total: {hours:.1f} hours")
|
|
|
|
# Show errors
|
|
if result.errors:
|
|
print(f"\nErrors:")
|
|
for err in result.errors:
|
|
print(f" - {err}")
|
|
|
|
# Approve if passed and requested
|
|
if result.passed:
|
|
if args.approve:
|
|
gate.approve()
|
|
print(f"\nStudy approved for optimization.")
|
|
else:
|
|
print(f"\nTo approve and start: atomizer validate {args.study} --approve")
|
|
|
|
# Save result
|
|
output_path = gate.save_result(result)
|
|
print(f"\nResult saved: {output_path}")
|
|
|
|
return 0 if result.passed else 1
|
|
|
|
except Exception as e:
|
|
print(f"\nError: {e}")
|
|
if args.verbose:
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
|
|
def cmd_list(args):
|
|
"""List available studies."""
|
|
project_root = find_project_root()
|
|
studies_dir = project_root / "studies"
|
|
|
|
print("Available Studies:")
|
|
print("=" * 60)
|
|
|
|
# List inbox items
|
|
inbox_dir = studies_dir / "_inbox"
|
|
if inbox_dir.exists():
|
|
inbox_items = [d for d in inbox_dir.iterdir() if d.is_dir() and not d.name.startswith(".")]
|
|
if inbox_items:
|
|
print("\nPending Intake (_inbox/):")
|
|
for item in sorted(inbox_items):
|
|
has_config = (item / "intake.yaml").exists()
|
|
has_model = bool(list(item.glob("**/*.sim")))
|
|
status = []
|
|
if has_config:
|
|
status.append("config")
|
|
if has_model:
|
|
status.append("model")
|
|
print(f" {item.name:<30} [{', '.join(status) or 'empty'}]")
|
|
|
|
# List active studies
|
|
print("\nActive Studies:")
|
|
for study_dir in sorted(studies_dir.iterdir()):
|
|
if (
|
|
study_dir.is_dir()
|
|
and not study_dir.name.startswith("_")
|
|
and not study_dir.name.startswith(".")
|
|
):
|
|
# Check status
|
|
has_spec = (study_dir / "atomizer_spec.json").exists() or (
|
|
study_dir / "optimization_config.json"
|
|
).exists()
|
|
has_db = (study_dir / "3_results" / "study.db").exists() or (
|
|
study_dir / "2_results" / "study.db"
|
|
).exists()
|
|
has_approval = (study_dir / ".validation_approved").exists()
|
|
|
|
status = []
|
|
if has_spec:
|
|
status.append("configured")
|
|
if has_approval:
|
|
status.append("approved")
|
|
if has_db:
|
|
status.append("has_results")
|
|
|
|
print(f" {study_dir.name:<30} [{', '.join(status) or 'new'}]")
|
|
|
|
return 0
|
|
|
|
|
|
def cmd_finalize(args):
|
|
"""Generate final report for a study."""
|
|
print(f"Finalize command not yet implemented for: {args.study}")
|
|
print("This will generate the interactive HTML report.")
|
|
return 0
|
|
|
|
|
|
def create_parser() -> argparse.ArgumentParser:
|
|
"""Create the argument parser."""
|
|
parser = argparse.ArgumentParser(
|
|
prog="atomizer",
|
|
description="Atomizer - FEA Optimization Command Line Interface",
|
|
)
|
|
parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
|
|
parser.add_argument("-q", "--quiet", action="store_true", help="Minimal output")
|
|
|
|
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
|
|
|
# intake command
|
|
intake_parser = subparsers.add_parser("intake", help="Process an intake folder")
|
|
intake_parser.add_argument("folder", help="Path to intake folder")
|
|
intake_parser.add_argument("--skip-baseline", action="store_true", help="Skip baseline solve")
|
|
intake_parser.add_argument(
|
|
"--interview", action="store_true", help="Continue to interview mode"
|
|
)
|
|
intake_parser.add_argument("--canvas", action="store_true", help="Open in canvas mode")
|
|
intake_parser.set_defaults(func=cmd_intake)
|
|
|
|
# validate command
|
|
validate_parser = subparsers.add_parser("validate", help="Validate a study")
|
|
validate_parser.add_argument("study", help="Study name or path")
|
|
validate_parser.add_argument("--skip-trials", action="store_true", help="Skip test trials")
|
|
validate_parser.add_argument("--trials", type=int, default=3, help="Number of test trials")
|
|
validate_parser.add_argument(
|
|
"--approve", action="store_true", help="Approve if validation passes"
|
|
)
|
|
validate_parser.set_defaults(func=cmd_validate)
|
|
|
|
# list command
|
|
list_parser = subparsers.add_parser("list", help="List studies")
|
|
list_parser.set_defaults(func=cmd_list)
|
|
|
|
# finalize command
|
|
finalize_parser = subparsers.add_parser("finalize", help="Generate final report")
|
|
finalize_parser.add_argument("study", help="Study name or path")
|
|
finalize_parser.add_argument("--format", choices=["html", "pdf", "all"], default="html")
|
|
finalize_parser.set_defaults(func=cmd_finalize)
|
|
|
|
return parser
|
|
|
|
|
|
def main(args=None):
|
|
"""Main entry point."""
|
|
parser = create_parser()
|
|
parsed_args = parser.parse_args(args)
|
|
|
|
setup_logging(getattr(parsed_args, "verbose", False))
|
|
|
|
if parsed_args.command is None:
|
|
parser.print_help()
|
|
return 0
|
|
|
|
return parsed_args.func(parsed_args)
|
|
|
|
|
|
# For typer/click compatibility
|
|
app = main
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|