"""Public validation entry points and internal dispatch helpers."""
from __future__ import annotations
from pathlib import Path
from typing import assert_never, cast
from buffalo_core.diagnostics import Diagnostic, DiagnosticLocation
from buffalo_wings.airfoil.internal.schema import (
AirfoilCamberCurveSpec,
AirfoilDefinitionSpec,
AirfoilDocumentSpec,
CurveDefinitionSpec,
FileAirfoilSpec,
PointsAirfoilSpec,
PolylineCurveSpec,
SplineAirfoilSpec,
SplineCurveSpec,
)
from buffalo_wings.airfoil.internal.schema.analytic import (
BiconvexAirfoilSpec,
BiconvexParabolaAirfoilSpec,
CircularArcAirfoilSpec,
CstAirfoilSpec,
EllipseAirfoilSpec,
FlatPlateAirfoilSpec,
ParsecAirfoilSpec,
PolygonAirfoilSpec,
)
from buffalo_wings.airfoil.internal.schema.constants import POINT2D_TUPLE_LENGTH
from buffalo_wings.airfoil.internal.schema.naca import (
Naca4AirfoilSpec,
Naca4ModifiedAirfoilSpec,
Naca5AirfoilSpec,
Naca5ModifiedAirfoilSpec,
Naca6AAirfoilSpec,
Naca6AirfoilSpec,
Naca16AirfoilSpec,
)
from .analytic import (
validate_biconvex_parabola_spec,
validate_biconvex_spec,
validate_circular_arc_spec,
validate_cst_spec,
validate_ellipse_spec,
validate_flat_plate_spec,
validate_joukowski_params,
validate_parsec_spec,
validate_polygon_params,
)
from .common import (
AirfoilDocumentValidationResult,
AirfoilValidationResult,
StructuredPayloadFormat,
build_support_error,
build_validation_error,
document_result_from_diagnostics,
result_from_diagnostics,
)
from .file_io import read_file_source_text, validate_file_spec
from .file_structured import (
detect_file_source_format,
validate_structured_points_source,
)
from .file_text import validate_lednicer_source, validate_selig_source
from .naca4 import validate_naca4_modified_spec, validate_naca4_spec
from .naca5 import validate_naca5_modified_spec, validate_naca5_spec
from .naca6 import validate_naca6_spec
from .naca6a import validate_naca6a_spec
from .naca16 import validate_naca16_spec
from .points import validate_points_spec
from .spline import validate_spline_spec
_MIN_CURVE_APPROXIMATION_POINTS = 2
_MIN_POLYLINE_CURVE_POINTS = 2
_MIN_SPLINE_CURVE_CONTROL_POINTS = 2
def _validate_schema_family( # noqa: PLR0911
spec: AirfoilDefinitionSpec,
) -> tuple[Diagnostic, ...]:
"""
Dispatch to one family-specific validator.
Raises
------
AssertionError
If a new schema type reaches this compatibility facade without a
matching validator branch.
"""
if isinstance(spec, Naca4AirfoilSpec):
return validate_naca4_spec(spec)
if isinstance(spec, Naca4ModifiedAirfoilSpec):
return validate_naca4_modified_spec(spec)
if isinstance(spec, Naca5AirfoilSpec):
return validate_naca5_spec(spec)
if isinstance(spec, Naca5ModifiedAirfoilSpec):
return validate_naca5_modified_spec(spec)
if isinstance(spec, Naca16AirfoilSpec):
return validate_naca16_spec(spec)
if isinstance(spec, Naca6AirfoilSpec):
return validate_naca6_spec(spec)
if isinstance(spec, Naca6AAirfoilSpec):
return validate_naca6a_spec(spec)
if isinstance(spec, FileAirfoilSpec):
return validate_file_spec(spec)
if isinstance(spec, PointsAirfoilSpec):
return validate_points_spec(spec)
if isinstance(spec, SplineAirfoilSpec):
return validate_spline_spec(spec)
if isinstance(spec, CstAirfoilSpec):
return validate_cst_spec(spec)
if isinstance(spec, ParsecAirfoilSpec):
return validate_parsec_spec(spec)
if isinstance(spec, BiconvexAirfoilSpec):
return validate_biconvex_spec(spec)
if isinstance(spec, BiconvexParabolaAirfoilSpec):
return validate_biconvex_parabola_spec(spec)
if isinstance(spec, EllipseAirfoilSpec):
return validate_ellipse_spec(spec)
if isinstance(spec, CircularArcAirfoilSpec):
return validate_circular_arc_spec(spec)
if isinstance(spec, FlatPlateAirfoilSpec):
return validate_flat_plate_spec(spec)
if isinstance(spec, PolygonAirfoilSpec):
return validate_polygon_params(spec.params)
return validate_joukowski_params(spec.params)
def _runtime_supports_spec(spec: AirfoilDefinitionSpec) -> bool:
"""
Return whether the current runtime can construct ``spec``.
Returns
-------
bool
``True`` when the current runtime implements the requested family.
"""
return isinstance(
spec,
Naca4AirfoilSpec
| Naca4ModifiedAirfoilSpec
| Naca5AirfoilSpec
| Naca5ModifiedAirfoilSpec
| EllipseAirfoilSpec
| SplineAirfoilSpec
| CstAirfoilSpec,
)
[docs]
def validate_spec(spec: AirfoilDefinitionSpec) -> AirfoilValidationResult:
"""Validate one schema-backed airfoil definition.
Parameters
----------
spec
Airfoil schema object to validate.
Returns
-------
AirfoilValidationResult
Validation result containing the original schema copy when valid
and a structured diagnostics report describing schema errors or
unsupported runtime families.
Notes
-----
This workflow reports malformed schema content and runtime-support
gaps through structured diagnostics instead of raising for ordinary
validation failures.
Use ``AirfoilFactory.from_spec(...)`` only after checking that the
returned result is valid and supported by the current runtime.
Examples
--------
>>> import buffalo_wings.airfoil as bwa
>>> spec = bwa.Naca4AirfoilSpec(
... params=bwa.Naca4AirfoilParamsSpec(
... m=0.02,
... p=0.4,
... max_thickness=0.12,
... )
... )
>>> result = bwa.validate_spec(spec)
>>> result.is_valid
True
"""
diagnostics = tuple(_validate_schema_family(spec))
if not _runtime_supports_spec(spec):
diagnostics = (*diagnostics, build_support_error(spec))
return result_from_diagnostics(spec, diagnostics)
[docs]
def validate_document(
document: AirfoilDocumentSpec,
) -> AirfoilDocumentValidationResult:
"""
Validate one document containing named airfoils and named curves.
Parameters
----------
document
Top-level airfoil document schema object to validate.
Returns
-------
AirfoilDocumentValidationResult
Validation result containing a structured diagnostic report for all
document-level airfoil and curve checks.
"""
diagnostics: list[Diagnostic] = []
for airfoil_name, airfoil_spec in document.airfoils.items():
diagnostics.extend(
_documentize_airfoil_diagnostics(
airfoil_name=airfoil_name,
diagnostics=validate_spec(airfoil_spec).diagnostics.entries,
)
)
for curve_name, curve_spec in document.curves.items():
diagnostics.extend(
_validate_document_curve(
curve_name=curve_name,
curve_spec=curve_spec,
document=document,
)
)
return document_result_from_diagnostics(document, tuple(diagnostics))
[docs]
def validate_file_airfoil_source(
spec: FileAirfoilSpec,
) -> AirfoilValidationResult:
"""Validate a file-backed airfoil source and its referenced contents.
Parameters
----------
spec
File-backed airfoil schema object describing the source path and
expected or auto-detected file format.
Returns
-------
AirfoilValidationResult
Validation result containing structured diagnostics for file-spec
errors, read failures, detected format metadata, and content
validation problems localized to file-source payload paths.
Notes
-----
This workflow first validates the outer ``FileAirfoilSpec`` and then
validates the referenced file contents when the file can be read.
Diagnostics may include informational entries, such as auto-detected
file format, in addition to error diagnostics.
Examples
--------
>>> from pathlib import Path
>>> import tempfile
>>> import buffalo_wings.airfoil as bwa
>>> with tempfile.TemporaryDirectory() as tmp_dir:
... file_path = Path(tmp_dir) / "airfoil.json"
... _ = file_path.write_text(
... (
... '{"format": "upper_lower", '
... '"upper": [[0.0, 0.0], [1.0, 0.0]], '
... '"lower": [[0.0, 0.0], [1.0, 0.0]]}'
... ),
... encoding="utf-8",
... )
... result = bwa.validate_file_airfoil_source(
... bwa.FileAirfoilSpec(path=str(file_path), format="auto")
... )
>>> result.diagnostics.has_infos
True
"""
diagnostics = list(validate_file_spec(spec))
if diagnostics:
return result_from_diagnostics(spec, tuple(diagnostics))
source_text, read_diagnostics = read_file_source_text(Path(spec.path))
diagnostics.extend(read_diagnostics)
if source_text is None or diagnostics:
return result_from_diagnostics(spec, tuple(diagnostics))
configured_format = spec.format
if configured_format == "auto":
source_format, detection_diagnostics = detect_file_source_format(
path=Path(spec.path),
text=source_text,
)
diagnostics.extend(detection_diagnostics)
else:
source_format = configured_format
if source_format == "selig":
diagnostics.extend(validate_selig_source(source_text))
elif source_format == "lednicer":
diagnostics.extend(validate_lednicer_source(source_text))
else:
if source_format == "surface_curve":
requested_format: StructuredPayloadFormat = "surface_curve"
elif source_format == "upper_lower":
requested_format = "upper_lower"
else:
assert_never(source_format) # pragma: no cover
diagnostics.extend(
validate_structured_points_source(
source_text,
requested_format=requested_format,
)
)
return result_from_diagnostics(spec, tuple(diagnostics))
def _documentize_airfoil_diagnostics(
*,
airfoil_name: str,
diagnostics: tuple[Diagnostic, ...],
) -> tuple[Diagnostic, ...]:
"""Relocate one airfoil-spec diagnostic tuple under document paths."""
relocated: list[Diagnostic] = []
for diagnostic in diagnostics:
location = diagnostic.location
field_suffix = (
""
if location is None or location.field_path is None
else (f".{location.field_path}")
)
relocated.append(
Diagnostic(
severity=diagnostic.severity,
code=diagnostic.code,
message=diagnostic.message,
location=DiagnosticLocation(
object_path="airfoil.document",
field_path=f"airfoils.{airfoil_name}{field_suffix}",
geometry_region=(
None if location is None else location.geometry_region
),
),
)
)
return tuple(relocated)
def _validate_document_curve(
*,
curve_name: str,
curve_spec: CurveDefinitionSpec,
document: AirfoilDocumentSpec,
) -> tuple[Diagnostic, ...]:
"""Validate one named document curve entry."""
if isinstance(curve_spec, AirfoilCamberCurveSpec):
return _validate_document_airfoil_camber_curve(
curve_name=curve_name,
curve_spec=curve_spec,
document=document,
)
if isinstance(curve_spec, PolylineCurveSpec):
return _validate_document_polyline_curve(
curve_name=curve_name,
curve_spec=curve_spec,
)
return _validate_document_spline_curve(
curve_name=curve_name,
curve_spec=curve_spec,
)
def _validate_document_airfoil_camber_curve(
*,
curve_name: str,
curve_spec: AirfoilCamberCurveSpec,
document: AirfoilDocumentSpec,
) -> tuple[Diagnostic, ...]:
"""Validate one named airfoil-derived camber curve entry."""
diagnostics: list[Diagnostic] = []
if curve_spec.airfoil == "":
diagnostics.append(
build_validation_error(
code="airfoil.document.curve.airfoil_required",
message="Document curve airfoil reference must be non-empty.",
field_path=f"curves.{curve_name}.airfoil",
object_path="airfoil.document",
)
)
elif curve_spec.airfoil not in document.airfoils:
diagnostics.append(
build_validation_error(
code="airfoil.document.curve.unknown_airfoil",
message=(
"Document curve references an unknown airfoil name: "
f"{curve_spec.airfoil!r}."
),
field_path=f"curves.{curve_name}.airfoil",
object_path="airfoil.document",
)
)
if curve_spec.approximation.spacing not in {"uniform", "cosine"}:
diagnostics.append(
build_validation_error(
code="airfoil.document.curve.bad_spacing",
message=(
"Document curve approximation spacing must be "
"'uniform' or 'cosine'."
),
field_path=f"curves.{curve_name}.approximation.spacing",
object_path="airfoil.document",
)
)
if curve_spec.approximation.num_points < _MIN_CURVE_APPROXIMATION_POINTS:
diagnostics.append(
build_validation_error(
code="airfoil.document.curve.bad_num_points",
message=(
"Document curve approximation num_points must be at "
"least 2."
),
field_path=f"curves.{curve_name}.approximation.num_points",
object_path="airfoil.document",
)
)
return tuple(diagnostics)
def _is_numeric_point2d(point: object) -> bool:
"""Return whether one runtime value looks like a numeric 2D point."""
if not isinstance(point, tuple | list):
return False
point_sequence = cast(tuple[object, ...] | list[object], point)
if len(point_sequence) != POINT2D_TUPLE_LENGTH:
return False
x_value: object = point_sequence[0]
y_value: object = point_sequence[1]
return isinstance(x_value, int | float) and isinstance(
y_value,
int | float,
)
def _validate_document_spline_curve(
*,
curve_name: str,
curve_spec: SplineCurveSpec,
) -> tuple[Diagnostic, ...]:
"""Validate one named spline-backed document curve entry."""
diagnostics: list[Diagnostic] = []
if curve_spec.representation != "bezier":
diagnostics.append(
build_validation_error(
code="airfoil.document.curve.spline.bad_representation",
message=(
"Document spline curve representation must be 'bezier'."
),
field_path=f"curves.{curve_name}.representation",
object_path="airfoil.document",
)
)
if len(curve_spec.control_points) < _MIN_SPLINE_CURVE_CONTROL_POINTS:
diagnostics.append(
build_validation_error(
code="airfoil.document.curve.spline.control_points_too_short",
message=(
"Document spline curve control_points must contain at "
"least two points."
),
field_path=f"curves.{curve_name}.control_points",
object_path="airfoil.document",
)
)
return tuple(diagnostics)
control_points = cast(list[object], curve_spec.control_points)
for index, point in enumerate(control_points):
if not _is_numeric_point2d(point):
diagnostics.append(
build_validation_error(
code=(
"airfoil.document.curve.spline.control_point_invalid"
),
message=(
"Document spline curve control_points entries must "
"be numeric two-component points."
),
field_path=(f"curves.{curve_name}.control_points.{index}"),
object_path="airfoil.document",
)
)
break
return tuple(diagnostics)
def _validate_document_polyline_curve(
*,
curve_name: str,
curve_spec: PolylineCurveSpec,
) -> tuple[Diagnostic, ...]:
"""Validate one named polyline-backed document curve entry."""
diagnostics: list[Diagnostic] = []
if len(curve_spec.points) < _MIN_POLYLINE_CURVE_POINTS:
diagnostics.append(
build_validation_error(
code="airfoil.document.curve.polyline.points_too_short",
message=(
"Document polyline curve points must contain at least "
"two points."
),
field_path=f"curves.{curve_name}.points",
object_path="airfoil.document",
)
)
return tuple(diagnostics)
points = cast(list[object], curve_spec.points)
for index, point in enumerate(points):
if not _is_numeric_point2d(point):
diagnostics.append(
build_validation_error(
code="airfoil.document.curve.polyline.point_invalid",
message=(
"Document polyline curve points entries must be "
"numeric two-component points."
),
field_path=f"curves.{curve_name}.points.{index}",
object_path="airfoil.document",
)
)
return tuple(diagnostics)
for index in range(len(curve_spec.points) - 1):
if curve_spec.points[index] == curve_spec.points[index + 1]:
diagnostics.append(
build_validation_error(
code=(
"airfoil.document.curve.polyline.zero_length_segment"
),
message=(
"Document polyline curve points must not contain "
"consecutive duplicate vertices."
),
field_path=f"curves.{curve_name}.points.{index + 1}",
object_path="airfoil.document",
)
)
break
return tuple(diagnostics)
__all__ = [
"validate_document",
"validate_file_airfoil_source",
"validate_spec",
]