"""Streamlines 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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
"""
from __future__ import annotations
import os
from typing import cast
from collections.abc import Sequence
from .. import _core
from .. import constant
from .._core_utils import _core_call
from .._base import (
PathBase,
SurfaceBase,
_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] = ["StreamlinesStreak", "Streamlines", "create_streamlines"]
_UNSET = object()
_STREAMLINES_DISPLAY_TYPES: set[str] = {
constant.PathsDisplayType.COMPLETE.value,
constant.PathsDisplayType.FILAMENT.value,
constant.PathsDisplayType.FILAMENT_ARROWS.value,
constant.PathsDisplayType.FILAMENT_SPHERES.value,
constant.PathsDisplayType.GROWING.value,
constant.PathsDisplayType.SPHERES_AND_LINES.value,
constant.PathsDisplayType.SPHERES.value,
constant.PathsDisplayType.DOTS.value,
constant.PathsDisplayType.LINES_OF_SPHERES.value,
constant.PathsDisplayType.LINES_OF_DOTS.value,
constant.PathsDisplayType.RIBBONS.value,
}
def _normalize_seed_coord(value: constant.StreamlinesSeedCoord | str) -> str:
if isinstance(value, constant.StreamlinesSeedCoord):
return value.value
normalized = str(value).strip().lower()
if normalized in {"ijk", "ijk_int"}:
return constant.StreamlinesSeedCoord.IJK.value
if normalized == constant.StreamlinesSeedCoord.IJK_REAL.value:
return normalized
if normalized == constant.StreamlinesSeedCoord.XYZ.value:
return normalized
raise InvalidArgumentError(
"seed_coord must be StreamlinesSeedCoord.IJK, StreamlinesSeedCoord.IJK_REAL, or StreamlinesSeedCoord.XYZ"
)
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_animate_direction(value: constant.AnimationDirection | str) -> str:
if isinstance(value, constant.AnimationDirection):
return value.value
direction = str(value).strip().lower()
if direction in {
constant.AnimationDirection.FORWARD.value,
constant.AnimationDirection.BACKWARD.value,
}:
return direction
raise InvalidArgumentError("animate_direction must be forward or backward")
def _normalize_streamlines_coloring(value: constant.Coloring | str) -> str:
mode = (
value.value
if isinstance(value, constant.Coloring)
else str(value).strip().lower()
)
if mode == constant.Coloring.MATERIAL.value:
raise InvalidArgumentError("material coloring is not supported for streamlines")
if mode not in {constant.Coloring.GEOMETRIC.value, constant.Coloring.SCALAR.value}:
raise InvalidArgumentError("coloring must be geometric or scalar")
return mode
def _normalize_line_type(value: constant.LineType | str) -> str:
line_type = (
value.value
if isinstance(value, constant.LineType)
else str(value).strip().lower()
)
if line_type not in {
constant.LineType.THIN.value,
constant.LineType.MEDIUM.value,
constant.LineType.THICK.value,
}:
raise InvalidArgumentError("line_type must be thin, medium, or thick")
return line_type
def _normalize_streamlines_display_type(
value: constant.PathsDisplayType | str,
) -> str:
display_type = (
value.value if isinstance(value, constant.PathsDisplayType) else str(value)
)
normalized = display_type.strip().lower()
if normalized in _STREAMLINES_DISPLAY_TYPES:
return normalized
allowed = ", ".join(sorted(_STREAMLINES_DISPLAY_TYPES))
raise InvalidArgumentError(f"display_type must be one of: {allowed}")
def _to_seed_triplet(
seed: Sequence[int | float], *, integral: bool
) -> list[int | float]:
if not isinstance(seed, Sequence) or len(seed) != 3:
raise InvalidArgumentError(
"each seed must be a sequence of exactly 3 numeric values"
)
values: list[int | float] = []
for item in seed:
if isinstance(item, bool) or not isinstance(item, (int, float)):
raise InvalidArgumentError("seed values must be numeric")
values.append(int(item) if integral else float(item))
return values
def _normalize_streak_int(value: object, label: str) -> int:
if isinstance(value, bool):
raise InvalidArgumentError(f"{label} must be an integer")
try:
return _coerce_int(value)
except (TypeError, ValueError) as exc:
raise InvalidArgumentError(f"{label} must be an integer") from exc
def _validate_streak_pair(release_interval: int, duration: int) -> tuple[int, int]:
if release_interval <= 0:
raise InvalidArgumentError("release_interval must be > 0")
if duration <= 0 or duration >= release_interval:
raise InvalidArgumentError("duration must be > 0 and < release_interval")
return release_interval, duration
def _resolve_streak_pair(
*,
current_release_interval: int,
current_duration: int,
release_interval: object = _UNSET,
duration: object = _UNSET,
) -> tuple[int | None, int | None]:
if release_interval is _UNSET and duration is _UNSET:
return None, None
next_release_interval = (
current_release_interval
if release_interval is _UNSET
else _normalize_streak_int(release_interval, "release_interval")
)
next_duration = (
current_duration
if duration is _UNSET
else _normalize_streak_int(duration, "duration")
)
_validate_streak_pair(next_release_interval, next_duration)
return (
None if release_interval is _UNSET else next_release_interval,
None if duration is _UNSET else next_duration,
)
class StreamlinesStreak:
"""Streakline controls bound to a :class:`Streamlines` surface.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> streak = sl.streak
>>> streak.duration = 2
"""
__slots__ = ("_parent",)
def __init__(self, parent: "Streamlines") -> None:
"""Create a convenience wrapper for streak controls.
Args:
parent: Owning :class:`Streamlines` object.
"""
self._parent = parent
@property
def duration(self) -> int:
"""`int`: Streak duration in frames.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> streak = sl.streak
>>> streak.duration = 2
"""
return self._parent.duration
@duration.setter
def duration(self, value: int) -> None:
self._parent.duration = value
[docs]
class Streamlines(PathBase):
"""Streamlines surface wrapper.
Notes:
``seed_coord`` cannot be changed while the streamline object still has
seeds. Delete them first with :meth:`delete_all_seeds`.
:meth:`seed_surface` requires a live ``Boundary``, ``Comp``, ``Coord``,
or ``Iso`` surface from the same dataset as the streamline object.
Streak controls use a single validation rule:
``release_interval > 0`` and ``0 < duration < release_interval``.
Invalid values are rejected before any core update, and
:meth:`copy` raises if the source object violates that contract.
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"),
... )
>>> sl = fv.vis.create_streamlines(
... ds,
... vector_func=ds.vector_functions[0],
... coloring=fv.constant.Coloring.SCALAR,
... scalar_func=ds.scalar_functions[0],
... display_type=fv.constant.PathsDisplayType.FILAMENT,
... seed_coord=fv.constant.StreamlinesSeedCoord.IJK_REAL,
... )
>>> sl.add_seeds([(1.0, 1.0, 1.0), (40.0, 20.0, 10.0)], grid=1)
>>> sl.calculate()
"""
_core_prefix = "streamlines_surf"
__slots__ = ("_dataset", "_streak")
def __init__(
self, phigs_obj: int, dataset_id: int, dataset: Dataset | None = None
) -> None:
"""Create a streamlines wrapper around an existing host object.
Args:
phigs_obj: Host PHIGS object identifier for the streamlines surface.
dataset_id: Host dataset identifier backing the surface.
dataset: Optional live :class:`Dataset` wrapper for function lookup
and validation.
"""
super().__init__(phigs_obj, dataset_id)
self._dataset = dataset
self._streak = StreamlinesStreak(self)
def _state(self, *, include_seeds: bool = False) -> dict[str, object]:
return _core_call(
_core.streamlines_surf_get_state, self._phigs_obj, bool(include_seeds)
)
def _current_streak_pair(self) -> tuple[int, int]:
state = self._state()
release_interval = _normalize_streak_int(
state.get("release_interval", 3), "release_interval"
)
duration = _normalize_streak_int(state.get("duration", 1), "duration")
return release_interval, duration
def _require_dataset(self) -> Dataset:
dataset = self._dataset
if dataset is None:
raise InvalidDatasetError(
"streamlines surface is missing a dataset reference"
)
dataset._ensure_valid()
return dataset
def _lookup_function_id(
self, name: str, mapping: dict[str, int], label: str
) -> int:
if not isinstance(name, str):
raise InvalidArgumentError(f"{label} function name must be a string")
needle = name.strip()
if not needle:
raise InvalidArgumentError(f"{label} function name cannot be empty")
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} function not found: {needle}")
@property
def display_type(self) -> constant.PathsDisplayType:
"""`PathsDisplayType`: Path display style.
Example usage:
.. code-block:: python
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.display_type = fv.constant.PathsDisplayType.FILAMENT
"""
return constant.PathsDisplayType(
str(
self._state().get(
"display_type", constant.PathsDisplayType.COMPLETE.value
)
).lower()
)
@display_type.setter
def display_type(self, value: constant.PathsDisplayType | str) -> None:
self._modify({"display_type": _normalize_streamlines_display_type(value)})
def _seed_entries_from_state(
self, state: dict[str, object] | None = None
) -> list[dict[str, object]]:
src = self._state(include_seeds=True) if state is None else state
entries = src.get("seeds", [])
if not isinstance(entries, list):
return []
return [
dict(cast(dict[str, object], entry))
for entry in cast(list[object], entries)
if isinstance(entry, dict)
]
def _seed_payload_from_entries(
self,
entries: Sequence[dict[str, object]],
seed_coord: str,
) -> tuple[list[list[int | float]], list[int]]:
coords: list[list[int | float]] = []
grids: list[int] = []
for entry in entries:
if seed_coord == constant.StreamlinesSeedCoord.XYZ.value:
xyz = cast(Sequence[int | float], entry.get("xyz", []))
coords.append(_to_seed_triplet(xyz, integral=False))
elif seed_coord == constant.StreamlinesSeedCoord.IJK.value:
ijk = cast(Sequence[int | float], entry.get("ijk", []))
coords.append(_to_seed_triplet(ijk, integral=True))
grids.append(_coerce_int(entry.get("grid", 1)))
else:
ijk = cast(Sequence[int | float], entry.get("ijk", []))
coords.append(_to_seed_triplet(ijk, integral=False))
grids.append(_coerce_int(entry.get("grid", 1)))
return coords, grids
def _rebuild_with_seed_entries(
self,
state: dict[str, object],
seed_entries: Sequence[dict[str, object]],
*,
seed_coord: str,
) -> None:
dataset = self._require_dataset()
vector_func_name = str(state.get("vector_func_name", "")).strip()
if not vector_func_name:
raise InvalidArgumentError(
"streamlines surface has no vector_func; cannot rebuild"
)
new_surf = create_streamlines(
dataset,
coloring=cast(constant.Coloring | str | None, state.get("coloring")),
geometric_color=cast(
constant.GeometricColor | int | None,
state.get("geometric_color"),
),
line_type=cast(constant.LineType | str | None, state.get("line_type")),
display_type=cast(
constant.PathsDisplayType | str | None,
state.get("display_type"),
),
sphere_scale=None
if state.get("sphere_scale") is None
else _coerce_float(state.get("sphere_scale")),
ribbon_width=None
if state.get("ribbon_width") is None
else _coerce_int(state.get("ribbon_width")),
transparency=None
if state.get("transparency") is None
else _coerce_float(state.get("transparency")),
visibility=cast(bool | None, state.get("visibility")),
scalar_func=cast(str | None, state.get("scalar_func_name") or None),
vector_func=vector_func_name,
animate=cast(bool | None, state.get("animate")),
animate_direction=cast(
constant.AnimationDirection | str | None,
state.get("animate_direction"),
),
animate_divs=None
if state.get("animate_divs") is None
else _coerce_int(state.get("animate_divs")),
colormap=cast(
Colormap | dict[str, object] | None,
state.get("scalar_colormap"),
),
legend=cast(Legend | dict[str, object] | None, state.get("legend")),
defer_apply=True,
calculate=False,
)
try:
release_interval = _normalize_streak_int(
state.get("release_interval", 3), "release_interval"
)
duration = _normalize_streak_int(state.get("duration", 1), "duration")
release_interval, duration = _validate_streak_pair(
release_interval, duration
)
calc_payload: dict[str, object] = {
"seed_coord": seed_coord,
"show_seeds": bool(state.get("show_seeds", True)),
"direction": str(
state.get("direction", constant.CalculationDirection.FORWARD.value)
),
"step": _coerce_int(state.get("step", 3)),
"time_limit": state.get("time_limit"),
"release_interval": release_interval,
"duration": duration,
}
_core_call(
_core.streamlines_surf_modify,
new_surf._phigs_obj,
calc_payload,
)
coords, grids = self._seed_payload_from_entries(seed_entries, seed_coord)
if coords:
if seed_coord == constant.StreamlinesSeedCoord.XYZ.value:
new_surf.add_seeds(coords)
else:
new_surf.add_seeds(coords, grids=grids)
except Exception:
new_surf.delete()
raise
old_phigs_obj = self._phigs_obj
self._phigs_obj = new_surf._phigs_obj
_core_call(_core.streamlines_surf_delete, old_phigs_obj)
@property
def sphere_scale(self) -> float:
"""`float`: Sphere/arrow size scale.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.sphere_scale = 1.25
"""
return _coerce_float(self._state().get("sphere_scale", 1.0))
@sphere_scale.setter
def sphere_scale(self, value: float) -> None:
"""Set sphere/arrow size scale."""
_core_call(
_core.streamlines_surf_modify,
self._phigs_obj,
{"sphere_scale": float(value)},
)
@property
def ribbon_width(self) -> int:
"""`int`: Ribbon width for ribbon display mode.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.ribbon_width = 24
"""
return _coerce_int(self._state().get("ribbon_width", 32))
@ribbon_width.setter
def ribbon_width(self, value: int) -> None:
"""Set ribbon width."""
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"ribbon_width": int(value)}
)
@property
def transparency(self) -> float:
"""`float`: Transparency percentage.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.transparency = 0.2
"""
return _coerce_float(self._state().get("transparency", 0.0))
@transparency.setter
def transparency(self, value: float) -> None:
"""Set transparency percentage."""
_core_call(
_core.streamlines_surf_modify,
self._phigs_obj,
{"transparency": float(value)},
)
@property
def scalar_func(self) -> str | None:
"""`str | None`: Selected scalar function name.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.scalar_func = "Density (Q1)"
"""
name = str(self._state().get("scalar_func_name", ""))
return name or None
@scalar_func.setter
def scalar_func(self, value: str) -> None:
"""Set scalar function by name (case-insensitive)."""
dataset = self._require_dataset()
func_id = self._lookup_function_id(
value, dataset._scalar_function_ids, "scalar"
)
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"scalar_func_id": func_id}
)
@property
def vector_func(self) -> str | None:
"""`str | None`: Selected vector function name.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.vector_func = "Momentum Vectors [PLOT3D]"
"""
name = str(self._state().get("vector_func_name", ""))
return name or None
@vector_func.setter
def vector_func(self, value: str) -> None:
"""Set vector function by name (case-insensitive)."""
dataset = self._require_dataset()
func_id = self._lookup_function_id(
value, dataset._vector_function_ids, "vector"
)
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"vector_func_id": func_id}
)
@property
def animate(self) -> bool:
"""`bool`: Animate path traversal in the viewer.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.animate = True
"""
return bool(self._state().get("animate", False))
@animate.setter
def animate(self, value: bool) -> None:
"""Enable or disable streamline animation."""
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"animate": bool(value)}
)
@property
def animate_direction(self) -> constant.AnimationDirection:
"""`AnimationDirection`: Direction of streamline animation.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.animate_direction = "forward"
"""
value = str(
self._state().get(
"animate_direction", constant.AnimationDirection.FORWARD.value
)
).lower()
return constant.AnimationDirection(value)
@animate_direction.setter
def animate_direction(self, value: constant.AnimationDirection | str) -> None:
"""Set animation direction."""
_core_call(
_core.streamlines_surf_modify,
self._phigs_obj,
{"animate_direction": _normalize_animate_direction(value)},
)
@property
def animate_divs(self) -> int:
"""`int`: Number of animation divisions.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.animate_divs = 30
"""
return _coerce_int(self._state().get("animate_divs", 25))
@animate_divs.setter
def animate_divs(self, value: int) -> None:
"""Set animation divisions."""
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"animate_divs": int(value)}
)
@property
def show_seeds(self) -> bool:
"""`bool`: Show or hide seed points in the viewer.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.show_seeds = True
"""
return bool(self._state().get("show_seeds", True))
@show_seeds.setter
def show_seeds(self, value: bool) -> None:
"""Set seed visibility."""
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"show_seeds": bool(value)}
)
@property
def seed_coord(self) -> constant.StreamlinesSeedCoord:
"""`StreamlinesSeedCoord`: Coordinate system used by seed operations.
Changing `seed_coord` while seeds exist is rejected by the core.
Delete existing seeds first with :meth:`delete_all_seeds`.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.seed_coord
"""
value = str(
self._state().get(
"seed_coord", constant.StreamlinesSeedCoord.IJK_REAL.value
)
).lower()
return constant.StreamlinesSeedCoord(value)
@seed_coord.setter
def seed_coord(self, value: constant.StreamlinesSeedCoord | str) -> None:
"""Set seed coordinate type (`ijk`, `ijk_real`, or `xyz`).
Changing `seed_coord` while seeds exist is rejected by the core.
Delete existing seeds first with :meth:`delete_all_seeds`.
"""
target = _normalize_seed_coord(value)
state = self._state()
current = str(
state.get("seed_coord", constant.StreamlinesSeedCoord.IJK_REAL.value)
).lower()
if target == current:
return
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"seed_coord": target}
)
@property
def num_seeds(self) -> int:
"""`int`: Number of seeds currently assigned (read-only).
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.num_seeds
"""
return _coerce_int(self._state().get("num_seeds", 0))
@property
def direction(self) -> constant.CalculationDirection:
"""`CalculationDirection`: Trace direction for calculation.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.direction = "both"
"""
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:
"""Set trace direction (`forward`, `backward`, or `both`)."""
_core_call(
_core.streamlines_surf_modify,
self._phigs_obj,
{"direction": _normalize_direction(value)},
)
@property
def step(self) -> int:
"""`int`: Integration step size.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.step = 3
"""
return _coerce_int(self._state().get("step", 3))
@step.setter
def step(self, value: int) -> None:
"""Set integration step size."""
_core_call(_core.streamlines_surf_modify, self._phigs_obj, {"step": int(value)})
@property
def time_limit(self) -> float | None:
"""`float | None`: Optional time limit for tracing.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.time_limit = 0.25
"""
value = self._state().get("time_limit")
if value is None:
return None
return _coerce_float(value)
@time_limit.setter
def time_limit(self, value: float | None) -> None:
"""Set or clear (`None`) the tracing time limit."""
payload_value: float | None
if value is None:
payload_value = None
else:
payload_value = float(value)
_core_call(
_core.streamlines_surf_modify,
self._phigs_obj,
{"time_limit": payload_value},
)
@property
def release_interval(self) -> int:
"""`int`: Streakline release interval.
Must be > 0, and the paired `duration` must remain smaller than this
value.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.release_interval = 3
"""
return _coerce_int(self._state().get("release_interval", 3))
@release_interval.setter
def release_interval(self, value: int) -> None:
"""Set streakline release interval.
`release_interval` must be > 0 and the current `duration` must remain
smaller than the new value.
"""
current_release_interval, current_duration = self._current_streak_pair()
validated_release_interval, _ = _resolve_streak_pair(
current_release_interval=current_release_interval,
current_duration=current_duration,
release_interval=value,
)
_core_call(
_core.streamlines_surf_modify,
self._phigs_obj,
{"release_interval": validated_release_interval},
)
@property
def duration(self) -> int:
"""`int`: Streakline duration.
Must be > 0 and < `release_interval`.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.duration = 1
"""
return _coerce_int(self._state().get("duration", 1))
@duration.setter
def duration(self, value: int) -> None:
"""Set streakline duration.
`duration` must be > 0 and < `release_interval`.
"""
current_release_interval, current_duration = self._current_streak_pair()
_, validated_duration = _resolve_streak_pair(
current_release_interval=current_release_interval,
current_duration=current_duration,
duration=value,
)
_core_call(
_core.streamlines_surf_modify,
self._phigs_obj,
{"duration": validated_duration},
)
@property
def streak(self) -> StreamlinesStreak:
"""`StreamlinesStreak`: Convenience access to streak controls.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> streak = sl.streak
"""
return self._streak
[docs]
def add_seeds(
self,
seeds: Sequence[Sequence[int | float]],
*,
grid: int = 1,
grids: Sequence[int] | None = None,
) -> None:
"""Add one or more seed points.
Args:
seeds: Sequence of seed coordinates. Use XYZ values when
``seed_coord`` is ``xyz``; otherwise use IJK or IJK-real
triplets.
grid: Default grid number to apply to all seeds when the active
seed coordinate mode requires grid ids.
grids: Optional per-seed grid numbers. When provided, the length
must match ``seeds``.
Example usage:
.. code-block:: python
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0), (10.0, 8.0, 6.0)], grid=1)
"""
if not isinstance(seeds, Sequence):
raise InvalidArgumentError("seeds must be a sequence of 3-value points")
if len(seeds) == 0:
return
seed_coord = self.seed_coord
integral = seed_coord == constant.StreamlinesSeedCoord.IJK
normalized_seeds = [_to_seed_triplet(seed, integral=integral) for seed in seeds]
payload: dict[str, object] = {"seeds": normalized_seeds}
if seed_coord != constant.StreamlinesSeedCoord.XYZ:
if grids is not None:
grid_values = [int(item) for item in grids]
if len(grid_values) != len(normalized_seeds):
raise InvalidArgumentError(
"grids must have the same length as seeds"
)
payload["grids"] = grid_values
else:
payload["grid"] = int(grid)
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"add_seeds": payload}
)
[docs]
def seed_surface(
self,
surface: SurfaceBase,
*,
inc: int = 3,
num_seeds: int = 10,
) -> None:
"""Seed from another live surface (`Boundary`, `Comp`, `Coord`, or `Iso`).
The surface must not be deleted and must belong to the same dataset
as this streamline object.
Args:
surface: Source surface used to generate streamline seeds.
inc: Grid-space increment used when ``seed_coord`` is ``ijk``.
num_seeds: Number of seeds to distribute when ``seed_coord`` is
``ijk_real`` or ``xyz``.
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"),
... )
>>> bnd = fv.vis.create_boundary(ds, types=fv.constant.BoundaryTypeSelection.ALL)
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.seed_surface(surface=bnd)
"""
if not isinstance(surface, SurfaceBase):
raise InvalidArgumentError(
"surface must be a Boundary, Comp, Coord, or Iso surface"
)
if getattr(surface, "_is_deleted", False):
raise InvalidArgumentError("surface was deleted")
surface_dataset_id = getattr(surface, "_dataset_id", None)
if surface_dataset_id != self._dataset_id:
raise InvalidArgumentError(
"surface must belong to the same dataset as the streamlines object"
)
phigs_obj = getattr(surface, "_phigs_obj", None)
if not isinstance(phigs_obj, int):
raise InvalidArgumentError("surface must be a FieldView surface object")
seed_coord = self.seed_coord
payload: dict[str, object] = {"surface_phigs_obj": int(phigs_obj)}
if seed_coord == constant.StreamlinesSeedCoord.IJK:
payload["inc"] = int(inc)
else:
payload["num_seeds"] = int(num_seeds)
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"seed_surface": payload}
)
[docs]
def delete_all_seeds(self) -> None:
"""Delete all seeds.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.delete_all_seeds()
"""
_core_call(
_core.streamlines_surf_modify, self._phigs_obj, {"delete_all_seeds": True}
)
[docs]
def calculate(self) -> None:
"""Run streamline calculation using current seeds and parameters.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
"""
_core_call(_core.streamlines_surf_calculate, self._phigs_obj)
[docs]
def modify(
self,
*,
coloring: constant.Coloring | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
line_type: constant.LineType | str | None = None,
display_type: constant.PathsDisplayType | str | None = None,
sphere_scale: float | None = None,
ribbon_width: int | None = None,
transparency: float | None = None,
visibility: bool | None = None,
scalar_func: str | None = None,
vector_func: str | None = None,
animate: bool | None = None,
animate_direction: constant.AnimationDirection | str | None = None,
animate_divs: int | None = None,
show_seeds: bool | None = None,
seed_coord: constant.StreamlinesSeedCoord | str | None = None,
direction: constant.CalculationDirection | str | None = None,
step: int | None = None,
time_limit: float | None | object = _UNSET,
release_interval: int | None = None,
duration: int | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
calculate: bool = False,
) -> None:
"""Update multiple properties in one call.
Calculation runs implicitly when any calculation parameter
(``direction``, ``step``, ``time_limit``, ``release_interval``,
``duration``) is changed, or when ``calculate=True``.
Args:
coloring: Coloring mode for the streamline geometry.
geometric_color: Geometric color id used when geometric coloring is
active.
line_type: Streamline line style.
display_type: Path display type such as complete, filament, or
ribbon.
sphere_scale: Sphere or arrow size scale.
ribbon_width: Ribbon width used by ribbon display modes.
transparency: Transparency amount for the streamline geometry.
visibility: Whether the streamline object is visible.
scalar_func: Scalar function name used for scalar coloring.
vector_func: Vector function name used for streamline integration.
animate: Whether streamline animation is enabled.
animate_direction: Direction used for streamline animation.
animate_divs: Number of animation divisions.
show_seeds: Whether seed markers are shown.
seed_coord: Seed coordinate system to use for subsequent seed
operations.
direction: Integration direction used for calculation.
step: Integration step size.
time_limit: Optional tracing time limit. Pass ``None`` to clear the
current limit.
release_interval: Optional streakline release interval.
duration: Optional streakline duration.
colormap: Optional colormap settings as a :class:`Colormap` or
payload dict.
legend: Optional legend settings as a :class:`Legend` or payload
dict.
calculate: When ``True``, run streamline calculation after
applying the updates.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.modify(visibility=False, show_seeds=False)
`release_interval` and `duration` are validated together before any
core call. Invalid combinations are rejected without partial updates.
"""
payload: dict[str, object] = {}
if coloring is not None:
payload["coloring"] = _normalize_streamlines_coloring(coloring)
if geometric_color is not None:
payload["geometric_color"] = int(geometric_color)
if line_type is not None:
payload["line_type"] = _normalize_line_type(line_type)
if display_type is not None:
payload["display_type"] = _normalize_streamlines_display_type(display_type)
if sphere_scale is not None:
payload["sphere_scale"] = float(sphere_scale)
if ribbon_width is not None:
payload["ribbon_width"] = int(ribbon_width)
if transparency is not None:
payload["transparency"] = float(transparency)
if visibility is not None:
payload["visibility"] = bool(visibility)
if scalar_func is not None:
dataset = self._require_dataset()
payload["scalar_func_id"] = self._lookup_function_id(
scalar_func, dataset._scalar_function_ids, "scalar"
)
if vector_func is not None:
dataset = self._require_dataset()
payload["vector_func_id"] = self._lookup_function_id(
vector_func, dataset._vector_function_ids, "vector"
)
if animate is not None:
payload["animate"] = bool(animate)
if animate_direction is not None:
payload["animate_direction"] = _normalize_animate_direction(
animate_direction
)
if animate_divs is not None:
payload["animate_divs"] = int(animate_divs)
if show_seeds is not None:
payload["show_seeds"] = bool(show_seeds)
if seed_coord is not None:
payload["seed_coord"] = _normalize_seed_coord(seed_coord)
if direction is not None:
payload["direction"] = _normalize_direction(direction)
if step is not None:
payload["step"] = int(step)
if time_limit is not _UNSET:
payload["time_limit"] = (
None if time_limit is None else _coerce_float(time_limit)
)
validated_release_interval: int | None = None
validated_duration: int | None = None
if release_interval is not None or duration is not None:
current_release_interval, current_duration = self._current_streak_pair()
# modify() uses None for "argument omitted", while the shared
# validator uses _UNSET for that role so it can distinguish
# omission from an explicit numeric value.
requested_release_interval = (
release_interval if release_interval is not None else _UNSET
)
requested_duration = duration if duration is not None else _UNSET
validated_release_interval, validated_duration = _resolve_streak_pair(
current_release_interval=current_release_interval,
current_duration=current_duration,
release_interval=requested_release_interval,
duration=requested_duration,
)
if validated_release_interval is not None:
payload["release_interval"] = validated_release_interval
if validated_duration is not None:
payload["duration"] = validated_duration
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))
implicit_calculate = any(
value is not None for value in (direction, step, release_interval, duration)
) or (time_limit is not _UNSET)
if not payload and not calculate and not implicit_calculate:
raise InvalidArgumentError("modify requires at least one property")
if payload:
_core_call(_core.streamlines_surf_modify, self._phigs_obj, payload)
if calculate or implicit_calculate:
self.calculate()
[docs]
def export(self, filename: str | os.PathLike[str], format: str = "fvp") -> None:
"""Export this streamlines surface to an FVP file.
Args:
filename: Destination export path.
format: Export format string. Only ``fvp`` is currently supported.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.calculate()
>>> sl.export("/tmp/streamlines.fvp", format="fvp")
"""
filename = _coerce_pathlike_str(filename, "filename")
if not filename.strip():
raise InvalidArgumentError("filename cannot be empty")
if not isinstance(format, str):
raise InvalidArgumentError("format must be a string")
fmt = format.strip().lower()
if fmt not in {"fvp", "fv particle path"}:
raise InvalidArgumentError("format must be fvp")
_core_call(_core.streamlines_surf_export, self._phigs_obj, filename, "fvp")
[docs]
def copy(self) -> "Streamlines":
"""Return a new streamlines surface with the same settings and seeds.
Raises :class:`InvalidArgumentError` if the current streak controls are
not valid for the Python API contract.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> copy_obj = sl.copy()
"""
dataset = self._require_dataset()
state = self._state(include_seeds=True)
vector_func_name = state.get("vector_func_name") or None
if vector_func_name is None:
raise InvalidArgumentError("cannot copy streamlines without vector_func")
release_interval = _normalize_streak_int(
state.get("release_interval", 3), "release_interval"
)
duration = _normalize_streak_int(state.get("duration", 1), "duration")
release_interval, duration = _validate_streak_pair(release_interval, duration)
surf = create_streamlines(
dataset,
coloring=cast(constant.Coloring | str | None, state.get("coloring")),
geometric_color=cast(
constant.GeometricColor | int | None,
state.get("geometric_color"),
),
line_type=cast(constant.LineType | str | None, state.get("line_type")),
display_type=cast(
constant.PathsDisplayType | str | None,
state.get("display_type"),
),
sphere_scale=None
if state.get("sphere_scale") is None
else _coerce_float(state.get("sphere_scale")),
ribbon_width=None
if state.get("ribbon_width") is None
else _coerce_int(state.get("ribbon_width")),
transparency=None
if state.get("transparency") is None
else _coerce_float(state.get("transparency")),
visibility=cast(bool | None, state.get("visibility")),
scalar_func=cast(str | None, state.get("scalar_func_name") or None),
vector_func=cast(str, vector_func_name),
animate=cast(bool | None, state.get("animate")),
animate_direction=cast(
constant.AnimationDirection | str | None,
state.get("animate_direction"),
),
animate_divs=None
if state.get("animate_divs") is None
else _coerce_int(state.get("animate_divs")),
show_seeds=cast(bool | None, state.get("show_seeds")),
seed_coord=cast(
constant.StreamlinesSeedCoord | str | None,
state.get("seed_coord"),
),
direction=cast(
constant.CalculationDirection | str | None,
state.get("direction"),
),
step=None if state.get("step") is None else _coerce_int(state.get("step")),
time_limit=state.get("time_limit"),
release_interval=release_interval,
duration=duration,
colormap=cast(
Colormap | dict[str, object] | None,
state.get("scalar_colormap"),
),
legend=cast(Legend | dict[str, object] | None, state.get("legend")),
calculate=False,
)
seed_entries = self._seed_entries_from_state(state)
if seed_entries:
target_seed_coord = str(
state.get("seed_coord", constant.StreamlinesSeedCoord.IJK_REAL.value)
).lower()
coords, grids = self._seed_payload_from_entries(
seed_entries, target_seed_coord
)
if target_seed_coord == constant.StreamlinesSeedCoord.XYZ.value:
surf.add_seeds(coords)
else:
surf.add_seeds(coords, grids=grids)
return surf
[docs]
def delete(self) -> None:
"""Delete this streamlines surface.
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"),
... )
>>> sl = fv.vis.create_streamlines(ds, vector_func=ds.vector_functions[0])
>>> sl.seed_coord = fv.constant.StreamlinesSeedCoord.IJK_REAL
>>> sl.add_seeds([(5.0, 5.0, 5.0)], grid=1)
>>> sl.calculate()
>>> sl.delete()
"""
self._delete_surface(_core.streamlines_surf_delete)
[docs]
def create_streamlines(
dataset: Dataset | None = None,
*,
coloring: constant.Coloring | str | None = None,
geometric_color: constant.GeometricColor | int | None = None,
line_type: constant.LineType | str | None = None,
display_type: constant.PathsDisplayType | str | None = None,
sphere_scale: float | None = None,
ribbon_width: int | None = None,
transparency: float | None = None,
visibility: bool | None = None,
scalar_func: str | None = None,
vector_func: str | None = None,
animate: bool | None = None,
animate_direction: constant.AnimationDirection | str | None = None,
animate_divs: int | None = None,
show_seeds: bool | None = None,
seed_coord: constant.StreamlinesSeedCoord | str | None = None,
direction: constant.CalculationDirection | str | None = None,
step: int | None = None,
time_limit: float | None | object = _UNSET,
release_interval: int | None = None,
duration: int | None = None,
colormap: Colormap | dict[str, object] | None = None,
legend: Legend | dict[str, object] | None = None,
defer_apply: bool = False,
calculate: bool = False,
) -> Streamlines:
"""Create a streamlines surface for the given dataset.
`vector_func` is required and can be any dataset vector function name.
When provided, `release_interval` and `duration` must satisfy
`release_interval > 0` and `0 < duration < release_interval`.
Args:
dataset: Dataset to attach the new streamlines surface to. When
omitted, the current dataset is used.
coloring: Coloring mode for the streamline geometry.
geometric_color: Geometric color id used when geometric coloring is
active.
line_type: Streamline line style.
display_type: Path display type such as complete, filament, or ribbon.
sphere_scale: Sphere or arrow size scale.
ribbon_width: Ribbon width used by ribbon display modes.
transparency: Transparency amount for the streamline geometry.
visibility: Whether the new streamline object is initially visible.
scalar_func: Optional scalar function name used for scalar coloring.
vector_func: Required vector function name used for streamline
integration.
animate: Whether streamline animation is enabled.
animate_direction: Direction used for streamline animation.
animate_divs: Number of animation divisions.
show_seeds: Whether seed markers are shown.
seed_coord: Seed coordinate system to use for subsequent seed
operations.
direction: Integration direction used for calculation.
step: Integration step size.
time_limit: Optional tracing time limit. Pass ``None`` to clear the
current limit.
release_interval: Optional streakline release interval.
duration: Optional streakline duration.
colormap: Optional colormap settings as a :class:`Colormap` or payload
dict.
legend: Optional legend settings as a :class:`Legend` or payload dict.
defer_apply: When ``True``, create the object without forcing an
immediate apply/calculation step.
calculate: When ``True``, run streamline 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"),
... )
>>> sl = fv.vis.create_streamlines(
... ds,
... vector_func=ds.vector_functions[0],
... coloring=fv.constant.Coloring.SCALAR,
... scalar_func=ds.scalar_functions[0],
... seed_coord=fv.constant.StreamlinesSeedCoord.IJK_REAL,
... )
>>> sl.add_seeds([(1.0, 1.0, 1.0), (40.0, 20.0, 10.0)], grid=1)
>>> sl.calculate()
"""
ds = _resolve_dataset_or_current(dataset)
if vector_func is None:
raise InvalidArgumentError("vector_func is required when creating streamlines")
if release_interval is not None or duration is not None:
_resolve_streak_pair(
current_release_interval=3,
current_duration=1,
release_interval=release_interval
if release_interval is not None
else _UNSET,
duration=duration if duration is not None else _UNSET,
)
if coloring is not None:
_normalize_streamlines_coloring(coloring)
if line_type is not None:
_normalize_line_type(line_type)
if animate_direction is not None:
_normalize_animate_direction(animate_direction)
has_modifications = any(
(value is not None and value is not _UNSET)
for value in (
coloring,
geometric_color,
line_type,
display_type,
sphere_scale,
ribbon_width,
transparency,
visibility,
scalar_func,
vector_func,
animate,
animate_direction,
animate_divs,
show_seeds,
seed_coord,
direction,
step,
time_limit,
release_interval,
duration,
colormap,
legend,
)
)
phigs_obj = _core_call(
_core.streamlines_surf_create,
ds.dataset_id,
defer_apply=bool(defer_apply or has_modifications),
)
surf = Streamlines(phigs_obj=phigs_obj, dataset_id=ds.dataset_id, dataset=ds)
if has_modifications or calculate:
try:
surf.modify(
coloring=coloring,
geometric_color=geometric_color,
line_type=line_type,
display_type=display_type,
sphere_scale=sphere_scale,
ribbon_width=ribbon_width,
transparency=transparency,
visibility=visibility,
scalar_func=scalar_func,
vector_func=vector_func,
animate=animate,
animate_direction=animate_direction,
animate_divs=animate_divs,
show_seeds=show_seeds,
seed_coord=seed_coord,
direction=direction,
step=step,
time_limit=time_limit,
release_interval=release_interval,
duration=duration,
colormap=colormap,
legend=legend,
calculate=calculate,
)
except Exception:
_core_call(_core.streamlines_surf_delete, phigs_obj)
raise
return surf
