"""Surface flow feature-extraction path helpers.
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"),
... )
>>> sf = fv.vis.create_surface_flows(
... ds,
... mode=fv.constant.SurfaceFlowMode.NO_SLIP,
... vector_func="Velocity Vectors [PLOT3D]",
... direction=fv.constant.CalculationDirection.BOTH,
... )
>>> sf.calculate()
"""
from __future__ import annotations
import os
from typing import TypedDict, 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, _coerce_int, _coerce_pathlike_str
__all__: list[str] = ["SurfaceFlows", "create_surface_flows"]
_UNSET = object()
_SURFACE_FLOW_MODES: dict[str, str] = {
"no slip": "no_slip",
"noslip": "no_slip",
"euler": "euler",
"offset": "offset",
}
_SURFACE_FLOW_SEEDING: dict[str, str] = {
"low density": "low_density",
"lowdensity": "low_density",
"medium density": "medium_density",
"mediumdensity": "medium_density",
"high density": "high_density",
"highdensity": "high_density",
"from file": "from_file",
}
_SURFACE_FLOW_PATH_VARIABLES: dict[str, str] = {
"duration": "Duration",
"wall shear magnitude": "Wall Shear Magnitude",
"wallshearmagnitude": "Wall Shear Magnitude",
}
_FEATURE_COLORING = {
constant.Coloring.GEOMETRIC.value,
constant.Coloring.SCALAR.value,
}
_FEATURE_LINE_TYPES = {
constant.LineType.THIN.value,
constant.LineType.MEDIUM.value,
constant.LineType.THICK.value,
}
class _FeatureStateKwargs(TypedDict):
visibility: bool | None
line_type: constant.LineType | str | None
geometric_color: constant.GeometricColor | int | None
coloring: constant.Coloring | str | None
colormap: Colormap | dict[str, object] | None
legend: Legend | dict[str, object] | None
class _SurfaceFlowStateKwargs(_FeatureStateKwargs):
mode: constant.SurfaceFlowMode | str
scalar_func: str | None
scalar_path_variable: str | None
vector_func: str | None
direction: constant.CalculationDirection | str | None
time_limit_enabled: bool | None
time_limit: object
seeding: constant.SurfaceFlowSeeding | str | None
seed_file: object
offset_distance: float | None
step: int | None
def _normalize_mode(value: constant.SurfaceFlowMode | str) -> str:
if isinstance(value, constant.SurfaceFlowMode):
return value.value
canonical = _SURFACE_FLOW_MODES.get(str(value).strip().lower().replace("_", " "))
if canonical is None:
allowed = ", ".join(sorted({v for v in _SURFACE_FLOW_MODES.values()}))
raise InvalidArgumentError(f"mode must be one of: {allowed}")
return canonical
def _normalize_feature_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 surface-flow path objects"
)
if mode not in _FEATURE_COLORING:
raise InvalidArgumentError("coloring must be geometric or scalar")
return mode
def _normalize_feature_line_type(value: constant.LineType | str) -> str:
if isinstance(value, constant.LineType):
return value.value
line_type = str(value).strip().lower()
if line_type not in _FEATURE_LINE_TYPES:
raise InvalidArgumentError("line_type must be thin, medium, or thick")
return line_type
def _normalize_direction(value: constant.CalculationDirection | str) -> str:
if isinstance(value, constant.CalculationDirection):
return value.value
direction = str(value).strip().lower()
if direction in {
constant.CalculationDirection.FORWARD.value,
constant.CalculationDirection.BACKWARD.value,
constant.CalculationDirection.BOTH.value,
}:
return direction
raise InvalidArgumentError("direction must be forward, backward, or both")
def _normalize_seeding(value: constant.SurfaceFlowSeeding | str) -> str:
if isinstance(value, constant.SurfaceFlowSeeding):
return value.value
canonical = _SURFACE_FLOW_SEEDING.get(str(value).strip().lower().replace("_", " "))
if canonical is None:
allowed = ", ".join(sorted({v for v in _SURFACE_FLOW_SEEDING.values()}))
raise InvalidArgumentError(f"seeding must be one of: {allowed}")
return canonical
def _feature_state_kwargs(state: dict[str, object]) -> _FeatureStateKwargs:
return {
"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")
),
"coloring": cast(constant.Coloring | str | None, state.get("coloring")),
"colormap": cast(
Colormap | dict[str, object] | None, state.get("scalar_colormap")
),
"legend": cast(Legend | dict[str, object] | None, state.get("legend")),
}
def _require_dataset(dataset: Dataset | None, label: str) -> Dataset:
if dataset is None:
raise InvalidDatasetError(f"{label} 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, label)
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, label)
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_surface_flow_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", "surface flows"
)
if dataset_name is not None:
return dataset_name
canonical = _SURFACE_FLOW_PATH_VARIABLES.get(name.lower().replace("_", " "))
if canonical is not None:
return canonical
allowed = ["Duration", "Wall Shear Magnitude"]
raise InvalidArgumentError(
"scalar_path_variable must be a dataset scalar or one of: " + ", ".join(allowed)
)
def _build_path_visual_payload(
*,
visibility: bool | None = None,
line_type: constant.LineType | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
coloring: constant.Coloring | str | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
) -> dict[str, object]:
payload: dict[str, object] = {}
if visibility is not None:
payload["visibility"] = bool(visibility)
if line_type is not None:
payload["line_type"] = _normalize_feature_line_type(line_type)
if geometric_color is not None:
payload["geometric_color"] = int(geometric_color)
if coloring is not None:
payload["coloring"] = _normalize_feature_coloring(coloring)
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 _build_surface_flows_payload(
*,
dataset: Dataset | None,
mode: object = _UNSET,
visibility: bool | None = None,
line_type: constant.LineType | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
coloring: constant.Coloring | str | None = None,
scalar_func: str | None = None,
scalar_path_variable: str | None = None,
vector_func: str | None = None,
direction: constant.CalculationDirection | str | None = None,
time_limit_enabled: bool | None = None,
time_limit: float | None | object = _UNSET,
seeding: constant.SurfaceFlowSeeding | str | None = None,
seed_file: str | os.PathLike[str] | None | object = _UNSET,
offset_distance: float | None = None,
step: int | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
) -> dict[str, object]:
payload = _build_path_visual_payload(
visibility=visibility,
line_type=line_type,
geometric_color=geometric_color,
coloring=coloring,
colormap=colormap,
legend=legend,
)
if mode is not _UNSET:
if not isinstance(mode, (constant.SurfaceFlowMode, str)):
raise InvalidArgumentError("mode must be a SurfaceFlowMode or string")
payload["mode"] = _normalize_mode(mode)
if scalar_path_variable is not None:
payload["scalar_path_variable"] = _normalize_surface_flow_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", "surface flows"
)
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:
stripped = vector_func.strip()
if not stripped:
raise InvalidArgumentError("vector_func cannot be empty")
if stripped.lower() == "none":
payload["vector_func"] = "none"
else:
func_id = _lookup_function_id(
dataset, stripped, "_vector_function_ids", "surface flows"
)
if func_id is not None:
payload["vector_func_id"] = func_id
else:
payload["vector_func_name"] = stripped
if direction is not None:
payload["direction"] = _normalize_direction(direction)
if time_limit_enabled is not None:
payload["time_limit_enabled"] = bool(time_limit_enabled)
if time_limit is not _UNSET:
if time_limit is None:
payload["time_limit"] = None
else:
try:
payload["time_limit"] = _coerce_float(time_limit)
except (TypeError, ValueError) as exc:
raise InvalidArgumentError(
"time_limit must be a float or None"
) from exc
if seeding is not None:
payload["seeding"] = _normalize_seeding(seeding)
if seed_file is not _UNSET:
if seed_file is None:
payload["seed_file"] = None
else:
normalized_seed_file = _coerce_pathlike_str(seed_file, "seed_file")
if not normalized_seed_file.strip():
raise InvalidArgumentError("seed_file cannot be empty")
payload["seed_file"] = normalized_seed_file
if offset_distance is not None:
try:
payload["offset_distance"] = _coerce_float(offset_distance)
except (TypeError, ValueError) as exc:
raise InvalidArgumentError("offset_distance must be a float") from exc
if step is not None:
payload["step"] = _coerce_int(step)
return payload
def _surface_flow_kwargs_from_state(
state: dict[str, object],
) -> _SurfaceFlowStateKwargs:
mode = state.get("mode", constant.SurfaceFlowMode.NO_SLIP.value)
include_offset_distance = str(mode).lower() == constant.SurfaceFlowMode.OFFSET.value
feature_kwargs = _feature_state_kwargs(state)
offset_distance = state.get("offset_distance")
step = state.get("step")
seed_file = state.get("seed_file")
seed_file_value = None
if isinstance(seed_file, str):
seed_file_value = seed_file.strip() or None
return {
"visibility": feature_kwargs["visibility"],
"line_type": feature_kwargs["line_type"],
"geometric_color": feature_kwargs["geometric_color"],
"coloring": feature_kwargs["coloring"],
"colormap": feature_kwargs["colormap"],
"legend": feature_kwargs["legend"],
"mode": str(mode),
"scalar_func": cast(str | None, state.get("scalar_func_name") or None),
"scalar_path_variable": cast(
str | None, state.get("scalar_path_variable") or None
),
"vector_func": cast(str | None, state.get("vector_func_name") or None),
"direction": cast(
constant.CalculationDirection | str | None, state.get("direction")
),
"time_limit_enabled": cast(bool | None, state.get("time_limit_enabled")),
"time_limit": state.get("time_limit"),
"seeding": cast(constant.SurfaceFlowSeeding | str | None, state.get("seeding")),
"seed_file": seed_file_value,
"offset_distance": None
if not include_offset_distance or offset_distance is None
else _coerce_float(offset_distance),
"step": None if step is None else _coerce_int(step),
}
[docs]
class SurfaceFlows(PathBase):
"""Surface restricted flow path 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"),
... )
>>> sf = fv.vis.create_surface_flows(
... ds,
... mode=fv.constant.SurfaceFlowMode.NO_SLIP,
... vector_func="Velocity Vectors [PLOT3D]",
... )
>>> sf.direction = fv.constant.CalculationDirection.BOTH
>>> sf.calculate()
"""
_core_prefix = "surface_flows"
__slots__ = ("_dataset", "_mode")
def __init__(
self,
phigs_obj: int,
dataset_id: int,
dataset: Dataset | None = None,
mode: constant.SurfaceFlowMode | str | None = None,
) -> None:
"""Create a surface-flows 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.
mode: Optional surface-flow extraction mode to cache locally.
"""
super().__init__(phigs_obj, dataset_id)
self._dataset = dataset
self._mode = _normalize_mode(mode) if mode is not None else None
def _state(self) -> dict[str, object]:
return _core_call(_core.surface_flows_get_state, self._phigs_obj)
@property
def display_type(self) -> constant.PathsDisplayType:
"""`PathsDisplayType`: Display representation for surface flows.
Only ``COMPLETE`` is supported.
"""
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(
"surface flows only support display_type='complete'"
)
@property
def mode(self) -> constant.SurfaceFlowMode:
"""`SurfaceFlowMode`: Surface-flow extraction mode."""
value = self._state().get(
"mode", self._mode or constant.SurfaceFlowMode.NO_SLIP.value
)
return constant.SurfaceFlowMode(str(value).lower())
@mode.setter
def mode(self, value: constant.SurfaceFlowMode | str) -> None:
new_mode = _normalize_mode(value)
if new_mode == self.mode.value:
return
self._rebuild(mode=new_mode)
@property
def scalar_func(self) -> str | None:
"""`Optional[str]`: Dataset scalar function cached for registry loading."""
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":
self._modify({"scalar_func": "none"})
return
func_id = _lookup_function_id(
self._dataset, name, "_scalar_function_ids", "surface flows"
)
if func_id is None:
raise InvalidArgumentError(
"scalar_func must be a dataset scalar function; use scalar_path_variable for path variables"
)
self._modify({"scalar_func_id": func_id})
@property
def scalar_path_variable(self) -> str | None:
"""`Optional[str]`: Active scalar coloring selection for dataset or feature variables."""
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:
self._modify(
{
"scalar_path_variable": _normalize_surface_flow_scalar_path_variable(
self._dataset, value
)
}
)
@property
def vector_func(self) -> str | None:
"""`Optional[str]`: Vector function used for the path calculation."""
name = str(self._state().get("vector_func_name", "")).strip()
return name or None
@vector_func.setter
def vector_func(self, value: str) -> None:
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":
self._modify({"vector_func": "none"})
return
func_id = _lookup_function_id(
self._dataset, name, "_vector_function_ids", "surface flows"
)
if func_id is not None:
self._modify({"vector_func_id": func_id})
else:
self._modify({"vector_func_name": name})
@property
def direction(self) -> constant.CalculationDirection:
"""`CalculationDirection`: Integration direction."""
value = str(
self._state().get("direction", constant.CalculationDirection.FORWARD.value)
).lower()
return constant.CalculationDirection(value)
@direction.setter
def direction(self, value: constant.CalculationDirection | str) -> None:
self._modify({"direction": _normalize_direction(value)})
@property
def time_limit_enabled(self) -> bool:
"""`bool`: Enable the duration-limit field."""
return bool(self._state().get("time_limit_enabled", False))
@time_limit_enabled.setter
def time_limit_enabled(self, value: bool) -> None:
self._modify({"time_limit_enabled": bool(value)})
@property
def time_limit(self) -> float | None:
"""`Optional[float]`: Duration-limit value, or ``None`` when disabled."""
value = self._state().get("time_limit")
return None if value is None else _coerce_float(value)
@time_limit.setter
def time_limit(self, value: float | None) -> None:
if value is None:
self._modify({"time_limit": None})
else:
self._modify({"time_limit": float(value)})
@property
def seeding(self) -> constant.SurfaceFlowSeeding:
"""`SurfaceFlowSeeding`: Seeding density/source mode."""
value = str(
self._state().get("seeding", constant.SurfaceFlowSeeding.LOW_DENSITY.value)
).lower()
return constant.SurfaceFlowSeeding(value)
@seeding.setter
def seeding(self, value: constant.SurfaceFlowSeeding | str) -> None:
self._modify({"seeding": _normalize_seeding(value)})
@property
def seed_file(self) -> str | None:
"""`Optional[str]`: Seed-file path when ``seeding`` is ``FROM_FILE``."""
value = self._state().get("seed_file")
if isinstance(value, str):
stripped = value.strip()
return stripped or None
return None
@seed_file.setter
def seed_file(self, value: str | os.PathLike[str] | None) -> None:
if value is None:
self._modify({"seed_file": None})
else:
normalized = _coerce_pathlike_str(value, "seed_file")
if not normalized.strip():
raise InvalidArgumentError("seed_file cannot be empty")
self._modify({"seed_file": normalized})
@property
def offset_distance(self) -> float:
"""`float`: Surface offset distance used by OFFSET mode."""
return _coerce_float(self._state().get("offset_distance", 0.0))
@offset_distance.setter
def offset_distance(self, value: float) -> None:
self._modify({"offset_distance": float(value)})
@property
def step(self) -> int:
"""`int`: Integration step count."""
return _coerce_int(self._state().get("step", 1))
@step.setter
def step(self, value: int) -> None:
self._modify({"step": int(value)})
[docs]
def calculate(self) -> None:
"""Calculate or recalculate the current surface-flow paths."""
_core_call(_core.surface_flows_calculate, self._phigs_obj)
def _rebuild(
self,
*,
mode: constant.SurfaceFlowMode | str,
visibility: bool | None = None,
line_type: constant.LineType | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
coloring: constant.Coloring | str | None = None,
scalar_func: str | None = None,
scalar_path_variable: str | None = None,
vector_func: str | None = None,
direction: constant.CalculationDirection | str | None = None,
time_limit_enabled: bool | None = None,
time_limit: object = _UNSET,
seeding: constant.SurfaceFlowSeeding | str | None = None,
seed_file: object = _UNSET,
offset_distance: float | None = None,
step: int | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
calculate: bool = False,
) -> None:
dataset = _require_dataset(self._dataset, "surface flows")
state = _surface_flow_kwargs_from_state(self._state())
new_surf = create_surface_flows(
dataset,
mode=mode,
visibility=state["visibility"] if visibility is None else visibility,
line_type=state["line_type"] if line_type is None else line_type,
geometric_color=state["geometric_color"]
if geometric_color is None
else geometric_color,
coloring=state["coloring"] if coloring is None else coloring,
scalar_func=state["scalar_func"] if scalar_func is None else scalar_func,
scalar_path_variable=(
state["scalar_path_variable"]
if scalar_path_variable is None
else scalar_path_variable
),
vector_func=state["vector_func"] if vector_func is None else vector_func,
direction=state["direction"] if direction is None else direction,
time_limit_enabled=state["time_limit_enabled"]
if time_limit_enabled is None
else time_limit_enabled,
time_limit=state["time_limit"] if time_limit is _UNSET else time_limit,
seeding=state["seeding"] if seeding is None else seeding,
seed_file=state["seed_file"] if seed_file is _UNSET else seed_file,
offset_distance=state["offset_distance"]
if offset_distance is None
else offset_distance,
step=state["step"] if step is None else step,
colormap=state["colormap"] if colormap is None else colormap,
legend=state["legend"] if legend is None else legend,
calculate=calculate,
)
old_phigs_obj = self._phigs_obj
self._phigs_obj = new_surf._phigs_obj
self._mode = new_surf._mode
_core_call(_core.surface_flows_delete, old_phigs_obj)
[docs]
def modify(
self,
*,
mode: object = _UNSET,
visibility: bool | None = None,
line_type: constant.LineType | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
coloring: constant.Coloring | str | None = None,
scalar_func: str | None = None,
scalar_path_variable: str | None = None,
vector_func: str | None = None,
direction: constant.CalculationDirection | str | None = None,
time_limit_enabled: bool | None = None,
time_limit: float | None | object = _UNSET,
seeding: constant.SurfaceFlowSeeding | str | None = None,
seed_file: str | os.PathLike[str] | None | object = _UNSET,
offset_distance: float | None = None,
step: int | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
calculate: bool = False,
) -> None:
"""Modify multiple surface-flow settings in one call.
Set ``calculate=True`` to run the calculation after the settings update.
Args:
mode: Optional surface-flow extraction mode.
visibility: Whether the object is visible.
line_type: Line style used for the extracted paths.
geometric_color: Geometric color id used when geometric coloring is
active.
coloring: Coloring mode for the extracted paths.
scalar_func: Scalar function name used for scalar coloring.
scalar_path_variable: Path-variable name used for scalar coloring.
vector_func: Vector function name used for path calculation.
direction: Integration direction.
time_limit_enabled: Whether the time-limit field is enabled.
time_limit: Optional tracing time limit.
seeding: Seeding mode.
seed_file: Optional seed-file path.
offset_distance: Surface offset distance.
step: Integration step count.
colormap: Scalar colormap options.
legend: Legend options.
calculate: When ``True``, run the calculation after updating the
settings.
"""
payload = _build_surface_flows_payload(
dataset=self._dataset,
mode=mode,
visibility=visibility,
line_type=line_type,
geometric_color=geometric_color,
coloring=coloring,
scalar_func=scalar_func,
scalar_path_variable=scalar_path_variable,
vector_func=vector_func,
direction=direction,
time_limit_enabled=time_limit_enabled,
time_limit=time_limit,
seeding=seeding,
seed_file=seed_file,
offset_distance=offset_distance,
step=step,
colormap=colormap,
legend=legend,
)
if not payload and not calculate:
raise InvalidArgumentError("modify requires at least one property")
payload_mode = payload.pop("mode", None)
if payload_mode is not None and payload_mode != self.mode.value:
self._rebuild(
mode=cast(constant.SurfaceFlowMode | str, payload_mode),
visibility=visibility,
line_type=line_type,
geometric_color=geometric_color,
coloring=coloring,
scalar_func=scalar_func,
scalar_path_variable=scalar_path_variable,
vector_func=vector_func,
direction=direction,
time_limit_enabled=time_limit_enabled,
time_limit=time_limit,
seeding=seeding,
seed_file=seed_file,
offset_distance=offset_distance,
step=step,
colormap=colormap,
legend=legend,
calculate=calculate,
)
return
if payload:
_core_call(_core.surface_flows_modify, self._phigs_obj, payload)
if calculate:
self.calculate()
[docs]
def copy(self) -> "SurfaceFlows":
"""Return a new surface-flow object with the same public settings."""
dataset = _require_dataset(self._dataset, "surface flows")
return create_surface_flows(
dataset, **_surface_flow_kwargs_from_state(self._state())
)
[docs]
def delete(self) -> None:
"""Delete this surface-flow object."""
self._delete_surface(_core.surface_flows_delete)
[docs]
def create_surface_flows(
dataset: Dataset | None = None,
*,
mode: constant.SurfaceFlowMode | str,
visibility: bool | None = None,
line_type: constant.LineType | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
coloring: constant.Coloring | str | None = None,
scalar_func: str | None = None,
scalar_path_variable: str | None = None,
vector_func: str | None = None,
direction: constant.CalculationDirection | str | None = None,
time_limit_enabled: bool | None = None,
time_limit: float | None | object = _UNSET,
seeding: constant.SurfaceFlowSeeding | str | None = None,
seed_file: str | os.PathLike[str] | None | object = _UNSET,
offset_distance: float | None = None,
step: int | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
calculate: bool = False,
) -> SurfaceFlows:
"""Create a surface restricted flow path object.
Args:
dataset: Dataset to attach the new feature object to. When omitted,
the current dataset is used.
mode: Surface-flow extraction mode.
visibility: Whether the object is initially visible.
line_type: Line style used for the extracted paths.
geometric_color: Geometric color id used when geometric coloring is
active.
coloring: Coloring mode for the extracted paths.
scalar_func: Scalar function name used for scalar coloring.
scalar_path_variable: Path-variable name used for scalar coloring.
vector_func: Vector function name used for path calculation.
direction: Integration direction.
time_limit_enabled: Whether the time-limit field is enabled.
time_limit: Optional tracing time limit.
seeding: Seeding mode.
seed_file: Optional seed-file path.
offset_distance: Surface offset distance.
step: Integration step count.
colormap: Scalar colormap options.
legend: Legend options.
calculate: When ``True``, run the calculation after creation.
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"),
... )
>>> sf = fv.vis.create_surface_flows(
... ds,
... mode=fv.constant.SurfaceFlowMode.NO_SLIP,
... vector_func="Velocity Vectors [PLOT3D]",
... direction=fv.constant.CalculationDirection.BOTH,
... )
>>> sf.calculate()
"""
ds = _resolve_dataset_or_current(dataset)
canonical_mode = _normalize_mode(mode)
payload = _build_surface_flows_payload(
dataset=ds,
visibility=visibility,
line_type=line_type,
geometric_color=geometric_color,
coloring=coloring,
scalar_func=scalar_func,
scalar_path_variable=scalar_path_variable,
vector_func=vector_func,
direction=direction,
time_limit_enabled=time_limit_enabled,
time_limit=time_limit,
seeding=seeding,
seed_file=seed_file,
offset_distance=offset_distance,
step=step,
colormap=colormap,
legend=legend,
)
phigs_obj = _core_call(_core.surface_flows_create, ds.dataset_id, canonical_mode)
surf = SurfaceFlows(
phigs_obj=phigs_obj, dataset_id=ds.dataset_id, dataset=ds, mode=canonical_mode
)
if payload or calculate:
try:
if payload:
_core_call(_core.surface_flows_modify, phigs_obj, payload)
if calculate:
surf.calculate()
except Exception:
_core_call(_core.surface_flows_delete, phigs_obj)
raise
return surf
