""" 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