|
| 1 | +# Copyright 2025 Pasteur Labs. All Rights Reserved. |
| 2 | +# SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +from typing import Any |
| 5 | + |
| 6 | +import equinox as eqx |
| 7 | +import jax |
| 8 | +import jax.numpy as jnp |
| 9 | +import jax_cfd.base as cfd |
| 10 | +from pydantic import BaseModel, Field |
| 11 | + |
| 12 | +from tesseract_core.runtime import Array, Differentiable, Float32 |
| 13 | +from tesseract_core.runtime.tree_transforms import filter_func, flatten_with_paths |
| 14 | + |
| 15 | + |
| 16 | +class InputSchema(BaseModel): |
| 17 | + v0: Differentiable[ |
| 18 | + Array[ |
| 19 | + ( |
| 20 | + None, |
| 21 | + None, |
| 22 | + None, |
| 23 | + ), |
| 24 | + Float32, |
| 25 | + ] |
| 26 | + ] = Field(description="3D Array defining the initial velocity field") |
| 27 | + density: float = Field(description="Density of the fluid") |
| 28 | + viscosity: float = Field(description="Viscosity of the fluid") |
| 29 | + inner_steps: int = Field( |
| 30 | + description="Number of solver steps for each timestep", default=25 |
| 31 | + ) |
| 32 | + outer_steps: int = Field(description="Number of timesteps steps", default=10) |
| 33 | + max_velocity: float = Field(description="Maximum velocity", default=2.0) |
| 34 | + cfl_safety_factor: float = Field(description="CFL safety factor", default=0.5) |
| 35 | + domain_size_x: float = Field(description="Domain size x", default=1.0) |
| 36 | + domain_size_y: float = Field(description="Domain size y", default=1.0) |
| 37 | + |
| 38 | + |
| 39 | +class OutputSchema(BaseModel): |
| 40 | + result: Differentiable[Array[(None, None, None), Float32]] = Field( |
| 41 | + description="3D Array defining the final velocity field" |
| 42 | + ) |
| 43 | + |
| 44 | + |
| 45 | +def cfd_fwd( |
| 46 | + v0: jnp.ndarray, |
| 47 | + density: float, |
| 48 | + viscosity: float, |
| 49 | + inner_steps: int, |
| 50 | + outer_steps: int, |
| 51 | + max_velocity: float, |
| 52 | + cfl_safety_factor: float, |
| 53 | + domain_size_x: float, |
| 54 | + domain_size_y: float, |
| 55 | +) -> tuple[jax.Array, jax.Array]: |
| 56 | + """Compute the final velocity field using the semi-implicit Navier-Stokes equations. |
| 57 | +
|
| 58 | + Args: |
| 59 | + v0: Initial velocity field. |
| 60 | + density: Density of the fluid. |
| 61 | + viscosity: Viscosity of the fluid. |
| 62 | + inner_steps: Number of solver steps for each timestep. |
| 63 | + outer_steps: Number of timesteps steps. |
| 64 | + max_velocity: Maximum velocity. |
| 65 | + cfl_safety_factor: CFL safety factor. |
| 66 | + domain_size_x: Domain size in x direction. |
| 67 | + domain_size_y: Domain size in y direction. |
| 68 | +
|
| 69 | + Returns: |
| 70 | + Final velocity field. |
| 71 | + """ |
| 72 | + vx0 = v0[..., 0] |
| 73 | + vy0 = v0[..., 1] |
| 74 | + bc = cfd.boundaries.HomogeneousBoundaryConditions( |
| 75 | + ( |
| 76 | + (cfd.boundaries.BCType.PERIODIC, cfd.boundaries.BCType.PERIODIC), |
| 77 | + (cfd.boundaries.BCType.PERIODIC, cfd.boundaries.BCType.PERIODIC), |
| 78 | + ) |
| 79 | + ) |
| 80 | + |
| 81 | + # reconstruct grid from input |
| 82 | + grid = cfd.grids.Grid( |
| 83 | + vx0.shape, domain=((0.0, domain_size_x), (0.0, domain_size_y)) |
| 84 | + ) |
| 85 | + |
| 86 | + vx0 = cfd.grids.GridArray(vx0, grid=grid, offset=(1.0, 0.5)) |
| 87 | + vy0 = cfd.grids.GridArray(vy0, grid=grid, offset=(0.5, 1.0)) |
| 88 | + |
| 89 | + # reconstruct GridVariable from input |
| 90 | + vx0 = cfd.grids.GridVariable(vx0, bc) |
| 91 | + vy0 = cfd.grids.GridVariable(vy0, bc) |
| 92 | + v0 = (vx0, vy0) |
| 93 | + |
| 94 | + # Choose a time step. |
| 95 | + dt = cfd.equations.stable_time_step( |
| 96 | + max_velocity, cfl_safety_factor, viscosity, grid |
| 97 | + ) |
| 98 | + |
| 99 | + # Define a step function and use it to compute a trajectory. |
| 100 | + step_fn = cfd.funcutils.repeated( |
| 101 | + cfd.equations.semi_implicit_navier_stokes( |
| 102 | + density=density, viscosity=viscosity, dt=dt, grid=grid |
| 103 | + ), |
| 104 | + steps=inner_steps, |
| 105 | + ) |
| 106 | + rollout_fn = cfd.funcutils.trajectory(step_fn, outer_steps) |
| 107 | + _, trajectory = jax.device_get(rollout_fn(v0)) |
| 108 | + vxn = trajectory[0].array.data[-1] |
| 109 | + vyn = trajectory[1].array.data[-1] |
| 110 | + return jnp.stack([vxn, vyn], axis=-1) |
| 111 | + |
| 112 | + |
| 113 | +@eqx.filter_jit |
| 114 | +def apply_jit(inputs: dict) -> dict: |
| 115 | + vn = cfd_fwd(**inputs) |
| 116 | + return dict(result=vn) |
| 117 | + |
| 118 | + |
| 119 | +def apply(inputs: InputSchema) -> OutputSchema: |
| 120 | + return apply_jit(inputs.model_dump()) |
| 121 | + |
| 122 | + |
| 123 | +def jacobian( |
| 124 | + inputs: InputSchema, |
| 125 | + jac_inputs: set[str], |
| 126 | + jac_outputs: set[str], |
| 127 | +): |
| 128 | + return jac_jit(inputs.model_dump(), tuple(jac_inputs), tuple(jac_outputs)) |
| 129 | + |
| 130 | + |
| 131 | +def jacobian_vector_product( |
| 132 | + inputs: InputSchema, |
| 133 | + jvp_inputs: set[str], |
| 134 | + jvp_outputs: set[str], |
| 135 | + tangent_vector: dict[str, Any], |
| 136 | +): |
| 137 | + return jvp_jit( |
| 138 | + inputs.model_dump(), |
| 139 | + tuple(jvp_inputs), |
| 140 | + tuple(jvp_outputs), |
| 141 | + tangent_vector, |
| 142 | + ) |
| 143 | + |
| 144 | + |
| 145 | +def vector_jacobian_product( |
| 146 | + inputs: InputSchema, |
| 147 | + vjp_inputs: set[str], |
| 148 | + vjp_outputs: set[str], |
| 149 | + cotangent_vector: dict[str, Any], |
| 150 | +): |
| 151 | + return vjp_jit( |
| 152 | + inputs.model_dump(), |
| 153 | + tuple(vjp_inputs), |
| 154 | + tuple(vjp_outputs), |
| 155 | + cotangent_vector, |
| 156 | + ) |
| 157 | + |
| 158 | + |
| 159 | +def abstract_eval(abstract_inputs): |
| 160 | + """Calculate output shape of apply from the shape of its inputs.""" |
| 161 | + is_shapedtype_dict = lambda x: type(x) is dict and (x.keys() == {"shape", "dtype"}) |
| 162 | + is_shapedtype_struct = lambda x: isinstance(x, jax.ShapeDtypeStruct) |
| 163 | + |
| 164 | + jaxified_inputs = jax.tree.map( |
| 165 | + lambda x: jax.ShapeDtypeStruct(**x) if is_shapedtype_dict(x) else x, |
| 166 | + abstract_inputs.model_dump(), |
| 167 | + is_leaf=is_shapedtype_dict, |
| 168 | + ) |
| 169 | + dynamic_inputs, static_inputs = eqx.partition( |
| 170 | + jaxified_inputs, filter_spec=is_shapedtype_struct |
| 171 | + ) |
| 172 | + |
| 173 | + def wrapped_apply(dynamic_inputs): |
| 174 | + inputs = eqx.combine(static_inputs, dynamic_inputs) |
| 175 | + return apply_jit(inputs) |
| 176 | + |
| 177 | + jax_shapes = jax.eval_shape(wrapped_apply, dynamic_inputs) |
| 178 | + return jax.tree.map( |
| 179 | + lambda x: ( |
| 180 | + {"shape": x.shape, "dtype": str(x.dtype)} if is_shapedtype_struct(x) else x |
| 181 | + ), |
| 182 | + jax_shapes, |
| 183 | + is_leaf=is_shapedtype_struct, |
| 184 | + ) |
| 185 | + |
| 186 | + |
| 187 | +@eqx.filter_jit |
| 188 | +def jac_jit( |
| 189 | + inputs: dict, |
| 190 | + jac_inputs: tuple[str], |
| 191 | + jac_outputs: tuple[str], |
| 192 | +): |
| 193 | + filtered_apply = filter_func(apply_jit, inputs, jac_outputs) |
| 194 | + return jax.jacrev(filtered_apply)( |
| 195 | + flatten_with_paths(inputs, include_paths=jac_inputs) |
| 196 | + ) |
| 197 | + |
| 198 | + |
| 199 | +@eqx.filter_jit |
| 200 | +def jvp_jit( |
| 201 | + inputs: dict, jvp_inputs: tuple[str], jvp_outputs: tuple[str], tangent_vector: dict |
| 202 | +): |
| 203 | + filtered_apply = filter_func(apply_jit, inputs, jvp_outputs) |
| 204 | + return jax.jvp( |
| 205 | + filtered_apply, |
| 206 | + [flatten_with_paths(inputs, include_paths=jvp_inputs)], |
| 207 | + [tangent_vector], |
| 208 | + )[1] |
| 209 | + |
| 210 | + |
| 211 | +@eqx.filter_jit |
| 212 | +def vjp_jit( |
| 213 | + inputs: dict, |
| 214 | + vjp_inputs: tuple[str], |
| 215 | + vjp_outputs: tuple[str], |
| 216 | + cotangent_vector: dict, |
| 217 | +): |
| 218 | + filtered_apply = filter_func(apply_jit, inputs, vjp_outputs) |
| 219 | + _, vjp_func = jax.vjp( |
| 220 | + filtered_apply, flatten_with_paths(inputs, include_paths=vjp_inputs) |
| 221 | + ) |
| 222 | + return vjp_func(cotangent_vector)[0] |
0 commit comments