# Use the existing connected PyFluent variable solver. import math task_directory = r"C:\Users\13977\AppData\Local\FluentAGI\data\runs\20261018-142547-90d0a4" s = solver.settings def checked(label, result): print(label, result) if isinstance(result, dict): if result.get("status") in ("error", "failed", "needs_review"): raise RuntimeError(str(result)) if result.get("error") or result.get("errors"): raise RuntimeError(str(result)) return result def numbers(value): if isinstance(value, bool): return [] if isinstance(value, (int, float)): return [float(value)] if isinstance(value, dict): return [number for child in value.values() for number in numbers(child)] if isinstance(value, (list, tuple)): return [number for child in value for number in numbers(child)] return [] def report_value(result, name): if isinstance(result, dict): if name in result: values = numbers(result[name]) if values and math.isfinite(values[0]): return values[0] children = result.values() elif isinstance(result, (list, tuple)): children = result else: children = [] for child in children: try: return report_value(child, name) except ValueError: pass raise ValueError("无法提取报告值：" + name + "；返回：" + str(result)) def scalar(setting, value): state = setting.get_state() if isinstance(state, dict): if "value" not in state: raise RuntimeError("无法识别数值字段：" + str(state)) update = {"value": value} if "option" in state: update["option"] = "value" setting.set_state(update) else: setting.set_state(value) def vector(setting, values): state = setting.get_state() if isinstance(state, (list, tuple)): setting.set_state(values) return if isinstance(state, dict): for keys in [ ["x", "y", "z"], ["x_component", "y_component", "z_component"], ["i", "j", "k"], ]: if all(key in state for key in keys): for key, value in zip(keys, values): scalar(getattr(setting, key), value) return raise RuntimeError("无法识别矢量字段：" + str(state)) if not s.setup.models.is_active(): raise RuntimeError("当前网格未激活。") bc = s.setup.boundary_conditions reports = s.solution.report_definitions wall_names = bc.wall.get_object_names() initial_iteration = s.solution.run_calculation.iter_count.get_state() print("CURRENT_STATE", { "walls": wall_names, "iterations": initial_iteration, "solver": s.setup.general.solver.get_state(), "compute_active": reports.compute.is_active(), }) s.file.write_case( file_name=task_directory + r"\before_report_activation_repair.cas.h5" ) if not reports.compute.is_active(): if initial_iteration != 0: raise RuntimeError("已有迭代但报告不可用，停止以保护结果。") s.solution.initialization.hybrid_initialize() print("INITIALIZATION_PERFORMED", True) else: print("EXISTING_INITIALIZED_STATE_PRESERVED", True) if not reports.compute.is_active(): raise RuntimeError("初始化后报告计算仍不可用。") surface_reports = reports.surface probe_names = [] for axis in ["x", "y", "z"]: for suffix, kind in [ ("min", "surface-vertexmin"), ("max", "surface-vertexmax"), ]: name = "fw_bound_" + axis + "_" + suffix if name not in surface_reports.get_object_names(): surface_reports[name] = {} report = surface_reports[name] report.report_type = kind field = axis + "-coordinate" if field not in report.field.allowed_values(): raise RuntimeError("坐标字段不可用：" + field) report.field = field probe_names.append(name) bounds = {} for zone in wall_names: for name in probe_names: surface_reports[name].surface_names = [zone] result = reports.compute(report_defs=probe_names) bounds[zone] = [report_value(result, name) for name in probe_names] print("WALL_BOUNDS", zone, bounds[zone]) wing, ground, farfield = [], [], [] for zone, bound in bounds.items(): xmin, xmax, ymin, ymax, zmin, zmax = bound if max(abs(ymin), abs(ymax)) < 1e-5: ground.append(zone) elif ( xmin > -0.031 and xmax < 0.491 and ymin > 0.089 and ymax < 0.351 and zmin > -0.619 and zmax < 0.619 ): wing.append(zone) else: farfield.append(zone) print("CLASSIFICATION", {"wing": wing, "ground": ground, "farfield": farfield}) if not wing or len(ground) != 1 or len(farfield) != 3: raise RuntimeError("边界分类与设计不符。") for zone in wing + ground: momentum = bc.wall[zone].momentum momentum.shear_condition = "No Slip" momentum.wall_motion = "Stationary Wall" momentum = bc.wall[ground[0]].momentum momentum.wall_motion = "Moving Wall" for name in ["relative", "rotating", "components"]: child = getattr(momentum, name) if child.is_active(): child.set_state(False) scalar(momentum.speed, 30.0) vector(momentum.direction, [1.0, 0.0, 0.0]) print("GROUND_READBACK", momentum.get_state()) for zone in farfield: momentum = bc.wall[zone].momentum momentum.wall_motion = "Stationary Wall" momentum.shear_condition = "Specified Shear" print("OUTER_WALL_READBACK", zone, momentum.get_state()) inlet = bc.velocity_inlet["inlet"] outlet = bc.pressure_outlet["outlet"] scalar(inlet.momentum.velocity, 30.0) scalar(outlet.momentum.gauge_pressure, 0.0) inlet.turbulence.turbulence_specification = "Intensity and Viscosity Ratio" scalar(inlet.turbulence.turbulent_intensity, 0.01) scalar(inlet.turbulence.turbulent_viscosity_ratio, 10.0) force_names = [] for name, direction in [ ("front_wing_drag_interface", [1.0, 0.0, 0.0]), ("front_wing_lift_interface", [0.0, 1.0, 0.0]), ]: if name not in reports.force.get_object_names(): reports.force[name] = {} report = reports.force[name] report.force_vector = direction report.zones = wing force_names.append(name) print("FORCE_SCOPE", name, report.get_state()) mass_names = [] for name, zone in [("fw_inlet_mass", "inlet"), ("fw_outlet_mass", "outlet")]: if name not in surface_reports.get_object_names(): surface_reports[name] = {} surface_reports[name].report_type = "surface-massflowrate" surface_reports[name].surface_names = [zone] mass_names.append(name) s.file.write_case_data( file_name=task_directory + r"\front_wing_boundary_repair_ready.cas.h5" ) if initial_iteration != 0: raise RuntimeError("检测到已有求解结果，已保留；不重复本轮首次求解流程。") print("RESIDUAL_SETTINGS", s.solution.monitor.residual.get_state()) s.setup.general.solver.time = "steady" s.file.start_transcript( file_name=task_directory + r"\front_wing_repaired_flow.trn" ) try: for block in range(6): s.solution.run_calculation.iterate(iter_count=50) result = reports.compute(report_defs=force_names) drag = report_value(result, force_names[0]) lift = report_value(result, force_names[1]) denominator = 0.5 * 1.225 * 30.0**2 * 0.552 print("FORCE_HISTORY", { "iteration": s.solution.run_calculation.iter_count.get_state(), "drag": drag, "lift": lift, "downforce": -lift, "CD": drag / denominator, "CL": lift / denominator, }) s.file.write_case_data( file_name=task_directory + r"\front_wing_repaired_steady.cas.h5" ) section = checked( "CENTRAL_SECTION", fluentagi.create_section(axis="z", position=0.0), ) s.setup.general.solver.time = "unsteady-2nd-order" s.solution.run_calculation.transient_controls.time_step_size = 0.0005 checked( "VELOCITY_SECTION_ANIMATION", fluentagi.record_animation( directory=task_directory, field="velocity-magnitude", frames=41, steps_per_frame=5, max_iter_per_step=40, surfaces=[section["surface"]], ), ) print("FINAL_FORCES", reports.compute(report_defs=force_names)) mass_result = reports.compute(report_defs=mass_names) print("MASS_FLOW_READBACK", mass_result) mass_values = [report_value(mass_result, name) for name in mass_names] scale = max(abs(value) for value in mass_values) print("RELATIVE_MASS_IMBALANCE", abs(sum(mass_values)) / max(scale, 1e-20)) checked( "FINAL_SECTION_EXPORT", fluentagi.export_results( directory=task_directory, fields=["pressure", "velocity-magnitude"], surfaces=[section["surface"]], ), ) paths = s.results.graphics.pathline if paths.is_active(): name = "front_wing_streamlines" if name not in paths.get_object_names(): paths[name] = {} print("STREAMLINE_STATE", paths[name].get_state()) print("STREAMLINE_FIELDS", getattr(paths[name], "child_names", [])) finally: s.file.stop_transcript() s.file.write_case_data( file_name=task_directory + r"\front_wing_repaired_flow_checkpoint.cas.h5" ) print("REFERENCE_AREA", 0.552) print("VALIDATION_PENDING", [ "all_step_convergence", "force_stability", "mesh_independence", "near_wall_resolution", "streamline_export", ])