wall_normal_profile Namespace Reference

Functions
	load_spectra_helpers ()
	Import the PICGRID and PETSc-Vec readers from generators/spectra.gen.
	find_payload (checkpoint_dir, window, field, moment, block)
	Locate one statistics-window payload inside a checkpoint bundle.
	homogeneous_average (numpy, field, wall_kji)
	Average a cell array over both directions that are not the wall-normal one.
	main (argv=None)
	Entry point.

Variables
	REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", ".."))
	SPECTRA_GEN = os.path.join(REPO_ROOT, "generators", "spectra.gen")
dict	AXIS_TO_KJI = {"Xi": 2, "Eta": 1, "Zeta": 0}
dict	AXIS_TO_COMPONENT = {"Xi": 0, "Eta": 1, "Zeta": 2}

Function Documentation

load_spectra_helpers()

wall_normal_profile.load_spectra_helpers

(

)

Import the PICGRID and PETSc-Vec readers from generators/spectra.gen.

Returns

The loaded module.

Definition at line 52 of file wall_normal_profile.py.

def load_spectra_helpers():
    """!
    @brief Import the PICGRID and PETSc-Vec readers from generators/spectra.gen.
    @return The loaded module.
    """
    if not os.path.isfile(SPECTRA_GEN):
        raise SystemExit(f"cannot find {SPECTRA_GEN}; run this from a PICurv checkout.")
    loader = importlib.machinery.SourceFileLoader("picurv_spectra_gen", SPECTRA_GEN)
    spec = importlib.util.spec_from_loader("picurv_spectra_gen", loader)
    module = importlib.util.module_from_spec(spec)
    loader.exec_module(module)
    return module
 
 

Here is the caller graph for this function:

find_payload()

wall_normal_profile.find_payload	(		checkpoint_dir,
			window,
			field,
			moment,
			block
	)

Locate one statistics-window payload inside a checkpoint bundle.

Parameters

[in]	checkpoint_dir	Checkpoint directory holding the window payloads.
[in]	window	Window name from monitor.yml.
[in]	field	Field name, e.g. Ucat.
[in]	moment	Moment name, e.g. first or second.
[in]	block	Block index.

Returns

Absolute path to the payload file.

Definition at line 66 of file wall_normal_profile.py.

def find_payload(checkpoint_dir, window, field, moment, block):
    """!
    @brief Locate one statistics-window payload inside a checkpoint bundle.
    @param[in] checkpoint_dir Checkpoint directory holding the window payloads.
    @param[in] window Window name from monitor.yml.
    @param[in] field Field name, e.g. Ucat.
    @param[in] moment Moment name, e.g. first or second.
    @param[in] block Block index.
    @return Absolute path to the payload file.
    """
    wanted = (window.lower(), field.lower(), moment.lower())
    candidates = []
    for root, _dirs, files in os.walk(checkpoint_dir):
        for name in files:
            if not name.endswith(".dat"):
                continue
            haystack = os.path.join(root, name).lower()
            if all(token in haystack for token in wanted) and f"block_{block:04d}" in haystack:
                candidates.append(os.path.join(root, name))
    if not candidates:
        raise SystemExit(
            f"no payload found for window={window!r} field={field!r} moment={moment!r} "
            f"block={block} under {checkpoint_dir}.\n"
            "Check that monitor.yml requested this moment and that the window reached "
            "'active' or 'complete' before the checkpoint was written."
        )
    if len(candidates) > 1:
        raise SystemExit("ambiguous payloads:\n  " + "\n  ".join(sorted(candidates)))
    return candidates[0]
 
 

Here is the caller graph for this function:

homogeneous_average()

wall_normal_profile.homogeneous_average	(		numpy,
			field,
			wall_kji
	)

Average a cell array over both directions that are not the wall-normal one.

Parameters

[in]	numpy	The numpy module.
[in]	field	Cell array shaped (nk, nj, ni) or (nk, nj, ni, 3).
[in]	wall_kji	Index of the wall-normal axis in (k, j, i) order.

Returns

Array indexed by the wall-normal coordinate, keeping any trailing component axis.

Definition at line 97 of file wall_normal_profile.py.

def homogeneous_average(numpy, field, wall_kji):
    """!
    @brief Average a cell array over both directions that are not the wall-normal one.
    @param[in] numpy The numpy module.
    @param[in] field Cell array shaped (nk, nj, ni) or (nk, nj, ni, 3).
    @param[in] wall_kji Index of the wall-normal axis in (k, j, i) order.
    @return Array indexed by the wall-normal coordinate, keeping any trailing component axis.
    """
    homogeneous = tuple(axis for axis in (0, 1, 2) if axis != wall_kji)
    return numpy.mean(field, axis=homogeneous)
 
 

Here is the caller graph for this function:

main()

wall_normal_profile.main

(

argv = None

)

Entry point.

Parameters

[in]

argv

Command-line style argument list supplied to the function.

Returns

Process exit status.

Definition at line 109 of file wall_normal_profile.py.

def main(argv=None):
    """!
    @brief Entry point.
    @param[in] argv Command-line style argument list supplied to the function.
    @return Process exit status.
    """
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--checkpoint", required=True,
                        help="Checkpoint directory containing the statistics window payloads.")
    parser.add_argument("--grid", required=True, help="Canonical PICGRID path for the run.")
    parser.add_argument("--window", default="stationary", help="Statistics window name.")
    parser.add_argument("--block", type=int, default=0, help="Block index.")
    parser.add_argument("--wall-axis", default="Eta", choices=sorted(AXIS_TO_KJI),
                        help="Wall-normal axis token.")
    parser.add_argument("--stream-axis", default="Zeta", choices=sorted(AXIS_TO_KJI),
                        help="Streamwise (driven) axis token.")
    parser.add_argument("--viscosity", type=float, required=True,
                        help="Kinematic viscosity in solver units (1/Re for length_ref=velocity_ref=1).")
    parser.add_argument("--half-height", type=float, default=1.0,
                        help="Channel half-height h in solver units.")
    parser.add_argument("--body-force", type=float,
                        help="Converged driving body force f. If given, u_tau = sqrt(f*h) is used "
                             "instead of the wall-shear estimate, which is the exact mean force balance.")
    parser.add_argument("--output", required=True, help="Output CSV path.")
    args = parser.parse_args(argv)
 
    helpers = load_spectra_helpers()
    numpy = helpers.require_numpy()
 
    blocks = helpers.read_picgrid_blocks(args.grid)
    if args.block >= len(blocks):
        raise SystemExit(f"block {args.block} is out of range; the grid holds {len(blocks)}.")
    node_dims = blocks[args.block]["dims"]
    nodes = blocks[args.block]["coords"]
 
    wall_kji = AXIS_TO_KJI[args.wall_axis]
    stream_c = AXIS_TO_COMPONENT[args.stream_axis]
    wall_c = AXIS_TO_COMPONENT[args.wall_axis]
    span_c = ({0, 1, 2} - {stream_c, wall_c}).pop()
 
    mean_path = find_payload(args.checkpoint, args.window, "Ucat", "first", args.block)
    second_path = find_payload(args.checkpoint, args.window, "Ucat", "second", args.block)
    mean = helpers.extract_interior_cells(helpers.read_petsc_vec_binary(mean_path), node_dims)
    second = helpers.extract_interior_cells(helpers.read_petsc_vec_binary(second_path), node_dims)
 
    # <u_i> and <u_i u_j> both reduce over the homogeneous directions; the
    # fluctuation intensities then follow as <u_i^2> - <u_i>^2. Averaging the
    # moments before differencing (rather than after) is what makes this a true
    # ensemble average over the homogeneous plane as well as over time.
    mean_p = homogeneous_average(numpy, mean, wall_kji)
    second_p = homogeneous_average(numpy, second, wall_kji)
    variance = numpy.maximum(second_p - mean_p ** 2, 0.0)
 
    # Wall-normal cell-centre coordinates, taken from the node coordinates of the
    # wall-normal axis. nodes is (KM, JM, IM, 3) in the same (k, j, i) order.
    axis_nodes = {0: nodes[:, 0, 0, :], 1: nodes[0, :, 0, :], 2: nodes[0, 0, :, :]}[wall_kji]
    axis_coord = axis_nodes[:, wall_c]
    y_centres = 0.5 * (axis_coord[:-1] + axis_coord[1:])
    if y_centres.size != mean_p.shape[0]:
        raise SystemExit(f"wall-normal cell count mismatch: grid gives {y_centres.size}, "
                         f"statistics give {mean_p.shape[0]}.")
 
    nu, h = args.viscosity, args.half_height
    if args.body_force is not None:
        u_tau = math.sqrt(args.body_force * h)
        source = f"sqrt(f*h) with f={args.body_force:.8e}"
    else:
        # Wall-shear estimate from the first cell: du/dy at the wall.
        dudy = abs(mean_p[0, stream_c]) / abs(y_centres[0] - axis_coord[0])
        u_tau = math.sqrt(nu * dudy)
        source = "sqrt(nu*du/dy) from the first cell (approximate; prefer --body-force)"
    if u_tau <= 0.0:
        raise SystemExit("computed a non-positive friction velocity; check the inputs.")
 
    wall_distance = numpy.minimum(y_centres - axis_coord[0], axis_coord[-1] - y_centres)
    with open(args.output, "w", encoding="utf-8") as handle:
        handle.write(f"# u_tau = {u_tau:.10e}  ({source})\n")
        handle.write(f"# Re_tau = u_tau*h/nu = {u_tau * h / nu:.6f}\n")
        handle.write(f"# nu = {nu:.10e}, h = {h:.10e}, window = {args.window}\n")
        handle.write("y,y_plus,U_plus,u_rms_plus,v_rms_plus,w_rms_plus,"
                     "uv_plus,U_plus_loglaw,U_plus_sublayer\n")
        for index in range(y_centres.size):
            y_plus = wall_distance[index] * u_tau / nu
            loglaw = (1.0 / 0.41) * math.log(y_plus) + 5.2 if y_plus > 0.0 else float("nan")
            handle.write(
                f"{y_centres[index]:.10e},{y_plus:.10e},"
                f"{mean_p[index, stream_c] / u_tau:.10e},"
                f"{math.sqrt(variance[index, stream_c]) / u_tau:.10e},"
                f"{math.sqrt(variance[index, wall_c]) / u_tau:.10e},"
                f"{math.sqrt(variance[index, span_c]) / u_tau:.10e},"
                f"{(second_p[index, stream_c] - mean_p[index, stream_c] ** 2) / (u_tau ** 2):.10e},"
                f"{loglaw:.10e},{y_plus:.10e}\n")
 
    print(f"u_tau = {u_tau:.10e}  ({source})")
    print(f"Re_tau = {u_tau * h / nu:.4f}")
    print(f"wrote {y_centres.size} wall-normal stations to {args.output}")
    return 0
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

REPO_ROOT

wall_normal_profile.REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", ".."))

Definition at line 43 of file wall_normal_profile.py.

SPECTRA_GEN

wall_normal_profile.SPECTRA_GEN = os.path.join(REPO_ROOT, "generators", "spectra.gen")

Definition at line 44 of file wall_normal_profile.py.

AXIS_TO_KJI

dict wall_normal_profile.AXIS_TO_KJI = {"Xi": 2, "Eta": 1, "Zeta": 0}

Definition at line 47 of file wall_normal_profile.py.

AXIS_TO_COMPONENT

dict wall_normal_profile.AXIS_TO_COMPONENT = {"Xi": 0, "Eta": 1, "Zeta": 2}

Definition at line 49 of file wall_normal_profile.py.