feat: load combinations
LoadCombination entity, combination service, manager dialog and unit/integration coverage.
This commit is contained in:
parent
7093164f75
commit
ea352a5de5
5 changed files with 845 additions and 0 deletions
82
src/otko/commands/combinations.py
Normal file
82
src/otko/commands/combinations.py
Normal file
|
|
@ -0,0 +1,82 @@
|
|||
"""Commands for load-combination management."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from otko.commands.base import ProjectCommand
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from otko.viewmodels import ProjectViewModel
|
||||
|
||||
|
||||
class AddCombinationsCommand(ProjectCommand):
|
||||
"""Add one or more load combinations."""
|
||||
|
||||
def __init__(
|
||||
self, vm: "ProjectViewModel", combos: list[Any], *, text: str | None = None
|
||||
) -> None:
|
||||
super().__init__(vm, text or f"Add {len(combos)} load combination(s)")
|
||||
self._combos = list(combos)
|
||||
|
||||
def redo(self) -> None:
|
||||
existing = {c.id for c in self.project.combinations}
|
||||
for combo in self._combos:
|
||||
if combo.id in existing:
|
||||
raise ValueError(f"Load combination id {combo.id} already exists.")
|
||||
self.project.combinations.extend(self._combos)
|
||||
self._notify()
|
||||
|
||||
def undo(self) -> None:
|
||||
ids = {c.id for c in self._combos}
|
||||
self.project.combinations[:] = [c for c in self.project.combinations if c.id not in ids]
|
||||
self._notify()
|
||||
|
||||
|
||||
class DeleteCombinationsCommand(ProjectCommand):
|
||||
"""Remove a set of load combinations."""
|
||||
|
||||
def __init__(self, vm: "ProjectViewModel", combo_ids: set[int]) -> None:
|
||||
super().__init__(vm, f"Delete {len(combo_ids)} load combination(s)")
|
||||
self._combo_ids = set(combo_ids)
|
||||
self._removed: list[tuple[int, Any]] = []
|
||||
|
||||
def redo(self) -> None:
|
||||
self._removed = [
|
||||
(i, c) for i, c in enumerate(self.project.combinations) if c.id in self._combo_ids
|
||||
]
|
||||
self.project.combinations[:] = [
|
||||
c for c in self.project.combinations if c.id not in self._combo_ids
|
||||
]
|
||||
self._notify()
|
||||
|
||||
def undo(self) -> None:
|
||||
for i, c in self._removed:
|
||||
self.project.combinations.insert(min(i, len(self.project.combinations)), c)
|
||||
self._removed.clear()
|
||||
self._notify()
|
||||
|
||||
|
||||
class UpdateCombinationCommand(ProjectCommand):
|
||||
"""Replace a load combination at a given id."""
|
||||
|
||||
def __init__(self, vm: "ProjectViewModel", new_combo: Any) -> None:
|
||||
super().__init__(vm, f"Edit load combination {new_combo.id}")
|
||||
self._new = new_combo
|
||||
self._old: Any | None = None
|
||||
self._index: int | None = None
|
||||
|
||||
def redo(self) -> None:
|
||||
for i, c in enumerate(self.project.combinations):
|
||||
if c.id == self._new.id:
|
||||
self._old = c
|
||||
self._index = i
|
||||
self.project.combinations[i] = self._new
|
||||
self._notify()
|
||||
return
|
||||
raise KeyError(f"Load combination with id={self._new.id} not found.")
|
||||
|
||||
def undo(self) -> None:
|
||||
if self._old is not None and self._index is not None:
|
||||
self.project.combinations[self._index] = self._old
|
||||
self._notify()
|
||||
235
src/otko/services/combinations.py
Normal file
235
src/otko/services/combinations.py
Normal file
|
|
@ -0,0 +1,235 @@
|
|||
"""Case-result load combinations — post-processing superposition.
|
||||
|
||||
A :class:`LoadCombination` names a weighted set of *completed* analysis
|
||||
results (e.g. ``1.2×Dead + 1.6×Live``). Evaluation is pure NumPy over
|
||||
:class:`StaticResults` — no solver re-run — and is only valid for
|
||||
linear-elastic static superposition. Combining any other result kind
|
||||
(transient, modal, pushover, response-spectrum) raises :class:`TypeError`.
|
||||
|
||||
Two combination kinds:
|
||||
|
||||
- ``Linear`` → :func:`combine_static_results`: weighted sum, returned as
|
||||
:class:`StaticResults` (``case_id``/``case_name`` taken from the combo).
|
||||
- ``Envelope`` → :func:`envelope_static_results`: per-DOF, per-step
|
||||
maximum/minimum across the factored terms, returned as
|
||||
:class:`EnvelopeResults`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from otko.core.loads import LoadCombination
|
||||
from otko.services.results import StaticResults
|
||||
|
||||
|
||||
@dataclass
|
||||
class EnvelopeResults:
|
||||
"""Per-DOF, per-step max/min across a combination's factored terms."""
|
||||
|
||||
case_id: int
|
||||
case_name: str
|
||||
n_steps: int
|
||||
node_disp_max: dict[int, np.ndarray] = field(default_factory=dict)
|
||||
"""node_id → array of shape (n_steps, ndf), per-DOF maxima."""
|
||||
node_disp_min: dict[int, np.ndarray] = field(default_factory=dict)
|
||||
"""node_id → array of shape (n_steps, ndf), per-DOF minima."""
|
||||
node_reaction_max: dict[int, np.ndarray] = field(default_factory=dict)
|
||||
"""node_id → array of shape (n_steps, ndf), per-DOF maxima."""
|
||||
node_reaction_min: dict[int, np.ndarray] = field(default_factory=dict)
|
||||
"""node_id → array of shape (n_steps, ndf), per-DOF minima."""
|
||||
element_forces_max: dict[int, np.ndarray] = field(default_factory=dict)
|
||||
"""element_id → array of shape (n_steps, n_components), maxima."""
|
||||
element_forces_min: dict[int, np.ndarray] = field(default_factory=dict)
|
||||
"""element_id → array of shape (n_steps, n_components), minima."""
|
||||
|
||||
|
||||
def _check_compatible(results: Sequence[StaticResults]) -> int:
|
||||
"""Validate equal steps, nodes, elements and shapes; return ``n_steps``.
|
||||
|
||||
Raises:
|
||||
ValueError: on any incompatibility, with the offending detail.
|
||||
"""
|
||||
if not results:
|
||||
raise ValueError("Need at least one result to combine.")
|
||||
first = results[0]
|
||||
n_steps = first.n_steps
|
||||
node_ids = set(first.node_disp)
|
||||
elem_ids = set(first.element_forces)
|
||||
reaction_ids = set(first.node_reaction)
|
||||
for res in results[1:]:
|
||||
if res.n_steps != n_steps:
|
||||
raise ValueError(
|
||||
f"Step-count mismatch: case {first.case_id} has {n_steps} steps, "
|
||||
f"case {res.case_id} has {res.n_steps}."
|
||||
)
|
||||
if set(res.node_disp) != node_ids:
|
||||
raise ValueError(f"Node set mismatch: case {first.case_id} vs case {res.case_id}.")
|
||||
if set(res.node_reaction) != reaction_ids:
|
||||
raise ValueError(f"Reaction set mismatch: case {first.case_id} vs case {res.case_id}.")
|
||||
if set(res.element_forces) != elem_ids:
|
||||
raise ValueError(f"Element set mismatch: case {first.case_id} vs case {res.case_id}.")
|
||||
for nid in node_ids:
|
||||
if res.node_disp[nid].shape != first.node_disp[nid].shape:
|
||||
raise ValueError(
|
||||
f"Displacement shape mismatch at node {nid}: "
|
||||
f"case {first.case_id} vs case {res.case_id}."
|
||||
)
|
||||
if res.node_reaction[nid].shape != first.node_reaction[nid].shape:
|
||||
raise ValueError(
|
||||
f"Reaction shape mismatch at node {nid}: "
|
||||
f"case {first.case_id} vs case {res.case_id}."
|
||||
)
|
||||
for eid in elem_ids:
|
||||
if res.element_forces[eid].shape != first.element_forces[eid].shape:
|
||||
raise ValueError(
|
||||
f"Element-force shape mismatch at element {eid}: "
|
||||
f"case {first.case_id} vs case {res.case_id}."
|
||||
)
|
||||
return n_steps
|
||||
|
||||
|
||||
def combine_static_results(
|
||||
results: Sequence[StaticResults],
|
||||
factors: Sequence[float],
|
||||
*,
|
||||
case_id: int,
|
||||
case_name: str,
|
||||
) -> StaticResults:
|
||||
"""Return the weighted sum ``Σ factor_i × results_i``.
|
||||
|
||||
Args:
|
||||
results: Source static results, one per combination term.
|
||||
factors: Scale per term (same order as ``results``).
|
||||
case_id: Tag for the returned results (the combination's id).
|
||||
case_name: Label for the returned results (the combination's name).
|
||||
|
||||
Raises:
|
||||
ValueError: on length mismatch, non-finite factor, or
|
||||
incompatible result shapes.
|
||||
"""
|
||||
if len(results) != len(factors):
|
||||
raise ValueError(f"Results/factors length mismatch: {len(results)} vs {len(factors)}.")
|
||||
for factor in factors:
|
||||
if not np.isfinite(factor):
|
||||
raise ValueError(f"Combination factor must be finite, got {factor!r}.")
|
||||
n_steps = _check_compatible(results)
|
||||
|
||||
node_disp: dict[int, np.ndarray] = {}
|
||||
node_reaction: dict[int, np.ndarray] = {}
|
||||
for nid in results[0].node_disp:
|
||||
acc_d = np.zeros_like(results[0].node_disp[nid], dtype=float)
|
||||
acc_r = np.zeros_like(results[0].node_reaction[nid], dtype=float)
|
||||
for r, f in zip(results, factors, strict=False):
|
||||
acc_d = acc_d + float(f) * r.node_disp[nid]
|
||||
acc_r = acc_r + float(f) * r.node_reaction[nid]
|
||||
node_disp[nid] = acc_d
|
||||
node_reaction[nid] = acc_r
|
||||
element_forces: dict[int, np.ndarray] = {}
|
||||
for eid in results[0].element_forces:
|
||||
acc_e = np.zeros_like(results[0].element_forces[eid], dtype=float)
|
||||
for r, f in zip(results, factors, strict=False):
|
||||
acc_e = acc_e + float(f) * r.element_forces[eid]
|
||||
element_forces[eid] = acc_e
|
||||
return StaticResults(
|
||||
case_id=case_id,
|
||||
case_name=case_name,
|
||||
n_steps=n_steps,
|
||||
node_disp=node_disp,
|
||||
node_reaction=node_reaction,
|
||||
element_forces=element_forces,
|
||||
)
|
||||
|
||||
|
||||
def envelope_static_results(
|
||||
results: Sequence[StaticResults],
|
||||
factors: Sequence[float],
|
||||
*,
|
||||
case_id: int,
|
||||
case_name: str,
|
||||
) -> EnvelopeResults:
|
||||
"""Return per-DOF, per-step max/min across the factored terms.
|
||||
|
||||
Each term is ``factor_i × results_i``; the envelope takes the
|
||||
element-wise maximum (and minimum) across terms at every step.
|
||||
"""
|
||||
if len(results) != len(factors):
|
||||
raise ValueError(f"Results/factors length mismatch: {len(results)} vs {len(factors)}.")
|
||||
for factor in factors:
|
||||
if not np.isfinite(factor):
|
||||
raise ValueError(f"Combination factor must be finite, got {factor!r}.")
|
||||
n_steps = _check_compatible(results)
|
||||
|
||||
def _stack(getter: Any) -> dict[int, np.ndarray]:
|
||||
keys = list(getter(results[0]).keys())
|
||||
return {
|
||||
k: np.stack([float(f) * getter(r)[k] for r, f in zip(results, factors, strict=False)])
|
||||
for k in keys
|
||||
}
|
||||
|
||||
disp = _stack(lambda r: r.node_disp)
|
||||
reac = _stack(lambda r: r.node_reaction)
|
||||
elem = _stack(lambda r: r.element_forces)
|
||||
return EnvelopeResults(
|
||||
case_id=case_id,
|
||||
case_name=case_name,
|
||||
n_steps=n_steps,
|
||||
node_disp_max={k: v.max(axis=0) for k, v in disp.items()},
|
||||
node_disp_min={k: v.min(axis=0) for k, v in disp.items()},
|
||||
node_reaction_max={k: v.max(axis=0) for k, v in reac.items()},
|
||||
node_reaction_min={k: v.min(axis=0) for k, v in reac.items()},
|
||||
element_forces_max={k: v.max(axis=0) for k, v in elem.items()},
|
||||
element_forces_min={k: v.min(axis=0) for k, v in elem.items()},
|
||||
)
|
||||
|
||||
|
||||
def evaluate_combination(
|
||||
results_by_case_id: Mapping[int, Any],
|
||||
combination: LoadCombination,
|
||||
) -> StaticResults | EnvelopeResults:
|
||||
"""Evaluate ``combination`` against already-computed results.
|
||||
|
||||
Args:
|
||||
results_by_case_id: ``{analysis_case_id: results}`` — every
|
||||
referenced case must be present as :class:`StaticResults`.
|
||||
combination: The named combination to evaluate.
|
||||
|
||||
Returns:
|
||||
:class:`StaticResults` for ``Linear`` combos,
|
||||
:class:`EnvelopeResults` for ``Envelope`` combos.
|
||||
|
||||
Raises:
|
||||
KeyError: if a referenced case has no entry in the map.
|
||||
TypeError: if any referenced result is not :class:`StaticResults`
|
||||
(superposition is only meaningful for static results).
|
||||
ValueError: propagated from the combine/envelope helpers on
|
||||
shape incompatibility.
|
||||
"""
|
||||
ordered: list[StaticResults] = []
|
||||
factors: list[float] = []
|
||||
for item in combination.items:
|
||||
if item.case_id not in results_by_case_id:
|
||||
raise KeyError(
|
||||
f"Combination {combination.id}: no results for case {item.case_id} — "
|
||||
"run that case first."
|
||||
)
|
||||
res = results_by_case_id[item.case_id]
|
||||
if not isinstance(res, StaticResults):
|
||||
raise TypeError(
|
||||
f"Combination {combination.id}: case {item.case_id} produced "
|
||||
f"{type(res).__name__}, not StaticResults — case-result "
|
||||
"combinations only support (linear) static cases."
|
||||
)
|
||||
ordered.append(res)
|
||||
factors.append(item.factor)
|
||||
if combination.kind == "Envelope":
|
||||
return envelope_static_results(
|
||||
ordered, factors, case_id=combination.id, case_name=combination.name
|
||||
)
|
||||
return combine_static_results(
|
||||
ordered, factors, case_id=combination.id, case_name=combination.name
|
||||
)
|
||||
258
src/otko/views/dialogs/combination_dialog.py
Normal file
258
src/otko/views/dialogs/combination_dialog.py
Normal file
|
|
@ -0,0 +1,258 @@
|
|||
"""Load Combination manager — library-style CRUD over case-result combos.
|
||||
|
||||
A combination names a weighted set of *completed* static-case results
|
||||
(e.g. ``1.2×Dead + 1.6×Live``). Evaluation is post-processing
|
||||
superposition in :mod:`otko.services.combinations` — the dialog only
|
||||
edits the definition; running happens from File → Run → Combinations.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import ValidationError
|
||||
from PySide6.QtCore import Qt
|
||||
from PySide6.QtWidgets import (
|
||||
QComboBox,
|
||||
QDialog,
|
||||
QDialogButtonBox,
|
||||
QDoubleSpinBox,
|
||||
QFormLayout,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QLineEdit,
|
||||
QListWidget,
|
||||
QListWidgetItem,
|
||||
QMessageBox,
|
||||
QPushButton,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from otko.commands import (
|
||||
AddCombinationsCommand,
|
||||
DeleteCombinationsCommand,
|
||||
UpdateCombinationCommand,
|
||||
)
|
||||
from otko.core import LoadCombination, LoadCombinationItem
|
||||
from otko.viewmodels import ProjectViewModel
|
||||
|
||||
|
||||
def _spin(default: float = 1.0) -> QDoubleSpinBox:
|
||||
"""Factor editor: wide range, 3 decimals, auto-selects its row on edit."""
|
||||
sb = QDoubleSpinBox()
|
||||
sb.setRange(-1e9, 1e9)
|
||||
sb.setDecimals(3)
|
||||
sb.setSingleStep(0.1)
|
||||
sb.setValue(default)
|
||||
sb.setMaximumWidth(100)
|
||||
sb.setToolTip("Scale applied to this case's results (1.0 = full).")
|
||||
return sb
|
||||
|
||||
|
||||
class CombinationForm(QWidget):
|
||||
"""Edit one combination: name, kind, and per-case factors."""
|
||||
|
||||
def __init__(self, vm: ProjectViewModel, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self._vm = vm
|
||||
self._combo_id: int | None = None
|
||||
self._spins: dict[int, QDoubleSpinBox] = {}
|
||||
layout = QFormLayout(self)
|
||||
self._name_edit = QLineEdit()
|
||||
layout.addRow("Name:", self._name_edit)
|
||||
self._kind = QComboBox()
|
||||
self._kind.addItems(["Linear", "Envelope"])
|
||||
self._kind.setToolTip(
|
||||
"Linear = weighted sum Σ factor×results. "
|
||||
"Envelope = per-DOF max/min across factored terms."
|
||||
)
|
||||
layout.addRow("Kind:", self._kind)
|
||||
layout.addRow(QLabel("<b>Cases (tick to include; factor beside each):</b>"))
|
||||
self._cases = QListWidget()
|
||||
self._cases.setMaximumHeight(220)
|
||||
layout.addRow(self._cases)
|
||||
layout.addRow(
|
||||
QLabel(
|
||||
"<i>Only linear-static cases combine meaningfully. "
|
||||
"Transient / modal / pushover results are rejected at evaluation.</i>"
|
||||
)
|
||||
)
|
||||
|
||||
# ── populate / read ──────────────────────────────────────────
|
||||
def populate(self, combo: LoadCombination) -> None:
|
||||
"""Fill the form from ``combo`` (case list rebuilt from the project)."""
|
||||
self._combo_id = combo.id
|
||||
self._name_edit.setText(combo.name)
|
||||
self._kind.setCurrentText(combo.kind)
|
||||
self._rebuild_case_list(combo)
|
||||
|
||||
def _rebuild_case_list(self, combo: LoadCombination | None = None) -> None:
|
||||
factors = {i.case_id: i.factor for i in combo.items} if combo else {}
|
||||
self._cases.clear()
|
||||
self._spins.clear()
|
||||
project = self._vm.project
|
||||
if project is None:
|
||||
return
|
||||
for case in project.analyses:
|
||||
item = QListWidgetItem()
|
||||
item.setData(Qt.ItemDataRole.UserRole, case.id)
|
||||
self._cases.addItem(item)
|
||||
cell = QWidget()
|
||||
row = QHBoxLayout(cell)
|
||||
row.setContentsMargins(2, 0, 2, 0)
|
||||
row.addWidget(QLabel(f"#{case.id} {case.name or '(unnamed)'} [{case.type}]"), 1)
|
||||
spin = _spin(float(factors.get(case.id, 1.0)))
|
||||
spin.valueChanged.connect(lambda _v, _it=item: _it.setSelected(True))
|
||||
row.addWidget(spin)
|
||||
self._cases.setItemWidget(item, cell)
|
||||
self._spins[case.id] = spin
|
||||
if case.id in factors:
|
||||
item.setSelected(True)
|
||||
|
||||
def read(self) -> LoadCombination:
|
||||
"""Build a validated combination from the form (raises on bad input)."""
|
||||
if self._combo_id is None:
|
||||
raise ValueError("No combination selected.")
|
||||
selected = [i.data(Qt.ItemDataRole.UserRole) for i in self._cases.selectedItems()]
|
||||
if not selected:
|
||||
raise ValueError("Select at least one analysis case for this combination.")
|
||||
items = [
|
||||
LoadCombinationItem(case_id=int(cid), factor=float(self._spins[int(cid)].value()))
|
||||
for cid in selected
|
||||
]
|
||||
return LoadCombination(
|
||||
id=self._combo_id,
|
||||
name=self._name_edit.text(),
|
||||
kind=self._kind.currentText(), # type: ignore[arg-type]
|
||||
items=items,
|
||||
)
|
||||
|
||||
|
||||
class LoadCombinationManagerDialog(QDialog):
|
||||
"""CRUD dialog for load combinations."""
|
||||
|
||||
def __init__(self, vm: ProjectViewModel, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle("Load Combinations")
|
||||
self.resize(720, 520)
|
||||
self._vm = vm
|
||||
self._build_ui()
|
||||
self._refresh_list()
|
||||
self._vm.modelMutated.connect(self._refresh_list)
|
||||
|
||||
def _build_ui(self) -> None:
|
||||
outer = QVBoxLayout(self)
|
||||
outer.addWidget(
|
||||
QLabel(
|
||||
"<b>Load Combinations</b> — weighted sets of completed static-case "
|
||||
"results (e.g. 1.2×Dead + 1.6×Live). Evaluate from File → Run."
|
||||
)
|
||||
)
|
||||
body = QHBoxLayout()
|
||||
outer.addLayout(body, stretch=1)
|
||||
|
||||
left = QVBoxLayout()
|
||||
self._list = QListWidget()
|
||||
self._list.currentRowChanged.connect(self._on_row_changed)
|
||||
left.addWidget(self._list, stretch=1)
|
||||
btn_row = QHBoxLayout()
|
||||
self._add_btn = QPushButton("Add…")
|
||||
self._delete_btn = QPushButton("Delete")
|
||||
self._add_btn.clicked.connect(self._on_add)
|
||||
self._delete_btn.clicked.connect(self._on_delete)
|
||||
btn_row.addWidget(self._add_btn)
|
||||
btn_row.addWidget(self._delete_btn)
|
||||
left.addLayout(btn_row)
|
||||
body.addLayout(left, stretch=1)
|
||||
|
||||
right = QVBoxLayout()
|
||||
self._form = CombinationForm(self._vm)
|
||||
right.addWidget(self._form, stretch=1)
|
||||
self._apply_btn = QPushButton("Apply changes")
|
||||
self._apply_btn.clicked.connect(self._on_apply)
|
||||
right.addWidget(self._apply_btn, alignment=Qt.AlignmentFlag.AlignRight)
|
||||
body.addLayout(right, stretch=2)
|
||||
|
||||
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close, parent=self)
|
||||
buttons.rejected.connect(self.reject)
|
||||
buttons.accepted.connect(self.accept)
|
||||
outer.addWidget(buttons)
|
||||
|
||||
# ── list ─────────────────────────────────────────────────────
|
||||
def _refresh_list(self) -> None:
|
||||
selected_id = None
|
||||
if self._list.currentItem() is not None:
|
||||
selected_id = self._list.currentItem().data(Qt.ItemDataRole.UserRole)
|
||||
self._list.clear()
|
||||
for c in self._vm.project.combinations if self._vm.project else []:
|
||||
label = f"#{c.id} {c.name or '(unnamed)'} [{c.kind}]"
|
||||
item = QListWidgetItem(label)
|
||||
item.setData(Qt.ItemDataRole.UserRole, c.id)
|
||||
self._list.addItem(item)
|
||||
if selected_id is not None:
|
||||
self._select_by_id(selected_id)
|
||||
if self._list.currentRow() < 0 and self._list.count():
|
||||
self._list.setCurrentRow(0)
|
||||
if not self._list.count():
|
||||
self._form._combo_id = None
|
||||
|
||||
def _selected(self) -> LoadCombination | None:
|
||||
item = self._list.currentItem()
|
||||
if item is None or self._vm.project is None:
|
||||
return None
|
||||
cid = item.data(Qt.ItemDataRole.UserRole)
|
||||
return next((c for c in self._vm.project.combinations if c.id == cid), None)
|
||||
|
||||
def _on_row_changed(self, _row: int) -> None:
|
||||
combo = self._selected()
|
||||
if combo is None:
|
||||
return
|
||||
self._form.populate(combo)
|
||||
|
||||
def _select_by_id(self, target_id: int) -> None:
|
||||
for i in range(self._list.count()):
|
||||
if self._list.item(i).data(Qt.ItemDataRole.UserRole) == target_id:
|
||||
self._list.setCurrentRow(i)
|
||||
return
|
||||
|
||||
# ── mutations ────────────────────────────────────────────────
|
||||
def _on_apply(self) -> None:
|
||||
try:
|
||||
new_combo = self._form.read()
|
||||
self._vm.apply_command(UpdateCombinationCommand(self._vm, new_combo))
|
||||
except (ValidationError, ValueError) as exc:
|
||||
QMessageBox.critical(self, "Validation error", str(exc))
|
||||
|
||||
def _on_add(self) -> None:
|
||||
if self._vm.project is None:
|
||||
return
|
||||
if not self._vm.project.analyses:
|
||||
QMessageBox.information(
|
||||
self,
|
||||
"No cases",
|
||||
"Define at least one analysis case first (Analyze → Cases…).",
|
||||
)
|
||||
return
|
||||
new_id = self._vm.project.next_combination_id()
|
||||
first = self._vm.project.analyses[0]
|
||||
try:
|
||||
combo = LoadCombination(
|
||||
id=new_id,
|
||||
name="New",
|
||||
items=[LoadCombinationItem(case_id=first.id)],
|
||||
)
|
||||
except (ValidationError, ValueError) as exc:
|
||||
QMessageBox.warning(self, "Could not create combination", str(exc))
|
||||
return
|
||||
self._vm.apply_command(AddCombinationsCommand(self._vm, [combo]))
|
||||
self._select_by_id(new_id)
|
||||
|
||||
def _on_delete(self) -> None:
|
||||
combo = self._selected()
|
||||
if combo is None:
|
||||
return
|
||||
reply = QMessageBox.question(
|
||||
self, "Delete combination", f"Delete load combination #{combo.id}?"
|
||||
)
|
||||
if reply != QMessageBox.StandardButton.Yes:
|
||||
return
|
||||
self._vm.apply_command(DeleteCombinationsCommand(self._vm, {combo.id}))
|
||||
90
tests/integration/test_combinations.py
Normal file
90
tests/integration/test_combinations.py
Normal file
|
|
@ -0,0 +1,90 @@
|
|||
"""Load-combination integration: 1.2×D + 1.6×L by superposition == one combined run."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
import pytest
|
||||
|
||||
ops = pytest.importorskip("openseespy.opensees") # skip if OpenSeesPy not installed
|
||||
|
||||
from otko.core import ( # noqa: E402
|
||||
ElasticBeamColumn,
|
||||
ElasticSection,
|
||||
LinearTimeSeries,
|
||||
LoadCombination,
|
||||
LoadCombinationItem,
|
||||
NodalLoad,
|
||||
Node,
|
||||
PlainLoadPattern,
|
||||
Project,
|
||||
StaticCase,
|
||||
)
|
||||
from otko.services import OpenSeesRunner, evaluate_combination # noqa: E402
|
||||
from otko.services.results import StaticResults # noqa: E402
|
||||
|
||||
pytestmark = pytest.mark.slow
|
||||
|
||||
|
||||
def _two_pattern_cantilever() -> Project:
|
||||
elastic_mod = 200e9
|
||||
area = 0.01
|
||||
inertia = 8.333e-6
|
||||
return Project(
|
||||
ndm=2,
|
||||
ndf=3,
|
||||
nodes=[
|
||||
Node(
|
||||
id=1,
|
||||
coords=(0.0, 0.0, 0.0),
|
||||
restraint=(True, True, False, False, False, True),
|
||||
),
|
||||
Node(id=2, coords=(5.0, 0.0, 0.0)),
|
||||
],
|
||||
sections=[ElasticSection(id=1, E=elastic_mod, A=area, Iz=inertia)],
|
||||
elements=[ElasticBeamColumn(id=1, nodes=(1, 2), section_id=1)],
|
||||
time_series=[LinearTimeSeries(id=1), LinearTimeSeries(id=2)],
|
||||
load_patterns=[
|
||||
PlainLoadPattern(
|
||||
id=1,
|
||||
name="Dead",
|
||||
time_series_id=1,
|
||||
nodal_loads=[NodalLoad(node_id=2, forces=(0.0, -1000.0, 0.0, 0.0, 0.0, 0.0))],
|
||||
),
|
||||
PlainLoadPattern(
|
||||
id=2,
|
||||
name="Live",
|
||||
time_series_id=2,
|
||||
nodal_loads=[NodalLoad(node_id=2, forces=(0.0, -500.0, 0.0, 0.0, 0.0, 0.0))],
|
||||
),
|
||||
],
|
||||
analyses=[
|
||||
StaticCase(id=1, name="Dead", pattern_ids=[1]),
|
||||
StaticCase(id=2, name="Live", pattern_ids=[2]),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def test_superposition_matches_combined_run() -> None:
|
||||
"""1.2*Dead + 1.6*Live via evaluate_combination == single StaticCase run."""
|
||||
proj = _two_pattern_cantilever()
|
||||
dead = OpenSeesRunner(proj).run(proj.analyses[0])
|
||||
live = OpenSeesRunner(proj).run(proj.analyses[1])
|
||||
combo = LoadCombination(
|
||||
id=1,
|
||||
name="1.2D+1.6L",
|
||||
kind="Linear",
|
||||
items=[
|
||||
LoadCombinationItem(case_id=1, factor=1.2),
|
||||
LoadCombinationItem(case_id=2, factor=1.6),
|
||||
],
|
||||
)
|
||||
combined = evaluate_combination({1: dead, 2: live}, combo)
|
||||
assert isinstance(combined, StaticResults)
|
||||
# Single run with both patterns factored identically.
|
||||
direct = OpenSeesRunner(proj).run(
|
||||
StaticCase(id=3, name="direct", pattern_ids=[1, 2], pattern_factors={1: 1.2, 2: 1.6})
|
||||
)
|
||||
assert math.isclose(
|
||||
combined.disp(node_id=2, dof=2), direct.disp(node_id=2, dof=2), rel_tol=1e-9
|
||||
)
|
||||
180
tests/unit/test_combinations.py
Normal file
180
tests/unit/test_combinations.py
Normal file
|
|
@ -0,0 +1,180 @@
|
|||
"""Case-result load combinations — core validation, project wiring, combine math."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from otko.core import (
|
||||
LoadCombination,
|
||||
LoadCombinationItem,
|
||||
Project,
|
||||
StaticCase,
|
||||
)
|
||||
from otko.services.combinations import (
|
||||
combine_static_results,
|
||||
envelope_static_results,
|
||||
evaluate_combination,
|
||||
)
|
||||
from otko.services.results import ModalResults, StaticResults
|
||||
|
||||
|
||||
def _static(
|
||||
case_id: int, disp_val: float, reac_val: float, elem_val: float, n_steps: int = 1
|
||||
) -> StaticResults:
|
||||
return StaticResults(
|
||||
case_id=case_id,
|
||||
case_name=f"case-{case_id}",
|
||||
n_steps=n_steps,
|
||||
node_disp={1: np.full((n_steps, 2), disp_val)},
|
||||
node_reaction={1: np.full((n_steps, 2), reac_val)},
|
||||
element_forces={1: np.full((n_steps, 3), elem_val)},
|
||||
)
|
||||
|
||||
|
||||
class TestEntity:
|
||||
def test_duplicate_case_rejected(self) -> None:
|
||||
with pytest.raises(ValueError, match="duplicate"):
|
||||
LoadCombination(
|
||||
id=1,
|
||||
name="bad",
|
||||
items=[
|
||||
LoadCombinationItem(case_id=1, factor=1.0),
|
||||
LoadCombinationItem(case_id=1, factor=2.0),
|
||||
],
|
||||
)
|
||||
|
||||
def test_non_finite_factor_rejected(self) -> None:
|
||||
with pytest.raises(ValueError, match="finite"):
|
||||
LoadCombination(id=1, items=[LoadCombinationItem(case_id=1, factor=float("inf"))])
|
||||
|
||||
def test_empty_items_rejected(self) -> None:
|
||||
with pytest.raises(ValueError):
|
||||
LoadCombination(id=1, items=[])
|
||||
|
||||
def test_round_trip(self) -> None:
|
||||
combo = LoadCombination(
|
||||
id=3,
|
||||
name="1.2D+1.6L",
|
||||
kind="Linear",
|
||||
items=[
|
||||
LoadCombinationItem(case_id=1, factor=1.2),
|
||||
LoadCombinationItem(case_id=2, factor=1.6),
|
||||
],
|
||||
)
|
||||
assert LoadCombination.model_validate(combo.model_dump()) == combo
|
||||
assert LoadCombination.model_validate_json(combo.model_dump_json()) == combo
|
||||
|
||||
|
||||
class TestProject:
|
||||
def test_next_id_and_lookup(self) -> None:
|
||||
proj = Project()
|
||||
assert proj.next_combination_id() == 1
|
||||
proj.combinations.append(LoadCombination(id=1, items=[LoadCombinationItem(case_id=1)]))
|
||||
assert proj.next_combination_id() == 2
|
||||
assert proj.combination(1).id == 1
|
||||
with pytest.raises(KeyError):
|
||||
proj.combination(99)
|
||||
|
||||
def test_duplicate_ids_rejected(self) -> None:
|
||||
with pytest.raises(ValueError, match="Duplicate combination"):
|
||||
Project(
|
||||
combinations=[
|
||||
LoadCombination(id=1, items=[LoadCombinationItem(case_id=1)]),
|
||||
LoadCombination(id=1, items=[LoadCombinationItem(case_id=2)]),
|
||||
]
|
||||
)
|
||||
|
||||
def test_validate_references_missing_case(self) -> None:
|
||||
proj = Project(
|
||||
analyses=[StaticCase(id=1, pattern_ids=[1])],
|
||||
combinations=[LoadCombination(id=1, items=[LoadCombinationItem(case_id=9)])],
|
||||
)
|
||||
with pytest.raises(ValueError, match="missing analysis case 9"):
|
||||
proj.validate_references()
|
||||
|
||||
def test_persistence_round_trip(self, tmp_path) -> None: # type: ignore[no-untyped-def]
|
||||
from otko.services import load_project, save_project
|
||||
|
||||
proj = Project(
|
||||
analyses=[StaticCase(id=1, pattern_ids=[1])],
|
||||
combinations=[
|
||||
LoadCombination(
|
||||
id=1,
|
||||
name="combo",
|
||||
items=[LoadCombinationItem(case_id=1, factor=1.2)],
|
||||
)
|
||||
],
|
||||
)
|
||||
path = save_project(proj, tmp_path / "c.osmodel")
|
||||
assert load_project(path).combinations[0].name == "combo"
|
||||
|
||||
|
||||
class TestCombine:
|
||||
def test_linear_weighted_sum(self) -> None:
|
||||
out = combine_static_results(
|
||||
[_static(1, 1.0, 10.0, 100.0), _static(2, 2.0, 20.0, 200.0)],
|
||||
[1.2, 1.6],
|
||||
case_id=7,
|
||||
case_name="combo",
|
||||
)
|
||||
assert out.case_id == 7
|
||||
assert out.node_disp[1][0, 0] == pytest.approx(1.2 * 1.0 + 1.6 * 2.0)
|
||||
assert out.node_reaction[1][0, 1] == pytest.approx(1.2 * 10.0 + 1.6 * 20.0)
|
||||
assert out.element_forces[1][0, 2] == pytest.approx(1.2 * 100.0 + 1.6 * 200.0)
|
||||
|
||||
def test_step_mismatch_rejected(self) -> None:
|
||||
with pytest.raises(ValueError, match="Step-count"):
|
||||
combine_static_results(
|
||||
[_static(1, 1.0, 1.0, 1.0, n_steps=1), _static(2, 1.0, 1.0, 1.0, n_steps=2)],
|
||||
[1.0, 1.0],
|
||||
case_id=1,
|
||||
case_name="x",
|
||||
)
|
||||
|
||||
def test_length_mismatch_rejected(self) -> None:
|
||||
with pytest.raises(ValueError, match="length mismatch"):
|
||||
combine_static_results([_static(1, 1, 1, 1)], [1.0, 2.0], case_id=1, case_name="x")
|
||||
|
||||
|
||||
class TestEnvelope:
|
||||
def test_max_min(self) -> None:
|
||||
out = envelope_static_results(
|
||||
[_static(1, 1.0, 10.0, 5.0), _static(2, -4.0, -20.0, 7.0)],
|
||||
[1.0, 1.0],
|
||||
case_id=3,
|
||||
case_name="env",
|
||||
)
|
||||
assert out.node_disp_max[1][0, 0] == pytest.approx(1.0)
|
||||
assert out.node_disp_min[1][0, 0] == pytest.approx(-4.0)
|
||||
assert out.node_reaction_max[1][0, 0] == pytest.approx(10.0)
|
||||
assert out.node_reaction_min[1][0, 0] == pytest.approx(-20.0)
|
||||
assert out.element_forces_max[1][0, 0] == pytest.approx(7.0)
|
||||
assert out.element_forces_min[1][0, 0] == pytest.approx(5.0)
|
||||
|
||||
|
||||
class TestEvaluate:
|
||||
def test_linear_dispatch(self) -> None:
|
||||
combo = LoadCombination(
|
||||
id=5,
|
||||
name="c",
|
||||
kind="Linear",
|
||||
items=[
|
||||
LoadCombinationItem(case_id=1, factor=2.0),
|
||||
LoadCombinationItem(case_id=2, factor=3.0),
|
||||
],
|
||||
)
|
||||
out = evaluate_combination({1: _static(1, 1, 1, 1), 2: _static(2, 1, 1, 1)}, combo)
|
||||
assert isinstance(out, StaticResults)
|
||||
assert out.node_disp[1][0, 0] == pytest.approx(5.0)
|
||||
|
||||
def test_missing_results_keyerror(self) -> None:
|
||||
combo = LoadCombination(id=1, items=[LoadCombinationItem(case_id=2)])
|
||||
with pytest.raises(KeyError, match="no results for case 2"):
|
||||
evaluate_combination({1: _static(1, 1, 1, 1)}, combo)
|
||||
|
||||
def test_non_static_typeerror(self) -> None:
|
||||
combo = LoadCombination(id=1, items=[LoadCombinationItem(case_id=1)])
|
||||
modal = ModalResults(case_id=1, case_name="m", eigenvalues=np.array([1.0]), mode_shapes={})
|
||||
with pytest.raises(TypeError, match="not StaticResults"):
|
||||
evaluate_combination({1: modal}, combo)
|
||||
Loading…
Reference in a new issue