"""Topology and element metadata for a :class:`DataSource`.
Two concerns live here:
- :class:`Topology` -- the connectivity + vertex coordinates of the
underlying mesh (or the bare points of a points data source). The
``cell_type`` discriminator is in place from day one so volume
export can be added later additively.
- :class:`ElementMeta` -- per-element scalar / vector attributes:
position (centroid), area, volume, normal. Optional free-form metadata
is allowed for "station name"-style annotations.
A :class:`GroupsDataSource` does *not* own its topology; it carries a
reference to the parent surface's topology plus a per-element group
index. That keeps groups data sources cheap and prevents the parent
mesh from drifting from the grouped view.
"""
from __future__ import annotations
__all__ = [
"CellType",
"Topology",
"ElementMeta",
]
from typing import Annotated, Any, Literal
import numpy as np
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
CellType = Literal["triangle", "point", "cell"]
"""Supported cell-type discriminators.
- ``triangle``: 3-node 2D triangular faces (the existing surface
format). ``connectivity`` has shape ``(n_elements, 3)``.
- ``point``: bare points / probes. ``connectivity`` is empty
(shape ``(0, 0)``).
- ``cell``: 3D volumetric cells. Reserved; not implemented in Phase 1
but held by the discriminator so Phase-N volume export does not
need a schema change.
"""
def _arr(value: Any, dtype: np.dtype) -> np.ndarray:
"""Coerce inputs to ``numpy.ndarray`` with the requested dtype.
Pydantic does not natively validate ``numpy.ndarray`` fields; this
helper plus the ``arbitrary_types_allowed=True`` config gives a
consistent point of entry.
"""
arr = np.asarray(value, dtype=dtype)
return arr
[docs]
class Topology(BaseModel):
"""Mesh connectivity + vertex coordinates for a data source.
Frozen and immutable. Geometric ops (rigid-body transformation,
rescale) produce a new :class:`Topology` rather than mutating in
place.
Attributes:
cell_type: One of :data:`CellType`. Locks the connectivity
schema.
connectivity: For ``triangle`` -> ``(n_elements, 3)`` int32
indices into ``vertices``. For ``point`` -> ``(0, 0)`` (no
connectivity). For ``cell`` (reserved) -> implementation
specific.
vertices: ``(n_vertices, 3)`` float64 vertex coordinates.
The shapes mirror the existing on-disk layout in
``cfdmod/io/xdmf.py`` exactly: ``/Triangles`` -> int32 (N, 3),
``/Geometry`` -> float64 (V, 3). No format change.
"""
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
cell_type: CellType
connectivity: np.ndarray
vertices: np.ndarray
@field_validator("connectivity", mode="before")
@classmethod
def _coerce_connectivity(cls, v: Any) -> np.ndarray:
return _arr(v, np.dtype("int32"))
@field_validator("vertices", mode="before")
@classmethod
def _coerce_vertices(cls, v: Any) -> np.ndarray:
return _arr(v, np.dtype("float64"))
@model_validator(mode="after")
def _check_shapes(self) -> "Topology":
if self.vertices.ndim != 2 or self.vertices.shape[1] != 3:
raise ValueError(
f"vertices must have shape (n_vertices, 3); got {self.vertices.shape}"
)
if self.cell_type == "triangle":
if self.connectivity.ndim != 2 or self.connectivity.shape[1] != 3:
raise ValueError(
"triangle connectivity must have shape (n_elements, 3); got "
f"{self.connectivity.shape}"
)
if self.vertices.shape[0] > 0 and self.connectivity.size > 0:
max_idx = int(self.connectivity.max())
if max_idx >= self.vertices.shape[0]:
raise ValueError(
f"connectivity references vertex {max_idx} but only "
f"{self.vertices.shape[0]} vertices were provided"
)
elif self.cell_type == "point":
if self.connectivity.size != 0:
raise ValueError(
"point topology must have empty connectivity; got "
f"shape {self.connectivity.shape}"
)
return self
@property
def n_elements(self) -> int:
"""Number of elements (triangles, points, or cells)."""
if self.cell_type == "point":
return self.vertices.shape[0]
return self.connectivity.shape[0]
@property
def n_vertices(self) -> int:
return self.vertices.shape[0]
[docs]
@classmethod
def points(cls, vertices: Any) -> "Topology":
"""Build a points topology from a ``(n_points, 3)`` array."""
verts = _arr(vertices, np.dtype("float64"))
return cls(
cell_type="point",
connectivity=np.empty((0, 0), dtype=np.int32),
vertices=verts,
)
[docs]
@classmethod
def triangles(cls, connectivity: Any, vertices: Any) -> "Topology":
"""Build a triangle topology from connectivity + vertex arrays."""
return cls(
cell_type="triangle",
connectivity=connectivity,
vertices=vertices,
)
class ElementMeta(BaseModel):
"""Per-element scalar / vector attributes.
Every column is optional; a points data source typically only sets
``position``, a surface data source typically sets ``position`` +
``area`` + ``normal``, a volume data source adds ``volume``.
All array columns share the leading axis ``n_elements``. Cross-
column shape checks happen at :class:`DataSource` construction
time, not here, because the canonical length lives there.
Free-form per-element metadata (e.g. station name) belongs in
``annotations``: a dict whose values are arrays of length
``n_elements`` or constants.
"""
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
position: np.ndarray | None = None
area: np.ndarray | None = None
volume: np.ndarray | None = None
normal: np.ndarray | None = None
annotations: Annotated[
dict[str, Any],
Field(default_factory=dict, description="Free-form per-element metadata."),
]
@field_validator("position", "normal", mode="before")
@classmethod
def _coerce_vec3(cls, v: Any) -> np.ndarray | None:
if v is None:
return None
return _arr(v, np.dtype("float64"))
@field_validator("area", "volume", mode="before")
@classmethod
def _coerce_scalar(cls, v: Any) -> np.ndarray | None:
if v is None:
return None
return _arr(v, np.dtype("float64"))
@model_validator(mode="after")
def _check_shapes(self) -> "ElementMeta":
for name in ("position", "normal"):
arr = getattr(self, name)
if arr is not None and (arr.ndim != 2 or arr.shape[1] != 3):
raise ValueError(f"{name} must have shape (n_elements, 3); got {arr.shape}")
for name in ("area", "volume"):
arr = getattr(self, name)
if arr is not None and arr.ndim != 1:
raise ValueError(f"{name} must have shape (n_elements,); got {arr.shape}")
return self