"""User-facing report configuration and internal config compilation."""
from __future__ import annotations
import math
import re
from collections.abc import Callable, Iterable, Mapping, Sequence
from typing import Any, Literal, TypeAlias
from urllib.parse import quote
import pydantic as pd
from . import models
from .resource_types import (
CaseResourceSelector,
ReportResourceInput,
ReportResourceSelector,
)
from .scene import SceneViewpoint, SceneVisualizationSetting
SummaryResourceType = Literal["geometry", "surface_mesh", "volume_mesh", "case"]
GeometrySummaryField = Literal["bodies", "patches", "edges"]
SurfaceMeshSummaryField = Literal["max_edge_length", "nodes", "triangles", "quadrilaterals"]
VolumeMeshSummaryField = Literal[
"first_layer_thickness",
"nodes",
"tetrahedrons",
"prisms",
"pyramids",
"hexahedrons",
]
CaseSummaryField = Literal[
"velocity",
"alpha",
"beta",
"turbulence_model",
"transition_model",
"pseudo_steps",
"physical_steps",
"area",
"oal",
"oah",
"oaw",
"wb",
"cl",
"cd",
"clf",
"clr",
"cs",
"cd_area",
]
SummaryField: TypeAlias = (
GeometrySummaryField | SurfaceMeshSummaryField | VolumeMeshSummaryField | CaseSummaryField
)
VisualizationLayout = Literal["grid", "single"]
VisualizationResolution = Literal["low", "high"]
ChartComparison = Literal["absolute", "delta"]
ChartStyle = Literal["case", "variable"]
ChartBackgroundView = Literal["left", "back", "top"]
ChartSeriesDisplay = Literal["individual", "cumulative", "summed"]
ForceScopeMode = Literal["total", "faces", "body_groups"]
CameraLengthUnit = Literal["m", "cm", "mm", "inch", "ft"]
BuiltInChartXVariable = Literal[
"alpha",
"beta",
"velocity",
"first_layer_thickness",
"surface_max_edge_length",
"pseudo_step",
"physical_step",
"coordinate_x",
"coordinate_y",
]
BuiltInChartYVariable = Literal[
"CL",
"CD",
"CFx",
"CFy",
"CFz",
"CMx",
"CMy",
"CMz",
"CLPressure",
"CDPressure",
"CFxPressure",
"CFyPressure",
"CFzPressure",
"CMxPressure",
"CMyPressure",
"CMzPressure",
"CLSkinFriction",
"CDSkinFriction",
"CFxSkinFriction",
"CFySkinFriction",
"CFzSkinFriction",
"CMxSkinFriction",
"CMySkinFriction",
"CMzSkinFriction",
]
VisualizationFieldType = Literal["surface", "slice", "isosurface", "streamline"]
VisualizationClip = Literal["none", "above", "below", "range"]
class _ReportSpec(pd.BaseModel):
"""Base model for the small set of options exposed by the Report UI."""
model_config = pd.ConfigDict(
extra="forbid",
arbitrary_types_allowed=True,
validate_default=True,
str_strip_whitespace=True,
allow_inf_nan=False,
)
_SUMMARY_FIELDS: dict[SummaryResourceType, dict[str, str]] = {
"geometry": {
"bodies": "bodies",
"patches": "patches",
"edges": "edges",
},
"surface_mesh": {
"max_edge_length": "maxEdgeLength",
"nodes": "nodes",
"triangles": "triangles",
"quadrilaterals": "quadrilaterals",
},
"volume_mesh": {
"first_layer_thickness": "firstLayerThickness",
"nodes": "nodes",
"tetrahedrons": "tetrahedrons",
"prisms": "prisms",
"pyramids": "pyramids",
"hexahedrons": "hexahedrons",
},
"case": {
"velocity": "velocity",
"alpha": "alphaAngle",
"beta": "betaAngle",
"turbulence_model": "turbulenceModel",
"transition_model": "transitionModel",
"pseudo_steps": "pseudoSteps",
"physical_steps": "physicalSteps",
"area": "area",
"oal": "oal",
"oah": "oah",
"oaw": "oaw",
"wb": "wb",
"cl": "cl",
"cd": "cd",
"clf": "clf",
"clr": "clr",
"cs": "cs",
"cd_area": "cdArea",
},
}
_DEFAULT_CASE_SUMMARY_FIELDS = frozenset(
{
"velocity",
"alpha",
"beta",
"turbulence_model",
"transition_model",
"pseudo_steps",
"physical_steps",
"cl",
"cd",
}
)
[docs]
class ReportSummary(_ReportSpec):
"""Customize one of the four summary tables included in every report."""
resource_type: SummaryResourceType
title: str | None = pd.Field(default=None, min_length=1)
fields: list[SummaryField] | None = None
force_scope: ForceScopeMode | None = None
force_scope_ids: list[str] | None = None
@pd.model_validator(mode="before")
@classmethod
def _normalize_fields(cls, values):
if not isinstance(values, Mapping):
return values
resource_type = values.get("resource_type")
fields = values.get("fields")
if fields is None or resource_type not in _SUMMARY_FIELDS:
return values
if isinstance(fields, str) or not isinstance(fields, (list, tuple)):
return values
normalized = _normalize_unique_strings(list(fields), "fields")
allowed = _SUMMARY_FIELDS[resource_type]
canonical_by_case = {field.casefold(): field for field in allowed}
canonical_fields: list[str] = []
unsupported: list[str] = []
for field in normalized:
canonical = canonical_by_case.get(field.casefold())
if canonical is None:
unsupported.append(field)
else:
canonical_fields.append(canonical)
if unsupported:
raise ValueError(
f"unsupported {resource_type} summary fields: "
+ ", ".join(unsupported)
+ "; choose from "
+ ", ".join(allowed)
)
if len(set(canonical_fields)) != len(canonical_fields):
raise ValueError("fields must not contain duplicate values")
return {**values, "fields": canonical_fields}
[docs]
@classmethod
def supported_fields(cls, resource_type: SummaryResourceType) -> tuple[SummaryField, ...]:
"""Return the accepted field names for one summary resource type."""
return tuple(_SUMMARY_FIELDS[resource_type])
@pd.field_validator("force_scope_ids")
@classmethod
def _normalize_force_scope_ids(cls, values: list[str] | None) -> list[str] | None:
if values is None:
return None
return _normalize_unique_strings(values, "force_scope_ids")
@pd.model_validator(mode="after")
def _validate_force_scope(self):
if (
self.force_scope is not None or self.force_scope_ids is not None
) and self.resource_type != "case":
raise ValueError("force_scope options are available only for case summaries")
_validate_force_scope_parameters(self.force_scope, self.force_scope_ids)
return self
[docs]
class ReportCamera(_ReportSpec):
"""Camera used by linked views, with optional input units for length values."""
position: tuple[float, float, float] | None = None
look_at: tuple[float, float, float] | None = None
pan_target: tuple[float, float, float] | None = None
up: tuple[float, float, float] | None = None
dimension: float | None = pd.Field(default=None, gt=0)
dimension_dir: Literal["width", "height", "diagonal"] | None = None
unit: CameraLengthUnit | None = None
[docs]
class ReportVisualizationSetup(_ReportSpec):
"""Configure one surface, slice, isosurface, or streamline output.
Use exact IDs returned by ``get_capabilities(resource)`` and check this setup
against that resource before creating or updating a Report. Name normalization
can suggest candidates but does not produce authoritative IDs.
"""
output: str = pd.Field(min_length=1)
type: VisualizationFieldType
field: str | None = pd.Field(default=None, min_length=1)
show: list[str] | None = None
hide: list[str] | None = None
log_scale: bool | None = None
range: tuple[float, float] | None = None
color_range: tuple[float, float] | None = None
theme: str | None = pd.Field(default=None, min_length=1)
contour_mode: Literal["surface", "contours", "both"] | None = None
contour_steps: int | None = pd.Field(default=None, gt=0)
contour_line_color: int | None = pd.Field(default=None, ge=0)
solid_color: int | None = pd.Field(default=None, ge=0)
clip: VisualizationClip | None = None
clip_value: float | None = None
clip_range: tuple[float, float] | None = None
show_color_map: bool | None = None
streamline_direction: Literal["upstream", "downstream", "both"] | None = None
render_type: Literal["ribbon", "line"] | None = None
tube_width: float | None = pd.Field(default=None, gt=0)
tube_width_unit: str | None = pd.Field(default=None, min_length=1)
ribbon_width: float | None = pd.Field(default=None, gt=0)
ribbon_width_unit: str | None = pd.Field(default=None, min_length=1)
ribbon_angle_scale: float | None = pd.Field(default=None, gt=0)
visualizer: Literal["lic"] | None = None
unit: str | None = pd.Field(default=None, min_length=1)
use_local_value: bool | None = None
time_frame: int | None = pd.Field(default=None, ge=0)
slice_variants: dict[str, Literal["flat", "crinkled"]] | None = None
wireframes: dict[str, bool] | None = None
custom_colors: list[tuple[float, str]] | None = pd.Field(default=None, min_length=2)
@pd.field_validator("show", "hide")
@classmethod
def _validate_visibility_ids(cls, values: list[str] | None, info) -> list[str] | None:
if values is None:
return None
return _normalize_unique_strings(values, info.field_name)
@pd.field_validator("slice_variants", "wireframes")
@classmethod
def _validate_id_mapping(cls, values: dict[str, Any] | None, info):
if values is None:
return None
normalized = {}
for item_id, value in values.items():
if not isinstance(item_id, str) or not item_id.strip():
raise TypeError(f"{info.field_name} keys must be non-empty strings")
normalized[item_id.strip()] = value
if len(normalized) != len(values):
raise ValueError(f"{info.field_name} must not contain duplicate IDs")
return normalized
@pd.field_validator("custom_colors")
@classmethod
def _validate_custom_colors(
cls, values: list[tuple[float, str]] | None
) -> list[tuple[float, str]] | None:
if values is None:
return None
positions: set[float] = set()
normalized = []
for position, color in values:
if position < 0 or position > 100:
raise ValueError("custom color positions must be between 0 and 100")
if position in positions:
raise ValueError("custom color positions must be unique")
if re.fullmatch(r"#[0-9a-fA-F]{6}(?:[0-9a-fA-F]{2})?", color) is None:
raise ValueError("custom colors must use #RRGGBB or #RRGGBBAA")
positions.add(position)
normalized.append((position, color.lower()))
return sorted(normalized)
@pd.model_validator(mode="after")
def _validate_setup(self):
_validate_range(self.range, "range")
_validate_range(self.color_range, "color_range")
_validate_range(self.clip_range, "clip_range")
overlap = set(self.show or []) & set(self.hide or [])
if overlap:
raise ValueError("show and hide must not contain the same IDs")
if self.clip == "range" and self.clip_range is None:
raise ValueError("clip_range is required when clip is 'range'")
if self.clip != "range" and self.clip_range is not None:
raise ValueError("clip_range is available only when clip is 'range'")
if self.clip not in ("above", "below") and self.clip_value is not None:
raise ValueError("clip_value is available only when clip is 'above' or 'below'")
contour_options = (self.contour_mode, self.contour_steps, self.contour_line_color)
if self.type not in ("surface", "slice") and any(
value is not None for value in contour_options
):
raise ValueError("contour options are available only for surface or slice outputs")
streamline_options = (
self.streamline_direction,
self.render_type,
self.tube_width,
self.tube_width_unit,
self.ribbon_width,
self.ribbon_width_unit,
self.ribbon_angle_scale,
)
if self.type != "streamline" and any(value is not None for value in streamline_options):
raise ValueError("streamline options are available only for streamline outputs")
if self.type != "slice" and self.slice_variants is not None:
raise ValueError("slice_variants is available only for slice outputs")
field_options = (
self.log_scale,
self.range,
self.color_range,
self.theme,
*contour_options,
self.clip,
self.clip_value,
self.clip_range,
self.show_color_map,
self.visualizer,
self.unit,
self.use_local_value,
self.time_frame,
self.custom_colors,
)
if self.field is None and any(value is not None for value in field_options):
raise ValueError("field is required when field display options are configured")
if self.custom_colors is not None:
if self.theme not in (None, "custom"):
raise ValueError("theme must be 'custom' when custom_colors is configured")
object.__setattr__(self, "theme", "custom")
return self
def _accept_visualization_setups(value):
if isinstance(value, SceneVisualizationSetting):
return value
if value is None:
return []
if isinstance(value, (ReportVisualizationSetup, Mapping)):
return [value]
return value
def _validate_visualization_setups(setups: list[ReportVisualizationSetup]) -> None:
outputs = [setup.output for setup in setups]
if len(set(outputs)) != len(outputs):
raise ValueError("setup must not configure the same output more than once")
if sum(setup.custom_colors is not None for setup in setups) > 1:
raise ValueError("custom_colors can be configured by only one visualization setup")
[docs]
class ReportVisualizationView(_ReportSpec):
"""Validated independent state stored for one visualization resource view."""
setup: list[ReportVisualizationSetup] | SceneVisualizationSetting = pd.Field(
default_factory=list
)
camera: ReportCamera | SceneViewpoint | None = None
@pd.field_validator("setup", mode="before")
@classmethod
def _accept_single_setup(cls, value):
return _accept_visualization_setups(value)
@pd.model_validator(mode="after")
def _validate_config(self):
if isinstance(self.setup, list):
_validate_visualization_setups(self.setup)
return self
[docs]
class ReportVisualization(_ReportSpec):
"""Add a configured 3D resource comparison section to a report.
AI-generated workflows should first load ``get_agent_guide()``. Before reusing one
setup across resources, obtain each resource's capabilities and call
``check_visualization_setup`` before creating or updating the Report.
"""
title: str = pd.Field(default="Visualization", min_length=1)
# The full config compiler resolves and validates resource objects or IDs against its pool.
resources: list[pd.SkipValidation[ReportResourceSelector]] | None = None
layout: VisualizationLayout = "grid"
resolution: VisualizationResolution | None = None
setup: list[ReportVisualizationSetup] | SceneVisualizationSetting = pd.Field(
default_factory=list
)
camera: ReportCamera | SceneViewpoint | None = None
views: dict[str, ReportVisualizationView] = pd.Field(default_factory=dict)
@pd.field_validator("setup", mode="before")
@classmethod
def _accept_single_setup(cls, value):
return _accept_visualization_setups(value)
@pd.field_validator("views", mode="before")
@classmethod
def _validate_views(cls, value):
if value is None:
return {}
if not isinstance(value, Mapping):
raise TypeError("views must map resource IDs to setup and camera configuration")
normalized = {}
for resource_id, config in value.items():
if not isinstance(resource_id, str) or not resource_id.strip():
raise TypeError("views keys must be non-empty resource ID strings")
resource_id = resource_id.strip()
if resource_id in normalized:
raise ValueError("views must not contain duplicate resource IDs")
normalized[resource_id] = ReportVisualizationView.model_validate(config)
return normalized
@pd.model_validator(mode="after")
def _validate_setups(self):
if isinstance(self.setup, list):
_validate_visualization_setups(self.setup)
return self
def _encode_axis_id_segment(value: str) -> str:
"""Match JavaScript's encodeURIComponent for persisted Chart2D axis ids."""
return quote(value, safe="-_.!~*'()")
_X_AXIS_IDS = {
"alpha": "axis:x:simulation:alpha-angle",
"beta": "axis:x:simulation:beta-angle",
"velocity": "axis:x:simulation:velocity",
"first_layer_thickness": "axis:x:simulation:first-layer-thickness",
"surface_max_edge_length": "axis:x:simulation:surface-max-edge-length",
"pseudo_step": "axis:x:step:pseudo",
"physical_step": "axis:x:step:physical",
"coordinate_x": "axis:x:spatial:X",
"coordinate_y": "axis:x:spatial:Y",
}
_FORCE_VARIABLES = ("CL", "CD", "CFx", "CFy", "CFz", "CMx", "CMy", "CMz")
_FORCE_HISTORY_VARIABLES = (
*_FORCE_VARIABLES,
*(f"{variable}Pressure" for variable in _FORCE_VARIABLES),
*(f"{variable}SkinFriction" for variable in _FORCE_VARIABLES),
)
_FORCE_Y_AXIS_IDS = {
variable.lower(): f"axis:y:force:{_encode_axis_id_segment(variable)}"
for variable in _FORCE_HISTORY_VARIABLES
}
_SERIES_VALUE_COLUMN = "__series__"
_Y_AXIS_IDS = _FORCE_Y_AXIS_IDS
_Y_VARIABLE_NAMES_BY_NORMALIZED = {
variable.lower(): variable for variable in _FORCE_HISTORY_VARIABLES
}
[docs]
class ReportChartVariable(_ReportSpec):
"""A CSV or user-defined-dynamics variable selected in the Report UI."""
source: Literal["csv", "udd"] = "csv"
file_name: str = pd.Field(min_length=1)
column: str = pd.Field(min_length=1)
model_config = pd.ConfigDict(frozen=True)
@pd.field_validator("file_name")
@classmethod
def _normalize_file_name(cls, value: str) -> str:
normalized = value.lstrip("/")
if normalized.startswith("results/"):
normalized = normalized[len("results/") :]
if not normalized:
raise ValueError("file_name must not be empty")
return normalized
[docs]
@classmethod
def csv(cls, file_name: str, column: str) -> "ReportChartVariable":
"""Select a CSV output column."""
return cls(source="csv", file_name=file_name, column=column)
[docs]
@classmethod
def udd(cls, file_name: str, column: str) -> "ReportChartVariable":
"""Select a user-defined-dynamics output column."""
return cls(source="udd", file_name=file_name, column=column)
[docs]
@classmethod
def csv_series(cls, file_name: str) -> "ReportChartVariable":
"""Select a metadata-defined series group from a CSV output."""
return cls(source="csv", file_name=file_name, column=_SERIES_VALUE_COLUMN)
[docs]
@classmethod
def udd_series(cls, file_name: str) -> "ReportChartVariable":
"""Select a metadata-defined series group from a UDD output."""
return cls(source="udd", file_name=file_name, column=_SERIES_VALUE_COLUMN)
[docs]
class ReportChart(_ReportSpec):
"""Add a 2D chart section using the controls available in the Report UI."""
title: str = pd.Field(default="2D Chart", min_length=1)
# The full config compiler resolves and validates Case objects or IDs against its pool.
resources: list[pd.SkipValidation[CaseResourceSelector]] | None = None
x: BuiltInChartXVariable = "alpha"
y: list[BuiltInChartYVariable | ReportChartVariable] = pd.Field(
default_factory=lambda: ["CL"], min_length=1, max_length=5
)
comparison: ChartComparison = "absolute"
x_range: tuple[float, float] | None = None
y_ranges: dict[BuiltInChartYVariable | ReportChartVariable, tuple[float, float]] | None = None
log_scale: bool = False
style: ChartStyle = "case"
background_view: ChartBackgroundView | None = None
force_scope: ForceScopeMode | None = None
force_scope_ids: list[str] | None = None
series: list[str] | None = None
series_display: ChartSeriesDisplay | None = None
@pd.field_validator("x", mode="before")
@classmethod
def _normalize_x_variable(cls, value):
if not isinstance(value, str):
return value
return _canonical_x_variable(value)
@pd.field_validator("y", mode="before")
@classmethod
def _normalize_y_variables(cls, value):
if isinstance(value, (str, ReportChartVariable)):
value = [value]
if not isinstance(value, (list, tuple)):
return value
return [
_canonical_y_variable(variable, "y") if isinstance(variable, str) else variable
for variable in value
]
@pd.field_validator("y_ranges", mode="before")
@classmethod
def _normalize_y_range_variables(cls, value):
if value is None or not isinstance(value, Mapping):
return value
normalized = {}
for variable, value_range in value.items():
canonical = (
_canonical_y_variable(variable, "y_ranges")
if isinstance(variable, str)
else variable
)
if canonical in normalized:
raise ValueError(f"y_ranges contains duplicate y variable {variable!r}")
normalized[canonical] = value_range
return normalized
@pd.field_validator("force_scope_ids", "series")
@classmethod
def _normalize_id_list(cls, values: list[str] | None, info) -> list[str] | None:
if values is None:
return None
return _normalize_unique_strings(values, info.field_name)
[docs]
@classmethod
def supported_x_variables(cls) -> tuple[BuiltInChartXVariable, ...]:
"""Return the built-in x-axis variables."""
return tuple(_X_AXIS_IDS)
[docs]
@classmethod
def supported_y_variables(cls) -> tuple[BuiltInChartYVariable, ...]:
"""Return the built-in y-axis variables."""
return _FORCE_HISTORY_VARIABLES
@pd.model_validator(mode="after")
def _validate_chart_options(self):
_validate_force_scope_parameters(self.force_scope, self.force_scope_ids)
x_axis_id = _x_axis_id(self.x)
y_axis_ids = [_y_axis_id(value, "y") for value in self.y]
if len(set(y_axis_ids)) != len(y_axis_ids):
raise ValueError("y must not contain duplicate variables")
_validate_chart_variables(self.x, self.y)
_validate_range(self.x_range, "x_range")
if self.y_ranges is not None:
normalized_ranges: dict[str, tuple[float, float]] = {}
for variable, value_range in self.y_ranges.items():
y_axis_id = _y_axis_id(variable, "y_ranges")
if y_axis_id not in y_axis_ids:
raise ValueError(f"y_ranges contains unselected y variable {variable!r}")
if y_axis_id in normalized_ranges:
raise ValueError(f"y_ranges contains duplicate y variable {variable!r}")
_validate_range(value_range, f"y_ranges[{variable!r}]")
normalized_ranges[y_axis_id] = value_range
if self.comparison == "delta" and self.log_scale:
raise ValueError("log_scale is unavailable when comparison is 'delta'")
if self.background_view is not None:
allowed_views = {
"axis:x:spatial:X": {"left", "top"},
"axis:x:spatial:Y": {"back", "top"},
}.get(x_axis_id)
if allowed_views is None:
raise ValueError(
"background_view is only available for coordinate_x or coordinate_y charts"
)
if self.background_view not in allowed_views:
raise ValueError(
f"background_view for {self.x} must be one of "
+ ", ".join(sorted(allowed_views))
)
is_step_history = self.x in {"pseudo_step", "physical_step"}
has_force_variable = any(
isinstance(variable, str) and variable in _FORCE_HISTORY_VARIABLES
for variable in self.y
)
if self.force_scope is not None and not (is_step_history and has_force_variable):
raise ValueError(
"force_scope requires pseudo_step or physical_step with a force Y variable"
)
has_metadata_variable = any(
isinstance(variable, ReportChartVariable) for variable in self.y
)
if (
self.series is not None or self.series_display is not None
) and not has_metadata_variable:
raise ValueError("series options require a CSV, UDD, or solver-metric Y variable")
return self
ReportSectionSpec: TypeAlias = ReportSummary | ReportVisualization | ReportChart
def _normalize_unique_strings(values: list[str], field_name: str) -> list[str]:
normalized: list[str] = []
for value in values:
if not isinstance(value, str):
raise TypeError(f"{field_name} must contain only strings")
cleaned = value.strip()
if not cleaned:
raise ValueError(f"{field_name} must not contain empty values")
if cleaned in normalized:
raise ValueError(f"{field_name} must not contain duplicate values")
normalized.append(cleaned)
return normalized
def _validate_force_scope_parameters(
mode: ForceScopeMode | None, included_ids: list[str] | None
) -> None:
if mode in (None, "total") and included_ids:
raise ValueError("force_scope_ids must be empty when force_scope is unset or 'total'")
if mode in ("faces", "body_groups") and not included_ids:
raise ValueError(f"force_scope_ids must not be empty when force_scope is {mode!r}")
def _normalize_variable(value: str, field_name: str) -> str:
if not isinstance(value, str):
raise TypeError(f"{field_name} variables must be strings")
normalized = value.strip()
if not normalized:
raise ValueError(f"{field_name} variables must not be empty")
return normalized
def _x_axis_id(value: str) -> str:
return _X_AXIS_IDS[_canonical_x_variable(value)]
def _y_axis_id(value: str | ReportChartVariable, field_name: str) -> str:
if isinstance(value, ReportChartVariable):
return (
f"axis:y:{value.source}:"
f"{_encode_axis_id_segment(value.file_name)}:"
f"{_encode_axis_id_segment(value.column)}"
)
canonical = _canonical_y_variable(value, field_name)
return _Y_AXIS_IDS[canonical.lower()]
def _canonical_x_variable(value: str) -> BuiltInChartXVariable:
normalized = _normalize_variable(value, "x")
canonical = next(
(variable for variable in _X_AXIS_IDS if variable.lower() == normalized.lower()), None
)
if canonical is not None:
return canonical
raise ValueError(f"unsupported x variable {value!r}; choose from " + ", ".join(_X_AXIS_IDS))
def _canonical_y_variable(value: str, field_name: str) -> BuiltInChartYVariable:
normalized = _normalize_variable(value, field_name)
canonical = _Y_VARIABLE_NAMES_BY_NORMALIZED.get(normalized.lower())
if canonical is not None:
return canonical
raise ValueError(
f"unsupported {field_name} variable {value!r}; choose from "
+ ", ".join(_FORCE_HISTORY_VARIABLES)
+ ", or use ReportChartVariable for CSV and UDD outputs"
)
def _validate_chart_variables(
x_variable: BuiltInChartXVariable,
y_variables: list[BuiltInChartYVariable | ReportChartVariable],
) -> None:
if x_variable in {
"alpha",
"beta",
"velocity",
"first_layer_thickness",
"surface_max_edge_length",
}:
incompatible_variables = [
value
for value in y_variables
if not (isinstance(value, str) and value in _FORCE_VARIABLES)
]
elif x_variable in {"coordinate_x", "coordinate_y"}:
incompatible_variables = [
value for value in y_variables if not isinstance(value, ReportChartVariable)
]
else:
incompatible_variables = [
value
for value in y_variables
if not (
isinstance(value, ReportChartVariable)
or (isinstance(value, str) and value in _FORCE_HISTORY_VARIABLES)
)
]
incompatible = [_y_axis_id(value, "y") for value in incompatible_variables]
if incompatible:
raise ValueError(
f"y variables are not available for x variable {_x_axis_id(x_variable)}: "
+ ", ".join(incompatible)
)
def _validate_range(value_range: tuple[float, float] | None, field_name: str) -> None:
if value_range is not None and value_range[0] >= value_range[1]:
raise ValueError(f"{field_name} minimum must be less than its maximum")
def _resource_value(asset: ReportResourceInput, name: str) -> Any:
try:
return getattr(asset, name, None)
except (AttributeError, NotImplementedError):
return None
def _normalize_aliases(aliases: Mapping[str, str] | None) -> dict[str, str]:
if aliases is None:
return {}
if not isinstance(aliases, Mapping):
raise TypeError("aliases must map cloud resource IDs to display names")
normalized: dict[str, str] = {}
for resource_id, alias in aliases.items():
if not isinstance(resource_id, str) or not resource_id.strip():
raise TypeError("aliases keys must be cloud resource IDs")
if not isinstance(alias, str) or not alias.strip():
raise TypeError("aliases values must be non-empty strings")
normalized[resource_id.strip()] = alias.strip()
return normalized
def _normalize_reference_id(reference: CaseResourceSelector | None) -> str | None:
if reference is None:
return None
reference_id = reference if isinstance(reference, str) else _resource_value(reference, "id")
if not isinstance(reference_id, str) or not reference_id.strip():
raise ValueError("reference must provide a non-empty cloud resource ID")
return reference_id.strip()
def _assign_default_reference(
resources: list[models.ReportResource], reference_id: str | None
) -> list[models.ReportResource]:
if reference_id is not None:
return resources
for index, resource in enumerate(resources):
if resource.type == "Case":
resources[index] = resource.model_copy(update={"is_reference": True})
break
return resources
def _build_resources(
assets: Iterable[ReportResourceInput],
aliases: Mapping[str, str] | None,
reference: CaseResourceSelector | None,
) -> list[models.ReportResource]:
if isinstance(assets, (str, bytes)):
raise TypeError("resources must be an iterable of Flow360 cloud resources")
asset_list = list(assets)
if not asset_list:
raise ValueError("resources must contain at least one Flow360 cloud resource")
normalized_aliases = _normalize_aliases(aliases)
reference_id = _normalize_reference_id(reference)
resources: list[models.ReportResource] = []
for asset in asset_list:
resource_type = _resource_value(asset, "_cloud_resource_type_name")
if resource_type not in ("Case", "Geometry", "SurfaceMesh", "VolumeMesh"):
raise TypeError(
"resources must contain only Flow360 Case, Geometry, SurfaceMesh, "
"or VolumeMesh objects"
)
resource_id = _resource_value(asset, "id")
if not isinstance(resource_id, str) or not resource_id.strip():
raise ValueError("every report resource must have a cloud resource ID")
resource_id = resource_id.strip()
original_name = _resource_value(asset, "name")
if not isinstance(original_name, str) or not original_name.strip():
raise ValueError(f"report resource {resource_id} must have a name")
original_name = original_name.strip()
resources.append(
models.ReportResource(
type=resource_type,
id=resource_id,
project_id=_resource_value(asset, "project_id"),
original_name=original_name,
name=normalized_aliases.get(resource_id, original_name),
is_reference=resource_id == reference_id,
)
)
known_ids = {resource.id for resource in resources}
unknown_aliases = set(normalized_aliases) - known_ids
if unknown_aliases:
raise ValueError(
"aliases contains unknown resource IDs: " + ", ".join(sorted(unknown_aliases))
)
if reference_id is not None:
reference_resource = next(
(resource for resource in resources if resource.id == reference_id), None
)
if reference_resource is None:
raise ValueError("reference must be one of the report resources")
if reference_resource.type != "Case":
raise ValueError("reference must be a Flow360 Case")
return _assign_default_reference(resources, reference_id)
def _selected_resources(
selected: Sequence[ReportResourceSelector] | None,
report_resources: list[models.ReportResource],
*,
section_name: str,
) -> list[models.ReportResource]:
if selected is None:
cases = [resource for resource in report_resources if resource.type == "Case"]
if cases:
return cases
first_type = report_resources[0].type
return [resource for resource in report_resources if resource.type == first_type]
if not selected:
raise ValueError(f"{section_name} resources must not be empty")
resources_by_id = {resource.id: resource for resource in report_resources}
resolved: list[models.ReportResource] = []
seen: set[str] = set()
for item in selected:
resource_id = item.strip() if isinstance(item, str) else _resource_value(item, "id")
if not isinstance(resource_id, str) or not resource_id.strip():
raise TypeError(
f"{section_name} resources must be report resource objects or resource IDs"
)
resource_id = resource_id.strip()
if resource_id in seen:
raise ValueError(f"{section_name} resources must not contain duplicates")
resource = resources_by_id.get(resource_id)
if resource is None:
raise ValueError(f"{section_name} selects resource {resource_id} not in the report")
seen.add(resource_id)
resolved.append(resource)
return resolved
def _compile_force_scope(
mode: ForceScopeMode | None, included_ids: list[str] | None
) -> models.ReportForceScope:
if mode is None or mode == "total":
return models.ReportTotalForceScope()
return models.ReportSelectedForceScope(
mode="bodyGroups" if mode == "body_groups" else "faces",
included_ids=included_ids or [],
)
def _compile_summary(section: ReportSummary):
field_map = _SUMMARY_FIELDS[section.resource_type]
if section.fields is not None:
selected_fields = set(section.fields)
elif section.resource_type == "case":
selected_fields = set(_DEFAULT_CASE_SUMMARY_FIELDS)
else:
selected_fields = set(field_map)
hidden_fields = [
internal for public, internal in field_map.items() if public not in selected_fields
]
if section.resource_type == "geometry":
return models.GeometrySummarySection(
title=section.title or "Geometry summary",
config=models.GeometrySummarySectionConfig(
table=models.GeometrySummaryTableConfig(
hidden_field_keys=hidden_fields,
)
),
)
if section.resource_type == "surface_mesh":
return models.SurfaceMeshSummarySection(
title=section.title or "Surface mesh summary",
config=models.SurfaceMeshSummarySectionConfig(
table=models.SurfaceMeshSummaryTableConfig(
hidden_field_keys=hidden_fields,
)
),
)
if section.resource_type == "volume_mesh":
return models.VolumeMeshSummarySection(
title=section.title or "Volume mesh summary",
config=models.VolumeMeshSummarySectionConfig(
table=models.VolumeMeshSummaryTableConfig(
hidden_field_keys=hidden_fields,
)
),
)
return models.CaseSummarySection(
title=section.title or "Case summary",
config=models.CaseSummarySectionConfig(
table=models.CaseSummaryTableConfig(
hidden_field_keys=hidden_fields,
),
force_scope=_compile_force_scope(section.force_scope, section.force_scope_ids),
),
)
_VISUALIZATION_FIELD_KEYS = {
"surface": "surfaces",
"slice": "slices",
"isosurface": "isosurfaces",
"streamline": "streamlines",
}
_VISUALIZATION_TYPES_BY_RESOURCE = {
"Case": frozenset(_VISUALIZATION_FIELD_KEYS),
"Geometry": frozenset(),
"SurfaceMesh": frozenset({"surface"}),
"VolumeMesh": frozenset({"slice"}),
}
_CLIP_TYPES = {"none": 0, "above": 1, "below": 2, "range": 3}
def _merge_setup_mapping(target: dict[str, Any], source: Mapping[str, Any], name: str) -> None:
for item_id, value in source.items():
if item_id in target and target[item_id] != value:
raise ValueError(f"{name} configures conflicting values for {item_id!r}")
target[item_id] = value
def _compile_visualization_setup( # pylint: disable=too-many-locals,too-many-branches
setups: list[ReportVisualizationSetup] | SceneVisualizationSetting,
resource_type: models.ResourceType,
) -> models.VisualizationViewSetup | None:
if isinstance(setups, SceneVisualizationSetting):
if setups.resource_type != resource_type:
raise ValueError(
f"scene visualization setting for {setups.resource_type} cannot be applied "
f"to a {resource_type} visualization; apply its viewpoint without its setting"
)
return setups._as_report_setup()
if not setups:
return None
allowed_types = _VISUALIZATION_TYPES_BY_RESOURCE[resource_type]
unsupported = sorted({setup.type for setup in setups} - allowed_types)
if unsupported:
if resource_type == "Geometry":
raise ValueError("Geometry visualizations do not expose setup controls in the web UI")
raise ValueError(
f"{resource_type} visualizations support only "
+ ", ".join(sorted(allowed_types))
+ " setup outputs"
)
visible_setting: dict[str, bool] = {}
visible_setting_by_field: dict[str, dict[str, bool]] = {}
field_setting: dict[str, models.FieldCacheData] = {}
time_frame_setting: dict[str, int] = {}
slice_variant_setting: dict[str, Literal["flat", "crinkled"]] = {}
wireframe_setting: dict[str, bool] = {}
custom_color_nodes = None
for setup in setups:
scoped_visibility: dict[str, bool] = {}
for item_id in setup.show or []:
scoped_visibility[item_id] = True
for item_id in setup.hide or []:
scoped_visibility[item_id] = False
if scoped_visibility:
_merge_setup_mapping(visible_setting, scoped_visibility, "show/hide")
field_key = f"{_VISUALIZATION_FIELD_KEYS[setup.type]}::{setup.output}"
visible_setting_by_field[field_key] = scoped_visibility
field_data = models.FieldCacheData(
field_name=setup.field,
log_scale=setup.log_scale,
range=setup.range,
color_scale_min_max=setup.color_range,
theme=setup.theme,
contour_steps=setup.contour_steps,
contour_line_color=setup.contour_line_color,
clip_type=_CLIP_TYPES[setup.clip] if setup.clip is not None else None,
clip_value=setup.clip_value,
clip_value_range=setup.clip_range,
solid_color=setup.solid_color,
show_color_map=setup.show_color_map,
contour_mode=setup.contour_mode,
tube_width=setup.tube_width,
tube_width_unit=setup.tube_width_unit,
ribbon_width=setup.ribbon_width,
ribbon_width_unit=setup.ribbon_width_unit,
streamline_direction=(
setup.streamline_direction.title()
if setup.streamline_direction is not None
else None
),
render_type=setup.render_type.title() if setup.render_type is not None else None,
ribbon_angle_scale=setup.ribbon_angle_scale,
visualizer=setup.visualizer,
selected_unit=setup.unit,
use_local_value=setup.use_local_value,
time_frame=setup.time_frame,
)
if field_data.model_dump(exclude_none=True):
field_setting[setup.output] = field_data
if setup.time_frame is not None:
time_frame_setting[setup.output] = setup.time_frame
if setup.slice_variants is not None:
_merge_setup_mapping(slice_variant_setting, setup.slice_variants, "slice_variants")
if setup.wireframes is not None:
_merge_setup_mapping(wireframe_setting, setup.wireframes, "wireframes")
if setup.custom_colors is not None:
custom_color_nodes = [
models.ColorNode(position=position, color=color)
for position, color in setup.custom_colors
]
return models.VisualizationViewSetup(
visible_setting=visible_setting or None,
field_setting=field_setting or None,
color_map_setting=(
models.ColorMapSetting(custom_color_nodes=custom_color_nodes)
if custom_color_nodes is not None
else None
),
time_frame_setting=time_frame_setting or None,
slice_variant_setting=slice_variant_setting or None,
wireframe_setting=wireframe_setting or None,
visible_setting_by_field=visible_setting_by_field or None,
)
_CAMERA_LENGTH_UNIT_TO_METERS: dict[CameraLengthUnit, float] = {
"m": 1.0,
"cm": 0.01,
"mm": 0.001,
"inch": 0.0254,
"ft": 0.3048,
}
def _meters_per_model_unit(simulation_json: Mapping[str, Any]) -> float:
"""Resolve the Workbench model scale with the same invalid-value fallbacks as the UI."""
asset_cache = simulation_json.get("private_attribute_asset_cache")
if not isinstance(asset_cache, Mapping):
return 1.0
project_length_unit = asset_cache.get("project_length_unit")
if not isinstance(project_length_unit, Mapping):
return 1.0
raw_unit = project_length_unit.get("units")
if isinstance(raw_unit, str) and raw_unit in _CAMERA_LENGTH_UNIT_TO_METERS:
unit = raw_unit
else:
unit = "m"
raw_factor = project_length_unit.get("value")
try:
factor = float(raw_factor) if not isinstance(raw_factor, bool) else 1.0
except (TypeError, ValueError):
factor = 1.0
if not math.isfinite(factor) or factor <= 0:
factor = 1.0
return factor * _CAMERA_LENGTH_UNIT_TO_METERS[unit]
def _compile_physical_camera_for_resource(
camera: ReportCamera | SceneViewpoint,
simulation_json: Mapping[str, Any],
) -> models.ReportCameraParams:
"""Convert physical camera lengths into one target resource's model coordinates."""
if isinstance(camera, SceneViewpoint):
values = camera._as_report_camera().model_dump()
meters_per_input_unit = 1.0
else:
values = camera.model_dump(exclude={"unit"})
if camera.unit is None:
raise ValueError("a physical per-view Report.Camera requires an explicit unit")
meters_per_input_unit = _CAMERA_LENGTH_UNIT_TO_METERS[camera.unit]
scale = meters_per_input_unit / _meters_per_model_unit(simulation_json)
for field_name in ("look_at", "pan_target"):
vector = values.get(field_name)
if vector is not None:
values[field_name] = tuple(component * scale for component in vector)
if values.get("dimension") is not None:
values["dimension"] *= scale
return models.ReportCameraParams(**values)
def _compile_camera(
camera: ReportCamera | SceneViewpoint | None, *, convert_lengths_to_meters: bool
) -> models.ReportCameraParams | None:
if camera is None:
return None
if isinstance(camera, SceneViewpoint):
return camera._as_report_camera()
values = camera.model_dump(exclude={"unit"})
if convert_lengths_to_meters:
scale = _CAMERA_LENGTH_UNIT_TO_METERS[camera.unit or "m"]
for field_name in ("look_at", "pan_target"):
vector = values.get(field_name)
if vector is not None:
values[field_name] = tuple(component * scale for component in vector)
if values.get("dimension") is not None:
values["dimension"] *= scale
return models.ReportCameraParams(**values)
def _compile_visualization_views(
views: dict[str, ReportVisualizationView],
selected_resources: list[models.ReportResource],
resource_type: models.ResourceType,
global_setup: models.VisualizationViewSetup | None,
resource_assets_by_id: Mapping[str, ReportResourceInput],
simulation_json_getter: Callable[[ReportResourceInput], Mapping[str, Any]] | None,
simulation_json_cache: dict[str, Mapping[str, Any]],
) -> dict[str, models.VisualizationViewState]:
selected_ids = {resource.id for resource in selected_resources}
compiled: dict[str, models.VisualizationViewState] = {}
for resource_id, view in views.items():
if resource_id not in selected_ids:
raise ValueError(
f"visualization view resource {resource_id} is not selected by the section"
)
if view.setup:
view_setup = _compile_visualization_setup(view.setup, resource_type)
else:
view_setup = global_setup.model_copy(deep=True) if global_setup is not None else None
camera_requires_unit_conversion = isinstance(view.camera, SceneViewpoint) or (
isinstance(view.camera, ReportCamera) and view.camera.unit is not None
)
if camera_requires_unit_conversion:
if simulation_json_getter is None:
raise ValueError(
"a per-view Scene viewpoint or Report.Camera with unit requires a "
"SimulationJSON getter for unit conversion"
)
if resource_id not in simulation_json_cache:
simulation_json = simulation_json_getter(resource_assets_by_id[resource_id])
if not isinstance(simulation_json, Mapping):
raise TypeError(
f"SimulationJSON for resource {resource_id!r} must be a mapping"
)
simulation_json_cache[resource_id] = simulation_json
view_camera = _compile_physical_camera_for_resource(
view.camera,
simulation_json_cache[resource_id],
)
else:
view_camera = _compile_camera(view.camera, convert_lengths_to_meters=False)
compiled[resource_id] = models.VisualizationViewState(
linked=False,
camera_params=view_camera,
setup=view_setup,
)
return compiled
def _compile_visualization(
section: ReportVisualization,
report_resources: list[models.ReportResource],
resource_assets_by_id: Mapping[str, ReportResourceInput],
simulation_json_getter: Callable[[ReportResourceInput], Mapping[str, Any]] | None,
simulation_json_cache: dict[str, Mapping[str, Any]],
) -> models.VisualizationSection:
resources = _selected_resources(
section.resources,
report_resources,
section_name="visualization",
)
if section.resources is None:
resources = resources[: models.MAX_VISUALIZATION_RESOURCES]
resource_types = {resource.type for resource in resources}
if len(resource_types) != 1:
raise ValueError("a visualization section can contain only one resource type")
if len(resources) > models.MAX_VISUALIZATION_RESOURCES:
raise ValueError(
"a visualization section supports at most "
f"{models.MAX_VISUALIZATION_RESOURCES} resources"
)
resource_type = resources[0].type
if resource_type != "Case" and section.resolution == "low":
raise ValueError("low resolution is available only for Case visualizations")
lod_level = 0 if resource_type != "Case" or section.resolution == "high" else 1
setup = _compile_visualization_setup(section.setup, resource_type)
camera = _compile_camera(section.camera, convert_lengths_to_meters=True)
views = _compile_visualization_views(
section.views,
resources,
resource_type,
setup,
resource_assets_by_id,
simulation_json_getter,
simulation_json_cache,
)
return models.VisualizationSection(
title=section.title,
config=models.VisualizationSectionConfig(
resources=models.VisualizationSectionResourcesConfig(
type=resource_type,
selected_ids=[resource.id for resource in resources],
),
display=models.VisualizationSectionDisplayConfig(
layout=section.layout,
lod_level=lod_level,
),
sync=models.VisualizationSectionSyncConfig(
camera=models.VisualizationCameraSyncConfig(global_params=camera),
settings=models.VisualizationSettingsSyncConfig(global_setup=setup),
),
views_by_resource_id=views,
),
)
def _chart_background(
x_axis_id: str, background_view: ChartBackgroundView | None
) -> models.Chart2DBackgroundConfig | None:
if background_view is None:
return None
viewpoint_by_axis = {
("axis:x:spatial:X", "left"): "chart2d:left",
("axis:x:spatial:X", "top"): "chart2d:x-top",
("axis:x:spatial:Y", "back"): "chart2d:back",
("axis:x:spatial:Y", "top"): "chart2d:y-top",
}
return models.Chart2DBackgroundConfig(
viewpoint_id=viewpoint_by_axis[(x_axis_id, background_view)]
)
def _compile_chart(
section: ReportChart, report_resources: list[models.ReportResource]
) -> models.Chart2DSection:
resources = _selected_resources(section.resources, report_resources, section_name="chart")
if any(resource.type != "Case" for resource in resources):
raise ValueError("chart sections can contain only Case resources")
if section.resources is None:
resources = resources[: models.MAX_CHART2D_CASES]
if len(resources) > models.MAX_CHART2D_CASES:
raise ValueError(f"a chart section supports at most {models.MAX_CHART2D_CASES} cases")
reference = next((resource for resource in report_resources if resource.is_reference), None)
if reference is not None and all(resource.id != reference.id for resource in resources):
if len(resources) == models.MAX_CHART2D_CASES:
resources = resources[:-1]
resources.append(reference)
x_axis_id = _x_axis_id(section.x)
y_axis_ids = [_y_axis_id(variable, "y") for variable in section.y]
y_ranges = {
_y_axis_id(variable, "y_ranges"): value_range
for variable, value_range in (section.y_ranges or {}).items()
}
y_axes = [
models.Chart2DAxisConfig(
range=(
models.Chart2DManualRange(values=y_ranges[y_axis_id])
if y_axis_id in y_ranges
else models.Chart2DGlobalAutoRange()
),
scale="log" if section.log_scale else "linear",
)
for y_axis_id in y_axis_ids
]
x_range = (
models.Chart2DManualRange(values=section.x_range)
if section.x_range is not None
else models.Chart2DGlobalAutoRange()
)
style_mapping = (
"colorByCaseLineByVariable" if section.style == "case" else "colorByVariableLineByCase"
)
return models.Chart2DSection(
title=section.title,
config=models.Chart2DSectionConfig(
resources=models.Chart2DSectionResourcesConfig(
selected_ids=[resource.id for resource in resources]
),
shared=models.Chart2DSharedConfig(
data=models.Chart2DSharedDataConfig(
x_axis_id=x_axis_id,
force_scope=(
_compile_force_scope(section.force_scope, section.force_scope_ids)
if section.force_scope is not None
else None
),
series_selection=(
models.Chart2DSeriesSelectionConfig(included_ids=section.series)
if section.series is not None
else None
),
series_display=section.series_display,
),
comparison=models.Chart2DComparisonConfig(mode=section.comparison),
axes=models.Chart2DAxesConfig(x=models.Chart2DAxisConfig(range=x_range)),
background=_chart_background(x_axis_id, section.background_view),
display=models.Chart2DDisplayConfig(style_mapping=style_mapping),
),
charts=[
models.Chart2DChartConfig(
data=models.Chart2DChartDataConfig(y_axis_ids=y_axis_ids),
axes=models.Chart2DAxesConfig(y=y_axes),
)
],
),
)
def build_report_config(
*,
resources: Iterable[ReportResourceInput],
sections: Iterable[ReportSectionSpec] | None = None,
reference: CaseResourceSelector | None = None,
aliases: Mapping[str, str] | None = None,
simulation_json_getter: Callable[[ReportResourceInput], Mapping[str, Any]] | None = None,
) -> models.ReportConfig:
"""Compile the public Report API into the web application's persisted config."""
if isinstance(resources, (str, bytes)):
raise TypeError("resources must be an iterable of Flow360 cloud resources")
resource_assets = list(resources)
report_resources = _build_resources(resource_assets, aliases, reference)
resource_assets_by_id = {
str(_resource_value(resource, "id")).strip(): resource for resource in resource_assets
}
simulation_json_cache: dict[str, Mapping[str, Any]] = {}
compiled_sections = []
seen_summaries: set[SummaryResourceType] = set()
for section in sections or []:
if isinstance(section, ReportSummary):
if section.resource_type in seen_summaries:
raise ValueError(f"only one {section.resource_type} summary can be configured")
seen_summaries.add(section.resource_type)
compiled_sections.append(_compile_summary(section))
elif isinstance(section, ReportVisualization):
compiled_sections.append(
_compile_visualization(
section,
report_resources,
resource_assets_by_id,
simulation_json_getter,
simulation_json_cache,
)
)
elif isinstance(section, ReportChart):
compiled_sections.append(_compile_chart(section, report_resources))
else:
raise TypeError(
"sections must contain only ReportSummary, ReportVisualization, or ReportChart"
)
return models.ReportConfig(resources=report_resources, sections=compiled_sections)
