Add is_commutative and is_distributive

b_asic/signal_flow_graph.py

@@ -2111,6 +2111,153 @@ class SFG(AbstractOperation):
     def is_constant(self) -> bool:
         return all(output.is_constant for output in self._output_operations)
 
+    @property
+    def is_commutative(self) -> bool:
+        """
+        Checks if all operations in the SFG are commutative.
+
+        An operation is considered commutative if it is not in B-ASIC Special Operations Module,
+        and its `is_commutative` property is `True`.
+
+        Returns
+        -------
+        bool: `True` if all operations are commutative, `False` otherwise.
+        """
+        return all(
+            (
+                op.is_commutative
+                if op.type_name() not in ["in", "out", "t", "c"]
+                else True
+            )
+            for op in self.split()
+        )
+
+    @property
+    def is_distributive(self) -> bool:
+        """
+        Checks if the SFG is distributive.
+
+        An operation is considered distributive if it can be applied to each element of a set separately.
+        For example, multiplication is distributive over addition, meaning that `a * (b + c)` is equivalent to `a * b + a * c`.
+
+        Returns
+        -------
+        bool: True if the SFG is distributive, False otherwise.
+
+        Examples
+        --------
+            >>> Mad_op = MAD(Input(), Input(), Input()) # Creates an instance of the Mad operation, MAD is defined in b_asic.core_operations
+            >>> Mad_sfg = Mad_op.to_sfg()  # The operation is turned into a sfg
+            >>> Mad_sfg.is_distributive    # True  # if the distributive property holds, False otherwise
+
+        """
+        structures = []
+        operations = self.get_operations_topological_order()
+        for op in operations:
+            if not (
+                op.type_name() == "in"
+                or op.type_name() == "out"
+                or op.type_name() == "c"
+                or op.type_name() == "t"
+            ):
+                structures.append(op)
+        return (
+            all(self.has_distributive_structure(op) for op in structures)
+            if len(structures) > 1
+            else False
+        )
+
+    def has_distributive_structure(self, op: Operation) -> bool:
+        """
+        Checks if the SFG contains distributive structures.
+        NOTE: a*b + c = a*(b + c/a) is considered distributive. Meaning that an algorithm transformation would require an additionat operation.
+
+        Parameters:
+        ----------
+        op : Operation
+            The operation that is the start of the structure to check for distributivity.
+
+        Returns:
+        -------
+            bool: True if a distributive structures is found, False otherwise.
+        """
+        # TODO Butterfly and SymmetricTwoportAdaptor needs to be converted to a SF using to_sfg() in order to be evaluated
+        if op.type_name() == 'mac':
+            return True
+        elif op.type_name() in ['mul', 'div']:
+            for subsequent_op in op.subsequent_operations:
+                if subsequent_op.type_name() in [
+                    'add',
+                    'sub',
+                    'addsub',
+                    'min',
+                    'max',
+                    'sqrt',
+                    'abs',
+                    'rec',
+                    'out',
+                    't',
+                ]:
+                    return True
+                elif subsequent_op.type_name() in ['mul', 'div']:
+                    for subsequent_op in subsequent_op.subsequent_operations:
+                        return self.has_distributive_structure(subsequent_op)
+                else:
+                    return False
+        elif op.type_name() in ['cmul', 'shift', 'rshift', 'lshift']:
+            for subsequent_op in op.subsequent_operations:
+                if subsequent_op.type_name() in [
+                    'add',
+                    'sub',
+                    'addsub',
+                    'min',
+                    'max',
+                    'out',
+                    't',
+                ]:
+                    return True
+                elif subsequent_op.type_name() in ['cmul', 'shift', 'rshift', 'lshift']:
+                    for subsequent_op in subsequent_op.subsequent_operations:
+                        return self.has_distributive_structure(subsequent_op)
+                else:
+                    return False
+        elif op.type_name() in ['add', 'sub', 'addsub']:
+            for subsequent_op in op.subsequent_operations:
+                if subsequent_op.type_name() in [
+                    'mul',
+                    'div',
+                    'min',
+                    'max',
+                    'out',
+                    'cmul',
+                    't',
+                ]:
+                    return True
+                elif subsequent_op.type_name() in ['add', 'sub', 'addsub']:
+                    for subsequent_op in subsequent_op.subsequent_operations:
+                        return self.has_distributive_structure(subsequent_op)
+                else:
+                    return False
+        elif op.type_name() in ['min', 'max']:
+            for subsequent_op in op.subsequent_operations:
+                if subsequent_op.type_name() in [
+                    'add',
+                    'sub',
+                    'addsub',
+                    'mul',
+                    'div',
+                    'cmul',
+                    'out',
+                    't',
+                ]:
+                    return True
+                elif subsequent_op.type_name() in ['min', 'max']:
+                    for subsequent_op in subsequent_op.subsequent_operations:
+                        return self.has_distributive_structure(subsequent_op)
+                else:
+                    return False
+        return False
+
     def get_used_type_names(self) -> List[TypeName]:
         """Get a list of all TypeNames used in the SFG."""
         ret = list({op.type_name() for op in self.operations})