diff --git a/docs/cdb/api.rst b/docs/cdb/api.rst index f7971d9..3e1981e 100644 --- a/docs/cdb/api.rst +++ b/docs/cdb/api.rst @@ -38,8 +38,8 @@ associated public API. CableLoad CableResult CrossSectionalData - _GroupData - _GroupLCData + Groups + GroupsLC _LoadCases _Node _NodeData @@ -48,7 +48,7 @@ associated public API. _NodeResidual Quads QuadData - _SecondaryGroupLCData + SecondaryGroupsLC _Spring _SpringData _SpringResult diff --git a/docs/cdb/test_setup.rst b/docs/cdb/test_setup.rst index 6cac1ac..a5e8035 100644 --- a/docs/cdb/test_setup.rst +++ b/docs/cdb/test_setup.rst @@ -94,11 +94,14 @@ the temporary environment variable approach, open an MSYS2 MINGW64 shell, naviga tests/cable_load tests/cable_result tests/cross_section + tests/groups + tests/groups_lc tests/node_data tests/node_load tests/node_residual tests/node_result tests/quad_data + tests/sec_groups_lc tests/spring_data tests/spring_result tests/truss_data diff --git a/docs/cdb/tests/groups.rst b/docs/cdb/tests/groups.rst new file mode 100644 index 0000000..798fc52 --- /dev/null +++ b/docs/cdb/tests/groups.rst @@ -0,0 +1,37 @@ +Groups +------ + +Related test suite: ``test_groups.py`` + +Expected CDB file name: ``GROUP_DATA.cdb`` + +Runs with: SOFiSTiK 2025 + +Version: 1 + +.. code-block:: text + + +PROG AQUA + HEAD MATERIAL AND SECTIONS + NORM EN 199X-200X + STEE NO 1 TYPE YC ES 210000.0 GAM 78.5 TITL 'S355' + PROF 1 TYPE CHS 60.0 10.0 MNO 1 + END + + +PROG SOFIMSHA + HEAD GEOMETRY REV-1-SOF-2025 + SYST 3D GDIR NEGZ GDIV 10 + NODE NO 1 X 00.0 Y 0.0 Z +0.0 FIX F + NODE NO 2 X 05.0 Y 0.0 Z -0.5 + NODE NO 3 X 10.0 Y 0.0 Z -1.0 FIX F + + GRP 3 TITL 'GRP 3' + SPRI NO 2 NA 2 NE 3 + CABL NO 6 NA 1 NE 2 NCS 1 + GRP 10 TITL 'GRP 10' + BEAM NO 1 NA 1 NE 2 NCS 1 + TRUS NO 1 NA 2 NE 3 NCS 1 + GRP 20 TITL 'GRP 20' + BEAM NO 2 NA 2 NE 3 NCS 1 + CABL NO 6 NA 1 NE 2 NCS 1 + END diff --git a/docs/cdb/tests/groups_lc.rst b/docs/cdb/tests/groups_lc.rst new file mode 100644 index 0000000..6796ca3 --- /dev/null +++ b/docs/cdb/tests/groups_lc.rst @@ -0,0 +1,53 @@ +GroupsLC +-------- + +Related test suite: ``test_groups_lc.py`` + +Expected CDB file name: ``GROUP_LC_DATA.cdb`` + +Runs with: SOFiSTiK 2025 + +Version: 1 + +.. code-block:: text + + +PROG AQUA + HEAD MATERIAL AND SECTIONS + NORM EN 199X-200X + STEE NO 1 TYPE YC ES 210000.0 GAM 78.5 TITL 'S355' + PROF 1 TYPE CHS 60.0 10.0 MNO 1 + END + + +PROG SOFIMSHA + HEAD GEOMETRY REV-1-SOF-2025 + SYST 3D GDIR NEGZ GDIV 10 + NODE NO 1 X 00.0 Y 0.0 Z +0.0 FIX F + NODE NO 2 X 05.0 Y 0.0 Z -0.5 + NODE NO 3 X 10.0 Y 0.0 Z -1.0 FIX F + + GRP 3 TITL 'GRP 3' + SPRI NO 2 NA 2 NE 3 + CABL NO 6 NA 1 NE 2 NCS 1 + GRP 10 TITL 'GRP 10' + BEAM NO 1 NA 1 NE 2 NCS 1 + TRUS NO 1 NA 2 NE 3 NCS 1 + GRP 20 TITL 'GRP 20' + BEAM NO 2 NA 2 NE 3 NCS 1 + CABL NO 6 NA 1 NE 2 NCS 1 + END + + +PROG ASE + HEAD DUMMY ANALYSES + LOOP#I 2 + SYST PROB LINE + GRP - VAL YES + IF (#I==0) + GRP 10 VAL OFF + ELSE + GRP 3 VAL OFF + ENDIF + LET#LC #I+1000 + LC #LC DLZ 1.0 TITL 'DUMMY' + END + ENDLOOP + END diff --git a/docs/cdb/tests/sec_groups_lc.rst b/docs/cdb/tests/sec_groups_lc.rst new file mode 100644 index 0000000..328f413 --- /dev/null +++ b/docs/cdb/tests/sec_groups_lc.rst @@ -0,0 +1,63 @@ +GroupsLC +-------- + +Related test suite: ``test_secondary_groups_lc.py`` + +Expected CDB file name: ``SEC_GROUPS_LC.cdb`` + +Runs with: SOFiSTiK 2025 + +Version: 1 + +.. code-block:: text + + +PROG AQUA + HEAD MATERIAL AND SECTIONS + NORM EN 199X-200X + STEE NO 1 TYPE YC ES 210000.0 GAM 78.5 TITL 'S355' + PROF 1 TYPE CHS 60.0 10.0 MNO 1 + END + + +PROG SOFIMSHA + HEAD GEOMETRY REV-1-SOF-2025 + SYST 3D GDIR NEGZ GDIV 10 + NODE NO 1 X 00.0 Y 0.0 Z +0.0 FIX F + NODE NO 2 X 05.0 Y 0.0 Z -0.5 + NODE NO 3 X 10.0 Y 0.0 Z -1.0 FIX F + + GRP 3 TITL 'GRP 3' + SPRI NO 2 NA 2 NE 3 + CABL NO 6 NA 1 NE 2 NCS 1 + GRP 10 TITL 'GRP 10' + BEAM NO 1 NA 1 NE 2 NCS 1 + TRUS NO 1 NA 2 NE 3 NCS 1 + GRP 20 TITL 'GRP 20' + BEAM NO 2 NA 2 NE 3 NCS 1 + BEAM NO 4 NA 1 NE 2 NCS 1 + CABL NO 6 NA 1 NE 2 NCS 1 + END + + +PROG SOFIMSHA + HEAD SECONDARY GROUPS + SYST REST + GRP 'TEST' TITL 'SECONDARY TEST GROUP' + BEAM (202,-204) + END + + +PROG ASE + HEAD DUMMY ANALYSES + LOOP#I 2 + SYST PROB LINE + GRP - VAL YES + IF (#I==0) + GRP 10 VAL OFF + GRP 'TEST' VAL OFF + ELSE + GRP 3 VAL OFF + GRP 'TEST' VAL YES + ENDIF + LET#LC #I+1000 + LC #LC DLZ 1.0 TITL 'DUMMY' + END + ENDLOOP + END diff --git a/src/py_sofistik_utils/cdb_reader/__init__.py b/src/py_sofistik_utils/cdb_reader/__init__.py index 582260f..40e6cc2 100644 --- a/src/py_sofistik_utils/cdb_reader/__init__.py +++ b/src/py_sofistik_utils/cdb_reader/__init__.py @@ -7,8 +7,8 @@ from . _internals.cable_load import CableLoad from . _internals.cable_result import CableResult from . _internals.cross_section_data import CrossSectionalData -from . _internals.group_data import _GroupData -from . _internals.group_lc_data import _GroupLCData +from . _internals.group_data import Groups +from . _internals.group_lc_data import GroupsLC from . _internals.load_cases import _LoadCases from . _internals.node import _Node from . _internals.node_data import _NodeData @@ -17,7 +17,7 @@ from . _internals.node_result import _NodeResult from . _internals.quad import Quads from . _internals.quad_data import QuadData -from . _internals.sec_group_lc_data import _SecondaryGroupLCData +from . _internals.sec_group_lc_data import SecondaryGroupsLC from . _internals.spring import _Spring from . _internals.spring_data import _SpringData from . _internals.spring_result import _SpringResult @@ -39,8 +39,8 @@ "CableLoad", "CableResult", "CrossSectionalData", - "_GroupData", - "_GroupLCData", + "Groups", + "GroupsLC", "_LoadCases", "_Node", "_NodeData", @@ -49,7 +49,7 @@ "_NodeResult", "Quads", "QuadData", - "_SecondaryGroupLCData", + "SecondaryGroupsLC", "_Spring", "_SpringData", "_SpringResult", diff --git a/src/py_sofistik_utils/cdb_reader/_internals/beam_data.py b/src/py_sofistik_utils/cdb_reader/_internals/beam_data.py index 8b17f99..4987e60 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/beam_data.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/beam_data.py @@ -5,7 +5,7 @@ from pandas import DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_dll import SofDll from . sofistik_classes import CBEAM, CBEAM_SCT from . sofistik_utilities import decode_beam_end_release @@ -261,13 +261,13 @@ def load(self) -> None: ) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() df = DataFrame(conv_data).sort_values("ELEM_ID", kind="mergesort") elem_ids = df["ELEM_ID"] - for grp, grp_range in group_data.iterator_beam(): + for grp, grp_range in group_data.iterator("BEAM"): if grp_range.stop == 0: continue left = elem_ids.searchsorted(grp_range.start, side="left") diff --git a/src/py_sofistik_utils/cdb_reader/_internals/beam_results.py b/src/py_sofistik_utils/cdb_reader/_internals/beam_results.py index d62793d..d92c089 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/beam_results.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/beam_results.py @@ -6,7 +6,7 @@ # local library specific imports from . beam_data import _BeamData -from . group_data import _GroupData +from . group_data import Groups from . sofistik_dll import SofDll from . sofistik_classes import CBEAM_FOR @@ -193,13 +193,13 @@ def load(self, load_cases: int | list[int]) -> None: data.extend(self._load(load_case)) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() df = DataFrame(data).sort_values("ELEM_ID", kind="mergesort") elem_ids = df["ELEM_ID"] - for grp, grp_range in group_data.iterator_beam(): + for grp, grp_range in group_data.iterator("BEAM"): if grp_range.stop == 0: continue left = elem_ids.searchsorted(grp_range.start, side="left") diff --git a/src/py_sofistik_utils/cdb_reader/_internals/beam_stresses.py b/src/py_sofistik_utils/cdb_reader/_internals/beam_stresses.py index 9237903..4a6a08e 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/beam_stresses.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/beam_stresses.py @@ -6,7 +6,7 @@ # local library specific imports from . beam_data import _BeamData -from . group_data import _GroupData +from . group_data import Groups from . sofistik_dll import SofDll from . sofistik_classes import CBEAM_STR from . sofistik_utilities import long_to_str @@ -190,13 +190,13 @@ def load(self, load_cases: int | list[int]) -> None: data.extend(self._load(load_case)) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() df = DataFrame(data).sort_values("ELEM_ID", kind="mergesort") elem_ids = df["ELEM_ID"] - for grp, grp_range in group_data.iterator_beam(): + for grp, grp_range in group_data.iterator("BEAM"): if grp_range.stop == 0: continue left = elem_ids.searchsorted(grp_range.start, side="left") diff --git a/src/py_sofistik_utils/cdb_reader/_internals/cable_data.py b/src/py_sofistik_utils/cdb_reader/_internals/cable_data.py index 5f3d418..ff17a4d 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/cable_data.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/cable_data.py @@ -5,7 +5,7 @@ from pandas import concat, DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_dll import SofDll from . sofistik_classes import CCABL @@ -163,13 +163,13 @@ def load(self) -> None: ) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() df = DataFrame(data).sort_values("ELEM_ID", kind="mergesort") elem_ids = df["ELEM_ID"] - for grp, grp_range in group_data.iterator_cable(): + for grp, grp_range in group_data.iterator("CABLE"): if grp_range.stop == 0: continue diff --git a/src/py_sofistik_utils/cdb_reader/_internals/cable_load.py b/src/py_sofistik_utils/cdb_reader/_internals/cable_load.py index 273a3a5..176d87d 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/cable_load.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/cable_load.py @@ -5,7 +5,7 @@ from pandas import concat, DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_classes import CCABL_LOA from . sofistik_dll import SofDll @@ -208,13 +208,13 @@ def load(self, load_cases: int | list[int]) -> None: data.extend(self._load(load_case)) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() df = DataFrame(data).sort_values("ELEM_ID", kind="mergesort") elem_ids = df["ELEM_ID"] - for grp, grp_range in group_data.iterator_cable(): + for grp, grp_range in group_data.iterator("CABLE"): if grp_range.stop == 0: continue diff --git a/src/py_sofistik_utils/cdb_reader/_internals/cable_result.py b/src/py_sofistik_utils/cdb_reader/_internals/cable_result.py index 4840740..cb58c93 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/cable_result.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/cable_result.py @@ -5,7 +5,7 @@ from pandas import concat, DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_dll import SofDll from . sofistik_classes import CCABL_RES @@ -170,13 +170,13 @@ def load(self, load_cases: int | list[int]) -> None: data.extend(self._load(load_case)) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() df = DataFrame(data).sort_values("ELEM_ID", kind="mergesort") elem_ids = df["ELEM_ID"] - for grp, grp_range in group_data.iterator_cable(): + for grp, grp_range in group_data.iterator("CABLE"): if grp_range.stop == 0: continue diff --git a/src/py_sofistik_utils/cdb_reader/_internals/group_data.py b/src/py_sofistik_utils/cdb_reader/_internals/group_data.py index 1484aee..9d26b9b 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/group_data.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/group_data.py @@ -1,30 +1,74 @@ # standard library imports from ctypes import byref, c_int, sizeof -from typing import Any, Generator +from typing import Generator # third party library imports -from pandas import concat, DataFrame +from pandas import DataFrame # local library specific imports -from . sofistik_dll import SofDll from . sofistik_classes import CGRP +from . sofistik_dll import SofDll from . sofistik_utilities import long_to_str -class _GroupData: +class Groups: + """This class provides methods and a data structure to: + + * access keys ``11/0`` of the CDB file; + * store the retrieved data in a convenient format; + * provide access to the data after the CDB is closed. + + The underlying data structure is a :class:`pandas.DataFrame` with the + following columns: + + * ``GROUP"`` group number + * ``GROUP_NAME"`` group nmae + * ``BEAM_MIN_ID"`` beam minimum id + * ``BEAM_MAX_ID"`` beam maximum id + * ``NUMBER_OF_BEAMS"`` number of beams + * ``TRUSS_MIN_ID"`` truss minimum id + * ``TRUSS_MAX_ID"`` truss maximum id + * ``NUMBER_OF_TRUSSES"`` number of trusses + * ``CABLE_MIN_ID"`` cable minimum id + * ``CABLE_MAX_ID"`` cable maximum id + * ``NUMBER_OF_CABLES"`` number of cables + * ``SPRING_MIN_ID"`` spring minimum id + * ``SPRING_MAX_ID"`` spring maximum id + * ``NUMBER_OF_SPRINGS"`` number of springs + * ``QUAD_MIN_ID"`` quad minimum id + * ``QUAD_MAX_ID"`` quad maximum id + * ``NUMBER_OF_QUADS`` number of quads + + The ``DataFrame`` uses a MultiIndex with level ``GROUP`` to enable fast + lookups via the `get_id_range`, `get_name` and `get_number` methods. + The index columns are not dropped from the ``DataFrame``. + + .. note:: + + Not all available quantities are retrieved and stored. In + particular: + + * ``INF``: bit-code of the group + * ``MNR``: material number of the group + * ``MBW``: material reinforcement number of the group + * ``IBB`` and ``IBD``: construction stage numbers + + are currently not included. + + This is a deliberate design choice and may be changed in the future + without breaking the existing API. """ - This class provides methods and data structure to: + _MAP = { + 100: ("BEAM_MIN_ID", "BEAM_MAX_ID", "NUMBER_OF_BEAMS"), + 150: ("TRUSS_MIN_ID", "TRUSS_MAX_ID", "NUMBER_OF_TRUSSES"), + 160: ("CABLE_MIN_ID", "CABLE_MAX_ID", "NUMBER_OF_CABLES"), + 170: ("SPRING_MIN_ID", "SPRING_MAX_ID", "NUMBER_OF_SPRINGS"), + 200: ("QUAD_MIN_ID", "QUAD_MAX_ID", "NUMBER_OF_QUADS"), + } - * access and load the key ``011/00`` of the CDB file; - * store these data in a convenient format; - * provide access to these data. - """ def __init__(self, dll: SofDll) -> None: - """The initializer of the ``_GroupData`` class. - """ - self._dll = dll self._data = DataFrame( - columns = [ + columns=[ "GROUP", "GROUP_NAME", "BEAM_MIN_ID", @@ -44,153 +88,71 @@ def __init__(self, dll: SofDll) -> None: "NUMBER_OF_QUADS" ] ) + self._dll = dll def clear(self) -> None: - """Clear all group data. + """Clear all the loaded data. """ self._data = self._data[0:0] - def get_beam_id_range(self, group_number: int) -> range: - """Return a `range` starting from the minimum beam element ID to the maximum ID + - 1, so that a check like ``max_id in get_beam_id_range(grp_nmb)`` return `True`. - - If no beam elements are present in the given ``group_number`` return ``range(0)``. + def get_data(self, deep: bool = True) -> DataFrame: + """Return the :class:`pandas.DataFrame` containing the loaded key + ``11/0``. Parameters ---------- - group_number: int - The group number - - Raises - ------ - RuntimeError - If the given ``group_number`` is not found. + 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). """ - mask = self._data["GROUP"] == group_number - - if mask.eq(False).all(): - raise RuntimeError(f"Group {group_number} not found!") - - if self._data.NUMBER_OF_BEAMS[mask].item() == 0: - return range(0) - - max_id = self._data.BEAM_MAX_ID[mask].item() - min_id = self._data.BEAM_MIN_ID[mask].item() - - return range(min_id, max_id + 1, 1) - - def get_cable_id_range(self, group_number: int) -> range: - """Return a `range` starting from the minimum cable element ID to the maximum ID + - 1, so that a check like ``max_id in get_cable_id_range(grp_nmb)`` return `True`. - - If no cable elements are present in the given ``group_number`` return ``range(0)``. - - Parameters - ---------- - group_number: int - The group number - - Raises - ------ - RuntimeError - If the given ``group_number`` is not found. - """ - mask = self._data["GROUP"] == group_number - - if mask.eq(False).all(): - raise RuntimeError(f"Group {group_number} not found!") - - if self._data.NUMBER_OF_CABLES[mask].item() == 0: - return range(0) - - max_id = self._data.CABLE_MAX_ID[mask].item() - min_id = self._data.CABLE_MIN_ID[mask].item() - - return range(min_id, max_id + 1, 1) + return self._data.copy(deep=deep) def get_groups(self) -> list[int]: """Return a `list` of groups. """ if self._data.GROUP.empty: - raise RuntimeError("No groups found! Check if load() has been called.") + raise RuntimeError("No groups found!") return self._data.GROUP.to_list() - def get_group_name(self, group_number: int) -> str: - """Return a string containing the group name, given its number. - - Parameters - ---------- - group_number: int - The group number - - Raises - ------ - RuntimeError - If the given ``group_number`` is not found. - """ - mask = self._data["GROUP"] == group_number + def get_id_range(self, quantity: str, group_number: int) -> range: + """Return a `range` starting from the minimum element ID to the maximum + ID + 1, so that a check like ``max_id in get_id_range("BEAM", + group_number)`` returns `True`. - if mask.eq(False).all(): - raise RuntimeError(f"Group {group_number} not found!") - - return str(self._data.GROUP_NAME[mask].item()) - - def get_group_number(self, group_name: str) -> int: - """Return the group number, given its name. + If no elements of the requested type are present in the given + ``group_number`` returns ``range(0)``. Parameters ---------- - group_name: str - The group name + quantity: str + The type of finite element for which the range is requested. Must + be one of: - Raises - ------ - RuntimeError - If the given ``group_name`` is not found. - """ - mask = self._data["GROUP_NAME"] == group_name.upper() + - ``"BEAM"`` + - ``"CABLE"`` + - ``"TRUSS"`` + - ``"SPRING"`` + - ``"QUAD"`` - if mask.eq(False).all(): - raise RuntimeError(f"Group \"{group_name}\" not found!") - - return int(self._data.GROUP[mask].item()) - - def get_quad_id_range(self, group_number: int) -> range: - """Return a `range` starting from the minimum quad element ID to the maximum ID - + 1, so that a check like ``max_id in get_quad_id_range(grp_nmb)`` return `True`. - - If no quad elements are present in the given ``group_number`` return ``range(0)``. - - Parameters - ---------- group_number: int The group number - - Raises - ------ - RuntimeError - If the given ``group_number`` is not found. """ - mask = self._data["GROUP"] == group_number - - if mask.eq(False).all(): - raise RuntimeError(f"Group {group_number} not found!") - - if self._data.NUMBER_OF_QUADS[mask].item() == 0: - return range(0) - - max_id = self._data.QUAD_MAX_ID[mask].item() - min_id = self._data.QUAD_MIN_ID[mask].item() - - return range(min_id, max_id + 1, 1) - - def get_spring_id_range(self, group_number: int) -> range: - """Return a `range` starting from the minimum spring element ID to the maximum ID - + 1, so that a check like ``max_id in get_spring_id_range(grp_nmb)`` return `True`. - - If no spring elements are present in the given ``group_number`` return ``range(0)`` - . + try: + return range( + self._data.at[group_number, f"{quantity}_MIN_ID"], # type: ignore + self._data.at[group_number, f"{quantity}_MAX_ID"] + 1 # type: ignore + ) + except (KeyError, ValueError) as e: + raise LookupError( + f"Range not found for group number {group_number} " + f"and quantity {quantity}!" + ) from e + + def get_name(self, group_number: int) -> str: + """Return a string containing the group name, given its number. Parameters ---------- @@ -199,164 +161,123 @@ def get_spring_id_range(self, group_number: int) -> range: Raises ------ - RuntimeError + LookupError If the given ``group_number`` is not found. """ - mask = self._data["GROUP"] == group_number - - if mask.eq(False).all(): - raise RuntimeError(f"Group {group_number} not found!") - - if self._data.NUMBER_OF_SPRINGS[mask].item() == 0: - return range(0) - - max_id = self._data.SPRING_MAX_ID[mask].item() - min_id = self._data.SPRING_MIN_ID[mask].item() - - return range(min_id, max_id + 1, 1) - - def get_truss_id_range(self, group_number: int) -> range: - """Return a `range` starting from the minimum truss element ID to the maximum ID - + 1, so that a check like ``max_id in get_truss_id_range(grp_nmb)`` return `True`. - - If no truss elements are present in the given ``group_number`` return ``range(0)``. + try: + return self._data.at[group_number, "GROUP_NAME"] # type: ignore + except (KeyError, ValueError) as e: + raise LookupError( + f"Name not found for group number {group_number}!" + ) from e + + def get_number(self, group_name: str) -> int: + """Return the group number, given its name. Parameters ---------- - group_number: int - The group number + group_name: str + The group name Raises ------ - RuntimeError - If the given ``group_number`` is not found. - """ - mask = self._data["GROUP"] == group_number - - if mask.eq(False).all(): - raise RuntimeError(f"Group {group_number} not found!") - - if self._data.NUMBER_OF_TRUSSES[mask].item() == 0: - return range(0) - - max_id = self._data.TRUSS_MAX_ID[mask].item() - min_id = self._data.TRUSS_MIN_ID[mask].item() - - return range(min_id, max_id + 1, 1) - - def iterator_beam(self) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the beam ID range. - """ - for grp in self.get_groups(): - yield (grp, self.get_beam_id_range(grp)) - - def iterator_cable(self) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the cable ID range. - """ - for grp in self.get_groups(): - yield (grp, self.get_cable_id_range(grp)) - - def iterator_quad(self) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the quad ID range. - """ - for grp in self.get_groups(): - yield (grp, self.get_quad_id_range(grp)) - - def iterator_spring(self) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the spring ID range. + LookupError + If the given ``group_name`` is not found. """ - for grp in self.get_groups(): - yield (grp, self.get_spring_id_range(grp)) + try: + return self._data.index[self._data["GROUP_NAME"] == group_name][0] # type: ignore + except (KeyError, ValueError) as e: + raise LookupError( + f"Group number not found for group name {group_name}!" + ) from e + + def iterator( + self, + quantity: str + ) -> Generator[tuple[int, range], None, None]: + """Yield a `tuple` containing the group number and the ``quantity`` ID + range for each group defined in the CDB. - def iterator_truss(self) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the truss ID range. + Parameters + ---------- + quantity: str + The type of finite element for which the range is requested. Must + be one of: + + - ``"BEAM"`` + - ``"CABLE"`` + - ``"TRUSS"`` + - ``"SPRING"`` + - ``"QUAD"`` + + See Also + -------- + `get_id_range()` """ for grp in self.get_groups(): - yield (grp, self.get_truss_id_range(grp)) + yield (grp, self.get_id_range(quantity, grp)) def load(self) -> None: - """Load the group data. + """Load group data (key 11/0) from the CDB. """ if self._dll.key_exist(11, 0): - g_data = CGRP() - rec_length = c_int(sizeof(g_data)) + group = CGRP() + rec_length = c_int(sizeof(group)) return_value = c_int(0) - self.clear() - - temp_container: list[list[Any]] = [] - count = 0 + data: dict[int, dict[str, float | int | str]] = {} + first_call = True while return_value.value < 2: return_value.value = self._dll.get( 1, 11, 0, - byref(g_data), + byref(group), byref(rec_length), - 0 if count == 0 else 1 + 0 if first_call else 1 ) - rec_length = c_int(sizeof(g_data)) - count += 1 - + rec_length = c_int(sizeof(group)) + first_call = False if return_value.value >= 2: break - temp_list: list[Any] = [0 for _ in range(17)] - - if g_data.m_typ == 0: - temp_list[0] = g_data.m_ng - g_name = "".join(long_to_str(g_data.m_text[_]) for _ in range(17)) - temp_list[1] = g_name.upper() - temp_container.append(temp_list) + if group.m_typ == 0: + name = "".join( + long_to_str(group.m_text[_]) for _ in range(17) + ).upper() + data.update( + { + group.m_ng: { + "GROUP": group.m_ng, + "GROUP_NAME": name, + "BEAM_MIN_ID": 0, + "BEAM_MAX_ID": 0, + "NUMBER_OF_BEAMS": 0, + "TRUSS_MIN_ID": 0, + "TRUSS_MAX_ID": 0, + "NUMBER_OF_TRUSSES": 0, + "CABLE_MIN_ID": 0, + "CABLE_MAX_ID": 0, + "NUMBER_OF_CABLES": 0, + "SPRING_MIN_ID": 0, + "SPRING_MAX_ID": 0, + "NUMBER_OF_SPRINGS": 0, + "QUAD_MIN_ID": 0, + "QUAD_MAX_ID": 0, + "NUMBER_OF_QUADS": 0 + } + } + ) else: - useful_data = True - match g_data.m_typ: - case 100: - type_index = 2 - case 150: - type_index = 5 - case 160: - type_index = 8 - case 170: - type_index = 11 - case 200: - type_index = 14 - case _: - useful_data = False - - if useful_data: - grp_index = [_[0] for _ in temp_container].index(g_data.m_ng) - temp_container[grp_index][type_index + 0] = g_data.m_min - temp_container[grp_index][type_index + 1] = g_data.m_max - temp_container[grp_index][type_index + 2] = g_data.m_num - - # preparing data for conversion to a pandas DataFrame - conv_data: list[dict[str, Any]] = [] - for item in temp_container: - conv_data.append({"GROUP": item[0], - "GROUP_NAME": item[1], - "BEAM_MIN_ID": item[2], - "BEAM_MAX_ID": item[3], - "NUMBER_OF_BEAMS": item[4], - "TRUSS_MIN_ID": item[5], - "TRUSS_MAX_ID": item[6], - "NUMBER_OF_TRUSSES": item[7], - "CABLE_MIN_ID": item[8], - "CABLE_MAX_ID": item[9], - "NUMBER_OF_CABLES": item[10], - "SPRING_MIN_ID": item[11], - "SPRING_MAX_ID": item[12], - "NUMBER_OF_SPRINGS": item[13], - "QUAD_MIN_ID": item[14], - "QUAD_MAX_ID": item[15], - "NUMBER_OF_QUADS": item[16]}) - - if self._data.empty: - self._data = DataFrame(conv_data) - else: - self._data = concat( - [self._data, DataFrame(conv_data)], - ignore_index=True - ) + if group.m_typ in self._MAP.keys(): + min_key, max_key, num_key = self._MAP[group.m_typ] + data[group.m_ng][min_key] = group.m_min + data[group.m_ng][max_key] = group.m_max + data[group.m_ng][num_key] = group.m_num + + # set indices for fast lookup + self._data = ( + DataFrame(data.values()).set_index(["GROUP"], drop=False) + ) diff --git a/src/py_sofistik_utils/cdb_reader/_internals/group_lc_data.py b/src/py_sofistik_utils/cdb_reader/_internals/group_lc_data.py index 8590385..24550d3 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/group_lc_data.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/group_lc_data.py @@ -1,425 +1,170 @@ # standard library imports from ctypes import byref, c_int, sizeof -from typing import Any, Generator +from typing import Generator # third party library imports from pandas import concat, DataFrame # local library specific imports -from . sofistik_dll import SofDll from . sofistik_classes import CGRP_LC +from . sofistik_dll import SofDll -class _GroupLCData: - """ - This class provides methods and data structure to: +class GroupsLC: + """This class provides methods and a data structure to: + + * access keys ``11/LC`` of the CDB file; + * store the retrieved data in a convenient format; + * provide access to the data after the CDB is closed. + + The underlying data structure is a :class:`pandas.DataFrame` with the + following columns: + + * ``GROUP"`` group number + * ``LOAD_CASE"`` the load case number + * ``IS_ACTIVE"`` indicates whether the group is active in the load case + + The ``DataFrame`` uses a MultiIndex with level ``GROUP`` and + ``LOAD_CASE`` (in this specific order) to enable fast lookups via the + `is_active` method. The index columns are not dropped from the + ``DataFrame``. - * access and load the key ``011/LC`` of the CDB file; - * store these data in a convenient format; - * provide access to these data. + .. note:: + + Not all available quantities are retrieved and stored. In + particular: + + * ``MNR``: material number of the group + * ``MBW``: material reinforcement number of the group + * ``IBB`` and ``IBD``: construction stage numbers + * ``MIN_ID``, ``MAX_ID`` and number of elements. + + are currently not included; however, ``MIN_ID``, ``MAX_ID``, and + the number of elements are available in the `Groups` class. + + This is a deliberate design choice and may be changed in the future + without breaking the existing API. """ def __init__(self, dll: SofDll) -> None: - """The initializer of the ``_GroupLCData`` class. - """ self._data = DataFrame( - columns = [ - "LOAD_CASE", + columns=[ "GROUP", - "BEAM_MIN_ID", - "BEAM_MAX_ID", - "NUMBER_OF_BEAMS", - "TRUSS_MIN_ID", - "TRUSS_MAX_ID", - "NUMBER_OF_TRUSSES", - "CABLE_MIN_ID", - "CABLE_MAX_ID", - "NUMBER_OF_CABLES", - "SPRING_MIN_ID", - "SPRING_MAX_ID", - "NUMBER_OF_SPRINGS", - "QUAD_MIN_ID", - "QUAD_MAX_ID", - "NUMBER_OF_QUADS", - "IS_ACTIVE" + "LOAD_CASE", + "IS_ACTIVE", ] ) self._dll = dll 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.drop(self._data[self._data.LOAD_CASE == load_case].index) + 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_active_groups(self, load_case: int) -> list[int]: - """For the given ``load_case``, return the `list` of active groups. + def get_data(self, deep: bool = True) -> DataFrame: + """Return the :class:`pandas.DataFrame` containing the loaded keys + ``11/LC``. Parameters ---------- - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` is not found. + 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). """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - active_mask = self._data["IS_ACTIVE"] == True - - return self._data.GROUP[lc_mask & active_mask].to_list() - - def get_beam_id_range(self, load_case: int, group_number: int) -> range: - """For the given ``load_case``, return a `range` starting from the minimum beam - element ID to the maximum ID + 1, so that a check like - ``max_id in get_beam_id_range(lc, grp_nmb)`` return `True`. - - If no beam elements are present in the given ``load_case`` and ``group_number`` - return ``range(0)``. - - Parameters - ---------- - group_number: int - The group number - load_case: int - The load_case number + return self._data.copy(deep=deep) - Raises - ------ - RuntimeError - If the given ``load_case`` or ``group_number`` are not found. + def is_active(self, group_number: int, load_case: int) -> bool: + """Return `True` if the group ``group_number`` is active in the + load case ``load_case``. """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_number - - if grp_mask.eq(False).all(): - err_msg = f"Group {group_number} not found in load case {load_case}!" - raise RuntimeError(err_msg) - - if self._data.NUMBER_OF_BEAMS[lc_mask & grp_mask].item() == 0: - return range(0) - - max_id = self._data.BEAM_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.BEAM_MIN_ID[lc_mask & grp_mask].item() + try: + return self._data.at[(group_number, load_case), "IS_ACTIVE"] # type: ignore + except (KeyError, ValueError) as e: + raise LookupError( + f"Group {group_number} not found in load case {load_case}!" + ) from e - return range(min_id, max_id + 1, 1) - - def get_cable_id_range(self, load_case: int, group_number: int) -> range: - """For the given ``load_case``, return a `range` starting from the minimum cable - element ID to the maximum ID + 1, so that a check like - ``max_id in get_cable_id_range(lc, grp_nmb)`` return `True`. - - If no cable elements are present in the given ``load_case`` and ``group_number`` - return ``range(0)``. + def load(self, load_cases: int | list[bool | int]) -> None: + """Retrieve group data for the given ``load_cases``. Parameters ---------- - group_number: int - The group number - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` or ``group_number`` are not found. - """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_number - - if grp_mask.eq(False).all(): - err_msg = f"Group {group_number} not found in load case {load_case}!" - raise RuntimeError(err_msg) - - if self._data.NUMBER_OF_CABLES[lc_mask & grp_mask].item() == 0: - return range(0) - - max_id = self._data.CABLE_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.CABLE_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) - - def get_quad_id_range(self, load_case: int, group_number: int) -> range: - """For the given ``load_case``, return a `range` starting from the minimum quad - element ID to the maximum ID + 1, so that a check like - ``max_id in get_quad_id_range(lc, grp_nmb)`` return `True`. - - If no quad elements are present in the given ``load_case`` and ``group_number`` - return ``range(0)``. - - Parameters - ---------- - group_number: int - The group number - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` or ``group_number`` are not found. - """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_number - - if grp_mask.eq(False).all(): - err_msg = f"Group {group_number} not found in load case {load_case}!" - raise RuntimeError(err_msg) - - if self._data.NUMBER_OF_QUADS[lc_mask & grp_mask].item() == 0: - return range(0) - - max_id = self._data.QUAD_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.QUAD_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) - - def get_spring_id_range(self, load_case: int, group_number: int) -> range: - """For the given ``load_case``, return a `range` starting from the minimum spring - element ID to the maximum ID + 1, so that a check like - ``max_id in get_spring_id_range(lc, grp_nmb)`` return `True`. - - If no spring elements are present in the given ``load_case`` and ``group_number`` - return ``range(0)``. - - Parameters - ---------- - group_number: int - The group number - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` or ``group_number`` are not found. - """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_number - - if grp_mask.eq(False).all(): - err_msg = f"Group {group_number} not found in load case {load_case}!" - raise RuntimeError(err_msg) - - if self._data.NUMBER_OF_SPRINGS[lc_mask & grp_mask].item() == 0: - return range(0) - - max_id = self._data.SPRING_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.SPRING_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) - - def get_truss_id_range(self, load_case: int, group_number: int) -> range: - """For the given ``load_case``, return a `range` starting from the minimum truss - element ID to the maximum ID + 1, so that a check like - ``max_id in get_truss_id_range(lc, grp_nmb)`` return `True`. - - If no truss elements are present in the given ``load_case`` and ``group_number`` - return ``range(0)``. - - Parameters - ---------- - group_number: int - The group number - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` or ``group_number`` are not found. - """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_number - - if grp_mask.eq(False).all(): - err_msg = f"Group {group_number} not found in load case {load_case}!" - raise RuntimeError(err_msg) - - if self._data.NUMBER_OF_TRUSSES[lc_mask & grp_mask].item() == 0: - return range(0) - - max_id = self._data.TRUSS_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.TRUSS_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) - - def group_is_active(self, load_case: int, group_number: int) -> bool: - """Return `True` if the given ``group_number`` is active in the given ``load_case``. - `False` otherwise. - - Parameters - ---------- - group_number: int - The group number - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` or ``group_number`` are not found. - """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_number - - if grp_mask.eq(False).all(): - err_msg = f"Group {group_number} not found in load case {load_case}!" - raise RuntimeError(err_msg) - - return bool(self._data.IS_ACTIVE[lc_mask & grp_mask].item()) - - def iterator_beam(self, load_case: int) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the beam ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_beam_id_range(load_case, grp)) - - def iterator_cable(self, load_case: int) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the cable ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_cable_id_range(load_case, grp)) - - def iterator_quad(self, load_case: int) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the quad ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_quad_id_range(load_case, grp)) - - def iterator_spring(self, load_case: int) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the spring ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_spring_id_range(load_case, grp)) - - def iterator_truss(self, load_case: int) -> Generator[tuple[int, range], None, None]: - """Yield a tuple containing the group number and the truss ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_truss_id_range(load_case, grp)) - - def load(self, load_case: int) -> None: - """Load the group data for the given ``load_case``. - """ - if self._dll.key_exist(11, load_case): - g_data = CGRP_LC() - rec_length = c_int(sizeof(g_data)) - return_value = c_int(0) - - self.clear(load_case) - - temp_container: list[list[Any]] = [] - count = 0 - while return_value.value < 2: - return_value.value = self._dll.get( - 1, - 11, - load_case, - byref(g_data), - byref(rec_length), - 0 if count == 0 else 1 - ) - - rec_length = c_int(sizeof(g_data)) - count += 1 - - if return_value.value >= 2 or g_data.m_ng > 999: - break - - temp_list: list[Any] = [0 for _ in range(18)] - - if g_data.m_typ == 0: - temp_list[0] = load_case - temp_list[1] = g_data.m_ng - temp_list[-1] = (2 & g_data.m_inf) > 0 - temp_container.append(temp_list) - - else: - useful_data = True - match g_data.m_typ: - case 100: - type_index = 2 - case 150: - type_index = 5 - case 160: - type_index = 8 - case 170: - type_index = 11 - case 200: - type_index = 14 - case _: - useful_data = False - - if useful_data: - grp_index = [_[1] for _ in temp_container].index(g_data.m_ng) - temp_container[grp_index][type_index + 0] = g_data.m_min - temp_container[grp_index][type_index + 1] = g_data.m_max - temp_container[grp_index][type_index + 2] = g_data.m_num - - # preparing data for conversion to a pandas DataFrame - conv_data: list[dict[str, Any]] = [] - for item in temp_container: - conv_data.append({"LOAD_CASE": item[0], - "GROUP": item[1], - "BEAM_MIN_ID": item[2], - "BEAM_MAX_ID": item[3], - "NUMBER_OF_BEAMS": item[4], - "TRUSS_MIN_ID": item[5], - "TRUSS_MAX_ID": item[6], - "NUMBER_OF_TRUSSES": item[7], - "CABLE_MIN_ID": item[8], - "CABLE_MAX_ID": item[9], - "NUMBER_OF_CABLES": item[10], - "SPRING_MIN_ID": item[11], - "SPRING_MAX_ID": item[12], - "NUMBER_OF_SPRINGS": item[13], - "QUAD_MIN_ID": item[14], - "QUAD_MAX_ID": item[15], - "NUMBER_OF_QUADS": item[16], - "IS_ACTIVE": item[17]}) - - if self._data.empty: - self._data = DataFrame(conv_data) - else: - self._data = concat( - [self._data, DataFrame(conv_data)], - ignore_index=True - ) - self._loaded_lc.add(load_case) + load_cases : int | list[int] + load case numbers + """ + if isinstance(load_cases, int): + load_cases = [load_cases] + else: + load_cases = list(set(load_cases)) # remove duplicated entries + + # load data + data: list[dict[str, int]] = [] + for load_case in load_cases: + if self._dll.key_exist(162, load_case): + self.clear(load_case) + data.extend(self._load(load_case)) + + df = DataFrame(data).sort_values("GROUP", kind="mergesort") + + # set indices for fast lookup + df = df.set_index(["GROUP", "LOAD_CASE"], drop=False) + + # merge data + if self._data.empty: + self._data = df + else: + self._data = concat([self._data, df]) + self._loaded_lc.update(load_cases) + + def _load(self, load_case: int) -> list[dict[str, bool | int]]: + """Load key ``11/load_case`` from the CDB. + """ + group = CGRP_LC() + rec_length = c_int(sizeof(group)) + return_value = c_int(0) + + data: dict[int, dict[str, bool | int]] = {} + first_call = True + while return_value.value < 2: + return_value.value = self._dll.get( + 1, + 11, + load_case, + byref(group), + byref(rec_length), + 0 if first_call else 1 + ) + + rec_length = c_int(sizeof(group)) + first_call = False + if return_value.value >= 2: + break + + if group.m_typ != 0 or group.m_ng > 999: + break + + data.update( + { + group.m_ng: { + "GROUP": group.m_ng, + "LOAD_CASE": load_case, + "IS_ACTIVE": bool(group.m_inf & 2) + } + } + ) + + return list(data.values()) diff --git a/src/py_sofistik_utils/cdb_reader/_internals/quad_data.py b/src/py_sofistik_utils/cdb_reader/_internals/quad_data.py index a694477..f4e1d21 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/quad_data.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/quad_data.py @@ -5,7 +5,7 @@ from pandas import DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_dll import SofDll from . sofistik_classes import CQUAD @@ -174,10 +174,10 @@ def load(self) -> None: elem_ids = df["ELEM_ID"] # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() - for grp, grp_range in group_data.iterator_quad(): + for grp, grp_range in group_data.iterator("QUAD"): if grp_range.stop == 0: continue left = elem_ids.searchsorted(grp_range.start, side="left") diff --git a/src/py_sofistik_utils/cdb_reader/_internals/sec_group_lc_data.py b/src/py_sofistik_utils/cdb_reader/_internals/sec_group_lc_data.py index 707b925..8a380c6 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/sec_group_lc_data.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/sec_group_lc_data.py @@ -1,7 +1,5 @@ # standard library imports -from copy import deepcopy from ctypes import byref, c_int, sizeof -from typing import Any, Generator # third party library imports from pandas import concat, DataFrame @@ -12,91 +10,57 @@ from . sofistik_utilities import long_to_str -class _SecondaryGroupLCData: - """The ``_SecondaryGroupLCData`` class provides methods and data structure to: +class SecondaryGroupsLC: + """This class provides methods and a data structure to: - * access and load the key ``011/LC`` of the CDB file; - * store these data in a convenient format; - * provide access to these data. + * access secondary groups info in keys ``11/LC`` of the CDB file; + * store the retrieved data in a convenient format; + * provide access to the data after the CDB is closed. - Only the secondary groups data are stored in this class. + The underlying data structure is a :class:`pandas.DataFrame` with the + following columns: + + * ``GROUP_NAME"`` group name + * ``LOAD_CASE"`` the load case number + * ``IS_ACTIVE"`` indicates whether the group is active in the load case + + The ``DataFrame`` uses a MultiIndex with level ``GROUP`` and + ``LOAD_CASE`` (in this specific order) to enable fast lookups via the + `is_active` method. The index columns are not dropped from the + ``DataFrame``. + + .. note:: + + Not all available quantities are retrieved and stored. In + particular: + + * ``MNR``: material number of the group + * ``MBW``: material reinforcement number of the group + * ``IBB`` and ``IBD``: construction stage numbers + * ``MIN_ID``, ``MAX_ID`` and number of elements. + + are currently not included. + + This is a deliberate design choice and may be changed in the future + without breaking the existing API. """ def __init__(self, dll: SofDll) -> None: - """The initializer of the ``_SecondaryGroupLCData`` class. - """ self._data = DataFrame( - columns = [ + columns=[ "LOAD_CASE", - "GROUP", - "BEAM_MIN_ID", - "BEAM_MAX_ID", - "NUMBER_OF_BEAMS", - "TRUSS_MIN_ID", - "TRUSS_MAX_ID", - "NUMBER_OF_TRUSSES", - "CABLE_MIN_ID", - "CABLE_MAX_ID", - "NUMBER_OF_CABLES", - "SPRING_MIN_ID", - "SPRING_MAX_ID", - "NUMBER_OF_SPRINGS", - "QUAD_MIN_ID", - "QUAD_MAX_ID", - "NUMBER_OF_QUADS", + "GROUP_NAME", "IS_ACTIVE" ] ) self._dll = dll self._loaded_lc: set[int] = set() - def clear(self, load_case: int) -> None: - """Clear the results for the given ``load_case`` number. - """ - if load_case not in self._loaded_lc: - return - - self._data = self._data.drop(self._data[self._data.LOAD_CASE == load_case].index) - self._loaded_lc.remove(load_case) - - def clear_all(self) -> None: - """Clear all group data. - """ - self._data = self._data[0:0] - self._loaded_lc.clear() - - def get_active_groups(self, load_case: int) -> list[str]: - """For the given ``load_case``, return the list of active groups. - - Parameters - ---------- - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` is not loaded. - """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - active_mask = self._data["IS_ACTIVE"] == True - - return self._data.GROUP[lc_mask & active_mask].to_list() - - def get_beam_id_range(self, load_case: int, group_name: str) -> range: - """For the given ``load_case``, return a range starting from the minimum beam - element ID to the maximum ID + 1, so that a check like - ``max_id in get_beam_id_range(lc, grp_nmb)`` return ``True``. - - If no beam elements are present in the given ``load_case`` and ``group_name``: - return ``range(0)``. + def active_groups(self, load_case: int) -> list[str]: + """For the given ``load_case``, return the list of active secondary + groups. Parameters ---------- - group_name: str - The group number load_case: int The load_case number @@ -106,322 +70,122 @@ def get_beam_id_range(self, load_case: int, group_name: str) -> range: If the given ``load_case`` is not loaded. """ if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") + raise RuntimeError(f"Load case {load_case} not loaded!") lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_name.upper() - - if (lc_mask & grp_mask).eq(False).all(): - return range(0) - - max_id = self._data.BEAM_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.BEAM_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) + active_mask = self._data["IS_ACTIVE"] == True # noqa: E712 - def get_cable_id_range(self, load_case: int, group_name: str) -> range: - """For the given ``load_case``, return a range starting from the minimum cable - element ID to the maximum ID + 1, so that a check like - ``max_id in get_cable_id_range(lc, grp_nmb)`` return ``True``. - - If no cable elements are present in the given ``load_case`` and ``group_name``: - return ``range(0)``. - - Parameters - ---------- - group_name: str - The group number - load_case: int - The load_case number + return self._data.GROUP_NAME[lc_mask & active_mask].to_list() - Raises - ------ - RuntimeError - If the given ``load_case`` is not loaded. + def clear(self, load_case: int) -> None: + """Clear the loaded data for the given ``load_case`` number. """ if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_name.upper() - - if (lc_mask & grp_mask).eq(False).all(): - return range(0) - - max_id = self._data.CABLE_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.CABLE_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) - - def get_quad_id_range(self, load_case: int, group_name: str) -> range: - """For the given ``load_case``, return a `range` starting from the minimum quad - element ID to the maximum ID + 1, so that a check like - ``max_id in get_quad_id_range(lc, grp_nmb)`` return `True`. - - If no quad elements are present in the given ``load_case`` and ``group_name``: - return ``range(0)``. + return - Parameters - ---------- - group_name: str - The group number - load_case: int - The load_case number + self._data = self._data[ + self._data.index.get_level_values("LOAD_CASE") != load_case + ] + self._loaded_lc.remove(load_case) - Raises - ------ - RuntimeError - If the given ``load_case`` is not loaded. + def clear_all(self) -> None: + """Clear the loaded data for all the load cases. """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_name.upper() - - if (lc_mask & grp_mask).eq(False).all(): - return range(0) - - max_id = self._data.QUAD_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.QUAD_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) - - def get_spring_id_range(self, load_case: int, group_name: str) -> range: - """For the given ``load_case``, return a range starting from the minimum spring - element ID to the maximum ID + 1, so that a check like - ``max_id in get_spring_id_range(lc, grp_nmb)`` return ``True``. + self._data = self._data[0:0] + self._loaded_lc.clear() - If no spring elements are present in the given ``load_case`` and ``group_name``: - return ``range(0)``. + def get_data(self, deep: bool = True) -> DataFrame: + """Return the :class:`pandas.DataFrame` containing the loaded key + ``11/0``. Parameters ---------- - group_name: str - The group number - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` is not loaded. + 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). """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_name.upper() + return self._data.copy(deep=deep) - if (lc_mask & grp_mask).eq(False).all(): - return range(0) - - max_id = self._data.SPRING_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.SPRING_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) - - def get_truss_id_range(self, load_case: int, group_name: str) -> range: - """For the given ``load_case``, return a range starting from the minimum truss - element ID to the maximum ID + 1, so that a check like - ``max_id in get_truss_id_range(lc, grp_nmb)`` return ``True``. - - If no truss elements are present in the given ``load_case`` and ``group_name``: - return ``range(0)``. - - Parameters - ---------- - group_name: str - The group number - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` is not loaded. + def is_active(self, group_name: str, load_case: int) -> bool: + """Return `True` if the group ``group_name`` is active in the load case + ``load_case``. """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") + try: + return self._data.at[(group_name, load_case), "IS_ACTIVE"] # type: ignore + except (KeyError, ValueError) as e: + raise LookupError( + f"Secondary group {group_name} not found in LC {load_case}!" + ) from e - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_name.upper() - - if (lc_mask & grp_mask).eq(False).all(): - return range(0) - - max_id = self._data.TRUSS_MAX_ID[lc_mask & grp_mask].item() - min_id = self._data.TRUSS_MIN_ID[lc_mask & grp_mask].item() - - return range(min_id, max_id + 1, 1) - - def group_is_active(self, load_case: int, group_name: str) -> bool: - """Return ``True`` if the given ``group_name:` is active` in the given ``load_case``. - ``False`` otherwise. + def load(self, load_cases: int | list[bool | int]) -> None: + """Retrieve group data for the given ``load_cases``. Parameters ---------- - group_name: str - The group number - load_case: int - The load_case number - - Raises - ------ - RuntimeError - If the given ``load_case`` is not loaded. - """ - if load_case not in self._loaded_lc: - raise RuntimeError(f"Load case {load_case} not found!") - - lc_mask = self._data["LOAD_CASE"] == load_case - grp_mask = self._data["GROUP"] == group_name.upper() - - if (lc_mask & grp_mask).eq(False).all(): - err_msg = f"Group {group_name:} not found in load case {load_case}!" - raise RuntimeError(err_msg) - - return bool(self._data.IS_ACTIVE[lc_mask & grp_mask].item()) - - def iterator_beam(self, load_case: int) -> Generator[tuple[str, range], None, None]: - """Yield a tuple containing the group number and the beam ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_beam_id_range(load_case, grp)) - - def iterator_cable(self, load_case: int) -> Generator[tuple[str, range], None, None]: - """Yield a tuple containing the group number and the cable ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_cable_id_range(load_case, grp)) - - def iterator_quad(self, load_case: int) -> Generator[tuple[str, range], None, None]: - """Yield a tuple containing the group number and the quad ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_quad_id_range(load_case, grp)) - - def iterator_spring(self, load_case: int) -> Generator[tuple[str, range], None, None]: - """Yield a tuple containing the group number and the spring ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_spring_id_range(load_case, grp)) - - def iterator_truss(self, load_case: int) -> Generator[tuple[str, range], None, None]: - """Yield a tuple containing the group number and the truss ID range for the given - ``load_case``. - """ - for grp in self.get_active_groups(load_case): - yield (grp, self.get_truss_id_range(load_case, grp)) - - def load(self, load_case: int) -> None: - """Load the group data for the given ``load_case``. - """ - if self._dll.key_exist(11, load_case): - g_data = CGRP_LC() - rec_length = c_int(sizeof(g_data)) - return_value = c_int(0) - - self.clear(load_case) - - temp_container: list[list[Any]] = [] - count = 0 - while return_value.value < 2: - return_value.value = self._dll.get( - 1, - 11, - load_case, - byref(g_data), - byref(rec_length), - 0 if count == 0 else 1 - ) - - rec_length = c_int(sizeof(g_data)) - count += 1 - - if return_value.value >= 2: - break - - if g_data.m_ng <= 999: - continue - - temp_list: list[Any] = [0 for _ in range(18)] - grp_name = long_to_str(g_data.m_ng) - - # dummy addition to avoid IndexError in the next if statement - if not temp_container: - temp_container.append(deepcopy(temp_list)) - - if grp_name != temp_container[-1][1]: - temp_list[0] = load_case - temp_list[1] = grp_name - temp_list[-1] = (2 & g_data.m_inf) > 0 - temp_container.append(deepcopy(temp_list)) - - match g_data.m_typ: - case 100: - type_index = 2 - case 150: - type_index = 5 - case 160: - type_index = 8 - case 170: - type_index = 11 - case 200: - type_index = 14 - case _: - continue - - grp_index = [_[1] for _ in temp_container].index(grp_name) - - if (temp_container[grp_index][type_index + 0] == 0 or - g_data.m_min < temp_container[grp_index][type_index + 0] - ): - temp_container[grp_index][type_index + 0] = g_data.m_min - - if g_data.m_max > temp_container[grp_index][type_index + 1]: - temp_container[grp_index][type_index + 1] = g_data.m_max - - temp_container[grp_index][type_index + 2] = g_data.m_num - - # manage case that there are no secondary group data for this load case - if not temp_container: - return - - # remove the dummy addition on line 400 - del temp_container[0] - - # preparing data for conversion to a pandas DataFrame - conv_data: list[dict[str, Any]] = [] - for item in temp_container: - conv_data.append({"LOAD_CASE": item[0], - "GROUP": item[1], - "BEAM_MIN_ID": item[2], - "BEAM_MAX_ID": item[3], - "NUMBER_OF_BEAMS": item[4], - "TRUSS_MIN_ID": item[5], - "TRUSS_MAX_ID": item[6], - "NUMBER_OF_TRUSSES": item[7], - "CABLE_MIN_ID": item[8], - "CABLE_MAX_ID": item[9], - "NUMBER_OF_CABLES": item[10], - "SPRING_MIN_ID": item[11], - "SPRING_MAX_ID": item[12], - "NUMBER_OF_SPRINGS": item[13], - "QUAD_MIN_ID": item[14], - "QUAD_MAX_ID": item[15], - "NUMBER_OF_QUADS": item[16], - "IS_ACTIVE": item[17]}) - - if self._data.empty: - self._data = DataFrame(conv_data) - else: - self._data = concat( - [self._data, DataFrame(conv_data)], - ignore_index=True - ) - self._loaded_lc.add(load_case) + load_cases : int | list[int] + load case numbers + """ + if isinstance(load_cases, int): + load_cases = [load_cases] + else: + load_cases = list(set(load_cases)) # remove duplicated entries + + # load data + data: list[dict[str, int]] = [] + for load_case in load_cases: + if self._dll.key_exist(162, load_case): + self.clear(load_case) + data.extend(self._load(load_case)) + + df = DataFrame(data).sort_values("GROUP_NAME", kind="mergesort") + + # set indices for fast lookup + df = df.set_index(["GROUP_NAME", "LOAD_CASE"], drop=False) + + # merge data + if self._data.empty: + self._data = df + else: + self._data = concat([self._data, df]) + self._loaded_lc.update(load_cases) + + def _load(self, load_case: int) -> list[dict[str, bool | int]]: + """Load secondary groups info in key ``11/load_case`` from the CDB. + """ + group = CGRP_LC() + rec_length = c_int(sizeof(group)) + return_value = c_int(0) + + data: dict[int, dict[str, bool | int]] = {} + first_call = True + while return_value.value < 2: + return_value.value = self._dll.get( + 1, + 11, + load_case, + byref(group), + byref(rec_length), + 0 if first_call else 1 + ) + + rec_length = c_int(sizeof(group)) + first_call = False + if return_value.value >= 2: + break + + if group.m_ng <= 999: + continue + + data.update( + { + group.m_ng: { + "GROUP_NAME": long_to_str(group.m_ng), + "LOAD_CASE": load_case, + "IS_ACTIVE": bool(group.m_inf & 2) + } + } + ) + + return list(data.values()) diff --git a/src/py_sofistik_utils/cdb_reader/_internals/spring_data.py b/src/py_sofistik_utils/cdb_reader/_internals/spring_data.py index 9435b24..7a7aa78 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/spring_data.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/spring_data.py @@ -5,7 +5,7 @@ from pandas import concat, DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_classes import CSPRI from . sofistik_dll import SofDll @@ -211,13 +211,13 @@ def load(self) -> None: ) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() temp_df = DataFrame(data).sort_values("ELEM_ID", kind="mergesort") elem_ids = temp_df["ELEM_ID"] - for grp, grp_range in group_data.iterator_spring(): + for grp, grp_range in group_data.iterator("SPRING"): if grp_range.stop == 0: continue diff --git a/src/py_sofistik_utils/cdb_reader/_internals/spring_result.py b/src/py_sofistik_utils/cdb_reader/_internals/spring_result.py index 6653051..bea028f 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/spring_result.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/spring_result.py @@ -5,7 +5,7 @@ from pandas import concat, DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_dll import SofDll from . sofistik_classes import CSPRI_RES @@ -186,13 +186,13 @@ def load(self, load_cases: int | list[int]) -> None: temp_list.extend(self._load(load_case)) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() temp_df = DataFrame(temp_list).sort_values("ELEM_ID", kind="mergesort") elem_ids = temp_df["ELEM_ID"] - for grp, grp_range in group_data.iterator_spring(): + for grp, grp_range in group_data.iterator("SPRING"): if grp_range.stop == 0: continue diff --git a/src/py_sofistik_utils/cdb_reader/_internals/truss_data.py b/src/py_sofistik_utils/cdb_reader/_internals/truss_data.py index 7de9b78..b0d7285 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/truss_data.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/truss_data.py @@ -5,7 +5,7 @@ from pandas import concat, DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_dll import SofDll from . sofistik_classes import CTRUS @@ -165,13 +165,13 @@ def load(self) -> None: ) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() temp_df = DataFrame(data).sort_values("ELEM_ID", kind="mergesort") elem_ids = temp_df["ELEM_ID"] - for grp, grp_range in group_data.iterator_truss(): + for grp, grp_range in group_data.iterator("TRUSS"): if grp_range.stop == 0: continue diff --git a/src/py_sofistik_utils/cdb_reader/_internals/truss_load.py b/src/py_sofistik_utils/cdb_reader/_internals/truss_load.py index 2f11cf0..35b533d 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/truss_load.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/truss_load.py @@ -5,7 +5,7 @@ from pandas import concat, DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_classes import CTRUS_LOA from . sofistik_dll import SofDll @@ -174,13 +174,13 @@ def load(self, load_cases: int | list[int]) -> None: temp_list.extend(self._load(load_case)) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() temp_df = DataFrame(temp_list).sort_values("ELEM_ID", kind="mergesort") elem_ids = temp_df["ELEM_ID"] - for grp, grp_range in group_data.iterator_truss(): + for grp, grp_range in group_data.iterator("TRUSS"): if grp_range.stop == 0: continue diff --git a/src/py_sofistik_utils/cdb_reader/_internals/truss_result.py b/src/py_sofistik_utils/cdb_reader/_internals/truss_result.py index 61b85df..5f7c216 100644 --- a/src/py_sofistik_utils/cdb_reader/_internals/truss_result.py +++ b/src/py_sofistik_utils/cdb_reader/_internals/truss_result.py @@ -5,7 +5,7 @@ from pandas import concat, DataFrame # local library specific imports -from . group_data import _GroupData +from . group_data import Groups from . sofistik_classes import CTRUS_RES from . sofistik_dll import SofDll @@ -155,13 +155,13 @@ def load(self, load_cases: int | list[int]) -> None: temp_list.extend(self._load(load_case)) # assigning groups - group_data = _GroupData(self._dll) + group_data = Groups(self._dll) group_data.load() temp_df = DataFrame(temp_list).sort_values("ELEM_ID", kind="mergesort") elem_ids = temp_df["ELEM_ID"] - for grp, grp_range in group_data.iterator_truss(): + for grp, grp_range in group_data.iterator("TRUSS"): if grp_range.stop == 0: continue diff --git a/src/py_sofistik_utils/cdb_reader/reader.py b/src/py_sofistik_utils/cdb_reader/reader.py index 20846cc..067dc54 100644 --- a/src/py_sofistik_utils/cdb_reader/reader.py +++ b/src/py_sofistik_utils/cdb_reader/reader.py @@ -6,12 +6,12 @@ from . _internals.beam import Beam from . _internals.cable import Cables from . _internals.cross_section_data import CrossSectionalData -from . _internals.group_data import _GroupData -from . _internals.group_lc_data import _GroupLCData +from . _internals.group_data import Groups +from . _internals.group_lc_data import GroupsLC from . _internals.load_cases import _LoadCases from . _internals.node import _Node from . _internals.quad import Quads -from . _internals.sec_group_lc_data import _SecondaryGroupLCData +from . _internals.sec_group_lc_data import SecondaryGroupsLC from . _internals.spring import _Spring from . _internals.sofistik_dll import SofDll from . _internals.truss import _Truss @@ -32,10 +32,10 @@ class SOFiSTiKCDBReader: # other cdb data cross_sections: CrossSectionalData - group_data: _GroupData - group_lc_data: _GroupLCData + groups: Groups + groups_lc: GroupsLC load_cases: _LoadCases - sec_group_lc_data: _SecondaryGroupLCData + sec_groups_lc: SecondaryGroupsLC def __init__( self, @@ -60,10 +60,10 @@ def __init__( # other cdb data self.cross_sections = CrossSectionalData(self._dll) - self.group_data = _GroupData(self._dll) - self.group_lc_data = _GroupLCData(self._dll) + self.groups = Groups(self._dll) + self.groups_lc = GroupsLC(self._dll) self.load_cases = _LoadCases(self._dll) - self.sec_group_lc_data = _SecondaryGroupLCData(self._dll) + self.sec_groups_lc = SecondaryGroupsLC(self._dll) def close(self) -> None: """Close the CDB database. diff --git a/tests/cdb_reader/test_groups.py b/tests/cdb_reader/test_groups.py new file mode 100644 index 0000000..65e66c3 --- /dev/null +++ b/tests/cdb_reader/test_groups.py @@ -0,0 +1,92 @@ +# 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 = [ + "GROUP", + "GROUP_NAME", + "BEAM_MIN_ID", + "BEAM_MAX_ID", + "NUMBER_OF_BEAMS", + "TRUSS_MIN_ID", + "TRUSS_MAX_ID", + "NUMBER_OF_TRUSSES", + "CABLE_MIN_ID", + "CABLE_MAX_ID", + "NUMBER_OF_CABLES", + "SPRING_MIN_ID", + "SPRING_MAX_ID", + "NUMBER_OF_SPRINGS", + "QUAD_MIN_ID", + "QUAD_MAX_ID", + "NUMBER_OF_QUADS" +] +_DATA = [ + (3, "GRP 3", 0, 0, 0, 0, 0, 0, 36, 36, 1, 32, 32, 1, 0, 0, 0), + (10, "GRP 10", 101, 101, 1, 101, 101, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0), + (20, "GRP 20", 202, 202, 1, 0, 0, 0, 206, 206, 1, 0, 0, 0, 0, 0, 0) +] + + +@skipUnless( + all([CDB_PATH, DLL_PATH, VERSION]), + "SOFiSTiK environment variables not set!" +) +class SOFiSTiKCDBReaderGroupDataTestSuite(TestCase): + def setUp(self) -> None: + self.cdb = SOFiSTiKCDBReader( + CDB_PATH, # type: ignore + "GROUP_DATA", + DLL_PATH, # type: ignore + VERSION # type: ignore + ) + self.cdb.open() + self.cdb.groups.load() + + def tearDown(self) -> None: + self.cdb.close() + + def test_data(self) -> None: + assert_frame_equal( + DataFrame(_DATA, columns=_COLUMNS).set_index(["GROUP"], drop=False), + self.cdb.groups.get_data(), + rtol=1E-7 + ) + + def test_get_after_clear(self) -> None: + self.cdb.groups.clear() + with self.subTest(msg="Check clear method"): + with self.assertRaises(LookupError): + self.cdb.groups.get_id_range("BEAM", 10) + + self.cdb.groups.load() + with self.subTest(msg="Check indexes management"): + self.test_get_id_range() + + def test_get_group_name(self) -> None: + self.assertEqual(self.cdb.groups.get_name(10), "GRP 10") + + def test_get_group_number(self) -> None: + self.assertEqual(self.cdb.groups.get_number("GRP 10"), 10) + + def test_get_id_range(self) -> None: + with self.subTest(): + self.assertEqual( + self.cdb.groups.get_id_range("BEAM", 10), + range(101, 102) + ) + + with self.subTest(): + self.assertTrue(101 in self.cdb.groups.get_id_range("BEAM", 10)) diff --git a/tests/cdb_reader/test_groups_lc.py b/tests/cdb_reader/test_groups_lc.py new file mode 100644 index 0000000..49bab70 --- /dev/null +++ b/tests/cdb_reader/test_groups_lc.py @@ -0,0 +1,67 @@ +# 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 = ["GROUP", "LOAD_CASE", "IS_ACTIVE"] +_DATA = [ + (3, 1000, True), + (3, 1001, False), + (10, 1000, False), + (10, 1001, True), + (20, 1000, True), + (20, 1001, True) +] + + +@skipUnless( + all([CDB_PATH, DLL_PATH, VERSION]), + "SOFiSTiK environment variables not set!" +) +class SOFiSTiKCDBReaderGroupLCTestSuite(TestCase): + def setUp(self) -> None: + self.cdb = SOFiSTiKCDBReader( + CDB_PATH, # type: ignore + "GROUP_LC_DATA", + DLL_PATH, # type: ignore + VERSION # type: ignore + ) + self.cdb.open() + self.cdb.groups_lc.load([1000, 1001]) + + def tearDown(self) -> None: + self.cdb.close() + + def test_data(self) -> None: + assert_frame_equal( + DataFrame( + _DATA, + columns=_COLUMNS + ).set_index(["GROUP", "LOAD_CASE"], drop=False), + self.cdb.groups_lc.get_data(), + rtol=1E-7 + ) + + def test_is_active(self) -> None: + self.assertFalse(self.cdb.groups_lc.is_active(10, 1000)) + + def test_is_active_after_clear(self) -> None: + self.cdb.groups_lc.clear(1000) + with self.subTest(msg="Check clear method"): + with self.assertRaises(LookupError): + self.cdb.groups_lc.is_active(10, 1000) + + self.cdb.groups_lc.load(1000) + with self.subTest(msg="Check indexes management"): + self.test_is_active() diff --git a/tests/cdb_reader/test_secondary_groups_lc.py b/tests/cdb_reader/test_secondary_groups_lc.py new file mode 100644 index 0000000..5dafab8 --- /dev/null +++ b/tests/cdb_reader/test_secondary_groups_lc.py @@ -0,0 +1,66 @@ +# 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 = ["GROUP_NAME", "LOAD_CASE", "IS_ACTIVE"] +_DATA = [ + ("TEST", 1000, False), + ("TEST", 1001, True) +] + + +@skipUnless( + all([CDB_PATH, DLL_PATH, VERSION]), + "SOFiSTiK environment variables not set!" +) +class SOFiSTiKCDBReaderSecondaryGroupLCTestSuite(TestCase): + def setUp(self) -> None: + self.cdb = SOFiSTiKCDBReader( + CDB_PATH, # type: ignore + "SEC_GROUPS_LC", + DLL_PATH, # type: ignore + VERSION # type: ignore + ) + self.cdb.open() + self.cdb.sec_groups_lc.load([1000, 1001]) + + def tearDown(self) -> None: + self.cdb.close() + + def test_active_groups(self) -> None: + self.assertEqual(self.cdb.sec_groups_lc.active_groups(1001), ["TEST"]) + + def test_data(self) -> None: + assert_frame_equal( + DataFrame( + _DATA, + columns=_COLUMNS + ).set_index(["GROUP_NAME", "LOAD_CASE"], drop=False), + self.cdb.sec_groups_lc.get_data(), + rtol=1E-7 + ) + + def test_is_active(self) -> None: + self.assertFalse(self.cdb.sec_groups_lc.is_active("TEST", 1000)) + + def test_is_active_after_clear(self) -> None: + self.cdb.sec_groups_lc.clear(1000) + with self.subTest(msg="Check clear method"): + with self.assertRaises(LookupError): + self.cdb.sec_groups_lc.is_active("TEST", 1000) + + self.cdb.sec_groups_lc.load(1000) + with self.subTest(msg="Check indexes management"): + self.test_is_active()