from __future__ import annotations
import importlib
from dataclasses import dataclass, field
from types import ModuleType
from typing import Protocol, cast
from collections.abc import Callable
from . import _core as _core_module
from ._core_utils import _core_call
from .exceptions import CoreError, InvalidDatasetError
__all__: list[str] = ["ObjectLists", "get_current_object", "get_all_objects"]
def _import_module(name: str) -> ModuleType:
return importlib.import_module(name, __package__)
def _import_attr(module_name: str, attr_name: str) -> object:
return cast(object, getattr(_import_module(module_name), attr_name))
class _DatasetObjectCtor(Protocol):
def __call__(
self, phigs_obj: int, dataset_id: int, dataset: object | None
) -> object: ...
class _Plot2DCtor(Protocol):
def __call__(
self, dataset_id: int, plot_id: int, dataset: object | None
) -> object: ...
class _AnnotationCtor(Protocol):
def __call__(self, phigs_obj: int) -> object: ...
_get_or_attach_dataset = cast(
Callable[[int], object], _import_attr(".data", "_get_or_attach_dataset")
)
_annotation_arrow_cls = cast(
_AnnotationCtor, _import_attr(".vis.annotation", "AnnotationArrow")
)
_annotation_text_cls = cast(
_AnnotationCtor, _import_attr(".vis.annotation", "AnnotationText")
)
_boundary_cls = cast(_DatasetObjectCtor, _import_attr(".vis.boundary", "Boundary"))
_comp_cls = cast(_DatasetObjectCtor, _import_attr(".vis.comp", "Comp"))
_coord_cls = cast(_DatasetObjectCtor, _import_attr(".vis.coord", "Coord"))
_iso_cls = cast(_DatasetObjectCtor, _import_attr(".vis.iso", "Iso"))
_particle_paths_cls = cast(
_DatasetObjectCtor, _import_attr(".vis.particle_paths", "ParticlePaths")
)
_plot2d_cls = cast(_Plot2DCtor, _import_attr(".vis.plot2d", "Plot2D"))
_reattachment_lines_cls = cast(
_DatasetObjectCtor,
_import_attr(".vis.reattachment_lines", "ReattachmentLines"),
)
_separation_lines_cls = cast(
_DatasetObjectCtor,
_import_attr(".vis.separation_lines", "SeparationLines"),
)
_streamlines_cls = cast(
_DatasetObjectCtor, _import_attr(".vis.streamlines", "Streamlines")
)
_surface_flows_cls = cast(
_DatasetObjectCtor, _import_attr(".vis.surface_flows", "SurfaceFlows")
)
_vortex_cores_cls = cast(
_DatasetObjectCtor, _import_attr(".vis.vortex_cores", "VortexCores")
)
[docs]
@dataclass
class ObjectLists:
"""Grouped scene objects returned by :func:`get_all_objects`.
Example usage:
.. code-block:: python
>>> import fieldview as fv
>>> objects = fv.get_all_objects()
>>> len(objects.coord_list)
1
>>> coord = objects.coord_list[0]
>>> coord.plane = fv.constant.Plane.Y
"""
boundary_list: list[object] = field(default_factory=list)
comp_list: list[object] = field(default_factory=list)
coord_list: list[object] = field(default_factory=list)
iso_list: list[object] = field(default_factory=list)
streamlines_list: list[object] = field(default_factory=list)
particle_paths_list: list[object] = field(default_factory=list)
plot2d_list: list[object] = field(default_factory=list)
vortex_cores_list: list[object] = field(default_factory=list)
surface_flows_list: list[object] = field(default_factory=list)
separation_lines_list: list[object] = field(default_factory=list)
reattachment_lines_list: list[object] = field(default_factory=list)
annotation_text_list: list[object] = field(default_factory=list)
annotation_arrow_list: list[object] = field(default_factory=list)
def _as_dict(metadata: object, label: str) -> dict[str, object]:
if not isinstance(metadata, dict):
raise CoreError(f"{label} returned invalid metadata.")
return cast(dict[str, object], metadata)
def _as_int(metadata: dict[str, object], key: str, label: str) -> int:
if key not in metadata:
raise CoreError(f"{label} is missing {key}.")
value = metadata[key]
if isinstance(value, bool) or not isinstance(value, (int, float, str)):
raise CoreError(f"{label} has invalid {key}.")
return int(value)
def _dataset_from_registry(dataset_id: int) -> object | None:
try:
return _get_or_attach_dataset(int(dataset_id))
except InvalidDatasetError:
return None
def _build_dataset_object(kind: str, metadata: dict[str, object]) -> object:
dataset_id = _as_int(metadata, "dataset_id", f"Object metadata for '{kind}'")
dataset_ref = _dataset_from_registry(dataset_id)
if kind == "plot2d":
return _plot2d_cls(
dataset_id,
_as_int(metadata, "plot_id", "Object metadata for 'plot2d'"),
dataset_ref,
)
phigs_obj = _as_int(metadata, "phigs_obj", f"Object metadata for '{kind}'")
dataset_types = {
"boundary": _boundary_cls,
"comp": _comp_cls,
"coord": _coord_cls,
"iso": _iso_cls,
"streamlines": _streamlines_cls,
"particle_paths": _particle_paths_cls,
"vortex_cores": _vortex_cores_cls,
"surface_flows": _surface_flows_cls,
"separation_lines": _separation_lines_cls,
"reattachment_lines": _reattachment_lines_cls,
}
cls = dataset_types.get(kind)
if cls is None:
raise CoreError(f"Unsupported object kind '{kind}'.")
return cls(phigs_obj, dataset_id, dataset_ref)
def _build_object(metadata: object) -> object:
item = _as_dict(metadata, "object query")
if "kind" not in item:
raise CoreError("Object metadata is missing required fields.")
kind = str(item["kind"])
if kind == "plot2d":
return _build_dataset_object(kind, item)
if "phigs_obj" not in item:
raise CoreError("Object metadata is missing required fields.")
if kind == "annotation_text":
return _annotation_text_cls(_as_int(item, "phigs_obj", "Object metadata"))
if kind == "annotation_arrow":
return _annotation_arrow_cls(_as_int(item, "phigs_obj", "Object metadata"))
return _build_dataset_object(kind, item)
def _build_group(metadata: dict[str, object], key: str) -> list[object]:
values = metadata.get(key, [])
if not isinstance(values, list):
raise CoreError(f"{key} metadata must be a list.")
return [_build_object(item) for item in cast(list[object], values)]
[docs]
def get_current_object() -> object | None:
"""Return the current scene object as its concrete wrapper type.
Returns ``None`` when FieldView has no current object.
Example usage:
.. code-block:: python
>>> import fieldview as fv
>>> current = fv.get_current_object()
>>> if isinstance(current, fv.vis.Coord):
... current.plane = fv.constant.Plane.Y
... elif isinstance(current, fv.vis.Iso):
... current.visibility = False
Notes:
Do not manually cast the result to another wrapper type. The returned
object is already the correct runtime type. Use ``isinstance`` before
accessing type-specific properties.
"""
func = getattr(_core_module, "get_current_object_metadata", None)
if not callable(func):
raise CoreError("FieldView host does not provide get_current_object().")
metadata = _core_call(func)
if metadata is None:
return None
return _build_object(metadata)
[docs]
def get_all_objects() -> ObjectLists:
"""Return all scene objects grouped by public wrapper type.
Example usage:
.. code-block:: python
>>> import fieldview as fv
>>> objects = fv.get_all_objects()
>>> for coord in objects.coord_list:
... coord.visibility = True
>>> for text in objects.annotation_text_list:
... print(text.text)
"""
func = getattr(_core_module, "get_all_objects_metadata", None)
if not callable(func):
raise CoreError("FieldView host does not provide get_all_objects().")
metadata = _core_call(func)
payload = _as_dict(metadata, "get_all_objects")
return ObjectLists(
boundary_list=_build_group(payload, "boundary_list"),
comp_list=_build_group(payload, "comp_list"),
coord_list=_build_group(payload, "coord_list"),
iso_list=_build_group(payload, "iso_list"),
streamlines_list=_build_group(payload, "streamlines_list"),
particle_paths_list=_build_group(payload, "particle_paths_list"),
plot2d_list=_build_group(payload, "plot2d_list"),
vortex_cores_list=_build_group(payload, "vortex_cores_list"),
surface_flows_list=_build_group(payload, "surface_flows_list"),
separation_lines_list=_build_group(payload, "separation_lines_list"),
reattachment_lines_list=_build_group(payload, "reattachment_lines_list"),
annotation_text_list=_build_group(payload, "annotation_text_list"),
annotation_arrow_list=_build_group(payload, "annotation_arrow_list"),
)
