"""High-level modal-analysis pipeline.
``ModalPipeline`` runs the full workflow end to end: for each operating point it
applies the MBC3 transform, solves for modes, then links the modes across
operating points, assembles the Campbell diagram, and flags resonance crossings.
The :class:`PipelineResult` exposes every intermediate product and convenience
exports to a state-space system, so a complete analysis is a few lines of code.
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from typing import TYPE_CHECKING
from vane.campbell.builder import build_campbell
from vane.campbell.excitation import (
DEFAULT_HARMONICS,
_validate_harmonics,
find_resonances,
)
from vane.io.mbc_transform import mbc3_transform
from vane.modal.eigensolver import modes_from_mbc
from vane.modal.identifier import identify_modes
from vane.sysid.state_space import state_space_from_mbc
if TYPE_CHECKING:
from collections.abc import Sequence
from vane.campbell.builder import CampbellDiagram
from vane.campbell.excitation import ResonanceCrossing
from vane.io.lin_reader import LinFile
from vane.io.mbc_transform import MBCResult
from vane.modal.eigensolver import ModalSolution
from vane.modal.identifier import IdentificationResult, ModeTrack
from vane.sysid.state_space import StateSpace
__all__ = ["ModalPipeline", "PipelineResult"]
_RPM_PARAMETER = "rotor_speed_rpm"
_WIND_PARAMETER = "wind_speed"
[docs]
@dataclass
class PipelineResult:
"""Outputs of a full modal-analysis run.
Parameters
----------
mbc_results : list[MBCResult]
The MBC transform result for each operating point.
solutions : list[ModalSolution]
The modal solution for each operating point.
identification : IdentificationResult
The cross-operating-point mode tracks.
campbell : CampbellDiagram
The assembled Campbell diagram.
resonances : list[ResonanceCrossing]
Detected resonance crossings (empty for non-rotor-speed runs).
parameter_name : str
Name of the operating parameter used.
"""
mbc_results: list[MBCResult]
solutions: list[ModalSolution]
identification: IdentificationResult
campbell: CampbellDiagram
resonances: list[ResonanceCrossing]
parameter_name: str
@property
def tracks(self) -> list[ModeTrack]:
"""Return the identified mode tracks."""
return self.identification.tracks
[docs]
def state_space(self, operating_point: int = 0) -> StateSpace:
"""Export the state-space system at one operating point.
Parameters
----------
operating_point : int, optional
Index of the operating point to export (default the first).
Returns
-------
StateSpace
The continuous-time system from that operating point's MBC result.
Raises
------
IndexError
If ``operating_point`` is negative or out of range.
"""
if not 0 <= operating_point < len(self.mbc_results):
msg = (
f"operating_point {operating_point} out of range "
f"[0, {len(self.mbc_results)})"
)
raise IndexError(msg)
return state_space_from_mbc(self.mbc_results[operating_point])
[docs]
class ModalPipeline:
"""End-to-end modal analysis from linearization files to a Campbell diagram."""
def __init__(
self,
*,
frequency_weight: float = 0.5,
mac_threshold: float = 0.5,
harmonics: Sequence[int] = DEFAULT_HARMONICS,
) -> None:
"""Configure the pipeline.
Parameters
----------
frequency_weight : float, optional
Frequency-continuity weight for cross-operating-point tracking.
mac_threshold : float, optional
Minimum MAC for a track to be extended.
harmonics : Sequence[int], optional
Excitation harmonics for resonance detection.
Raises
------
ValueError
If ``frequency_weight`` or ``mac_threshold`` is not in ``[0, 1]``, or
``harmonics`` contains a non-positive, fractional, non-finite, or
duplicate value.
"""
for name, value in (
("frequency_weight", frequency_weight),
("mac_threshold", mac_threshold),
):
if not 0.0 <= value <= 1.0:
msg = f"{name} must be in [0, 1], got {value}"
raise ValueError(msg)
self.frequency_weight = frequency_weight
self.mac_threshold = mac_threshold
self.harmonics = tuple(_validate_harmonics(harmonics))
[docs]
def run(
self,
operating_points: Sequence[Sequence[LinFile]],
*,
parameter_name: str = _RPM_PARAMETER,
) -> PipelineResult:
"""Run the full pipeline over a set of operating points.
Parameters
----------
operating_points : Sequence[Sequence[LinFile]]
One entry per operating point, each a sequence of linearization files
spanning that point's azimuth sweep.
parameter_name : str, optional
The operating parameter to plot against: ``"rotor_speed_rpm"`` (the
default) or ``"wind_speed"``.
Returns
-------
PipelineResult
Every intermediate product of the analysis. Operating points are
ordered by ascending operating parameter regardless of input order, so
tracking and resonance detection see a monotonic sweep.
Raises
------
ValueError
If ``operating_points`` is empty, ``parameter_name`` is unknown, or the
operating-parameter values are non-finite or not distinct across points.
"""
if not operating_points:
msg = "operating_points must contain at least one operating point"
raise ValueError(msg)
if parameter_name not in (_RPM_PARAMETER, _WIND_PARAMETER):
msg = (
f"parameter_name must be '{_RPM_PARAMETER}' or '{_WIND_PARAMETER}', "
f"got '{parameter_name}'"
)
raise ValueError(msg)
mbc_results = [mbc3_transform(list(point)) for point in operating_points]
solutions = [modes_from_mbc(result) for result in mbc_results]
parameter_values = [
self._parameter_value(result, parameter_name) for result in mbc_results
]
self._validate_parameter_values(parameter_values, parameter_name)
# Sort by the operating parameter so cross-operating-point tracking and
# resonance interpolation see a monotonic sweep regardless of input order.
order = sorted(range(len(parameter_values)), key=parameter_values.__getitem__)
mbc_results = [mbc_results[i] for i in order]
solutions = [solutions[i] for i in order]
parameter_values = [parameter_values[i] for i in order]
identification = identify_modes(
solutions,
frequency_weight=self.frequency_weight,
mac_threshold=self.mac_threshold,
)
campbell = build_campbell(
identification, parameter_values, parameter_name=parameter_name
)
resonances = (
find_resonances(campbell, self.harmonics)
if parameter_name == _RPM_PARAMETER
else []
)
return PipelineResult(
mbc_results=mbc_results,
solutions=solutions,
identification=identification,
campbell=campbell,
resonances=resonances,
parameter_name=parameter_name,
)
@staticmethod
def _parameter_value(result: MBCResult, parameter_name: str) -> float:
"""Return the operating-parameter value from an MBC result."""
if parameter_name == _RPM_PARAMETER:
return result.rotor_speed_rpm
return result.wind_speed
@staticmethod
def _validate_parameter_values(
values: Sequence[float], parameter_name: str
) -> None:
"""Validate the operating parameter is finite and distinct across points.
A non-finite operating parameter or two operating points sharing the same
value makes the Campbell diagram and resonance interpolation ambiguous, so
both are rejected rather than silently producing an untrustworthy sweep.
Parameters
----------
values : Sequence[float]
The operating-parameter value at each operating point.
parameter_name : str
The operating-parameter name, for the error message.
Raises
------
ValueError
If any value is non-finite or two operating points share a value.
"""
if not all(math.isfinite(value) for value in values):
msg = (
f"operating parameter '{parameter_name}' has non-finite "
f"values: {values}"
)
raise ValueError(msg)
if len(set(values)) != len(values):
msg = (
f"operating points must have distinct '{parameter_name}' values; "
f"got duplicates in {values}"
)
raise ValueError(msg)