"""Manifest-backed visualization capabilities for Flow360 resources."""
from __future__ import annotations
import math
from collections.abc import Iterable, Mapping
from typing import Any, Literal, NamedTuple, TypeAlias
import pydantic as pd
from ._flow360_adapter import DEFAULT_FLOW360_ADAPTER
from .models import ReportBaseModel, ResourceType
from .resource_types import ReportResourceInput
VisualizationOutputType = Literal["surface", "slice", "isosurface", "streamline"]
_SOLID_GEOMETRY_TYPE = "SolidGeometry"
_STREAMLINE_ARRAY_TYPE = "StreamlineArray"
_ROOT_SURFACE_IDS = frozenset({"volumeMesh", "boundaries"})
_ROOT_EXCLUDED_SOLID_IDS = frozenset({"qcriterion", "slices", "isosurfaces"})
_FIELD_GROUP_TYPES: dict[str, VisualizationOutputType] = {
"surfaces": "surface",
"boundaries": "surface",
"slices": "slice",
"isosurfaces": "isosurface",
"streamlines": "streamline",
}
_ALLOWED_OUTPUT_TYPES: dict[ResourceType, frozenset[VisualizationOutputType]] = {
"Case": frozenset({"surface", "slice", "isosurface", "streamline"}),
"Geometry": frozenset(),
"SurfaceMesh": frozenset({"surface"}),
"VolumeMesh": frozenset({"slice"}),
}
_NON_FIELD_SECTION_NAMES = frozenset(
{
"index",
"indices",
"position",
"normal",
"color",
"edgePosition",
"edgeIndices",
"rotationAngle",
"elementGroupId",
}
)
_VECTOR_COMPONENTS = ("X", "Y", "Z")
_VOLUME_MESH_CONFIG_OWNER_ID = "mesh-diagnostic-volume"
FieldBounds: TypeAlias = tuple[float, float]
__all__ = [
"ResourceVisualizationCapabilities",
"VisualizationFieldCapability",
"VisualizationOutputCapability",
"VisualizationTargetCapability",
"get_capabilities",
]
class _ManifestFieldBounds(NamedTuple):
min_value: float | tuple[float, ...]
max_value: float | tuple[float, ...]
min_magnitude: float | None
max_magnitude: float | None
class VisualizationFieldCapability(ReportBaseModel):
"""One field addressable under a visualization output."""
name: str = pd.Field(min_length=1)
dimension: int = pd.Field(default=1, ge=1)
source_name: str | None = None
component: Literal["X", "Y", "Z"] | None = None
bounds: FieldBounds | None = None
class VisualizationTargetCapability(ReportBaseModel):
"""One visibility target addressable under a visualization output."""
id: str = pd.Field(min_length=1)
type: str | None = None
controller_owner_id: str | None = None
class VisualizationOutputCapability(ReportBaseModel):
"""Fields and visibility targets scoped to one exact output owner ID."""
id: str = pd.Field(min_length=1)
type: VisualizationOutputType
controller_owner_ids: list[str] = pd.Field(default_factory=list)
fields: list[VisualizationFieldCapability] = pd.Field(default_factory=list)
visibility_targets: list[VisualizationTargetCapability] = pd.Field(default_factory=list)
@property
def field_names(self) -> tuple[str, ...]:
"""Return field names in manifest order."""
return tuple(field.name for field in self.fields)
def get_field(self, field_name: str) -> VisualizationFieldCapability | None:
"""Return the capability for one exact field name."""
return next((field for field in self.fields if field.name == field_name), None)
@property
def visibility_target_ids(self) -> tuple[str, ...]:
"""Return visibility target IDs in manifest order."""
return tuple(target.id for target in self.visibility_targets)
class ResourceVisualizationCapabilities(ReportBaseModel):
"""Manifest-declared visualization capabilities for one cloud resource."""
resource_id: str = pd.Field(min_length=1)
resource_type: ResourceType
manifest_version: str | None = None
outputs: list[VisualizationOutputCapability] = pd.Field(default_factory=list)
def get_output(self, output_id: str) -> VisualizationOutputCapability | None:
"""Return the capability for an exact output owner ID."""
return next((output for output in self.outputs if output.id == output_id), None)
[docs]
def get_capabilities(resource: ReportResourceInput) -> ResourceVisualizationCapabilities:
"""Return Manifest-declared visualization capabilities for one resource.
AI-generated workflows should call this function for every selected resource
before choosing exact output, field, or visibility-target IDs. See
``get_agent_guide()`` for the required cross-resource selection workflow.
"""
adapter = DEFAULT_FLOW360_ADAPTER
environment = adapter.current_environment()
resource_type, resource_id = adapter.resource_identity(resource)
manifest = adapter.get_visualization_manifest(resource, environment=environment)
return parse_visualization_capabilities(
resource_id=resource_id,
resource_type=resource_type,
manifest=manifest,
)
class _OutputBinding:
"""Internal config-owner to controller-owner mapping matching Workbench."""
def __init__(
self,
*,
config_owner_id: str,
controller_owner_ids: list[str],
output_type: VisualizationOutputType,
member_ids_by_controller_owner: dict[str, list[str]],
):
self.config_owner_id = config_owner_id
self.controller_owner_ids = controller_owner_ids
self.output_type = output_type
self.member_ids_by_controller_owner = member_ids_by_controller_owner
self.member_ids = list(
dict.fromkeys(
member_id
for member_ids in member_ids_by_controller_owner.values()
for member_id in member_ids
)
)
def _as_record(value: Any) -> Mapping[str, Any] | None:
return value if isinstance(value, Mapping) else None
def _string_ids(value: Any) -> list[str]:
if not isinstance(value, list):
return []
return [str(item) for item in value]
def _attribution_ids(node: Mapping[str, Any], name: str) -> list[str]:
attributions = _as_record(node.get("attributions"))
return _string_ids(attributions.get(name)) if attributions else []
def _is_root_group(node: Mapping[str, Any]) -> bool:
properties = _as_record(node.get("properties"))
return (
node.get("type") == "GeometryGroup"
and properties is not None
and properties.get("type") == 0
)
def _manifest_nodes(manifest: Any) -> list[Mapping[str, Any]]:
if isinstance(manifest, Mapping):
manifest = manifest.get("manifest")
if not isinstance(manifest, list):
raise ValueError("visualization manifest must be a list of nodes")
nodes = [_as_record(node) for node in manifest]
if any(node is None for node in nodes):
raise ValueError("visualization manifest nodes must be objects")
return [node for node in nodes if node is not None]
def _categorize_outputs(
nodes: list[Mapping[str, Any]],
) -> list[tuple[str, VisualizationOutputType]]:
nodes_by_id = {str(node.get("id")): node for node in nodes if node.get("id") is not None}
root = next((node for node in nodes if _is_root_group(node)), None)
if root is None:
raise ValueError("visualization manifest does not contain a root GeometryGroup")
root_members = set(_attribution_ids(root, "members"))
if not root_members:
return []
outputs: list[tuple[str, VisualizationOutputType]] = []
visited: set[str] = set()
def append(output_id: str, output_type: VisualizationOutputType) -> None:
item = (output_id, output_type)
if item not in outputs:
outputs.append(item)
def traverse(node_id: str, parent_id: str | None) -> None:
if node_id in visited:
return
visited.add(node_id)
node = nodes_by_id.get(node_id)
if node is None:
return
manifest_type = node.get("type")
members = _attribution_ids(node, "members")
is_in_root = node_id in root_members
if manifest_type == _STREAMLINE_ARRAY_TYPE and is_in_root:
append(node_id, "streamline")
return
if node_id in _ROOT_SURFACE_IDS and manifest_type == _SOLID_GEOMETRY_TYPE and is_in_root:
append(node_id, "surface")
return
resources = _as_record(node.get("resources"))
if (
manifest_type == _SOLID_GEOMETRY_TYPE
and is_in_root
and resources is not None
and resources.get("buffers") is not None
and node_id not in _ROOT_EXCLUDED_SOLID_IDS
):
append(node_id, "surface")
return
if node_id == "qcriterion" and manifest_type == _SOLID_GEOMETRY_TYPE and is_in_root:
append(node_id, "isosurface")
return
if node_id == "isosurfaces" and is_in_root and _has_solid_members(node, nodes_by_id):
append(node_id, "isosurface")
return
if node_id == "slices" and manifest_type == _SOLID_GEOMETRY_TYPE and is_in_root:
append(node_id, "slice")
return
if node_id == "slices" and is_in_root and _has_solid_members(node, nodes_by_id):
append(node_id, "slice")
return
if parent_id == "boundaries" and manifest_type == _SOLID_GEOMETRY_TYPE:
append(node_id, "surface")
return
if parent_id in _FIELD_GROUP_TYPES and manifest_type == "GeometryGroup":
append(node_id, _FIELD_GROUP_TYPES[parent_id])
return
for member_id in members:
traverse(member_id, node_id)
traverse(str(root.get("id")), None)
return outputs
def _has_solid_members(
node: Mapping[str, Any], nodes_by_id: Mapping[str, Mapping[str, Any]]
) -> bool:
return any(
nodes_by_id.get(member_id, {}).get("type") == _SOLID_GEOMETRY_TYPE
for member_id in _attribution_ids(node, "members")
)
def _collect_descendant_ids(
nodes_by_id: Mapping[str, Mapping[str, Any]],
root_id: str,
*,
manifest_type: str,
) -> list[str]:
visited: set[str] = set()
result: list[str] = []
def traverse(node_id: str) -> None:
if node_id in visited:
return
visited.add(node_id)
node = nodes_by_id.get(node_id)
if node is None:
return
if node.get("type") == manifest_type:
result.append(node_id)
return
for member_id in _attribution_ids(node, "members"):
traverse(member_id)
traverse(root_id)
return result
def _visibility_member_ids(
nodes_by_id: Mapping[str, Mapping[str, Any]],
output_id: str,
output_type: VisualizationOutputType,
) -> list[str]:
node = nodes_by_id.get(output_id)
if node is None:
return []
if output_type == "streamline":
return [output_id]
if output_type in ("slice", "isosurface"):
return _collect_descendant_ids(nodes_by_id, output_id, manifest_type=_SOLID_GEOMETRY_TYPE)
solid_ids = (
[output_id]
if node.get("type") == _SOLID_GEOMETRY_TYPE
else _collect_descendant_ids(nodes_by_id, output_id, manifest_type=_SOLID_GEOMETRY_TYPE)
)
member_ids: list[str] = []
seen: set[str] = set()
for solid_id in solid_ids:
solid = nodes_by_id.get(solid_id)
if solid is None:
continue
for member_id in (
*_attribution_ids(solid, "faces"),
*_attribution_ids(solid, "edges"),
):
if member_id not in seen:
seen.add(member_id)
member_ids.append(member_id)
return member_ids
def _build_bindings(
nodes: list[Mapping[str, Any]], resource_type: ResourceType
) -> list[_OutputBinding]:
nodes_by_id = {str(node.get("id")): node for node in nodes if node.get("id") is not None}
allowed_types = _ALLOWED_OUTPUT_TYPES[resource_type]
outputs = [item for item in _categorize_outputs(nodes) if item[1] in allowed_types]
if resource_type == "VolumeMesh":
slice_ids = [output_id for output_id, output_type in outputs if output_type == "slice"]
if not slice_ids:
return []
return [
_OutputBinding(
config_owner_id=_VOLUME_MESH_CONFIG_OWNER_ID,
controller_owner_ids=slice_ids,
output_type="slice",
member_ids_by_controller_owner={
output_id: _visibility_member_ids(nodes_by_id, output_id, "slice")
for output_id in slice_ids
},
)
]
return [
_OutputBinding(
config_owner_id=output_id,
controller_owner_ids=[output_id],
output_type=output_type,
member_ids_by_controller_owner={
output_id: _visibility_member_ids(nodes_by_id, output_id, output_type)
},
)
for output_id, output_type in outputs
]
def _iter_buffer_records(buffers: Any) -> Iterable[Mapping[str, Any]]:
record = _as_record(buffers)
if record is None:
return
yield record
levels = record.get("levels")
if isinstance(levels, list):
for level in levels:
level_record = _as_record(level)
if level_record is not None:
yield from _iter_buffer_records(level_record.get("resource", level_record))
animation = _as_record(record.get("animation"))
frames = animation.get("frames") if animation else record.get("frames")
if isinstance(frames, list):
for frame in frames:
frame_record = _as_record(frame)
if frame_record is not None:
yield from _iter_buffer_records(frame_record.get("resource"))
def _field_dimensions(node: Mapping[str, Any]) -> dict[str, int]:
resources = _as_record(node.get("resources"))
if resources is None:
return {}
dimensions: dict[str, int] = {}
for buffer_record in _iter_buffer_records(resources.get("buffers")):
sections = buffer_record.get("sections")
if not isinstance(sections, list):
continue
for section in sections:
section_record = _as_record(section)
if section_record is None:
continue
name = section_record.get("name")
if not isinstance(name, str) or not name or name in _NON_FIELD_SECTION_NAMES:
continue
dimension = section_record.get("dimension")
dimensions[name] = max(
dimensions.get(name, 1),
dimension if isinstance(dimension, int) and dimension > 0 else 1,
)
return dimensions
def _finite_bounds_value(value: Any) -> float | tuple[float, ...] | None:
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return float(value) if math.isfinite(value) else None
if not isinstance(value, list) or not value:
return None
values: list[float] = []
for item in value:
if isinstance(item, bool) or not isinstance(item, (int, float)) or not math.isfinite(item):
return None
values.append(float(item))
return tuple(values)
def _parse_bounds_records(value: Any) -> dict[str, _ManifestFieldBounds]:
if not isinstance(value, list):
return {}
parsed: dict[str, _ManifestFieldBounds] = {}
for item in value:
record = _as_record(item)
if record is None:
continue
name = record.get("name")
if not isinstance(name, str) or not name or name in parsed:
continue
min_value = _finite_bounds_value(record.get("minVal", record.get("min_value")))
max_value = _finite_bounds_value(record.get("maxVal", record.get("max_value")))
if min_value is None or max_value is None:
continue
min_magnitude = _finite_bounds_value(record.get("minMag"))
max_magnitude = _finite_bounds_value(record.get("maxMag"))
parsed[name] = _ManifestFieldBounds(
min_value=min_value,
max_value=max_value,
min_magnitude=min_magnitude if isinstance(min_magnitude, float) else None,
max_magnitude=max_magnitude if isinstance(max_magnitude, float) else None,
)
return parsed
def _field_bounds_by_name(node: Mapping[str, Any]) -> dict[str, _ManifestFieldBounds]:
resources = _as_record(node.get("resources"))
buffers = _as_record(resources.get("buffers")) if resources else None
direct = _parse_bounds_records(buffers.get("bounds") if buffers else None)
if direct:
return direct
properties = _as_record(node.get("properties"))
buffer_locations = _as_record(properties.get("bufferLocations")) if properties else None
indices = buffer_locations.get("indices") if buffer_locations else None
if not isinstance(indices, list) or not indices:
return {}
first_index = _as_record(indices[0])
return _parse_bounds_records(first_index.get("bounds") if first_index else None)
def _select_field_bounds(
bounds: _ManifestFieldBounds,
component_index: int | None,
) -> FieldBounds | None:
min_value = bounds.min_value
max_value = bounds.max_value
if component_index is not None:
selected_min = (
min_value[component_index]
if isinstance(min_value, tuple) and component_index < len(min_value)
else min_value
if isinstance(min_value, float)
else None
)
selected_max = (
max_value[component_index]
if isinstance(max_value, tuple) and component_index < len(max_value)
else max_value
if isinstance(max_value, float)
else None
)
elif isinstance(min_value, tuple) or isinstance(max_value, tuple):
selected_min = bounds.min_magnitude
selected_max = bounds.max_magnitude
else:
selected_min = min_value
selected_max = max_value
if selected_min is None or selected_max is None:
return None
return (min(selected_min, selected_max), max(selected_min, selected_max))
def _aggregate_field_bounds(
bounds_by_model: Iterable[Mapping[str, _ManifestFieldBounds]],
field_name: str,
component_index: int | None = None,
) -> FieldBounds | None:
minimum = math.inf
maximum = -math.inf
found = False
for model_bounds in bounds_by_model:
manifest_bounds = model_bounds.get(field_name)
if manifest_bounds is None:
continue
selected = _select_field_bounds(manifest_bounds, component_index)
if selected is None:
continue
found = True
minimum = min(minimum, selected[0])
maximum = max(maximum, selected[1])
return (minimum, maximum) if found else None
def _field_model_ids(
nodes_by_id: Mapping[str, Mapping[str, Any]],
owner_id: str,
output_type: VisualizationOutputType,
) -> list[str]:
if output_type == "streamline":
return [owner_id]
owner = nodes_by_id.get(owner_id)
if owner is None:
return []
if owner.get("type") == _SOLID_GEOMETRY_TYPE:
return [owner_id]
return _collect_descendant_ids(nodes_by_id, owner_id, manifest_type=_SOLID_GEOMETRY_TYPE)
def _available_base_fields(
nodes_by_id: Mapping[str, Mapping[str, Any]], binding: _OutputBinding
) -> tuple[list[str], dict[str, int]]:
available: list[str] | None = None
dimension_by_name: dict[str, int] = {}
for owner_id in binding.controller_owner_ids:
owner_names: list[str] = []
owner_seen: set[str] = set()
owner_dimensions: dict[str, int] = {}
for model_id in _field_model_ids(nodes_by_id, owner_id, binding.output_type):
node = nodes_by_id.get(model_id)
if node is None:
continue
for name, dimension in _field_dimensions(node).items():
if name not in owner_seen:
owner_seen.add(name)
owner_names.append(name)
owner_dimensions[name] = max(owner_dimensions.get(name, 1), dimension)
if available is None:
available = owner_names
dimension_by_name = owner_dimensions
continue
owner_name_set = set(owner_names)
available = [name for name in available if name in owner_name_set]
for name in list(dimension_by_name):
if name not in owner_name_set:
dimension_by_name.pop(name, None)
continue
dimension_by_name[name] = min(
dimension_by_name[name],
owner_dimensions[name],
)
return available or [], dimension_by_name
def _field_capabilities(
nodes_by_id: Mapping[str, Mapping[str, Any]], binding: _OutputBinding
) -> list[VisualizationFieldCapability]:
names, dimensions = _available_base_fields(nodes_by_id, binding)
model_ids: list[str] = []
seen_model_ids: set[str] = set()
for owner_id in binding.controller_owner_ids:
for model_id in _field_model_ids(nodes_by_id, owner_id, binding.output_type):
if model_id not in seen_model_ids:
seen_model_ids.add(model_id)
model_ids.append(model_id)
bounds_by_model = [
_field_bounds_by_name(nodes_by_id[model_id])
for model_id in model_ids
if model_id in nodes_by_id
]
fields: list[VisualizationFieldCapability] = []
for name in names:
dimension = dimensions.get(name, 1)
fields.append(
VisualizationFieldCapability(
name=name,
dimension=dimension,
bounds=_aggregate_field_bounds(bounds_by_model, name),
)
)
components = _VECTOR_COMPONENTS[: min(dimension, 3)] if dimension > 1 else ()
for component_index, component in enumerate(components):
fields.append(
VisualizationFieldCapability(
name=f"{name}_{component}",
dimension=1,
source_name=name,
component=component,
bounds=_aggregate_field_bounds(
bounds_by_model,
name,
component_index,
),
)
)
return fields
def parse_visualization_capabilities(
*,
resource_id: str,
resource_type: ResourceType,
manifest: Any,
) -> ResourceVisualizationCapabilities:
"""Parse one visualization manifest into Report-addressable capabilities."""
nodes = _manifest_nodes(manifest)
nodes_by_id = {str(node.get("id")): node for node in nodes if node.get("id") is not None}
root = next((node for node in nodes if _is_root_group(node)), None)
bindings = _build_bindings(nodes, resource_type)
outputs: list[VisualizationOutputCapability] = []
for binding in bindings:
controller_owner_id_by_member_id: dict[str, str] = {}
for controller_owner_id, member_ids in binding.member_ids_by_controller_owner.items():
for member_id in member_ids:
controller_owner_id_by_member_id.setdefault(member_id, controller_owner_id)
targets = [
VisualizationTargetCapability(
id=member_id,
controller_owner_id=controller_owner_id_by_member_id.get(member_id),
type=(
str(nodes_by_id[member_id].get("type"))
if member_id in nodes_by_id and nodes_by_id[member_id].get("type") is not None
else None
),
)
for member_id in binding.member_ids
]
outputs.append(
VisualizationOutputCapability(
id=binding.config_owner_id,
type=binding.output_type,
controller_owner_ids=binding.controller_owner_ids,
fields=_field_capabilities(nodes_by_id, binding),
visibility_targets=targets,
)
)
manifest_version = root.get("version") if root is not None else None
return ResourceVisualizationCapabilities(
resource_id=resource_id,
resource_type=resource_type,
manifest_version=manifest_version if isinstance(manifest_version, str) else None,
outputs=outputs,
)
