feat: Multi-file introspection for FEM/SIM/PRT with PyNastran parsing
This commit is contained in:
@@ -5282,21 +5282,126 @@ async def get_nx_expressions(study_id: str):
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get expressions: {str(e)}")
|
||||
|
||||
|
||||
@router.get("/studies/{study_id}/nx/introspect/{part_name}")
|
||||
async def introspect_specific_part(study_id: str, part_name: str, force: bool = False):
|
||||
def introspect_fem_file(fem_path: Path) -> dict:
|
||||
"""
|
||||
Introspect a specific .prt file in the model directory.
|
||||
Introspect a .fem or .afm file using PyNastran.
|
||||
|
||||
Use this to get expressions from component parts (e.g., M1_Blank.prt)
|
||||
rather than just the main assembly.
|
||||
Extracts: element counts, material info, property info, node count, etc.
|
||||
"""
|
||||
result = {
|
||||
"file_type": "fem",
|
||||
"file_name": fem_path.name,
|
||||
"success": False,
|
||||
}
|
||||
|
||||
try:
|
||||
from pyNastran.bdf.bdf import BDF
|
||||
|
||||
# Read BDF/FEM file
|
||||
bdf = BDF()
|
||||
bdf.read_bdf(str(fem_path), xref=False)
|
||||
|
||||
# Extract info
|
||||
result["success"] = True
|
||||
result["nodes"] = {
|
||||
"count": len(bdf.nodes),
|
||||
}
|
||||
|
||||
# Element counts by type
|
||||
element_counts = {}
|
||||
for eid, elem in bdf.elements.items():
|
||||
elem_type = elem.type
|
||||
element_counts[elem_type] = element_counts.get(elem_type, 0) + 1
|
||||
result["elements"] = {
|
||||
"total": len(bdf.elements),
|
||||
"by_type": element_counts,
|
||||
}
|
||||
|
||||
# Materials
|
||||
materials = []
|
||||
for mid, mat in bdf.materials.items():
|
||||
mat_info = {"id": mid, "type": mat.type}
|
||||
if hasattr(mat, "e"):
|
||||
mat_info["E"] = mat.e
|
||||
if hasattr(mat, "nu"):
|
||||
mat_info["nu"] = mat.nu
|
||||
if hasattr(mat, "rho"):
|
||||
mat_info["rho"] = mat.rho
|
||||
materials.append(mat_info)
|
||||
result["materials"] = materials
|
||||
|
||||
# Properties
|
||||
properties = []
|
||||
for pid, prop in bdf.properties.items():
|
||||
prop_info = {"id": pid, "type": prop.type}
|
||||
properties.append(prop_info)
|
||||
result["properties"] = properties
|
||||
|
||||
# Coordinate systems
|
||||
result["coord_systems"] = len(bdf.coords)
|
||||
|
||||
# Loads and constraints summary
|
||||
result["loads"] = {
|
||||
"load_cases": len(bdf.loads),
|
||||
}
|
||||
result["constraints"] = {
|
||||
"spcs": len(bdf.spcs),
|
||||
"mpcs": len(bdf.mpcs),
|
||||
}
|
||||
|
||||
except ImportError:
|
||||
result["error"] = "PyNastran not available"
|
||||
except Exception as e:
|
||||
result["error"] = str(e)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def introspect_sim_file(sim_path: Path) -> dict:
|
||||
"""
|
||||
Introspect a .sim file - extract solution info.
|
||||
|
||||
Note: .sim is a binary NX format, so we extract what we can from associated files.
|
||||
"""
|
||||
result = {
|
||||
"file_type": "sim",
|
||||
"file_name": sim_path.name,
|
||||
"success": True,
|
||||
"note": "SIM files are binary NX format. Use NX introspection for full details.",
|
||||
}
|
||||
|
||||
# Check for associated .dat file that might have been exported
|
||||
dat_file = sim_path.parent / (sim_path.stem + ".dat")
|
||||
if not dat_file.exists():
|
||||
# Try common naming patterns
|
||||
for f in sim_path.parent.glob("*solution*.dat"):
|
||||
dat_file = f
|
||||
break
|
||||
|
||||
if dat_file.exists():
|
||||
result["associated_dat"] = dat_file.name
|
||||
# Could parse the DAT file for solution info if needed
|
||||
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/studies/{study_id}/nx/introspect/{file_name}")
|
||||
async def introspect_specific_file(study_id: str, file_name: str, force: bool = False):
|
||||
"""
|
||||
Introspect a specific file in the model directory.
|
||||
|
||||
Supports ALL NX file types:
|
||||
- .prt / _i.prt - Uses NX journal to extract expressions, mass, etc.
|
||||
- .fem / .afm - Uses PyNastran to extract mesh info, materials, properties
|
||||
- .sim - Extracts simulation/solution info
|
||||
|
||||
Args:
|
||||
study_id: Study identifier
|
||||
part_name: Name of the part file (e.g., "M1_Blank.prt" or "M1_Blank")
|
||||
file_name: Name of the file (e.g., "M1_Blank.prt", "M1_Blank_fem1.fem")
|
||||
force: Force re-introspection even if cached
|
||||
|
||||
Returns:
|
||||
JSON with introspection results for the specific part
|
||||
JSON with introspection results appropriate for the file type
|
||||
"""
|
||||
try:
|
||||
study_dir = resolve_study_path(study_id)
|
||||
@@ -5318,74 +5423,124 @@ async def introspect_specific_part(study_id: str, part_name: str, force: bool =
|
||||
if model_dir is None:
|
||||
raise HTTPException(status_code=404, detail=f"No model directory found for {study_id}")
|
||||
|
||||
# Normalize part name
|
||||
if not part_name.lower().endswith(".prt"):
|
||||
part_name = part_name + ".prt"
|
||||
# Determine file type and find the file
|
||||
file_name_lower = file_name.lower()
|
||||
target_file = None
|
||||
file_type = None
|
||||
|
||||
# Find the part file (case-insensitive)
|
||||
prt_file = None
|
||||
for f in model_dir.glob("*.prt"):
|
||||
if f.name.lower() == part_name.lower():
|
||||
prt_file = f
|
||||
break
|
||||
# Add extension if missing
|
||||
if "." not in file_name:
|
||||
# Try common extensions in order
|
||||
for ext in [".prt", ".fem", ".afm", ".sim"]:
|
||||
candidate = model_dir / (file_name + ext)
|
||||
if candidate.exists():
|
||||
target_file = candidate
|
||||
file_type = ext[1:] # Remove leading dot
|
||||
break
|
||||
else:
|
||||
# Has extension, find exact match (case-insensitive)
|
||||
for f in model_dir.iterdir():
|
||||
if f.name.lower() == file_name_lower:
|
||||
target_file = f
|
||||
if "_i.prt" in f.name.lower():
|
||||
file_type = "idealized"
|
||||
elif f.suffix.lower() == ".prt":
|
||||
file_type = "prt"
|
||||
elif f.suffix.lower() == ".fem":
|
||||
file_type = "fem"
|
||||
elif f.suffix.lower() == ".afm":
|
||||
file_type = "afm"
|
||||
elif f.suffix.lower() == ".sim":
|
||||
file_type = "sim"
|
||||
break
|
||||
|
||||
if prt_file is None:
|
||||
# List available parts for helpful error
|
||||
available = [f.name for f in model_dir.glob("*.prt")]
|
||||
if target_file is None:
|
||||
# List available files for helpful error
|
||||
available = [
|
||||
f.name
|
||||
for f in model_dir.iterdir()
|
||||
if f.is_file() and f.suffix.lower() in [".prt", ".fem", ".afm", ".sim"]
|
||||
]
|
||||
raise HTTPException(
|
||||
status_code=404, detail=f"Part '{part_name}' not found. Available: {available}"
|
||||
status_code=404, detail=f"File '{file_name}' not found. Available: {available}"
|
||||
)
|
||||
|
||||
# Check cache
|
||||
cache_file = model_dir / f"_introspection_{prt_file.stem}.json"
|
||||
cache_file = model_dir / f"_introspection_{target_file.stem}.json"
|
||||
if cache_file.exists() and not force:
|
||||
try:
|
||||
with open(cache_file, "r") as f:
|
||||
cached = json.load(f)
|
||||
return {
|
||||
"study_id": study_id,
|
||||
"part_name": prt_file.name,
|
||||
"file_name": target_file.name,
|
||||
"file_type": file_type,
|
||||
"cached": True,
|
||||
"introspection": cached,
|
||||
}
|
||||
except:
|
||||
pass
|
||||
|
||||
# Run introspection
|
||||
# Run appropriate introspection based on file type
|
||||
result = None
|
||||
|
||||
if file_type in ["prt", "idealized"]:
|
||||
# Use NX journal introspection for .prt files
|
||||
try:
|
||||
from optimization_engine.extractors.introspect_part import introspect_part
|
||||
|
||||
result = introspect_part(str(target_file), str(model_dir), verbose=False)
|
||||
except ImportError:
|
||||
result = {"success": False, "error": "introspect_part module not available"}
|
||||
except Exception as e:
|
||||
result = {"success": False, "error": str(e)}
|
||||
|
||||
elif file_type in ["fem", "afm"]:
|
||||
# Use PyNastran for FEM files
|
||||
result = introspect_fem_file(target_file)
|
||||
|
||||
elif file_type == "sim":
|
||||
# Extract what we can from SIM file
|
||||
result = introspect_sim_file(target_file)
|
||||
|
||||
if result is None:
|
||||
result = {"success": False, "error": f"Unknown file type: {file_type}"}
|
||||
|
||||
# Cache results
|
||||
try:
|
||||
from optimization_engine.extractors.introspect_part import introspect_part
|
||||
|
||||
result = introspect_part(str(prt_file), str(model_dir), verbose=False)
|
||||
|
||||
# Cache results
|
||||
with open(cache_file, "w") as f:
|
||||
json.dump(result, f, indent=2)
|
||||
except Exception:
|
||||
pass # Caching is optional
|
||||
|
||||
return {
|
||||
"study_id": study_id,
|
||||
"part_name": prt_file.name,
|
||||
"cached": False,
|
||||
"introspection": result,
|
||||
}
|
||||
|
||||
except ImportError:
|
||||
raise HTTPException(status_code=500, detail="introspect_part module not available")
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Introspection failed: {str(e)}")
|
||||
return {
|
||||
"study_id": study_id,
|
||||
"file_name": target_file.name,
|
||||
"file_type": file_type,
|
||||
"cached": False,
|
||||
"introspection": result,
|
||||
}
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Failed to introspect part: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to introspect file: {str(e)}")
|
||||
|
||||
|
||||
@router.get("/studies/{study_id}/nx/parts")
|
||||
async def list_model_parts(study_id: str):
|
||||
async def list_model_files(study_id: str):
|
||||
"""
|
||||
List all .prt files in the study's model directory.
|
||||
List all NX model files in the study's model directory.
|
||||
|
||||
Includes ALL file types in the NX dependency chain:
|
||||
- .sim (Simulation) - loads, BCs, solution settings
|
||||
- .afm (Assembly FEM) - assembly of component FEMs
|
||||
- .fem (FEM) - mesh, materials, properties
|
||||
- _i.prt (Idealized) - simplified geometry for meshing
|
||||
- .prt (Geometry) - full CAD geometry with expressions
|
||||
|
||||
Returns:
|
||||
JSON with list of available parts that can be introspected
|
||||
JSON with categorized list of all model files
|
||||
"""
|
||||
try:
|
||||
study_dir = resolve_study_path(study_id)
|
||||
@@ -5407,28 +5562,62 @@ async def list_model_parts(study_id: str):
|
||||
if model_dir is None:
|
||||
raise HTTPException(status_code=404, detail=f"No model directory found for {study_id}")
|
||||
|
||||
# Collect all part files
|
||||
parts = []
|
||||
for f in sorted(model_dir.glob("*.prt")):
|
||||
is_idealized = "_i.prt" in f.name.lower()
|
||||
parts.append(
|
||||
{
|
||||
"name": f.name,
|
||||
"stem": f.stem,
|
||||
"is_idealized": is_idealized,
|
||||
"size_kb": round(f.stat().st_size / 1024, 1),
|
||||
"has_cache": (model_dir / f"_introspection_{f.stem}.json").exists(),
|
||||
}
|
||||
)
|
||||
# Collect all NX files by type
|
||||
files = {
|
||||
"sim": [], # Simulation files
|
||||
"afm": [], # Assembly FEM files
|
||||
"fem": [], # FEM files
|
||||
"idealized": [], # Idealized parts (*_i.prt)
|
||||
"prt": [], # Geometry parts
|
||||
}
|
||||
|
||||
for f in sorted(model_dir.iterdir()):
|
||||
if not f.is_file():
|
||||
continue
|
||||
|
||||
name_lower = f.name.lower()
|
||||
file_info = {
|
||||
"name": f.name,
|
||||
"stem": f.stem,
|
||||
"size_kb": round(f.stat().st_size / 1024, 1),
|
||||
"has_cache": (model_dir / f"_introspection_{f.stem}.json").exists(),
|
||||
}
|
||||
|
||||
if name_lower.endswith(".sim"):
|
||||
file_info["type"] = "sim"
|
||||
file_info["description"] = "Simulation (loads, BCs, solutions)"
|
||||
files["sim"].append(file_info)
|
||||
elif name_lower.endswith(".afm"):
|
||||
file_info["type"] = "afm"
|
||||
file_info["description"] = "Assembly FEM"
|
||||
files["afm"].append(file_info)
|
||||
elif name_lower.endswith(".fem"):
|
||||
file_info["type"] = "fem"
|
||||
file_info["description"] = "FEM (mesh, materials, properties)"
|
||||
files["fem"].append(file_info)
|
||||
elif "_i.prt" in name_lower:
|
||||
file_info["type"] = "idealized"
|
||||
file_info["description"] = "Idealized geometry for meshing"
|
||||
files["idealized"].append(file_info)
|
||||
elif name_lower.endswith(".prt"):
|
||||
file_info["type"] = "prt"
|
||||
file_info["description"] = "CAD geometry with expressions"
|
||||
files["prt"].append(file_info)
|
||||
|
||||
# Build flat list for dropdown (all introspectable files)
|
||||
all_files = []
|
||||
for category in ["sim", "afm", "fem", "idealized", "prt"]:
|
||||
all_files.extend(files[category])
|
||||
|
||||
return {
|
||||
"study_id": study_id,
|
||||
"model_dir": str(model_dir),
|
||||
"parts": parts,
|
||||
"count": len(parts),
|
||||
"files": files,
|
||||
"all_files": all_files,
|
||||
"count": len(all_files),
|
||||
}
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Failed to list parts: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to list model files: {str(e)}")
|
||||
|
||||
Reference in New Issue
Block a user