"""Cloud resource for interactive Flexcompute Flow reports."""
from __future__ import annotations
from collections.abc import Iterable, Mapping
from datetime import datetime
from typing import Any
import pydantic as pd
from pydantic.alias_generators import to_camel
from ._flow360_adapter import (
DEFAULT_FLOW360_ADAPTER,
DEFAULT_ROOT_FOLDER_ID,
Flow360Adapter,
)
from .apply_to_new import build_apply_to_new_report_config
from .models import ReportAssociatedResource, ReportConfig, ReportInfo, ResourceType
from .resource_types import CaseResourceSelector, ReportResourceInput
from .scene import Scene
from .spec import (
ReportCamera,
ReportChart,
ReportChartVariable,
ReportSectionSpec,
ReportSummary,
ReportVisualization,
ReportVisualizationSetup,
ReportVisualizationView,
build_report_config,
)
class _CreateReportRequest(pd.BaseModel):
model_config = pd.ConfigDict(extra="forbid", populate_by_name=True, alias_generator=to_camel)
name: str
description: str
tags: list[str]
parent_folder_id: str
config_json: str
def to_dict(self) -> dict:
"""Return the request payload with API aliases."""
return self.model_dump(mode="json", by_alias=True, exclude_none=True)
class _ReportResourceRef(pd.BaseModel):
model_config = pd.ConfigDict(extra="forbid")
id: str
type: ResourceType
class _UpdateReportRequest(pd.BaseModel):
model_config = pd.ConfigDict(extra="forbid", populate_by_name=True, alias_generator=to_camel)
name: str | None = None
description: str | None = None
tags: list[str] | None = None
parent_folder_id: str | None = None
associated_resources: list[_ReportResourceRef] | None = None
config_json: str | None = None
def to_dict(self) -> dict:
"""Return only explicitly updated fields with API aliases."""
return self.model_dump(mode="json", by_alias=True, exclude_none=True)
def _clean_tags(tags: Iterable[str] | None) -> list[str]:
"""Trim and deduplicate report tags while preserving order."""
if isinstance(tags, str):
raise TypeError("tags must be an iterable of strings, not a string")
cleaned: list[str] = []
seen: set[str] = set()
for tag in tags or []:
if not isinstance(tag, str):
raise TypeError("tags must contain only strings")
normalized = tag.strip()
if normalized and normalized not in seen:
cleaned.append(normalized)
seen.add(normalized)
return cleaned
[docs]
class Report:
"""An interactive Report resource created on the Flow360 platform."""
Camera = ReportCamera
Chart = ReportChart
ChartVariable = ReportChartVariable
Summary = ReportSummary
Visualization = ReportVisualization
VisualizationSetup = ReportVisualizationSetup
VisualizationView = ReportVisualizationView
Scene = Scene
_cloud_resource_type_name = "Report"
_adapter = DEFAULT_FLOW360_ADAPTER
def __init__(
self,
*,
info: ReportInfo,
environment: Any,
config: ReportConfig,
adapter: Flow360Adapter,
):
self._info = info
self._environment = environment
self._config = config
self._flow360_adapter = adapter
@classmethod
def _create_from_config( # pylint: disable=too-many-arguments
cls,
*,
name: str,
config: ReportConfig,
description: str,
tags: Iterable[str] | None,
parent_folder_id: str,
environment: Any,
adapter: Flow360Adapter,
) -> "Report":
"""Create a Report from an SDK-managed persisted configuration."""
normalized_name = name.strip()
normalized_parent_folder_id = parent_folder_id.strip()
if not normalized_name:
raise ValueError("name must not be empty")
if not normalized_parent_folder_id:
raise ValueError("parent_folder_id must not be empty")
request = _CreateReportRequest(
name=normalized_name,
description=description.strip(),
tags=_clean_tags(tags),
parent_folder_id=normalized_parent_folder_id,
config_json=config.to_json(),
)
response = adapter.create_report(request.to_dict(), environment=environment)
info = ReportInfo.model_validate(response)
return cls(info=info, environment=environment, config=config, adapter=adapter)
[docs]
@classmethod
def from_cloud(cls, report_id: str) -> "Report":
"""Load an interactive Report from the active Flow360 environment."""
if not isinstance(report_id, str):
raise TypeError("report_id must be a string")
normalized_report_id = report_id.strip()
if not normalized_report_id:
raise ValueError("report_id must not be empty")
adapter = cls._adapter
environment = adapter.current_environment()
response = adapter.get_report(normalized_report_id, environment=environment)
return cls._from_cloud_response(
response=response,
environment=environment,
adapter=adapter,
report_id=normalized_report_id,
)
@classmethod
def _from_cloud_response(
cls,
*,
response: Mapping[str, Any],
environment: Any,
adapter: Flow360Adapter,
report_id: str,
) -> "Report":
"""Build a Report from a cloud response that includes persisted config."""
raw_config = response.get("configJson")
if isinstance(raw_config, str):
config = ReportConfig.model_validate_json(raw_config)
elif isinstance(raw_config, Mapping):
config = ReportConfig.model_validate(raw_config)
response = {**response, "configJson": config.to_json()}
else:
raise ValueError(f"Report {report_id} does not contain a valid configJson")
info = ReportInfo.model_validate(response)
return cls(info=info, environment=environment, config=config, adapter=adapter)
[docs]
@classmethod
def create( # pylint: disable=too-many-arguments,too-many-locals
cls,
*,
name: str,
resources: Iterable[ReportResourceInput],
sections: Iterable[ReportSectionSpec] | None = None,
reference: CaseResourceSelector | None = None,
aliases: Mapping[str, str] | None = None,
description: str = "",
tags: Iterable[str] | None = None,
parent_folder_id: str = DEFAULT_ROOT_FOLDER_ID,
) -> "Report":
"""Create an interactive report and return its cloud resource.
AI-generated workflows should load ``get_agent_guide()`` before constructing
sections. For visualizations, discover every resource's capabilities and run
``check_visualization_setup(...)`` before calling this method.
"""
adapter = cls._adapter
environment = adapter.current_environment()
config = build_report_config(
resources=resources,
sections=sections,
reference=reference,
aliases=aliases,
simulation_json_getter=lambda resource: adapter.get_simulation_json(
resource,
environment=environment,
),
)
return cls._create_from_config(
name=name,
config=config,
description=description,
tags=tags,
parent_folder_id=parent_folder_id,
environment=environment,
adapter=adapter,
)
[docs]
def apply_to_new( # pylint: disable=too-many-arguments
self,
*,
name: str,
resources: Iterable[ReportResourceInput],
reference: CaseResourceSelector | None = None,
aliases: Mapping[str, str] | None = None,
description: str = "",
tags: Iterable[str] | None = None,
parent_folder_id: str | None = None,
) -> "Report":
"""Create a new Report by applying this Report to new cloud resources."""
config = build_apply_to_new_report_config(
source_config=self._config,
resources=resources,
reference=reference,
aliases=aliases,
)
target_folder_id = (
parent_folder_id
if parent_folder_id is not None
else self.parent_folder_id or DEFAULT_ROOT_FOLDER_ID
)
return type(self)._create_from_config(
name=name,
config=config,
description=description,
tags=tags,
parent_folder_id=target_folder_id,
environment=self._environment,
adapter=self._flow360_adapter,
)
[docs]
def refresh(self) -> "Report":
"""Reload this Report's metadata and persisted configuration."""
response = self._flow360_adapter.get_report(self.id, environment=self._environment)
refreshed = type(self)._from_cloud_response(
response=response,
environment=self._environment,
adapter=self._flow360_adapter,
report_id=self.id,
)
self._info = refreshed._info
self._config = refreshed._config
return self
[docs]
def update( # pylint: disable=too-many-arguments
self,
*,
resources: Iterable[ReportResourceInput] | None = None,
sections: Iterable[ReportSectionSpec] | None = None,
reference: CaseResourceSelector | None = None,
aliases: Mapping[str, str] | None = None,
name: str | None = None,
description: str | None = None,
tags: Iterable[str] | None = None,
parent_folder_id: str | None = None,
) -> "Report":
"""Update this Report in place while preserving its cloud ID and URL."""
if (resources is None) != (sections is None):
raise ValueError(
"resources and sections must be provided together when updating config"
)
config_changed = resources is not None
if not config_changed:
if reference is not None or aliases is not None:
raise ValueError("reference and aliases require resources and sections")
config = self._config.model_copy(deep=True)
else:
config = build_report_config(
resources=resources,
sections=sections,
reference=reference,
aliases=aliases,
simulation_json_getter=lambda resource: self._flow360_adapter.get_simulation_json(
resource,
environment=self._environment,
),
)
if name is not None and (not isinstance(name, str) or not name.strip()):
raise ValueError("name must not be empty")
if description is not None and not isinstance(description, str):
raise TypeError("description must be a string")
if parent_folder_id is not None and (
not isinstance(parent_folder_id, str) or not parent_folder_id.strip()
):
raise ValueError("parent_folder_id must not be empty")
request = _UpdateReportRequest(
name=name.strip() if name is not None else None,
description=description.strip() if description is not None else None,
tags=_clean_tags(tags) if tags is not None else None,
parent_folder_id=parent_folder_id.strip() if parent_folder_id is not None else None,
associated_resources=(
[
_ReportResourceRef(id=resource.id, type=resource.type)
for resource in config.resources
]
if config_changed
else None
),
config_json=config.to_json() if config_changed else None,
)
payload = request.to_dict()
if not payload:
return self
response = self._flow360_adapter.update_report(
self.id,
payload,
environment=self._environment,
)
self._info = ReportInfo.model_validate(response)
self._config = config
return self
@property
def id(self) -> str:
"""Cloud report ID."""
return self._info.id
@property
def project_id(self) -> str | None:
"""Project ID created for this report."""
return self._info.project_id
@property
def name(self) -> str:
"""Report name."""
return self._info.name
@property
def description(self) -> str | None:
"""Report description."""
return self._info.description
@property
def tags(self) -> list[str]:
"""Report tags."""
return list(self._info.tags)
@property
def parent_folder_id(self) -> str | None:
"""Parent folder ID."""
return self._info.parent_folder_id
@property
def workspace_id(self) -> str | None:
"""Workspace containing the report."""
return self._info.workspace_id
@property
def status(self) -> str:
"""Current report generation status."""
return self._info.status
@property
def viewed(self) -> bool | None:
"""Whether the report has been viewed."""
return self._info.viewed
@property
def is_deleted(self) -> bool | None:
"""Whether the report has been deleted."""
return self._info.is_deleted
@property
def associated_resources(self) -> list[ReportAssociatedResource]:
"""Resources associated with this report."""
return list(self._info.associated_resources)
@property
def config(self) -> ReportConfig:
"""A defensive copy of the configuration persisted for this Report."""
return self._config.model_copy(deep=True)
@property
def config_json(self) -> str:
"""The persisted Report configuration serialized as JSON."""
return self._config.to_json()
@property
def created_at(self) -> datetime | None:
"""Report creation time."""
return self._info.created_at
@property
def updated_at(self) -> datetime | None:
"""Report last update time."""
return self._info.updated_at
@property
def web_url(self) -> str:
"""Browser URL for this report in the environment where it was created."""
return self._flow360_adapter.web_url(self.id, environment=self._environment)
def __repr__(self) -> str:
return f"Report(id={self.id!r}, name={self.name!r}, status={self.status!r})"
