refactor indexed grammar intersection

Author: bygu4

pyformlang/fst/fst.py

@@ -7,10 +7,6 @@
 from networkx import MultiDiGraph
 from networkx.drawing.nx_pydot import write_dot
 
-from pyformlang.indexed_grammar import IndexedGrammar, Rules, \
-    DuplicationRule, ProductionRule, EndRule, ConsumptionRule
-from pyformlang.indexed_grammar.reduced_rule import ReducedRule
-
 from .utils import StateRenaming
 from ..objects.finite_automaton_objects import State, Symbol, Epsilon
 from ..objects.finite_automaton_objects.utils import to_state, to_symbol
@@ -216,7 +212,7 @@ def translate(self,
             The translation of the input word
         """
         # (remaining in the input, generated so far, current_state)
-        input_word = [to_symbol(symbol) for symbol in input_word]
+        input_word = [to_symbol(x) for x in input_word if x != Epsilon()]
         to_process: List[Tuple[List[Symbol], List[Symbol], State]] = []
         seen_by_state = {state: [] for state in self.states}
         for start_state in self._start_states:
@@ -244,134 +240,6 @@ def translate(self,
                                        generated + list(output_symbols),
                                        next_state))
 
-    def intersection(self, indexed_grammar: IndexedGrammar) -> IndexedGrammar:
-        """ Compute the intersection with an other object
-
-        Equivalent to:
-          >> fst and indexed_grammar
-        """
-        rules = indexed_grammar.rules
-        new_rules: List[ReducedRule] = [EndRule("T", str(Epsilon()))]
-        self._extract_consumption_rules_intersection(rules, new_rules)
-        self._extract_indexed_grammar_rules_intersection(rules, new_rules)
-        self._extract_terminals_intersection(rules, new_rules)
-        self._extract_epsilon_transitions_intersection(new_rules)
-        self._extract_fst_delta_intersection(new_rules)
-        self._extract_fst_epsilon_intersection(new_rules)
-        self._extract_fst_duplication_rules_intersection(new_rules)
-        rules = Rules(new_rules, rules.optim)
-        return IndexedGrammar(rules).remove_useless_rules()
-
-    def _extract_fst_duplication_rules_intersection(
-            self,
-            new_rules: List[ReducedRule]) \
-                -> None:
-        for state_p in self._final_states:
-            for start_state in self._start_states:
-                new_rules.append(DuplicationRule(
-                    "S",
-                    str((start_state, "S", state_p)),
-                    "T"))
-
-    def _extract_fst_epsilon_intersection(
-            self,
-            new_rules: List[ReducedRule]) \
-                -> None:
-        for state_p in self._states:
-            new_rules.append(EndRule(
-                str((state_p, Epsilon(), state_p)), str(Epsilon())))
-
-    def _extract_fst_delta_intersection(
-            self,
-            new_rules:List[ReducedRule]) \
-                -> None:
-        for key, pair in self._delta.items():
-            state_p = key[0]
-            terminal = key[1]
-            for transition in pair:
-                state_q = transition[0]
-                symbol = transition[1]
-                new_rules.append(EndRule(str((state_p, terminal, state_q)),
-                                         symbol))
-
-    def _extract_epsilon_transitions_intersection(
-            self,
-            new_rules: List[ReducedRule]) \
-                -> None:
-        for state_p in self._states:
-            for state_q in self._states:
-                for state_r in self._states:
-                    new_rules.append(DuplicationRule(
-                        str((state_p, Epsilon(), state_q)),
-                        str((state_p, Epsilon(), state_r)),
-                        str((state_r, Epsilon(), state_q))))
-
-    def _extract_indexed_grammar_rules_intersection(
-            self,
-            rules: Rules,
-            new_rules: List[ReducedRule]) \
-                -> None:
-        for rule in rules.rules:
-            if isinstance(rule, DuplicationRule):
-                for state_p in self._states:
-                    for state_q in self._states:
-                        for state_r in self._states:
-                            new_rules.append(DuplicationRule(
-                                str((state_p, rule.left_term, state_q)),
-                                str((state_p, rule.right_terms[0], state_r)),
-                                str((state_r, rule.right_terms[1], state_q))))
-            elif isinstance(rule, ProductionRule):
-                for state_p in self._states:
-                    for state_q in self._states:
-                        new_rules.append(ProductionRule(
-                            str((state_p, rule.left_term, state_q)),
-                            str((state_p, rule.right_term, state_q)),
-                            str(rule.production)))
-            elif isinstance(rule, EndRule):
-                for state_p in self._states:
-                    for state_q in self._states:
-                        new_rules.append(DuplicationRule(
-                            str((state_p, rule.left_term, state_q)),
-                            str((state_p, rule.right_term, state_q)),
-                            "T"))
-
-    def _extract_terminals_intersection(
-            self,
-            rules: Rules,
-            new_rules: List[ReducedRule]) \
-                -> None:
-        terminals = rules.terminals
-        for terminal in terminals:
-            for state_p in self._states:
-                for state_q in self._states:
-                    for state_r in self._states:
-                        new_rules.append(DuplicationRule(
-                            str((state_p, terminal, state_q)),
-                            str((state_p, Epsilon(), state_r)),
-                            str((state_r, terminal, state_q))))
-                        new_rules.append(DuplicationRule(
-                            str((state_p, terminal, state_q)),
-                            str((state_p, terminal, state_r)),
-                            str((state_r, Epsilon(), state_q))))
-
-    def _extract_consumption_rules_intersection(
-            self,
-            rules: Rules,
-            new_rules: List[ReducedRule]) \
-                -> None:
-        consumptions = rules.consumption_rules
-        for consumption_rule in consumptions:
-            for consumption in consumptions[consumption_rule]:
-                for state_r in self._states:
-                    for state_s in self._states:
-                        new_rules.append(ConsumptionRule(
-                            consumption.f_parameter,
-                            str((state_r, consumption.left_term, state_s)),
-                            str((state_r, consumption.right_term, state_s))))
-
-    def __and__(self, other: IndexedGrammar) -> IndexedGrammar:
-        return self.intersection(other)
-
     def union(self, other_fst: "FST") -> "FST":
         """
         Makes the union of two fst

pyformlang/fst/tests/test_fst.py

@@ -5,9 +5,6 @@
 import pytest
 
 from pyformlang.fst import FST
-from pyformlang.indexed_grammar import (
-    DuplicationRule, ProductionRule, EndRule,
-    ConsumptionRule, IndexedGrammar, Rules)
 
 
 @pytest.fixture
@@ -94,34 +91,6 @@ def test_translate(self):
         assert ["b", "c"] in translation
         assert ["b"] + ["c"] * 9 in translation
 
-    def test_intersection_indexed_grammar(self):
-        """ Test the intersection with indexed grammar """
-        l_rules = []
-        rules = Rules(l_rules)
-        indexed_grammar = IndexedGrammar(rules)
-        fst = FST()
-        intersection = fst & indexed_grammar
-        assert intersection.is_empty()
-
-        l_rules.append(ProductionRule("S", "D", "f"))
-        l_rules.append(DuplicationRule("D", "A", "B"))
-        l_rules.append(ConsumptionRule("f", "A", "Afinal"))
-        l_rules.append(ConsumptionRule("f", "B", "Bfinal"))
-        l_rules.append(EndRule("Afinal", "a"))
-        l_rules.append(EndRule("Bfinal", "b"))
-
-        rules = Rules(l_rules)
-        indexed_grammar = IndexedGrammar(rules)
-        intersection = fst.intersection(indexed_grammar)
-        assert intersection.is_empty()
-
-        fst.add_start_state("q0")
-        fst.add_final_state("final")
-        fst.add_transition("q0", "a", "q1", ["a"])
-        fst.add_transition("q1", "b", "final", ["b"])
-        intersection = fst.intersection(indexed_grammar)
-        assert not intersection.is_empty()
-
     def test_union(self, fst0, fst1):
         """ Tests the union"""
         fst_union = fst0.union(fst1)

pyformlang/indexed_grammar/indexed_grammar.py

@@ -4,15 +4,16 @@
 
 # pylint: disable=cell-var-from-loop
 
-from typing import Dict, List, Set, FrozenSet, Tuple, Hashable, Any
+from typing import Dict, List, Set, FrozenSet, Tuple, Hashable
 
 from pyformlang.cfg import CFGObject, Variable, Terminal
-from pyformlang.finite_automaton import FiniteAutomaton
-from pyformlang.regular_expression import Regex
+from pyformlang.fst import FST
 
 from .rules import Rules
+from .reduced_rule import ReducedRule
 from .duplication_rule import DuplicationRule
 from .production_rule import ProductionRule
+from .consumption_rule import ConsumptionRule
 from .end_rule import EndRule
 from .utils import exists, addrec_bis
 from ..objects.cfg_objects.utils import to_variable
@@ -47,7 +48,7 @@ def __init__(self,
         # Mark all end symbols
         for non_terminal_a in non_terminals:
             if exists(self._rules.rules,
-                      lambda x: x.is_end_rule()
+                      lambda x: isinstance(x, EndRule)
                       and x.left_term == non_terminal_a):
                 self._marked[non_terminal_a].add(frozenset())
 
@@ -363,7 +364,7 @@ def remove_useless_rules(self) -> "IndexedGrammar":
         rules = Rules(l_rules, self._rules.optim)
         return IndexedGrammar(rules)
 
-    def intersection(self, other: Any) -> "IndexedGrammar":
+    def intersection(self, other: FST) -> "IndexedGrammar":
         """ Computes the intersection of the current indexed grammar with the \
         other object
 
@@ -387,14 +388,18 @@ def intersection(self, other: Any) -> "IndexedGrammar":
             When trying to intersection with something else than a regular
             expression or a finite automaton
         """
-        if isinstance(other, Regex):
-            other = other.to_epsilon_nfa()
-        if isinstance(other, FiniteAutomaton):
-            fst = other.to_fst()
-            return fst.intersection(self)
-        raise NotImplementedError
-
-    def __and__(self, other: Any) -> "IndexedGrammar":
+        new_rules: List[ReducedRule] = [EndRule("T", "epsilon")]
+        self._extract_consumption_rules_intersection(other, new_rules)
+        self._extract_indexed_grammar_rules_intersection(other, new_rules)
+        self._extract_terminals_intersection(other, new_rules)
+        self._extract_epsilon_transitions_intersection(other, new_rules)
+        self._extract_fst_delta_intersection(other, new_rules)
+        self._extract_fst_epsilon_intersection(other, new_rules)
+        self._extract_fst_duplication_rules_intersection(other, new_rules)
+        rules = Rules(new_rules, self.rules.optim)
+        return IndexedGrammar(rules).remove_useless_rules()
+
+    def __and__(self, other: FST) -> "IndexedGrammar":
         """ Computes the intersection of the current indexed grammar with the
         other object
 
@@ -409,3 +414,110 @@ def __and__(self, other: Any) -> "IndexedGrammar":
             The indexed grammar which useless rules
         """
         return self.intersection(other)
+
+    def _extract_fst_duplication_rules_intersection(
+            self,
+            other: FST,
+            new_rules: List[ReducedRule]) \
+                -> None:
+        for final_state in other.final_states:
+            for start_state in other.start_states:
+                new_rules.append(DuplicationRule(
+                    "S",
+                    (start_state, "S", final_state),
+                    "T"))
+
+    def _extract_fst_epsilon_intersection(
+            self,
+            other: FST,
+            new_rules: List[ReducedRule]) \
+                -> None:
+        for state in other.states:
+            new_rules.append(EndRule(
+                (state, "epsilon", state),
+                "epsilon"))
+
+    def _extract_fst_delta_intersection(
+            self,
+            other: FST,
+            new_rules: List[ReducedRule]) \
+                -> None:
+        for (s_from, symb_from), (s_to, symb_to) in other:
+            new_rules.append(EndRule(
+                (s_from, symb_from, s_to),
+                symb_to))
+
+    def _extract_epsilon_transitions_intersection(
+            self,
+            other: FST,
+            new_rules: List[ReducedRule]) \
+                -> None:
+        for state_p in other.states:
+            for state_q in other.states:
+                for state_r in other.states:
+                    new_rules.append(DuplicationRule(
+                        (state_p, "epsilon", state_q),
+                        (state_p, "epsilon", state_r),
+                        (state_r, "epsilon", state_q)))
+
+    def _extract_indexed_grammar_rules_intersection(
+            self,
+            other: FST,
+            new_rules: List[ReducedRule]) \
+                -> None:
+        for rule in self.rules.rules:
+            if isinstance(rule, DuplicationRule):
+                for state_p in other.states:
+                    for state_q in other.states:
+                        for state_r in other.states:
+                            new_rules.append(DuplicationRule(
+                                (state_p, rule.left_term, state_q),
+                                (state_p, rule.right_terms[0], state_r),
+                                (state_r, rule.right_terms[1], state_q)))
+            elif isinstance(rule, ProductionRule):
+                for state_p in other.states:
+                    for state_q in other.states:
+                        new_rules.append(ProductionRule(
+                            (state_p, rule.left_term, state_q),
+                            (state_p, rule.right_term, state_q),
+                            rule.production))
+            elif isinstance(rule, EndRule):
+                for state_p in other.states:
+                    for state_q in other.states:
+                        new_rules.append(DuplicationRule(
+                            (state_p, rule.left_term, state_q),
+                            (state_p, rule.right_term, state_q),
+                            "T"))
+
+    def _extract_terminals_intersection(
+            self,
+            other: FST,
+            new_rules: List[ReducedRule]) \
+                -> None:
+        for terminal in self.rules.terminals:
+            for state_p in other.states:
+                for state_q in other.states:
+                    for state_r in other.states:
+                        new_rules.append(DuplicationRule(
+                            (state_p, terminal, state_q),
+                            (state_p, "epsilon", state_r),
+                            (state_r, terminal, state_q)))
+                        new_rules.append(DuplicationRule(
+                            (state_p, terminal, state_q),
+                            (state_p, terminal, state_r),
+                            (state_r, "epsilon", state_q)))
+
+    def _extract_consumption_rules_intersection(
+            self,
+            other: FST,
+            new_rules: List[ReducedRule]) \
+                -> None:
+        consumptions = self.rules.consumption_rules
+        for terminal in consumptions:
+            for consumption in consumptions[terminal]:
+                for state_r in other.states:
+                    for state_s in other.states:
+                        new_rules.append(ConsumptionRule(
+                            consumption.f_parameter,
+                            (state_r, consumption.left_term, state_s),
+                            (state_r, consumption.right_term, state_s)))