"""Dataset loading, access, and query helpers for FieldView.
The :mod:`fieldview.data` namespace covers the full dataset lifecycle: loading
supported file formats, selecting or discovering the current dataset, probing
values, accessing registered scalar/vector/position arrays, working with
transient data, and retrieving session or window metadata tied to the active
scene.
Typical usage:
ds = fv.data.load_plot3d("grid.g", "sol.q")
"""
from __future__ import annotations
import json
import importlib
import os
from array import array
from dataclasses import dataclass
from enum import Enum
from typing import (
NamedTuple,
Protocol,
SupportsFloat,
SupportsIndex,
SupportsInt,
cast,
)
from collections.abc import Callable, Iterable, Iterator, Mapping
from . import constant
from . import _core as _core_module
from ._core import Dataset as _CoreDataset
from ._core_utils import _core_call
from .exceptions import (
CoreError,
InvalidArgumentError,
InvalidDatasetError,
_send_fv_error_msg,
)
from .utils import Range, _coerce_pathlike_str, _coerce_xyz_triplet
DatasetArraySnapshotView = _core_module.DatasetArraySnapshotView
__all__: list[str] = [
"DuplicationOptions",
"Dataset",
"get_current",
"set_current",
"load_plot3d",
"load_overflow2",
"load_fvuns",
"load_vtk_structured",
"load_acusolve_direct",
"load_acusolve_fvuns",
"load_ansys_cfx_fvuns",
"load_fluent_cff_direct",
"load_append_sampled_data",
"load_pw_common",
"load_cfdpp_fvuns",
"load_cgns_structured",
"load_cgns_unstructured",
"load_cgns_unstructured_hybrid",
"load_cobalt_fvuns",
"load_converge_fvuns",
"load_ensight",
"load_flow3d_animation",
"load_flow3d_restart",
"load_fluent_fvuns",
"load_fluent_unstructured",
"load_fluent_cas_dat_direct",
"load_fluent_direct",
"load_havoc",
"load_lsdyna_d3plot",
"load_lsdyna_state",
"load_nparc_wind",
"load_openfoam_direct",
"load_openfoam_fvuns",
"load_sc_tetra",
"load_sc_flow",
"load_scryu",
"load_sc_stream",
"load_starccm_fvuns",
"load_starcd_fvuns",
"load_stl",
"load_surface_sampled_data",
"load_tecplot_360",
"load_ultrafluidx_direct",
"load_vtk_unstructured_hybrid",
"load_wind_structured",
"load_wind_unstructured",
"load_xdb_import",
"reader_aliases",
"dataset",
"Vec3",
"ScalarProbe",
"VectorProbe",
"ProbeResult",
"DatasetArraySnapshotView",
"ScalarArrayRef",
"ScalarRegistry",
"probe",
"probe_ijk",
"TransientInfo",
"IntegrationResult",
"DatasetBounds",
"WindowInfo",
"WindowList",
"WindowSplitResult",
"CameraPose",
"CameraExactState",
"ViewState",
"CameraState",
"VectorArrayRef",
"VectorRegistry",
"PerspectiveState",
"SessionState",
"get_session_state",
]
_current_dataset: Dataset | None = None
_dataset_registry: dict[int, "Dataset"] = {}
_IntLike = SupportsInt | SupportsIndex | str | bytes
_FloatLike = SupportsFloat | SupportsIndex | str | bytes
_PathLikeStr = str | os.PathLike[str]
class _NumpyArrayLike(Protocol):
ndim: int
shape: tuple[int, ...]
def copy(self) -> object: ...
def setflags(self, write: bool = ...) -> None: ...
class _NumpyModule(Protocol):
float32: object
def asarray(self, values: object, dtype: object = ...) -> _NumpyArrayLike: ...
def ascontiguousarray(self, values: object, dtype: object = ...) -> object: ...
class _ReaderLoader(Protocol):
def load(self, **kwargs: object) -> Dataset: ...
class _ReaderWithServerConfig(Protocol):
def server_config(self, server_config: object) -> None: ...
def _load_numpy() -> _NumpyModule | None:
try:
module = importlib.import_module("numpy")
return cast(_NumpyModule, cast(object, module))
except ImportError:
return None
def _core_getattr(obj: object, name: str) -> object:
return cast(object, _core_call(getattr, obj, name))
def _core_setattr(obj: object, name: str, value: object) -> None:
_core_call(setattr, obj, name, value)
def _core_getstr(obj: object, name: str) -> str:
return str(_core_getattr(obj, name))
def _core_getbool(obj: object, name: str) -> bool:
return bool(_core_getattr(obj, name))
def _core_getint(obj: object, name: str) -> int:
return int(cast(_IntLike, _core_getattr(obj, name)))
def _core_getfloat(obj: object, name: str) -> float:
return float(cast(_FloatLike, _core_getattr(obj, name)))
def _to_int(value: object) -> int:
return int(cast(_IntLike, value))
def _to_float(value: object) -> float:
return float(cast(_FloatLike, value))
class DuplicationOptions:
"""Options for dataset duplication operations.
Args:
operation: One of "none", "mirror", "rotate", or "translate".
mirror_axes: constant.Axes to mirror, e.g. ``constant.Axes.XY``.
rotate_axis: constant.Axis to rotate around, e.g. ``constant.Axis.Z``.
rotate_copies: Number of copies for rotate. Default 2.
rotate_sweep: Sweep angle for rotate, in degrees. Default 360.0.
translate_axes: constant.Axes to translate, e.g. ``constant.Axes.XYZ``.
translate_copies: Tuple of (x, y, z) copies.
translate_deltas: Tuple of (dx, dy, dz) offsets.
"""
__slots__ = (
"operation",
"mirror_axes",
"rotate_axis",
"rotate_copies",
"rotate_sweep",
"translate_axes",
"translate_copies_x",
"translate_copies_y",
"translate_copies_z",
"translate_delta_x",
"translate_delta_y",
"translate_delta_z",
)
def __init__(
self,
operation: str | constant.DuplicateOperation = constant.DuplicateOperation.NONE,
mirror_axes: str | constant.Axes | None = None,
rotate_axis: str | constant.Axis | None = None,
rotate_copies: int | None = None,
rotate_sweep: float | None = None,
translate_axes: str | constant.Axes | None = None,
translate_copies_x: int | None = None,
translate_copies_y: int | None = None,
translate_copies_z: int | None = None,
translate_delta_x: float | None = None,
translate_delta_y: float | None = None,
translate_delta_z: float | None = None,
):
if isinstance(operation, constant.DuplicateOperation):
operation = operation.value
self.operation = operation
self.mirror_axes = (
mirror_axes.value if isinstance(mirror_axes, constant.Axes) else mirror_axes
)
self.rotate_axis = (
rotate_axis.value if isinstance(rotate_axis, constant.Axis) else rotate_axis
)
self.rotate_copies = rotate_copies
self.rotate_sweep = rotate_sweep
self.translate_axes = (
translate_axes.value
if isinstance(translate_axes, constant.Axes)
else translate_axes
)
self.translate_copies_x = translate_copies_x
self.translate_copies_y = translate_copies_y
self.translate_copies_z = translate_copies_z
self.translate_delta_x = translate_delta_x
self.translate_delta_y = translate_delta_y
self.translate_delta_z = translate_delta_z
def to_dict(self) -> dict[str, object]:
"""Return a dict payload for the duplication command."""
data: dict[str, object] = {"operation": self.operation}
if self.mirror_axes is not None:
data["mirror_axes"] = self.mirror_axes
if self.rotate_axis is not None:
data["rotate_axis"] = self.rotate_axis
if self.rotate_copies is not None:
data["rotate_copies"] = self.rotate_copies
if self.rotate_sweep is not None:
data["rotate_sweep"] = self.rotate_sweep
if self.translate_axes is not None:
data["translate_axes"] = self.translate_axes
if self.translate_copies_x is not None:
data["translate_copies_x"] = self.translate_copies_x
if self.translate_copies_y is not None:
data["translate_copies_y"] = self.translate_copies_y
if self.translate_copies_z is not None:
data["translate_copies_z"] = self.translate_copies_z
if self.translate_delta_x is not None:
data["translate_delta_x"] = self.translate_delta_x
if self.translate_delta_y is not None:
data["translate_delta_y"] = self.translate_delta_y
if self.translate_delta_z is not None:
data["translate_delta_z"] = self.translate_delta_z
return data
class Vec3(NamedTuple):
"""Three-component point or vector."""
x: float
y: float
z: float
class ScalarProbe(NamedTuple):
"""Function name and scalar value pair."""
func: str | None
value: float | None
class VectorProbe(NamedTuple):
"""Function name and vector value pair."""
func: str | None
value: Vec3 | None
[docs]
@dataclass(frozen=True)
class ProbeResult:
"""Probe result for probed scalar/iso/threshold/vector functions.
``hit`` reports whether the probe resolved to a dataset location.
When ``hit`` is ``False``, function names may still be populated while
sampled values remain ``None``. For IJK misses, ``point`` may also be
``None`` because no displayed-space location was resolved.
"""
hit: bool
point: Vec3 | None
region: int | None
grid_index: int | None
grid: int | None
ijk: tuple[float, float, float] | None
scalar: ScalarProbe
iso: ScalarProbe
threshold: ScalarProbe
vector: VectorProbe
[docs]
@dataclass(frozen=True)
class IntegrationResult:
"""Surface integration result.
Instances are returned by ``surface.integrate()`` and the coord/iso
``integrate_partial_surface()`` methods.
Common fields:
- ``integral_type`` identifies the host-side integration mode.
- ``scalar_function`` is the scalar function that was integrated.
- ``area`` is the surface integral ∫ f dA of the active scalar.
- ``sum`` is the total integral of the active scalar over the surface.
- ``average`` is the area-weighted mean, equal to ``sum / area``
when the host provides it.
When the target surface has normals and an active vector function,
``has_surface_normals`` is ``True`` and the normal/vector flux fields
such as ``sum_nx`` and ``sum_v_dot_n`` may also be populated.
Example:
.. code-block:: python
>>> import os
>>> import fieldview as fv
>>> data_dir = os.path.join(fv.home, "examples", "f18")
>>> ds = fv.data.load_plot3d(
... os.path.join(data_dir, "f18i9b_g_bin"),
... os.path.join(data_dir, "f18i9b_q_bin"),
... )
>>> surf = fv.vis.create_boundary(ds)
>>> surf.scalar_func = "Pressure [PLOT3D]"
>>> result = surf.integrate()
>>> result.area, result.sum, result.average
"""
integral_type: str
scalar_function: str | None
area: float
sum: float
average: float | None
surface: str | None = None
has_surface_normals: bool = False
vector_function: str | None = None
sum_nx: float | None = None
sum_ny: float | None = None
sum_nz: float | None = None
sum_v_dot_n: float | None = None
def _coerce_scalar_values_buffer(values: object) -> object:
np = _load_numpy()
try:
if np is not None:
arr = np.asarray(values, dtype=np.float32)
if arr.ndim != 1:
raise InvalidArgumentError("Scalar values must be one-dimensional.")
return np.ascontiguousarray(arr, dtype=np.float32)
if isinstance(values, array):
if values.typecode == "f":
return values
return array("f", values)
return array(
"f",
[float(cast(_FloatLike, item)) for item in cast(Iterable[object], values)],
)
except InvalidArgumentError:
raise
except (TypeError, ValueError) as exc:
raise InvalidArgumentError(
"Scalar values must be a one-dimensional array-like of floats."
) from exc
def _coerce_vector_values_buffer(values: object) -> object:
np = _load_numpy()
try:
if np is not None:
arr = np.asarray(values, dtype=np.float32)
if arr.ndim != 2 or arr.shape[1] != 3:
raise InvalidArgumentError("Vector values must have shape (n, 3).")
return np.ascontiguousarray(arr, dtype=np.float32)
rows: list[tuple[float, float, float]] = []
for row in cast(Iterable[object], values):
if not isinstance(row, (list, tuple)):
raise InvalidArgumentError("Vector values must have shape (n, 3).")
components = tuple(cast(tuple[object, ...], row))
if len(components) != 3:
raise InvalidArgumentError("Vector values must have shape (n, 3).")
rows.append(
(
_to_float(components[0]),
_to_float(components[1]),
_to_float(components[2]),
)
)
if not rows:
raise InvalidArgumentError("Vector values must have shape (n, 3).")
flat = array("f")
for row in rows:
flat.extend(row)
return flat
except InvalidArgumentError:
raise
except (TypeError, ValueError) as exc:
raise InvalidArgumentError(
"Vector values must be an array-like of shape (n, 3)."
) from exc
def _extract_created_function_id(payload: Mapping[str, object], kind: str) -> int:
function_id_value = payload.get("function_id")
if function_id_value is None:
raise CoreError(
f"FieldView did not return 'function_id' when creating the {kind} function."
)
try:
function_id = int(cast(_IntLike, function_id_value))
except (TypeError, ValueError) as exc:
raise CoreError(
f"FieldView returned a non-integer 'function_id' for the created {kind} function."
) from exc
if function_id < 0:
raise CoreError(
f"FieldView returned an invalid 'function_id' for the created {kind} function."
)
return function_id
[docs]
class ScalarArrayRef:
"""Reference to a named scalar function within a dataset."""
__slots__ = ("_dataset", "_name", "_function_id")
def __init__(self, dataset: "Dataset", name: str, function_id: int):
self._dataset = dataset
self._name = name
self._function_id = int(function_id)
@property
def name(self) -> str:
return self._name
@property
def function_id(self) -> int:
return self._function_id
[docs]
def snapshot(self, *, grid: int = 1) -> DatasetArraySnapshotView:
"""Return a read-only snapshot view for this scalar on ``grid``.
If you convert the snapshot with :func:`numpy.asarray`, the resulting
NumPy view keeps the snapshot backing alive even if the local
``snap`` variable is deleted.
"""
self._dataset._ensure_valid()
return _core_call(
self._dataset._core._scalar_array_snapshot, self._function_id, int(grid)
)
[docs]
def to_numpy(self, *, grid: int = 1, copy: bool = False) -> object:
"""Return this scalar snapshot as a NumPy array.
Args:
grid: 1-based grid number.
copy: When ``False`` (default), return a read-only NumPy view over
the snapshot buffer when possible. When ``True``, return an
independent writable NumPy copy.
Notes:
With ``copy=False``, the returned NumPy array keeps the snapshot
backing alive, so the array remains valid even if the local
``snap`` variable goes out of scope.
"""
np = _load_numpy()
if np is None:
raise CoreError(
"NumPy is not installed; use snapshot() for a buffer-compatible view."
)
snap = self.snapshot(grid=grid)
arr = np.asarray(snap)
if copy:
return arr.copy()
arr.setflags(write=False)
return arr
def __repr__(self) -> str:
return f"ScalarArrayRef(name={self._name!r}, function_id={self._function_id})"
[docs]
class VectorArrayRef:
"""Reference to a named vector function within a dataset."""
__slots__ = ("_dataset", "_name", "_function_id")
def __init__(self, dataset: "Dataset", name: str, function_id: int):
self._dataset = dataset
self._name = name
self._function_id = int(function_id)
@property
def name(self) -> str:
return self._name
@property
def function_id(self) -> int:
return self._function_id
[docs]
def snapshot(self, *, grid: int = 1) -> DatasetArraySnapshotView:
"""Return a read-only snapshot view for this vector on ``grid``.
If you convert the snapshot with :func:`numpy.asarray`, the resulting
NumPy view keeps the snapshot backing alive even if the local
``snap`` variable is deleted.
"""
self._dataset._ensure_valid()
return _core_call(
self._dataset._core._vector_array_snapshot, self._function_id, int(grid)
)
[docs]
def to_numpy(self, *, grid: int = 1, copy: bool = False) -> object:
"""Return this vector snapshot as a NumPy array.
Args:
grid: 1-based grid number.
copy: When ``False`` (default), return a read-only NumPy view over
the snapshot buffer when possible. When ``True``, return an
independent writable NumPy copy.
Notes:
With ``copy=False``, the returned NumPy array keeps the snapshot
backing alive, so the array remains valid even if the local
``snap`` variable goes out of scope.
"""
np = _load_numpy()
if np is None:
raise CoreError(
"NumPy is not installed; use snapshot() for a buffer-compatible view."
)
snap = self.snapshot(grid=grid)
arr = np.asarray(snap)
if copy:
return arr.copy()
arr.setflags(write=False)
return arr
def __repr__(self) -> str:
return f"VectorArrayRef(name={self._name!r}, function_id={self._function_id})"
[docs]
class ScalarRegistry:
"""Container-like access to dataset scalar functions."""
__slots__ = ("_dataset",)
def __init__(self, dataset: "Dataset"):
self._dataset = dataset
def __contains__(self, name: object) -> bool:
return isinstance(name, str) and name in self._dataset._scalar_function_ids
def __getitem__(self, name: str) -> ScalarArrayRef:
self._dataset._ensure_valid()
function_id = self._dataset._scalar_function_ids.get(name)
if function_id is None:
raise KeyError(name)
return ScalarArrayRef(self._dataset, name, function_id)
def __iter__(self) -> Iterator[str]:
self._dataset._ensure_valid()
return iter(self._dataset._scalar_functions)
def __len__(self) -> int:
self._dataset._ensure_valid()
return len(self._dataset._scalar_functions)
[docs]
def snapshot(self, name: str, *, grid: int = 1) -> DatasetArraySnapshotView:
"""Return a read-only snapshot view for the named scalar function."""
return self[name].snapshot(grid=grid)
[docs]
def to_numpy(self, name: str, *, grid: int = 1, copy: bool = False) -> object:
"""Return the named scalar function as a NumPy array.
Args:
name: Scalar function name.
grid: 1-based grid number.
copy: When ``False`` (default), return a read-only NumPy view over
the snapshot buffer when possible. When ``True``, return an
independent writable NumPy copy.
"""
return self[name].to_numpy(grid=grid, copy=copy)
[docs]
def create(self, name: str, values: object, *, grid: int = 1) -> ScalarArrayRef:
"""Create or replace a named scalar function from array-like values."""
if not isinstance(name, str) or not name:
raise InvalidArgumentError("Scalar name must be a non-empty string.")
buffer_values = _coerce_scalar_values_buffer(values)
payload = _core_call(
self._dataset._core._create_scalar_array, name, int(grid), buffer_values
)
_refresh_function_lists(self._dataset)
function_id = _extract_created_function_id(payload, "scalar")
return ScalarArrayRef(self._dataset, name, function_id)
[docs]
class VectorRegistry:
"""Container-like access to dataset vector functions."""
__slots__ = ("_dataset",)
def __init__(self, dataset: "Dataset"):
self._dataset = dataset
def __contains__(self, name: object) -> bool:
return isinstance(name, str) and name in self._dataset._vector_function_ids
def __getitem__(self, name: str) -> VectorArrayRef:
self._dataset._ensure_valid()
function_id = self._dataset._vector_function_ids.get(name)
if function_id is None:
raise KeyError(name)
return VectorArrayRef(self._dataset, name, function_id)
def __iter__(self) -> Iterator[str]:
self._dataset._ensure_valid()
return iter(self._dataset._vector_functions)
def __len__(self) -> int:
self._dataset._ensure_valid()
return len(self._dataset._vector_functions)
[docs]
def snapshot(self, name: str, *, grid: int = 1) -> DatasetArraySnapshotView:
"""Return a read-only snapshot view for the named vector function."""
return self[name].snapshot(grid=grid)
[docs]
def to_numpy(self, name: str, *, grid: int = 1, copy: bool = False) -> object:
"""Return the named vector function as a NumPy array.
Args:
name: Vector function name.
grid: 1-based grid number.
copy: When ``False`` (default), return a read-only NumPy view over
the snapshot buffer when possible. When ``True``, return an
independent writable NumPy copy.
"""
return self[name].to_numpy(grid=grid, copy=copy)
[docs]
def create(self, name: str, values: object, *, grid: int = 1) -> VectorArrayRef:
"""Create or replace a named vector function from array-like values."""
if not isinstance(name, str) or not name:
raise InvalidArgumentError("Vector name must be a non-empty string.")
buffer_values = _coerce_vector_values_buffer(values)
payload = _core_call(
self._dataset._core._create_vector_array, name, int(grid), buffer_values
)
_refresh_function_lists(self._dataset)
function_id = _extract_created_function_id(payload, "vector")
return VectorArrayRef(self._dataset, name, function_id)
[docs]
class PositionRegistry:
"""Access to native XYZ node positions for dataset grids.
This registry is grid-oriented rather than function-oriented. Reads are
snapshot-only and return ``(n, 3)`` XYZ data for the requested grid.
"""
__slots__ = ("_dataset",)
def __init__(self, dataset: "Dataset"):
self._dataset = dataset
[docs]
def snapshot(self, *, grid: int = 1) -> DatasetArraySnapshotView:
"""Return a read-only snapshot view of native XYZ positions on ``grid``.
If you convert the snapshot with :func:`numpy.asarray`, the resulting
NumPy view keeps the snapshot backing alive even if the local
``snap`` variable is deleted.
"""
self._dataset._ensure_valid()
return _core_call(self._dataset._core._position_array_snapshot, int(grid))
[docs]
def to_numpy(self, *, grid: int = 1, copy: bool = False) -> object:
"""Return native XYZ positions as a NumPy array.
Args:
grid: 1-based grid number.
copy: When ``False`` (default), return a read-only NumPy view over
the snapshot buffer when possible. When ``True``, return an
independent writable NumPy copy.
Returns:
A NumPy array with shape ``(n, 3)`` in native XYZ order.
Notes:
With ``copy=False``, the returned NumPy array keeps the snapshot
backing alive, so the array remains valid even if the local
``snap`` variable goes out of scope.
"""
np = _load_numpy()
if np is None:
raise CoreError(
"NumPy is not installed; use snapshot() for a buffer-compatible view."
)
snap = self.snapshot(grid=grid)
arr = np.asarray(snap)
if copy:
return arr.copy()
arr.setflags(write=False)
return arr
[docs]
@dataclass(frozen=True)
class WindowInfo:
"""Metadata for a single graphics window."""
window: int
current: bool
scene_index: int
parent_window: int | None
label: str | None
environment: str | None
view_sync_enabled: bool
background: str | None
dataset_ids: list[int]
environment_id: int | None = None
background_image: str | None = None
[docs]
@dataclass(frozen=True)
class WindowList:
"""Window enumeration payload returned by :mod:`fieldview.layout`."""
current_window: int | None
windows: list[WindowInfo]
[docs]
@dataclass(frozen=True)
class WindowSplitResult:
"""Result metadata for a split-window operation."""
source_window: int
new_window: int
mode: str
orientation: str
[docs]
@dataclass(frozen=True)
class PerspectiveState:
"""Perspective state for a graphics window."""
enabled: bool
angle: float
@dataclass(frozen=True)
class CameraExactState:
"""Exact FieldView transform state for deterministic camera replay.
This stores the raw world/view transform pieces that classic ``.vct``
restore uses internally. It remains available as a low-level payload type,
while :class:`ViewState` is the primary deterministic replay object exposed
by :mod:`fieldview.camera`.
"""
zoom: float
scale: float
rotation_angle: float
rotation_axis: tuple[float, float, float]
translation: tuple[float, float, float]
perspective_z: float
rotation_center_on: bool = False
rotation_center: tuple[float, float, float] = (0.0, 0.0, 0.0)
def to_payload(self) -> dict[str, object]:
"""Return the host JSON payload used by camera replay."""
return {
"zoom": self.zoom,
"scale": self.scale,
"rotation_angle": self.rotation_angle,
"rotation_axis": list(self.rotation_axis),
"translation": list(self.translation),
"perspective_z": self.perspective_z,
"rotation_center_on": self.rotation_center_on,
"rotation_center": list(self.rotation_center),
}
[docs]
@dataclass(frozen=True)
class CameraPose:
"""Human-readable camera pose summary.
This describes the visible camera tuple in terms of ``eye``, ``target``,
and ``up`` plus the current perspective mode. It is useful for inspection
and approximate camera manipulation, but unlike :class:`ViewState` it is
not the authoritative deterministic replay format for FieldView views.
"""
eye: tuple[float, float, float]
target: tuple[float, float, float]
up: tuple[float, float, float]
perspective_enabled: bool
perspective_angle: float
CameraState = CameraPose
[docs]
@dataclass(frozen=True)
class ViewState:
"""Deterministic FieldView view state.
This is the exact replay representation for camera/view restore. It stores
the same raw transform pieces that classic ``.vct`` restore uses along with
the public perspective settings needed by the host command handler.
"""
perspective_enabled: bool
perspective_angle: float
perspective_z: float
rotation_angle: float
rotation_axis: tuple[float, float, float]
translation: tuple[float, float, float]
scale: float
zoom: float
rotation_center_on: bool = False
rotation_center: tuple[float, float, float] = (0.0, 0.0, 0.0)
[docs]
@classmethod
def from_exact(
cls,
exact_state: CameraExactState,
*,
perspective_enabled: bool,
perspective_angle: float,
) -> ViewState:
"""Create a deterministic view state from the exact transform payload."""
return cls(
perspective_enabled=perspective_enabled,
perspective_angle=perspective_angle,
perspective_z=exact_state.perspective_z,
rotation_angle=exact_state.rotation_angle,
rotation_axis=exact_state.rotation_axis,
translation=exact_state.translation,
scale=exact_state.scale,
zoom=exact_state.zoom,
rotation_center_on=exact_state.rotation_center_on,
rotation_center=exact_state.rotation_center,
)
[docs]
def to_exact_state(self) -> CameraExactState:
"""Return the underlying exact transform payload."""
return CameraExactState(
zoom=self.zoom,
scale=self.scale,
rotation_angle=self.rotation_angle,
rotation_axis=self.rotation_axis,
translation=self.translation,
perspective_z=self.perspective_z,
rotation_center_on=self.rotation_center_on,
rotation_center=self.rotation_center,
)
def to_payload(self) -> dict[str, object]:
"""Return the host JSON payload for deterministic replay."""
payload = self.to_exact_state().to_payload()
payload["perspective_enabled"] = self.perspective_enabled
payload["perspective_angle"] = self.perspective_angle
return payload
[docs]
@dataclass(frozen=True)
class TransientInfo:
"""Transient dataset state.
Instances are returned by :meth:`Dataset.transient_info`.
Example:
.. code-block:: python
import os
import fieldview as fv
uns_file = os.path.join(fv.home, "examples", "rectangular_duct", "rect_duct_010.uns")
ds = fv.data.load_fvuns(uns_file, transient=True)
info = ds.transient_info()
print(info.time_step_values[:3])
"""
time_step: int
solution_time: float
total_time_steps: int
has_solution_times: bool
time_step_range: Range
solution_time_range: Range
time_step_values: tuple[int, ...]
solution_time_values: tuple[float, ...]
def _coerce_enum(value: object) -> object:
if isinstance(value, Enum):
return cast(object, value.value)
return value
def _reader_factory(name: str) -> Callable[..., object]:
from . import _readers
return cast(Callable[..., object], getattr(_readers, name))
def _reader_load(reader: object, **kwargs: object) -> Dataset:
return cast(_ReaderLoader, reader).load(**kwargs)
def _load_json_dict(value: object, label: str = "JSON payload") -> dict[str, object]:
payload = cast(object, json.loads(str(value)))
if isinstance(payload, dict):
return dict(cast(dict[str, object], payload))
raise CoreError(f"{label} must decode to a JSON object")
class DuplicationMirrorConfig:
__slots__ = ("_parent", "_axes")
def __init__(self, parent: "DuplicationController") -> None:
self._parent = parent
self._axes: str | None = None
@property
def axes(self) -> str | None:
return self._axes
@axes.setter
def axes(self, value: object) -> None:
coerced = _coerce_enum(value)
self._axes = None if coerced is None else str(coerced)
self._parent._apply("mirror")
class DuplicationRotateConfig:
__slots__ = ("_parent", "_axis", "_copies", "_sweep")
def __init__(self, parent: "DuplicationController") -> None:
self._parent = parent
self._axis = constant.Axis.Y.value
self._copies = 2
self._sweep = 360.0
@property
def axis(self) -> str:
return str(self._axis)
@axis.setter
def axis(self, value: object) -> None:
self._axis = str(_coerce_enum(value))
self._parent._apply("rotate")
@property
def copies(self) -> int:
return self._copies
@copies.setter
def copies(self, value: object) -> None:
self._copies = _to_int(value)
self._parent._apply("rotate")
@property
def sweep(self) -> float:
return self._sweep
@sweep.setter
def sweep(self, value: object) -> None:
self._sweep = _to_float(value)
self._parent._apply("rotate")
class DuplicationTranslateConfig:
__slots__ = ("_parent", "_axes", "_copies", "_deltas")
def __init__(self, parent: "DuplicationController") -> None:
self._parent = parent
self._axes: str | None = None
self._copies = (2, 2, 2)
self._deltas = (0.0, 0.0, 0.0)
@property
def axes(self) -> str | None:
return self._axes
@axes.setter
def axes(self, value: object) -> None:
coerced = _coerce_enum(value)
self._axes = None if coerced is None else str(coerced)
self._parent._apply("translate")
@property
def copies(self) -> tuple[int, int, int]:
return self._copies
@copies.setter
def copies(self, value: object) -> None:
if not isinstance(value, (list, tuple)):
raise InvalidArgumentError("translate.copies must be a 3-tuple")
items = tuple(cast(tuple[object, ...], value))
if len(items) != 3:
raise InvalidArgumentError("translate.copies must be a 3-tuple")
self._copies = (
_to_int(items[0]),
_to_int(items[1]),
_to_int(items[2]),
)
self._parent._apply("translate")
@property
def deltas(self) -> tuple[float, float, float]:
return self._deltas
@deltas.setter
def deltas(self, value: object) -> None:
if not isinstance(value, (list, tuple)):
raise InvalidArgumentError("translate.deltas must be a 3-tuple")
items = tuple(cast(tuple[object, ...], value))
if len(items) != 3:
raise InvalidArgumentError("translate.deltas must be a 3-tuple")
self._deltas = (
_to_float(items[0]),
_to_float(items[1]),
_to_float(items[2]),
)
self._parent._apply("translate")
class DuplicationController:
__slots__ = ("_dataset", "mirror", "rotate", "translate")
def __init__(self, dataset: "Dataset") -> None:
self._dataset = dataset
self.mirror = DuplicationMirrorConfig(self)
self.rotate = DuplicationRotateConfig(self)
self.translate = DuplicationTranslateConfig(self)
def __call__(self, options: DuplicationOptions | dict[str, object]) -> object:
if isinstance(options, DuplicationOptions):
payload = options.to_dict()
elif isinstance(options, dict):
payload = options
else:
raise InvalidArgumentError(
"duplication options must be a DuplicationOptions or dict"
)
return _core_call(self._dataset._core.duplicate, payload)
def clear(self) -> None:
_core_call(self._dataset._core.duplicate, {"operation": "none"})
def _apply(self, operation: str) -> None:
payload: dict[str, object] = {"operation": operation}
if operation == "mirror":
if self.mirror.axes is None:
return
payload["mirror_axes"] = self.mirror.axes
elif operation == "rotate":
payload["rotate_axis"] = self.rotate.axis
payload["rotate_copies"] = self.rotate.copies
payload["rotate_sweep"] = self.rotate.sweep
elif operation == "translate":
if self.translate.axes is None:
return
payload["translate_axes"] = self.translate.axes
payload["translate_copies_x"] = self.translate.copies[0]
payload["translate_copies_y"] = self.translate.copies[1]
payload["translate_copies_z"] = self.translate.copies[2]
payload["translate_delta_x"] = self.translate.deltas[0]
payload["translate_delta_y"] = self.translate.deltas[1]
payload["translate_delta_z"] = self.translate.deltas[2]
_core_call(self._dataset._core.duplicate, payload)
class TransformController:
__slots__ = (
"_dataset",
"_scale",
"_scale_set",
"_translate",
"_translate_set",
"_rotate1",
"_rotate1_set",
"_rotate2",
"_rotate2_set",
"_rotate3",
"_rotate3_set",
"_pending",
)
def __init__(self, dataset: "Dataset") -> None:
self._dataset = dataset
self._scale = (1.0, 1.0, 1.0)
self._scale_set = False
self._translate = (0.0, 0.0, 0.0)
self._translate_set = False
self._rotate1 = (constant.Axis.Y.value, 0.0)
self._rotate1_set = False
self._rotate2 = (constant.Axis.Y.value, 0.0)
self._rotate2_set = False
self._rotate3 = (constant.Axis.Y.value, 0.0)
self._rotate3_set = False
self._pending = False
def clear(self) -> None:
defaults: dict[str, object] = {
"scale_x": 1.0,
"scale_y": 1.0,
"scale_z": 1.0,
"translate_x": 0.0,
"translate_y": 0.0,
"translate_z": 0.0,
"rotate1_axis": constant.Axis.Y.value,
"rotate1_angle": 0.0,
"rotate2_axis": constant.Axis.Y.value,
"rotate2_angle": 0.0,
"rotate3_axis": constant.Axis.Y.value,
"rotate3_angle": 0.0,
}
if _core_getint(self._dataset._core, "dataset_id") >= 0:
_core_call(self._dataset._core.transform, defaults)
self._scale = (1.0, 1.0, 1.0)
self._scale_set = False
self._translate = (0.0, 0.0, 0.0)
self._translate_set = False
self._rotate1 = (constant.Axis.Y.value, 0.0)
self._rotate1_set = False
self._rotate2 = (constant.Axis.Y.value, 0.0)
self._rotate2_set = False
self._rotate3 = (constant.Axis.Y.value, 0.0)
self._rotate3_set = False
self._pending = False
def _set_vec3(self, name: str, value: object) -> tuple[float, float, float]:
if not isinstance(value, (list, tuple)):
raise InvalidArgumentError("transform.{0} must be a 3-tuple".format(name))
items = tuple(cast(tuple[object, ...], value))
if len(items) != 3:
raise InvalidArgumentError("transform.{0} must be a 3-tuple".format(name))
return (_to_float(items[0]), _to_float(items[1]), _to_float(items[2]))
def _set_rotation(self, name: str, value: object) -> tuple[str, float]:
if not isinstance(value, (list, tuple)):
raise InvalidArgumentError(
"transform.{0} must be (axis, angle)".format(name)
)
items = tuple(cast(tuple[object, ...], value))
if len(items) != 2:
raise InvalidArgumentError(
"transform.{0} must be (axis, angle)".format(name)
)
axis = _coerce_enum(items[0])
if axis is None:
raise InvalidArgumentError(
"transform.{0} axis must be provided".format(name)
)
return (str(axis), _to_float(items[1]))
def _build_payload(self) -> dict[str, object]:
payload: dict[str, object] = {}
if self._scale_set:
payload["scale_x"] = self._scale[0]
payload["scale_y"] = self._scale[1]
payload["scale_z"] = self._scale[2]
if self._translate_set:
payload["translate_x"] = self._translate[0]
payload["translate_y"] = self._translate[1]
payload["translate_z"] = self._translate[2]
if self._rotate1_set:
payload["rotate1_axis"] = self._rotate1[0]
payload["rotate1_angle"] = self._rotate1[1]
if self._rotate2_set:
payload["rotate2_axis"] = self._rotate2[0]
payload["rotate2_angle"] = self._rotate2[1]
if self._rotate3_set:
payload["rotate3_axis"] = self._rotate3[0]
payload["rotate3_angle"] = self._rotate3[1]
return payload
def _apply(self) -> None:
payload = self._build_payload()
if not payload:
return
if _core_getint(self._dataset._core, "dataset_id") < 0:
self._pending = True
return
_core_call(self._dataset._core.transform, payload)
self._pending = False
def apply_pending(self) -> None:
if not self._pending:
return
if _core_getint(self._dataset._core, "dataset_id") < 0:
return
payload = self._build_payload()
if not payload:
self._pending = False
return
_core_call(self._dataset._core.transform, payload)
self._pending = False
@property
def scale(self) -> tuple[float, float, float]:
return self._scale
@scale.setter
def scale(self, value: object) -> None:
self._scale = self._set_vec3("scale", value)
self._scale_set = True
self._apply()
@property
def translate(self) -> tuple[float, float, float]:
return self._translate
@translate.setter
def translate(self, value: object) -> None:
self._translate = self._set_vec3("translate", value)
self._translate_set = True
self._apply()
@property
def rotate1(self) -> tuple[str, float]:
return self._rotate1
@rotate1.setter
def rotate1(self, value: object) -> None:
self._rotate1 = self._set_rotation("rotate1", value)
self._rotate1_set = True
self._apply()
@property
def rotate2(self) -> tuple[str, float]:
return self._rotate2
@rotate2.setter
def rotate2(self, value: object) -> None:
self._rotate2 = self._set_rotation("rotate2", value)
self._rotate2_set = True
self._apply()
@property
def rotate3(self) -> tuple[str, float]:
return self._rotate3
@rotate3.setter
def rotate3(self, value: object) -> None:
self._rotate3 = self._set_rotation("rotate3", value)
self._rotate3_set = True
self._apply()
[docs]
class Dataset:
"""Base dataset wrapper.
A Dataset instance tracks a loaded dataset in FieldView. If a new load
happens with ``input_mode=REPLACE``, existing dataset objects are invalidated.
"""
__slots__ = (
"_core",
"_duplication",
"_transform",
"_scalar_registry",
"_vector_registry",
"_position_registry",
"_scalar_functions",
"_vector_functions",
"_boundary_types",
"_scalar_function_ids",
"_vector_function_ids",
"_load_recipe",
)
_core: _CoreDataset
_duplication: DuplicationController
_transform: TransformController
_scalar_registry: ScalarRegistry
_vector_registry: VectorRegistry
_position_registry: PositionRegistry
_scalar_functions: list[str]
_vector_functions: list[str]
_boundary_types: list[str]
_scalar_function_ids: dict[str, int]
_vector_function_ids: dict[str, int]
_load_recipe: dict[str, object] | None
def __init__(self, core: _CoreDataset | None = None) -> None:
self._core = core or _CoreDataset()
self._duplication = DuplicationController(self)
self._transform = TransformController(self)
self._scalar_registry = ScalarRegistry(self)
self._vector_registry = VectorRegistry(self)
self._position_registry = PositionRegistry(self)
self._scalar_functions = []
self._vector_functions = []
self._boundary_types = []
self._scalar_function_ids = {}
self._vector_function_ids = {}
self._load_recipe = None
def __getattr__(self, name: str) -> object:
return _core_getattr(self._core, name)
@property
def dataset_id(self) -> int:
"""Return the host dataset identifier."""
return _core_getint(self._core, "dataset_id")
@property
def server(self) -> str:
"""Get or set the current server name."""
return _core_getstr(self._core, "server")
@server.setter
def server(self, value: str) -> None:
"""Set the current server name."""
_core_setattr(self._core, "server", value)
@property
def input_mode(self) -> str:
"""Return the dataset input mode label."""
return _core_getstr(self._core, "input_mode")
@property
def data_format(self) -> str:
"""Return the dataset format label."""
self._ensure_valid()
return _core_getstr(self._core, "data_format")
@property
def is_unstructured(self) -> bool:
"""Return True when the host dataset is unstructured."""
self._ensure_valid()
return self._is_unstructured_cached()
def _is_unstructured_cached(self) -> bool:
"""Return the cached dataset topology after prior validation."""
return _core_getbool(self._core, "is_unstructured")
@property
def grid_file(self) -> str:
"""Return the grid (or combined) file path."""
self._ensure_valid()
return _core_getstr(self._core, "grid_file")
@property
def result_file(self) -> str:
"""Return the results file path, if any."""
return _core_getstr(self._core, "result_file")
@property
def transient(self) -> bool:
"""Return True if the dataset is transient."""
return _core_getbool(self._core, "transient")
@property
def has_solution_times(self) -> bool:
"""Return True if solution times are available."""
return _core_getbool(self._core, "has_solution_times")
@property
def cur_time_step(self) -> int:
"""Return the current transient time step index."""
return _core_getint(self._core, "cur_time_step")
@property
def cur_solution_time(self) -> float:
"""Return the current transient solution time."""
return _core_getfloat(self._core, "cur_solution_time")
@property
def total_time_steps(self) -> int:
"""Return the total number of transient time steps."""
return _core_getint(self._core, "total_time_steps")
@property
def time_step_range(self) -> Range:
"""Return the min/max transient time-step range."""
return Range(
_core_getint(self._core, "time_step_min"),
_core_getint(self._core, "time_step_max"),
)
@property
def solution_time_range(self) -> Range:
"""Return the min/max transient solution-time range."""
return Range(
_core_getfloat(self._core, "solution_time_min"),
_core_getfloat(self._core, "solution_time_max"),
)
@property
def num_grids(self) -> int:
"""Return the number of grids in this dataset."""
self._ensure_valid()
return _core_getint(self._core, "num_grids")
@property
def scalar_functions(self) -> list[str]:
"""Return scalar function names loaded for this dataset (sorted).
The list is populated at load time and updated by APIs that add functions.
"""
self._ensure_valid()
return list(self._scalar_functions)
@property
def scalars(self) -> ScalarRegistry:
"""Return scalar-function registry access for this dataset."""
self._ensure_valid()
return self._scalar_registry
@property
def vector_functions(self) -> list[str]:
"""Return vector function names loaded for this dataset (sorted).
The list is populated at load time and updated by APIs that add functions.
"""
self._ensure_valid()
return list(self._vector_functions)
@property
def vectors(self) -> VectorRegistry:
"""Return vector-function registry access for this dataset."""
self._ensure_valid()
return self._vector_registry
@property
def positions(self) -> PositionRegistry:
"""Return native XYZ position snapshots for this dataset."""
self._ensure_valid()
return self._position_registry
@property
def boundary_types(self) -> list[str]:
"""Return boundary type names loaded for this dataset."""
self._ensure_valid()
return list(self._boundary_types)
def _ensure_valid(self) -> None:
try:
self._core._ensure_valid()
except RuntimeError as exc:
msg = str(exc)
if _is_invalid_dataset_message(msg):
raise InvalidDatasetError(msg) from exc
if msg:
_send_fv_error_msg(msg)
raise CoreError(msg) from exc
@property
def xmin(self) -> float:
"""Return the minimum X bound."""
return _core_getfloat(self._core, "xmin")
@property
def xmax(self) -> float:
"""Return the maximum X bound."""
return _core_getfloat(self._core, "xmax")
@property
def ymin(self) -> float:
"""Return the minimum Y bound."""
return _core_getfloat(self._core, "ymin")
@property
def ymax(self) -> float:
"""Return the maximum Y bound."""
return _core_getfloat(self._core, "ymax")
@property
def zmin(self) -> float:
"""Return the minimum Z bound."""
return _core_getfloat(self._core, "zmin")
@property
def zmax(self) -> float:
"""Return the maximum Z bound."""
return _core_getfloat(self._core, "zmax")
@property
def rmin(self) -> float:
"""Return the minimum R bound."""
return _core_getfloat(self._core, "rmin")
@property
def rmax(self) -> float:
"""Return the maximum R bound."""
return _core_getfloat(self._core, "rmax")
@property
def tmin(self) -> float:
"""Return the minimum T bound."""
return _core_getfloat(self._core, "tmin")
@property
def tmax(self) -> float:
"""Return the maximum T bound."""
return _core_getfloat(self._core, "tmax")
@property
def visibility(self) -> bool:
"""Get or set dataset visibility state."""
return _core_getbool(self._core, "visibility")
@visibility.setter
def visibility(self, value: bool) -> None:
"""Set dataset visibility state."""
_core_setattr(self._core, "visibility", bool(value))
[docs]
def dump(self) -> None:
"""Print dataset state to stdout."""
time_step_range = self.time_step_range
solution_time_range = self.solution_time_range
items = (
("dataset_id", self.dataset_id),
("server", self.server),
("input_mode", self.input_mode),
("data_format", self.data_format),
("grid_file", self.grid_file),
("result_file", self.result_file),
("transient", self.transient),
("has_solution_times", self.has_solution_times),
("cur_time_step", self.cur_time_step),
("cur_solution_time", self.cur_solution_time),
("total_time_steps", self.total_time_steps),
("time_step_min", time_step_range.min),
("time_step_max", time_step_range.max),
("solution_time_min", solution_time_range.min),
("solution_time_max", solution_time_range.max),
("num_grids", self.num_grids),
("xmin", self.xmin),
("xmax", self.xmax),
("ymin", self.ymin),
("ymax", self.ymax),
("zmin", self.zmin),
("zmax", self.zmax),
("rmin", self.rmin),
("rmax", self.rmax),
("tmin", self.tmin),
("tmax", self.tmax),
("visibility", self.visibility),
)
print("Dataset:")
for key, value in items:
print(f" {key}: {value}")
[docs]
def server_config(self, server_config: str | constant.ServerConfig) -> None:
"""Select a server configuration.
Args:
server_config: ``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a configuration name string (without the ``.srv`` extension)
under the sconfig folder.
"""
if isinstance(server_config, constant.ServerConfig):
server_config = server_config.value
if isinstance(server_config, str):
lowered = server_config.lower()
if lowered == "manual":
raise InvalidArgumentError("server_config 'manual' is not supported.")
if lowered in {
"local",
"local_parallel",
"localparallel",
"local-parallel",
}:
server_config = "local_parallel" if lowered != "local" else "local"
_core_call(self._core.server_config, server_config, None)
[docs]
def set_transient(
self, *, time_step: int | None = None, solution_time: float | None = None
) -> None:
"""Set the active transient time step or solution time.
Args:
time_step: Discrete transient time-step number to activate.
solution_time: Continuous solution-time value to activate.
Example:
.. code-block:: python
import os
import fieldview as fv
data_dir = os.path.join(fv.home, "examples", "rectangular_duct")
ds = fv.data.load_fvuns(os.path.join(data_dir, "rect_duct_010.uns"), transient=True)
ds.set_transient(time_step=25)
"""
if time_step is None and solution_time is None:
raise InvalidArgumentError(
"set_transient requires time_step or solution_time."
)
_core_call(self._core.set_transient, time_step, solution_time)
[docs]
def transient_info(self) -> TransientInfo:
"""Return transient dataset information.
Example:
.. code-block:: python
import os
import fieldview as fv
data_dir = os.path.join(fv.home, "examples", "rectangular_duct")
ds = fv.data.load_fvuns(os.path.join(data_dir, "rect_duct_010.uns"), transient=True)
info = ds.transient_info()
print(info.time_step, info.total_time_steps)
print(info.time_step_values[:3])
"""
self._ensure_valid()
payload = _core_call(self._core.transient_info)
return _parse_transient_info(payload)
[docs]
def sweep_time(
self,
*,
from_time_step: int | None = None,
to_time_step: int | None = None,
from_time_step_index: int | None = None,
to_time_step_index: int | None = None,
from_solution_time: float | None = None,
to_solution_time: float | None = None,
from_solution_time_index: int | None = None,
to_solution_time_index: int | None = None,
loop: bool = False,
skip: int = 0,
cycles: int = 1,
delta_time: float | None = None,
streaklines_filename: str | os.PathLike[str] | None = None,
extracts_database_name: str | os.PathLike[str] | None = None,
export_surfaces: object = None,
) -> None:
"""Run a transient sweep for this dataset.
By default, this sweeps all available time steps from the first to the
last step. ``delta_time=None`` restores the dataset's original
solution times (disables any active delta-time override). Index-based
modes resolve endpoints through :meth:`transient_info` before
dispatching the sweep to the core.
``export_surfaces`` accepts tuple entries like
``(coord, "coord_sweep", "csv")`` or ``(coord, "coord_sweep")``.
Args:
from_time_step: Starting transient step value. Use together with
``to_time_step``.
to_time_step: Ending transient step value. Use together with
``from_time_step``.
from_time_step_index: Starting transient step index. ``0`` means
the first step; negative values count from the end (``-1`` is
the last step). Indices are resolved against
:attr:`TransientInfo.time_step_values`.
to_time_step_index: Ending transient step index. ``-1`` means the
last step; negative values count from the end.
from_solution_time: Starting solution time value. Use together
with ``to_solution_time``.
to_solution_time: Ending solution time value. Use together with
``from_solution_time``.
from_solution_time_index: Starting solution time index. Indices
are resolved against :attr:`TransientInfo.solution_time_values`.
to_solution_time_index: Ending solution time index. Negative
values count from the end.
loop: Whether the host should loop the sweep continuously.
skip: Number of intermediate steps to skip between displayed or
exported states.
cycles: Number of sweep cycles to run.
delta_time: Optional delta-time override for animated pathline or
streakline calculations. If omitted, any prior override is
cleared and original dataset solution times are used.
streaklines_filename: Optional output filename for streakline
export generated during the sweep.
extracts_database_name: Optional extracts database name to write
during the sweep.
export_surfaces: Optional iterable of surface export requests, each
expressed as ``(surface, filename)`` or
``(surface, filename, format)``.
Example:
.. code-block:: python
import os
import fieldview as fv
data_dir = os.path.join(fv.home, "examples", "rectangular_duct")
ds = fv.data.load_fvuns(os.path.join(data_dir, "rect_duct_010.uns"), transient=True)
ds.sweep_time()
ds.sweep_time(from_time_step=1, to_time_step=10)
ds.sweep_time(from_time_step_index=-10, to_time_step_index=-1) # last 10 frames
ds.sweep_time(from_solution_time=0.0, to_solution_time=0.3, cycles=1)
ds.sweep_time(from_solution_time_index=0, to_solution_time_index=-1)
"""
self._ensure_valid()
if not self.transient:
raise InvalidArgumentError("dataset is not transient")
use_time_steps = from_time_step is not None or to_time_step is not None
use_time_step_indices = (
from_time_step_index is not None or to_time_step_index is not None
)
use_solution_times = (
from_solution_time is not None or to_solution_time is not None
)
use_solution_time_indices = (
from_solution_time_index is not None or to_solution_time_index is not None
)
mode_count = sum(
(
use_time_steps,
use_time_step_indices,
use_solution_times,
use_solution_time_indices,
)
)
if mode_count == 0:
use_time_step_indices = True
from_time_step_index = 0
to_time_step_index = -1
elif mode_count > 1:
raise InvalidArgumentError(
"specify exactly one of time-step values, time-step indices, solution-time values, or solution-time indices"
)
if use_time_steps:
if from_time_step is None or to_time_step is None:
raise InvalidArgumentError(
"from_time_step and to_time_step must both be provided"
)
if isinstance(from_time_step, bool) or not isinstance(from_time_step, int):
raise InvalidArgumentError("from_time_step must be an integer")
if isinstance(to_time_step, bool) or not isinstance(to_time_step, int):
raise InvalidArgumentError("to_time_step must be an integer")
if from_time_step > to_time_step:
raise InvalidArgumentError("from_time_step must be <= to_time_step")
elif use_time_step_indices:
if from_time_step_index is None or to_time_step_index is None:
raise InvalidArgumentError(
"from_time_step_index and to_time_step_index must both be provided"
)
if isinstance(from_time_step_index, bool) or not isinstance(
from_time_step_index, int
):
raise InvalidArgumentError("from_time_step_index must be an integer")
if isinstance(to_time_step_index, bool) or not isinstance(
to_time_step_index, int
):
raise InvalidArgumentError("to_time_step_index must be an integer")
elif use_solution_times:
if from_solution_time is None or to_solution_time is None:
raise InvalidArgumentError(
"from_solution_time and to_solution_time must both be provided"
)
if isinstance(from_solution_time, bool) or not isinstance(
from_solution_time, (int, float)
):
raise InvalidArgumentError("from_solution_time must be numeric")
if isinstance(to_solution_time, bool) or not isinstance(
to_solution_time, (int, float)
):
raise InvalidArgumentError("to_solution_time must be numeric")
if from_solution_time > to_solution_time:
raise InvalidArgumentError(
"from_solution_time must be <= to_solution_time"
)
else:
if from_solution_time_index is None or to_solution_time_index is None:
raise InvalidArgumentError(
"from_solution_time_index and to_solution_time_index must both be provided"
)
if isinstance(from_solution_time_index, bool) or not isinstance(
from_solution_time_index, int
):
raise InvalidArgumentError(
"from_solution_time_index must be an integer"
)
if isinstance(to_solution_time_index, bool) or not isinstance(
to_solution_time_index, int
):
raise InvalidArgumentError("to_solution_time_index must be an integer")
if isinstance(skip, bool) or not isinstance(skip, int):
raise InvalidArgumentError("skip must be an integer")
if skip < 0:
raise InvalidArgumentError("skip must be >= 0")
if isinstance(cycles, bool) or not isinstance(cycles, int):
raise InvalidArgumentError("cycles must be an integer")
if cycles <= 0:
raise InvalidArgumentError("cycles must be > 0")
loop_periods = _normalize_sweep_loop(loop)
time_step_value_range: tuple[int, int] | None = None
time_step_index_range: tuple[int, int] | None = None
solution_time_value_range: tuple[float, float] | None = None
solution_time_index_range: tuple[int, int] | None = None
if use_time_steps:
time_step_value_range = (
cast(int, from_time_step),
cast(int, to_time_step),
)
elif use_time_step_indices:
time_step_index_range = (
cast(int, from_time_step_index),
cast(int, to_time_step_index),
)
elif use_solution_times:
solution_time_value_range = (
float(cast(float | int, from_solution_time)),
float(cast(float | int, to_solution_time)),
)
else:
solution_time_index_range = (
cast(int, from_solution_time_index),
cast(int, to_solution_time_index),
)
payload: dict[str, object] = {}
if time_step_value_range is not None:
payload["from_time_step"], payload["to_time_step"] = time_step_value_range
elif time_step_index_range is not None:
# Index modes must query transient_info() so they can resolve to
# concrete core-facing transient step values before dispatch.
info = self.transient_info()
from_time_step_index_value, to_time_step_index_value = time_step_index_range
num_time_steps = len(info.time_step_values)
from_index = _normalize_sweep_index(
from_time_step_index_value,
label="from_time_step_index",
total_steps=num_time_steps,
collection_name="time steps",
)
to_index = _normalize_sweep_index(
to_time_step_index_value,
label="to_time_step_index",
total_steps=num_time_steps,
collection_name="time steps",
)
if from_index > to_index:
raise InvalidArgumentError(
"from_time_step_index must resolve to a step at or before to_time_step_index"
)
payload["from_time_step"] = info.time_step_values[from_index]
payload["to_time_step"] = info.time_step_values[to_index]
elif solution_time_value_range is not None:
(
payload["from_solution_time"],
payload["to_solution_time"],
) = solution_time_value_range
else:
info = self.transient_info()
if not info.has_solution_times:
raise InvalidArgumentError("dataset has no solution times")
(
from_solution_time_index_value,
to_solution_time_index_value,
) = cast(tuple[int, int], solution_time_index_range)
num_solution_times = len(info.solution_time_values)
from_index = _normalize_sweep_index(
from_solution_time_index_value,
label="from_solution_time_index",
total_steps=num_solution_times,
collection_name="solution times",
)
to_index = _normalize_sweep_index(
to_solution_time_index_value,
label="to_solution_time_index",
total_steps=num_solution_times,
collection_name="solution times",
)
if from_index > to_index:
raise InvalidArgumentError(
"from_solution_time_index must resolve to a time at or before to_solution_time_index"
)
payload["from_solution_time"] = info.solution_time_values[from_index]
payload["to_solution_time"] = info.solution_time_values[to_index]
payload["loop"] = loop_periods
payload["skip"] = int(skip)
payload["cycles"] = int(cycles)
if delta_time is not None:
if isinstance(delta_time, bool) or not isinstance(delta_time, (int, float)):
raise InvalidArgumentError("delta_time must be numeric")
payload["delta_time"] = float(delta_time)
if streaklines_filename is not None:
streaklines_filename = _coerce_pathlike_str(
streaklines_filename, "streaklines_filename"
)
if not streaklines_filename.strip():
raise InvalidArgumentError(
"streaklines_filename must be a non-empty string"
)
payload["streaklines_filename"] = streaklines_filename.strip()
if extracts_database_name is not None:
extracts_database_name = _coerce_pathlike_str(
extracts_database_name, "extracts_database_name"
)
if not extracts_database_name.strip():
raise InvalidArgumentError(
"extracts_database_name must be a non-empty string"
)
payload["extracts_database_name"] = extracts_database_name.strip()
export_target_count = sum(
value is not None
for value in (streaklines_filename, extracts_database_name, export_surfaces)
)
if export_target_count > 1:
raise InvalidArgumentError(
"Only one of streaklines_filename, extracts_database_name, or export_surfaces may be specified"
)
normalized_exports = _normalize_sweep_export_surfaces(self, export_surfaces)
if normalized_exports is not None:
payload["export_surfaces"] = normalized_exports
_core_call(self._core.sweep_time, payload)
@property
def duplication(self) -> DuplicationController:
"""Access duplication controls (mirror/rotate/translate)."""
return self._duplication
@property
def transform(self) -> TransformController:
"""Access transform controls (scale/translate/rotate)."""
return self._transform
[docs]
def duplication_clear(self) -> None:
"""Clear duplication state for this dataset."""
_core_call(self._core.duplicate, {"operation": "none"})
[docs]
def copy(self) -> "Dataset":
"""Return an independent dataset copy loaded in append mode."""
self._ensure_valid()
recipe = self._load_recipe
if recipe is None:
raise InvalidArgumentError(
"copy() requires a dataset loaded with fieldview.data load_* helpers."
)
return _load_dataset_from_recipe(recipe, mode=constant.InputMode.APPEND.value)
def _load_dataset_from_recipe(recipe: dict[str, object], *, mode: str) -> Dataset:
kind = str(recipe.get("kind", ""))
server = str(recipe.get("server", ""))
options = dict(cast(dict[str, object], recipe.get("options", {})))
core = _CoreDataset()
if server:
core.server = server
core.input_mode = mode
if kind == "plot3d":
_core_call(
core.load_plot3d,
recipe.get("grid_file", ""),
recipe.get("q_file", ""),
recipe.get("function_file", ""),
recipe.get("function_name_file", ""),
mode,
options,
)
elif kind == "generic":
file = str(recipe.get("file", ""))
results_file = str(recipe.get("results_file", ""))
if results_file:
filename = ""
grid_file = file
results = results_file
else:
filename = file
grid_file = ""
results = ""
_core_call(
cast(Callable[..., object], getattr(core, "load_generic")),
str(recipe.get("reader_name", "")),
filename,
grid_file,
results,
mode,
options,
None,
None,
)
else:
raise InvalidArgumentError("copy() is not supported for this dataset type.")
clone = _reset_after_load(Dataset(core))
clone._load_recipe = dict(recipe)
return clone
def _apply_if_supported(reader: object, attr: str, value: object) -> None:
if value is None:
return
if hasattr(reader, attr):
setattr(reader, attr, value)
def _apply_common_reader_options(
reader: object,
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
grid_processing: str | constant.GridProcessing | None = None,
) -> None:
if input_mode is not None:
setattr(reader, "input_mode", input_mode)
if grid_processing is not None:
setattr(reader, "grid_processing", grid_processing)
# Match FVX data input semantics: omitted server_config means direct/local-serial.
cast(_ReaderWithServerConfig, reader).server_config(
"" if server_config is None else server_config
)
def _apply_transient_reader_options(
reader: object,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
) -> None:
_apply_if_supported(reader, "transient", transient)
_apply_if_supported(reader, "read_as_steady_state", read_as_steady_state)
_apply_if_supported(
reader, "changing_number_of_grids_over_time", changing_number_of_grids_over_time
)
_apply_if_supported(reader, "initial_time_index", initial_time_index)
_apply_if_supported(reader, "initial_time_step", initial_time_step)
_apply_if_supported(reader, "initial_solution_time", initial_solution_time)
def _apply_generic_reader_options(
reader: object,
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
extended_variables: bool | None = None,
duplicate_boundaries: bool | None = None,
) -> None:
_apply_common_reader_options(reader, input_mode, server_config, grid_processing)
_apply_transient_reader_options(
reader,
transient,
read_as_steady_state,
changing_number_of_grids_over_time,
initial_time_index,
initial_time_step,
initial_solution_time,
)
_apply_if_supported(reader, "boundary_only", boundary_only)
_apply_if_supported(reader, "extended_variables", extended_variables)
_apply_if_supported(reader, "duplicate_boundaries", duplicate_boundaries)
def _load_generic_reader(
factory: str | Callable[..., object],
file: _PathLikeStr = "",
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
extended_variables: bool | None = None,
duplicate_boundaries: bool | None = None,
) -> Dataset:
if isinstance(factory, str):
factory = _reader_factory(factory)
reader = factory()
_apply_generic_reader_options(
reader,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
extended_variables=extended_variables,
duplicate_boundaries=duplicate_boundaries,
)
return _reader_load(reader, file=file, results_file=results_file)
def load_generic(
reader_name: str,
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
extended_variables: bool | None = None,
duplicate_boundaries: bool | None = None,
) -> Dataset:
"""Load a generic reader and return a Dataset.
Args:
reader_name: Reader alias (e.g. "fvuns", "ensight").
file: "Grid Data", "Grids and Results Data", or "Grid or Combined Data".
results_file: "Results Data" file for two-file formats.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings when supported by the reader.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load, when supported.
grid_processing: Optional grid-processing mode override.
boundary_only: When supported, restrict loading to boundary/surface
data.
extended_variables: When supported, request extended-variable import.
duplicate_boundaries: When supported, preserve duplicate boundary
information during import.
"""
reader = _reader_factory("_generic")(reader_name=reader_name)
_apply_generic_reader_options(
reader,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
extended_variables=extended_variables,
duplicate_boundaries=duplicate_boundaries,
)
return _reader_load(reader, file=file, results_file=results_file)
[docs]
def load_plot3d(
grid_file: _PathLikeStr,
q_file: _PathLikeStr = "",
function_file: _PathLikeStr = "",
function_name_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
merged_series: bool | None = None,
auto_detect_format: bool | None = None,
file_format: str | constant.Plot3dFileFormat | None = None,
auto_partition: bool | None = None,
ghost_cells: int | None = None,
iblanks: bool | None = None,
multi_grid: bool | None = None,
coords: str | constant.Plot3dCoords | None = None,
) -> Dataset:
"""Load Plot3D data and return a Dataset.
Args:
grid_file: Plot3D grid file path.
q_file: Optional Plot3D solution/Q file path.
function_file: Optional Plot3D function file path.
function_name_file: Optional Plot3D function-name file path.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
merged_series: When ``True``, treat matching Plot3D files as a merged
time series.
auto_detect_format, file_format: Optional Plot3D format detection or
explicit format override.
auto_partition: When ``True``, enable reader-side auto partitioning.
ghost_cells: Number of ghost-cell layers to interpret.
iblanks: When ``True``, enable iblank handling.
multi_grid: When ``True``, treat the input as a multigrid dataset.
coords: Optional Plot3D coordinate interpretation override.
Example usage:
.. code-block:: python
>>> import fieldview as fv
>>> grid_file = "/path/to/grid.bin"
>>> q_file = "/path/to/q.bin"
>>> ds = fv.data.load_plot3d(grid_file, q_file)
"""
reader = _reader_factory("_plot3d")()
_apply_common_reader_options(reader, input_mode, server_config, grid_processing)
_apply_transient_reader_options(
reader,
transient,
read_as_steady_state,
changing_number_of_grids_over_time,
)
_apply_if_supported(reader, "merged_series", merged_series)
_apply_if_supported(reader, "auto_detect_format", auto_detect_format)
_apply_if_supported(reader, "file_format", file_format)
_apply_if_supported(reader, "auto_partition", auto_partition)
_apply_if_supported(reader, "ghost_cells", ghost_cells)
_apply_if_supported(reader, "iblanks", iblanks)
_apply_if_supported(reader, "multi_grid", multi_grid)
_apply_if_supported(reader, "coords", coords)
return _reader_load(
reader,
grid_file=grid_file,
q_file=q_file,
function_file=function_file,
function_name_file=function_name_file,
)
[docs]
def load_overflow2(
grid_file: _PathLikeStr,
q_file: _PathLikeStr = "",
function_file: _PathLikeStr = "",
function_name_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
merged_series: bool | None = None,
auto_detect_format: bool | None = None,
file_format: str | constant.Plot3dFileFormat | None = None,
auto_partition: bool | None = None,
ghost_cells: int | None = None,
iblanks: bool | None = None,
multi_grid: bool | None = None,
coords: str | constant.Plot3dCoords | None = None,
) -> Dataset:
"""Load Overflow-2 (Plot3D-compatible) data and return a Dataset.
Args:
grid_file: Plot3D-compatible grid file path.
q_file: Optional solution/Q file path.
function_file: Optional function file path.
function_name_file: Optional function-name file path.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
merged_series: When ``True``, treat matching files as a merged time
series.
auto_detect_format, file_format: Optional format detection or explicit
format override.
auto_partition: When ``True``, enable reader-side auto partitioning.
ghost_cells: Number of ghost-cell layers to interpret.
iblanks: When ``True``, enable iblank handling.
multi_grid: When ``True``, treat the input as a multigrid dataset.
coords: Optional coordinate interpretation override.
"""
reader = _reader_factory("_overflow2")()
_apply_common_reader_options(reader, input_mode, server_config, grid_processing)
_apply_transient_reader_options(
reader,
transient,
read_as_steady_state,
changing_number_of_grids_over_time,
)
_apply_if_supported(reader, "merged_series", merged_series)
_apply_if_supported(reader, "auto_detect_format", auto_detect_format)
_apply_if_supported(reader, "file_format", file_format)
_apply_if_supported(reader, "auto_partition", auto_partition)
_apply_if_supported(reader, "ghost_cells", ghost_cells)
_apply_if_supported(reader, "iblanks", iblanks)
_apply_if_supported(reader, "multi_grid", multi_grid)
_apply_if_supported(reader, "coords", coords)
return _reader_load(
reader,
grid_file=grid_file,
q_file=q_file,
function_file=function_file,
function_name_file=function_name_file,
)
[docs]
def load_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load FV-UNS data and return a Dataset.
Args:
file: Primary FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
Example usage:
.. code-block:: python
>>> import fieldview as fv
>>> file = "/path/to/model.fvuns"
>>> results_file = "/path/to/results.fvuns"
>>> ds = fv.data.load_fvuns(file=file, results_file=results_file)
"""
return _load_generic_reader(
"_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_vtk_structured(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
) -> Dataset:
"""Load VTK structured data and return a Dataset.
Args:
file: Primary VTK structured file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
"""
return _load_generic_reader(
"_vtk_structured",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
)
[docs]
def load_acusolve_direct(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
extended_variables: bool | None = None,
duplicate_boundaries: bool | None = None,
) -> Dataset:
"""Load AcuSolve direct data and return a Dataset.
Args:
file: Primary AcuSolve input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
extended_variables: When ``True``, request extended-variable import.
duplicate_boundaries: When ``True``, preserve duplicate boundary
information during import.
"""
return _load_generic_reader(
"_acusolve_direct",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
grid_processing=grid_processing,
boundary_only=boundary_only,
extended_variables=extended_variables,
duplicate_boundaries=duplicate_boundaries,
)
[docs]
def load_acusolve_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load AcuSolve FV-UNS data and return a Dataset.
Args:
file: Primary AcuSolve FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_acusolve_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_ansys_cfx_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load ANSYS-CFX FV-UNS data and return a Dataset.
Args:
file: Primary ANSYS-CFX FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_ansys_cfx_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_fluent_cff_direct(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
) -> Dataset:
"""Load Fluent CFF direct data and return a Dataset.
Args:
file: Primary Fluent CFF input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
"""
return _load_generic_reader(
"_fluent_cff_direct",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
)
[docs]
def load_append_sampled_data(file: _PathLikeStr) -> Dataset:
"""Load Append Sampled Data and return a Dataset.
Args:
file: Input file path for the sampled-data dataset.
"""
return _load_generic_reader(
"_append_sampled_data",
file=file,
)
[docs]
def load_pw_common(file: _PathLikeStr) -> Dataset:
"""Load pw common data and return a Dataset.
Args:
file: Input file path for the pw common dataset.
"""
return _load_generic_reader(
"_pw_common",
file=file,
)
[docs]
def load_cfdpp_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load CFD++ FV-UNS data and return a Dataset.
Args:
file: Primary CFD++ FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_cfdpp_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_cgns_structured(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
grid_processing: str | constant.GridProcessing | None = None,
) -> Dataset:
"""Load CGNS structured data and return a Dataset.
Args:
file: Primary CGNS structured file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
grid_processing: Optional grid-processing mode override.
"""
return _load_generic_reader(
"_cgns_structured",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
grid_processing=grid_processing,
)
[docs]
def load_cgns_unstructured(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load CGNS unstructured data and return a Dataset.
Args:
file: Primary CGNS unstructured file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_cgns_unstructured",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_cgns_unstructured_hybrid(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load CGNS unstructured hybrid data and return a Dataset.
Args:
file: Primary CGNS hybrid file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_cgns_unstructured_hybrid",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_cobalt_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load COBALT FV-UNS data and return a Dataset.
Args:
file: Primary COBALT FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_cobalt_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_converge_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load CONVERGE FV-UNS data and return a Dataset.
Args:
file: Primary CONVERGE FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_converge_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_ensight(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load Ensight data and return a Dataset.
Args:
file: Primary Ensight input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_ensight",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_flow3d_animation(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
) -> Dataset:
"""Load FLOW-3D animation data and return a Dataset.
Args:
file: Primary FLOW-3D animation file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
"""
return _load_generic_reader(
"_flow3d_animation",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
)
[docs]
def load_flow3d_restart(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
) -> Dataset:
"""Load FLOW-3D restart data and return a Dataset.
Args:
file: Primary FLOW-3D restart file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
"""
return _load_generic_reader(
"_flow3d_restart",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
)
[docs]
def load_fluent_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load FLUENT FV-UNS data and return a Dataset.
Args:
file: Primary FLUENT FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_fluent_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_fluent_unstructured(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
) -> Dataset:
"""Load Fluent unstructured data and return a Dataset.
Args:
file: Primary Fluent unstructured file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
"""
return _load_generic_reader(
"_fluent_unstructured",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
)
[docs]
def load_fluent_cas_dat_direct(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load FLUENT cas/dat direct data and return a Dataset.
Args:
file: Primary FLUENT cas/dat input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_fluent_cas_dat_direct",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_fluent_direct(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
) -> Dataset:
"""Load Fluent direct data and return a Dataset.
Args:
file: Primary Fluent direct input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
"""
return _load_generic_reader(
"_fluent_direct",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
)
[docs]
def load_havoc(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load HAVOC data and return a Dataset.
Args:
file: Primary HAVOC input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_havoc",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_lsdyna_d3plot(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load LS-DYNA d3plot data and return a Dataset.
Args:
file: Primary LS-DYNA d3plot file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_lsdyna_d3plot",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_lsdyna_state(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load LS-DYNA state data and return a Dataset.
Args:
file: Primary LS-DYNA state file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_lsdyna_state",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_nparc_wind(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
) -> Dataset:
"""Load NPARC/WIND data and return a Dataset.
Args:
file: Primary NPARC/WIND input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
"""
return _load_generic_reader(
"_nparc_wind",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
)
[docs]
def load_openfoam_direct(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load OpenFOAM direct data and return a Dataset.
Args:
file: Primary OpenFOAM direct input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_openfoam_direct",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_openfoam_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load OpenFOAM FV-UNS data and return a Dataset.
Args:
file: Primary OpenFOAM FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_openfoam_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_sc_tetra(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load SC/Tetra data and return a Dataset.
Args:
file: Primary SC/Tetra input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_sc_tetra",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_sc_flow(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load SCFLOW data and return a Dataset.
Args:
file: Primary SCFLOW input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_sc_flow",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_scryu(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load SCRYU data and return a Dataset.
Args:
file: Primary SCRYU input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_scryu",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_sc_stream(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load SCSTREAM data and return a Dataset.
Args:
file: Primary SCSTREAM input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_sc_stream",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_starccm_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load STAR-CCM+ FV-UNS data and return a Dataset.
Args:
file: Primary STAR-CCM+ FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_starccm_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_starcd_fvuns(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load STAR-CD FV-UNS data and return a Dataset.
Args:
file: Primary STAR-CD FV-UNS file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_starcd_fvuns",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_stl(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
) -> Dataset:
"""Load STL data and return a Dataset.
Args:
file: Primary STL input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
"""
return _load_generic_reader(
"_stl",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
)
[docs]
def load_surface_sampled_data(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load surface sampled data and return a Dataset.
Args:
file: Primary surface-sampled-data input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_surface_sampled_data",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_tecplot_360(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load Tecplot 360 data and return a Dataset.
Args:
file: Primary Tecplot 360 input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_tecplot_360",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_ultrafluidx_direct(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load ultraFluidX data and return a Dataset.
Args:
file: Primary ultraFluidX input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_ultrafluidx_direct",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_vtk_unstructured_hybrid(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load VTK unstructured/hybrid data and return a Dataset.
Args:
file: Primary VTK unstructured or hybrid input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_vtk_unstructured_hybrid",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_wind_structured(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
) -> Dataset:
"""Load WIND structured data and return a Dataset.
Args:
file: Primary WIND structured input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
"""
return _load_generic_reader(
"_wind_structured",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
)
[docs]
def load_wind_unstructured(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
boundary_only: bool | None = None,
) -> Dataset:
"""Load WIND unstructured data and return a Dataset.
Args:
file: Primary WIND unstructured input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
boundary_only: When ``True``, load only boundary/surface data.
"""
return _load_generic_reader(
"_wind_unstructured",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
grid_processing=grid_processing,
boundary_only=boundary_only,
)
[docs]
def load_xdb_import(
file: _PathLikeStr,
results_file: _PathLikeStr = "",
input_mode: str | constant.InputMode | None = None,
server_config: str | constant.ServerConfig | None = None,
transient: bool | None = None,
read_as_steady_state: bool | None = None,
changing_number_of_grids_over_time: bool | None = None,
grid_processing: str | constant.GridProcessing | None = None,
initial_time_index: int | None = None,
initial_time_step: int | None = None,
initial_solution_time: float | None = None,
) -> Dataset:
"""Load XDB import data and return a Dataset.
Args:
file: Primary XDB import input file path.
results_file: Optional companion results file.
input_mode: Optional dataset load mode. Use ``constant.InputMode.REPLACE``
or ``constant.InputMode.APPEND``.
server_config: Optional server configuration. Use
``constant.ServerConfig.LOCAL``, ``constant.ServerConfig.LOCAL_PARALLEL``,
or a named ``.srv`` configuration.
transient, read_as_steady_state, changing_number_of_grids_over_time:
Optional transient-reader settings.
grid_processing: Optional grid-processing mode override.
initial_time_index, initial_time_step, initial_solution_time:
Optional initial transient selection after load.
"""
return _load_generic_reader(
"_xdb_import",
file=file,
results_file=results_file,
input_mode=input_mode,
server_config=server_config,
transient=transient,
read_as_steady_state=read_as_steady_state,
changing_number_of_grids_over_time=changing_number_of_grids_over_time,
initial_time_index=initial_time_index,
initial_time_step=initial_time_step,
initial_solution_time=initial_solution_time,
grid_processing=grid_processing,
)
def _reset_after_load(dataset: Dataset) -> Dataset:
global _current_dataset
dataset.duplication_clear()
dataset.transform.clear()
_refresh_function_lists(dataset)
_dataset_registry[dataset.dataset_id] = dataset
_current_dataset = dataset
return dataset
def _get_or_attach_dataset(dataset_id: int) -> Dataset:
"""Return a live dataset wrapper for an existing host dataset id.
Replace-style host reloads invalidate previously attached wrappers for the
same dataset slot. Append-style reads leave existing dataset wrappers
valid and allocate new dataset ids for the appended data.
"""
dataset_id = int(dataset_id)
dataset_ref = _dataset_registry.get(dataset_id)
if dataset_ref is not None:
try:
dataset_ref._core._ensure_valid()
except RuntimeError as exc:
if _is_invalid_dataset_message(str(exc)):
_dataset_registry.pop(dataset_id, None)
else:
msg = str(exc).strip()
raise CoreError(
msg or "Unexpected FieldView error while validating cached dataset."
) from exc
else:
return dataset_ref
attach_existing = getattr(_core_module, "dataset_attach_existing", None)
if not callable(attach_existing):
raise InvalidDatasetError(
"FieldView host does not provide live dataset attach support. "
+ "Load it via fieldview.data.load_*."
)
try:
core_dataset = _core_call(attach_existing, dataset_id)
except CoreError as exc:
raise InvalidDatasetError(str(exc)) from exc
dataset_ref = Dataset(core=cast(_CoreDataset, core_dataset))
_refresh_function_lists(dataset_ref)
_dataset_registry[dataset_ref.dataset_id] = dataset_ref
return dataset_ref
def _refresh_function_lists(dataset: Dataset) -> None:
scalar_payload = _load_json_dict(
_core_call(dataset._core._list_functions_json, "scalar"),
"scalar function list",
)
vector_payload = _load_json_dict(
_core_call(dataset._core._list_functions_json, "vector"),
"vector function list",
)
dataset._scalar_functions, dataset._scalar_function_ids = _parse_function_list(
scalar_payload
)
dataset._vector_functions, dataset._vector_function_ids = _parse_function_list(
vector_payload
)
dataset._boundary_types = []
list_boundary_types = getattr(dataset._core, "_list_boundary_types_json", None)
if list_boundary_types is None:
return
boundary_payload = _load_json_dict(
_core_call(cast(Callable[..., object], list_boundary_types)),
"boundary type list",
)
dataset._boundary_types = _parse_boundary_type_list(boundary_payload)
def _is_invalid_dataset_message(message: str) -> bool:
lower_msg = message.lower()
return "dataset is invalid" in lower_msg or "replaced by another load" in lower_msg
def _refresh_current_dataset_function_lists() -> None:
"""Refresh cached function lists for the current dataset wrapper when available."""
try:
dataset = get_current()
except InvalidDatasetError:
return
_refresh_function_lists(dataset)
def _parse_function_list(
payload: dict[str, object],
) -> tuple[list[str], dict[str, int]]:
names: list[str] = []
mapping: dict[str, int] = {}
functions = payload.get("functions", [])
if not isinstance(functions, list):
return names, mapping
for entry in cast(list[object], functions):
if not isinstance(entry, dict):
continue
entry_map = cast(dict[str, object], entry)
name = entry_map.get("name")
func_id = entry_map.get("id")
if not isinstance(name, str) or not name:
continue
try:
func_id_int = _to_int(func_id)
except (TypeError, ValueError):
continue
names.append(name)
mapping.setdefault(name, func_id_int)
return names, mapping
def _parse_boundary_type_list(payload: dict[str, object]) -> list[str]:
candidates = payload.get("boundary_types")
if candidates is None:
candidates = payload.get("types")
names: list[str] = []
if not isinstance(candidates, list):
return names
for entry in cast(list[object], candidates):
if isinstance(entry, str):
name = entry.strip()
elif isinstance(entry, dict):
name = str(cast(dict[str, object], entry).get("name", "")).strip()
else:
continue
if name:
names.append(name)
return names
def _resolve_dataset_or_current(dataset: Dataset | None) -> Dataset:
if dataset is None:
dataset = get_current()
if not isinstance(dataset, Dataset):
raise InvalidArgumentError("dataset must be a Dataset instance")
if dataset.dataset_id < 0:
raise InvalidDatasetError("dataset is invalid (not loaded)")
dataset._ensure_valid()
return dataset
def _validate_ijk_point(value: object) -> tuple[int, int, int]:
if not isinstance(value, (list, tuple)):
raise InvalidArgumentError("ijk_point must be a 3-integer sequence")
components = tuple(cast(tuple[object, ...], value))
if len(components) != 3:
raise InvalidArgumentError("ijk_point must be a 3-integer sequence")
result: list[int] = []
for index, component in enumerate(components, start=1):
if isinstance(component, bool) or not isinstance(component, int):
raise InvalidArgumentError(f"ijk_point[{index}] must be an integer")
result.append(int(component))
return result[0], result[1], result[2]
def _validate_grid_index(value: object) -> int:
if isinstance(value, bool) or not isinstance(value, int):
raise InvalidArgumentError("grid_index must be an integer")
if value < 0:
raise InvalidArgumentError("grid_index must be >= 0")
return int(value)
def _validate_grid(value: object) -> int:
if isinstance(value, bool) or not isinstance(value, int):
raise InvalidArgumentError("grid must be an integer")
if value < 1:
raise InvalidArgumentError("grid must be >= 1")
return int(value)
def _lookup_probe_function_id(
dataset: Dataset, value: object, attr: str, label: str
) -> int:
if not isinstance(value, str):
raise InvalidArgumentError(f"{label}_func must be a string")
needle = value.strip()
if not needle:
raise InvalidArgumentError(f"{label}_func cannot be empty")
mapping = getattr(dataset, attr, None)
if not isinstance(mapping, dict):
raise InvalidArgumentError(
f"{label}_func could not be resolved for the dataset"
)
mapping = cast(dict[str, int], mapping)
needle_lower = needle.lower()
for func_name, func_id in mapping.items():
if func_name.lower() == needle_lower:
return func_id
raise InvalidArgumentError(f"{label}_func not found: {needle}")
def _resolve_probe_function_overrides(
dataset: Dataset,
*,
scalar_func: object = None,
vector_func: object = None,
iso_func: object = None,
threshold_func: object = None,
) -> tuple[int | None, int | None, int | None, int | None]:
scalar_func_id = None
vector_func_id = None
iso_func_id = None
threshold_func_id = None
if scalar_func is not None:
scalar_func_id = _lookup_probe_function_id(
dataset, scalar_func, "_scalar_function_ids", "scalar"
)
if vector_func is not None:
vector_func_id = _lookup_probe_function_id(
dataset, vector_func, "_vector_function_ids", "vector"
)
if iso_func is not None:
# FieldView's geometric/iso register uses dataset scalar-function IDs.
iso_func_id = _lookup_probe_function_id(
dataset, iso_func, "_scalar_function_ids", "iso"
)
if threshold_func is not None:
# Threshold selection also comes from the dataset scalar-function registry.
threshold_func_id = _lookup_probe_function_id(
dataset, threshold_func, "_scalar_function_ids", "threshold"
)
return scalar_func_id, vector_func_id, iso_func_id, threshold_func_id
def _validate_tolerance(value: object) -> float:
if value is None:
return 1.0e-5
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise InvalidArgumentError("tolerance must be a number")
return float(value)
def _get_surface_target(surface: object) -> tuple[str, int]:
prefix = getattr(surface, "_core_prefix", "")
phigs_obj = getattr(surface, "_phigs_obj", None)
if getattr(surface, "_is_deleted", False):
raise InvalidArgumentError("surface was deleted")
if not isinstance(prefix, str) or not isinstance(phigs_obj, int):
raise InvalidArgumentError("surface must be a FieldView surface object")
return prefix, int(phigs_obj)
def _as_vec3(value: object, label: str) -> Vec3:
if not isinstance(value, (list, tuple)):
raise CoreError(f"Probe response field '{label}' must be a 3-item sequence")
components = tuple(cast(tuple[object, ...], value))
if len(components) != 3:
raise CoreError(f"Probe response field '{label}' must be a 3-item sequence")
try:
return Vec3(
_to_float(components[0]),
_to_float(components[1]),
_to_float(components[2]),
)
except (TypeError, ValueError) as exc:
raise CoreError(
f"Probe response field '{label}' must contain numeric values"
) from exc
def _as_optional_scalar_probe(value: object, label: str) -> ScalarProbe:
if not isinstance(value, dict):
raise CoreError(f"Probe response field '{label}' must be an object")
value_map = cast(dict[str, object], value)
func = value_map.get("func")
if func is not None and not isinstance(func, str):
raise CoreError(f"Probe response field '{label}.func' must be a string or null")
raw_value = value_map.get("value")
if raw_value is None:
probe_value = None
elif isinstance(raw_value, bool) or not isinstance(raw_value, (int, float)):
raise CoreError(f"Probe response field '{label}.value' must be numeric or null")
else:
probe_value = float(raw_value)
return ScalarProbe(func, probe_value)
def _as_optional_vector_probe(value: object, label: str) -> VectorProbe:
if not isinstance(value, dict):
raise CoreError(f"Probe response field '{label}' must be an object")
value_map = cast(dict[str, object], value)
func = value_map.get("func")
if func is not None and not isinstance(func, str):
raise CoreError(f"Probe response field '{label}.func' must be a string or null")
raw_value = value_map.get("value")
vector_value = None if raw_value is None else _as_vec3(raw_value, f"{label}.value")
return VectorProbe(func, vector_value)
def _parse_probe_result(payload: dict[str, object]) -> ProbeResult:
hit = payload.get("hit")
if not isinstance(hit, bool):
raise CoreError("Probe response field 'hit' must be boolean")
raw_point = payload.get("point")
point = None if raw_point is None else _as_vec3(raw_point, "point")
if hit and point is None:
raise CoreError("Probe response field 'point' must be present when hit is true")
region = payload.get("region")
if region is not None:
if isinstance(region, bool) or not isinstance(region, int):
raise CoreError("Probe response field 'region' must be an integer or null")
region = int(region)
grid_index = payload.get("grid_index")
if grid_index is not None:
if isinstance(grid_index, bool) or not isinstance(grid_index, int):
raise CoreError(
"Probe response field 'grid_index' must be an integer or null"
)
grid_index = int(grid_index)
grid = payload.get("grid")
if grid is not None:
if isinstance(grid, bool) or not isinstance(grid, int):
raise CoreError("Probe response field 'grid' must be an integer or null")
grid = int(grid)
raw_ijk = payload.get("ijk")
ijk = None
if raw_ijk is not None:
if not isinstance(raw_ijk, (list, tuple)):
raise CoreError(
"Probe response field 'ijk' must be a 3-item sequence or null"
)
components = tuple(cast(tuple[object, ...], raw_ijk))
if len(components) != 3:
raise CoreError(
"Probe response field 'ijk' must be a 3-item sequence or null"
)
try:
ijk = (
_to_float(components[0]),
_to_float(components[1]),
_to_float(components[2]),
)
except (TypeError, ValueError) as exc:
raise CoreError(
"Probe response field 'ijk' must contain numeric values"
) from exc
return ProbeResult(
hit=bool(hit),
point=point,
region=region,
grid_index=grid_index,
grid=grid,
ijk=ijk,
scalar=_as_optional_scalar_probe(payload.get("scalar"), "scalar"),
iso=_as_optional_scalar_probe(payload.get("iso"), "iso"),
threshold=_as_optional_scalar_probe(payload.get("threshold"), "threshold"),
vector=_as_optional_vector_probe(payload.get("vector"), "vector"),
)
def _parse_integration_result(payload: dict[str, object]) -> IntegrationResult:
if not isinstance(payload, dict):
raise CoreError("Integration response must be an object")
integral_type = payload.get("integral_type")
if integral_type is not None and not isinstance(integral_type, str):
raise CoreError("Integration response field 'integral_type' must be a string")
scalar_function = payload.get("scalar_function")
if scalar_function is not None and not isinstance(scalar_function, str):
raise CoreError(
"Integration response field 'scalar_function' must be a string or null"
)
surface = payload.get("surface")
if surface is not None and not isinstance(surface, str):
raise CoreError("Integration response field 'surface' must be a string or null")
area = payload.get("area")
total = payload.get("sum")
if isinstance(area, bool) or not isinstance(area, (int, float)):
raise CoreError("Integration response field 'area' must be numeric")
if isinstance(total, bool) or not isinstance(total, (int, float)):
raise CoreError("Integration response field 'sum' must be numeric")
average_value = payload.get("average")
if average_value is None:
average = None
elif isinstance(average_value, bool) or not isinstance(average_value, (int, float)):
raise CoreError("Integration response field 'average' must be numeric or null")
else:
average = float(average_value)
has_surface_normals = payload.get("has_surface_normals", False)
if not isinstance(has_surface_normals, bool):
raise CoreError(
"Integration response field 'has_surface_normals' must be boolean"
)
vector_function = payload.get("vector_function")
if vector_function is not None and not isinstance(vector_function, str):
raise CoreError(
"Integration response field 'vector_function' must be a string or null"
)
def optional_float(key: str) -> float | None:
value = payload.get(key)
if value is None:
return None
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise CoreError(
f"Integration response field '{key}' must be numeric or null"
)
return float(value)
return IntegrationResult(
integral_type=str(integral_type or ""),
scalar_function=scalar_function,
area=float(area),
sum=float(total),
average=average,
surface=surface,
has_surface_normals=has_surface_normals,
vector_function=vector_function,
sum_nx=optional_float("sum_nx"),
sum_ny=optional_float("sum_ny"),
sum_nz=optional_float("sum_nz"),
sum_v_dot_n=optional_float("sum_v_dot_n"),
)
def _parse_transient_int_values(
payload: dict[str, object], key: str
) -> tuple[int, ...]:
raw_values = payload.get(key, ())
if raw_values is None:
return ()
if not isinstance(raw_values, (list, tuple)):
raise CoreError(f"Transient response field '{key}' must be a sequence")
values_seq = tuple(cast(tuple[object, ...], raw_values))
values: list[int] = []
for index, value in enumerate(values_seq, start=1):
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise CoreError(
f"Transient response field '{key}[{index}]' must be numeric"
)
converted = int(value)
if converted != value:
raise CoreError(
f"Transient response field '{key}[{index}]' must be an integer"
)
values.append(converted)
return tuple(values)
def _parse_transient_float_values(
payload: dict[str, object], key: str
) -> tuple[float, ...]:
raw_values = payload.get(key, ())
if raw_values is None:
return ()
if not isinstance(raw_values, (list, tuple)):
raise CoreError(f"Transient response field '{key}' must be a sequence")
values_seq = tuple(cast(tuple[object, ...], raw_values))
values: list[float] = []
for index, value in enumerate(values_seq, start=1):
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise CoreError(
f"Transient response field '{key}[{index}]' must be numeric"
)
values.append(float(value))
return tuple(values)
def _parse_transient_info(payload: dict[str, object]) -> TransientInfo:
if not isinstance(payload, dict):
raise CoreError("Transient response must be an object")
time_step = payload.get("cur_time_step")
if isinstance(time_step, bool) or not isinstance(time_step, int):
raise CoreError("Transient response field 'cur_time_step' must be an integer")
solution_time = payload.get("cur_solution_time")
if isinstance(solution_time, bool) or not isinstance(solution_time, (int, float)):
raise CoreError("Transient response field 'cur_solution_time' must be numeric")
total_time_steps = payload.get("total_time_steps")
if isinstance(total_time_steps, bool) or not isinstance(total_time_steps, int):
raise CoreError(
"Transient response field 'total_time_steps' must be an integer"
)
has_solution_times = payload.get("has_solution_times", False)
if not isinstance(has_solution_times, bool):
raise CoreError("Transient response field 'has_solution_times' must be boolean")
time_step_min = payload.get("time_step_min")
time_step_max = payload.get("time_step_max")
if isinstance(time_step_min, bool) or not isinstance(time_step_min, int):
raise CoreError("Transient response field 'time_step_min' must be an integer")
if isinstance(time_step_max, bool) or not isinstance(time_step_max, int):
raise CoreError("Transient response field 'time_step_max' must be an integer")
solution_time_min = payload.get("solution_time_min")
solution_time_max = payload.get("solution_time_max")
if isinstance(solution_time_min, bool) or not isinstance(
solution_time_min, (int, float)
):
raise CoreError("Transient response field 'solution_time_min' must be numeric")
if isinstance(solution_time_max, bool) or not isinstance(
solution_time_max, (int, float)
):
raise CoreError("Transient response field 'solution_time_max' must be numeric")
return TransientInfo(
time_step=int(time_step),
solution_time=float(solution_time),
total_time_steps=int(total_time_steps),
has_solution_times=has_solution_times,
time_step_range=Range(int(time_step_min), int(time_step_max)),
solution_time_range=Range(float(solution_time_min), float(solution_time_max)),
time_step_values=_parse_transient_int_values(payload, "time_step_values"),
solution_time_values=_parse_transient_float_values(
payload, "solution_time_values"
),
)
def _normalize_sweep_loop(value: object) -> int:
if value is False or value is None:
return 1
if value is True:
raise InvalidArgumentError(
"loop=True is ambiguous; use an integer number of periods"
)
if isinstance(value, bool) or not isinstance(value, int):
raise InvalidArgumentError("loop must be False or a positive integer")
if value <= 0:
raise InvalidArgumentError("loop must be False or a positive integer")
return int(value)
def _normalize_sweep_index(
value: int,
*,
label: str,
total_steps: int,
collection_name: str,
) -> int:
if total_steps <= 0:
raise InvalidArgumentError(f"dataset has no transient {collection_name}")
index = value
if index < 0:
index += total_steps
if index < 0 or index >= total_steps:
raise InvalidArgumentError(
f"{label} index {value} is out of range for {total_steps} transient {collection_name}"
)
return index
def _normalize_sweep_export_surfaces(
dataset: "Dataset",
export_surfaces: object,
) -> list[dict[str, object]] | None:
if export_surfaces is None:
return None
if not isinstance(export_surfaces, (list, tuple)):
raise InvalidArgumentError(
"export_surfaces must be a non-empty list of export requests"
)
export_entries = tuple(cast(tuple[object, ...], export_surfaces))
if not export_entries:
raise InvalidArgumentError(
"export_surfaces must be a non-empty list of export requests"
)
if len(export_entries) != 1:
raise InvalidArgumentError(
"export_surfaces currently supports exactly one surface entry"
)
normalized: list[dict[str, object]] = []
for index, entry in enumerate(export_entries, start=1):
if not isinstance(entry, (list, tuple)):
raise InvalidArgumentError(f"export_surfaces[{index}] must be a tuple")
entry_items = tuple(cast(tuple[object, ...], entry))
surface: object
basename: object
export_type: object
if len(entry_items) == 2:
surface, basename = entry_items
export_type = "text"
elif len(entry_items) == 3:
surface = entry_items[0]
basename = entry_items[1]
export_type = entry_items[2]
else:
raise InvalidArgumentError(
f"export_surfaces[{index}] tuple entries must be (surface, basename) or (surface, basename, type)"
)
prefix, phigs_obj = _get_surface_target(surface)
dataset_id = getattr(surface, "_dataset_id", None)
if dataset_id != dataset.dataset_id:
raise InvalidArgumentError(
f"export_surfaces[{index}].surface must belong to the target dataset"
)
basename_path = _coerce_pathlike_str(
basename, f"export_surfaces[{index}].basename"
)
if not basename_path.strip():
raise InvalidArgumentError(
f"export_surfaces[{index}].basename must be a non-empty string"
)
if not isinstance(export_type, str):
raise InvalidArgumentError(
f"export_surfaces[{index}].type must be a string"
)
export_type = export_type.strip().lower()
if export_type not in {"text", "mat-file", "csv"}:
raise InvalidArgumentError(
f"export_surfaces[{index}].type must be 'text', 'mat-file', or 'csv'"
)
normalized.append(
{
"surface_kind": prefix,
"phigs_obj": phigs_obj,
"basename": basename_path.strip(),
"type": export_type,
}
)
return normalized
[docs]
def probe(
point: object,
dataset: Dataset | None = None,
*,
scalar_func: object = None,
vector_func: object = None,
iso_func: object = None,
threshold_func: object = None,
) -> ProbeResult:
"""Probe selected functions at a displayed-space point.
Args:
point: Displayed-space XYZ probe location as a 3-number sequence.
dataset: Target dataset. When omitted, the current dataset is used.
scalar_func: Optional scalar function name to make current before probing.
When omitted, the existing current scalar is used.
vector_func: Optional vector function name to make current before probing.
When omitted, the existing current vector is used.
iso_func: Optional iso/geometric function name to make current before probing.
When omitted, the existing current iso/geometric function is used.
threshold_func: Optional threshold function name to make current before
probing. When omitted, the existing current threshold function is used.
Returns:
ProbeResult with the probed location, resolved grid metadata, and sampled
scalar/vector/iso/threshold values.
Notes:
Any provided function override updates FieldView's current function
selection before probing and leaves that selection changed after the
call returns. If an override is invalid or cannot be selected, the
previous current-function state is preserved.
Example:
.. code-block:: python
import os
import fieldview as fv
data_dir = os.path.join(fv.home, "examples", "f18")
ds = fv.data.load_plot3d(
os.path.join(data_dir, "f18i9b_g_bin"),
os.path.join(data_dir, "f18i9b_q_bin"),
)
result = fv.data.probe(
(0.5, 0.5, 0.5),
dataset=ds,
scalar_func="Pressure",
vector_func="Velocity",
)
print(result.scalar)
print(result.vector)
"""
resolved_dataset = _resolve_dataset_or_current(dataset)
x, y, z = _coerce_xyz_triplet(point, "point")
scalar_func_id, vector_func_id, iso_func_id, threshold_func_id = (
_resolve_probe_function_overrides(
resolved_dataset,
scalar_func=scalar_func,
vector_func=vector_func,
iso_func=iso_func,
threshold_func=threshold_func,
)
)
payload = _load_json_dict(
_core_call(
resolved_dataset._core._probe_json,
x,
y,
z,
scalar_func_id,
vector_func_id,
iso_func_id,
threshold_func_id,
),
"probe result",
)
return _parse_probe_result(payload)
[docs]
def probe_ijk(
ijk_point: object,
grid_index: int | None = None,
*,
grid: int | None = None,
dataset: Dataset | None = None,
scalar_func: object = None,
vector_func: object = None,
iso_func: object = None,
threshold_func: object = None,
) -> ProbeResult:
"""Probe selected functions at a structured-grid IJK point.
Args:
ijk_point: Structured-grid IJK probe location as a 3-integer sequence.
grid_index: Optional 0-based actual-grid index.
grid: Optional 1-based grid number, matching the convention used by other
public APIs. Provide either ``grid`` or ``grid_index``. If both are
provided, they must refer to the same grid.
dataset: Target dataset. When omitted, the current dataset is used.
scalar_func: Optional scalar function name to make current before probing.
When omitted, the existing current scalar is used.
vector_func: Optional vector function name to make current before probing.
When omitted, the existing current vector is used.
iso_func: Optional iso/geometric function name to make current before probing.
When omitted, the existing current iso/geometric function is used.
threshold_func: Optional threshold function name to make current before
probing. When omitted, the existing current threshold function is used.
Returns:
ProbeResult with the resolved displayed-space point, grid metadata, and
sampled scalar/vector/iso/threshold values.
Notes:
Any provided function override updates FieldView's current function
selection before probing and leaves that selection changed after the
call returns. If an override is invalid or cannot be selected, the
previous current-function state is preserved.
Example:
.. code-block:: python
import os
import fieldview as fv
data_dir = os.path.join(fv.home, "examples", "f18")
ds = fv.data.load_plot3d(
os.path.join(data_dir, "f18i9b_g_bin"),
os.path.join(data_dir, "f18i9b_q_bin"),
)
result = fv.data.probe_ijk((2, 3, 4), grid=1, dataset=ds, scalar_func="Pressure")
print(result.point)
print(result.scalar)
"""
resolved_dataset = _resolve_dataset_or_current(dataset)
i, j, k = _validate_ijk_point(ijk_point)
if grid is None and grid_index is None:
raise InvalidArgumentError("probe_ijk requires grid or grid_index")
resolved_grid_index = None
if grid is not None:
resolved_grid_index = _validate_grid(grid) - 1
if grid_index is not None:
validated_grid_index = _validate_grid_index(grid_index)
if (
resolved_grid_index is not None
and resolved_grid_index != validated_grid_index
):
raise InvalidArgumentError(
"grid and grid_index must refer to the same grid"
)
resolved_grid_index = validated_grid_index
scalar_func_id, vector_func_id, iso_func_id, threshold_func_id = (
_resolve_probe_function_overrides(
resolved_dataset,
scalar_func=scalar_func,
vector_func=vector_func,
iso_func=iso_func,
threshold_func=threshold_func,
)
)
payload = _load_json_dict(
_core_call(
resolved_dataset._core._probe_ijk_json,
i,
j,
k,
resolved_grid_index,
scalar_func_id,
vector_func_id,
iso_func_id,
threshold_func_id,
),
"probe_ijk result",
)
return _parse_probe_result(payload)
def _integrate_surface(surface: object) -> IntegrationResult:
"""Integrate the current scalar on a single surface."""
prefix, phigs_obj = _get_surface_target(surface)
funcs = {
"boundary_surf": getattr(_core_module, "boundary_surf_integrate", None),
"comp_surf": getattr(_core_module, "comp_surf_integrate", None),
"coord_surf": getattr(_core_module, "coord_surf_integrate", None),
"iso_surf": getattr(_core_module, "iso_surf_integrate", None),
}
func = funcs.get(prefix)
if not callable(func):
raise InvalidArgumentError(
"surface must be a Boundary, Comp, Coord, or Iso surface"
)
payload = cast(dict[str, object], _core_call(func, phigs_obj))
return _parse_integration_result(payload)
def _integrate_partial_surface(
surface: object,
point: object,
tolerance: float | None = None,
) -> IntegrationResult | None:
"""Integrate the connected partial region on a coord or iso surface."""
prefix, phigs_obj = _get_surface_target(surface)
x, y, z = _coerce_xyz_triplet(point, "point")
tol = _validate_tolerance(tolerance)
funcs = {
"coord_surf": getattr(_core_module, "coord_surf_integrate_partial", None),
"iso_surf": getattr(_core_module, "iso_surf_integrate_partial", None),
}
func = funcs.get(prefix)
if not callable(func):
raise InvalidArgumentError("surface must be a Coord or Iso surface")
payload = _core_call(func, phigs_obj, x, y, z, tol)
if payload is None:
return None
return _parse_integration_result(cast(dict[str, object], payload))
_READER_ALIASES = {
"plot3d": "plot3d",
"overflow2": "plot3d",
"fvuns": "unstructured",
"vtk_structured": "vtk_structured",
"acusolve_direct": "acusolve_direct",
"acusolve_fvuns": "acusolve_unstructured",
"ansys_cfx_fvuns": "ansys-cfx_unstructured",
"fluent_cff_direct": "fluent_cff_cas/dat_direct",
"append_sampled_data": "append_sampled_data",
"pw_common": "pw common",
"cfdpp_fvuns": "cfd++_unstructured",
"cgns_structured": "cgns_structured",
"cgns_unstructured": "cgns_unstructured",
"cgns_unstructured_hybrid": "cgns_unstructured/hybrid",
"cobalt_fvuns": "cobalt_unstructured",
"converge_fvuns": "converge_unstructured",
"ensight": "ensight",
"flow3d_animation": "flow-3d animation",
"flow3d_restart": "flow-3d restart",
"fluent_fvuns": "fluent_unstructured",
"fluent_unstructured": "fluent_unstructured",
"fluent_cas_dat_direct": "fluent_cas/dat_direct",
"fluent_direct": "fluent_direct",
"havoc": "havoc",
"lsdyna_d3plot": "lsdyna_d3plot",
"lsdyna_state": "lsdyna",
"nparc_wind": "nparc/wind",
"openfoam_direct": "openfoam",
"openfoam_fvuns": "openfoam_unstructured",
"sc_tetra": "sc_tetra",
"sc_flow": "sc_flow",
"scryu": "scryu",
"sc_stream": "sc_stream",
"starccm_fvuns": "starccm_unstructured",
"starcd_fvuns": "starcd_unstructured",
"stl": "stl",
"surface_sampled_data": "surface_sampled_data",
"tecplot_360": "tecplot_360",
"ultrafluidx_direct": "ufx",
"vtk_unstructured_hybrid": "vtk_unstructured/hybrid",
"wind_structured": "wind_structured",
"wind_unstructured": "wind_unstructured",
"xdb_import": "xdb_import",
}
def reader_aliases() -> dict[str, str]:
"""Return a copy of reader aliases supported by the generic reader."""
return dict(_READER_ALIASES)
[docs]
@dataclass
class DatasetBounds:
"""Axis-aligned bounding box for a dataset.
Cartesian axes are always populated. ``rmin``/``rmax`` and ``tmin``/``tmax``
are non-zero only for datasets with a cylindrical coordinate system.
"""
xmin: float = 0.0
xmax: float = 0.0
ymin: float = 0.0
ymax: float = 0.0
zmin: float = 0.0
zmax: float = 0.0
rmin: float = 0.0
rmax: float = 0.0
tmin: float = 0.0
tmax: float = 0.0
[docs]
@dataclass
class SessionState:
"""Snapshot of the current FieldView session.
Returned by :func:`get_session_state`. Contains dataset metadata, domain
bounds, function lists, and all existing surface objects in a single call.
When ``data_loaded`` is ``False``, ``dataset_id`` is ``None`` and all lists
are empty.
The returned fields are grouped roughly as follows:
- Dataset identity: ``data_loaded``, ``dataset_id``, ``data_format``,
``reader_name``,
``grid_file``, ``result_file``, ``num_grids``
- Domain extents: ``bounds``
- Transient state: ``transient``, ``has_solution_times``,
``cur_time_step``, ``total_time_steps``
- Available functions: ``scalar_functions``, ``vector_functions``
- Existing scene objects: ``objects``
- Window/layout state: ``windows``
Example:
.. code-block:: python
>>> import os
>>> import fieldview as fv
>>> data_dir = os.path.join(fv.home, "examples", "f18")
>>> ds = fv.data.load_plot3d(
... os.path.join(data_dir, "f18i9b_g_bin"),
... os.path.join(data_dir, "f18i9b_q_bin"),
... )
>>> state = fv.data.get_session_state()
>>> state.dataset_id, state.num_grids
>>> state.bounds.xmin, state.bounds.xmax
>>> len(state.scalar_functions), len(state.objects.boundary_list)
"""
data_loaded: bool
dataset_id: int | None
data_format: str
reader_name: str
grid_file: str
result_file: str
num_grids: int
bounds: DatasetBounds
transient: bool
has_solution_times: bool
cur_time_step: int
total_time_steps: int
scalar_functions: list[str]
vector_functions: list[str]
objects: object # ObjectLists from fieldview._objects; typed as object to avoid circular import
windows: WindowList | None = None
def _build_window_info(payload: dict[str, object]) -> WindowInfo:
parent_window = payload.get("parent_window")
environment_id = payload.get("environment_id")
dataset_ids = [
_to_int(dataset_id)
for dataset_id in cast(Iterable[object], payload.get("dataset_ids", []))
]
return WindowInfo(
window=_to_int(payload.get("window", 0)),
current=bool(payload.get("current", False)),
scene_index=_to_int(payload.get("scene_index", -1)),
parent_window=None if parent_window is None else _to_int(parent_window),
label=None if payload.get("label") is None else str(payload.get("label")),
environment=None
if payload.get("environment") is None
else str(payload.get("environment")),
view_sync_enabled=bool(payload.get("view_sync_enabled", False)),
background=None
if payload.get("background") is None
else str(payload.get("background")),
dataset_ids=dataset_ids,
environment_id=None if environment_id is None else _to_int(environment_id),
background_image=None
if payload.get("background_image") is None
else str(payload.get("background_image")),
)
def _build_window_list(payload: dict[str, object] | None) -> WindowList | None:
if not isinstance(payload, dict):
return None
current_window = payload.get("current_window")
if current_window is not None:
current_window = _to_int(current_window)
windows = [
_build_window_info(cast(dict[str, object], item))
for item in cast(Iterable[object], payload.get("windows", []))
if isinstance(item, dict)
]
return WindowList(current_window=current_window, windows=windows)
def _build_window_split_result(payload: dict[str, object]) -> WindowSplitResult:
return WindowSplitResult(
source_window=_to_int(payload.get("source_window", 0)),
new_window=_to_int(payload.get("new_window", 0)),
mode=str(payload.get("mode", "")),
orientation=str(payload.get("orientation", "")),
)
def _build_perspective_state(payload: dict[str, object]) -> PerspectiveState:
return PerspectiveState(
enabled=bool(payload.get("enabled", False)),
angle=_to_float(payload.get("angle", 0.0)),
)
def _build_camera_pose(payload: dict[str, object]) -> CameraPose:
def _tuple3(key: str) -> tuple[float, float, float]:
values = payload.get(key, (0.0, 0.0, 0.0))
if not isinstance(values, (list, tuple)):
values = (0.0, 0.0, 0.0)
values_tuple = tuple(cast(tuple[object, ...], values))
if len(values_tuple) != 3:
values_tuple = (0.0, 0.0, 0.0)
return (
_to_float(values_tuple[0]),
_to_float(values_tuple[1]),
_to_float(values_tuple[2]),
)
return CameraPose(
eye=_tuple3("eye"),
target=_tuple3("target"),
up=_tuple3("up"),
perspective_enabled=bool(payload.get("perspective_enabled", False)),
perspective_angle=_to_float(payload.get("perspective_angle", 0.0)),
)
def _build_view_state(payload: dict[str, object]) -> ViewState:
exact_state_obj = payload.get("exact_state")
if not isinstance(exact_state_obj, dict):
exact_state_obj = {}
exact_payload = cast(dict[object, object], exact_state_obj)
def _tuple3_value(
value: object, default: tuple[float, float, float]
) -> tuple[float, float, float]:
if not isinstance(value, (list, tuple)):
value = default
values_tuple = tuple(cast(tuple[object, ...], value))
if len(values_tuple) != 3:
values_tuple = default
return (
_to_float(values_tuple[0]),
_to_float(values_tuple[1]),
_to_float(values_tuple[2]),
)
exact_state = CameraExactState(
zoom=_to_float(exact_payload.get("zoom", 1.0)),
scale=_to_float(exact_payload.get("scale", 1.0)),
rotation_angle=_to_float(exact_payload.get("rotation_angle", 0.0)),
rotation_axis=_tuple3_value(
exact_payload.get("rotation_axis", (0.0, 0.0, 1.0)),
(0.0, 0.0, 1.0),
),
translation=_tuple3_value(
exact_payload.get("translation", (0.0, 0.0, 0.0)),
(0.0, 0.0, 0.0),
),
perspective_z=_to_float(exact_payload.get("perspective_z", 0.0)),
rotation_center_on=bool(exact_payload.get("rotation_center_on", False)),
rotation_center=_tuple3_value(
exact_payload.get("rotation_center", (0.0, 0.0, 0.0)),
(0.0, 0.0, 0.0),
),
)
return ViewState.from_exact(
exact_state,
perspective_enabled=bool(payload.get("perspective_enabled", False)),
perspective_angle=_to_float(payload.get("perspective_angle", 0.0)),
)
[docs]
def get_session_state(include_functions: bool = True) -> SessionState:
"""Return a snapshot of the current FieldView session.
Retrieves dataset metadata, domain bounds, function lists, existing scene
objects, and window/layout state in a single dispatcher round-trip. If no
dataset is loaded, ``data_loaded`` is ``False`` and most fields are empty
or ``None``.
As a side effect, the current dataset is attached into the Python registry
so that a subsequent :func:`get_current` call returns the cached wrapper
at no extra cost.
This is the recommended first call for any automation script entering a
live FieldView session with unknown state.
The returned :class:`SessionState` is a compact snapshot with:
- dataset metadata and file names
- full domain bounds
- transient/time-step information
- scalar/vector function name lists
- all current graphics objects grouped in ``state.objects``
- current window/layout information in ``state.windows``
Append-vs-replace invalidation semantics:
- Replace-style reloads invalidate previously attached wrappers for the
replaced dataset slot. Re-acquire wrappers after the reload completes.
- Append-style reads allocate a new dataset id and leave existing wrappers
valid.
Example:
.. code-block:: python
>>> import os
>>> import fieldview as fv
>>> data_dir = os.path.join(fv.home, "examples", "f18")
>>> ds = fv.data.load_plot3d(
... os.path.join(data_dir, "f18i9b_g_bin"),
... os.path.join(data_dir, "f18i9b_q_bin"),
... )
>>> state = fv.data.get_session_state()
>>> current = fv.data.get_current() # cached after get_session_state()
>>> mid_x = (state.bounds.xmin + state.bounds.xmax) / 2.0
>>> len(state.scalar_functions), len(state.objects.boundary_list)
:param include_functions: When ``False``, scalar/vector function lists are
omitted from the response. Useful for large datasets where the script
only needs object/bounds context.
"""
from ._objects import ObjectLists, _build_group # local import avoids circular
_get_session_state_fn = getattr(_core_module, "get_session_state", None)
if not callable(_get_session_state_fn):
raise CoreError("get_session_state: not supported by this FieldView build")
# parse_response_dict_or_throw in C++ already validates "ok" and strips it,
# so the dict we receive here has no "ok" key — errors throw before we get here.
resp = cast(dict[str, object], _core_call(_get_session_state_fn, include_functions))
data_loaded = bool(resp.get("data_loaded", False))
bounds = DatasetBounds(
xmin=_to_float(resp.get("xmin", 0.0)),
xmax=_to_float(resp.get("xmax", 0.0)),
ymin=_to_float(resp.get("ymin", 0.0)),
ymax=_to_float(resp.get("ymax", 0.0)),
zmin=_to_float(resp.get("zmin", 0.0)),
zmax=_to_float(resp.get("zmax", 0.0)),
rmin=_to_float(resp.get("rmin", 0.0)),
rmax=_to_float(resp.get("rmax", 0.0)),
tmin=_to_float(resp.get("tmin", 0.0)),
tmax=_to_float(resp.get("tmax", 0.0)),
)
objects = ObjectLists(
boundary_list=_build_group(resp, "boundary_list"),
comp_list=_build_group(resp, "comp_list"),
coord_list=_build_group(resp, "coord_list"),
iso_list=_build_group(resp, "iso_list"),
streamlines_list=_build_group(resp, "streamlines_list"),
particle_paths_list=_build_group(resp, "particle_paths_list")
if "particle_paths_list" in resp
else [],
plot2d_list=_build_group(resp, "plot2d_list") if "plot2d_list" in resp else [],
vortex_cores_list=_build_group(resp, "vortex_cores_list")
if "vortex_cores_list" in resp
else [],
surface_flows_list=_build_group(resp, "surface_flows_list")
if "surface_flows_list" in resp
else [],
separation_lines_list=_build_group(resp, "separation_lines_list")
if "separation_lines_list" in resp
else [],
reattachment_lines_list=_build_group(resp, "reattachment_lines_list")
if "reattachment_lines_list" in resp
else [],
annotation_text_list=_build_group(resp, "annotation_text_list"),
annotation_arrow_list=_build_group(resp, "annotation_arrow_list"),
)
# Side effect: attach the current dataset into the registry so get_current()
# returns a cached wrapper without a second round-trip.
if data_loaded:
dataset_id = resp.get("dataset_id")
if dataset_id is not None:
try:
_get_or_attach_dataset(_to_int(dataset_id))
except InvalidDatasetError:
pass
scalar_funcs = [
str(cast(dict[str, object], f).get("name", ""))
if isinstance(f, dict)
else str(f)
for f in cast(Iterable[object], resp.get("scalar_functions", []))
]
vector_funcs = [
str(cast(dict[str, object], f).get("name", ""))
if isinstance(f, dict)
else str(f)
for f in cast(Iterable[object], resp.get("vector_functions", []))
]
windows = _build_window_list(cast(dict[str, object] | None, resp.get("windows")))
dataset_id_value = resp.get("dataset_id")
dataset_id = int(dataset_id_value) if isinstance(dataset_id_value, int) else None
return SessionState(
data_loaded=data_loaded,
dataset_id=dataset_id,
data_format=str(resp.get("data_format", "")),
reader_name=str(resp.get("reader_name", "")),
grid_file=str(resp.get("grid_file", "")),
result_file=str(resp.get("result_file", "")),
num_grids=_to_int(resp.get("num_grids", 0)),
bounds=bounds,
transient=bool(resp.get("transient", False)),
has_solution_times=bool(resp.get("has_solution_times", False)),
cur_time_step=_to_int(resp.get("cur_time_step", 0)),
total_time_steps=_to_int(resp.get("total_time_steps", 0)),
scalar_functions=scalar_funcs,
vector_functions=vector_funcs,
objects=objects,
windows=windows,
)
[docs]
def get_current() -> Dataset:
"""Return the current dataset from the live FieldView session.
If the dataset was loaded through the FieldView UI before Python attached
to the session, this call reconstructs and caches a live :class:`Dataset`
wrapper on demand instead of failing on a Python-registry miss.
Replace-style reloads invalidate the previous wrapper for that dataset
slot. Append-style reads allocate a new dataset id and leave existing
wrappers valid.
Example:
.. code-block:: python
>>> import os
>>> import fieldview as fv
>>> data_dir = os.path.join(fv.home, "examples", "f18")
>>> ds = fv.data.load_plot3d(
... os.path.join(data_dir, "f18i9b_g_bin"),
... os.path.join(data_dir, "f18i9b_q_bin"),
... )
>>> current = fv.data.get_current()
>>> current.dataset_id == ds.dataset_id
"""
global _current_dataset
get_current_id = getattr(_core_module, "dataset_get_current_id", None)
if callable(get_current_id):
try:
dsnum = _to_int(_core_call(get_current_id))
except CoreError as exc:
raise InvalidDatasetError(str(exc)) from exc
if dsnum < 0:
raise InvalidDatasetError(
"No current dataset is set. Load a dataset or call fieldview.data.set_current()."
)
dataset_ref = _get_or_attach_dataset(dsnum)
_current_dataset = dataset_ref
return dataset_ref
current_dataset = _current_dataset
if current_dataset is None:
raise InvalidDatasetError(
"No current dataset is set. Load a dataset or call fieldview.data.set_current()."
)
current_dataset._ensure_valid()
return current_dataset
def set_current(ds: Dataset) -> None:
"""Set the current dataset used by create_* helpers when dataset is omitted.
Args:
ds: Dataset instance to make current for subsequent helper calls that
omit an explicit dataset argument.
"""
if not isinstance(ds, Dataset):
raise InvalidArgumentError("ds must be a Dataset instance")
if ds.dataset_id < 0:
raise InvalidDatasetError("dataset is invalid (not loaded)")
ds._ensure_valid()
set_current_id = getattr(_core_module, "dataset_set_current_id", None)
if callable(set_current_id):
_core_call(set_current_id, int(ds.dataset_id))
_dataset_registry[ds.dataset_id] = ds
global _current_dataset
_current_dataset = ds
dataset: type[Dataset] = Dataset
