"""Manifest-backed compatibility checks for visualization setups."""
from __future__ import annotations
from collections.abc import Sequence
from typing import Literal, TypeAlias
import pydantic as pd
from .capabilities import (
ResourceVisualizationCapabilities,
VisualizationOutputCapability,
VisualizationOutputType,
)
from .models import ReportBaseModel, ResourceType
from .scene import SceneVisualizationSetting
from .spec import ReportVisualizationSetup
VisualizationSetupIssueCode = Literal[
"resource-type-mismatch",
"missing-output",
"output-type-mismatch",
"missing-field-name",
"missing-visibility-target",
"missing-time-frame-target",
"missing-slice-variant",
]
VisualizationSetupIssueSeverity = Literal["error", "warning"]
VisualizationSetupInput: TypeAlias = (
ReportVisualizationSetup | Sequence[ReportVisualizationSetup] | SceneVisualizationSetting
)
_SCENE_FIELD_TYPE_TO_OUTPUT_TYPE: dict[str, VisualizationOutputType] = {
"surfaces": "surface",
"slices": "slice",
"isosurfaces": "isosurface",
"streamlines": "streamline",
}
__all__ = [
"VisualizationSetupCheckResult",
"VisualizationSetupIssue",
"check_visualization_setup",
]
class VisualizationSetupIssue(ReportBaseModel):
"""One incompatibility between a visualization setup and resource capabilities."""
code: VisualizationSetupIssueCode
severity: VisualizationSetupIssueSeverity
message: str = pd.Field(min_length=1)
path: str = pd.Field(min_length=1)
resource_id: str = pd.Field(min_length=1)
output: str | None = None
field: str | None = None
target: str | None = None
expected_type: VisualizationOutputType | None = None
actual_type: VisualizationOutputType | None = None
available_values: list[str] = pd.Field(default_factory=list)
class VisualizationSetupCheckResult(ReportBaseModel):
"""Structured result returned by :func:`check_visualization_setup`."""
resource_id: str = pd.Field(min_length=1)
resource_type: ResourceType
issues: list[VisualizationSetupIssue] = pd.Field(default_factory=list)
@pd.computed_field
@property
def is_compatible(self) -> bool:
"""Whether every checked setup reference is available."""
return not self.issues
@property
def errors(self) -> list[VisualizationSetupIssue]:
"""Return blocking compatibility issues."""
return [issue for issue in self.issues if issue.severity == "error"]
@property
def warnings(self) -> list[VisualizationSetupIssue]:
"""Return non-blocking compatibility issues."""
return [issue for issue in self.issues if issue.severity == "warning"]
def _available_output_ids(capabilities: ResourceVisualizationCapabilities) -> list[str]:
return [output.id for output in capabilities.outputs]
def _available_scene_owner_ids(
capabilities: ResourceVisualizationCapabilities,
output_type: VisualizationOutputType,
) -> list[str]:
return list(
dict.fromkeys(
owner_id
for output in capabilities.outputs
if output.type == output_type
for owner_id in (output.controller_owner_ids or [output.id])
)
)
def _get_scene_output(
capabilities: ResourceVisualizationCapabilities,
owner_id: str,
) -> VisualizationOutputCapability | None:
output = capabilities.get_output(owner_id)
if output is not None:
return output
return next(
(
candidate
for candidate in capabilities.outputs
if owner_id in candidate.controller_owner_ids
),
None,
)
def _available_target_ids(
output: VisualizationOutputCapability,
*,
controller_owner_id: str | None = None,
) -> list[str]:
has_owner_metadata = any(
target.controller_owner_id is not None for target in output.visibility_targets
)
target_ids = [
target.id
for target in output.visibility_targets
if controller_owner_id is None
or not has_owner_metadata
or target.controller_owner_id == controller_owner_id
]
# Workbench addresses the owner itself when an output has no child visibility targets.
return target_ids or [controller_owner_id or output.id]
def _missing_output_issue(
*,
capabilities: ResourceVisualizationCapabilities,
output_id: str,
path: str,
message: str | None = None,
available_values: Sequence[str] | None = None,
) -> VisualizationSetupIssue:
return VisualizationSetupIssue(
code="missing-output",
severity="error",
message=message
or f'Output "{output_id}" is not available for resource "{capabilities.resource_id}".',
path=path,
resource_id=capabilities.resource_id,
output=output_id,
available_values=(
list(available_values)
if available_values is not None
else _available_output_ids(capabilities)
),
)
def _check_output_type(
*,
output: VisualizationOutputCapability,
expected_type: VisualizationOutputType,
capabilities: ResourceVisualizationCapabilities,
path: str,
) -> VisualizationSetupIssue | None:
if output.type == expected_type:
return None
return VisualizationSetupIssue(
code="output-type-mismatch",
severity="error",
message=(
f'Output "{output.id}" has type "{output.type}", not "{expected_type}", '
f'for resource "{capabilities.resource_id}".'
),
path=path,
resource_id=capabilities.resource_id,
output=output.id,
expected_type=expected_type,
actual_type=output.type,
available_values=[output.type],
)
def _check_field(
*,
field_name: str | None,
output: VisualizationOutputCapability,
capabilities: ResourceVisualizationCapabilities,
path: str,
) -> VisualizationSetupIssue | None:
if field_name in (None, "", "None") or output.get_field(field_name) is not None:
return None
return VisualizationSetupIssue(
code="missing-field-name",
severity="warning",
message=(
f'Field "{field_name}" is not available under output "{output.id}" '
f'for resource "{capabilities.resource_id}".'
),
path=path,
resource_id=capabilities.resource_id,
output=output.id,
field=field_name,
available_values=list(output.field_names),
)
def _check_targets(
*,
target_ids: Sequence[str],
output: VisualizationOutputCapability,
capabilities: ResourceVisualizationCapabilities,
path: str,
code: Literal["missing-visibility-target", "missing-slice-variant"],
controller_owner_id: str | None = None,
) -> list[VisualizationSetupIssue]:
available_target_ids = _available_target_ids(
output,
controller_owner_id=controller_owner_id,
)
available_target_id_set = set(available_target_ids)
label = "Slice variant target" if code == "missing-slice-variant" else "Visibility target"
return [
VisualizationSetupIssue(
code=code,
severity="warning",
message=(
f'{label} "{target_id}" is not available under output "{output.id}" '
f'for resource "{capabilities.resource_id}".'
),
path=f'{path}["{target_id}"]',
resource_id=capabilities.resource_id,
output=output.id,
target=target_id,
available_values=available_target_ids,
)
for target_id in target_ids
if target_id not in available_target_id_set
]
def _check_manual_setup(
*,
setup: ReportVisualizationSetup,
capabilities: ResourceVisualizationCapabilities,
path: str,
) -> list[VisualizationSetupIssue]:
output = capabilities.get_output(setup.output)
if output is None:
return [
_missing_output_issue(
capabilities=capabilities,
output_id=setup.output,
path=f"{path}.output",
)
]
issues: list[VisualizationSetupIssue] = []
type_issue = _check_output_type(
output=output,
expected_type=setup.type,
capabilities=capabilities,
path=f"{path}.type",
)
if type_issue is not None:
issues.append(type_issue)
field_issue = _check_field(
field_name=setup.field,
output=output,
capabilities=capabilities,
path=f"{path}.field",
)
if field_issue is not None:
issues.append(field_issue)
slice_variant_ids = list(setup.slice_variants or {})
slice_variant_id_set = set(slice_variant_ids)
issues.extend(
_check_targets(
target_ids=[
target_id for target_id in setup.show or [] if target_id not in slice_variant_id_set
],
output=output,
capabilities=capabilities,
path=f"{path}.show",
code="missing-visibility-target",
)
)
issues.extend(
_check_targets(
target_ids=[
target_id for target_id in setup.hide or [] if target_id not in slice_variant_id_set
],
output=output,
capabilities=capabilities,
path=f"{path}.hide",
code="missing-visibility-target",
)
)
issues.extend(
_check_targets(
target_ids=slice_variant_ids,
output=output,
capabilities=capabilities,
path=f"{path}.slice_variants",
code="missing-slice-variant",
)
)
return issues
def _parse_scene_field_key(field_key: str) -> tuple[VisualizationOutputType, str] | None:
field_type, separator, output_id = field_key.partition("::")
output_type = _SCENE_FIELD_TYPE_TO_OUTPUT_TYPE.get(field_type)
if not separator or not output_id or output_type is None:
return None
return output_type, output_id
def _check_scene_setup(
*,
setup: SceneVisualizationSetting,
capabilities: ResourceVisualizationCapabilities,
) -> list[VisualizationSetupIssue]:
if setup.resource_type != capabilities.resource_type:
return [
VisualizationSetupIssue(
code="resource-type-mismatch",
severity="error",
message=(
f'Scene setting has resource type "{setup.resource_type}", not '
f'"{capabilities.resource_type}" for resource "{capabilities.resource_id}".'
),
path="setup.resource_type",
resource_id=capabilities.resource_id,
)
]
issues: list[VisualizationSetupIssue] = []
payload = setup.payload
field_settings = payload.field_setting or {}
missing_field_output_ids: set[str] = set()
for output_id, field_setting in field_settings.items():
output = capabilities.get_output(output_id)
if output is None:
issues.append(
_missing_output_issue(
capabilities=capabilities,
output_id=output_id,
path=f'setup.payload.field_setting["{output_id}"]',
)
)
missing_field_output_ids.add(output_id)
continue
field_issue = _check_field(
field_name=field_setting.field_name,
output=output,
capabilities=capabilities,
path=f'setup.payload.field_setting["{output_id}"].field_name',
)
if field_issue is not None:
issues.append(field_issue)
for output_id in payload.time_frame_setting or {}:
if capabilities.get_output(output_id) is not None:
continue
issues.append(
VisualizationSetupIssue(
code="missing-time-frame-target",
severity="warning",
message=(
f'Time-frame target "{output_id}" is not available for resource '
f'"{capabilities.resource_id}".'
),
path=f'setup.payload.time_frame_setting["{output_id}"]',
resource_id=capabilities.resource_id,
output=output_id,
available_values=_available_output_ids(capabilities),
)
)
slice_variant_id_set = set(payload.slice_variant_setting or {})
for field_key, visibility_setting in (payload.visible_setting_by_field or {}).items():
parsed = _parse_scene_field_key(field_key)
# Workbench Scenes retain inactive diagnostic groups whose targets are all hidden.
# They do not affect the rendered setup unless that output also has a field setting.
if not any(visibility_setting.values()) and (
parsed is None or parsed[1] not in field_settings
):
continue
if parsed is None:
message = f'Visibility owner key "{field_key}" does not match an available output.'
issues.append(
_missing_output_issue(
capabilities=capabilities,
output_id=field_key,
path=f'setup.payload.visible_setting_by_field["{field_key}"]',
message=message,
)
)
continue
expected_type, output_id = parsed
output = _get_scene_output(capabilities, output_id)
if output is None:
if output_id not in missing_field_output_ids:
issues.append(
_missing_output_issue(
capabilities=capabilities,
output_id=output_id,
path=f'setup.payload.visible_setting_by_field["{field_key}"]',
available_values=_available_scene_owner_ids(
capabilities,
expected_type,
),
)
)
continue
type_issue = _check_output_type(
output=output,
expected_type=expected_type,
capabilities=capabilities,
path=f'setup.payload.visible_setting_by_field["{field_key}"]',
)
if type_issue is not None:
issues.append(type_issue)
active_target_ids = [
target_id for target_id, visible in visibility_setting.items() if visible
]
issues.extend(
_check_targets(
target_ids=[
target_id
for target_id in active_target_ids
if expected_type != "slice" or target_id not in slice_variant_id_set
],
output=output,
capabilities=capabilities,
path=f'setup.payload.visible_setting_by_field["{field_key}"]',
code="missing-visibility-target",
controller_owner_id=output_id,
)
)
if expected_type == "slice":
issues.extend(
_check_targets(
target_ids=[
target_id
for target_id in active_target_ids
if target_id in slice_variant_id_set
],
output=output,
capabilities=capabilities,
path="setup.payload.slice_variant_setting",
code="missing-slice-variant",
controller_owner_id=output_id,
)
)
return issues
[docs]
def check_visualization_setup(
setup: VisualizationSetupInput,
capabilities: ResourceVisualizationCapabilities,
) -> VisualizationSetupCheckResult:
"""Check exact setup references against one resource's visualization capabilities.
Run this check for every selected resource before ``Report.create(...)`` or a
configuration-changing ``report.update(...)``. This function performs no network
requests and does not mutate or translate the setup. Obtain ``capabilities``
separately with :func:`get_capabilities` and reuse them for configuration decisions.
AI-generated workflows should follow the complete sequence in ``get_agent_guide()``.
"""
if not isinstance(capabilities, ResourceVisualizationCapabilities):
raise TypeError("capabilities must be a ResourceVisualizationCapabilities object")
if isinstance(setup, SceneVisualizationSetting):
issues = _check_scene_setup(setup=setup, capabilities=capabilities)
elif isinstance(setup, ReportVisualizationSetup):
issues = _check_manual_setup(setup=setup, capabilities=capabilities, path="setup")
elif isinstance(setup, Sequence) and not isinstance(setup, (str, bytes)):
issues = []
for index, item in enumerate(setup):
if not isinstance(item, ReportVisualizationSetup):
raise TypeError(
"setup sequences must contain only Report.VisualizationSetup objects"
)
issues.extend(
_check_manual_setup(
setup=item,
capabilities=capabilities,
path=f"setup[{index}]",
)
)
else:
raise TypeError(
"setup must be a Report.VisualizationSetup, a sequence of setups, "
"or a Scene visualization setting"
)
return VisualizationSetupCheckResult(
resource_id=capabilities.resource_id,
resource_type=capabilities.resource_type,
issues=issues,
)
