Source code for xgc_analysis.analysis.transport.neoclassical_workflow

"""Optional file output and standalone neoclassical analysis workflow."""

from __future__ import annotations

from pathlib import Path
import numpy as np

from .neoclassical import NeoResult, PROFILE_INFO, compute_neoclassical_transport
from .neoclassical_plots import save_neoclassical_svg_plots


[docs] def write_neoclassical_bp(result: NeoResult, path: str | Path) -> Path: """Write one documented ADIOS BP package with averaging provenance. Parameters ---------- result Previously computed analysis result. path Input HDF5 path or output BP path, according to the routine. Returns ------- Path written ADIOS BP package. """ import adios2 path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) with adios2.Stream(str(path), "w") as stream: stream.write_attribute("description", "Time-averaged XGC and local neoclassical transport comparison on the outer-midplane flux-surface grid.") stream.write_attribute("units", "mixed; see each variable's units attribute") stream.write_attribute("source_product", result.source) stream.write_attribute("flux_surface_average", result.fsa) stream.write_attribute("gstep_start", int(result.gstep_range[0])) stream.write_attribute("gstep_end", int(result.gstep_range[1])) stream.write_attribute("time_start_s", float(result.time_range_s[0])) stream.write_attribute("time_end_s", float(result.time_range_s[1])) stream.write_attribute("sample_count", int(result.sample_count)) stream.write_attribute("smoothing_sigma_surfaces", float(result.smoothing_sigma_surfaces)) stream.write_attribute("denominator_smoothing_sigma_surfaces", float(result.denominator_smoothing_sigma_surfaces)) stream.write_attribute("current_drive_enabled", int(result.current_drive)) stream.write_attribute("analytical_domain", "psi_pol_norm < 1; analytical values at the separatrix and in the SOL are NaN") for name, values in result.profiles.items(): description, units = PROFILE_INFO[name] stream.write(name, np.asarray(values, dtype=np.float64)) stream.write_attribute("description", description, variable_name=name) stream.write_attribute("units", units, variable_name=name) return path
[docs] def run_neoclassical_transport( simulation, *, output_bp: str | Path | None = None, plot_directory: str | Path | None = None, plot_quantities=None, show_uncertainty: bool = True, uncertainty_sigma: float = 1.0, **compute_options, ) -> NeoResult: """Run the analysis from a script, optionally writing BP and SVG plots. ``simulation`` is an initialized :class:`xgc_analysis.simulation.Simulation`. All other computation options are forwarded to :func:`compute_neoclassical_transport`. The GUI calls the compute and plot functions directly and never requests output files. Parameters ---------- simulation Initialized XGC Simulation containing the needed geometry and diagnostics. output_bp Optional path for the documented BP output. plot_directory Optional directory for SVG plots. plot_quantities Optional subset of SVG quantity names. show_uncertainty Shade the temporal spread when plotting. uncertainty_sigma Width of the shaded band in standard deviations. compute_options Keyword options forwarded to compute_neoclassical_transport. Returns ------- NeoResult calculated profiles; requested files are written as side effects. """ result = compute_neoclassical_transport(simulation, **compute_options) if output_bp is not None: write_neoclassical_bp(result, output_bp) if plot_directory is not None: save_neoclassical_svg_plots(result, plot_directory, quantities=plot_quantities, show_uncertainty=show_uncertainty, uncertainty_sigma=uncertainty_sigma) return result