#!/usr/bin/env python3 """Run the 24-rack, 979.2 kW AI data center cooling example. Requires Python 3.10+ and Requests. Edit the constants below, set VANELLUS_API_KEY, then run this file to submit, follow progress, and download results. Uses API credits. Adapted from the Vanellus Python automation template: https://docs.vanellus.tech/guide/how-to-guides/run-simulations/automate-a-simulation-with-python/ The 10.2 kW server load and 1,105 CFM supply-flow reference are inspired by NVIDIA DGX H100 specifications (airflow at 80% PWM): https://docs.nvidia.com/dgx/dgxh100-user-guide/dgxh100-user-guide.pdf The 300 Pa / 0.65 m³/s quadratic fan curve is illustrative, not measured DGX data. Cooling cabinet dimensions follow the STULZ CyberAir 3PRO ASR 2410 CW envelope, pages 16–17, with an assumed 0.5 m underfloor discharge extension: https://www.stulz.com/fileadmin/user_upload/Final_images/STULZ_NL/Other/STULZ_CyberAir_3PRO_CW_Brochure_2403_EN.pdf Cooling uses a fixed supply temperature with no capacity limit. Floor openings have no grille resistance. This is an illustrative example, not a validated design. """ from collections.abc import Iterator import io import json import math import os from pathlib import Path import threading import time from typing import Any import zipfile # Edit these case settings before running; dimensions are in meters. MAX_ITERATIONS = 1000 NUMERICS = "stable" MESH_SETTINGS = { "max_cell_size": 0.20, "target_wall_distance": 100.0, "max_growth_rate": 1.5, "boundary_layer_growth_rate": 1.3, "fuse_threshold": 0.0001, } TARGET_RESIDUAL = 0.001 TIMEOUT_CANCEL_SECONDS = 2400 TIMEOUT_KILL_SECONDS = 2460 OUTPUT_ROOT = Path("results") ROOM_SIZE = [12.0, 16.0, 3.5] PLENUM_DEPTH = 1.0 RAISED_FLOOR_THICKNESS = 0.10 RACKS_PER_ROW = 6 RACK_SIZE = [0.7, 1.2, 52 * 0.04445] ROW_X = (ROOM_SIZE[0] - RACKS_PER_ROW * RACK_SIZE[0]) / 2.0 POD_Y = [3.1, 9.0] SERVER_POWER_W = 10_200.0 SERVER_PORT_SIZE = [0.44, 0.30] SERVER_CENTER_HEIGHTS = [0.40, 0.87, 1.44, 1.91] SERVER_PORT_MESH_RESOLUTION = 4 REFERENCE_FLOW_M3_S = 1105.0 * (0.028316846592 / 60.0) FAN_SHUTOFF_PRESSURE_PA = 300.0 FAN_ZERO_PRESSURE_FLOW_M3_S = 0.65 TILE_SIZE = [0.60, 0.62] CABINET_SIZE = [0.98, 3.35, 2.495] UNIT_BOTTOM = -0.5 PORT_SIZE = [0.8, 3.1] PLANT_Y = [2.05, 6.0, 9.95, 13.9] INSTALLED_UNITS = {"crah_w1", "crah_w2", "crah_w3", "crah_w4", "crah_e1", "crah_e3", "crah_e4"} STANDBY = "crah_w2" CRAH_PORT_MESH_RESOLUTION = 4 SUPPLY_TEMPERATURE_C = 20.0 # The shared documentation runtime uses these types and terminal states. JsonObject = dict[str, Any] DOWNLOADABLE_STATES = { "canceled", "dry_run", "max_iterations_reached", "monitor_converged", "residual_converged", } def server_curve() -> JsonObject: """Sample the illustrative whole-server fan curve, retaining the reference flow.""" # The endpoint and pressure capability are assumptions documented above. flow_rates = sorted({FAN_ZERO_PRESSURE_FLOW_M3_S * index / 80.0 for index in range(81)} | {REFERENCE_FLOW_M3_S}) pressures = [ max(0.0, FAN_SHUTOFF_PRESSURE_PA * (1.0 - (flow / FAN_ZERO_PRESSURE_FLOW_M3_S) ** 2)) for flow in flow_rates ] pressures[-1] = 0.0 return {"flow_rates": flow_rates, "static_pressures": pressures} def rectangle( side: str, center: list[float], size: list[float], condition: dict[str, Any], *, mesh_resolution: int, ) -> dict[str, Any]: """Return a rectangular boundary-condition payload.""" return { "boundary": { "rectangle": { "side": side, "center": center, "size": size, "mesh_resolution": mesh_resolution, } }, "condition": condition, } def insulating_solid(tag: str, origin: list[float], size: list[float]) -> dict[str, Any]: """Return an insulating cuboid used for a containment panel.""" return { "tag": tag, "bbox": {"origin": origin, "size": size}, "thermal_model": "insulating", "boundary_conditions": [], } def _rack_component( *, pod_index: int, row_index: int, column_index: int, curve: dict[str, Any], ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """Build one blanked rack, its four linked fans, and nearby probes.""" x = ROW_X + column_index * RACK_SIZE[0] y = POD_Y[pod_index] + row_index * (RACK_SIZE[1] + 1.5) rack_tag = f"rack_p{pod_index + 1}_{row_index + 1}_{column_index + 1:02d}" front_side = "y_max" if row_index == 0 else "y_min" exhaust_side = "y_min" if row_index == 0 else "y_max" front_y = y + RACK_SIZE[1] if row_index == 0 else y exhaust_y = y if row_index == 0 else y + RACK_SIZE[1] probe_direction = 1.0 if row_index == 0 else -1.0 center_x = x + RACK_SIZE[0] / 2.0 boundaries: list[dict[str, Any]] = [] monitors: list[dict[str, Any]] = [] for server_index, center_z in enumerate(SERVER_CENTER_HEIGHTS, start=1): server_tag = f"{rack_tag}_dgx_{server_index:02d}" boundaries.extend( [ rectangle( front_side, [center_x, center_z], list(SERVER_PORT_SIZE), {"linked_fan_intake": {"tag": server_tag}}, mesh_resolution=SERVER_PORT_MESH_RESOLUTION, ), rectangle( exhaust_side, [center_x, center_z], list(SERVER_PORT_SIZE), { "linked_fan_exhaust": { "tag": server_tag, "flow_rates": curve["flow_rates"], "static_pressures": curve["static_pressures"], "thermal_power": SERVER_POWER_W, "relaxation_factor": 0.01, "turbulent_intensity": 0.1, "viscosity_ratio": 100.0, } }, mesh_resolution=SERVER_PORT_MESH_RESOLUTION, ), ] ) # Probe just outside each port to compare server inlet and outlet states. for role, probe_y in ( ("intake", front_y + probe_direction * 0.08), ("exhaust", exhaust_y - probe_direction * 0.08), ): for field in ("temperature", "pressure"): monitors.append( { "probe": { "name": f"{server_tag}_{role}_{field}", "field": field, "position": [center_x, probe_y, center_z], } } ) component = { "tag": rack_tag, "bbox": {"origin": [x, y, 0.0], "size": list(RACK_SIZE)}, "thermal_model": "insulating", "boundary_conditions": boundaries, } return component, monitors def raised_floor_components( equipment_openings: list[dict[str, list[float]]], *, tile_openings: list[dict[str, list[float]]], ) -> list[dict[str, Any]]: """Partition the raised floor around the 24 vents and seven CRAH footprints.""" holes = [ ( opening["center"][0] - opening["size"][0] / 2.0, opening["center"][1] - opening["size"][1] / 2.0, opening["center"][0] + opening["size"][0] / 2.0, opening["center"][1] + opening["size"][1] / 2.0, ) for opening in [*tile_openings, *equipment_openings] ] y_coordinates = sorted({0.0, ROOM_SIZE[1], *(coordinate for hole in holes for coordinate in hole[1::2])}) panels: list[list[float]] = [] for y_min, y_max in zip(y_coordinates, y_coordinates[1:]): y_center = (y_min + y_max) / 2.0 blocked_intervals = sorted( (x_min, x_max) for x_min, hole_y_min, x_max, hole_y_max in holes if hole_y_min < y_center < hole_y_max ) x_min = 0.0 spans: list[tuple[float, float]] = [] for blocked_min, blocked_max in blocked_intervals: if blocked_min > x_min: spans.append((x_min, blocked_min)) x_min = max(x_min, blocked_max) if x_min < ROOM_SIZE[0]: spans.append((x_min, ROOM_SIZE[0])) for span_min, span_max in spans: # Merge vertically adjacent panels with identical horizontal spans. previous = next( ( panel for panel in panels if math.isclose(panel[0], span_min) and math.isclose(panel[2], span_max) and math.isclose(panel[3], y_min) ), None, ) if previous is None: panels.append([span_min, y_min, span_max, y_max]) else: previous[3] = y_max return [ insulating_solid( f"raised_floor_{index:03d}", [x_min, y_min, -RAISED_FLOOR_THICKNESS], [x_max - x_min, y_max - y_min, RAISED_FLOOR_THICKNESS], ) for index, (x_min, y_min, x_max, y_max) in enumerate(panels, start=1) ] def build_request() -> JsonObject: """Build the published case using the editable constants above.""" # The illustrative curve is shared by all 96 heated server pairs. curves = server_curve() columns = RACKS_PER_ROW server_count = len(POD_Y) * 2 * columns * len(SERVER_CENTER_HEIGHTS) duty_count = len(INSTALLED_UNITS - {STANDBY}) components, openings = [], [] monitors = [ {"mass_balance": {"relative": True}}, {"energy_balance": {"relative": True}}, {"minimum": {"name": "air_temperature_minimum", "field": "temperature"}}, {"maximum": {"name": "air_temperature_maximum", "field": "temperature"}}, {"mean": {"name": "air_temperature_mean", "field": "temperature"}}, ] for pod, y in enumerate(POD_Y): for row in range(2): for column in range(RACKS_PER_ROW): rack, probes = _rack_component(pod_index=pod, row_index=row, column_index=column, curve=curves) components.append(rack) monitors.extend(probes) openings.append( { "center": [ROW_X + (column + 0.5) * RACK_SIZE[0], y + (1.53 if row == 0 else 2.37)], "size": list(TILE_SIZE), } ) length = columns * RACK_SIZE[0] height = RACK_SIZE[2] aisle_y = y + RACK_SIZE[1] components.extend( [ insulating_solid(f"cold_aisle_p{pod + 1}_roof", [ROW_X, aisle_y, height], [length, 1.5, 0.05]), insulating_solid( f"cold_aisle_p{pod + 1}_west_door", [ROW_X - 0.05, aisle_y, 0], [0.05, 1.5, height + 0.05] ), insulating_solid( f"cold_aisle_p{pod + 1}_east_door", [ROW_X + length, aisle_y, 0], [0.05, 1.5, height + 0.05] ), ] ) # Allocate NVIDIA's reference server airflow equally across the duty units. flow_per_unit = server_count * REFERENCE_FLOW_M3_S / duty_count supply_speed = flow_per_unit / (PORT_SIZE[0] * PORT_SIZE[1]) penetrations = [] for bank, x in (("w", 0.25), ("e", ROOM_SIZE[0] - 0.25 - CABINET_SIZE[0])): for index, y in enumerate(PLANT_Y, 1): tag = f"crah_{bank}{index}" bbox = {"origin": [x, y - CABINET_SIZE[1] / 2, 0], "size": list(CABINET_SIZE)} if tag not in INSTALLED_UNITS: continue unit = insulating_solid( tag, [x, bbox["origin"][1], UNIT_BOTTOM], [*CABINET_SIZE[:2], CABINET_SIZE[2] - UNIT_BOTTOM] ) center = [x + CABINET_SIZE[0] / 2, y] if tag != STANDBY: unit["boundary_conditions"] = [ rectangle( "z_max", center, list(PORT_SIZE), {"flow_rate_outlet": {"target_flow_rate": flow_per_unit, "relaxation_factor": 0.1}}, mesh_resolution=CRAH_PORT_MESH_RESOLUTION, ), rectangle( "z_min", center, list(PORT_SIZE), { "velocity_inlet": { "speed": supply_speed, "temperature": SUPPLY_TEMPERATURE_C, "turbulent_intensity": 0.1, "viscosity_ratio": 100.0, } }, mesh_resolution=CRAH_PORT_MESH_RESOLUTION, ), ] for role, height in (("intake", CABINET_SIZE[2] + 0.08), ("discharge", UNIT_BOTTOM - 0.08)): for field in ("temperature", "pressure"): monitors.append( { "probe": { "name": f"{tag}_{role}_{field}", "field": field, "position": [*center, height], } } ) components.append(unit) penetrations.append({"center": center, "size": CABINET_SIZE[:2]}) components.extend(raised_floor_components(penetrations, tile_openings=openings)) request = { "fluid_properties": {"air": {}}, "turbulence_model": "k_omega_sst", "boussinesq": True, "gravity": [0.0, 0.0, -9.81], "mesh_settings": dict(MESH_SETTINGS), "numerics": NUMERICS, "max_iterations": MAX_ITERATIONS, "target_residual": TARGET_RESIDUAL, "timeout_cancel": TIMEOUT_CANCEL_SECONDS, "timeout_kill": TIMEOUT_KILL_SECONDS, "dry_run": False, # Derive the domain from the room dimensions so the floor and walls agree. "domain": { "origin": [0, 0, -PLENUM_DEPTH], "size": [*ROOM_SIZE[:2], ROOM_SIZE[2] + PLENUM_DEPTH], "boundary_conditions": [], }, "cuboid_components": components, "monitors": monitors, } return request def requests_client() -> Any: """Load Requests only when the downloaded client starts network activity.""" # Documentation builds import build_request(), which does not need the HTTP dependency. try: import requests except ModuleNotFoundError as error: raise RuntimeError("Install the client requirements before running a simulation") from error return requests def api_url(path: str) -> str: """Resolve a path against the public Vanellus API.""" return f"https://api.vanellus.tech/{path.lstrip('/')}" def request_headers(api_key: str, accept: str) -> dict[str, str]: """Return authentication and response-format headers for one request.""" # Keep the API key in a header rather than placing it in a URL or request body. return {"X-API-Key": api_key, "Accept": accept} def require_success(response: Any, action: str) -> None: """Raise a concise error containing the API response body when a request fails.""" # The response body normally contains more useful validation detail than a traceback. if response.status_code != 200: raise RuntimeError(f"{action} failed with HTTP {response.status_code}: {response.text}") # ANCHOR: submit_request def submit_request(request: JsonObject, api_key: str) -> int: """Submit one simulation request and return its server-generated ID.""" requests = requests_client() response = requests.post( api_url("/simulations"), json=request, headers=request_headers(api_key, "application/json"), timeout=120, ) require_success(response, "Simulation submission") submission = response.json() simulation_id = int(submission["id"]) print(f"Submitted simulation {simulation_id}") # Accepted requests can still contain setup warnings. for warning in submission.get("warnings", []): print(f"API warning: {warning}") return simulation_id # ANCHOR_END: submit_request # ANCHOR: progress_updates def iter_sse_data(response: Any) -> Iterator[str]: """Yield complete data payloads from a server-sent-events response.""" data_lines: list[str] = [] # One SSE event ends at a blank line and may contain more than one data line. for raw_line in response.iter_lines(decode_unicode=True): if raw_line is None: continue if isinstance(raw_line, bytes): raw_line = raw_line.decode(response.encoding or "utf-8", errors="replace") line = raw_line.rstrip("\r\n") if not line: if data_lines: yield "\n".join(data_lines) data_lines.clear() elif line.startswith("data:"): data_lines.append(line.removeprefix("data:").removeprefix(" ")) if data_lines: yield "\n".join(data_lines) def format_progress_value(value: Any) -> str: """Format one live diagnostic, preserving the API's null marker for non-finite output.""" return "NaN" if value is None else f"{float(value):.3e}" def print_progress_updates( api_key: str, simulation_id: int, stop_event: threading.Event, ) -> None: """Print live residual updates until the run or the client thread stops.""" requests = requests_client() try: with requests.get( api_url(f"/simulations/{simulation_id}/progress_stream"), headers=request_headers(api_key, "text/event-stream"), stream=True, timeout=(10, None), ) as response: require_success(response, "Progress stream") for payload in iter_sse_data(response): if stop_event.is_set(): return update = json.loads(payload) residuals = update.get("residuals", {}) residual_text = ", ".join( f"{name}={format_progress_value(value)}" for name, value in sorted(residuals.items()) ) print(f"Iteration {update['iteration']}: {residual_text}", flush=True) except (requests.RequestException, RuntimeError, json.JSONDecodeError) as error: # Status polling remains authoritative if the optional stream disconnects. if not stop_event.is_set(): print(f"Progress stream stopped: {error}") # ANCHOR_END: progress_updates def get_status(api_key: str, simulation_id: int) -> JsonObject: """Fetch the current lifecycle and solver status for one simulation.""" requests = requests_client() response = requests.get( api_url(f"/simulations/{simulation_id}/status"), headers=request_headers(api_key, "application/json"), timeout=120, ) require_success(response, "Status request") return response.json() def cancel_simulation(api_key: str, simulation_id: int) -> None: """Request a recoverable stop at the next solver iteration boundary.""" requests = requests_client() # Cancellation preserves partial results, unlike an immediate kill request. response = requests.post( api_url(f"/simulations/{simulation_id}/cancel"), headers=request_headers(api_key, "application/json"), timeout=120, ) require_success(response, "Cancellation") print(f"Cancellation requested for simulation {simulation_id}") # ANCHOR: wait_for_completion def wait_for_completion( api_key: str, simulation_id: int, dry_run: bool, ) -> JsonObject: """Stream progress while polling the authoritative simulation status.""" stop_event = threading.Event() reported_cells = False progress_thread: threading.Thread | None = None # Dry runs have no iterative residual history, so they only need status polling. if not dry_run: progress_thread = threading.Thread( target=print_progress_updates, kwargs={ "api_key": api_key, "simulation_id": simulation_id, "stop_event": stop_event, }, daemon=True, ) progress_thread.start() try: while True: status = get_status(api_key=api_key, simulation_id=simulation_id) if status.get("num_cells") and not reported_cells: print(f"Mesh cells: {int(status['num_cells']):,}", flush=True) reported_cells = True if status.get("status") in {"diverged", "error", "killed"} or status.get("error"): raise RuntimeError(f"Simulation stopped without results: {status}") if status.get("completed"): return status time.sleep(2) except KeyboardInterrupt: # Ctrl-C asks the server to preserve a partial result at an iteration boundary. cancel_simulation(api_key=api_key, simulation_id=simulation_id) raise SystemExit(130) from None finally: stop_event.set() if progress_thread is not None: progress_thread.join(timeout=5) # ANCHOR_END: wait_for_completion def extract_zip_safely(archive_bytes: bytes, output_directory: Path) -> None: """Extract a result archive after rejecting paths outside the destination.""" output_directory.mkdir(parents=True, exist_ok=True) output_root = output_directory.resolve() # Validate every member before extracting any file from the downloaded archive. with zipfile.ZipFile(io.BytesIO(archive_bytes)) as archive: for member in archive.infolist(): destination = (output_directory / member.filename).resolve() if destination != output_root and not destination.is_relative_to(output_root): raise RuntimeError(f"Unsafe path in result archive: {member.filename!r}") archive.extractall(output_directory) # ANCHOR: download_results def download_results( api_key: str, simulation_id: int, output_directory: Path, ) -> None: """Download and safely extract every available result artifact.""" requests = requests_client() response = requests.get( api_url(f"/simulations/{simulation_id}/download"), headers=request_headers(api_key, "application/zip"), timeout=None, ) require_success(response, "Results download") extract_zip_safely(response.content, output_directory) print(f"Extracted results to {output_directory}") # ANCHOR_END: download_results def run_simulation( request: JsonObject, api_key: str, ) -> Path: """Submit, follow, download, and return this case's result directory.""" simulation_id = submit_request(request, api_key=api_key) status = wait_for_completion(api_key=api_key, simulation_id=simulation_id, dry_run=request["dry_run"]) output_directory = OUTPUT_ROOT / f"simulation-{simulation_id}" download_results( api_key=api_key, simulation_id=simulation_id, output_directory=output_directory / "fields", ) # Match the website rebuild tool's result layout for direct offline rendering. for directory in (output_directory, output_directory / "fields"): (directory / "submitted-request.json").write_text(json.dumps(request, indent=2) + "\n", encoding="utf-8") (directory / "status.json").write_text(json.dumps(status, indent=2) + "\n", encoding="utf-8") final_state = str(status["status"]) print(f"Simulation finished with status {final_state}") if final_state == "max_iterations_reached": print("Warning: inspect convergence before using this result") elif final_state == "canceled": print("Warning: the downloaded result is from a canceled simulation") elif final_state not in DOWNLOADABLE_STATES: raise RuntimeError(f"Simulation ended as {final_state}; diagnostic results were retained") return output_directory def main() -> None: """Read credentials and run the case configured at the top of this file.""" # Keep the secret in the environment rather than the source or result files. api_key = os.environ.get("VANELLUS_API_KEY") if not api_key: raise SystemExit("Set VANELLUS_API_KEY before running this script") run_simulation(build_request(), api_key=api_key) if __name__ == "__main__": main()