test: relocate service-layer tests to tests/services/
Mirrors the source layout (core/services split) so unit/ holds core-only tests and services get their own home.
This commit is contained in:
parent
dd9fa26b52
commit
7093164f75
9 changed files with 144 additions and 79 deletions
1
tests/services/__init__.py
Normal file
1
tests/services/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""Service-layer tests: runner, persistence, export, results I/O."""
|
||||
197
tests/services/test_element_forces.py
Normal file
197
tests/services/test_element_forces.py
Normal file
|
|
@ -0,0 +1,197 @@
|
|||
"""Unit tests for element_forces service."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from otko.core import (
|
||||
ElasticBeamColumn,
|
||||
ElasticSection,
|
||||
Node,
|
||||
Project,
|
||||
)
|
||||
from otko.services.element_forces import (
|
||||
DiagramData,
|
||||
ForceComponent,
|
||||
auto_scale,
|
||||
extract_diagram_data,
|
||||
)
|
||||
from otko.services.results import StaticResults
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# Fixtures: a tiny 3D project with 2 elements + canned force results.
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def project_3d() -> Project:
|
||||
return Project(
|
||||
nodes=[
|
||||
Node(id=1, coords=(0.0, 0.0, 0.0)),
|
||||
Node(id=2, coords=(3.0, 0.0, 0.0)),
|
||||
Node(id=3, coords=(6.0, 0.0, 0.0)),
|
||||
],
|
||||
sections=[ElasticSection(id=1, E=200e9, A=0.01, Iz=1e-5, Iy=1e-5)],
|
||||
elements=[
|
||||
ElasticBeamColumn(id=10, nodes=(1, 2), section_id=1),
|
||||
ElasticBeamColumn(id=20, nodes=(2, 3), section_id=1),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def static_3d_results() -> StaticResults:
|
||||
"""One step of analysis. Element 10 carries [N=100, …]; element 20
|
||||
carries [N=-50, …] (compression). 12 components, single time step."""
|
||||
forces_10 = np.array([[100.0, 5.0, 7.0, 1.0, 8.0, 9.0, -100.0, -5.0, -7.0, -1.0, -8.0, -9.0]])
|
||||
forces_20 = np.array([[-50.0, 0.0, 0.0, 0.0, 0.0, 0.0, 50.0, 0.0, 0.0, 0.0, 0.0, 0.0]])
|
||||
return StaticResults(
|
||||
case_id=1,
|
||||
case_name="tst",
|
||||
n_steps=1,
|
||||
element_forces={10: forces_10, 20: forces_20},
|
||||
)
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# extract_diagram_data
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_extract_axial_yields_correct_per_end_values(
|
||||
project_3d: Project,
|
||||
static_3d_results: StaticResults,
|
||||
) -> None:
|
||||
data = extract_diagram_data(project_3d, static_3d_results, ForceComponent.N)
|
||||
assert data.component is ForceComponent.N
|
||||
# Both elements participate; preserve project order (10 first, then 20).
|
||||
assert list(data.element_ids) == [10, 20]
|
||||
# Sign-flipped j-end so diagrams show continuous internal forces:
|
||||
# raw end-i = [100, -50], raw end-j = [-100, 50]
|
||||
# flipped end-j = [100, -50] → internal force same along each element.
|
||||
np.testing.assert_array_equal(data.values_i, [100.0, -50.0])
|
||||
np.testing.assert_array_equal(data.values_j, [100.0, -50.0])
|
||||
assert data.abs_max == 100.0
|
||||
|
||||
|
||||
def test_extract_skips_elements_without_force_data(
|
||||
project_3d: Project,
|
||||
) -> None:
|
||||
results = StaticResults(case_id=1, case_name="x", n_steps=1, element_forces={})
|
||||
data = extract_diagram_data(project_3d, results, ForceComponent.N)
|
||||
assert data.element_ids.size == 0
|
||||
assert data.abs_max == 0.0
|
||||
|
||||
|
||||
def test_extract_handles_2d_force_vectors(project_3d: Project) -> None:
|
||||
"""2D OpenSees beams return 6 components: [N, Vy, Mz] × 2 ends.
|
||||
Asking for V3 / M2 in this case should yield no rows for that element."""
|
||||
forces_10 = np.array([[100.0, 5.0, 9.0, -100.0, -5.0, -9.0]])
|
||||
results = StaticResults(case_id=1, case_name="2d", n_steps=1, element_forces={10: forces_10})
|
||||
n_data = extract_diagram_data(project_3d, results, ForceComponent.N)
|
||||
np.testing.assert_array_equal(n_data.values_i, [100.0])
|
||||
v3_data = extract_diagram_data(project_3d, results, ForceComponent.V3)
|
||||
assert v3_data.element_ids.size == 0 # V3 not available in 2D output
|
||||
|
||||
|
||||
def test_extract_uses_specified_step(project_3d: Project) -> None:
|
||||
"""Multi-step pushover: pick the first step explicitly."""
|
||||
f = np.zeros((3, 12))
|
||||
f[0, 0] = 10.0 # step 0 axial @ node i
|
||||
f[1, 0] = 20.0
|
||||
f[2, 0] = 30.0
|
||||
results = StaticResults(case_id=1, case_name="push", n_steps=3, element_forces={10: f})
|
||||
data = extract_diagram_data(project_3d, results, ForceComponent.N, step=0)
|
||||
np.testing.assert_array_equal(data.values_i, [10.0])
|
||||
data = extract_diagram_data(project_3d, results, ForceComponent.N, step=2)
|
||||
np.testing.assert_array_equal(data.values_i, [30.0])
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# auto_scale
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_auto_scale_scales_to_target_fraction(
|
||||
project_3d: Project,
|
||||
static_3d_results: StaticResults,
|
||||
) -> None:
|
||||
data = extract_diagram_data(project_3d, static_3d_results, ForceComponent.N)
|
||||
# Bounding-box diagonal of the 3-node line is 6.0 along X.
|
||||
# target_fraction defaults to 0.08 → max diagram height = 0.48.
|
||||
# abs_max=100 → expected scale ≈ 0.0048.
|
||||
scale = auto_scale(project_3d, data)
|
||||
assert scale == pytest.approx((6.0 * 0.08) / 100.0, rel=1e-9)
|
||||
|
||||
|
||||
def test_auto_scale_zero_force_returns_unity(project_3d: Project) -> None:
|
||||
empty = DiagramData(
|
||||
component=ForceComponent.N,
|
||||
element_ids=np.empty(0, dtype=int),
|
||||
values_i=np.empty(0),
|
||||
values_j=np.empty(0),
|
||||
abs_max=0.0,
|
||||
)
|
||||
assert auto_scale(project_3d, empty) == 1.0
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# Truss-specific force extraction (separate localForce layout)
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
@pytest.fixture
|
||||
def truss_project_2d() -> Project:
|
||||
from otko.core import ElasticUniaxial, TrussElement
|
||||
|
||||
return Project(
|
||||
ndm=2,
|
||||
ndf=2,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0.0, 0.0, 0.0)),
|
||||
Node(id=2, coords=(3.0, 0.0, 0.0)),
|
||||
],
|
||||
materials=[ElasticUniaxial(id=1, E=200e9)],
|
||||
elements=[
|
||||
TrussElement(id=1, nodes=(1, 2), area=1e-3, material_id=1),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def test_truss_axial_uses_correct_index_map(truss_project_2d: Project) -> None:
|
||||
"""2D truss localForce is [N_i, 0, N_j, 0] — N_j is at index 2, not 3.
|
||||
|
||||
Before the fix, extract_diagram_data assumed the frame layout where
|
||||
index 3 means N_j. Truss's index 3 is the zero shear-y at end j, so
|
||||
every diagram value_j came back as 0 and the force diagram drew
|
||||
nothing meaningful for trusses.
|
||||
"""
|
||||
# 2D truss with tension N = +1500 N → local force vector [-N, 0, +N, 0]
|
||||
# (equilibrium signs: end-i points out, end-j points in).
|
||||
forces = np.array([[-1500.0, 0.0, 1500.0, 0.0]])
|
||||
results = StaticResults(
|
||||
case_id=1,
|
||||
case_name="tst",
|
||||
n_steps=1,
|
||||
element_forces={1: forces},
|
||||
)
|
||||
data = extract_diagram_data(truss_project_2d, results, ForceComponent.N)
|
||||
assert data.element_ids.tolist() == [1]
|
||||
assert data.values_i[0] == pytest.approx(-1500.0)
|
||||
# values_j is sign-flipped (equilibrium convention) so the diagram
|
||||
# draws a single-sign line along the bar.
|
||||
assert data.values_j[0] == pytest.approx(-1500.0)
|
||||
|
||||
|
||||
def test_truss_has_no_shear_or_moment_components(truss_project_2d: Project) -> None:
|
||||
"""Requesting V2 / V3 / M3 on a truss returns an empty diagram."""
|
||||
forces = np.array([[100.0, 0.0, -100.0, 0.0]])
|
||||
results = StaticResults(
|
||||
case_id=1,
|
||||
case_name="tst",
|
||||
n_steps=1,
|
||||
element_forces={1: forces},
|
||||
)
|
||||
for comp in (ForceComponent.V2, ForceComponent.V3, ForceComponent.M3):
|
||||
data = extract_diagram_data(truss_project_2d, results, comp)
|
||||
assert data.element_ids.size == 0, f"{comp.value} should yield no data"
|
||||
240
tests/services/test_export.py
Normal file
240
tests/services/test_export.py
Normal file
|
|
@ -0,0 +1,240 @@
|
|||
"""Unit tests for the script exporter (services.export).
|
||||
|
||||
The exporter drives OpenSeesRunner with a RecordingOps stand-in and
|
||||
renders the captured calls as OpenSeesPy / Tcl. No solver, no Qt.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from otko import __version__
|
||||
from otko.core import (
|
||||
ElasticBeamColumn,
|
||||
ElasticIsotropic,
|
||||
ElasticSection,
|
||||
LinearTimeSeries,
|
||||
NodalLoad,
|
||||
Node,
|
||||
PathTimeSeries,
|
||||
PlainLoadPattern,
|
||||
Project,
|
||||
StaticCase,
|
||||
Steel01,
|
||||
TrussElement,
|
||||
)
|
||||
from otko.services.export import (
|
||||
PINNED_OPENSEESPY,
|
||||
RecordingOps,
|
||||
_render_opspy,
|
||||
_render_tcl,
|
||||
export_opspy,
|
||||
export_tcl,
|
||||
)
|
||||
|
||||
|
||||
# ───────────────────────── fixtures ─────────────────────────
|
||||
def _truss_2d() -> Project:
|
||||
"""2D truss + Linear pattern + StaticCase #1."""
|
||||
return Project(
|
||||
ndm=2,
|
||||
ndf=2,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True, True, False, False, False, False)),
|
||||
Node(id=2, coords=(4, 0, 0), restraint=(False, True, False, False, False, False)),
|
||||
Node(id=3, coords=(2, 3, 0)),
|
||||
],
|
||||
materials=[Steel01(id=1, Fy=420e6, E0=200e9, b=0.01)],
|
||||
elements=[
|
||||
TrussElement(id=1, nodes=(1, 3), area=1e-3, material_id=1),
|
||||
TrussElement(id=2, nodes=(2, 3), area=1e-3, material_id=1),
|
||||
],
|
||||
time_series=[LinearTimeSeries(id=1)],
|
||||
load_patterns=[
|
||||
PlainLoadPattern(
|
||||
id=1,
|
||||
time_series_id=1,
|
||||
nodal_loads=[NodalLoad(node_id=3, forces=(0, -1000, 0, 0, 0, 0))],
|
||||
)
|
||||
],
|
||||
analyses=[StaticCase(id=1, name="push", pattern_ids=[1])],
|
||||
)
|
||||
|
||||
|
||||
def _frame_3d() -> Project:
|
||||
"""3D cantilever column (exercises geomTransf tags)."""
|
||||
return Project(
|
||||
ndm=3,
|
||||
ndf=6,
|
||||
nodes=[
|
||||
Node(
|
||||
id=1,
|
||||
coords=(0, 0, 0),
|
||||
restraint=(True, True, True, True, True, True),
|
||||
),
|
||||
Node(id=2, coords=(0, 0, 3)),
|
||||
],
|
||||
materials=[ElasticIsotropic(id=1, E=200e9, nu=0.3)],
|
||||
sections=[ElasticSection(id=1, E=200e9, A=0.01, Iz=1e-4, Iy=1e-4, G=80e9, J=1e-6)],
|
||||
elements=[ElasticBeamColumn(id=1, nodes=(1, 2), section_id=1)],
|
||||
)
|
||||
|
||||
|
||||
def _path_series_project() -> Project:
|
||||
"""Single truss + PathTimeSeries(dt) + StaticCase."""
|
||||
return Project(
|
||||
ndm=2,
|
||||
ndf=2,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True, True, False, False, False, False)),
|
||||
Node(id=2, coords=(1, 0, 0)),
|
||||
],
|
||||
materials=[Steel01(id=1, Fy=250e6, E0=200e9, b=0.01)],
|
||||
elements=[TrussElement(id=1, nodes=(1, 2), area=1e-3, material_id=1)],
|
||||
time_series=[PathTimeSeries(id=7, dt=0.02, values=[0.0, 1.0, 0.5], factor=2.0)],
|
||||
load_patterns=[
|
||||
PlainLoadPattern(
|
||||
id=1,
|
||||
time_series_id=7,
|
||||
nodal_loads=[NodalLoad(node_id=2, forces=(100, 0, 0, 0, 0, 0))],
|
||||
)
|
||||
],
|
||||
analyses=[StaticCase(id=1, name="transient-prep", pattern_ids=[1])],
|
||||
)
|
||||
|
||||
|
||||
# ───────────────────────── header ─────────────────────────
|
||||
class TestHeader:
|
||||
def test_opspy_header_pins_solver_version_and_units(self) -> None:
|
||||
script = export_opspy(_truss_2d())
|
||||
assert PINNED_OPENSEESPY == "openseespy==3.8.0.0"
|
||||
assert "openseespy==3.8.0.0" in script
|
||||
assert __version__ in script
|
||||
assert _truss_2d().meta.units.value in script
|
||||
|
||||
def test_tcl_header_pins_solver_version_and_units(self) -> None:
|
||||
script = export_tcl(_truss_2d())
|
||||
assert "openseespy==3.8.0.0" in script
|
||||
assert __version__ in script
|
||||
|
||||
|
||||
# ───────────────────────── model-only ─────────────────────────
|
||||
class TestModelOnly:
|
||||
def test_opspy_model_only_commands(self) -> None:
|
||||
script = export_opspy(_truss_2d())
|
||||
assert "import openseespy.opensees as ops" in script
|
||||
assert "ops.wipe()" in script
|
||||
assert script.index("ops.wipe()") < script.index("ops.model(")
|
||||
assert 'ops.model("basic", "-ndm", 2, "-ndf", 2)' in script
|
||||
assert "ops.node(1, 0.0, 0.0)" in script
|
||||
assert "ops.fix(1, 1, 1)" in script
|
||||
assert 'ops.uniaxialMaterial("Steel01", 1,' in script
|
||||
assert 'ops.element("truss", 1,' in script
|
||||
# Model-only scope: no patterns, no analysis.
|
||||
assert "ops.pattern(" not in script
|
||||
assert "ops.analyze(" not in script
|
||||
assert "model only" in script
|
||||
|
||||
def test_tcl_model_only_commands(self) -> None:
|
||||
script = export_tcl(_truss_2d())
|
||||
lines = script.splitlines()
|
||||
assert "wipe" in lines
|
||||
assert "model basic -ndm 2 -ndf 2" in lines
|
||||
assert "node 1 0.0 0.0" in lines
|
||||
assert "fix 1 1 1" in lines
|
||||
assert any(line.startswith("uniaxialMaterial Steel01 1 ") for line in lines)
|
||||
assert any(line.startswith("element truss 1 ") for line in lines)
|
||||
assert not any(line.startswith("pattern ") for line in lines)
|
||||
|
||||
def test_no_query_commands_leak(self) -> None:
|
||||
script = export_opspy(_truss_2d(), case_id=1)
|
||||
for banned in ("nodeDisp", "nodeReaction", "eleResponse", "eleForce", "reactions"):
|
||||
assert banned not in script
|
||||
|
||||
|
||||
# ───────────────────────── with case ─────────────────────────
|
||||
class TestWithCase:
|
||||
def test_opspy_static_case_appends_analysis(self) -> None:
|
||||
script = export_opspy(_truss_2d(), case_id=1)
|
||||
assert 'ops.timeSeries("Linear", 1, "-factor", 1.0)' in script
|
||||
assert 'ops.pattern("Plain", 1, 1)' in script
|
||||
assert "ops.load(3, 0.0, -1000.0)" in script
|
||||
assert "ops.system(" in script
|
||||
assert "ops.analyze(1)" in script
|
||||
assert "Static case #1" in script
|
||||
|
||||
def test_tcl_static_case_appends_analysis(self) -> None:
|
||||
script = export_tcl(_truss_2d(), case_id=1)
|
||||
assert "timeSeries Linear 1 -factor 1.0" in script
|
||||
assert "pattern Plain 1 1" in script
|
||||
assert "load 3 0.0 -1000.0" in script
|
||||
assert "analyze 1" in script
|
||||
|
||||
def test_unknown_case_id_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="999"):
|
||||
export_opspy(_truss_2d(), case_id=999)
|
||||
with pytest.raises(ValueError, match="999"):
|
||||
export_tcl(_truss_2d(), case_id=999)
|
||||
|
||||
|
||||
# ───────────────────────── gotchas ─────────────────────────
|
||||
class TestGotchas:
|
||||
def test_path_timeseries_dt_preserved(self) -> None:
|
||||
py = export_opspy(_path_series_project(), case_id=1)
|
||||
tcl = export_tcl(_path_series_project(), case_id=1)
|
||||
assert '"Path", 7, "-dt", 0.02' in py
|
||||
assert "timeSeries Path 7 -dt 0.02" in tcl
|
||||
|
||||
def test_geomtransf_tags_preserved(self) -> None:
|
||||
py = export_opspy(_frame_3d())
|
||||
tcl = export_tcl(_frame_3d())
|
||||
assert 'ops.geomTransf("Linear", 1, 1.0, 0.0, 0.0)' in py
|
||||
assert 'ops.element("elasticBeamColumn", 1, 1, 2, 1, 1,' in py
|
||||
assert "geomTransf Linear 1 1.0 0.0 0.0" in tcl
|
||||
assert "element elasticBeamColumn 1 1 2 1 1" in tcl
|
||||
|
||||
def test_fix_uses_dof_subset(self) -> None:
|
||||
"""2D-frame (ndf=3) fix emits exactly (Ux, Uy, Rz)."""
|
||||
proj = Project(
|
||||
ndm=2,
|
||||
ndf=3,
|
||||
nodes=[
|
||||
Node(
|
||||
id=1,
|
||||
coords=(0, 0, 0),
|
||||
restraint=(True, True, False, False, False, True),
|
||||
),
|
||||
Node(id=2, coords=(0, 3, 0)),
|
||||
],
|
||||
sections=[ElasticSection(id=1, E=200e9, A=0.01, Iz=1e-4)],
|
||||
elements=[ElasticBeamColumn(id=1, nodes=(1, 2), section_id=1)],
|
||||
)
|
||||
assert "ops.fix(1, 1, 1, 1)" in export_opspy(proj)
|
||||
assert "fix 1 1 1 1" in export_tcl(proj).splitlines()
|
||||
|
||||
|
||||
# ───────────────────────── unsupported ─────────────────────────
|
||||
class TestUnsupported:
|
||||
def test_unknown_op_raises_with_marker(self) -> None:
|
||||
rec = RecordingOps()
|
||||
rec.someFutureCommand(1, 2)
|
||||
with pytest.raises(NotImplementedError, match="UNSUPPORTED"):
|
||||
_render_opspy(rec.calls, _truss_2d(), None)
|
||||
with pytest.raises(NotImplementedError, match="UNSUPPORTED"):
|
||||
_render_tcl(rec.calls, _truss_2d(), None)
|
||||
|
||||
def test_no_top_level_openseespy_import(self) -> None:
|
||||
"""services.export and the touched views must not import openseespy."""
|
||||
import re
|
||||
|
||||
pattern = re.compile(r"^\s*(import|from)\s+openseespy", re.MULTILINE)
|
||||
root = Path(__file__).resolve().parents[2]
|
||||
for rel in (
|
||||
"src/otko/services/export.py",
|
||||
"src/otko/views/menu_builder.py",
|
||||
"src/otko/views/action_handlers.py",
|
||||
):
|
||||
text = (root / rel).read_text(encoding="utf-8")
|
||||
assert pattern.search(text) is None, rel
|
||||
193
tests/services/test_imposed_support_motion.py
Normal file
193
tests/services/test_imposed_support_motion.py
Normal file
|
|
@ -0,0 +1,193 @@
|
|||
"""ImposedSupportMotionPattern — model validation + runner command emission.
|
||||
|
||||
The emission tests inject a ``MagicMock`` as the ops module (the
|
||||
``test_runner_translation.py`` pattern); the physics itself is verified in
|
||||
``tests/integration/test_runner_imposed_motion.py`` against a uniform-
|
||||
excitation reference.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from otko.core import (
|
||||
ElasticUniaxial,
|
||||
ImposedSupportMotionPattern,
|
||||
Node,
|
||||
PathTimeSeries,
|
||||
Project,
|
||||
ZeroLengthElement,
|
||||
)
|
||||
from otko.services.opensees_runner import (
|
||||
_IMPOSED_VEL_TS_OFFSET,
|
||||
OpenSeesRunner,
|
||||
)
|
||||
|
||||
|
||||
def _project(**overrides) -> Project: # type: ignore[no-untyped-def]
|
||||
"""Grounded zeroLength + mass node, disp record on the support's X."""
|
||||
defaults = dict(
|
||||
ndm=3,
|
||||
ndf=6,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True,) * 6),
|
||||
Node(
|
||||
id=2,
|
||||
coords=(0, 0, 0),
|
||||
restraint=(False, True, True, True, True, True),
|
||||
mass=(1.0, 0.0, 0.0, 0.0, 0.0, 0.0),
|
||||
),
|
||||
],
|
||||
materials=[ElasticUniaxial(id=1, E=100.0)],
|
||||
elements=[
|
||||
ZeroLengthElement(
|
||||
id=1,
|
||||
nodes=(1, 2),
|
||||
material_ids=(1,),
|
||||
dofs=(1,),
|
||||
)
|
||||
],
|
||||
time_series=[
|
||||
PathTimeSeries(id=1, dt=0.1, values=[0.0, 1.0, 4.0, 9.0, 16.0]),
|
||||
],
|
||||
load_patterns=[
|
||||
ImposedSupportMotionPattern(
|
||||
id=7,
|
||||
direction=1,
|
||||
disp_series_id=1,
|
||||
node_ids=[1],
|
||||
)
|
||||
],
|
||||
)
|
||||
defaults.update(overrides)
|
||||
return Project(**defaults)
|
||||
|
||||
|
||||
# ───────────────────────── model / references ─────────────────────────
|
||||
def test_round_trip_preserves_pattern() -> None:
|
||||
project = _project()
|
||||
clone = Project.model_validate(project.model_dump())
|
||||
pat = clone.load_patterns[0]
|
||||
assert isinstance(pat, ImposedSupportMotionPattern)
|
||||
assert pat.direction == 1
|
||||
assert pat.disp_series_id == 1
|
||||
assert pat.node_ids == [1]
|
||||
|
||||
|
||||
def test_validate_references_missing_series_and_node() -> None:
|
||||
project = _project()
|
||||
project.load_patterns[0].disp_series_id = 99
|
||||
with pytest.raises(ValueError, match="missing time series 99"):
|
||||
project.validate_references()
|
||||
|
||||
project = _project()
|
||||
project.load_patterns[0].node_ids = [42]
|
||||
with pytest.raises(ValueError, match="drives missing node 42"):
|
||||
project.validate_references()
|
||||
|
||||
|
||||
# ───────────────────────── emission ─────────────────────────
|
||||
def _emit(project: Project) -> MagicMock:
|
||||
ops = MagicMock()
|
||||
runner = OpenSeesRunner(project, ops_module=ops)
|
||||
runner._emit_patterns_for_case([7])
|
||||
return ops
|
||||
|
||||
|
||||
def test_emission_sequence() -> None:
|
||||
ops = _emit(_project())
|
||||
|
||||
# Derived velocity series: central differences of the disp record.
|
||||
ts_calls = ops.timeSeries.call_args_list
|
||||
assert ts_calls[0].args[:2] == ("Path", 1) # the disp record itself
|
||||
kind, tag, flag_dt, dt, flag_vals, *rest = ts_calls[1].args
|
||||
assert (kind, tag, flag_dt, dt, flag_vals) == (
|
||||
"Path",
|
||||
_IMPOSED_VEL_TS_OFFSET + 7,
|
||||
"-dt",
|
||||
0.1,
|
||||
"-values",
|
||||
)
|
||||
vel = rest[: rest.index("-factor")]
|
||||
expected = np.gradient(np.array([0.0, 1.0, 4.0, 9.0, 16.0]), 0.1)
|
||||
assert np.allclose(vel, expected)
|
||||
|
||||
# The X fix is swapped for the imposed motion, then the pattern lands.
|
||||
ops.remove.assert_called_once_with("sp", 1, 1)
|
||||
ops.pattern.assert_called_once_with("MultipleSupport", 7)
|
||||
ops.groundMotion.assert_called_once_with(
|
||||
7,
|
||||
"Plain",
|
||||
"-disp",
|
||||
1,
|
||||
"-vel",
|
||||
_IMPOSED_VEL_TS_OFFSET + 7,
|
||||
"-fact",
|
||||
1.0,
|
||||
)
|
||||
ops.imposedMotion.assert_called_once_with(1, 1, 7)
|
||||
|
||||
|
||||
def test_emission_order_fix_removed_before_pattern() -> None:
|
||||
ops = _emit(_project())
|
||||
names = [c[0] for c in ops.method_calls]
|
||||
assert names.index("remove") < names.index("pattern")
|
||||
assert names.index("pattern") < names.index("groundMotion")
|
||||
assert names.index("groundMotion") < names.index("imposedMotion")
|
||||
|
||||
|
||||
def test_unrestrained_node_rejected() -> None:
|
||||
project = _project()
|
||||
project.nodes[0].restraint = (False, True, True, True, True, True)
|
||||
with pytest.raises(ValueError, match="must be restrained in direction 1"):
|
||||
_emit(project)
|
||||
|
||||
|
||||
def test_non_dt_path_series_rejected() -> None:
|
||||
project = _project(
|
||||
time_series=[
|
||||
PathTimeSeries(
|
||||
id=1,
|
||||
times=[0.0, 0.1, 0.3],
|
||||
values=[0.0, 1.0, 4.0],
|
||||
)
|
||||
]
|
||||
)
|
||||
with pytest.raises(ValueError, match="PathTimeSeries with uniform"):
|
||||
_emit(project)
|
||||
|
||||
|
||||
def test_vel_tag_collision_rejected() -> None:
|
||||
project = _project()
|
||||
project.time_series.append(PathTimeSeries(id=_IMPOSED_VEL_TS_OFFSET + 7, dt=0.1, values=[0.0]))
|
||||
with pytest.raises(ValueError, match="collides with an existing time series"):
|
||||
_emit(project)
|
||||
|
||||
|
||||
def test_direction_beyond_ndf_rejected() -> None:
|
||||
project = _project(
|
||||
ndm=2,
|
||||
ndf=2,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True, True, False, False, False, False)),
|
||||
Node(
|
||||
id=2,
|
||||
coords=(0, 0, 0),
|
||||
restraint=(False, True, False, False, False, False),
|
||||
mass=(1.0, 0.0, 0.0, 0.0, 0.0, 0.0),
|
||||
),
|
||||
],
|
||||
load_patterns=[
|
||||
ImposedSupportMotionPattern(
|
||||
id=7,
|
||||
direction=3,
|
||||
disp_series_id=1,
|
||||
node_ids=[1],
|
||||
)
|
||||
],
|
||||
)
|
||||
with pytest.raises(ValueError, match="exceeds ndf=2"):
|
||||
_emit(project)
|
||||
69
tests/services/test_peer_record.py
Normal file
69
tests/services/test_peer_record.py
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
"""PEER record parser tests — NGA + old SMD headers + plain values."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import textwrap
|
||||
|
||||
import pytest
|
||||
|
||||
from otko.services.peer_record import (
|
||||
parse_peer_record,
|
||||
parse_plain_values,
|
||||
)
|
||||
|
||||
|
||||
def _write(tmp_path, text: str, name: str = "rec.at2"): # type: ignore[no-untyped-def]
|
||||
p = tmp_path / name
|
||||
p.write_text(text)
|
||||
return p
|
||||
|
||||
|
||||
def test_parse_nga_format(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||||
"""New PEER NGA header: '3930 0.00500 NPTS, DT' on line 4."""
|
||||
content = textwrap.dedent("""\
|
||||
PACIFIC ENGINEERING AND ANALYSIS STRONG-MOTION DATA
|
||||
IMPERIAL VALLEY 10/15/79 2319, EL CENTRO ARRAY 6, 230
|
||||
ACCELERATION TIME HISTORY IN UNITS OF G
|
||||
5 0.01 NPTS, DT
|
||||
0.001 0.002 0.003
|
||||
0.004 0.005
|
||||
""")
|
||||
path = _write(tmp_path, content)
|
||||
dt, npts, vals = parse_peer_record(path)
|
||||
assert dt == pytest.approx(0.01)
|
||||
assert npts == 5
|
||||
assert vals == pytest.approx([0.001, 0.002, 0.003, 0.004, 0.005])
|
||||
|
||||
|
||||
def test_parse_old_smd_format(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||||
"""Old SMD header: 'NPTS= 3930, DT= .00500 SEC'."""
|
||||
content = textwrap.dedent("""\
|
||||
PACIFIC ENGINEERING
|
||||
IMPERIAL VALLEY
|
||||
ACCELERATION
|
||||
NPTS= 5, DT= .01000 SEC
|
||||
0.1 0.2 0.3
|
||||
0.4 0.5
|
||||
""")
|
||||
path = _write(tmp_path, content)
|
||||
dt, npts, vals = parse_peer_record(path)
|
||||
assert dt == pytest.approx(0.01)
|
||||
assert npts == 5
|
||||
assert vals == pytest.approx([0.1, 0.2, 0.3, 0.4, 0.5])
|
||||
|
||||
|
||||
def test_parse_peer_with_no_header_raises(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||||
path = _write(tmp_path, "0.1 0.2 0.3 0.4 0.5")
|
||||
with pytest.raises(ValueError):
|
||||
parse_peer_record(path)
|
||||
|
||||
|
||||
def test_parse_plain_values(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||||
path = _write(tmp_path, "0.1 0.2\n0.3 0.4\n0.5", "vals.txt")
|
||||
assert parse_plain_values(path) == pytest.approx([0.1, 0.2, 0.3, 0.4, 0.5])
|
||||
|
||||
|
||||
def test_parse_plain_values_rejects_empty(tmp_path) -> None: # type: ignore[no-untyped-def]
|
||||
path = _write(tmp_path, "# just a comment", "empty.txt")
|
||||
with pytest.raises(ValueError):
|
||||
parse_plain_values(path)
|
||||
107
tests/services/test_persistence.py
Normal file
107
tests/services/test_persistence.py
Normal file
|
|
@ -0,0 +1,107 @@
|
|||
"""Unit tests for the persistence service — JSON round-trip."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from otko.core import (
|
||||
Concrete02,
|
||||
ElasticBeamColumn,
|
||||
ElasticSection,
|
||||
LinearTimeSeries,
|
||||
NodalLoad,
|
||||
Node,
|
||||
PlainLoadPattern,
|
||||
Project,
|
||||
ProjectMeta,
|
||||
Steel01,
|
||||
TrussElement,
|
||||
UnitSystem,
|
||||
)
|
||||
from otko.services import PROJECT_FILE_SUFFIX, load_project, save_project
|
||||
|
||||
|
||||
def _sample_project() -> Project:
|
||||
return Project(
|
||||
meta=ProjectMeta(name="Sample", author="Ozan", units=UnitSystem.SI_M_N),
|
||||
ndm=3,
|
||||
ndf=6,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True,) * 6),
|
||||
Node(id=2, coords=(0, 0, 3.0)),
|
||||
],
|
||||
materials=[
|
||||
Steel01(id=1, Fy=420e6, E0=200e9, b=0.01),
|
||||
Concrete02(
|
||||
id=2,
|
||||
fpc=-30e6,
|
||||
epsc0=-0.002,
|
||||
fpcu=-15e6,
|
||||
epsU=-0.005,
|
||||
ft=3e6,
|
||||
Ets=2e9,
|
||||
**{"lambda": 0.1},
|
||||
),
|
||||
],
|
||||
sections=[ElasticSection(id=1, E=200e9, A=0.01, Iz=8.33e-6, Iy=8.33e-6, G=80e9, J=1e-6)],
|
||||
elements=[
|
||||
ElasticBeamColumn(id=1, nodes=(1, 2), section_id=1, geom_transf="PDelta"),
|
||||
TrussElement(id=2, nodes=(1, 2), area=1e-3, material_id=1),
|
||||
],
|
||||
time_series=[LinearTimeSeries(id=1, factor=1.0)],
|
||||
load_patterns=[
|
||||
PlainLoadPattern(
|
||||
id=1,
|
||||
time_series_id=1,
|
||||
nodal_loads=[NodalLoad(node_id=2, forces=(0, 0, -10e3, 0, 0, 0))],
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def test_round_trip_preserves_everything(tmp_path: Path) -> None:
|
||||
original = _sample_project()
|
||||
target = save_project(original, tmp_path / "model")
|
||||
assert target.suffix == PROJECT_FILE_SUFFIX
|
||||
assert target.exists()
|
||||
|
||||
restored = load_project(target)
|
||||
assert restored.model_dump(by_alias=True) == original.model_dump(by_alias=True)
|
||||
|
||||
|
||||
def test_save_appends_suffix_when_missing(tmp_path: Path) -> None:
|
||||
p = Project()
|
||||
out = save_project(p, tmp_path / "no_extension")
|
||||
assert out.suffix == PROJECT_FILE_SUFFIX
|
||||
|
||||
|
||||
def test_save_keeps_existing_suffix(tmp_path: Path) -> None:
|
||||
p = Project()
|
||||
out = save_project(p, tmp_path / "with_ext.osmodel")
|
||||
assert out.name == "with_ext.osmodel"
|
||||
|
||||
|
||||
def test_load_missing_file_raises() -> None:
|
||||
with pytest.raises(FileNotFoundError):
|
||||
load_project("/no/such/path/x.osmodel")
|
||||
|
||||
|
||||
def test_load_corrupt_file_raises_validation_error(tmp_path: Path) -> None:
|
||||
bad = tmp_path / "bad.osmodel"
|
||||
bad.write_text('{"ndm": 2, "ndf": 6}', encoding="utf-8") # invalid combo
|
||||
with pytest.raises(ValidationError):
|
||||
load_project(bad)
|
||||
|
||||
|
||||
def test_polymorphic_collection_dispatches_correctly(tmp_path: Path) -> None:
|
||||
"""Materials list must restore as Steel01 + Concrete02 — not generic Entity."""
|
||||
original = _sample_project()
|
||||
out = save_project(original, tmp_path / "x")
|
||||
restored = load_project(out)
|
||||
assert isinstance(restored.materials[0], Steel01)
|
||||
assert isinstance(restored.materials[1], Concrete02)
|
||||
assert isinstance(restored.elements[0], ElasticBeamColumn)
|
||||
assert isinstance(restored.elements[1], TrussElement)
|
||||
300
tests/services/test_runner_translation.py
Normal file
300
tests/services/test_runner_translation.py
Normal file
|
|
@ -0,0 +1,300 @@
|
|||
"""Translation tests for OpenSeesRunner.
|
||||
|
||||
These tests inject a ``Mock()`` as the ``ops`` module and assert that
|
||||
the runner emits the correct command sequence — no real OpenSees
|
||||
needed. The companion ``test_runner_static.py`` etc. exercise the
|
||||
solver with a real ``ops`` for analytical verification.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, call
|
||||
|
||||
import pytest
|
||||
|
||||
from otko.core import (
|
||||
ElasticBeamColumn,
|
||||
ElasticIsotropic,
|
||||
ElasticSection,
|
||||
EqualDOFConstraint,
|
||||
HystereticSM,
|
||||
LinearTimeSeries,
|
||||
NodalLoad,
|
||||
Node,
|
||||
PlainLoadPattern,
|
||||
Project,
|
||||
StaticCase,
|
||||
Steel01,
|
||||
TrussElement,
|
||||
ZeroLengthElement,
|
||||
)
|
||||
from otko.services.opensees_runner import OpenSeesRunner, _dof_indices
|
||||
|
||||
|
||||
# ───────────────────────── helpers ─────────────────────────
|
||||
def _truss_2d() -> Project:
|
||||
return Project(
|
||||
ndm=2,
|
||||
ndf=2,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True, True, False, False, False, False)),
|
||||
Node(id=2, coords=(4, 0, 0), restraint=(False, True, False, False, False, False)),
|
||||
Node(id=3, coords=(2, 3, 0)),
|
||||
],
|
||||
materials=[Steel01(id=1, Fy=420e6, E0=200e9, b=0.01)],
|
||||
elements=[
|
||||
TrussElement(id=1, nodes=(1, 3), area=1e-3, material_id=1),
|
||||
TrussElement(id=2, nodes=(2, 3), area=1e-3, material_id=1),
|
||||
],
|
||||
time_series=[LinearTimeSeries(id=1)],
|
||||
load_patterns=[
|
||||
PlainLoadPattern(
|
||||
id=1,
|
||||
time_series_id=1,
|
||||
nodal_loads=[NodalLoad(node_id=3, forces=(0, -1000, 0, 0, 0, 0))],
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def _portal_3d() -> Project:
|
||||
return Project(
|
||||
ndm=3,
|
||||
ndf=6,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True,) * 6),
|
||||
Node(id=2, coords=(0, 0, 3)),
|
||||
],
|
||||
materials=[ElasticIsotropic(id=1, E=200e9, nu=0.3)],
|
||||
sections=[ElasticSection(id=1, E=200e9, A=0.01, Iz=1e-4, Iy=1e-4, G=80e9, J=1e-6)],
|
||||
elements=[ElasticBeamColumn(id=1, nodes=(1, 2), section_id=1)],
|
||||
)
|
||||
|
||||
|
||||
def _isolator_3d() -> Project:
|
||||
"""A grounded HystereticSM isolator (zeroLength) under a -cMass frame —
|
||||
the wire-rope benchmark's building block in miniature."""
|
||||
return Project(
|
||||
ndm=3,
|
||||
ndf=6,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True,) * 6), # grounded
|
||||
Node(id=2, coords=(0, 0, 0)), # coincident
|
||||
Node(id=3, coords=(0, 0, 3)),
|
||||
],
|
||||
materials=[
|
||||
HystereticSM(
|
||||
id=1,
|
||||
name="axial",
|
||||
pos_env=[(1.57, 0.00207), (69.1, 0.0399)],
|
||||
neg_env=[(-1.4, -0.00057), (-15.31, -0.0483)],
|
||||
),
|
||||
HystereticSM(id=2, name="shear", pos_env=[(0.12, 0.00067), (9.21, 0.0804)]),
|
||||
],
|
||||
sections=[
|
||||
ElasticSection(
|
||||
id=1, E=2.1e8, A=9.13e-4, Iz=7.373e-7, Iy=7.373e-7, G=8.08e7, J=2.494e-8
|
||||
),
|
||||
],
|
||||
elements=[
|
||||
ZeroLengthElement(id=1, nodes=(1, 2), material_ids=(1, 2), dofs=(3, 1)),
|
||||
ElasticBeamColumn(
|
||||
id=2,
|
||||
nodes=(2, 3),
|
||||
section_id=1,
|
||||
rho=0.00717,
|
||||
consistent_mass=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def _shear_frame_2d() -> Project:
|
||||
return Project(
|
||||
ndm=2,
|
||||
ndf=3,
|
||||
nodes=[
|
||||
Node(id=1, coords=(0, 0, 0), restraint=(True, True, False, False, False, True)),
|
||||
Node(id=2, coords=(4, 0, 0), restraint=(True, True, False, False, False, True)),
|
||||
Node(id=3, coords=(0, 3, 0)),
|
||||
Node(id=4, coords=(4, 3, 0)),
|
||||
],
|
||||
mp_constraints=[EqualDOFConstraint(retained_node=3, constrained_node=4, dofs=(2, 3))],
|
||||
sections=[ElasticSection(id=1, E=200e9, A=0.01, Iz=1e-4)],
|
||||
elements=[ElasticBeamColumn(id=1, nodes=(1, 3), section_id=1)],
|
||||
)
|
||||
|
||||
|
||||
# ───────────────────────── _dof_indices ─────────────────────────
|
||||
class TestDofIndices:
|
||||
@pytest.mark.parametrize(
|
||||
("ndm", "ndf", "expected"),
|
||||
[(2, 2, (0, 1)), (2, 3, (0, 1, 5)), (3, 3, (0, 1, 2)), (3, 6, (0, 1, 2, 3, 4, 5))],
|
||||
)
|
||||
def test_valid_combinations(self, ndm: int, ndf: int, expected: tuple[int, ...]) -> None:
|
||||
assert _dof_indices(ndm, ndf) == expected
|
||||
|
||||
def test_invalid_combination_raises(self) -> None:
|
||||
with pytest.raises(ValueError):
|
||||
_dof_indices(2, 6)
|
||||
|
||||
|
||||
# ───────────────────────── command order ─────────────────────────
|
||||
class TestBuildCommandOrder:
|
||||
def test_wipe_first_then_model(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_truss_2d(), ops_module=ops).build()
|
||||
names = [c[0] for c in ops.method_calls]
|
||||
assert names[0] == "wipe"
|
||||
assert names[1] == "model"
|
||||
|
||||
def test_nodes_before_fixes(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_truss_2d(), ops_module=ops).build()
|
||||
names = [c[0] for c in ops.method_calls]
|
||||
assert names.index("node") < names.index("fix")
|
||||
if "equalDOF" in names:
|
||||
assert names.index("fix") < names.index("equalDOF")
|
||||
|
||||
def test_materials_before_elements(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_truss_2d(), ops_module=ops).build()
|
||||
names = [c[0] for c in ops.method_calls]
|
||||
assert names.index("uniaxialMaterial") < names.index("element")
|
||||
|
||||
def test_geom_transf_before_frame_element(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_portal_3d(), ops_module=ops).build()
|
||||
names = [c[0] for c in ops.method_calls]
|
||||
assert "geomTransf" in names
|
||||
assert names.index("geomTransf") < names.index("element")
|
||||
|
||||
|
||||
# ───────────────────────── per-command emission ─────────────────────────
|
||||
class TestNodeEmission:
|
||||
def test_2d_truss_passes_only_two_coords(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_truss_2d(), ops_module=ops).build()
|
||||
node_calls = [c for c in ops.method_calls if c[0] == "node"]
|
||||
# Each: (1, 0, 0) or (4, 0, 0) or (2, 3, 0); only first two coords go in.
|
||||
assert node_calls[0] == call.node(1, 0.0, 0.0)
|
||||
assert node_calls[1] == call.node(2, 4.0, 0.0)
|
||||
assert node_calls[2] == call.node(3, 2.0, 3.0)
|
||||
|
||||
def test_3d_passes_all_three_coords(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_portal_3d(), ops_module=ops).build()
|
||||
node_calls = [c for c in ops.method_calls if c[0] == "node"]
|
||||
assert node_calls[0] == call.node(1, 0.0, 0.0, 0.0)
|
||||
assert node_calls[1] == call.node(2, 0.0, 0.0, 3.0)
|
||||
|
||||
|
||||
class TestFixEmission:
|
||||
def test_2d_truss_fix_count_matches_ndf(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_truss_2d(), ops_module=ops).build()
|
||||
fix_calls = [c for c in ops.method_calls if c[0] == "fix"]
|
||||
for c in fix_calls:
|
||||
args = c.args
|
||||
# tag + ndf flags
|
||||
assert len(args) == 1 + 2
|
||||
|
||||
def test_3d_pin_passes_six_flags(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_portal_3d(), ops_module=ops).build()
|
||||
fix_calls = [c for c in ops.method_calls if c[0] == "fix"]
|
||||
assert fix_calls[0] == call.fix(1, 1, 1, 1, 1, 1, 1)
|
||||
|
||||
def test_equal_dof_constraint_emits(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_shear_frame_2d(), ops_module=ops).build()
|
||||
ops.equalDOF.assert_called_with(3, 4, 2, 3)
|
||||
|
||||
|
||||
class TestMaterialEmission:
|
||||
def test_steel01_basic(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_truss_2d(), ops_module=ops).build()
|
||||
ops.uniaxialMaterial.assert_any_call("Steel01", 1, 420e6, 200e9, 0.01)
|
||||
|
||||
def test_hystereticsm_pos_and_neg_envelopes(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_isolator_3d(), ops_module=ops).build()
|
||||
# Pairs are flattened in (force, deformation) command order, both envelopes.
|
||||
ops.uniaxialMaterial.assert_any_call(
|
||||
"HystereticSM",
|
||||
1,
|
||||
"-posEnv",
|
||||
1.57,
|
||||
0.00207,
|
||||
69.1,
|
||||
0.0399,
|
||||
"-negEnv",
|
||||
-1.4,
|
||||
-0.00057,
|
||||
-15.31,
|
||||
-0.0483,
|
||||
)
|
||||
|
||||
def test_hystereticsm_symmetric_omits_neg_envelope(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_isolator_3d(), ops_module=ops).build()
|
||||
# No neg_env ⇒ only -posEnv is emitted (OpenSees mirrors it).
|
||||
ops.uniaxialMaterial.assert_any_call(
|
||||
"HystereticSM",
|
||||
2,
|
||||
"-posEnv",
|
||||
0.12,
|
||||
0.00067,
|
||||
9.21,
|
||||
0.0804,
|
||||
)
|
||||
|
||||
|
||||
class TestElementEmission:
|
||||
def test_truss_2d(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_truss_2d(), ops_module=ops).build()
|
||||
ops.element.assert_any_call("truss", 1, 1, 3, 1e-3, 1, "-rho", 0.0)
|
||||
|
||||
def test_elastic_beam_column_uses_allocated_transf_tag(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_portal_3d(), ops_module=ops).build()
|
||||
# geomTransf tag 1 was allocated for "Linear"; element should use it.
|
||||
ops.element.assert_any_call("elasticBeamColumn", 1, 1, 2, 1, 1, "-mass", 0.0)
|
||||
|
||||
def test_elastic_beam_column_cmass_appends_flag(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_isolator_3d(), ops_module=ops).build()
|
||||
# cMass=True ⇒ the consistent-mass flag trails the -mass density.
|
||||
ops.element.assert_any_call("elasticBeamColumn", 2, 2, 3, 1, 1, "-mass", 0.00717, "-cMass")
|
||||
|
||||
def test_zero_length_emits_mat_and_dir_lists(self) -> None:
|
||||
ops = MagicMock()
|
||||
OpenSeesRunner(_isolator_3d(), ops_module=ops).build()
|
||||
ops.element.assert_any_call("zeroLength", 1, 1, 2, "-mat", 1, 2, "-dir", 3, 1)
|
||||
|
||||
|
||||
class TestPatternEmission:
|
||||
def test_plain_pattern_emits_nested_loads(self) -> None:
|
||||
"""The full run path emits patterns; build() does not."""
|
||||
ops = MagicMock()
|
||||
runner = OpenSeesRunner(_truss_2d(), ops_module=ops)
|
||||
runner.build()
|
||||
runner._emit_patterns_for_case([1])
|
||||
ops.timeSeries.assert_called_with("Linear", 1, "-factor", 1.0)
|
||||
ops.pattern.assert_called_with("Plain", 1, 1)
|
||||
# Nodal load on node 3, force vector sliced to (Fx, Fy) for 2D-2DOF
|
||||
ops.load.assert_called_with(3, 0.0, -1000.0)
|
||||
|
||||
|
||||
# ───────────────────────── reference validation ─────────────────────────
|
||||
def test_build_calls_validate_references() -> None:
|
||||
p = _truss_2d()
|
||||
# break a reference
|
||||
p.elements[0].nodes = (1, 99)
|
||||
ops = MagicMock()
|
||||
with pytest.raises(ValueError, match="missing node 99"):
|
||||
OpenSeesRunner(p, ops_module=ops).build()
|
||||
# No commands should have reached ops after validation failed.
|
||||
ops.element.assert_not_called()
|
||||
128
tests/services/test_section_bbox.py
Normal file
128
tests/services/test_section_bbox.py
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
"""Tests for bbox_for_section — used to draw extruded sections in 3D."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
import pytest
|
||||
|
||||
from otko.core import (
|
||||
CircularPatch,
|
||||
ElasticSection,
|
||||
FiberSection,
|
||||
Project,
|
||||
RectangularPatch,
|
||||
StraightLayer,
|
||||
)
|
||||
from otko.services.section_bbox import bbox_for_section
|
||||
|
||||
|
||||
# ── ElasticSection ────────────────────────────────────────────────
|
||||
def test_elastic_section_back_solves_rectangle() -> None:
|
||||
"""An ElasticSection is treated as a rectangle: A = b·h and
|
||||
Iz = b·h³/12 determine (b, h) uniquely."""
|
||||
# Known rectangle: b = 0.3, h = 0.5 → A = 0.15, Iz = 3.125e-3
|
||||
b, h = 0.30, 0.50
|
||||
A = b * h
|
||||
Iz = b * h**3 / 12.0
|
||||
Iy = h * b**3 / 12.0
|
||||
sec = ElasticSection(
|
||||
id=1,
|
||||
name="R",
|
||||
E=200e9,
|
||||
A=A,
|
||||
Iz=Iz,
|
||||
Iy=Iy,
|
||||
G=80e9,
|
||||
J=1e-6,
|
||||
)
|
||||
dims = bbox_for_section(sec)
|
||||
assert dims is not None
|
||||
w_y, h_z = dims
|
||||
assert w_y == pytest.approx(b, rel=1e-9)
|
||||
assert h_z == pytest.approx(h, rel=1e-9)
|
||||
|
||||
|
||||
# ── FiberSection ──────────────────────────────────────────────────
|
||||
def test_fiber_section_bbox_from_rect_patch() -> None:
|
||||
sec = FiberSection(
|
||||
id=1,
|
||||
patches=[
|
||||
RectangularPatch(
|
||||
material_id=1,
|
||||
n_fib_y=4,
|
||||
n_fib_z=4,
|
||||
y_i=-0.15,
|
||||
z_i=-0.25,
|
||||
y_j=0.15,
|
||||
z_j=0.25,
|
||||
)
|
||||
],
|
||||
)
|
||||
dims = bbox_for_section(sec)
|
||||
assert dims is not None
|
||||
w_y, h_z = dims
|
||||
assert w_y == pytest.approx(0.30)
|
||||
assert h_z == pytest.approx(0.50)
|
||||
|
||||
|
||||
def test_fiber_section_bbox_includes_circ_patches() -> None:
|
||||
sec = FiberSection(
|
||||
id=1,
|
||||
patches=[
|
||||
CircularPatch(
|
||||
material_id=1,
|
||||
n_fib_circ=8,
|
||||
n_fib_rad=2,
|
||||
y_center=0.0,
|
||||
z_center=0.0,
|
||||
r_outer=0.2,
|
||||
)
|
||||
],
|
||||
)
|
||||
dims = bbox_for_section(sec)
|
||||
assert dims is not None
|
||||
assert dims[0] == pytest.approx(0.40)
|
||||
assert dims[1] == pytest.approx(0.40)
|
||||
|
||||
|
||||
def test_fiber_section_bbox_grows_for_layers() -> None:
|
||||
"""A bar layer that extends past the patches must widen the bbox."""
|
||||
sec = FiberSection(
|
||||
id=1,
|
||||
patches=[
|
||||
RectangularPatch(
|
||||
material_id=1,
|
||||
n_fib_y=2,
|
||||
n_fib_z=2,
|
||||
y_i=-0.10,
|
||||
z_i=-0.10,
|
||||
y_j=0.10,
|
||||
z_j=0.10,
|
||||
)
|
||||
],
|
||||
layers=[
|
||||
StraightLayer(
|
||||
material_id=2,
|
||||
n_bars=3,
|
||||
bar_area=1e-4,
|
||||
y_start=-0.15,
|
||||
z_start=0.12,
|
||||
y_end=0.15,
|
||||
z_end=0.12,
|
||||
)
|
||||
],
|
||||
)
|
||||
dims = bbox_for_section(sec)
|
||||
assert dims is not None
|
||||
# y extents: layer [-0.15, 0.15] (wider than patch [-0.10, 0.10]) → 0.30
|
||||
# z extents: patch [-0.10, 0.10] + layer z=0.12 → [-0.10, 0.12] → 0.22
|
||||
# plus the bar radius sqrt(bar_area / pi) on each rebar side.
|
||||
r = math.sqrt(1e-4 / math.pi)
|
||||
assert dims[0] == pytest.approx(0.30 + 2 * r)
|
||||
assert dims[1] == pytest.approx(0.22 + r)
|
||||
|
||||
|
||||
def test_empty_fiber_section_returns_none() -> None:
|
||||
sec = FiberSection(id=1)
|
||||
assert bbox_for_section(sec) is None
|
||||
108
tests/services/test_transient_and_export.py
Normal file
108
tests/services/test_transient_and_export.py
Normal file
|
|
@ -0,0 +1,108 @@
|
|||
"""Tests for new TransientResults accessors and animation_export service."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from otko.services.results import TransientResults
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_transient_h5(tmp_path: Path) -> Path:
|
||||
"""Synthetic 5-step transient with disp/vel/accel for nodes 1, 2."""
|
||||
import h5py
|
||||
|
||||
h5_path = tmp_path / "fake_case.h5"
|
||||
n_steps = 5
|
||||
ndf = 6
|
||||
with h5py.File(h5_path, "w") as f:
|
||||
f.create_dataset("time", data=np.linspace(0.0, 0.04, n_steps))
|
||||
for nid in (1, 2):
|
||||
base = nid * 10.0
|
||||
f.create_dataset(f"nodes/{nid}/disp", data=np.full((n_steps, ndf), base))
|
||||
f.create_dataset(f"nodes/{nid}/vel", data=np.full((n_steps, ndf), base + 0.1))
|
||||
f.create_dataset(f"nodes/{nid}/accel", data=np.full((n_steps, ndf), base + 0.2))
|
||||
f.create_dataset("elements/100/forces", data=np.full((n_steps, 12), 5.0))
|
||||
return h5_path
|
||||
|
||||
|
||||
def test_node_disp_vel_accel_history_round_trip(fake_transient_h5: Path) -> None:
|
||||
r = TransientResults(case_id=1, case_name="t", h5_path=fake_transient_h5, n_steps=5, dt=0.01)
|
||||
np.testing.assert_array_equal(r.node_disp_history(1), np.full((5, 6), 10.0))
|
||||
np.testing.assert_array_equal(r.node_vel_history(1), np.full((5, 6), 10.1))
|
||||
np.testing.assert_array_equal(r.node_accel_history(1), np.full((5, 6), 10.2))
|
||||
np.testing.assert_array_equal(r.node_disp_history(2), np.full((5, 6), 20.0))
|
||||
|
||||
|
||||
def test_missing_history_raises_keyerror(tmp_path: Path) -> None:
|
||||
"""Older runs (pre-Phase-7c) only stored disp; vel/accel must error
|
||||
explicitly so the caller knows to re-run, not silently return zeros."""
|
||||
import h5py
|
||||
|
||||
h5_path = tmp_path / "old_case.h5"
|
||||
with h5py.File(h5_path, "w") as f:
|
||||
f.create_dataset("time", data=np.array([0.0, 0.01]))
|
||||
f.create_dataset("nodes/1/disp", data=np.zeros((2, 6)))
|
||||
|
||||
r = TransientResults(case_id=1, case_name="x", h5_path=h5_path, n_steps=2, dt=0.01)
|
||||
# disp works:
|
||||
assert r.node_disp_history(1).shape == (2, 6)
|
||||
# vel/accel raise:
|
||||
with pytest.raises(KeyError, match="vel"):
|
||||
r.node_vel_history(1)
|
||||
with pytest.raises(KeyError, match="accel"):
|
||||
r.node_accel_history(1)
|
||||
|
||||
|
||||
def test_export_mode_shape_video_writes_file(tmp_path: Path) -> None:
|
||||
"""Smoke: export ⇒ produces a non-empty file."""
|
||||
pytest.importorskip("imageio")
|
||||
import pyvista as pv
|
||||
|
||||
from otko.services.animation_export import export_mode_shape_video
|
||||
|
||||
plotter = pv.Plotter(off_screen=True, window_size=(160, 120))
|
||||
sphere = pv.Sphere(radius=1.0)
|
||||
plotter.add_mesh(sphere)
|
||||
|
||||
captured = []
|
||||
|
||||
def set_phase(phase: float) -> None:
|
||||
captured.append(phase)
|
||||
|
||||
out = tmp_path / "anim.gif"
|
||||
export_mode_shape_video(plotter, set_phase, out, n_frames=4, fps=4)
|
||||
plotter.close()
|
||||
|
||||
assert out.exists()
|
||||
assert out.stat().st_size > 0
|
||||
# 4 frames called → 4 phases recorded.
|
||||
assert len(captured) == 4
|
||||
|
||||
|
||||
def test_export_time_history_video_decimates(tmp_path: Path) -> None:
|
||||
pytest.importorskip("imageio")
|
||||
import pyvista as pv
|
||||
|
||||
from otko.services.animation_export import (
|
||||
export_time_history_video,
|
||||
)
|
||||
|
||||
plotter = pv.Plotter(off_screen=True, window_size=(160, 120))
|
||||
plotter.add_mesh(pv.Cube())
|
||||
|
||||
seen_steps = []
|
||||
|
||||
def set_step(step: int) -> None:
|
||||
seen_steps.append(step)
|
||||
|
||||
out = tmp_path / "th.gif"
|
||||
# 100 steps with every=10 → 10 frames captured.
|
||||
export_time_history_video(plotter, set_step, out, n_steps=100, fps=4, every=10)
|
||||
plotter.close()
|
||||
|
||||
assert out.exists()
|
||||
assert seen_steps == [0, 10, 20, 30, 40, 50, 60, 70, 80, 90]
|
||||
Loading…
Reference in a new issue