diff --git a/docs/cdb/test_setup.rst b/docs/cdb/test_setup.rst index 10c9ae1..45baab0 100644 --- a/docs/cdb/test_setup.rst +++ b/docs/cdb/test_setup.rst @@ -93,3 +93,4 @@ the temporary environment variable approach, open an MSYS2 MINGW64 shell, naviga tests/spring_data tests/spring_result tests/truss_data + tests/truss_load diff --git a/docs/cdb/tests/truss_load.rst b/docs/cdb/tests/truss_load.rst new file mode 100644 index 0000000..50b0c05 --- /dev/null +++ b/docs/cdb/tests/truss_load.rst @@ -0,0 +1,67 @@ +TrussLoad +--------- + +Related test suite: ``test_truss_load.py`` + +Expected CDB file name: ``TRUSS_LOAD.cdb`` + +Runs with: SOFiSTiK 2025 + +Version: 1 + +.. code-block:: text + + +PROG AQUA + HEAD MATERIAL AND SECTIONS + NORM EN 199X-200X + STEE 1 S 355 FY 345 FT 470 GAM 78.5 TMAX 40.0 TITL 'S355' + PROF NO 1 'HEA' 220 MNO 1 + END + + +PROG SOFIMSHA + HEAD GEOMETRY REV-1-SOF-2025 + SYST 3D GDIV 10 GDIR NEGZ + + LET#COUNT 1 + LOOP#I 4 + LOOP#J 3 + NODE NO #COUNT X 0.0+#I Y 0.0+#J Z 0.0 + LET#COUNT #COUNT+1 + ENDLOOP + ENDLOOP + + GRP 500 TITL 'TRUSS-1' + TRUS NO 1 NA 1 NE 2 NCS 1 + TRUS NO 9 NA 2 NE 3 NCS 1 + + GRP 501 TITL 'TRUSS-2' + TRUS NO 1 NA 7 NE 8 NCS 1 + TRUS NO 4 NA 11 NE 12 NCS 1 + END + + +PROG SOFILOAD + HEAD TRUSS LOADS + LC 1 TITL 'LC-1-PG' + TRUS 5001 TYPE PG PA +1.0 PE -1.0 + LC 2 TITL 'LC-2-PXX' + TRUS GRP 500 TYPE PXX +2.0 + LC 3 TITL 'LC-3-PYY' + TRUS GRP 501 TYPE PYY -3.0 + LC 4 TITL 'LC-4-PZZ' + TRUS 5011 TYPE PZZ -4.0 + LC 5 TITL 'LC-5-PXP' + TRUS GRP 501 TYPE PXP +5.0 + LC 6 TITL 'LC-6-PYP' + TRUS GRP 500 TYPE PYP -6.0 + LC 7 TITL 'LC-6-PZP' + TRUS 5009 TYPE PZP -7.0 + LC 8 TITL 'LC-8-EX' + TRUS 5001 TYPE EX -8.0 + TRUS 5014 TYPE EX -8.0 + LC 9 TITL 'LC-9-WX' + TRUS GRP 501 TYPE WX +9.0 + LC 10 TITL 'LC-10-DT' + TRUS GRP 501 TYPE DT -10.0 + LC 11 TITL 'LC-11-VX' + TRUS GRP (500, 501, 1) TYPE VX 11.0 + END diff --git a/src/py_sofistik_utils/cdb_reader/_internal/truss_load.py b/src/py_sofistik_utils/cdb_reader/_internal/truss_load.py index aebf543..dfa9a53 100644 --- a/src/py_sofistik_utils/cdb_reader/_internal/truss_load.py +++ b/src/py_sofistik_utils/cdb_reader/_internal/truss_load.py @@ -1,6 +1,5 @@ # standard library imports from ctypes import byref, c_int, sizeof -from typing import Any # third party library imports from pandas import concat, DataFrame @@ -17,11 +16,25 @@ class _TrussLoad: * store these data in a convenient format; * provide access to these data. """ + _LOAD_TYPE_MAP = { + 10: "PG", + 11: "PXX", + 12: "PYY", + 13: "PZZ", + 30: "EX", + 31: "WX", + 60: "DT", + 61: "DT", + 70: "VX", + 80: "VX", + 111: "PXP", + 212: "PYP", + 313: "PZP" + } + def __init__(self, dll: SofDll) -> None: - """The initializer of the ``_TrussLoad`` class. - """ self._data = DataFrame( - columns = [ + columns=[ "LOAD_CASE", "GROUP", "ELEM_ID", @@ -35,158 +48,196 @@ def __init__(self, dll: SofDll) -> None: self._loaded_lc: set[int] = set() def clear(self, load_case: int) -> None: - """Clear the results for the given ``load_case`` number. + """Clear the loaded data for the given ``load_case`` number. """ if load_case not in self._loaded_lc: return - self._data = self._data.loc[~(self._data["LOAD_CASE"] == load_case), :] + self._data = self._data[ + self._data.index.get_level_values("LOAD_CASE") != load_case + ] self._loaded_lc.remove(load_case) def clear_all(self) -> None: - """Clear all group data. + """Clear the loaded data for all the load cases. """ self._data = self._data[0:0] self._loaded_lc.clear() - def get_element_load( + def data(self, deep: bool = True) -> DataFrame: + """Return the :class:`pandas.DataFrame` containing the loaded keys + ``151/LC``. + + Parameters + ---------- + deep : bool, default True + When ``deep=True``, a new object will be created with a copy of the + calling object's data and indices. Modifications to the data or + indices of the copy will not be reflected in the original object + (refer to :meth:`pandas.DataFrame.copy` documentation for details). + """ + return self._data.copy(deep=deep) + + def get( self, - element_number: int, + element_id: int, load_case: int, - load_type: str - ) -> Any: - """Return the cable connectivity for the given ``element_number``. + load_type: str, + point: str = "PA", + default: float | None = None + ) -> float: + """Retrieve the requested truss load. Parameters ---------- - ``element_number``: int - The cable element number - ``load_case``: int - The load case number - ``load_type``: str - The load type + element_id : int + Truss element number + load_case : int + Load case number + load_type : str + Load type to retrieve. Must be one of: + + - ``"PG"`` + - ``"PXX"`` + - ``"PYY"`` + - ``"PZZ"`` + - ``"EX"`` + - ``"WX"`` + - ``"DT"`` + - ``"VX"`` + - ``"PXP"`` + - ``"PYP"`` + - ``"PZP"`` + + point : str, default "PA" + Location on the truss where the load is applied; either the start + (``"PA"``) or the end (``"PE"``) + default : float or None, default None + Value to return if the requested load is not found + + Returns + ------- + value : float + The requested load if found. Otherwise, returns ``default`` when it + is not None. Raises ------ LookupError - If the requested data is not found. + If the requested load is not found and ``default`` is None. """ - raise NotImplementedError - e_mask = self._data["ELEM_ID"] == element_number - lc_mask = self._data["LOAD_CASE"] == load_case - lt_mask = self._data["TYPE"] == load_type - - if (e_mask & lc_mask & lt_mask).eq(False).all(): - err_msg = f"LC {load_case}, LT {load_type}, EL_ID {element_number} not found!" - raise LookupError(err_msg) - - return self._data[e_mask & lc_mask & lt_mask].copy(deep=True) + try: + return self._data.at[ + (element_id, load_case, load_type), point + ] # type: ignore + except (KeyError, ValueError) as e: + if default is not None: + return default + raise LookupError( + f"Truss load entry not found for element id {element_id}, load" + f" case {load_case}, load type {load_type} and point {point}!" + ) from e def load(self, load_cases: int | list[int]) -> None: - """Load cable element loads for the given the ``load_cases``. - - If a load case is not found, a warning is raised only if ``echo_level`` is ``> 0``. + """Retrieve truss load data for the given ``load_cases``. If a load + case is not found, a warning is raised only if ``echo_level > 0``. Parameters ---------- - ``load_cases``: int | list[int], load case numbers + load_cases : int | list[int] + load case numbers + + Notes + ----- + Wind and snow loads are not implemented and will raise a runtime error + if they are present in the requested load case. """ if isinstance(load_cases, int): load_cases = [load_cases] + else: + load_cases = list(set(load_cases)) # remove duplicated entries + # load data + temp_list: list[dict[str, float | int | str]] = [] for load_case in load_cases: if self._dll.key_exist(151, load_case): self.clear(load_case) + temp_list.extend(self._load(load_case)) - # load data - data = DataFrame(self._load(load_case)) + # assigning groups + group_data = _GroupData(self._dll) + group_data.load() - # merge data - if self._data.empty: - self._data = data - else: - self._data = concat([self._data, data], ignore_index=True) - self._loaded_lc.add(load_case) + temp_df = DataFrame(temp_list).sort_values("ELEM_ID", kind="mergesort") + elem_ids = temp_df["ELEM_ID"] - else: + for grp, grp_range in group_data.iterator_truss(): + if grp_range.stop == 0: continue - # assigning groups - group_data = _GroupData(self._dll) - group_data.load() + left = elem_ids.searchsorted(grp_range.start, side="left") + right = elem_ids.searchsorted(grp_range.stop - 1, side="right") + temp_df.loc[temp_df.index[left:right], "GROUP"] = grp + + # set indices for fast lookup + temp_df = temp_df.set_index( + ["ELEM_ID", "LOAD_CASE", "TYPE"], + drop=False + ) - for grp, cable_range in group_data.iterator_truss(): - self._data.loc[self._data.ELEM_ID.isin(cable_range), "GROUP"] = grp + # merge data + if self._data.empty: + self._data = temp_df + else: + self._data = concat([self._data, temp_df]) + self._loaded_lc.update(load_cases) - def _load(self, load_case: int) -> list[dict[str, Any]]: + def set_echo_level(self, echo_level: int) -> None: + """Set the echo level. """ + self._echo_level = echo_level + + def _load(self, load_case: int) -> list[dict[str, float | int | str]]: """ - cabl = CTRUS_LOA() - record_length = c_int(sizeof(cabl)) + """ + trus = CTRUS_LOA() + record_length = c_int(sizeof(trus)) return_value = c_int(0) - data: list[dict[str, Any]] = [] - count = 0 + data: list[dict[str, float | int | str]] = [] + first_call = True while return_value.value < 2: return_value.value = self._dll.get( 1, 151, load_case, - byref(cabl), + byref(trus), byref(record_length), - 0 if count == 0 else 1 + 0 if first_call else 1 ) - record_length = c_int(sizeof(cabl)) - count += 1 - + record_length = c_int(sizeof(trus)) + first_call = False if return_value.value >= 2: break - match cabl.m_typ: - case 10: - type_ = "PG" - case 11: - type_ = "PXX" - case 12: - type_ = "PYY" - case 13: - type_ = "PZZ" - case 30: - type_ = "EX" - case 31: - type_ = "WX" - case 60: - type_ = "T" - case 61: - type_ = "DT" - case 70: - type_ = "VX" - case 80: - type_ = "VX" - case 111: - type_ = "PXP" - case 212: - type_ = "PYP" - case 313: - type_ = "PZP" - case _: - raise RuntimeError(f"Unknown type: {cabl.m_typ}!") + try: + type_ = _TrussLoad._LOAD_TYPE_MAP[trus.m_typ] + except KeyError as e: + raise RuntimeError( + f"Unknown truss load type {trus.m_typ} for element " + f"{trus.m_nr}!" + ) from e data.append( { "LOAD_CASE": load_case, "GROUP": 0, - "ELEM_ID": cabl.m_nr, + "ELEM_ID": trus.m_nr, "TYPE": type_, - "PA": cabl.m_pa, - "PE": cabl.m_pe, + "PA": trus.m_pa, + "PE": trus.m_pe, } ) return data - - def set_echo_level(self, echo_level: int) -> None: - """Set the echo level. - """ - self._echo_level = echo_level diff --git a/tests/cdb_reader/test_truss_load.py b/tests/cdb_reader/test_truss_load.py new file mode 100644 index 0000000..420b397 --- /dev/null +++ b/tests/cdb_reader/test_truss_load.py @@ -0,0 +1,114 @@ +# standard library imports +from os import environ +from unittest import skipUnless, TestCase + +# third party library imports +from pandas import DataFrame +from pandas.testing import assert_frame_equal + +# local library specific imports +from py_sofistik_utils import SOFiSTiKCDBReader + + +CDB_PATH = environ.get("SOFISTIK_CDB_PATH") +DLL_PATH = environ.get("SOFISTIK_DLL_PATH") +VERSION = environ.get("SOFISTIK_VERSION") + + +_COLUMNS = ["LOAD_CASE", "GROUP", "ELEM_ID", "TYPE", "PA", "PE"] + +_DATA = [ + (1, 500, 5001, "PG", +1.0, -1.0), + (2, 500, 5001, "PXX", +2.0, +2.0), + (6, 500, 5001, "PYP", -6.0, -6.0), + (8, 500, 5001, "EX", -0.008, -0.008), + (11, 500, 5001, "VX", +11.0, +11.0), + (2, 500, 5009, "PXX", +2.0, +2.0), + (6, 500, 5009, "PYP", -6.0, -6.0), + (7, 500, 5009, "PZP", -7.0, -7.0), + (11, 500, 5009, "VX", +11.0, +11.0), + (3, 501, 5011, "PYY", -3.0, -3.0), + (4, 501, 5011, "PZZ", -4.0, -4.0), + (5, 501, 5011, "PXP", +5.0, +5.0), + (9, 501, 5011, "WX", +0.009, +0.009), + (10, 501, 5011, "DT", -10.0, -10.0), + (11, 501, 5011, "VX", +11.0, +11.0), + (3, 501, 5014, "PYY", -3.0, -3.0), + (5, 501, 5014, "PXP", +5.0, +5.0), + (8, 501, 5014, "EX", -0.008, -0.008), + (9, 501, 5014, "WX", +0.009, +0.009), + (10, 501, 5014, "DT", -10.0, -10.0), + (11, 501, 5014, "VX", +11.0, +11.0) + ] + + +@skipUnless( + all([CDB_PATH, DLL_PATH, VERSION]), + "SOFiSTiK environment variables not set!" +) +class SOFiSTiKCDBReaderTrussLoadTestSuite(TestCase): + """Tests for the `_TrussLoad` class. + """ + def setUp(self) -> None: + self.data = DataFrame(_DATA, columns=_COLUMNS).set_index( + ["ELEM_ID", "LOAD_CASE", "TYPE"], drop=False + ) + self.load_cases = list(range(1, 12, 1)) + + self.cdb = SOFiSTiKCDBReader(CDB_PATH, "TRUSS_LOAD", DLL_PATH, int(VERSION)) # type: ignore + self.cdb.initialize() + self.cdb.truss_load.load(self.load_cases) + + def tearDown(self) -> None: + self.cdb.close() + + def test_data(self) -> None: + """Test for the `data` method. + """ + # NOTE: + # Float values loaded from the CDB contain inherent numerical noise + # (e.g. -0.008 is represented as -0.00800000037997961). The chosen + # tolerance rtol=1e-7 is stricter than pandas default and reflects the + # maximum relative error observed in practice, ensuring stable and + # reproducible comparisons. + assert_frame_equal(self.data, self.cdb.truss_load.data(), rtol=1E-7) + + def test_get(self) -> None: + """Test for the `get` method. + """ + with self.subTest(msg="Existing entry"): + self.assertEqual(self.cdb.truss_load.get(5009, 7, "PZP", "PA"), -7) + + with self.subTest(msg="Non existing entry with default"): + self.assertEqual(self.cdb.truss_load.get(9009, 7, "PZP", "PA", -3), -3) + + def test_get_after_clear(self) -> None: + """Test for the `get` method after a `clear` call. + """ + self.cdb.truss_load.clear(7) + with self.subTest(msg="Check clear method"): + with self.assertRaises(LookupError): + self.cdb.truss_load.get(5009, 7, "PZP", "PA") + + self.cdb.truss_load.load(7) + with self.subTest(msg="Check indexes management"): + self.test_get() + + def test_get_after_clear_all(self) -> None: + """Test for the `get` method after a `clear_all` call. + """ + self.cdb.truss_load.clear_all() + with self.subTest(msg="Check clear_all method"): + with self.assertRaises(LookupError): + self.cdb.truss_load.get(5009, 7, "PZP", "PA") + + self.cdb.truss_load.load(self.load_cases) + with self.subTest(msg="Check indexes management"): + self.assertEqual(self.cdb.truss_load.get(5009, 7, "PZP", "PA"), -7) + + def test_load_with_duplicated_load_cases(self) -> None: + """Test for the `load` method with duplicated entries. + """ + self.cdb.truss_load.clear_all() + self.cdb.truss_load.load(self.load_cases + [10]) + self.assertEqual(self.cdb.truss_load.get(5009, 7, "PZP", "PA"), -7)