make library_column_order a shared parent class as a move toward consistency

cdisc_rules_engine/constants/permissibility.py

@@ -2,3 +2,4 @@
 EXPECTED = "Exp"
 PERMISSIBLE = "Perm"
 PERMISSIBILITY_KEY = "core"
+PERMISSIBILITY_DEFAULT = PERMISSIBLE

cdisc_rules_engine/operations/base_operation.py

@@ -1,8 +1,4 @@
 from cdisc_rules_engine.models.operation_params import OperationParams
-from cdisc_rules_engine.constants.permissibility import (
-    PERMISSIBLE,
-    PERMISSIBILITY_KEY,
-)
 from abc import abstractmethod
 from typing import List
 import pandas as pd
@@ -237,14 +233,6 @@ def _get_variables_metadata_from_standard(self) -> List[dict]:
             dataset_path=self.params.dataset_path,
         )
 
-    def get_allowed_variable_permissibility(self, variable_metadata: dict):
-        """
-        Returns the permissibility value of a variable allowed in the current domain
-        """
-        if PERMISSIBILITY_KEY in variable_metadata:
-            return variable_metadata[PERMISSIBILITY_KEY]
-        return PERMISSIBLE
-
     def _get_variable_names_list(self, domain, dataframe):
         # get variables metadata from the standard model
         variables_metadata: List[dict] = (

cdisc_rules_engine/operations/expected_variables.py

@@ -1,9 +1,11 @@
-from cdisc_rules_engine.operations.base_operation import BaseOperation
-from cdisc_rules_engine.constants.permissibility import EXPECTED
-from typing import List
+from cdisc_rules_engine.constants.permissibility import (
+    EXPECTED,
+    PERMISSIBILITY_KEY,
+)
+from cdisc_rules_engine.operations.library_column_order import LibraryColumnOrder
 
 
-class ExpectedVariables(BaseOperation):
+class ExpectedVariables(LibraryColumnOrder):
     def _execute_operation(self):
         """
         Fetches required variables for a given domain from the CDISC library.
@@ -17,16 +19,6 @@ def _execute_operation(self):
         The lists with column names are sorted
         in accordance to "ordinal" key of library metadata.
         """
-
-        # get variables metadata from the standard/model
-        variables_metadata: List[dict] = self._get_variables_metadata_from_standard()
-
-        return list(
-            {
-                BaseOperation._replace_variable_wildcard(
-                    var["name"], self.params.domain
-                ): None
-                for var in variables_metadata
-                if self.get_allowed_variable_permissibility(var) == EXPECTED
-            }.keys()
-        )
+        self.params.key_name = PERMISSIBILITY_KEY
+        self.params.key_value = EXPECTED
+        return super()._execute_operation()

cdisc_rules_engine/operations/get_dataset_filtered_variables.py

@@ -1,8 +1,7 @@
-from typing import List
-from cdisc_rules_engine.operations.base_operation import BaseOperation
+from cdisc_rules_engine.operations.library_column_order import LibraryColumnOrder
 
 
-class GetDatasetFilteredVariables(BaseOperation):
+class GetDatasetFilteredVariables(LibraryColumnOrder):
     def _execute_operation(self):
         """
         Filter variables from the dataset based on specified criteria.
@@ -11,21 +10,7 @@ def _execute_operation(self):
         - key_name: The metadata key to filter by (e.g., "role", "type", etc.)
         - key_value: The value to match for the filter key (e.g., "Timing", "Identifier", etc.)
         """
-        filter_key = self.params.key_name
-        filter_value = self.params.key_value
-
-        # Get variables metadata from the standard model for the current domain
-        variables_metadata: List[dict] = self._get_variables_metadata_from_standard()
-
-        # Filter variables based on the specified criteria
-        filtered_variables = [
-            var for var in variables_metadata if var.get(filter_key) == filter_value
-        ]
-
-        # Replace variable wildcards with actual domain names
-        variable_names_list = self._replace_variable_wildcards(
-            filtered_variables, self.params.domain
-        )
+        variable_names_list = super()._execute_operation()
 
         # Get actual column names from the dataset that match our filtered list
         dataset_columns = self.params.dataframe.columns.tolist()

cdisc_rules_engine/operations/library_column_order.py

@@ -1,6 +1,5 @@
 from cdisc_rules_engine.operations.base_operation import BaseOperation
 from typing import List
-from collections import OrderedDict
 
 
 class LibraryColumnOrder(BaseOperation):
@@ -16,16 +15,28 @@ def _execute_operation(self):
         Length of Series is equal to the length of given dataframe.
         The lists with column names are sorted
         in accordance to "ordinal" key of library metadata.
-        """
 
-        # get variables metadata , for custom domains from model; for non-custom from IG and model
+        If key_name and key_value are provided, filter variables based on specified criteria.
+
+        Optional parameters:
+        - key_name: The metadata key to filter by (e.g., "role", "type", etc.)
+        - key_value: The value to match for the filter key (e.g., "Timing", "Identifier", etc.)
+        """
+        # Get variables metadata from the standard model for the current domain
         variables_metadata: List[dict] = self._get_variables_metadata_from_standard()
 
-        # create a list of variable names in accordance to the "ordinal" key
-        variable_names_list = BaseOperation._replace_variable_wildcards(
-            variables_metadata,
-            self.data_service.get_raw_dataset_metadata(
-                dataset_name=self.params.dataset_path, datasets=self.params.datasets
-            ).wildcard_replacement,
+        # Filter variables based on the specified criteria
+
+        if self.params.key_name:
+            variables_metadata = [
+                var
+                for var in variables_metadata
+                if var.get(self.params.key_name) == self.params.key_value
+            ]
+
+        # Replace variable wildcards with actual domain names
+        variable_names_list = self._replace_variable_wildcards(
+            variables_metadata, self.params.domain
         )
-        return list(OrderedDict.fromkeys(variable_names_list))
+
+        return variable_names_list

cdisc_rules_engine/operations/permissible_variables.py

@@ -1,9 +1,8 @@
-from cdisc_rules_engine.operations.base_operation import BaseOperation
-from cdisc_rules_engine.constants.permissibility import PERMISSIBLE
-from typing import List
+from cdisc_rules_engine.constants.permissibility import PERMISSIBILITY_KEY, PERMISSIBLE
+from cdisc_rules_engine.operations.library_column_order import LibraryColumnOrder
 
 
-class PermissibleVariables(BaseOperation):
+class PermissibleVariables(LibraryColumnOrder):
     def _execute_operation(self):
         """
         Fetches required variables for a given domain from the CDISC library.
@@ -17,16 +16,6 @@ def _execute_operation(self):
         The lists with column names are sorted
         in accordance to "ordinal" key of library metadata.
         """
-
-        # get variables metadata from the standard model
-        variables_metadata: List[dict] = self._get_variables_metadata_from_standard()
-
-        return list(
-            {
-                BaseOperation._replace_variable_wildcard(
-                    var["name"], self.params.domain
-                ): None
-                for var in variables_metadata
-                if self.get_allowed_variable_permissibility(var) == PERMISSIBLE
-            }.keys()
-        )
+        self.params.key_name = PERMISSIBILITY_KEY
+        self.params.key_value = PERMISSIBLE
+        return super()._execute_operation()

cdisc_rules_engine/operations/required_variables.py

@@ -1,9 +1,8 @@
-from typing import List
-from cdisc_rules_engine.operations.base_operation import BaseOperation
-from cdisc_rules_engine.constants.permissibility import REQUIRED
+from cdisc_rules_engine.constants.permissibility import PERMISSIBILITY_KEY, REQUIRED
+from cdisc_rules_engine.operations.library_column_order import LibraryColumnOrder
 
 
-class RequiredVariables(BaseOperation):
+class RequiredVariables(LibraryColumnOrder):
     def _execute_operation(self):
         """
         Fetches required variables for a given domain from the CDISC library.
@@ -17,15 +16,6 @@ def _execute_operation(self):
         The lists with column names are sorted
         in accordance to "ordinal" key of library metadata.
         """
-
-        # get variables metadata from the standard model
-        variables_metadata: List[dict] = self._get_variables_metadata_from_standard()
-        return list(
-            {
-                BaseOperation._replace_variable_wildcard(
-                    var["name"], self.params.domain
-                ): None
-                for var in variables_metadata
-                if self.get_allowed_variable_permissibility(var) == REQUIRED
-            }.keys()
-        )
+        self.params.key_name = PERMISSIBILITY_KEY
+        self.params.key_value = REQUIRED
+        return super()._execute_operation()

cdisc_rules_engine/utilities/sdtm_utilities.py

@@ -25,6 +25,10 @@
     SPECIAL_PURPOSE,
     SPECIAL_PURPOSE_MODEL,
 )
+from cdisc_rules_engine.constants.permissibility import (
+    PERMISSIBILITY_DEFAULT,
+    PERMISSIBILITY_KEY,
+)
 from cdisc_rules_engine.enums.variable_roles import VariableRoles
 from cdisc_rules_engine.models.library_metadata_container import (
     LibraryMetadataContainer,
@@ -224,6 +228,7 @@ def get_variables_metadata_from_standard(  # noqa
                 )
             else:
                 variables_metadata = ig_variables
+    set_default_variable_permissibility(variables_metadata)
     return variables_metadata
 
 
@@ -403,6 +408,7 @@ def get_variables_metadata_from_standard_model(  # noqa
             timing_metadata,
         ]:
             replace_variable_wildcards(var_list, original_domain, variables_metadata)
+            set_default_variable_permissibility(variables_metadata)
         return variables_metadata
     else:
         # First, try to get class metadata and check for classVariables
@@ -424,6 +430,7 @@ def get_variables_metadata_from_standard_model(  # noqa
             replace_variable_wildcards(
                 class_variables, original_domain, variables_metadata
             )
+            set_default_variable_permissibility(variables_metadata)
             return variables_metadata
         else:
             # Second, check if domain exists in model datasets
@@ -446,6 +453,7 @@ def get_variables_metadata_from_standard_model(  # noqa
                     dataset_variables, original_domain, variables_metadata
                 )
                 variables_metadata.sort(key=lambda item: int(item["ordinal"]))
+                set_default_variable_permissibility(variables_metadata)
                 return variables_metadata
             # Third, fall back to standard datasets
             if IG_domain_details:
@@ -465,6 +473,7 @@ def get_variables_metadata_from_standard_model(  # noqa
                 replace_variable_wildcards(
                     dataset_variables, original_domain, variables_metadata
                 )
+                set_default_variable_permissibility(variables_metadata)
                 return variables_metadata
         return None
 
@@ -487,6 +496,12 @@ def replace_variable_wildcards(var_list, domain, target_list):
         target_list.append(var_copy)
 
 
+def set_default_variable_permissibility(var_list):
+    for variable_metadata in var_list:
+        if PERMISSIBILITY_KEY not in variable_metadata:
+            variable_metadata[PERMISSIBILITY_KEY] = PERMISSIBILITY_DEFAULT
+
+
 def get_all_model_wildcard_variables(model_details: dict):
     return {
         classVariable["name"]