"""Vortex cores surface helpers.
Example usage:
.. 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"),
... )
>>> vc = fv.vis.create_vortex_cores(
... ds,
... method=fv.constant.VortexCoreMethod.VORTICITY_ALIGNMENT,
... vector_func="Vorticity Vectors [PLOT3D]",
... )
>>> vc.coloring = fv.constant.Coloring.SCALAR
"""
from __future__ import annotations
from typing import cast
from .. import _core
from .. import constant
from .._core_utils import _core_call
from .._base import (
PathBase,
_flatten_colormap,
_flatten_legend_options,
_prefix_payload,
)
from ..data import Dataset, _resolve_dataset_or_current
from ..exceptions import InvalidArgumentError, InvalidDatasetError
from ..utils import Colormap, Legend, _coerce_float
__all__: list[str] = ["VortexCores", "create_vortex_cores"]
_VORTEX_CORE_METHODS: dict[str, str] = {
"vorticity alignment": "vorticity alignment",
"vorticityalignment": "vorticity alignment",
"eigenmode analysis": "eigenmode analysis",
"eigenmodeanalysis": "eigenmode analysis",
}
_VORTEX_CORE_PATH_VARIABLES: dict[str, str] = {
"vortex strength": "Vortex Strength",
"vortexstrength": "Vortex Strength",
}
def _normalize_method(value: constant.VortexCoreMethod | str) -> str:
if isinstance(value, constant.VortexCoreMethod):
return value.value
canonical = _VORTEX_CORE_METHODS.get(str(value).strip().lower().replace("_", " "))
if canonical is None:
allowed = ", ".join(sorted({v for v in _VORTEX_CORE_METHODS.values()}))
raise InvalidArgumentError(f"method must be one of: {allowed}")
return canonical
def _normalize_coloring(value: constant.Coloring | str) -> str:
if isinstance(value, constant.Coloring):
mode = value.value
else:
mode = str(value).strip().lower()
if mode == constant.Coloring.MATERIAL.value:
raise InvalidArgumentError(
"material coloring is not supported for vortex cores"
)
return mode
def _normalize_line_type(value: constant.LineType | str) -> str:
if isinstance(value, constant.LineType):
return value.value
return str(value).strip().lower()
def _build_modify_payload(
*,
dataset: Dataset | None,
visibility: bool | None = None,
line_type: constant.LineType | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
min_vortex_strength_enabled: bool | None = None,
min_vortex_strength: float | None = None,
coloring: constant.Coloring | str | None = None,
scalar_func: str | None = None,
scalar_path_variable: str | None = None,
vector_func: str | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
) -> dict[str, object]:
payload: dict[str, object] = {}
normalized_min_vortex_strength: float | None = None
if visibility is not None:
payload["visibility"] = bool(visibility)
if line_type is not None:
payload["line_type"] = _normalize_line_type(line_type)
if geometric_color is not None:
payload["geometric_color"] = int(geometric_color)
if min_vortex_strength_enabled is not None:
payload["min_vortex_strength_enabled"] = bool(min_vortex_strength_enabled)
if min_vortex_strength is not None:
try:
normalized_min_vortex_strength = _coerce_float(min_vortex_strength)
except (TypeError, ValueError) as exc:
raise InvalidArgumentError("min_vortex_strength must be a float") from exc
if (
normalized_min_vortex_strength is not None
and min_vortex_strength_enabled is not False
):
payload["min_vortex_strength"] = normalized_min_vortex_strength
if coloring is not None:
payload["coloring"] = _normalize_coloring(coloring)
if scalar_path_variable is not None:
payload["scalar_path_variable"] = _normalize_scalar_path_variable(
dataset, scalar_path_variable
)
if scalar_func is not None:
stripped = scalar_func.strip()
if not stripped:
raise InvalidArgumentError("scalar_func cannot be empty")
if stripped.lower() == "none":
payload["scalar_func"] = "none"
else:
func_id = _lookup_function_id(
dataset, stripped, "_scalar_function_ids", "scalar"
)
if func_id is None:
raise InvalidArgumentError(
"scalar_func must be a dataset scalar function; use scalar_path_variable for path variables"
)
payload["scalar_func_id"] = func_id
if vector_func is not None:
payload.update(_normalize_vector_func_name(dataset, vector_func))
if colormap is not None:
if isinstance(colormap, Colormap):
cmap_payload = _flatten_colormap(colormap.to_payload())
elif isinstance(colormap, dict):
cmap_payload = _flatten_colormap(colormap)
else:
raise InvalidArgumentError("colormap must be a Colormap or dict")
payload.update(_prefix_payload("scalar_colormap_", cmap_payload))
if legend is not None:
if isinstance(legend, Legend):
legend_payload = legend.to_payload()
elif isinstance(legend, dict):
legend_payload = _flatten_legend_options(legend)
else:
raise InvalidArgumentError("legend must be a Legend or dict")
payload.update(_prefix_payload("legend_", legend_payload))
return payload
def _require_dataset(dataset: Dataset | None) -> Dataset:
if dataset is None:
raise InvalidDatasetError("vortex cores surface is missing a dataset reference")
dataset._ensure_valid()
return dataset
def _lookup_function_id(
dataset: Dataset | None,
name: str,
attr: str,
_label: str,
) -> int | None:
ds = _require_dataset(dataset)
mapping = cast(dict[str, int], getattr(ds, attr, {}))
needle = name.strip().lower()
for func_name, func_id in mapping.items():
if func_name.lower() == needle:
return func_id
return None
def _lookup_function_name(
dataset: Dataset | None,
name: str,
attr: str,
_label: str,
) -> str | None:
ds = _require_dataset(dataset)
mapping = cast(dict[str, int], getattr(ds, attr, {}))
needle = name.strip().lower()
for func_name in mapping:
if func_name.lower() == needle:
return str(func_name)
return None
def _normalize_vector_func_name(
dataset: Dataset | None, value: str
) -> dict[str, object]:
if not isinstance(value, str):
raise InvalidArgumentError("vector_func must be a string")
name = value.strip()
if not name:
raise InvalidArgumentError("vector_func cannot be empty")
if name.lower() == "none":
return {"vector_func": "none"}
func_id = _lookup_function_id(dataset, name, "_vector_function_ids", "vector")
if func_id is None:
raise InvalidArgumentError("vector_func must be a dataset vector function")
return {"vector_func_id": func_id}
def _normalize_scalar_path_variable(dataset: Dataset | None, value: str) -> str:
if not isinstance(value, str):
raise InvalidArgumentError("scalar_path_variable must be a string")
name = value.strip()
if not name:
raise InvalidArgumentError("scalar_path_variable cannot be empty")
dataset_name = _lookup_function_name(
dataset, name, "_scalar_function_ids", "scalar"
)
if dataset_name is not None:
return dataset_name
canonical = _VORTEX_CORE_PATH_VARIABLES.get(name.lower().replace("_", " "))
if canonical is not None:
return canonical
raise InvalidArgumentError(
"scalar_path_variable must be a dataset scalar or 'Vortex Strength'"
)
[docs]
class VortexCores(PathBase):
"""Vortex cores surface wrapper.
Example usage:
.. 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"),
... )
>>> vc = fv.vis.create_vortex_cores(
... ds,
... method="Vorticity Alignment",
... vector_func="Vorticity Vectors [PLOT3D]",
... )
>>> vc.coloring = fv.constant.Coloring.SCALAR
"""
_core_prefix = "vortex_cores_surf"
__slots__ = ("_dataset", "_method")
def __init__(
self,
phigs_obj: int,
dataset_id: int,
dataset: Dataset | None = None,
method: constant.VortexCoreMethod | str | None = None,
) -> None:
"""Create a vortex-cores wrapper around an existing host object.
Args:
phigs_obj: Host PHIGS object identifier for the feature object.
dataset_id: Host dataset identifier backing the object.
dataset: Optional live :class:`Dataset` wrapper for validation.
method: Optional vortex-core extraction method to cache locally.
"""
super().__init__(phigs_obj, dataset_id)
self._dataset = dataset
self._method = _normalize_method(method) if method is not None else None
def _state(self) -> dict[str, object]:
return _core_call(_core.vortex_cores_surf_get_state, self._phigs_obj)
def _read_method(self, state: dict[str, object] | None = None) -> str | None:
if self._method is not None:
return self._method
snapshot = self._state() if state is None else state
method_raw = snapshot.get("method")
if isinstance(method_raw, str) and method_raw.strip():
method = _normalize_method(method_raw)
self._method = method
return method
return self._method
@property
def method(self) -> str | None:
"""`Optional[str]`: Active vortex-core extraction method.
Example usage:
.. code-block:: python
>>> vc.method
"""
return self._read_method()
@property
def display_type(self) -> constant.PathsDisplayType:
"""`PathsDisplayType`: Display representation for vortex cores.
Only ``COMPLETE`` is supported.
Example usage:
.. code-block:: python
>>> vc.display_type = fv.constant.PathsDisplayType.COMPLETE
"""
return constant.PathsDisplayType.COMPLETE
@display_type.setter
def display_type(self, value: constant.PathsDisplayType | str) -> None:
normalized = (
value.value
if isinstance(value, constant.PathsDisplayType)
else str(value).strip().lower()
)
if normalized != constant.PathsDisplayType.COMPLETE.value:
raise InvalidArgumentError(
"vortex cores only support display_type='complete'"
)
@property
def min_vortex_strength_enabled(self) -> bool:
"""`bool`: Enable/disable the MIN VORTEX STRENGTH filter.
Example usage:
.. code-block:: python
>>> vc.min_vortex_strength_enabled = True
"""
return bool(self._state().get("min_vortex_strength_enabled", False))
@min_vortex_strength_enabled.setter
def min_vortex_strength_enabled(self, value: bool) -> None:
_core_call(
_core.vortex_cores_surf_modify,
self._phigs_obj,
{"min_vortex_strength_enabled": bool(value)},
)
@property
def min_vortex_strength(self) -> float:
"""`float`: Current MIN VORTEX STRENGTH slider value.
Example usage:
.. code-block:: python
>>> vc.min_vortex_strength = 0.5
"""
return _coerce_float(self._state().get("min_vortex_strength", 0.0))
@min_vortex_strength.setter
def min_vortex_strength(self, value: float) -> None:
try:
val = float(value)
except (TypeError, ValueError) as exc:
raise InvalidArgumentError("min_vortex_strength must be a float") from exc
_core_call(
_core.vortex_cores_surf_modify,
self._phigs_obj,
{"min_vortex_strength": val},
)
@property
def scalar_func(self) -> str | None:
"""`Optional[str]`: Dataset scalar function cached for registry loading.
Example usage:
.. code-block:: python
>>> vc.scalar_func = "Density (Q1)"
"""
name = str(self._state().get("scalar_func_name", "")).strip()
return name or None
@scalar_func.setter
def scalar_func(self, value: str) -> None:
if not isinstance(value, str):
raise InvalidArgumentError("scalar_func must be a string")
name = value.strip()
if not name:
raise InvalidArgumentError("scalar_func cannot be empty")
if name.lower() == "none":
_core_call(
_core.vortex_cores_surf_modify, self._phigs_obj, {"scalar_func": "none"}
)
return
func_id = _lookup_function_id(
self._dataset, name, "_scalar_function_ids", "scalar"
)
if func_id is None:
raise InvalidArgumentError(
"scalar_func must be a dataset scalar function; use scalar_path_variable for path variables"
)
_core_call(
_core.vortex_cores_surf_modify, self._phigs_obj, {"scalar_func_id": func_id}
)
@property
def scalar_path_variable(self) -> str | None:
"""`Optional[str]`: Active scalar coloring selection for dataset or feature variables.
Example usage:
.. code-block:: python
>>> vc.scalar_path_variable = "Vortex Strength"
"""
name = str(self._state().get("scalar_path_variable", "")).strip()
return name or None
@scalar_path_variable.setter
def scalar_path_variable(self, value: str) -> None:
_core_call(
_core.vortex_cores_surf_modify,
self._phigs_obj,
{
"scalar_path_variable": _normalize_scalar_path_variable(
self._dataset, value
)
},
)
@property
def vector_func(self) -> str | None:
"""`Optional[str]`: Vector variable name.
Example usage:
.. code-block:: python
>>> vc.vector_func = "Velocity"
"""
name = str(self._state().get("vector_func_name", "")).strip()
return name or None
@vector_func.setter
def vector_func(self, value: str) -> None:
_core_call(
_core.vortex_cores_surf_modify,
self._phigs_obj,
_normalize_vector_func_name(self._dataset, value),
)
# -- methods --
[docs]
def modify(
self,
*,
visibility: bool | None = None,
line_type: constant.LineType | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
min_vortex_strength_enabled: bool | None = None,
min_vortex_strength: float | None = None,
coloring: constant.Coloring | str | None = None,
scalar_func: str | None = None,
scalar_path_variable: str | None = None,
vector_func: str | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
) -> None:
"""Modify multiple vortex-cores display attributes in one call.
Args:
visibility: Whether the object is visible.
line_type: Line style used for the extracted cores.
geometric_color: Geometric color id used when geometric coloring is
active.
min_vortex_strength_enabled: Whether the minimum-vortex-strength
filter is enabled.
min_vortex_strength: Minimum vortex-strength filter value.
coloring: Coloring mode for the extracted cores.
scalar_func: Scalar function name used for scalar coloring.
scalar_path_variable: Path-variable name used for scalar coloring.
vector_func: Vector variable name used for the extraction or
display.
colormap: Scalar colormap options.
legend: Legend options.
Example usage:
.. code-block:: python
>>> vc.modify(
... visibility=True,
... coloring="scalar",
... scalar_path_variable="Vortex Strength",
... )
"""
payload = _build_modify_payload(
dataset=self._dataset,
visibility=visibility,
line_type=line_type,
geometric_color=geometric_color,
min_vortex_strength_enabled=min_vortex_strength_enabled,
min_vortex_strength=min_vortex_strength,
coloring=coloring,
scalar_func=scalar_func,
scalar_path_variable=scalar_path_variable,
vector_func=vector_func,
colormap=colormap,
legend=legend,
)
if not payload:
raise InvalidArgumentError("modify requires at least one property")
_core_call(_core.vortex_cores_surf_modify, self._phigs_obj, payload)
[docs]
def copy(self) -> "VortexCores":
"""Return a new vortex-cores surface with the same settings.
Example usage:
.. code-block:: python
>>> copy_obj = vc.copy()
"""
dataset = _require_dataset(self._dataset)
state = self._state()
method = self._read_method(state)
if method is None:
raise InvalidArgumentError("cannot copy vortex cores without method")
vector_func_raw = state.get("vector_func_name")
vector_func = (
vector_func_raw.strip() if isinstance(vector_func_raw, str) else ""
)
if not vector_func:
raise InvalidArgumentError("cannot copy vortex cores without vector_func")
scalar_func_raw = state.get("scalar_func_name")
scalar_func_name = (
scalar_func_raw.strip() if isinstance(scalar_func_raw, str) else ""
)
scalar_func: str | None = scalar_func_name or None
scalar_path_raw = state.get("scalar_path_variable")
scalar_path_name = (
scalar_path_raw.strip() if isinstance(scalar_path_raw, str) else ""
)
scalar_path_variable: str | None = scalar_path_name or None
min_vortex_strength_enabled = cast(
bool | None, state.get("min_vortex_strength_enabled")
)
min_vortex_strength = (
None
if state.get("min_vortex_strength") is None
else _coerce_float(state.get("min_vortex_strength"))
)
copied = create_vortex_cores(
dataset,
method=method,
vector_func=vector_func,
visibility=cast(bool | None, state.get("visibility")),
line_type=cast(constant.LineType | str | None, state.get("line_type")),
geometric_color=cast(
constant.GeometricColor | int | None,
state.get("geometric_color"),
),
min_vortex_strength_enabled=min_vortex_strength_enabled,
min_vortex_strength=min_vortex_strength,
coloring=cast(constant.Coloring | str | None, state.get("coloring")),
scalar_func=scalar_func,
scalar_path_variable=scalar_path_variable,
colormap=cast(
Colormap | dict[str, object] | None,
state.get("scalar_colormap"),
),
legend=cast(Legend | dict[str, object] | None, state.get("legend")),
)
return copied
[docs]
def delete(self) -> None:
"""Delete this vortex-cores surface.
Example usage:
.. code-block:: python
>>> vc.delete()
"""
self._delete_surface(_core.vortex_cores_surf_delete)
[docs]
def create_vortex_cores(
dataset: Dataset | None = None,
*,
method: constant.VortexCoreMethod | str,
vector_func: str,
visibility: bool | None = None,
line_type: constant.LineType | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
min_vortex_strength_enabled: bool | None = None,
min_vortex_strength: float | None = None,
coloring: constant.Coloring | str | None = None,
scalar_func: str | None = None,
scalar_path_variable: str | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
) -> VortexCores:
"""Create a vortex-cores surface.
`method` and `vector_func` are required.
`method` must be ``"Vorticity Alignment"`` or
``"Eigenmode Analysis"``.
Args:
dataset: Dataset to attach the new feature object to. When omitted,
the current dataset is used.
method: Vortex-core extraction method.
vector_func: Vector variable name used for the extraction.
visibility: Whether the object is initially visible.
line_type: Line style used for the extracted cores.
geometric_color: Geometric color id used when geometric coloring is
active.
min_vortex_strength_enabled: Whether the minimum-vortex-strength
filter is enabled.
min_vortex_strength: Minimum vortex-strength filter value.
coloring: Coloring mode for the extracted cores.
scalar_func: Scalar function name used for scalar coloring.
scalar_path_variable: Path-variable name used for scalar coloring.
colormap: Scalar colormap options.
legend: Legend options.
Example usage:
.. 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"),
... )
>>> vc = fv.vis.create_vortex_cores(
... ds,
... method="Vorticity Alignment",
... vector_func="Vorticity Vectors [PLOT3D]",
... coloring=fv.constant.Coloring.SCALAR,
... scalar_path_variable="Vortex Strength",
... )
"""
ds = _resolve_dataset_or_current(dataset)
canonical_method = _normalize_method(method)
phigs_obj = _core_call(
_core.vortex_cores_surf_create,
ds.dataset_id,
canonical_method,
defer_apply=True,
)
surf = VortexCores(
phigs_obj=phigs_obj,
dataset_id=ds.dataset_id,
dataset=ds,
method=canonical_method,
)
try:
surf.modify(
visibility=visibility,
line_type=line_type,
geometric_color=geometric_color,
min_vortex_strength_enabled=min_vortex_strength_enabled,
min_vortex_strength=min_vortex_strength,
coloring=coloring,
scalar_func=scalar_func,
scalar_path_variable=scalar_path_variable,
vector_func=vector_func,
colormap=colormap,
legend=legend,
)
if min_vortex_strength_enabled is False and min_vortex_strength is not None:
surf.min_vortex_strength = min_vortex_strength
except Exception:
_core_call(_core.vortex_cores_surf_delete, phigs_obj)
raise
return surf
