Merge pull request #28 from jmhorcas/master

SAT to BDD and efficient implementation of Product Distribution

.gitignore

@@ -3,4 +3,6 @@ __pycache__
 *.egg-info*
 env
 
-.vscode
+.vscode
+
+build/

flamapy/metamodels/bdd_metamodel/models/bdd_model.py

@@ -48,6 +48,10 @@ def nof_nodes(self) -> int:
         """Return number of nodes in the BDD."""
         return len(self.bdd)
 
+    def get_node(self, index: int) -> Function:
+        """Return the node at the given position (index)."""
+        return self.bdd.var(self.bdd.var_at_level(index))
+
     @staticmethod
     def level(node: Function) -> int:
         """Return the level of the node.
@@ -96,4 +100,14 @@ def get_low_node(node: Function) -> Function:
 
         If the arc is complemented it returns the negation of the left node.
         """
-        return ~node.low if node.negated and node.low.var is not None else node.low
+        return node.low
+
+    @staticmethod
+    def get_value(node: Function, complemented: bool = False) -> int:
+        """Return the value (id) of the node considering complemented arcs."""
+        value = node.node
+        if BDDModel.is_terminal_n0(node):
+            value = 1 if complemented else 0
+        elif BDDModel.is_terminal_n1(node):
+            value = 0 if complemented else 1
+        return value

flamapy/metamodels/bdd_metamodel/operations/bdd_product_distribution.py

@@ -1,58 +1,116 @@
-from typing import Optional
+import math
+from collections import defaultdict
+
+from dd.autoref import Function
 
-from flamapy.metamodels.configuration_metamodel.models.configuration import Configuration
 from flamapy.metamodels.bdd_metamodel.models import BDDModel
 from flamapy.metamodels.bdd_metamodel.operations.interfaces import ProductDistribution
-from flamapy.metamodels.bdd_metamodel.operations import BDDProducts
 
 
 class BDDProductDistribution(ProductDistribution):
-    """The Product Distribution (PD) algorithm determines the number of solutions
-    having a given number of variables.
-
-    This is a brute-force implementation that enumerates all solutions for accounting them.
 
-    Ref.: [Heradio et al. 2019. Supporting the Statistical Analysis of Variability Models. SPLC.
-    (https://doi.org/10.1109/ICSE.2019.00091)]
-    """
-
-    def __init__(self, partial_configuration: Optional[Configuration] = None) -> None:
+    def __init__(self) -> None:
         self.result: list[int] = []
         self.bdd_model = None
-        self.partial_configuration = partial_configuration
 
     def execute(self, model: BDDModel) -> 'BDDProductDistribution':
         self.bdd_model = model
-        self.result = product_distribution(self.bdd_model, self.partial_configuration)
+        self.result = product_distribution(self.bdd_model)
         return self
 
     def get_result(self) -> list[int]:
         return self.result
 
     def product_distribution(self) -> list[int]:
-        return product_distribution(self.bdd_model, self.partial_configuration)
+        return product_distribution(self.bdd_model)
+
 
-    def serialize(self, filepath: str) -> None:
-        result = self.get_result()
-        serialize(result, filepath)
+def product_distribution(bdd_model: BDDModel) -> list[int]:
+    """Computes the distribution of the number of activated features per product.
 
+    That is,
+        + How many products have 0 features activated?
+        + How many products have 1 feature activated?
+        + ...
+        + How many products have all features activated?
 
-def product_distribution(bdd_model: BDDModel,
-                         p_config: Optional[Configuration] = None) -> list[int]:
-    """It accounts for how many solutions have no variables, one variable,
-    two variables, ..., all variables.
+    For detailed information, see the paper: 
+        Heradio, R., Fernandez-Amoros, D., Mayr-Dorn, C., Egyed, A.:
+        Supporting the statistical analysis of variability models. 
+        In: 41st International Conference on Software Engineering (ICSE), pp. 843-853. 2019.
+        DOI: https://doi.org/10.1109/ICSE.2019.00091
 
-    It enumerates all solutions and filters them.
+    The operation returns a list that stores:
+        + In index 0, the number of products with 0 features activated.
+        + In index 1, the number of products with 1 feature activated.
+        ...
+        + In index n, the number of products with n features activated.
     """
-    products = BDDProducts(p_config).execute(bdd_model).get_result()
-    dist: list[int] = []
-    for i in range(len(bdd_model.variables) + 1):
-        dist.append(sum(len(p.elements) == i for p in products))
-    return dist
-
-
-def serialize(prod_dist: list[int], filepath: str) -> None:
-    with open(filepath, mode='w', encoding='utf8') as file:
-        file.write('Features, Products\n')
-        for features, products in enumerate(prod_dist):
-            file.write(f'{features}, {products}\n')
+    root = bdd_model.root
+    id_root = BDDModel.get_value(root, root.negated)
+    dist: dict[int, list[int]] = {0: [], 1: [1]}
+    mark: dict[int, bool] = defaultdict(bool)
+    get_prod_dist(root, dist, mark, root.negated)
+    return dist[id_root]
+
+
+def get_prod_dist(node: Function, 
+                  dist: dict[int, list[int]], 
+                  mark: dict[int, bool], 
+                  complemented: bool) -> None:
+    id_node = BDDModel.get_value(node, complemented)
+    mark[id_node] = not mark[id_node]
+
+    if not BDDModel.is_terminal_node(node):
+
+        # traverse
+        low = BDDModel.get_low_node(node)
+        id_low = BDDModel.get_value(low, complemented)
+
+        if mark[id_node] != mark[id_low]:
+            get_prod_dist(low, dist, mark, complemented ^ low.negated)
+
+        # compute low_dist to account for the removed nodes through low
+        removed_nodes = BDDModel.index(low) - BDDModel.index(node) - 1
+        low_dist = [0] * (removed_nodes + len(dist[id_low]))
+        for i in range(removed_nodes + 1):
+            for j in range(len(dist[id_low])):
+                low_dist[i + j] = low_dist[i + j] + dist[id_low][j] * math.comb(removed_nodes, i)
+
+        # traverse
+        high = BDDModel.get_high_node(node)
+        id_high = BDDModel.get_value(high, complemented)
+
+        high = BDDModel.get_high_node(node)
+        id_high = BDDModel.get_value(high, complemented)
+        if mark[id_node] != mark[id_high]:
+            get_prod_dist(high, dist, mark, complemented ^ high.negated)
+
+        # compute high_dist to account for the removed nodes through high
+        removed_nodes = BDDModel.index(high) - BDDModel.index(node) - 1
+        high_dist = [0] * (removed_nodes + len(dist[id_high]))
+        for i in range(removed_nodes + 1):
+            for j in range(len(dist[id_high])):
+                high_dist[i + j] = high_dist[i + j] + dist[id_high][j] * (
+                    math.comb(removed_nodes, i))
+        combine_distributions(id_node, dist, low_dist, high_dist)
+
+
+def combine_distributions(id_node: int, 
+                          dist: dict[int, list[int]], 
+                          low_dist: list[int], 
+                          high_dist: list[int]) -> None:
+    # combine low and high distributions
+    if len(low_dist) > len(high_dist):
+        #dist_length = len(dist[id_low])
+        dist_length = len(low_dist)
+    else:
+        #dist_length = len(dist[id_high]) + 1
+        dist_length = len(high_dist) + 1
+
+    node_dist = [0] * dist_length
+    for i, value in enumerate(low_dist):
+        node_dist[i] = value
+    for i, value in enumerate(high_dist):
+        node_dist[i + 1] = node_dist[i + 1] + value
+    dist[id_node] = node_dist

flamapy/metamodels/bdd_metamodel/transformations/sat_to_bdd.py

@@ -0,0 +1,35 @@
+from flamapy.core.transformations import ModelToModel
+
+from flamapy.metamodels.bdd_metamodel.models import BDDModel
+from flamapy.metamodels.pysat_metamodel.models import PySATModel
+
+
+class SATToBDD(ModelToModel):
+
+    @staticmethod
+    def get_source_extension() -> str:
+        return 'pysat'
+
+    @staticmethod
+    def get_destination_extension() -> str:
+        return 'bdd'
+
+    def __init__(self, source_model: PySATModel) -> None:
+        self.source_model = source_model
+        self.destination_model = BDDModel()
+
+    def transform(self) -> BDDModel:
+        # Transform clauses to textual CNF notation required by the BDD
+        not_connective = BDDModel.NOT
+        or_connective = ' ' + BDDModel.OR + ' '
+        and_connective = ' ' + BDDModel.AND + ' '
+        cnf_list = []
+        for clause in self.source_model.get_all_clauses().clauses:
+            cnf_list.append('(' + or_connective.join(list(map(lambda l:
+                            not_connective + self.source_model.features[abs(l)] if l < 0 else
+                            self.source_model.features[abs(l)], clause))) + ')')
+
+        cnf_formula = and_connective.join(cnf_list)
+        self.destination_model.from_textual_cnf(cnf_formula, list(self.source_model.variables.keys()))
+
+        return self.destination_model

setup.py

@@ -7,7 +7,7 @@
 
 setuptools.setup(
     name="flamapy-bdd",
-    version="1.0.0",
+    version="1.0.1",
     author="Flamapy",
     author_email="flamapy@us.es",
     description="bdd-plugin for the automated analysis of feature models",
@@ -22,8 +22,8 @@
     ],
     python_requires='>=3.9',
     install_requires=[
-        'flamapy~=1.0.0',
-        'flamapy-fm~=1.0.0',
+        'flamapy~=1.0.1',
+        'flamapy-fm~=1.0.1',
         'dd>=0.5.6'
         'graphviz~=0.20',
     ],
@@ -37,6 +37,6 @@
         ]
     },
     dependency_links=[
-        'flamapy~=1.0.0',
+        'flamapy~=1.0.1',
     ]
 )