Spaces:
Running on Zero
Running on Zero
| import pandas as pd | |
| import numpy as np | |
| import os | |
| import ezdxf | |
| from pathlib import Path | |
| def process_mesh(input_csv_path: str, dxf_dir_path: str, max_faces=500, precision=2) -> str: | |
| """ | |
| V66 Geometry Engine: Optimized for Low-Disk usage on Hugging Face. | |
| Processes coordinate lines directly from dynamic session paths. | |
| """ | |
| print(f" -> Processing Optimized Mesh from: {dxf_dir_path}") | |
| dxf_dir = Path(dxf_dir_path) | |
| if not dxf_dir.exists(): | |
| raise FileNotFoundError(f"DXF directory not found at {dxf_dir}") | |
| all_points = [] | |
| dxf_files = sorted(list(dxf_dir.glob("*.dxf"))) | |
| # --- STRATEGY 1: FRAME SAMPLING --- | |
| if len(dxf_files) > 60: | |
| print(f" -> High frame count ({len(dxf_files)}). Downsampling for performance.") | |
| dxf_files = dxf_files[::2] | |
| for dxf_file in dxf_files: | |
| try: | |
| # Safely isolate frame index numerical components | |
| frame_id = int(''.join(filter(str.isdigit, dxf_file.stem)) or 0) | |
| doc = ezdxf.readfile(dxf_file) | |
| msp = doc.modelspace() | |
| # --- STRATEGY 2: VERTEX DECIMATION --- | |
| entities = msp.query('LINE') | |
| sampling_rate = 2 if len(entities) > 500 else 1 | |
| for i, entity in enumerate(entities): | |
| if i % sampling_rate != 0: | |
| continue | |
| all_points.append({'t': frame_id, 'x': entity.dxf.start.x, 'y': entity.dxf.start.y, 'z': entity.dxf.start.z}) | |
| all_points.append({'t': frame_id, 'x': entity.dxf.end.x, 'y': entity.dxf.end.y, 'z': entity.dxf.end.z}) | |
| except Exception: | |
| continue | |
| if not all_points: | |
| raise ValueError("No valid point geometry vectors could be parsed from DXF tracking layers.") | |
| df_vertices = pd.DataFrame(all_points) | |
| final_results = [] | |
| center_x, center_y = df_vertices['x'].mean(), df_vertices['y'].mean() | |
| for frame_id, group in df_vertices.groupby('t'): | |
| if len(group) < 2: continue | |
| group = group.drop_duplicates(subset=['x', 'y']).copy() | |
| group['dist'] = np.sqrt((group['x'] - center_x)**2 + (group['y'] - center_y)**2) | |
| group = group.sort_values('dist') | |
| # --- STRATEGY 3: FACE LIMITING --- | |
| window_size = 2 | |
| rolling = group.rolling(window=window_size) | |
| for i, window in enumerate(rolling): | |
| if len(window) < window_size or i > max_faces: | |
| continue | |
| pts = window[['x', 'y', 'z']].values | |
| cp_x, cp_y = np.mean(pts[:, 0]), np.mean(pts[:, 1]) | |
| avg_dist = np.mean(window['dist']) | |
| d_min, d_max = group['dist'].min(), group['dist'].max() | |
| span = (d_max - d_min) if d_max != d_min else 1 | |
| norm = (avg_dist - d_min) / span | |
| r, g, b = int(255 * norm), int(255 * (1 - norm)), int(255 * (0.5 + 0.5 * np.sin(norm * np.pi))) | |
| dynamic_z = float(frame_id + 1.0) * (r + g + b) | |
| v1 = pts[1] - pts[0] | |
| v2 = np.array([0, 0, dynamic_z]) | |
| normal = np.cross(v1, v2) | |
| norm_val = np.linalg.norm(normal) | |
| normal_unit = normal / norm_val if norm_val > 1e-9 else np.array([0, 0, 1]) | |
| # --- STRATEGY 4: STRING FORMATTING & ROUNDING --- | |
| row = (f"{frame_id}," | |
| f"{cp_x:.{precision}f}," | |
| f"{cp_y:.{precision}f}," | |
| f"{dynamic_z:.1f}," | |
| f"{r},{g},{b}," | |
| f"{normal_unit[0]:.2f},{normal_unit[1]:.2f},{normal_unit[2]:.2f}," | |
| f"{np.linalg.norm(v1):.2f}," | |
| f"{np.arccos(np.clip(np.dot([0,0,1], normal_unit), -1.0, 1.0)):.2f}") | |
| final_results.append(row) | |
| # Save output to the exact same temporary folder directory as the input CSV path | |
| output_path = os.path.dirname(input_csv_path) + "/final_12d_features.csv" | |
| with open(output_path, 'w') as f: | |
| f.write("t,x,y,z,R,G,B,N_x,N_y,N_z,d,th\n") | |
| f.write("\n".join(final_results)) | |
| return output_path |