/Video)” ) parser.add_argument( “–skip-video-encode”, action=“store_true”, help=“Skip ffmpeg video encoding after frame generation” ) parser.add_argument( “–framerate”, type=int, default=90, help=“Input framerate for ffmpeg (default: 90)” ) parser.add_argument( “–output-fps”, type=int, default=30, help=“Output video framerate (default: 30)” ) # Colorbar bounds parser.add_argument( “–d2-vmin”, type=float, default=-2.0, help=“Min value for strain-rate colorbar (default: -2.0)” ) parser.add_argument( “–d2-vmax”, type=float, default=2.0, help=“Max value for strain-rate colorbar (default: 2.0)” ) parser.add_argument( “–vel-vmin”, type=float, default=0.0, help=“Min value for velocity colorbar (default: 0.0)” ) parser.add_argument( “–vel-vmax”, type=float, default=1.0, help=“Max value for velocity colorbar (default: 1.0)” ) args = parser.parse_args()

Default output directory

output_dir = args.folderToSave if args.folderToSave else os.path.join(args.caseToProcess, “Video”)

return RuntimeConfig( cpus=args.CPUs, n_snapshots=args.nGFS, grids_per_r=args.GridsPerR, tsnap=args.tsnap, zmin=args.ZMIN, zmax=args.ZMAX, rmax=args.RMAX, case_dir=args.caseToProcess, output_dir=output_dir, skip_video_encode=args.skip_video_encode, framerate=args.framerate, output_fps=args.output_fps, d2_vmin=args.d2_vmin, d2_vmax=args.d2_vmax, vel_vmin=args.vel_vmin, vel_vmax=args.vel_vmax, )

def ensure_directory(path: str) -> None:

    if not os.path.isdir(path):
        os.makedirs(path, exist_ok=True)


def run_helper(command: Sequence[str], cwd: Optional[str] = None) -> Sequence[str]:

Run a helper executable and return its stderr as decoded lines.

The compiled helpers deliberately emit their payload to stderr, so stdout is ignored and we return the informative stderr content.

Note: Basilisk executables have issues with very long absolute paths, so we support running from a specific working directory with relative paths.

    process = sp.Popen(command, stdout=sp.PIPE, stderr=sp.PIPE, cwd=cwd)
    _, stderr = process.communicate()
    if process.returncode != 0:
        raise RuntimeError(
            f"Command {' '.join(command)} failed with code {process.returncode}:\n"
            f"{stderr.decode('utf-8')}"
        )
    return stderr.decode("utf-8").split("\n")


def get_facets(filename: str, case_dir: str):

Shells out to the compiled getFacet executable, which extracts the volume-of-fluid (VOF) interface as a sequence of line segments. Since the simulation uses axisymmetric coordinates, only the r >= 0 half is computed. This function mirrors each segment about r=0.

Args: filename: Relative path to snapshot file (e.g., ‘intermediate/snapshot-0.0100’) case_dir: Absolute path to case directory (used as cwd)

Returns: list[tuple]: Sequence of line segments, each as ((r1, z1), (r2, z2)).

    temp2 = run_helper([HELPER_GETFACET, filename], cwd=case_dir)
    segs = []
    skip = False
    if len(temp2) > 1e2:
        for n1 in range(len(temp2)):
            temp3 = temp2[n1].split(" ")
            if temp3 == [""]:
                skip = False
                continue
            if not skip and n1 + 1 < len(temp2):
                temp4 = temp2[n1 + 1].split(" ")
                r1, z1 = np.array([float(temp3[1]), float(temp3[0])])
                r2, z2 = np.array([float(temp4[1]), float(temp4[0])])
                segs.append(((r1, z1), (r2, z2)))
                segs.append(((-r1, z1), (-r2, z2)))
                skip = True
    return segs


def get_field(filename: str, case_dir: str, zmin: float, zmax: float, rmax: float, nr: int) -> FieldData:

Shells out to the compiled getData executable, which samples the strain-rate and velocity fields on a structured grid.

Args: filename: Relative path to snapshot file (e.g., ‘intermediate/snapshot-0.0100’) case_dir: Absolute path to case directory (used as cwd) zmin: Minimum axial coordinate for sampling domain zmax: Maximum axial coordinate for sampling domain rmax: Maximum radial coordinate (positive branch only) nr: Number of grid points in radial direction

Returns: FieldData: Structured container with reshaped 2D arrays.

    temp2 = run_helper(
        [
            HELPER_GETDATA,
            filename,
            str(zmin),
            str(0),
            str(zmax),
            str(rmax),
            str(nr),
        ],
        cwd=case_dir,
    )
    Rtemp, Ztemp, D2temp, veltemp = [], [], [], []

    for n1 in range(len(temp2)):
        temp3 = temp2[n1].split(" ")
        if temp3 == [""]:
            continue
        Ztemp.append(float(temp3[0]))
        Rtemp.append(float(temp3[1]))
        D2temp.append(float(temp3[2]))
        veltemp.append(float(temp3[3]))

    R = np.asarray(Rtemp)
    Z = np.asarray(Ztemp)
    D2 = np.asarray(D2temp)
    vel = np.asarray(veltemp)
    nz = int(len(Z) / nr)

    log_status(f"{os.path.basename(filename)}: nz = {nz}")

    R.resize((nz, nr))
    Z.resize((nz, nr))
    D2.resize((nz, nr))
    vel.resize((nz, nr))

    return FieldData(R=R, Z=Z, strain_rate=D2, velocity=vel, nz=nz)


def build_snapshot_info(index: int, config: RuntimeConfig) -> SnapshotInfo:

time = config.tsnap * index source = os.path.join(config.case_dir, “intermediate”, f”snapshot-{time:.4f}“) target = os.path.join(config.output_dir, f”{int(time * 1000):08d}.png”) return SnapshotInfo(index=index, time=time, source=source, target=target)

def draw_domain_outline(ax, bounds: DomainBounds, style: PlotStyle) -> None:

    ax.plot(
        [0, 0],
        [bounds.zmin, bounds.zmax],
        "-.",
        color=style.zero_axis_color,
        linewidth=style.line_width,
    )
    ax.plot(
        [bounds.rmin, bounds.rmin],
        [bounds.zmin, bounds.zmax],
        "-",
        color=style.axis_color,
        linewidth=style.line_width,
    )
    ax.plot(
        [bounds.rmin, bounds.rmax],
        [bounds.zmin, bounds.zmin],
        "-",
        color=style.axis_color,
        linewidth=style.line_width,
    )
    ax.plot(
        [bounds.rmin, bounds.rmax],
        [bounds.zmax, bounds.zmax],
        "-",
        color=style.axis_color,
        linewidth=style.line_width,
    )
    ax.plot(
        [bounds.rmax, bounds.rmax],
        [bounds.zmin, bounds.zmax],
        "-",
        color=style.axis_color,
        linewidth=style.line_width,
    )


def add_colorbar(fig, ax, mappable, *, align: str, label: str, style: PlotStyle):

l, b, w, h = ax.get_position().bounds if align == “left”: position = [l - style.left_colorbar_offset, b, style.colorbar_width, h] else: position = [l + w + style.right_colorbar_offset, b, style.colorbar_width, h] cb_ax = fig.add_axes(position) colorbar = plt.colorbar(mappable, cax=cb_ax, orientation=“vertical”) colorbar.set_label(label, fontsize=style.tick_label_size, labelpad=5) colorbar.ax.tick_params(labelsize=style.tick_label_size) colorbar.ax.yaxis.set_major_formatter(StrMethodFormatter(“{x:,.2f}”)) if align == “left”: colorbar.ax.yaxis.set_ticks_position(“left”) colorbar.ax.yaxis.set_label_position(“left”) return colorbar

def plot_snapshot( field_data: FieldData, facets, bounds: DomainBounds, snapshot: SnapshotInfo, config: RuntimeConfig, style: PlotStyle, ) -> None:

    Render and persist a single snapshot figure.

    Visualization:
    - Left side: log10(D:D) strain-rate field
    - Right side: velocity magnitude

fig, ax = plt.subplots() fig.set_size_inches(*style.figure_size)

draw_domain_outline(ax, bounds, style) line_segments = LineCollection( facets, linewidths=4, colors=style.interface_color, linestyle=“solid” ) ax.add_collection(line_segments)

rminp, rmaxp = field_data.radial_extent zminp, zmaxp = field_data.axial_extent

Left: Strain-rate field (D:D)

cntrl1 = ax.imshow( field_data.strain_rate, cmap=“hot_r”, interpolation=“Bilinear”, origin=“lower”, extent=[-rminp, -rmaxp, zminp, zmaxp], vmax=config.d2_vmax, vmin=config.d2_vmin, )

Right: Velocity magnitude

cntrl2 = ax.imshow( field_data.velocity, interpolation=“Bilinear”, cmap=“Purples”, origin=“lower”, extent=[rminp, rmaxp, zminp, zmaxp], vmax=config.vel_vmax, vmin=config.vel_vmin, )

ax.set_aspect(“equal”) ax.set_xlim(bounds.rmin, bounds.rmax) ax.set_ylim(bounds.zmin, bounds.zmax) ax.set_title(f”\(t/\\tau_0\) = {snapshot.time:4.3f}“, fontsize=style.tick_label_size) ax.axis(”off”)

add_colorbar( fig, ax, cntrl1, align=“left”, label=r”\(\log_{10}\left(\boldsymbol{\mathcal{D}:\mathcal{D}}\right)\)“, style=style, ) add_colorbar( fig, ax, cntrl2, align=”right”, label=r”\(\|\boldsymbol{u}\|\)“, style=style, )

plt.savefig(snapshot.target, bbox_inches=“tight”) plt.close(fig)

def process_timestep(index: int, config: RuntimeConfig, style: PlotStyle) -> None:

    Worker executed for every timestep index.

    Performs availability checks, loads helper outputs, and calls plot_snapshot.

snapshot = build_snapshot_info(index, config) if not os.path.exists(snapshot.source): log_status(f”Missing: {os.path.basename(snapshot.source)}“, level=”WARN”) return if os.path.exists(snapshot.target): log_status(f”Exists, skipping: {os.path.basename(snapshot.target)}“) return

Show relative path: CaseNo/intermediate/filename

src_parts = snapshot.source.split(os.sep) src_rel = os.sep.join(src_parts[-3:]) if len(src_parts) >= 3 else snapshot.source log_status(f”Processing {src_rel}“)

Relative path for Basilisk helpers (they crash with very long absolute paths)

rel_snapshot = os.path.join(“intermediate”, f”snapshot-{snapshot.time:.4f}“) case_dir = os.path.abspath(config.case_dir)

try: facets = get_facets(rel_snapshot, case_dir) nr = int(config.grids_per_r * config.rmax) field_data = get_field( rel_snapshot, case_dir, config.zmin, config.zmax, config.rmax, nr ) plot_snapshot(field_data, facets, config.bounds, snapshot, config, style)

Show relative path: CaseNo/Video/filename

tgt_parts = snapshot.target.split(os.sep) tgt_rel = os.sep.join(tgt_parts[-3:]) if len(tgt_parts) >= 3 else snapshot.target log_status(f”Saved: {tgt_rel}“)

except Exception as err: log_status( f”Error at {src_rel} (t={snapshot.time:.4f}): {err}“, level=”ERROR” ) raise

def encode_video(config: RuntimeConfig) -> None:

    Run ffmpeg to stitch PNG frames into an MP4 video.

    The output video is saved in the case directory with the case number
    as filename (e.g., simulationCases/1000/1000.mp4).

Extract case number from path

case_no = os.path.basename(config.case_dir)

Output path: /.mp4

output_path = os.path.join(config.case_dir, f”{case_no}.mp4”) input_pattern = os.path.join(config.output_dir, “*.png”)

cmd = [ “ffmpeg”, “-y”, “-framerate”, str(config.framerate), “-pattern_type”, “glob”, “-i”, input_pattern, “-vf”, “pad=ceil(iw/2)2:ceil(ih/2)2”, “-c:v”, “libx264”, “-r”, str(config.output_fps), “-pix_fmt”, “yuv420p”, output_path]

log_status(f”Encoding video: {output_path}“) result = sp.run(cmd, capture_output=True, text=True) if result.returncode != 0: log_status(f”ffmpeg error: {result.stderr}“, level=”ERROR”) raise RuntimeError(f”ffmpeg failed with code {result.returncode}“) log_status(f”Video saved: {output_path}“)

def main():

    config = parse_arguments()
    ensure_directory(config.output_dir)

    log_status(f"Processing case: {config.case_dir}")
    log_status(f"Domain: R=[{config.rmin:.2f},{config.rmax:.2f}], Z=[{config.zmin:.2f},{config.zmax:.2f}]")

    with mp.Pool(processes=config.cpus) as pool:
        worker = partial(process_timestep, config=config, style=PLOT_STYLE)
        pool.map(worker, range(config.n_snapshots))

    # Encode video unless skipped
    if not config.skip_video_encode:
        encode_video(config)


if __name__ == "__main__":
    main()