Anto01
ba0b9a1fae
Backend: - spec.py: New AtomizerSpec REST API endpoints - spec_manager.py: SpecManager service for unified config - interview_engine.py: Study creation interview logic - claude.py: Enhanced Claude API with context - optimization.py: Extended optimization endpoints - context_builder.py, session_manager.py: Improved services Frontend: - Chat components: Enhanced message rendering, tool call cards - Hooks: useClaudeCode, useSpecWebSocket, improved useChat - Pages: Updated Dashboard, Analysis, Insights, Setup, Home - Components: ParallelCoordinatesPlot, ParetoPlot improvements - App.tsx: Route updates for canvas/studio Infrastructure: - vite.config.ts: Build configuration updates - start/stop-dashboard.bat: Script improvements
455 lines
17 KiB
Python
455 lines
17 KiB
Python
"""
|
|
Interview Engine - Guided Study Creation through Conversation
|
|
|
|
Provides a structured interview flow for creating optimization studies.
|
|
Claude uses this to gather information step-by-step, building a complete
|
|
atomizer_spec.json through natural conversation.
|
|
"""
|
|
|
|
from typing import Dict, Any, List, Optional, Literal
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from datetime import datetime
|
|
import json
|
|
|
|
|
|
class InterviewState(str, Enum):
|
|
"""Current phase of the interview"""
|
|
NOT_STARTED = "not_started"
|
|
GATHERING_BASICS = "gathering_basics" # Name, description, goals
|
|
GATHERING_MODEL = "gathering_model" # Model file, solver type
|
|
GATHERING_VARIABLES = "gathering_variables" # Design variables
|
|
GATHERING_EXTRACTORS = "gathering_extractors" # Physics extractors
|
|
GATHERING_OBJECTIVES = "gathering_objectives" # Objectives
|
|
GATHERING_CONSTRAINTS = "gathering_constraints" # Constraints
|
|
GATHERING_SETTINGS = "gathering_settings" # Algorithm, trials
|
|
REVIEW = "review" # Review before creation
|
|
COMPLETED = "completed"
|
|
|
|
|
|
@dataclass
|
|
class InterviewData:
|
|
"""Accumulated data from the interview"""
|
|
# Basics
|
|
study_name: Optional[str] = None
|
|
category: Optional[str] = None
|
|
description: Optional[str] = None
|
|
goals: List[str] = field(default_factory=list)
|
|
|
|
# Model
|
|
sim_file: Optional[str] = None
|
|
prt_file: Optional[str] = None
|
|
solver_type: str = "nastran"
|
|
|
|
# Design variables
|
|
design_variables: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
# Extractors
|
|
extractors: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
# Objectives
|
|
objectives: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
# Constraints
|
|
constraints: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
# Settings
|
|
algorithm: str = "TPE"
|
|
max_trials: int = 100
|
|
|
|
def to_spec(self) -> Dict[str, Any]:
|
|
"""Convert interview data to atomizer_spec.json format"""
|
|
# Generate IDs for each element
|
|
dvs_with_ids = []
|
|
for i, dv in enumerate(self.design_variables):
|
|
dv_copy = dv.copy()
|
|
dv_copy['id'] = f"dv_{i+1:03d}"
|
|
dv_copy['canvas_position'] = {'x': 50, 'y': 100 + i * 80}
|
|
dvs_with_ids.append(dv_copy)
|
|
|
|
exts_with_ids = []
|
|
for i, ext in enumerate(self.extractors):
|
|
ext_copy = ext.copy()
|
|
ext_copy['id'] = f"ext_{i+1:03d}"
|
|
ext_copy['canvas_position'] = {'x': 400, 'y': 100 + i * 80}
|
|
exts_with_ids.append(ext_copy)
|
|
|
|
objs_with_ids = []
|
|
for i, obj in enumerate(self.objectives):
|
|
obj_copy = obj.copy()
|
|
obj_copy['id'] = f"obj_{i+1:03d}"
|
|
obj_copy['canvas_position'] = {'x': 750, 'y': 100 + i * 80}
|
|
objs_with_ids.append(obj_copy)
|
|
|
|
cons_with_ids = []
|
|
for i, con in enumerate(self.constraints):
|
|
con_copy = con.copy()
|
|
con_copy['id'] = f"con_{i+1:03d}"
|
|
con_copy['canvas_position'] = {'x': 750, 'y': 400 + i * 80}
|
|
cons_with_ids.append(con_copy)
|
|
|
|
return {
|
|
"meta": {
|
|
"version": "2.0",
|
|
"study_name": self.study_name or "untitled_study",
|
|
"description": self.description or "",
|
|
"created_at": datetime.now().isoformat(),
|
|
"created_by": "interview",
|
|
"modified_at": datetime.now().isoformat(),
|
|
"modified_by": "interview"
|
|
},
|
|
"model": {
|
|
"sim": {
|
|
"path": self.sim_file or "",
|
|
"solver": self.solver_type
|
|
}
|
|
},
|
|
"design_variables": dvs_with_ids,
|
|
"extractors": exts_with_ids,
|
|
"objectives": objs_with_ids,
|
|
"constraints": cons_with_ids,
|
|
"optimization": {
|
|
"algorithm": {
|
|
"type": self.algorithm
|
|
},
|
|
"budget": {
|
|
"max_trials": self.max_trials
|
|
}
|
|
},
|
|
"canvas": {
|
|
"edges": [],
|
|
"layout_version": "2.0"
|
|
}
|
|
}
|
|
|
|
|
|
class InterviewEngine:
|
|
"""
|
|
Manages the interview flow for study creation.
|
|
|
|
Usage:
|
|
1. Create engine: engine = InterviewEngine()
|
|
2. Start interview: engine.start()
|
|
3. Record answers: engine.record_answer("study_name", "bracket_opt")
|
|
4. Check progress: engine.get_progress()
|
|
5. Generate spec: engine.finalize()
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.state = InterviewState.NOT_STARTED
|
|
self.data = InterviewData()
|
|
self.questions_asked: List[str] = []
|
|
self.errors: List[str] = []
|
|
|
|
def start(self) -> Dict[str, Any]:
|
|
"""Start the interview process"""
|
|
self.state = InterviewState.GATHERING_BASICS
|
|
return {
|
|
"state": self.state.value,
|
|
"message": "Let's create a new optimization study! I'll guide you through the process.",
|
|
"next_questions": self.get_current_questions()
|
|
}
|
|
|
|
def get_current_questions(self) -> List[Dict[str, Any]]:
|
|
"""Get the questions for the current interview state"""
|
|
questions = {
|
|
InterviewState.GATHERING_BASICS: [
|
|
{
|
|
"field": "study_name",
|
|
"question": "What would you like to name this study?",
|
|
"hint": "Use snake_case, e.g., 'bracket_mass_optimization'",
|
|
"required": True
|
|
},
|
|
{
|
|
"field": "category",
|
|
"question": "What category should this study be in?",
|
|
"hint": "e.g., 'Simple_Bracket', 'M1_Mirror', or leave blank for root",
|
|
"required": False
|
|
},
|
|
{
|
|
"field": "description",
|
|
"question": "Briefly describe what you're trying to optimize",
|
|
"hint": "e.g., 'Minimize bracket mass while maintaining stiffness'",
|
|
"required": True
|
|
}
|
|
],
|
|
InterviewState.GATHERING_MODEL: [
|
|
{
|
|
"field": "sim_file",
|
|
"question": "What is the path to your simulation (.sim) file?",
|
|
"hint": "Relative path from the study folder, e.g., '1_setup/Model_sim1.sim'",
|
|
"required": True
|
|
}
|
|
],
|
|
InterviewState.GATHERING_VARIABLES: [
|
|
{
|
|
"field": "design_variable",
|
|
"question": "What parameters do you want to optimize?",
|
|
"hint": "Tell me the NX expression names and their bounds",
|
|
"required": True,
|
|
"multi": True
|
|
}
|
|
],
|
|
InterviewState.GATHERING_EXTRACTORS: [
|
|
{
|
|
"field": "extractor",
|
|
"question": "What physics quantities do you want to extract from FEA?",
|
|
"hint": "e.g., mass, max displacement, max stress, frequency, Zernike WFE",
|
|
"required": True,
|
|
"multi": True
|
|
}
|
|
],
|
|
InterviewState.GATHERING_OBJECTIVES: [
|
|
{
|
|
"field": "objective",
|
|
"question": "What do you want to optimize?",
|
|
"hint": "Tell me which extracted quantities to minimize or maximize",
|
|
"required": True,
|
|
"multi": True
|
|
}
|
|
],
|
|
InterviewState.GATHERING_CONSTRAINTS: [
|
|
{
|
|
"field": "constraint",
|
|
"question": "Do you have any constraints? (e.g., max stress, min frequency)",
|
|
"hint": "You can say 'none' if you don't have any",
|
|
"required": False,
|
|
"multi": True
|
|
}
|
|
],
|
|
InterviewState.GATHERING_SETTINGS: [
|
|
{
|
|
"field": "algorithm",
|
|
"question": "Which optimization algorithm would you like to use?",
|
|
"hint": "Options: TPE (default), CMA-ES, NSGA-II, RandomSearch",
|
|
"required": False
|
|
},
|
|
{
|
|
"field": "max_trials",
|
|
"question": "How many trials (FEA evaluations) should we run?",
|
|
"hint": "Default is 100. More trials = better results but longer runtime",
|
|
"required": False
|
|
}
|
|
],
|
|
InterviewState.REVIEW: [
|
|
{
|
|
"field": "confirm",
|
|
"question": "Does this configuration look correct? (yes/no)",
|
|
"required": True
|
|
}
|
|
]
|
|
}
|
|
return questions.get(self.state, [])
|
|
|
|
def record_answer(self, field: str, value: Any) -> Dict[str, Any]:
|
|
"""Record an answer and potentially advance the state"""
|
|
self.questions_asked.append(field)
|
|
|
|
# Handle different field types
|
|
if field == "study_name":
|
|
self.data.study_name = value
|
|
elif field == "category":
|
|
self.data.category = value if value else None
|
|
elif field == "description":
|
|
self.data.description = value
|
|
elif field == "sim_file":
|
|
self.data.sim_file = value
|
|
elif field == "design_variable":
|
|
# Value should be a dict with name, min, max, etc.
|
|
if isinstance(value, dict):
|
|
self.data.design_variables.append(value)
|
|
elif isinstance(value, list):
|
|
self.data.design_variables.extend(value)
|
|
elif field == "extractor":
|
|
if isinstance(value, dict):
|
|
self.data.extractors.append(value)
|
|
elif isinstance(value, list):
|
|
self.data.extractors.extend(value)
|
|
elif field == "objective":
|
|
if isinstance(value, dict):
|
|
self.data.objectives.append(value)
|
|
elif isinstance(value, list):
|
|
self.data.objectives.extend(value)
|
|
elif field == "constraint":
|
|
if value and value.lower() not in ["none", "no", "skip"]:
|
|
if isinstance(value, dict):
|
|
self.data.constraints.append(value)
|
|
elif isinstance(value, list):
|
|
self.data.constraints.extend(value)
|
|
elif field == "algorithm":
|
|
if value in ["TPE", "CMA-ES", "NSGA-II", "RandomSearch"]:
|
|
self.data.algorithm = value
|
|
elif field == "max_trials":
|
|
try:
|
|
self.data.max_trials = int(value)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
elif field == "confirm":
|
|
if value.lower() in ["yes", "y", "confirm", "ok"]:
|
|
self.state = InterviewState.COMPLETED
|
|
|
|
return {
|
|
"state": self.state.value,
|
|
"recorded": {field: value},
|
|
"data_so_far": self.get_summary()
|
|
}
|
|
|
|
def advance_state(self) -> Dict[str, Any]:
|
|
"""Advance to the next interview state"""
|
|
state_order = [
|
|
InterviewState.NOT_STARTED,
|
|
InterviewState.GATHERING_BASICS,
|
|
InterviewState.GATHERING_MODEL,
|
|
InterviewState.GATHERING_VARIABLES,
|
|
InterviewState.GATHERING_EXTRACTORS,
|
|
InterviewState.GATHERING_OBJECTIVES,
|
|
InterviewState.GATHERING_CONSTRAINTS,
|
|
InterviewState.GATHERING_SETTINGS,
|
|
InterviewState.REVIEW,
|
|
InterviewState.COMPLETED
|
|
]
|
|
|
|
current_idx = state_order.index(self.state)
|
|
if current_idx < len(state_order) - 1:
|
|
self.state = state_order[current_idx + 1]
|
|
|
|
return {
|
|
"state": self.state.value,
|
|
"next_questions": self.get_current_questions()
|
|
}
|
|
|
|
def get_summary(self) -> Dict[str, Any]:
|
|
"""Get a summary of collected data"""
|
|
return {
|
|
"study_name": self.data.study_name,
|
|
"category": self.data.category,
|
|
"description": self.data.description,
|
|
"model": self.data.sim_file,
|
|
"design_variables": len(self.data.design_variables),
|
|
"extractors": len(self.data.extractors),
|
|
"objectives": len(self.data.objectives),
|
|
"constraints": len(self.data.constraints),
|
|
"algorithm": self.data.algorithm,
|
|
"max_trials": self.data.max_trials
|
|
}
|
|
|
|
def get_progress(self) -> Dict[str, Any]:
|
|
"""Get interview progress information"""
|
|
state_progress = {
|
|
InterviewState.NOT_STARTED: 0,
|
|
InterviewState.GATHERING_BASICS: 15,
|
|
InterviewState.GATHERING_MODEL: 25,
|
|
InterviewState.GATHERING_VARIABLES: 40,
|
|
InterviewState.GATHERING_EXTRACTORS: 55,
|
|
InterviewState.GATHERING_OBJECTIVES: 70,
|
|
InterviewState.GATHERING_CONSTRAINTS: 80,
|
|
InterviewState.GATHERING_SETTINGS: 90,
|
|
InterviewState.REVIEW: 95,
|
|
InterviewState.COMPLETED: 100
|
|
}
|
|
|
|
return {
|
|
"state": self.state.value,
|
|
"progress_percent": state_progress.get(self.state, 0),
|
|
"summary": self.get_summary(),
|
|
"current_questions": self.get_current_questions()
|
|
}
|
|
|
|
def validate(self) -> Dict[str, Any]:
|
|
"""Validate the collected data before finalizing"""
|
|
errors = []
|
|
warnings = []
|
|
|
|
# Required fields
|
|
if not self.data.study_name:
|
|
errors.append("Study name is required")
|
|
|
|
if not self.data.design_variables:
|
|
errors.append("At least one design variable is required")
|
|
|
|
if not self.data.extractors:
|
|
errors.append("At least one extractor is required")
|
|
|
|
if not self.data.objectives:
|
|
errors.append("At least one objective is required")
|
|
|
|
# Warnings
|
|
if not self.data.sim_file:
|
|
warnings.append("No simulation file specified - you'll need to add one manually")
|
|
|
|
if not self.data.constraints:
|
|
warnings.append("No constraints defined - optimization will be unconstrained")
|
|
|
|
return {
|
|
"valid": len(errors) == 0,
|
|
"errors": errors,
|
|
"warnings": warnings
|
|
}
|
|
|
|
def finalize(self) -> Dict[str, Any]:
|
|
"""Generate the final atomizer_spec.json"""
|
|
validation = self.validate()
|
|
|
|
if not validation["valid"]:
|
|
return {
|
|
"success": False,
|
|
"errors": validation["errors"]
|
|
}
|
|
|
|
spec = self.data.to_spec()
|
|
|
|
return {
|
|
"success": True,
|
|
"spec": spec,
|
|
"warnings": validation.get("warnings", [])
|
|
}
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Serialize engine state for persistence"""
|
|
return {
|
|
"state": self.state.value,
|
|
"data": {
|
|
"study_name": self.data.study_name,
|
|
"category": self.data.category,
|
|
"description": self.data.description,
|
|
"goals": self.data.goals,
|
|
"sim_file": self.data.sim_file,
|
|
"prt_file": self.data.prt_file,
|
|
"solver_type": self.data.solver_type,
|
|
"design_variables": self.data.design_variables,
|
|
"extractors": self.data.extractors,
|
|
"objectives": self.data.objectives,
|
|
"constraints": self.data.constraints,
|
|
"algorithm": self.data.algorithm,
|
|
"max_trials": self.data.max_trials
|
|
},
|
|
"questions_asked": self.questions_asked,
|
|
"errors": self.errors
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "InterviewEngine":
|
|
"""Restore engine from serialized state"""
|
|
engine = cls()
|
|
engine.state = InterviewState(data.get("state", "not_started"))
|
|
|
|
d = data.get("data", {})
|
|
engine.data.study_name = d.get("study_name")
|
|
engine.data.category = d.get("category")
|
|
engine.data.description = d.get("description")
|
|
engine.data.goals = d.get("goals", [])
|
|
engine.data.sim_file = d.get("sim_file")
|
|
engine.data.prt_file = d.get("prt_file")
|
|
engine.data.solver_type = d.get("solver_type", "nastran")
|
|
engine.data.design_variables = d.get("design_variables", [])
|
|
engine.data.extractors = d.get("extractors", [])
|
|
engine.data.objectives = d.get("objectives", [])
|
|
engine.data.constraints = d.get("constraints", [])
|
|
engine.data.algorithm = d.get("algorithm", "TPE")
|
|
engine.data.max_trials = d.get("max_trials", 100)
|
|
|
|
engine.questions_asked = data.get("questions_asked", [])
|
|
engine.errors = data.get("errors", [])
|
|
|
|
return engine
|