diff --git a/docs/automata/atomaton.rst b/docs/automata/atomaton.rst index e738518..1f5b3b0 100644 --- a/docs/automata/atomaton.rst +++ b/docs/automata/atomaton.rst @@ -36,6 +36,17 @@ the special case where the reverse is deterministic. atomic_states(nfa) # states whose right language is a union of atoms is_atomic(nfa.reverse()) # iff nfa.determinize() is minimal +Maximized prime átomaton +======================== + +The maximized prime átomaton (:class:`MaximizedPrimeAtomaton`) is the dual of +the canonical RFSA :cite:`MaarandTamm2022`: the reverse of the canonical RFSA of +the reversed language, just as the átomaton is the reverse of the minimal DFA of +the reversed language. Its states are the maximized prime atoms, and the right +language of each lies between its atom and its maximized atom :cite:`Tamm2015`. +Unlike the átomaton it need not be atomic, so it is a plain +:class:`~sofic.automata.nfa.NFA` subclass. + API === @@ -45,3 +56,6 @@ API .. autoclass:: AtomicAutomaton .. autoclass:: Atomaton .. autoclass:: MaximizedPrimeAtomaton + :members: from_language, from_canonical_rfsa, dual + +.. autofunction:: sofic.automata.canonical_extraction.maximized_prime_atomaton_from_language diff --git a/docs/automata/learning.rst b/docs/automata/learning.rst index 4c53400..18b879c 100644 --- a/docs/automata/learning.rst +++ b/docs/automata/learning.rst @@ -10,11 +10,21 @@ Learning Active learning (NL\*) ====================== -Active learning of maximized prime átomatons via NL\* with a membership -teacher, following Angluin-style learning and its nondeterministic extension -:cite:`Angluin1987,Bollig2009`: - -.. autofunction:: sofic.automata.learning.learn_maximized_prime_atomaton +NL\* :cite:`Bollig2009` extends Angluin's L\* :cite:`Angluin1987` to +nondeterministic automata. It keeps an RFSA-closed, RFSA-consistent observation +table whose prime rows become the hypothesis states, and adds every suffix of a +counterexample as a new experiment. When the equivalence oracle accepts, the +hypothesis is the canonical RFSA of the target (:doc:`rfsa`). Running NL\* on +the reversed target and reversing the result learns the maximized prime +átomaton (:doc:`atomaton`). + +:class:`~sofic.automata.active.AutomatonEquivalenceOracle` answers equivalence +queries exactly against a target automaton, returning a shortest +counterexample. + +.. autofunction:: sofic.automata.learning.learn_rfsa_nlstar +.. autofunction:: sofic.automata.learning.learn_prime_atomaton_nlstar +.. autofunction:: sofic.automata.learning.learn_rfsa_from_language Active learning (L\*, TTT, Mealy) ================================= @@ -65,6 +75,7 @@ equivalence test: :members: .. autoclass:: sofic.automata.active.LanguageMembershipOracle .. autoclass:: sofic.automata.active.FunctionMembershipOracle +.. autoclass:: sofic.automata.active.AutomatonEquivalenceOracle .. autoclass:: sofic.automata.active.ExhaustiveEquivalenceOracle .. autoclass:: sofic.automata.active.RandomWalkEquivalenceOracle .. autoclass:: sofic.automata.active.TransducerOutputOracle diff --git a/docs/automata/nwa.rst b/docs/automata/nwa.rst index 2fabf4e..6b060da 100644 --- a/docs/automata/nwa.rst +++ b/docs/automata/nwa.rst @@ -53,6 +53,11 @@ visible roles. Use :meth:`NestedWordAutomaton.to_vpa` to encode an NWA as a VPA; by default symbols are tagged with their role so overlapping NWA alphabets still become a disjoint visible alphabet. +The closure operations (``union``, ``intersection``, ``complement``, +``difference``, ``concat``, ``kleene_star``) and decision procedures +(``is_empty``, ``is_universal``, ``includes``, ``equivalent``) run on the tagged +VPA encoding (see :doc:`vpa`) and are translated back to an NWA. + API === @@ -60,6 +65,8 @@ API :members: from_visible_word, validate .. autoclass:: NestedWordAutomaton - :members: add_call_transition, add_return_transition, add_internal_transition, recognizes, recognizes_visible, from_vpa, to_vpa + :members: add_call_transition, add_return_transition, add_internal_transition, recognizes, recognizes_visible, + from_vpa, to_vpa, union, intersection, complement, difference, concat, kleene_star, is_empty, + is_universal, includes, equivalent .. autofunction:: sofic.automata.nwa_simulation.recognizes_nwa diff --git a/docs/automata/observation_table.rst b/docs/automata/observation_table.rst index a384074..0e7deed 100644 --- a/docs/automata/observation_table.rst +++ b/docs/automata/observation_table.rst @@ -17,4 +17,5 @@ API .. autofunction:: sofic.automata.canonical_extraction.observation_to_canonical_rfsa .. autofunction:: sofic.automata.canonical_extraction.observation_to_atomaton +.. autofunction:: sofic.automata.canonical_extraction.observation_to_maximized_prime_atomaton .. autofunction:: sofic.automata.canonical_extraction.observation_to_minimal_dfa diff --git a/docs/automata/rfsa.rst b/docs/automata/rfsa.rst index ea30abd..bd21737 100644 --- a/docs/automata/rfsa.rst +++ b/docs/automata/rfsa.rst @@ -5,15 +5,39 @@ RFSA *** -Residual finite state automata (:class:`ResidualFiniteStateAutomaton`) and their -canonical form (:class:`CanonicalRFSA`) follow the residual-language theory of -Denis, Lemay, and Terlutte :cite:`Denis2002`. The extraction helpers documented -here expose RFSA-oriented entry points without claiming a fully minimized RFSA -pipeline beyond the implemented automata-backed construction. +Residual finite state automata (:class:`ResidualFiniteStateAutomaton`) are NFAs +whose every state accepts a residual (left quotient) of the language +:cite:`Denis2002`. :meth:`ResidualFiniteStateAutomaton.validate` checks this +exactly against the minimal DFA. + +The canonical RFSA (:class:`CanonicalRFSA`) has one state per *prime* +residual -- a non-empty residual that is not the union of the residuals strictly +inside it -- with initial states the primes contained in the language, accepting +states the primes containing the empty word, and a transition +:math:`p \xrightarrow{a} p'` whenever :math:`L_{p'} \subseteq a^{-1} L_p` +:cite:`Denis2002`. It is never larger than the minimal DFA and can be +exponentially smaller: for :math:`\Sigma^* a \Sigma^n` the minimal DFA has +:math:`2^{n+1}` states and the canonical RFSA :math:`n + 2`. + +Reversing a canonical RFSA gives the maximized prime átomaton of the reversed +language (:meth:`CanonicalRFSA.dual`; see :doc:`atomaton`). NL\* learns the +canonical RFSA from queries (:doc:`learning`). + +.. code-block:: python + + from sofic.automata.rfsa import CanonicalRFSA + + rfsa = CanonicalRFSA.from_language(nfa) + rfsa.validate() # every state accepts a residual + rfsa.dual() # maximized prime átomaton of the reverse API === .. autoclass:: ResidualFiniteStateAutomaton .. autoclass:: CanonicalRFSA - :members: from_language, from_observation_table + :members: from_language, from_observation_table, dual + +.. autofunction:: sofic.automata.canonical_extraction.canonical_rfsa_from_language +.. autoclass:: sofic.automata.canonical_extraction.ResidualTable + :members: includes, is_covered, prime_states diff --git a/docs/automata/vpa.rst b/docs/automata/vpa.rst index 35b54ce..ae6c36d 100644 --- a/docs/automata/vpa.rst +++ b/docs/automata/vpa.rst @@ -6,107 +6,116 @@ Visibly Pushdown Automata ************************* :class:`VisiblyPushdownAutomaton` partitions the alphabet into call, return, -and internal symbols. Call transitions push a stack symbol, return transitions -may either be guarded by a stack symbol or left unguarded as a wildcard over -ordinary stack entries, and internal transitions leave the stack untouched. -The model and its nested-word connection follow Alur and Madhusudan -:cite:`AlurMadhusudan2009`. +and internal symbols :cite:`AlurMadhusudan2009`. Call transitions push a stack +symbol, internal transitions leave the stack alone, and return transitions pop +it. A return guarded by a stack symbol fires only when that symbol is on top; a +wildcard return (no stack symbol) fires on every stack symbol, and also on the +empty stack when the VPA has a ``bottom_stack_symbol``. A return on the empty +stack is a *pending return*; it is possible only with a bottom symbol and leaves +the stack empty. Acceptance is by final state, so words may end with *pending +calls* still on the stack. + +Operations and decisions +======================== + +Every closure operation returns a concrete automaton: + +* ``union`` (disjoint sum) and ``intersection`` (synchronized product); +* ``determinize`` -- the summary construction of :cite:`AlurMadhusudan2009`, + whose states pair a summary relation with the set of current states. The + result is a complete :class:`DeterministicVisiblyPushdownAutomaton`, and + :meth:`DeterministicVisiblyPushdownAutomaton.from_vpa` uses it whenever its + input is nondeterministic; +* ``complement`` (determinize, then flip accepting states) and ``difference``; +* ``concat`` and ``kleene_star``. Each factor is read from an empty stack of its + own: the finite control records whether the current factor's stack is empty + and pushes that bit with every symbol, so a return that would pop a pending + call of an earlier factor counts as a pending return of the current one. + +Emptiness is decided by saturating the relation of well-matched summaries and +then searching states reachable with pending calls or pending returns; +``accepted_word`` returns a witness. ``is_universal``, ``includes``, +``equivalent``, and ``has_unmatched_word`` build on it. +:class:`~sofic.automata.nwa.NestedWordAutomaton` exposes the same operations by +delegating through :meth:`~sofic.automata.nwa.NestedWordAutomaton.to_vpa`. -Canonical forms -=============== +.. code-block:: python + + balanced.union(other).equivalent(other.union(balanced)) # True + balanced.complement().complement().equivalent(balanced) # True + balanced.concat(balanced).accepted_word() # e.g. ('(', ')') + +Canonical and modular forms +=========================== + +General visibly pushdown languages have no unique minimal deterministic VPA, and +exact unrestricted minimization is NP-complete :cite:`Gauwin2020`. Canonical +forms exist for well-matched languages, or once calls are assigned to modules +:cite:`AlurKumarMadhusudanViswanathan2005`. -The VPA module includes four deterministic canonical forms: +``CanonicalVisiblyPushdownAutomaton`` + The Myhill-Nerode canonical deterministic VPA of a well-matched language. + Its states are classes of the finite algebra of well-matched summaries, + refined jointly for top-level contexts and for contexts inside a pending + call. With an empty call alphabet it is the minimal DFA. Languages with a + pending call or return raise + :exc:`~sofic.exceptions.NonWellMatchedLanguageError`. ``SingleEntryVisiblyPushdownAutomaton`` A k-module SEVPA. Modules partition the states, calls are assigned to - modules by ``call_partition``, every non-base module has one entry state in - ``entry_states``, and every call pushes ``(caller_state, call_symbol)``. + modules by ``call_partition``, every non-base module has one entry state, and + every call pushes ``(caller_state, call_symbol)``. ``MultipleEntryVisiblyPushdownAutomaton`` - A k-module MEVPA. Calls are still assigned to modules, but a module may have - several entries. The pushed call stack symbol must depend only on the source - state. + A k-module MEVPA. A module may have several entries, and the pushed symbol + depends only on the caller state. ``CallDrivenAutomaton`` - A CDA, used here as the shared modular generalization. The target of a call - transition is determined by the call symbol, independent of the source state. - -``CanonicalVisiblyPushdownAutomaton`` - The Myhill-Nerode canonical deterministic VPA. It is constructed from the - finite algebra of well-matched summaries induced by a deterministic VPA and - quotiented by finite right-context acceptance signatures. If the call - alphabet is empty, this construction specializes to the usual minimal DFA - right congruence. - -The modular ``minimize`` constructors require deterministic input and fixed -module/call metadata. They intentionally do not attempt arbitrary VPA -minimization: visibly pushdown automata do not have unique minimum recognizers -in general, and exact unrestricted minimization is NP-complete. - -Operations -========== - -Finite automata expose the usual regular operations directly on ``DFA`` and -``NFA`` instances: ``union``, ``intersection``/``intersect``, ``complement``, -``difference``, ``concat``/``concatenate``, and ``kleene_star``/``star``. - -VPAs expose the same operation names. These return -``CompositeVisiblyPushdownAutomaton`` instances, which are exact VPA language -expressions with a ``recognizes`` method. This keeps concatenation and Kleene -star correct even when an operand accepts with pending stack content; concrete -graph normalization for those composite VPAs is intentionally left separate -from the operation API. - -Constructor sketch -================== + The shared modular generalization: a call's target depends only on the call + symbol. + +:func:`~sofic.automata.vpa_constructions.to_single_entry` and +:func:`~sofic.automata.vpa_constructions.to_multiple_entry` convert any VPA of a +well-matched language into these forms (default: one module per call symbol). +On a call the state resets to the module's entry and the caller is pushed, so +the automaton forgets its caller; that is why pending calls -- and hence +non-well-matched languages, which raise +:exc:`~sofic.exceptions.NonWellMatchedLanguageError` -- are out of scope. The +modular ``minimize`` constructors call these conversions when no modules are +given, and otherwise quotient the supplied module structure. .. code-block:: python + SingleEntryVisiblyPushdownAutomaton.minimize(vpa) # convert, then minimize SingleEntryVisiblyPushdownAutomaton.minimize( - vpa, - call_partition={"call": "module"}, - modules={"main": {"q0"}, "module": {"entry", "body"}}, - entry_states={"module": "entry"}, - ) - - MultipleEntryVisiblyPushdownAutomaton.minimize( - vpa, - modules={"main": {"q0"}, "module": {"entry0", "entry1"}}, - call_partition={"call0": "module", "call1": "module"}, - ) - - CallDrivenAutomaton.minimize( - vpa, - modules={"main": {"q0"}, "module": {"entry"}}, - call_partition={"call": "module"}, + sevpa, call_partition={"call": "module"}, modules=sevpa.modules, ) - CanonicalVisiblyPushdownAutomaton.from_vpa(vpa) -References -========== - -The summary and SEVPA constructions follow Alur, Kumar, Madhusudan, and -Viswanathan :cite:`AlurKumarMadhusudanViswanathan2005`. The unrestricted -minimization limitation follows Gauwin, Muscholl, and Raskin -:cite:`Gauwin2020`. - API === .. autoclass:: VisiblyPushdownAutomaton - :members: union, intersection, intersect, complement, difference, concat, concatenate, kleene_star, star + :members: union, intersection, complement, difference, concat, kleene_star, determinize, is_empty, + accepted_word, is_universal, includes, equivalent, has_unmatched_word .. autoclass:: DeterministicVisiblyPushdownAutomaton - -.. autoclass:: CompositeVisiblyPushdownAutomaton + :members: from_vpa .. autoclass:: SingleEntryVisiblyPushdownAutomaton + :members: minimize .. autoclass:: MultipleEntryVisiblyPushdownAutomaton + :members: minimize .. autoclass:: CallDrivenAutomaton + :members: minimize .. autoclass:: CanonicalVisiblyPushdownAutomaton + :members: from_vpa + +.. automodule:: sofic.automata.vpa_constructions + :members: normalize, determinize, complement, concat, kleene_star, well_matched_summaries, accepted_word, + has_unmatched_word, to_single_entry, to_multiple_entry .. autofunction:: sofic.automata.vpa_simulation.recognizes_vpa diff --git a/docs/references.bib b/docs/references.bib index afbdd2f..897990b 100644 --- a/docs/references.bib +++ b/docs/references.bib @@ -158,6 +158,28 @@ @article{Denis2002 doi = {10.3233/FUN-2002-51402}, } +@inproceedings{Tamm2015, + author = {Tamm, Hellis}, + title = {Generalization of the Double-Reversal Method of Finding a Canonical Residual Finite State Automaton}, + booktitle = {Descriptional Complexity of Formal Systems}, + series = {Lecture Notes in Computer Science}, + volume = {9118}, + pages = {268--279}, + publisher = {Springer}, + year = {2015}, + doi = {10.1007/978-3-319-19225-3_23}, +} + +@inproceedings{MaarandTamm2022, + author = {Maarand, Hendrik and Tamm, Hellis}, + title = {Yet Another Canonical Nondeterministic Automaton}, + booktitle = {Descriptional Complexity of Formal Systems}, + series = {Lecture Notes in Computer Science}, + publisher = {Springer}, + year = {2022}, + doi = {10.1007/978-3-031-13257-5_14}, +} + @inproceedings{BrzozowskiTamm2011, author = {Brzozowski, Janusz A. and Tamm, Hellis}, title = {Theory of {\'A}tomata}, diff --git a/sofic/automata/__init__.py b/sofic/automata/__init__.py index 720bf4e..4ed99ff 100644 --- a/sofic/automata/__init__.py +++ b/sofic/automata/__init__.py @@ -2,6 +2,7 @@ # ``atomaton`` is an intentional pun on atomic automaton. from sofic.automata.active import ( + AutomatonEquivalenceOracle, EquivalenceOracle, ExhaustiveEquivalenceOracle, FunctionMealyOracle, @@ -62,6 +63,7 @@ validate_idfa_string, ) from sofic.automata.languages import AutomatonLanguage, RegularLanguage +from sofic.automata.learning import learn_prime_atomaton_nlstar, learn_rfsa_from_language, learn_rfsa_nlstar from sofic.automata.nfa import NFA from sofic.automata.nwa import NestedWord, NestedWordAutomaton from sofic.automata.observation import ObservationTable @@ -94,17 +96,10 @@ from sofic.automata.vpa import ( CallDrivenAutomaton, CanonicalVisiblyPushdownAutomaton, - CompositeVisiblyPushdownAutomaton, DeterministicVisiblyPushdownAutomaton, MultipleEntryVisiblyPushdownAutomaton, SingleEntryVisiblyPushdownAutomaton, VisiblyPushdownAutomaton, - complement_vpa, - concat_vpa, - difference_vpa, - intersection_vpa, - kleene_star_vpa, - union_vpa, ) from sofic.automata.wheeler import ( WheelerError, @@ -125,6 +120,7 @@ __all__ = [ "Atomaton", "AtomicAutomaton", + "AutomatonEquivalenceOracle", "AutomatonLanguage", "BuchiAutomaton", "EquivalenceOracle", @@ -141,7 +137,6 @@ "CallDrivenAutomaton", "CanonicalVisiblyPushdownAutomaton", "CanonicalRFSA", - "CompositeVisiblyPushdownAutomaton", "DFA", "DeterministicVisiblyPushdownAutomaton", "DyckAlphabet", @@ -173,24 +168,20 @@ "colex_width", "cartesian_product_gg", "cartesian_product_tt", - "complement_vpa", "compose_tg", "compose_tt", - "concat_vpa", "complete", "count_accessible_idfa", "count_flag_sequences", "count_icdfa", "count_icdfa_empty", "determinize", - "difference_vpa", "dfa_to_icdfa_string", "equivalent", "first_icdfa_empty_string", "first_idfa_string", "flags_from_string", "icdfa_string_to_dfa", - "intersection_vpa", "is_input_consistent", "is_well_matched", "is_wheeler", @@ -199,7 +190,6 @@ "iter_idfa_strings", "last_icdfa_empty_string", "MISSING_TRANSITION", - "kleene_star_vpa", "learn_dfa_edsm", "learn_dfa_from_language", "learn_dfa_lstar", @@ -209,6 +199,9 @@ "learn_mealy_from_transducer", "learn_mealy_lstar", "learn_pfa_alergia", + "learn_prime_atomaton_nlstar", + "learn_rfsa_from_language", + "learn_rfsa_nlstar", "learn_sofic_dyck_shift_papni", "maximum_colex_relation", "minimize", @@ -224,7 +217,6 @@ "trim", "transduce_generator", "unrank_idfa_string", - "union_vpa", "validate_idfa_string", "validate_icdfa_empty_string", "wheeler_canonical_form", diff --git a/sofic/automata/active.py b/sofic/automata/active.py index df47fd0..01393f3 100644 --- a/sofic/automata/active.py +++ b/sofic/automata/active.py @@ -13,8 +13,8 @@ * a redundancy-free **discrimination-tree** learner in the TTT family :cite:`KearnsVazirani1994,Isberner2014`. -These complement the existing NL\* átomaton learner -(:func:`sofic.automata.learning.learn_maximized_prime_atomaton`). +These complement the NL\* canonical-RFSA learner +(:func:`sofic.automata.learning.learn_rfsa_nlstar`). """ from __future__ import annotations @@ -36,6 +36,7 @@ "LanguageMembershipOracle", "FunctionMealyOracle", "TransducerOutputOracle", + "AutomatonEquivalenceOracle", "ExhaustiveEquivalenceOracle", "RandomWalkEquivalenceOracle", "MealyExhaustiveEquivalenceOracle", @@ -162,6 +163,49 @@ def find_counterexample(self, hypothesis: DFA) -> Word | None: return None +class AutomatonEquivalenceOracle: + """Exact equivalence against a target finite automaton. + + Returns a shortest word on which the hypothesis and the target disagree, + found by breadth-first search over the product of their subset + constructions, or ``None`` when they recognize the same language. + """ + + def __init__(self, target: Any, alphabet: Iterable[Any] | None = None) -> None: + self._target = target + self._alphabet = None if alphabet is None else frozenset(alphabet) + + def find_counterexample(self, hypothesis: Any) -> Word | None: + from sofic.automata.algorithms import _transition_alphabet + + symbols = (self._alphabet or frozenset()) | _transition_alphabet(self._target) + symbols |= _transition_alphabet(hypothesis) + ordered = tuple(sorted(symbols, key=repr)) + start = (_closure(hypothesis, hypothesis.initial_states), _closure(self._target, self._target.initial_states)) + seen = {start} + queue: list[tuple[tuple[frozenset[Any], frozenset[Any]], Word]] = [(start, ())] + for (left, right), word in queue: + if bool(left & hypothesis.accepting_states) != bool(right & self._target.accepting_states): + return word + for symbol in ordered: + successor = (_step(hypothesis, left, symbol), _step(self._target, right, symbol)) + if successor not in seen: + seen.add(successor) + queue.append((successor, (*word, symbol))) + return None + + +def _closure(aut: Any, states: Iterable[Any]) -> frozenset[Any]: + return frozenset(aut.epsilon_closure(set(states))) + + +def _step(aut: Any, states: frozenset[Any], symbol: Any) -> frozenset[Any]: + targets: set[Any] = set() + for state in states: + targets |= aut.delta(state, symbol) + return _closure(aut, targets) + + class RandomWalkEquivalenceOracle: """Randomized equivalence test drawing random input words for DFA hypotheses.""" diff --git a/sofic/automata/atomaton.py b/sofic/automata/atomaton.py index 51f9862..10e2b89 100644 --- a/sofic/automata/atomaton.py +++ b/sofic/automata/atomaton.py @@ -80,8 +80,15 @@ def to_minimal_dfa_via_double_reversal(self) -> DFA: return minimize(self, algorithm="brzozowski") -class MaximizedPrimeAtomaton(AtomicAutomaton): - """Maximized prime átomaton — dual of the canonical RFSA.""" +class MaximizedPrimeAtomaton(NFA): + """Maximized prime átomaton: the dual of the canonical RFSA :cite:`MaarandTamm2022`. + + It is the reverse of the canonical RFSA of the reversed language, just as the + átomaton is the reverse of the minimal DFA of the reversed language. Its + states are the maximized prime atoms; the right language of each lies between + its atom and its maximized atom :cite:`Tamm2015`, so -- unlike the átomaton -- + it need not be atomic. + """ @classmethod def from_language(cls, language: RegularLanguage | NFA, **kwargs: Any) -> MaximizedPrimeAtomaton: @@ -97,6 +104,13 @@ def from_observation_table(cls, table: ObservationTable, **kwargs: Any) -> Maxim @classmethod def from_canonical_rfsa(cls, rfsa: CanonicalRFSA, **kwargs: Any) -> MaximizedPrimeAtomaton: - from sofic.automata.canonical_dual import dual_atomaton_from_rfsa + """Return the maximized prime átomaton of the language ``rfsa`` recognizes.""" + from sofic.automata.canonical_extraction import maximized_prime_atomaton_from_language + + return maximized_prime_atomaton_from_language(rfsa) + + def dual(self) -> CanonicalRFSA: + """Return the reverse automaton: the canonical RFSA of the reversed language.""" + from sofic.automata.canonical_dual import dual_rfsa_from_atomaton - return dual_atomaton_from_rfsa(rfsa) + return dual_rfsa_from_atomaton(self) diff --git a/sofic/automata/canonical_dual.py b/sofic/automata/canonical_dual.py index b8b7d43..4ba8469 100644 --- a/sofic/automata/canonical_dual.py +++ b/sofic/automata/canonical_dual.py @@ -1,4 +1,9 @@ -"""Dual constructions between canonical RFSA and maximized prime átomaton.""" +"""Duality between canonical RFSAs and maximized prime átomata. + +The maximized prime átomaton of ``L`` is the reverse of the canonical RFSA of +the reversed language :cite:`MaarandTamm2022`. Reversing either object therefore gives +the other one for the reversed language, and reversing twice is the identity. +""" from __future__ import annotations @@ -7,12 +12,14 @@ def dual_atomaton_from_rfsa(rfsa: CanonicalRFSA) -> MaximizedPrimeAtomaton: - from sofic.automata.canonical_extraction import maximized_prime_atomaton_from_language + """Reverse the canonical RFSA of ``L`` into the maximized prime átomaton of ``L^R``.""" + from sofic.automata.canonical_extraction import _reverse_into - return maximized_prime_atomaton_from_language(rfsa) + return _reverse_into(MaximizedPrimeAtomaton, rfsa) def dual_rfsa_from_atomaton(atomaton: MaximizedPrimeAtomaton) -> CanonicalRFSA: - from sofic.automata.canonical_extraction import canonical_rfsa_from_language + """Reverse the maximized prime átomaton of ``L`` into the canonical RFSA of ``L^R``.""" + from sofic.automata.canonical_extraction import _reverse_into - return canonical_rfsa_from_language(atomaton) + return _reverse_into(CanonicalRFSA, atomaton) diff --git a/sofic/automata/canonical_extraction.py b/sofic/automata/canonical_extraction.py index 533f913..61a3d90 100644 --- a/sofic/automata/canonical_extraction.py +++ b/sofic/automata/canonical_extraction.py @@ -1,22 +1,38 @@ -"""Canonical automaton extraction from languages and observation tables.""" +"""Canonical automaton extraction from languages and observation tables. + +Every construction works on the minimal complete DFA ``D`` of the language +``L``. Its states are the residuals (left quotients) ``u^{-1} L``, and inclusion +and union questions between residuals are decided exactly on ``D``: + +* ``L_p <= L_q`` holds iff no pair reachable from ``(p, q)`` in ``D x D`` is + accepting in the first component only; +* ``L_q <= union(L_p for p in S)`` holds iff no pair reachable from ``(q, S)`` in + ``D x subsets(D)`` accepts on the left with no accepting state on the right. + +A residual is *prime* when it is non-empty and not the union of the residuals +strictly contained in it :cite:`Denis2002`. +""" from __future__ import annotations +from collections import deque +from collections.abc import Hashable, Iterable +from dataclasses import dataclass from typing import Any from sofic.automata.atomaton import Atomaton, MaximizedPrimeAtomaton +from sofic.automata.base import LabeledAutomaton from sofic.automata.dfa import DFA -from sofic.automata.languages.automaton_ops import ( - minimal_dfa_from_language, -) from sofic.automata.languages.base import AutomatonLanguage, RegularLanguage, as_language from sofic.automata.nfa import NFA from sofic.automata.observation import ObservationTable from sofic.automata.rfsa import CanonicalRFSA +from sofic.graph import ATTR_SYMBOL +from sofic.states import sequential_labels -def _language_automaton(language: RegularLanguage | NFA | DFA) -> NFA | DFA: - if isinstance(language, (NFA, DFA)): +def _language_automaton(language: RegularLanguage | LabeledAutomaton) -> LabeledAutomaton: + if isinstance(language, LabeledAutomaton): return language lang = as_language(language) # type: ignore[arg-type] if isinstance(lang, AutomatonLanguage): @@ -24,22 +40,165 @@ def _language_automaton(language: RegularLanguage | NFA | DFA) -> NFA | DFA: raise TypeError(f"cannot extract automaton from {type(language)!r}") -def canonical_rfsa_from_language(language: RegularLanguage | NFA | DFA) -> CanonicalRFSA: - """Build canonical RFSA from prime residuals of the language. +@dataclass(frozen=True) +class ResidualTable: + """Minimal complete DFA of a language, viewed as its table of residuals.""" + + dfa: DFA + alphabet: tuple[Any, ...] + start: Hashable + accepting: frozenset[Hashable] + delta: dict[tuple[Hashable, Any], Hashable] + + @classmethod + def from_automaton(cls, aut: LabeledAutomaton) -> ResidualTable: + from sofic.automata.algorithms import _transition_alphabet, complete, minimize + + symbols = _transition_alphabet(aut) + source = aut if isinstance(aut, (NFA, DFA)) else NFA(**_automaton_kwargs(aut)) + dfa = complete(minimize(source, alphabet=symbols), symbols) + if not dfa.initial_states: + empty = DFA(input_alphabet=frozenset(symbols), initial_states=frozenset({"empty"})) + empty.graph.add_state("empty") + dfa = complete(empty, symbols) + delta = { + (transition.source, transition.data[ATTR_SYMBOL]): transition.target for transition in dfa.transitions() + } + return cls( + dfa=dfa, + alphabet=tuple(sorted(symbols, key=repr)), + start=next(iter(dfa.initial_states)), + accepting=frozenset(dfa.accepting_states), + delta=delta, + ) + + @property + def states(self) -> tuple[Hashable, ...]: + return tuple(self.dfa.states()) + + def is_empty(self, state: Hashable) -> bool: + """Whether the residual ``L_state`` is empty.""" + seen = {state} + queue = deque([state]) + while queue: + current = queue.popleft() + if current in self.accepting: + return False + for symbol in self.alphabet: + target = self.delta[(current, symbol)] + if target not in seen: + seen.add(target) + queue.append(target) + return True + + def includes(self, smaller: Hashable, larger: Hashable) -> bool: + """Whether ``L_smaller <= L_larger``.""" + seen = {(smaller, larger)} + queue = deque(seen) + while queue: + left, right = queue.popleft() + if left in self.accepting and right not in self.accepting: + return False + for symbol in self.alphabet: + pair = (self.delta[(left, symbol)], self.delta[(right, symbol)]) + if pair not in seen: + seen.add(pair) + queue.append(pair) + return True + + def is_covered(self, state: Hashable, cover: Iterable[Hashable]) -> bool: + """Whether ``L_state`` is contained in the union of ``L_p`` for ``p`` in ``cover``.""" + start = (state, frozenset(cover)) + seen = {start} + queue = deque([start]) + while queue: + left, right = queue.popleft() + if left in self.accepting and not (right & self.accepting): + return False + for symbol in self.alphabet: + pair = (self.delta[(left, symbol)], frozenset(self.delta[(q, symbol)] for q in right)) + if pair not in seen: + seen.add(pair) + queue.append(pair) + return True + + def prime_states(self) -> tuple[Hashable, ...]: + """States whose residual is prime, in breadth-first order from the start.""" + nonempty = [q for q in self._bfs_order() if not self.is_empty(q)] + primes = [] + for q in nonempty: + strictly_smaller = [p for p in nonempty if p != q and self.includes(p, q)] + if not self.is_covered(q, strictly_smaller): + primes.append(q) + return tuple(primes) + + def _bfs_order(self) -> list[Hashable]: + order = [self.start] + seen = {self.start} + for state in order: + for symbol in self.alphabet: + target = self.delta[(state, symbol)] + if target not in seen: + seen.add(target) + order.append(target) + return order + + def residual_automaton(self, state: Hashable) -> DFA: + """DFA recognizing the residual ``L_state``.""" + result = self.dfa.copy() + result.initial_states = frozenset({state}) + return result + + def left_language_automaton(self, state: Hashable) -> DFA: + """DFA recognizing the words that lead from the start to ``state``.""" + result = self.dfa.copy() + result.accepting_states = frozenset({state}) + return result + + +def _automaton_kwargs(aut: LabeledAutomaton) -> dict[str, Any]: + return { + "input_alphabet": aut.input_alphabet, + "initial_states": aut.initial_states, + "accepting_states": aut.accepting_states, + "graph": aut.graph.copy(), + } + + +def _labels(count: int) -> tuple[Hashable, ...]: + return sequential_labels(count) if count <= 26 else tuple(range(count)) - Phase 2 placeholder: currently reuses the minimal DFA structure directly. + +def canonical_rfsa_from_language(language: RegularLanguage | LabeledAutomaton) -> CanonicalRFSA: + r"""Build the canonical residual finite-state automaton of a regular language. + + Following :cite:`Denis2002`, the states are the prime residuals of ``L``; the + initial states are the primes contained in ``L``; the accepting states are + the primes containing the empty word; and there is a transition + :math:`p \xrightarrow{a} p'` exactly when :math:`L_{p'} \subseteq a^{-1} L_p`. + The canonical RFSA is saturated (it has every such transition) and is never + larger than the minimal DFA, often exponentially smaller. """ - aut = _language_automaton(language) - dfa = minimal_dfa_from_language(aut) - return CanonicalRFSA( - input_alphabet=dfa.input_alphabet, - initial_states=dfa.initial_states, - accepting_states=dfa.accepting_states, - graph=dfa.graph.copy(), + table = ResidualTable.from_automaton(_language_automaton(language)) + primes = table.prime_states() + name = dict(zip(primes, _labels(len(primes)), strict=True)) + rfsa = CanonicalRFSA( + input_alphabet=frozenset(table.alphabet), + initial_states=frozenset(name[p] for p in primes if table.includes(p, table.start)), + accepting_states=frozenset(name[p] for p in primes if p in table.accepting), ) + for p in primes: + rfsa.graph.add_state(name[p]) + for p in primes: + for symbol in table.alphabet: + successor = table.delta[(p, symbol)] + for target in primes: + if table.includes(target, successor): + rfsa.add_transition(name[p], name[target], symbol) + return rfsa -def atomaton_from_language(language: RegularLanguage | NFA | DFA) -> Atomaton: +def atomaton_from_language(language: RegularLanguage | LabeledAutomaton) -> Atomaton: """Build átomaton via double-reversal pipeline. The átomaton is the *reverse of the minimal DFA of the reverse language* @@ -48,6 +207,8 @@ def atomaton_from_language(language: RegularLanguage | NFA | DFA) -> Atomaton: of ``language``, which is Brzozowski's minimization rather than the átomaton. """ + from sofic.automata.languages.automaton_ops import minimal_dfa_from_language + aut = _language_automaton(language) dfa = minimal_dfa_from_language(aut) rev = dfa.reverse().determinize().minimize() @@ -60,18 +221,27 @@ def atomaton_from_language(language: RegularLanguage | NFA | DFA) -> Atomaton: ) -def maximized_prime_atomaton_from_language(language: RegularLanguage | NFA | DFA) -> MaximizedPrimeAtomaton: - """Build maximized prime átomaton from a language. +def maximized_prime_atomaton_from_language( + language: RegularLanguage | LabeledAutomaton, +) -> MaximizedPrimeAtomaton: + r"""Build the maximized prime átomaton of a regular language :cite:`MaarandTamm2022`. - Phase 2 placeholder: currently reuses the minimal DFA structure directly. + It is the reverse of the canonical RFSA of :math:`L^R`. Equivalently, it is + the subautomaton of the maximized átomaton of ``L`` -- the reverse of the + saturated minimal DFA of :math:`L^R` -- on the maximized atoms whose + quotients of :math:`L^R` are prime :cite:`Tamm2015`. """ - aut = _language_automaton(language) - lang = AutomatonLanguage(minimal_dfa_from_language(aut)) - return MaximizedPrimeAtomaton( - input_alphabet=lang.automaton.input_alphabet, - initial_states=lang.automaton.initial_states, - accepting_states=lang.automaton.accepting_states, - graph=lang.automaton.graph.copy(), + reversed_rfsa = canonical_rfsa_from_language(_language_automaton(language).reverse()) + return _reverse_into(MaximizedPrimeAtomaton, reversed_rfsa) + + +def _reverse_into(cls: type[NFA], aut: NFA) -> Any: + reversed_aut = aut.reverse() + return cls( + input_alphabet=reversed_aut.input_alphabet, + initial_states=reversed_aut.initial_states, + accepting_states=reversed_aut.accepting_states, + graph=reversed_aut.graph.copy(), ) @@ -108,8 +278,27 @@ def signature(word: tuple[Any, ...]) -> tuple[bool, ...]: def observation_to_canonical_rfsa(table: ObservationTable) -> CanonicalRFSA: - dfa = observation_to_minimal_dfa(table) - return canonical_rfsa_from_language(dfa) + r"""NL\*-style RFSA extraction from an RFSA-closed, RFSA-consistent table. + + States are the rows of the access words that are prime among all rows of + the table, exactly as in the NL\* hypothesis :cite:`Bollig2009`. + """ + from sofic.automata.learning import _hypothesis, _primes + + experiments = sorted(table.experiments, key=lambda w: (len(w), w)) + if () in experiments: + experiments.remove(()) + experiments.insert(0, ()) + symbols = tuple(sorted(_infer_alphabet(table), key=repr)) + access = sorted(table.access_words, key=lambda w: (len(w), w)) + + def row(word: tuple[Any, ...]) -> tuple[bool, ...]: + return tuple(table.membership.get(word + exp, False) for exp in experiments) + + upper = {u: row(u) for u in access} + lower = {u + (a,): row(u + (a,)) for u in access for a in symbols} + primes_upper = _primes(set(upper.values()) | set(lower.values()), len(experiments)) & set(upper.values()) + return _hypothesis(access, symbols, upper, primes_upper, row) def observation_to_atomaton(table: ObservationTable) -> Atomaton: diff --git a/sofic/automata/languages/__init__.py b/sofic/automata/languages/__init__.py index 5f3224b..6756b45 100644 --- a/sofic/automata/languages/__init__.py +++ b/sofic/automata/languages/__init__.py @@ -1,6 +1,6 @@ """Regular-language algebra for automata constructions.""" -from sofic.automata.languages.atoms import atoms, is_prime_atom, prime_atoms +from sofic.automata.languages.atoms import atoms, prime_atoms from sofic.automata.languages.base import AutomatonLanguage, ExplicitLanguage, RegularLanguage from sofic.automata.languages.operations import ( complement, @@ -25,7 +25,6 @@ "difference", "intersection", "is_composed_residual", - "is_prime_atom", "kleene_star", "left_quotient", "left_quotients", diff --git a/sofic/automata/languages/_quotient_utils.py b/sofic/automata/languages/_quotient_utils.py index 8adceb7..ec136da 100644 --- a/sofic/automata/languages/_quotient_utils.py +++ b/sofic/automata/languages/_quotient_utils.py @@ -6,7 +6,7 @@ from typing import Any from sofic.automata.languages.automaton_ops import minimal_dfa_from_language -from sofic.automata.languages.base import AutomatonLanguage, ExplicitLanguage, RegularLanguage, as_language +from sofic.automata.languages.base import AutomatonLanguage, ExplicitLanguage def _words_up_to(length: int, alphabet: frozenset[Any]) -> list[tuple[Any, ...]]: @@ -18,29 +18,12 @@ def _words_up_to(length: int, alphabet: frozenset[Any]) -> list[tuple[Any, ...]] return [word + (symbol,) for word in shorter for symbol in alphabet] -def _languages_equal(left: RegularLanguage, right: RegularLanguage, alphabet: frozenset[Any], max_len: int = 6) -> bool: - for length in range(max_len + 1): - for word in _words_up_to(length, alphabet): - if (word in left) != (word in right): - return False - return True - - def _residual_from_state(aut: AutomatonLanguage, state) -> AutomatonLanguage: sub = aut.automaton.copy() sub.initial_states = frozenset({state}) return AutomatonLanguage(minimal_dfa_from_language(sub)) -def _alphabet_of(language: RegularLanguage) -> frozenset[Any]: - lang = as_language(language) # type: ignore[arg-type] - if isinstance(lang, ExplicitLanguage): - return lang.alphabet - if isinstance(lang, AutomatonLanguage): - return lang.automaton.input_alphabet - return frozenset() - - def _suffixes_if_prefix(word: tuple[Any, ...], prefix: Sequence[Any]) -> set[tuple[Any, ...]]: p = tuple(prefix) if word[: len(p)] == p: diff --git a/sofic/automata/languages/atoms.py b/sofic/automata/languages/atoms.py index dcdfa25..38de8a9 100644 --- a/sofic/automata/languages/atoms.py +++ b/sofic/automata/languages/atoms.py @@ -1,31 +1,41 @@ -"""Atoms and prime atoms of regular languages.""" +r"""Atoms and prime atoms of regular languages :cite:`BrzozowskiTamm2014`. + +An *atom* of ``L`` is a non-empty intersection of complemented or +uncomplemented left quotients of ``L``. The atoms correspond one-to-one with +the states ``q`` of the minimal DFA of the reversed language :math:`L^R`: the +atom :math:`A_q` is the reverse of the set of words leading to ``q``, and these +are exactly the right languages of the átomaton's states. An atom is *prime* +when its matching quotient of :math:`L^R` is a prime residual; prime atoms label +the states of the maximized prime átomaton. +""" from __future__ import annotations -from sofic.automata.languages._quotient_utils import _alphabet_of, _languages_equal, _residual_from_state -from sofic.automata.languages.automaton_ops import minimal_dfa_from_language from sofic.automata.languages.base import AutomatonLanguage, RegularLanguage, as_language -from sofic.automata.languages.quotients import left_quotients -def atoms(language: RegularLanguage) -> frozenset[RegularLanguage]: +def _reversed_table(language: RegularLanguage): + from sofic.automata.canonical_extraction import ResidualTable + lang = as_language(language) # type: ignore[arg-type] - result: set[RegularLanguage] = set(left_quotients(language)) - if isinstance(lang, AutomatonLanguage) and not result: - dfa = minimal_dfa_from_language(lang.automaton) - for state in dfa.states(): - result.add(_residual_from_state(AutomatonLanguage(dfa), state)) - return frozenset(result) + if not isinstance(lang, AutomatonLanguage): + raise TypeError("atoms are defined for automaton-backed regular languages") + return ResidualTable.from_automaton(lang.automaton.reverse()) -def prime_atoms(language: RegularLanguage) -> frozenset[RegularLanguage]: - all_atoms = atoms(language) - return frozenset(atom for atom in all_atoms if is_prime_atom(atom, all_atoms)) +def _atom(table, state) -> AutomatonLanguage: + from sofic.automata.languages.automaton_ops import minimal_dfa_from_language + + return AutomatonLanguage(minimal_dfa_from_language(table.left_language_automaton(state).reverse())) -def is_prime_atom(atom: RegularLanguage, all_atoms: frozenset[RegularLanguage]) -> bool: - others = [candidate for candidate in all_atoms if candidate is not atom] - if not others: - return True - alphabet = _alphabet_of(atom) - return all(not _languages_equal(atom, other, alphabet) for other in others) +def atoms(language: RegularLanguage) -> frozenset[RegularLanguage]: + """Return the atoms of ``language``.""" + table = _reversed_table(language) + return frozenset(_atom(table, q) for q in table.states if not table.is_empty(q)) + + +def prime_atoms(language: RegularLanguage) -> frozenset[RegularLanguage]: + """Return the prime atoms of ``language``.""" + table = _reversed_table(language) + return frozenset(_atom(table, q) for q in table.prime_states()) diff --git a/sofic/automata/languages/residuals.py b/sofic/automata/languages/residuals.py index ddaee4f..5e26d79 100644 --- a/sofic/automata/languages/residuals.py +++ b/sofic/automata/languages/residuals.py @@ -1,25 +1,65 @@ -"""Prime and composed residual languages.""" +"""Prime and composed residual languages :cite:`Denis2002`. + +A residual (left quotient) of ``L`` is *composed* when it is the union of the +residuals strictly contained in it, and *prime* otherwise (the empty residual +is the empty union, so it is composed). For automaton-backed languages these +questions are decided exactly on the minimal DFA; for an +:class:`~sofic.automata.languages.base.ExplicitLanguage` (a finite labeled +sample) they are answered on the sample only. +""" from __future__ import annotations -from sofic.automata.languages._quotient_utils import ( - _alphabet_of, - _is_union_of_others, - _languages_equal, -) -from sofic.automata.languages.base import ExplicitLanguage, RegularLanguage +from sofic.automata.languages._quotient_utils import _is_union_of_others +from sofic.automata.languages.base import AutomatonLanguage, ExplicitLanguage, RegularLanguage, as_language from sofic.automata.languages.quotients import left_quotients def prime_residuals(language: RegularLanguage) -> frozenset[RegularLanguage]: - all_residuals = left_quotients(language) + """Return the prime residuals of ``language``.""" + lang = as_language(language) # type: ignore[arg-type] + if isinstance(lang, AutomatonLanguage): + from sofic.automata.canonical_extraction import ResidualTable + + table = ResidualTable.from_automaton(lang.automaton) + return frozenset(AutomatonLanguage(table.residual_automaton(q)) for q in table.prime_states()) + all_residuals = left_quotients(lang) return frozenset(r for r in all_residuals if not is_composed_residual(r, all_residuals)) def is_composed_residual(residual: RegularLanguage, all_residuals: frozenset[RegularLanguage]) -> bool: + """Return whether ``residual`` is the union of the members of ``all_residuals`` strictly inside it.""" if isinstance(residual, ExplicitLanguage): others = [r for r in all_residuals if r is not residual and isinstance(r, ExplicitLanguage)] return _is_union_of_others(residual, others) - others = [r for r in all_residuals if r is not residual] - alphabet = _alphabet_of(residual) - return any(_languages_equal(residual, other, alphabet) for other in others) + + from sofic.automata.languages.automaton_ops import union_nfa + + target = as_language(residual).automaton # type: ignore[union-attr] + inside = [] + for other in all_residuals: + if other is residual: + continue + candidate = as_language(other).automaton # type: ignore[union-attr] + if _is_subset(candidate, target) and not _is_subset(target, candidate): + inside.append(candidate) + if not inside: + return _is_empty(target) + union = inside[0] + for candidate in inside[1:]: + union = union_nfa(union, candidate) + return _is_subset(target, union) + + +def _is_subset(smaller, larger) -> bool: + from sofic.automata.algorithms import _transition_alphabet + from sofic.automata.languages.automaton_ops import difference_dfa + + alphabet = _transition_alphabet(smaller) | _transition_alphabet(larger) + return _is_empty(difference_dfa(smaller, larger, alphabet=alphabet)) + + +def _is_empty(aut) -> bool: + from sofic.automata.algorithms import trim + + return not trim(aut).accepting_states diff --git a/sofic/automata/learning.py b/sofic/automata/learning.py index 021cc07..d461d52 100644 --- a/sofic/automata/learning.py +++ b/sofic/automata/learning.py @@ -1,79 +1,202 @@ -"""NL* active learning for maximized prime átomata.""" +r"""NL\*: Angluin-style active learning of canonical RFSAs :cite:`Bollig2009`. + +NL\* keeps an observation table with rows for the access words ``U`` and their +one-symbol extensions ``U Sigma``, and columns for a suffix-closed set ``V`` of +experiments. Rows are compared pointwise: ``r <= r'`` when ``r'`` accepts every +experiment ``r`` accepts, and the join of rows is their pointwise ``or``. A row +is *prime* when it is not the join of the rows strictly below it. + +The table is + +* **RFSA-closed** when every row of ``U Sigma`` is the join of the prime rows of + ``U`` below it, and +* **RFSA-consistent** when ``row(u') <= row(u)`` implies + ``row(u' a) <= row(u a)`` for all ``u, u'`` in ``U`` and symbols ``a``. + +A closed, consistent table yields the hypothesis whose states are the prime +rows of ``U``. Counterexamples add all of their suffixes to ``V``. When the +equivalence oracle accepts, the hypothesis is the canonical RFSA of the target. +""" from __future__ import annotations -from collections.abc import Sequence +from collections.abc import Iterable, Sequence from typing import Any +from sofic.automata.active import ( + AutomatonEquivalenceOracle, + EquivalenceOracle, + ExhaustiveEquivalenceOracle, + LanguageMembershipOracle, + MembershipOracle, + _MembershipCache, +) from sofic.automata.atomaton import MaximizedPrimeAtomaton -from sofic.automata.languages.base import RegularLanguage -from sofic.automata.observation import ObservationTable +from sofic.automata.base import LabeledAutomaton +from sofic.automata.rfsa import CanonicalRFSA +Word = tuple[Any, ...] +Row = tuple[bool, ...] + + +def _leq(left: Row, right: Row) -> bool: + return all(not a or b for a, b in zip(left, right, strict=True)) + + +def _join(rows: Iterable[Row], width: int) -> Row: + result = [False] * width + for row in rows: + result = [a or b for a, b in zip(result, row, strict=True)] + return tuple(result) -def learn_maximized_prime_atomaton( - teacher: RegularLanguage, - alphabet: frozenset[Any], - max_rounds: int = 32, -) -> MaximizedPrimeAtomaton: - """Simulated NL* teacher loop using membership queries.""" - access_words: set[tuple[Any, ...]] = {()} - experiments: set[tuple[Any, ...]] = {()} - - def membership(word: Sequence[Any]) -> bool: - return tuple(word) in teacher - - def close_table(max_depth: int = 6) -> None: - changed = True - while changed: - changed = False - for word in list(access_words): - if len(word) >= max_depth: - continue - for symbol in alphabet: - successor = word + (symbol,) - if successor not in access_words: - access_words.add(successor) - changed = True + +def _primes(rows: set[Row], width: int) -> set[Row]: + return {row for row in rows if _join((r for r in rows if r != row and _leq(r, row)), width) != row} + + +def learn_rfsa_nlstar( + alphabet: Iterable[Any], + membership: MembershipOracle, + equivalence: EquivalenceOracle, + *, + max_rounds: int = 100, +) -> CanonicalRFSA: + r"""Learn the canonical RFSA of the target language with NL\* :cite:`Bollig2009`.""" + symbols = tuple(sorted(alphabet, key=repr)) + member = _MembershipCache(membership).member + access: list[Word] = [()] + experiments: list[Word] = [()] + + def row(word: Word) -> Row: + return tuple(member(word + suffix) for suffix in experiments) for _ in range(max_rounds): - close_table() - table = _build_table(access_words, experiments, membership) - hypothesis = table.to_maximized_prime_atomaton() - counterexample = _find_counterexample(teacher, hypothesis, alphabet) + while True: + width = len(experiments) + upper = {u: row(u) for u in access} + lower = {u + (a,): row(u + (a,)) for u in access for a in symbols} + all_primes = _primes(set(upper.values()) | set(lower.values()), width) + primes_upper = all_primes & set(upper.values()) + + unclosed = next( + ( + w + for w, r in sorted(lower.items(), key=lambda item: (len(item[0]), repr(item[0]))) + if r in all_primes and r not in upper.values() + ), + None, + ) + if unclosed is not None: + access.append(unclosed) + continue + + inconsistency = _find_inconsistency(access, symbols, upper, row, experiments) + if inconsistency is not None: + experiments.append(inconsistency) + continue + break + + hypothesis = _hypothesis(access, symbols, upper, primes_upper, row) + counterexample = equivalence.find_counterexample(hypothesis) if counterexample is None: return hypothesis - for length in range(len(counterexample) + 1): - experiments.add(counterexample[length:]) - - return _build_table(access_words, experiments, membership).to_maximized_prime_atomaton() - - -def _build_table( - access_words: set[tuple[Any, ...]], - experiments: set[tuple[Any, ...]], - membership, -) -> ObservationTable: - membership_map: dict[tuple[Any, ...], bool] = {} - for prefix in access_words: - for suffix in experiments: - membership_map[prefix + suffix] = membership(prefix + suffix) - return ObservationTable( - access_words=frozenset(access_words), - experiments=frozenset(experiments), - membership=membership_map, + for start in range(len(counterexample) + 1): + suffix = tuple(counterexample[start:]) + if suffix not in experiments: + experiments.append(suffix) + raise RuntimeError(f"NL* did not converge within {max_rounds} equivalence rounds") + + +def _find_inconsistency(access, symbols, upper, row, experiments) -> Word | None: + for u in access: + for other in access: + if u == other or not _leq(upper[other], upper[u]): + continue + for symbol in symbols: + below, above = row(other + (symbol,)), row(u + (symbol,)) + if not _leq(below, above): + index = next(i for i, (b, a) in enumerate(zip(below, above, strict=True)) if b and not a) + return (symbol, *experiments[index]) + return None + + +def _hypothesis(access, symbols, upper, primes_upper, row) -> CanonicalRFSA: + representative: dict[Row, Word] = {} + for u in sorted(access, key=lambda w: (len(w), repr(w))): + if upper[u] in primes_upper: + representative.setdefault(upper[u], u) + states = sorted(representative, key=lambda r: (len(representative[r]), repr(representative[r]))) + name = {r: index for index, r in enumerate(states)} + epsilon_row = upper[()] + rfsa = CanonicalRFSA( + input_alphabet=frozenset(symbols), + initial_states=frozenset(name[r] for r in states if _leq(r, epsilon_row)), + accepting_states=frozenset(name[r] for r in states if r[0]), ) + for r in states: + rfsa.graph.add_state(name[r]) + for r in states: + for symbol in symbols: + successor = row(representative[r] + (symbol,)) + for target in states: + if _leq(target, successor): + rfsa.add_transition(name[r], name[target], symbol) + return rfsa -def _find_counterexample( - teacher: RegularLanguage, - hypothesis: MaximizedPrimeAtomaton, - alphabet: frozenset[Any], - max_len: int = 8, -) -> tuple[Any, ...] | None: - from sofic.automata.languages._quotient_utils import _words_up_to +class _ReversedMembership: + def __init__(self, membership: MembershipOracle) -> None: + self._membership = membership - for length in range(max_len + 1): - for word in _words_up_to(length, alphabet): - if (word in teacher) != hypothesis.recognizes(word): - return word - return None + def member(self, word: Sequence[Any]) -> bool: + return bool(self._membership.member(tuple(reversed(tuple(word))))) + + +class _ReversedEquivalence: + def __init__(self, equivalence: EquivalenceOracle) -> None: + self._equivalence = equivalence + + def find_counterexample(self, hypothesis: Any) -> Word | None: + counterexample = self._equivalence.find_counterexample(hypothesis.reverse()) + return None if counterexample is None else tuple(reversed(tuple(counterexample))) + + +def learn_prime_atomaton_nlstar( + alphabet: Iterable[Any], + membership: MembershipOracle, + equivalence: EquivalenceOracle, + *, + max_rounds: int = 100, +) -> MaximizedPrimeAtomaton: + r"""Learn the maximized prime átomaton by running NL\* on the reversed target. + + The maximized prime átomaton of ``L`` is the reverse of the canonical RFSA of + :math:`L^R` :cite:`MaarandTamm2022`, so NL\* is run against reversed + membership and equivalence oracles and its result reversed. + """ + from sofic.automata.canonical_dual import dual_atomaton_from_rfsa + + reversed_rfsa = learn_rfsa_nlstar( + alphabet, _ReversedMembership(membership), _ReversedEquivalence(equivalence), max_rounds=max_rounds + ) + return dual_atomaton_from_rfsa(reversed_rfsa) + + +def learn_rfsa_from_language( + target: Any, + alphabet: Iterable[Any], + *, + max_length: int = 12, + max_rounds: int = 100, +) -> CanonicalRFSA: + r"""Learn the canonical RFSA of ``target`` with NL\*. + + Uses an exact equivalence oracle when ``target`` is a finite automaton and a + bounded exhaustive one (words up to ``max_length``) otherwise. + """ + membership = LanguageMembershipOracle(target) + if isinstance(target, LabeledAutomaton): + equivalence: EquivalenceOracle = AutomatonEquivalenceOracle(target, alphabet) + else: + equivalence = ExhaustiveEquivalenceOracle(membership, alphabet, max_length=max_length) + return learn_rfsa_nlstar(alphabet, membership, equivalence, max_rounds=max_rounds) diff --git a/sofic/automata/nwa.py b/sofic/automata/nwa.py index 70f1ac0..94eb930 100644 --- a/sofic/automata/nwa.py +++ b/sofic/automata/nwa.py @@ -221,6 +221,46 @@ def recognizes_visible(self, symbols: Sequence[Any]) -> bool: ) return self.recognizes(word) + def union(self, other: NestedWordAutomaton) -> NestedWordAutomaton: + """Return an NWA recognizing the union with ``other``.""" + return _from_tagged_vpa(self.to_vpa().union(other.to_vpa())) + + def intersection(self, other: NestedWordAutomaton) -> NestedWordAutomaton: + """Return an NWA recognizing the intersection with ``other``.""" + return _from_tagged_vpa(self.to_vpa().intersection(other.to_vpa())) + + def complement(self) -> NestedWordAutomaton: + """Return an NWA recognizing the complement over this NWA's role alphabets.""" + return _from_tagged_vpa(self.to_vpa().complement()) + + def difference(self, other: NestedWordAutomaton) -> NestedWordAutomaton: + """Return an NWA recognizing this language minus ``other``.""" + return _from_tagged_vpa(self.to_vpa().difference(other.to_vpa())) + + def concat(self, other: NestedWordAutomaton) -> NestedWordAutomaton: + """Return an NWA recognizing concatenation with ``other``.""" + return _from_tagged_vpa(self.to_vpa().concat(other.to_vpa())) + + def kleene_star(self) -> NestedWordAutomaton: + """Return an NWA recognizing the Kleene star.""" + return _from_tagged_vpa(self.to_vpa().kleene_star()) + + def is_empty(self) -> bool: + """Return whether no nested word is accepted.""" + return self.to_vpa().is_empty() + + def is_universal(self) -> bool: + """Return whether every nested word over the role alphabets is accepted.""" + return self.to_vpa().is_universal() + + def includes(self, other: NestedWordAutomaton) -> bool: + """Return whether ``other``'s language is contained in this one.""" + return self.to_vpa().includes(other.to_vpa()) + + def equivalent(self, other: NestedWordAutomaton) -> bool: + """Return whether both NWAs recognize the same nested words.""" + return self.to_vpa().equivalent(other.to_vpa()) + @classmethod def from_vpa(cls, vpa: VisiblyPushdownAutomaton) -> NestedWordAutomaton: """Copy a visibly pushdown automaton into an equivalent NWA view.""" @@ -316,6 +356,32 @@ def to_vpa(self, *, tag_symbols: bool = True) -> VisiblyPushdownAutomaton: return result +def _from_tagged_vpa(vpa: VisiblyPushdownAutomaton) -> NestedWordAutomaton: + """Inverse of :meth:`NestedWordAutomaton.to_vpa` with role-tagged symbols.""" + result = NestedWordAutomaton( + call_alphabet=frozenset(symbol for _kind, symbol in vpa.call_alphabet), + return_alphabet=frozenset(symbol for _kind, symbol in vpa.return_alphabet), + internal_alphabet=frozenset(symbol for _kind, symbol in vpa.internal_alphabet), + hier_alphabet=vpa.stack_alphabet, + bottom_hier_state=vpa.bottom_stack_symbol, + initial_state=vpa.initial_state, + accepting_states=vpa.accepting_states, + ) + for state in vpa.states(): + result.graph.add_state(state) + for transition in vpa.transitions(): + data = transition.data + kind, (_role, symbol) = data.get(ATTR_KIND), data.get(ATTR_SYMBOL) + if kind == KIND_CALL: + result.add_call_transition(transition.source, transition.target, symbol, data.get(ATTR_STACK_SYMBOL)) + elif kind == KIND_RETURN: + for hier_state in _return_hier_states(vpa.stack_alphabet, data.get(ATTR_STACK_SYMBOL)): + result.add_return_transition(transition.source, transition.target, symbol, hier_state) + else: + result.add_internal_transition(transition.source, transition.target, symbol) + return result + + def _require_disjoint_visible_alphabets( call_alphabet: frozenset[Any], return_alphabet: frozenset[Any], diff --git a/sofic/automata/rfsa.py b/sofic/automata/rfsa.py index 61933c5..846b867 100644 --- a/sofic/automata/rfsa.py +++ b/sofic/automata/rfsa.py @@ -8,19 +8,35 @@ from sofic.automata.nfa import NFA if TYPE_CHECKING: + from sofic.automata.atomaton import MaximizedPrimeAtomaton from sofic.automata.observation import ObservationTable class ResidualFiniteStateAutomaton(NFA): - """NFA whose states accept residual languages of the recognized language.""" + """NFA whose states accept residual languages of the recognized language :cite:`Denis2002`.""" def validate(self) -> None: super().validate() - # Phase 2: verify each state's right language is in Res(L(R)) + from sofic.automata.algorithms import equivalent + from sofic.automata.canonical_extraction import ResidualTable + + if not self.initial_states: + return + table = ResidualTable.from_automaton(self) + residuals = [table.residual_automaton(q) for q in table.states] + for state in self.states(): + right = NFA( + input_alphabet=self.input_alphabet, + initial_states=frozenset({state}), + accepting_states=self.accepting_states, + graph=self.graph, + ) + if not any(equivalent(right, residual, frozenset(table.alphabet)) for residual in residuals): + self._require(False, f"right language of state {state!r} is not a residual of the language") class CanonicalRFSA(ResidualFiniteStateAutomaton): - """Canonical residual finite-state automaton R(L).""" + """Canonical residual finite-state automaton R(L) :cite:`Denis2002`.""" @classmethod def from_language(cls, language: RegularLanguage | NFA, **kwargs: Any) -> CanonicalRFSA: @@ -33,3 +49,9 @@ def from_observation_table(cls, table: ObservationTable, **kwargs: Any) -> Canon from sofic.automata.canonical_extraction import observation_to_canonical_rfsa return observation_to_canonical_rfsa(table) + + def dual(self) -> MaximizedPrimeAtomaton: + """Return the reverse automaton: the maximized prime átomaton of the reversed language.""" + from sofic.automata.canonical_dual import dual_atomaton_from_rfsa + + return dual_atomaton_from_rfsa(self) diff --git a/sofic/automata/vpa.py b/sofic/automata/vpa.py index 088cdcd..c70c828 100644 --- a/sofic/automata/vpa.py +++ b/sofic/automata/vpa.py @@ -6,8 +6,9 @@ from collections.abc import Hashable, Iterable, Mapping, Sequence from typing import Any +from sofic.automata import vpa_constructions as vc from sofic.base import StateMachine -from sofic.exceptions import NonDeterministicError +from sofic.exceptions import NonDeterministicError, NonWellMatchedLanguageError from sofic.graph import ( ATTR_KIND, ATTR_STACK_SYMBOL, @@ -116,9 +117,10 @@ def add_return_transition( ) -> int: """Add a return transition. - If ``stack_symbol`` is omitted, the transition is a wildcard over - non-bottom stack symbols. This preserves the historical unguarded - return behavior of :class:`VisiblyPushdownAutomaton`. + If ``stack_symbol`` is omitted, the transition is a wildcard: it fires + on every stack symbol, and also on the empty stack (leaving it empty) + when the VPA has a ``bottom_stack_symbol``. A return guarded by the + bottom symbol fires only on the empty stack. """ data = {**attrs, ATTR_KIND: KIND_RETURN, ATTR_SYMBOL: symbol} if stack_symbol is not None: @@ -183,187 +185,66 @@ def recognizes(self, word: Sequence[Any]) -> bool: return recognizes_vpa(self, word) - def union(self, other: VisiblyPushdownAutomaton) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for the union with ``other``.""" - return union_vpa(self, other) + def union(self, other: VisiblyPushdownAutomaton) -> VisiblyPushdownAutomaton: + """Return a VPA recognizing the union with ``other``.""" + return _binary(vc.union, self, other) - def intersection(self, other: VisiblyPushdownAutomaton) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for the intersection with ``other``.""" - return intersection_vpa(self, other) + def intersection(self, other: VisiblyPushdownAutomaton) -> VisiblyPushdownAutomaton: + """Return a VPA recognizing the intersection with ``other`` (synchronized product).""" + return _binary(vc.intersection, self, other) - def intersect(self, other: VisiblyPushdownAutomaton) -> CompositeVisiblyPushdownAutomaton: - """Alias for :meth:`intersection`.""" - return self.intersection(other) + def complement(self) -> DeterministicVisiblyPushdownAutomaton: + """Return a deterministic VPA recognizing the complement over this visible alphabet.""" + return vc.denormalize(vc.complement(vc.normalize(self)), DeterministicVisiblyPushdownAutomaton) - def complement(self) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for the complement over this visible alphabet.""" - return complement_vpa(self) + def difference(self, other: VisiblyPushdownAutomaton) -> VisiblyPushdownAutomaton: + """Return a VPA recognizing this language minus ``other``.""" + return _binary(vc.difference, self, other) - def difference(self, other: VisiblyPushdownAutomaton) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for this language minus ``other``.""" - return difference_vpa(self, other) + def concat(self, other: VisiblyPushdownAutomaton) -> VisiblyPushdownAutomaton: + """Return a VPA recognizing concatenation with ``other``. - def concat(self, other: VisiblyPushdownAutomaton) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for concatenation with ``other``.""" - return concat_vpa(self, other) + Each factor is read from an empty stack of its own: a return in the right + factor that would pop a pending call of the left factor is a pending + return of the right factor. + """ + return _binary(vc.concat, self, other) - def concatenate(self, other: VisiblyPushdownAutomaton) -> CompositeVisiblyPushdownAutomaton: - """Alias for :meth:`concat`.""" - return self.concat(other) + def kleene_star(self) -> VisiblyPushdownAutomaton: + """Return a VPA recognizing the Kleene star; each factor starts from its own empty stack.""" + return vc.denormalize(vc.kleene_star(vc.normalize(self))) - def kleene_star(self) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for the Kleene star of this language.""" - return kleene_star_vpa(self) + def determinize(self) -> DeterministicVisiblyPushdownAutomaton: + """Return an equivalent complete deterministic VPA :cite:`AlurMadhusudan2009`.""" + return vc.determinize_vpa(self) - def star(self) -> CompositeVisiblyPushdownAutomaton: - """Alias for :meth:`kleene_star`.""" - return self.kleene_star() + def is_empty(self) -> bool: + """Return whether the language is empty.""" + return vc.is_empty(vc.normalize(self)) + def accepted_word(self) -> tuple[Any, ...] | None: + """Return a short accepted word, or ``None`` when the language is empty.""" + return vc.accepted_word(vc.normalize(self)) -class CompositeVisiblyPushdownAutomaton(VisiblyPushdownAutomaton): - """Lazy VPA language expression built from standard closure operations. + def is_universal(self) -> bool: + """Return whether every word over the visible alphabet is accepted.""" + return vc.is_empty(vc.complement(vc.normalize(self))) - Composite VPAs keep exact language semantics for operations whose concrete - graph construction would otherwise need a larger normalization pass. They - still expose the regular VPA membership API through :meth:`recognizes`. - """ + def includes(self, other: VisiblyPushdownAutomaton) -> bool: + """Return whether ``other``'s language is contained in this one.""" + return vc.is_empty(vc.difference(vc.normalize(other), vc.normalize(self))) - operation: str - operands: tuple[VisiblyPushdownAutomaton, ...] + def equivalent(self, other: VisiblyPushdownAutomaton) -> bool: + """Return whether both VPAs recognize the same language.""" + return self.includes(other) and other.includes(self) - def __init__( - self, - *, - operation: str, - operands: Iterable[VisiblyPushdownAutomaton], - ) -> None: - operands = tuple(operands) - if not operands: - raise ValueError("CompositeVisiblyPushdownAutomaton requires at least one operand") - call_alphabet, return_alphabet, internal_alphabet = _merge_visible_alphabets(operands) - super().__init__( - call_alphabet=call_alphabet, - return_alphabet=return_alphabet, - internal_alphabet=internal_alphabet, - stack_alphabet=frozenset(), - ) - self.operation = operation - self.operands = operands + def has_unmatched_word(self) -> bool: + """Return whether some accepted word has a pending call or a pending return.""" + return vc.has_unmatched_word(vc.normalize(self)) - def validate(self) -> None: - self._require( - self.operation in {"union", "intersection", "complement", "difference", "concat", "kleene_star"}, - f"unknown composite VPA operation {self.operation!r}", - ) - if self.operation in {"complement", "kleene_star"}: - self._require(len(self.operands) == 1, f"{self.operation} requires one operand") - elif self.operation in {"difference", "concat"}: - self._require(len(self.operands) == 2, f"{self.operation} requires two operands") - else: - self._require(len(self.operands) >= 2, f"{self.operation} requires at least two operands") - for operand in self.operands: - operand.validate() - call_alphabet, return_alphabet, internal_alphabet = _merge_visible_alphabets(self.operands) - self._require(call_alphabet == self.call_alphabet, "composite call alphabet is stale") - self._require(return_alphabet == self.return_alphabet, "composite return alphabet is stale") - self._require(internal_alphabet == self.internal_alphabet, "composite internal alphabet is stale") - def recognizes(self, word: Sequence[Any]) -> bool: - word = tuple(word) - if any(symbol not in self.input_alphabet for symbol in word): - return False - if self.operation == "union": - return any(operand.recognizes(word) for operand in self.operands) - if self.operation == "intersection": - return all(operand.recognizes(word) for operand in self.operands) - if self.operation == "complement": - return not self.operands[0].recognizes(word) - if self.operation == "difference": - return self.operands[0].recognizes(word) and not self.operands[1].recognizes(word) - if self.operation == "concat": - left, right = self.operands - return any( - left.recognizes(word[:index]) and right.recognizes(word[index:]) for index in range(len(word) + 1) - ) - if self.operation == "kleene_star": - operand = self.operands[0] - accepted = [False] * (len(word) + 1) - accepted[0] = True - for end in range(1, len(word) + 1): - accepted[end] = any(accepted[start] and operand.recognizes(word[start:end]) for start in range(end)) - return accepted[-1] - raise ValueError(f"unknown composite VPA operation {self.operation!r}") - - -def union_vpa( - left: VisiblyPushdownAutomaton, - right: VisiblyPushdownAutomaton, - *rest: VisiblyPushdownAutomaton, -) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for the union of the operands.""" - return CompositeVisiblyPushdownAutomaton(operation="union", operands=(left, right, *rest)) - - -def intersection_vpa( - left: VisiblyPushdownAutomaton, - right: VisiblyPushdownAutomaton, - *rest: VisiblyPushdownAutomaton, -) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for the intersection of the operands.""" - return CompositeVisiblyPushdownAutomaton(operation="intersection", operands=(left, right, *rest)) - - -def complement_vpa(vpa: VisiblyPushdownAutomaton) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for complement over ``vpa``'s visible alphabet.""" - return CompositeVisiblyPushdownAutomaton(operation="complement", operands=(vpa,)) - - -def difference_vpa( - left: VisiblyPushdownAutomaton, - right: VisiblyPushdownAutomaton, -) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for ``left`` minus ``right``.""" - return CompositeVisiblyPushdownAutomaton(operation="difference", operands=(left, right)) - - -def concat_vpa( - left: VisiblyPushdownAutomaton, - right: VisiblyPushdownAutomaton, -) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for language concatenation.""" - return CompositeVisiblyPushdownAutomaton(operation="concat", operands=(left, right)) - - -def kleene_star_vpa(vpa: VisiblyPushdownAutomaton) -> CompositeVisiblyPushdownAutomaton: - """Return a VPA recognizer for Kleene star.""" - return CompositeVisiblyPushdownAutomaton(operation="kleene_star", operands=(vpa,)) - - -def _merge_visible_alphabets( - vpas: Iterable[VisiblyPushdownAutomaton], -) -> tuple[frozenset[Any], frozenset[Any], frozenset[Any]]: - call_symbols: set[Any] = set() - return_symbols: set[Any] = set() - internal_symbols: set[Any] = set() - owners: dict[Any, str] = {} - for vpa in vpas: - for kind, symbols in ( - ("call", vpa.call_alphabet), - ("return", vpa.return_alphabet), - ("internal", vpa.internal_alphabet), - ): - for symbol in symbols: - existing = owners.get(symbol) - if existing is not None and existing != kind: - raise ValueError(f"symbol {symbol!r} is both {existing} and {kind}") - owners[symbol] = kind - if kind == "call": - call_symbols.add(symbol) - elif kind == "return": - return_symbols.add(symbol) - else: - internal_symbols.add(symbol) - return frozenset(call_symbols), frozenset(return_symbols), frozenset(internal_symbols) +def _binary(operation, left: VisiblyPushdownAutomaton, right: VisiblyPushdownAutomaton) -> VisiblyPushdownAutomaton: + return vc.denormalize(operation(vc.normalize(left), vc.normalize(right))) class DeterministicVisiblyPushdownAutomaton(VisiblyPushdownAutomaton): @@ -473,7 +354,14 @@ def return_successor(self, state: Hashable, symbol: Any, stack_symbol: Any) -> H @classmethod def from_vpa(cls, vpa: VisiblyPushdownAutomaton) -> DeterministicVisiblyPushdownAutomaton: - """Copy ``vpa`` into a deterministic VPA and validate determinism.""" + """Return ``vpa`` as a deterministic VPA, determinizing it when needed. + + An already deterministic ``vpa`` is copied with its states unchanged; + otherwise the summary construction of :cite:`AlurMadhusudan2009` is + applied (see :func:`~sofic.automata.vpa_constructions.determinize`). + """ + if vpa.initial_state is None or not _is_deterministic(vpa): + return vc.determinize_vpa(vpa) result = cls( input_alphabet=vpa.input_alphabet, call_alphabet=vpa.call_alphabet, @@ -489,6 +377,24 @@ def from_vpa(cls, vpa: VisiblyPushdownAutomaton) -> DeterministicVisiblyPushdown return result +def _is_deterministic(vpa: VisiblyPushdownAutomaton) -> bool: + probe = DeterministicVisiblyPushdownAutomaton( + call_alphabet=vpa.call_alphabet, + return_alphabet=vpa.return_alphabet, + internal_alphabet=vpa.internal_alphabet, + stack_alphabet=vpa.stack_alphabet, + bottom_stack_symbol=vpa.bottom_stack_symbol, + initial_state=vpa.initial_state, + accepting_states=vpa.accepting_states, + graph=vpa.graph, + ) + try: + probe._check_determinism() + except NonDeterministicError: + return False + return True + + class CallDrivenAutomaton(DeterministicVisiblyPushdownAutomaton): """Deterministic modular VPA whose call target depends only on the call symbol.""" @@ -770,15 +676,21 @@ def __init__( @classmethod def from_vpa(cls, vpa: VisiblyPushdownAutomaton) -> CanonicalVisiblyPushdownAutomaton: - """Build the Myhill-Nerode canonical deterministic VPA for ``vpa``. - - The construction is finite for deterministic VPAs because states are - summary classes of well-matched factors. With an empty call alphabet, - this specializes to the usual minimal DFA right-congruence construction. + """Build the Myhill-Nerode canonical deterministic VPA of a well-matched language. + + States are classes of the summary algebra of well-matched factors, so + the form is canonical for well-matched languages. Raises + :class:`~sofic.exceptions.NonWellMatchedLanguageError` when ``vpa`` + accepts a word with a pending call or return; general VPLs have no + unique minimal deterministic VPA :cite:`AlurKumarMadhusudanViswanathan2005`. + With an empty call alphabet this is the minimal DFA. """ + if vpa.has_unmatched_word(): + raise NonWellMatchedLanguageError( + "the canonical VPA is defined for well-matched languages; this one accepts a pending call or return" + ) det = DeterministicVisiblyPushdownAutomaton.from_vpa(vpa) - algebra = _SummaryAlgebra.from_vpa(det) - return algebra.to_canonical_vpa(cls, det) + return _SummaryAlgebra(det).to_canonical_vpa(cls) @classmethod def minimize(cls, vpa: VisiblyPushdownAutomaton) -> CanonicalVisiblyPushdownAutomaton: @@ -833,8 +745,21 @@ def _minimize_modular_vpa( call_entries = _metadata_or_argument(vpa, "call_entries", call_entries, None) entry_states = _metadata_or_argument(vpa, "entry_states", entry_states, None) - if modules is None or call_partition is None: - raise NotImplementedError("modular VPA minimization requires fixed modules and call_partition") + if modules is None: + convert = vc.to_multiple_entry if form == "mevpa" else vc.to_single_entry + converted = convert(vpa, call_partition) + return _minimize_modular_vpa( + target_cls, + converted, + modules=converted.modules, + call_partition=converted.call_partition, + base_module=converted.base_module, + call_entries=converted.call_entries, + entry_states=getattr(converted, "entry_states", None) if form != "cda" else None, + form=form, + ) + if call_partition is None: + raise ValueError("a call_partition is required when modules are given") det = _deterministic_view(vpa) modules = _normalize_modules(modules) @@ -904,12 +829,14 @@ def _infer_call_entries( entries: dict[Any, Hashable] = {} for (_source, symbol), (target, _stack) in det.call_transition_map().items(): if symbol not in call_partition: - raise NotImplementedError(f"call_partition missing call symbol {symbol!r}") + raise ValueError(f"call_partition missing call symbol {symbol!r}") existing = entries.get(symbol, _MISSING) if existing is _MISSING: entries[symbol] = target elif existing != target: - raise NotImplementedError(f"call target for {symbol!r} depends on source state") + raise ValueError( + f"call target for {symbol!r} depends on the source state; omit modules to convert the VPA first" + ) return entries @@ -941,7 +868,9 @@ def _infer_entry_states( if module == base_module: continue if len(states) != 1: - raise NotImplementedError("SEVPA minimization requires one inferred entry per non-base module") + raise ValueError( + "SEVPA minimization requires one entry per non-base module; omit modules to convert the VPA first" + ) result[module] = next(iter(states)) return result return {} @@ -1172,96 +1101,152 @@ def _canonical_stack_symbol(stack_symbol: Any, block_of: Mapping[Hashable, Hasha class _SummaryAlgebra: - def __init__( - self, - *, - state_order: tuple[Hashable, ...], - summaries: frozenset[tuple[int | None, ...]], - identity: tuple[int | None, ...], - internal_summaries: Mapping[Any, tuple[int | None, ...]], - class_of: Mapping[tuple[int | None, ...], int], - representatives: Mapping[int, tuple[int | None, ...]], - ) -> None: - self.state_order = state_order - self.summaries = summaries - self.identity = identity - self.internal_summaries = dict(internal_summaries) - self.class_of = dict(class_of) - self.representatives = dict(representatives) + """Finite algebra of well-matched summaries with top-level and nested congruences. + + A summary maps each source state to the state reached along a well-matched + word (``None`` when the run dies). Top-level summaries are identified when + every well-matched continuation accepts both or neither; nested summaries + (inside a pending call) are identified when, for every enclosing context, + returning from them lands in the same class. Both partitions are refined + together until internal steps, calls, and returns are well defined on + classes, which makes the quotient canonical. + """ - @classmethod - def from_vpa(cls, vpa: DeterministicVisiblyPushdownAutomaton) -> _SummaryAlgebra: - state_order = tuple(sorted(vpa.states(), key=repr)) - state_index = {state: index for index, state in enumerate(state_order)} - internal_summaries = { - symbol: _internal_summary(vpa, state_order, state_index, symbol) + def __init__(self, vpa: DeterministicVisiblyPushdownAutomaton) -> None: + self.vpa = vpa + self.state_order = tuple(sorted(vpa.states(), key=repr)) + state_index = {state: index for index, state in enumerate(self.state_order)} + self.internal_summaries = { + symbol: _internal_summary(vpa, self.state_order, state_index, symbol) for symbol in sorted(vpa.internal_alphabet, key=repr) } - identity = tuple(range(len(state_order))) - summaries = _close_summary_algebra(vpa, state_order, state_index, identity, internal_summaries) - class_of, representatives = _quotient_summaries(vpa, state_order, state_index, summaries, identity) - return cls( - state_order=state_order, - summaries=frozenset(summaries), - identity=identity, - internal_summaries=internal_summaries, - class_of=class_of, - representatives=representatives, + self.identity = tuple(range(len(self.state_order))) + self.summaries = sorted( + _close_summary_algebra(vpa, self.state_order, state_index, self.identity, self.internal_summaries), + key=repr, ) + self.calls = sorted(vpa.call_alphabet, key=repr) + self.returns = sorted(vpa.return_alphabet, key=repr) + self._wrap_cache: dict[tuple, tuple] = {} + self.top, self.nested = self._refine() + + def wrap(self, inner: tuple, call: Any, ret: Any) -> tuple: + key = (inner, call, ret) + if key not in self._wrap_cache: + self._wrap_cache[key] = _wrap_summary(self.vpa, self.state_order, inner, call, ret) + return self._wrap_cache[key] + + def accepts(self, summary: tuple) -> bool: + return _summary_accepts(self.vpa, self.state_order, summary) + + def _refine(self) -> tuple[dict[tuple, int], dict[tuple, int]]: + top = {s: int(self.accepts(s)) for s in self.summaries} + nested = dict.fromkeys(self.summaries, 0) + internals = list(self.internal_summaries.values()) + pairs = [(c, r) for c in self.calls for r in self.returns] + while True: + top_signature = { + s: ( + top[s], + tuple(top[_compose_summary(s, a)] for a in internals), + tuple(top[_compose_summary(s, self.wrap(x, c, r))] for x in self.summaries for c, r in pairs), + ) + for s in self.summaries + } + nested_signature = { + s: ( + nested[s], + tuple(nested[_compose_summary(s, a)] for a in internals), + tuple(nested[_compose_summary(s, self.wrap(x, c, r))] for x in self.summaries for c, r in pairs), + tuple( + (top[_compose_summary(o, self.wrap(s, c, r))], nested[_compose_summary(o, self.wrap(s, c, r))]) + for o in self.summaries + for c, r in pairs + ), + ) + for s in self.summaries + } + new_top = _number_blocks(top_signature, first=top_signature[self.identity]) + new_nested = _number_blocks(nested_signature, first=nested_signature[self.identity]) + stable = len(set(new_top.values())) == len(set(top.values())) and len(set(new_nested.values())) == len( + set(nested.values()) + ) + top, nested = new_top, new_nested + if stable: + return top, nested + + def to_canonical_vpa(self, cls: type[CanonicalVisiblyPushdownAutomaton]) -> CanonicalVisiblyPushdownAutomaton: + representative: dict[Hashable, tuple] = {} + for s in self.summaries: + representative.setdefault(self.top[s], s) + representative.setdefault(("nested", self.nested[s]), s) + + def label(summary: tuple, nested: bool) -> Hashable: + return ("nested", self.nested[summary]) if nested else self.top[summary] + + initial = label(self.identity, False) + entry = label(self.identity, True) + reachable: set[Hashable] = set() + internals, calls, returns = set(), set(), set() + pushed: set[tuple[Hashable, Any]] = set() + frontier = [initial] + while frontier: + while frontier: + state = frontier.pop() + if state in reachable: + continue + reachable.add(state) + summary = representative[state] + for symbol, step in self.internal_summaries.items(): + target = label(_compose_summary(summary, step), isinstance(state, tuple)) + internals.add((state, symbol, target)) + frontier.append(target) + for call in self.calls: + calls.add((state, call, entry, (state, call))) + pushed.add((state, call)) + frontier.append(entry) + # Returns pop a pushed (outer, call) from a nested state; both sets + # grow together, so repeat until no new state appears. + for state in [s for s in reachable if isinstance(s, tuple)]: + for outer, call in list(pushed): + for ret in self.returns: + combined = _compose_summary(representative[outer], self.wrap(representative[state], call, ret)) + target = label(combined, isinstance(outer, tuple)) + returns.add((state, ret, (outer, call), target)) + if target not in reachable: + frontier.append(target) - def to_canonical_vpa( - self, - cls: type[CanonicalVisiblyPushdownAutomaton], - source: DeterministicVisiblyPushdownAutomaton, - ) -> CanonicalVisiblyPushdownAutomaton: - identity_class = self.class_of[self.identity] - states = frozenset(self.representatives) - stack_alphabet = frozenset( - (summary_class, symbol) for summary_class in states for symbol in source.call_alphabet - ) - accepting_states = frozenset( - summary_class - for summary_class, summary in self.representatives.items() - if _summary_accepts(source, self.state_order, summary) - ) result = cls( - input_alphabet=source.input_alphabet, - call_alphabet=source.call_alphabet, - return_alphabet=source.return_alphabet, - internal_alphabet=source.internal_alphabet, - stack_alphabet=stack_alphabet, + input_alphabet=self.vpa.input_alphabet, + call_alphabet=self.vpa.call_alphabet, + return_alphabet=self.vpa.return_alphabet, + internal_alphabet=self.vpa.internal_alphabet, + stack_alphabet=frozenset(pushed), bottom_stack_symbol=None, - initial_state=identity_class, - accepting_states=accepting_states, - summary_representatives=self.representatives, + initial_state=initial, + accepting_states=frozenset( + s for s in reachable if not isinstance(s, tuple) and self.accepts(representative[s]) + ), + summary_representatives={s: representative[s] for s in reachable}, ) - for state in states: + for state in sorted(reachable, key=repr): result.graph.add_state(state) - - for summary_class, summary in sorted(self.representatives.items(), key=repr): - for symbol, internal in sorted(self.internal_summaries.items(), key=lambda item: repr(item[0])): - target_summary = _compose_summary(summary, internal) - result.add_internal_transition(summary_class, self.class_of[target_summary], symbol) - for symbol in sorted(source.call_alphabet, key=repr): - result.add_call_transition(summary_class, identity_class, symbol, (summary_class, symbol)) - - for inner_class, inner in sorted(self.representatives.items(), key=repr): - for outer_class, outer in sorted(self.representatives.items(), key=repr): - for call_symbol in sorted(source.call_alphabet, key=repr): - for return_symbol in sorted(source.return_alphabet, key=repr): - wrapped = _wrap_summary(source, self.state_order, inner, call_symbol, return_symbol) - target = _compose_summary(outer, wrapped) - result.add_return_transition( - inner_class, - self.class_of[target], - return_symbol, - (outer_class, call_symbol), - ) - + for source, symbol, target in sorted(internals, key=repr): + result.add_internal_transition(source, target, symbol) + for source, symbol, target, push in sorted(calls, key=repr): + result.add_call_transition(source, target, symbol, push) + for source, symbol, guard, target in sorted(returns, key=repr): + result.add_return_transition(source, target, symbol, guard) result.validate() return result +def _number_blocks(signatures: Mapping[tuple, Any], *, first: Any) -> dict[tuple, int]: + ordered = sorted(set(signatures.values()), key=lambda sig: (sig != first, repr(sig))) + index = {signature: position for position, signature in enumerate(ordered)} + return {summary: index[signature] for summary, signature in signatures.items()} + + def _internal_summary( vpa: DeterministicVisiblyPushdownAutomaton, state_order: Sequence[Hashable], @@ -1334,33 +1319,6 @@ def _wrap_summary( return tuple(result) -def _quotient_summaries( - vpa: DeterministicVisiblyPushdownAutomaton, - state_order: Sequence[Hashable], - state_index: Mapping[Hashable, int], - summaries: set[tuple[int | None, ...]], - identity: tuple[int | None, ...], -) -> tuple[dict[tuple[int | None, ...], int], dict[int, tuple[int | None, ...]]]: - contexts = tuple(sorted(summaries, key=repr)) - signatures = { - summary: tuple(_summary_accepts(vpa, state_order, _compose_summary(summary, context)) for context in contexts) - for summary in summaries - } - identity_signature = signatures[identity] - ordered_signatures = sorted( - set(signatures.values()), key=lambda signature: (signature != identity_signature, signature) - ) - signature_class = {signature: index for index, signature in enumerate(ordered_signatures)} - class_of = {summary: signature_class[signature] for summary, signature in signatures.items()} - representatives = { - index: min( - (summary for summary, signature in signatures.items() if signature_class[signature] == index), key=repr - ) - for index in signature_class.values() - } - return class_of, representatives - - def _summary_accepts( vpa: DeterministicVisiblyPushdownAutomaton, state_order: Sequence[Hashable], diff --git a/sofic/automata/vpa_constructions.py b/sofic/automata/vpa_constructions.py new file mode 100644 index 0000000..a49c733 --- /dev/null +++ b/sofic/automata/vpa_constructions.py @@ -0,0 +1,716 @@ +r"""Concrete constructions and decision procedures for visibly pushdown automata. + +Every construction works on a *normalized* form of a VPA: the bottom of the +stack is an explicit symbol ``⊥``, and wildcard returns are expanded into +guarded returns. A return guarded by ``⊥`` is a *pending* return: it fires on +the empty stack and leaves it empty. A VPA without ``⊥``-guarded returns rejects +pending returns. Acceptance is by final state, whatever the stack holds, so +pending calls are allowed :cite:`AlurMadhusudan2009`. + +* :func:`determinize` is the summary construction of Alur and Madhusudan: a + deterministic state is a pair ``(S, R)`` where ``S`` relates the state after + the last pending call to the current state along well-matched factors, and + ``R`` is the set of current states. The result is complete, so + :func:`complement` flips its accepting states. +* :func:`concat` and :func:`kleene_star` track, in the finite control, whether + the current factor's stack is empty, and push that bit with every symbol. A + return seen while the bit is set is a pending return of the current factor, + whatever lies below on the physical stack. +* :func:`is_empty` saturates the relation of well-matched summaries and then + explores states reachable with pending returns (stack empty) or pending calls. +* :func:`to_single_entry` and :func:`to_multiple_entry` convert a VPA of a + well-matched language into a modular form following the summary construction + of :cite:`AlurKumarMadhusudanViswanathan2005`: on a call the state resets to + the entry of the call's module and the caller is pushed. +""" + +from __future__ import annotations + +import heapq +from collections import defaultdict +from collections.abc import Hashable, Mapping +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +from sofic.exceptions import NonWellMatchedLanguageError +from sofic.graph import ATTR_KIND, ATTR_STACK_SYMBOL, ATTR_SYMBOL, KIND_CALL, KIND_INTERNAL, KIND_RETURN + +if TYPE_CHECKING: + from sofic.automata.vpa import ( + DeterministicVisiblyPushdownAutomaton, + MultipleEntryVisiblyPushdownAutomaton, + SingleEntryVisiblyPushdownAutomaton, + VisiblyPushdownAutomaton, + ) + +#: Bottom-of-stack symbol used by constructed VPAs. +BOTTOM = "⊥" + +Word = tuple[Any, ...] + + +@dataclass +class NormalVPA: + """VPA with an explicit bottom symbol and only guarded returns.""" + + call_alphabet: frozenset[Any] + return_alphabet: frozenset[Any] + internal_alphabet: frozenset[Any] + states: set[Hashable] = field(default_factory=set) + initial: set[Hashable] = field(default_factory=set) + accepting: set[Hashable] = field(default_factory=set) + calls: set[tuple[Hashable, Any, Hashable, Any]] = field(default_factory=set) + internals: set[tuple[Hashable, Any, Hashable]] = field(default_factory=set) + returns: set[tuple[Hashable, Any, Any, Hashable]] = field(default_factory=set) + + @property + def stack_symbols(self) -> set[Any]: + pushed = {g for _q, _a, _t, g in self.calls} + guarded = {g for _q, _a, g, _t in self.returns if g != BOTTOM} + return pushed | guarded + + def call_map(self) -> dict[tuple[Hashable, Any], set[tuple[Hashable, Any]]]: + result: dict[tuple[Hashable, Any], set[tuple[Hashable, Any]]] = defaultdict(set) + for q, a, t, g in self.calls: + result[(q, a)].add((t, g)) + return result + + def internal_map(self) -> dict[tuple[Hashable, Any], set[Hashable]]: + result: dict[tuple[Hashable, Any], set[Hashable]] = defaultdict(set) + for q, a, t in self.internals: + result[(q, a)].add(t) + return result + + def return_map(self) -> dict[tuple[Hashable, Any], set[tuple[Any, Hashable]]]: + result: dict[tuple[Hashable, Any], set[tuple[Any, Hashable]]] = defaultdict(set) + for q, a, g, t in self.returns: + result[(q, a)].add((g, t)) + return result + + def copy_outgoing(self, source: Hashable, new_source: Hashable, map_target=lambda t: t) -> None: + """Give ``new_source`` copies of ``source``'s outgoing transitions.""" + self.calls |= {(new_source, a, map_target(t), g) for q, a, t, g in list(self.calls) if q == source} + self.internals |= {(new_source, a, map_target(t)) for q, a, t in list(self.internals) if q == source} + self.returns |= {(new_source, a, g, map_target(t)) for q, a, g, t in list(self.returns) if q == source} + + +def normalize(vpa: VisiblyPushdownAutomaton) -> NormalVPA: + """Return the normalized form of ``vpa``. + + A wildcard return (no stack symbol) fires on every stack symbol, and also on + the empty stack when ``vpa`` has a bottom symbol; a return guarded by the + bottom symbol fires only on the empty stack. + """ + bottom = vpa.bottom_stack_symbol + gamma = {g for g in vpa.stack_alphabet if g != bottom} + for transition in vpa.transitions(): + if transition.data.get(ATTR_KIND) == KIND_CALL: + gamma.add(transition.data.get(ATTR_STACK_SYMBOL)) + result = NormalVPA( + call_alphabet=frozenset(vpa.call_alphabet), + return_alphabet=frozenset(vpa.return_alphabet), + internal_alphabet=frozenset(vpa.internal_alphabet), + states=set(vpa.states()), + initial=set() if vpa.initial_state is None else {vpa.initial_state}, + accepting=set(vpa.accepting_states), + ) + for transition in vpa.transitions(): + data = transition.data + kind, symbol = data.get(ATTR_KIND), data.get(ATTR_SYMBOL) + if symbol is None: + continue + source, target = transition.source, transition.target + if kind == KIND_CALL: + result.calls.add((source, symbol, target, data.get(ATTR_STACK_SYMBOL))) + elif kind == KIND_INTERNAL: + result.internals.add((source, symbol, target)) + elif kind == KIND_RETURN: + guard = data.get(ATTR_STACK_SYMBOL) + if guard is None: + guards = set(gamma) | ({BOTTOM} if bottom is not None else set()) + elif bottom is not None and guard == bottom: + guards = {BOTTOM} + else: + guards = {guard} + result.returns |= {(source, symbol, g, target) for g in guards} + return result + + +def _merge_alphabets(*machines: NormalVPA) -> tuple[frozenset[Any], frozenset[Any], frozenset[Any]]: + calls = frozenset().union(*(m.call_alphabet for m in machines)) + returns = frozenset().union(*(m.return_alphabet for m in machines)) + internals = frozenset().union(*(m.internal_alphabet for m in machines)) + if len(calls) + len(returns) + len(internals) != len(calls | returns | internals): + raise ValueError("operands disagree on which symbols are calls, returns, and internals") + return calls, returns, internals + + +def _with_alphabets(machine: NormalVPA, calls, returns, internals) -> NormalVPA: + machine.call_alphabet, machine.return_alphabet, machine.internal_alphabet = calls, returns, internals + return machine + + +def _forward_trim(machine: NormalVPA) -> NormalVPA: + """Drop states unreachable from the initial states (ignoring the stack).""" + successors: dict[Hashable, set[Hashable]] = defaultdict(set) + for q, _a, t, _g in machine.calls: + successors[q].add(t) + for q, _a, t in machine.internals: + successors[q].add(t) + for q, _a, _g, t in machine.returns: + successors[q].add(t) + keep = set(machine.initial) + stack = list(keep) + while stack: + for target in successors[stack.pop()]: + if target not in keep: + keep.add(target) + stack.append(target) + machine.states &= keep + machine.accepting &= keep + machine.calls = {c for c in machine.calls if c[0] in keep} + machine.internals = {i for i in machine.internals if i[0] in keep} + machine.returns = {r for r in machine.returns if r[0] in keep} + return machine + + +def denormalize(machine: NormalVPA, cls: type | None = None, **extra: Any) -> Any: + """Build a VPA of class ``cls`` from a normalized form, relabeling states and stack symbols as integers.""" + from sofic.automata.vpa import VisiblyPushdownAutomaton + + cls = cls or VisiblyPushdownAutomaton + machine = _forward_trim(machine) + if len(machine.initial) > 1: + start = ("start", len(machine.states)) + machine.states.add(start) + for state in sorted(machine.initial, key=repr): + machine.copy_outgoing(state, start) + if machine.initial & machine.accepting: + machine.accepting.add(start) + machine.initial = {start} + + state_name = {state: index for index, state in enumerate(sorted(machine.states, key=repr))} + stack_name = {symbol: index for index, symbol in enumerate(sorted(machine.stack_symbols, key=repr))} + has_bottom = any(g == BOTTOM for _q, _a, g, _t in machine.returns) + stack_alphabet = set(stack_name.values()) | ({BOTTOM} if has_bottom else set()) + result = cls( + call_alphabet=machine.call_alphabet, + return_alphabet=machine.return_alphabet, + internal_alphabet=machine.internal_alphabet, + stack_alphabet=frozenset(stack_alphabet), + bottom_stack_symbol=BOTTOM if has_bottom else None, + initial_state=state_name[next(iter(machine.initial))] if machine.initial else None, + accepting_states=frozenset(state_name[q] for q in machine.accepting), + **extra, + ) + for state in state_name.values(): + result.graph.add_state(state) + for q, a, t, g in sorted(machine.calls, key=repr): + result.graph.add_transition( + state_name[q], state_name[t], **{ATTR_KIND: KIND_CALL, ATTR_SYMBOL: a, ATTR_STACK_SYMBOL: stack_name[g]} + ) + for q, a, t in sorted(machine.internals, key=repr): + result.graph.add_transition(state_name[q], state_name[t], **{ATTR_KIND: KIND_INTERNAL, ATTR_SYMBOL: a}) + for q, a, g, t in sorted(machine.returns, key=repr): + guard = BOTTOM if g == BOTTOM else stack_name[g] + result.graph.add_transition( + state_name[q], state_name[t], **{ATTR_KIND: KIND_RETURN, ATTR_SYMBOL: a, ATTR_STACK_SYMBOL: guard} + ) + return result + + +# --------------------------------------------------------------------- boolean operations + + +def union(left: NormalVPA, right: NormalVPA) -> NormalVPA: + """Disjoint sum of two normalized VPAs.""" + alphabets = _merge_alphabets(left, right) + result = NormalVPA(*alphabets) + for tag, machine in ((0, left), (1, right)): + + def lift(g: Any, tag: int = tag) -> Any: + return BOTTOM if g == BOTTOM else (tag, g) + + result.states |= {(tag, q) for q in machine.states} + result.initial |= {(tag, q) for q in machine.initial} + result.accepting |= {(tag, q) for q in machine.accepting} + result.calls |= {((tag, q), a, (tag, t), (tag, g)) for q, a, t, g in machine.calls} + result.internals |= {((tag, q), a, (tag, t)) for q, a, t in machine.internals} + result.returns |= {((tag, q), a, lift(g), (tag, t)) for q, a, g, t in machine.returns} + return result + + +def intersection(left: NormalVPA, right: NormalVPA) -> NormalVPA: + """Synchronized product of two normalized VPAs (reachable part).""" + alphabets = _merge_alphabets(left, right) + result = NormalVPA(*alphabets) + lc, li, lr = left.call_map(), left.internal_map(), left.return_map() + rc, ri, rr = right.call_map(), right.internal_map(), right.return_map() + frontier = [(p, q) for p in left.initial for q in right.initial] + result.initial = set(frontier) + result.states = set(frontier) + while frontier: + state = frontier.pop() + p, q = state + if p in left.accepting and q in right.accepting: + result.accepting.add(state) + successors = [] + for a in alphabets[2]: + for t1 in li.get((p, a), ()): + for t2 in ri.get((q, a), ()): + result.internals.add((state, a, (t1, t2))) + successors.append((t1, t2)) + for a in alphabets[0]: + for t1, g1 in lc.get((p, a), ()): + for t2, g2 in rc.get((q, a), ()): + result.calls.add((state, a, (t1, t2), (g1, g2))) + successors.append((t1, t2)) + for a in alphabets[1]: + for g1, t1 in lr.get((p, a), ()): + for g2, t2 in rr.get((q, a), ()): + if (g1 == BOTTOM) != (g2 == BOTTOM): + continue + guard = BOTTOM if g1 == BOTTOM else (g1, g2) + result.returns.add((state, a, guard, (t1, t2))) + successors.append((t1, t2)) + for successor in successors: + if successor not in result.states: + result.states.add(successor) + frontier.append(successor) + return result + + +def determinize(machine: NormalVPA, alphabets: tuple[frozenset, frozenset, frozenset] | None = None) -> NormalVPA: + """Alur-Madhusudan summary construction; the result is deterministic and complete.""" + calls_a, returns_a, internals_a = alphabets or ( + machine.call_alphabet, + machine.return_alphabet, + machine.internal_alphabet, + ) + cmap, imap, rmap = machine.call_map(), machine.internal_map(), machine.return_map() + identity = frozenset((q, q) for q in machine.states) + start = (identity, frozenset(machine.initial)) + result = NormalVPA(calls_a, returns_a, internals_a, states={start}, initial={start}) + + def after_internal(pairs, a): + return frozenset((p, t) for p, q in pairs for t in imap.get((q, a), ())) + + def after_pending_return(pairs, a): + return frozenset((p, t) for p, q in pairs for g, t in rmap.get((q, a), ()) if g == BOTTOM) + + def after_matched_return(caller_pairs, c, pairs, a): + inner: dict[Hashable, set[Hashable]] = defaultdict(set) + for q2, q3 in pairs: + inner[q2].add(q3) + out = set() + for p, q1 in caller_pairs: + for q2, g in cmap.get((q1, c), ()): + for q3 in inner.get(q2, ()): + out |= {(p, t) for guard, t in rmap.get((q3, a), ()) if guard == g} + return frozenset(out) + + stack_symbols: list[tuple[frozenset, frozenset, Any]] = [] + known_stack: set[tuple[frozenset, frozenset, Any]] = set() + done: set[tuple[Any, Any]] = set() + pending = [start] + while True: + while pending: + state = pending.pop() + pairs, current = state + if current & machine.accepting: + result.accepting.add(state) + successors = [] + for a in internals_a: + target = (after_internal(pairs, a), frozenset(t for q in current for t in imap.get((q, a), ()))) + successors.append(target) + result.internals.add((state, a, target)) + for c in calls_a: + entered = frozenset(t for q in current for t, _g in cmap.get((q, c), ())) + target = (identity, entered) + symbol = (pairs, current, c) + if symbol not in known_stack: + known_stack.add(symbol) + stack_symbols.append(symbol) + successors.append(target) + result.calls.add((state, c, target, symbol)) + for a in returns_a: + moved = after_pending_return(pairs, a) + current_after = frozenset(t for q in current for g, t in rmap.get((q, a), ()) if g == BOTTOM) + target = (moved, current_after) + successors.append(target) + result.returns.add((state, a, BOTTOM, target)) + for successor in successors: + if successor not in result.states: + result.states.add(successor) + pending.append(successor) + new_work = False + for state in list(result.states): + pairs, _current = state + for symbol in list(stack_symbols): + if (state, symbol) in done: + continue + done.add((state, symbol)) + caller_pairs, caller_current, c = symbol + for a in returns_a: + moved = after_matched_return(caller_pairs, c, pairs, a) + reached = after_matched_return(frozenset((q, q) for q in caller_current), c, pairs, a) + target = (moved, frozenset(t for _p, t in reached)) + result.returns.add((state, a, symbol, target)) + if target not in result.states: + result.states.add(target) + pending.append(target) + new_work = True + if not new_work and not pending: + return result + + +def complement(machine: NormalVPA, alphabets=None) -> NormalVPA: + """Complement over the visible alphabet (or ``alphabets``).""" + deterministic = determinize(machine, alphabets) + deterministic.accepting = deterministic.states - deterministic.accepting + return deterministic + + +def difference(left: NormalVPA, right: NormalVPA) -> NormalVPA: + alphabets = _merge_alphabets(left, right) + return intersection(_with_alphabets(left, *alphabets), complement(right, alphabets)) + + +# ----------------------------------------------------------------- structural operations + + +def concat(left: NormalVPA, right: NormalVPA) -> NormalVPA: + """Concatenation; the right factor starts with an empty stack of its own.""" + alphabets = _merge_alphabets(left, right) + result = NormalVPA(*alphabets) + left_tops = [("A", g) for g in left.stack_symbols] + [BOTTOM] + + for q in left.states: + result.states.add(("A", q)) + result.initial = {("A", q) for q in left.initial} + result.calls |= {(("A", q), a, ("A", t), ("A", g)) for q, a, t, g in left.calls} + result.internals |= {(("A", q), a, ("A", t)) for q, a, t in left.internals} + result.returns |= {(("A", q), a, g if g == BOTTOM else ("A", g), ("A", t)) for q, a, g, t in left.returns} + + for q in right.states: + for empty in (True, False): + state = ("B", q, empty) + result.states.add(state) + result.internals |= {(state, a, ("B", t, empty)) for p, a, t in right.internals if p == q} + result.calls |= {(state, a, ("B", t, False), ("B", g, empty)) for p, a, t, g in right.calls if p == q} + for p, a, g, t in right.returns: + if p != q: + continue + if empty and g == BOTTOM: + result.returns |= {(state, a, top, ("B", t, True)) for top in left_tops} + elif not empty and g != BOTTOM: + result.returns |= {(state, a, ("B", g, bit), ("B", t, bit)) for bit in (True, False)} + if q in right.accepting: + result.accepting.add(state) + + epsilon_in_right = bool(right.initial & right.accepting) + for q in left.accepting: + for start in right.initial: + result.copy_outgoing(("B", start, True), ("A", q)) + if epsilon_in_right: + result.accepting.add(("A", q)) + return result + + +def kleene_star(machine: NormalVPA) -> NormalVPA: + """Kleene star; each factor starts with an empty stack of its own.""" + result = NormalVPA(machine.call_alphabet, machine.return_alphabet, machine.internal_alphabet) + tops = [(g, bit) for g in machine.stack_symbols for bit in (True, False)] + [BOTTOM] + for q in machine.states: + for empty in (True, False): + state = (q, empty) + result.states.add(state) + result.internals |= {(state, a, (t, empty)) for p, a, t in machine.internals if p == q} + result.calls |= {(state, a, (t, False), (g, empty)) for p, a, t, g in machine.calls if p == q} + for p, a, g, t in machine.returns: + if p != q: + continue + if empty and g == BOTTOM: + result.returns |= {(state, a, top, (t, True)) for top in tops} + elif not empty and g != BOTTOM: + result.returns |= {(state, a, (g, bit), (t, bit)) for bit in (True, False)} + if q in machine.accepting: + result.accepting.add(state) + start = ("star-start",) + result.states.add(start) + result.initial = {start} + result.accepting.add(start) + restart_sources = [start] + [(q, empty) for q in machine.accepting for empty in (True, False)] + for source in restart_sources: + for initial in machine.initial: + result.copy_outgoing((initial, True), source) + return result + + +# --------------------------------------------------------------------- decision procedures + + +def well_matched_summaries(machine: NormalVPA) -> dict[tuple[Hashable, Hashable], Word]: + """Pairs ``(p, q)`` joined by a well-matched word, each with a witness word.""" + witness: dict[tuple[Hashable, Hashable], Word] = {(q, q): () for q in machine.states} + rmap = machine.return_map() + changed = True + while changed: + changed = False + updates: dict[tuple[Hashable, Hashable], Word] = {} + by_source: dict[Hashable, list[tuple[Hashable, Word]]] = defaultdict(list) + for (p, q), word in witness.items(): + by_source[p].append((q, word)) + for (p, q), word in witness.items(): + for source, a, target in machine.internals: + if source == q: + updates.setdefault((p, target), (*word, a)) + for r, word2 in by_source.get(q, ()): + updates.setdefault((p, r), word + word2) + for p0, c, p, g in machine.calls: + for q, word in by_source.get(p, ()): + for a in machine.return_alphabet: + for guard, target in rmap.get((q, a), ()): + if guard == g: + updates.setdefault((p0, target), (c, *word, a)) + for pair, word in updates.items(): + if pair not in witness or len(word) < len(witness[pair]): + if pair not in witness: + changed = True + witness[pair] = word + return witness + + +def accepted_word(machine: NormalVPA) -> Word | None: + """Return a short accepted word, or ``None`` when the language is empty.""" + summaries = well_matched_summaries(machine) + by_source: dict[Hashable, list[tuple[Hashable, Word]]] = defaultdict(list) + for (p, q), word in summaries.items(): + if word: + by_source[p].append((q, word)) + queue: list[tuple[int, int, Hashable, bool, Word]] = [] + counter = 0 + for q in machine.initial: + heapq.heappush(queue, (0, counter, q, True, ())) + counter += 1 + seen: set[tuple[Hashable, bool]] = set() + while queue: + _length, _tie, state, stack_empty, word = heapq.heappop(queue) + if (state, stack_empty) in seen: + continue + seen.add((state, stack_empty)) + if state in machine.accepting: + return word + moves: list[tuple[Hashable, bool, Word]] = [(t, stack_empty, word + w) for t, w in by_source.get(state, ())] + moves += [(t, False, (*word, a)) for q, a, t, _g in machine.calls if q == state] + if stack_empty: + moves += [(t, True, (*word, a)) for q, a, g, t in machine.returns if q == state and g == BOTTOM] + for target, empty, extended in moves: + if (target, empty) not in seen: + heapq.heappush(queue, (len(extended), counter, target, empty, extended)) + counter += 1 + return None + + +def is_empty(machine: NormalVPA) -> bool: + return accepted_word(machine) is None + + +def _unmatched_tracker(alphabets: tuple[frozenset, frozenset, frozenset]) -> NormalVPA: + """Deterministic VPA accepting the words with a pending call or a pending return.""" + tracker = NormalVPA(*alphabets) + for pending_return in (False, True): + for empty in (True, False): + state = (pending_return, empty) + tracker.states.add(state) + if pending_return or not empty: + tracker.accepting.add(state) + tracker.internals |= {(state, a, state) for a in alphabets[2]} + tracker.calls |= {(state, c, (pending_return, False), empty) for c in alphabets[0]} + for a in alphabets[1]: + if empty: + tracker.returns.add((state, a, BOTTOM, (True, True))) + else: + tracker.returns |= {(state, a, bit, (pending_return, bit)) for bit in (True, False)} + tracker.initial = {(False, True)} + return tracker + + +def has_unmatched_word(machine: NormalVPA) -> bool: + """Whether the language contains a word with a pending call or a pending return.""" + alphabets = (machine.call_alphabet, machine.return_alphabet, machine.internal_alphabet) + return not is_empty(intersection(machine, _unmatched_tracker(alphabets))) + + +# ------------------------------------------------------------------------ modular forms + + +def _modular_conversion( + vpa: VisiblyPushdownAutomaton, + call_partition: Mapping[Any, Hashable] | None, + *, + multiple_entry: bool, +) -> Any: + from sofic.automata.vpa import MultipleEntryVisiblyPushdownAutomaton, SingleEntryVisiblyPushdownAutomaton + + machine = normalize(vpa) + if has_unmatched_word(machine): + raise NonWellMatchedLanguageError( + "modular (single- or multiple-entry) forms are built here for well-matched languages only" + ) + base = "base" + partition = dict(call_partition) if call_partition is not None else {c: ("module", c) for c in vpa.call_alphabet} + if base in partition.values(): + raise ValueError("calls cannot target the base module") + missing = set(vpa.call_alphabet) - set(partition) + if missing: + raise ValueError(f"call_partition is missing calls {sorted(missing, key=repr)!r}") + + cmap, imap, rmap = machine.call_map(), machine.internal_map(), machine.return_map() + identity = frozenset((q, q) for q in machine.states) + + def compose_internal(pairs, a): + return frozenset((p, t) for p, q in pairs for t in imap.get((q, a), ())) + + def wrap(caller_pairs, c, pairs, a): + inner: dict[Hashable, set[Hashable]] = defaultdict(set) + for q2, q3 in pairs: + inner[q2].add(q3) + out = set() + for p, q1 in caller_pairs: + for q2, g in cmap.get((q1, c), ()): + for q3 in inner.get(q2, ()): + out |= {(p, t) for guard, t in rmap.get((q3, a), ()) if guard == g} + return frozenset(out) + + # A raw state is (module, entering_call, summary); entering_call is only kept + # for the multiple-entry form, where it replaces the call symbol on the stack. + def entry(c): + return (partition[c], c if multiple_entry else None, identity) + + start = (base, None, identity) + label: dict[tuple, int] = {start: 0} + order = [start] + internals: set[tuple[int, Any, int]] = set() + calls: set[tuple[int, Any, int, Any]] = set() + returns: set[tuple[int, Any, Any, int]] = set() + pushes: list[tuple[int, Any]] = [] + done: set[tuple[int, Any]] = set() + + def name(raw): + if raw not in label: + label[raw] = len(order) + order.append(raw) + return label[raw] + + index = 0 + while True: + while index < len(order): + raw = order[index] + module, entered, pairs = raw + source = label[raw] + for a in vpa.internal_alphabet: + internals.add((source, a, name((module, entered, compose_internal(pairs, a))))) + for c in vpa.call_alphabet: + push = source if multiple_entry else (source, c) + calls.add((source, c, name(entry(c)), push)) + if (source, c) not in pushes: + pushes.append((source, c)) + index += 1 + grew = False + for raw in list(order): + module, entered, pairs = raw + source = label[raw] + for caller, c in list(pushes): + if (source, (caller, c)) in done: + continue + if multiple_entry and entered != c: + continue + if not multiple_entry and module != partition[c]: + continue + done.add((source, (caller, c))) + caller_module, caller_entered, caller_pairs = order[caller] + guard = caller if multiple_entry else (caller, c) + for a in vpa.return_alphabet: + before = len(order) + target = name((caller_module, caller_entered, wrap(caller_pairs, c, pairs, a))) + returns.add((source, a, guard, target)) + grew = grew or len(order) > before + if not grew and index >= len(order): + break + + modules: dict[Hashable, set[int]] = defaultdict(set) + for raw, state in label.items(): + modules[raw[0]].add(state) + accepting = { + label[raw] + for raw in order + if raw[0] == base and any(p in machine.initial and q in machine.accepting for p, q in raw[2]) + } + used_partition = {c: m for c, m in partition.items() if c in vpa.call_alphabet} + call_entries = {c: label[entry(c)] for c in vpa.call_alphabet} + common = { + "call_alphabet": vpa.call_alphabet, + "return_alphabet": vpa.return_alphabet, + "internal_alphabet": vpa.internal_alphabet, + "stack_alphabet": frozenset(push for _s, _c, _t, push in calls), + "initial_state": 0, + "accepting_states": frozenset(accepting), + "modules": {m: frozenset(states) for m, states in modules.items()}, + "base_module": base, + "call_partition": used_partition, + "call_entries": call_entries, + } + if multiple_entry: + entries: dict[Hashable, set[int]] = defaultdict(set) + entries[base].add(0) + for c, state in call_entries.items(): + entries[partition[c]].add(state) + result = MultipleEntryVisiblyPushdownAutomaton( + entry_states={m: frozenset(s) for m, s in entries.items()}, **common + ) + else: + result = SingleEntryVisiblyPushdownAutomaton( + entry_states={partition[c]: state for c, state in call_entries.items()}, **common + ) + for state in range(len(order)): + result.graph.add_state(state) + for s, c, t, push in sorted(calls, key=repr): + result.add_call_transition(s, t, c, push) + for s, a, t in sorted(internals, key=repr): + result.add_internal_transition(s, t, a) + for s, a, guard, t in sorted(returns, key=repr): + result.add_return_transition(s, t, a, guard) + result.validate() + return result + + +def to_single_entry( + vpa: VisiblyPushdownAutomaton, call_partition: Mapping[Any, Hashable] | None = None +) -> SingleEntryVisiblyPushdownAutomaton: + """Convert a VPA of a well-matched language into a single-entry modular VPA. + + ``call_partition`` maps each call symbol to a module (default: one module + per call symbol). Raises + :class:`~sofic.exceptions.NonWellMatchedLanguageError` when the language has + a word with a pending call or return, which a single-entry automaton that + forgets its caller cannot accept. + """ + return _modular_conversion(vpa, call_partition, multiple_entry=False) + + +def to_multiple_entry( + vpa: VisiblyPushdownAutomaton, call_partition: Mapping[Any, Hashable] | None = None +) -> MultipleEntryVisiblyPushdownAutomaton: + """Convert a VPA of a well-matched language into a multiple-entry modular VPA. + + Each module has one entry per call symbol assigned to it, and calls push + only the caller state. + """ + return _modular_conversion(vpa, call_partition, multiple_entry=True) + + +def determinize_vpa(vpa: VisiblyPushdownAutomaton) -> DeterministicVisiblyPushdownAutomaton: + from sofic.automata.vpa import DeterministicVisiblyPushdownAutomaton + + return denormalize(determinize(normalize(vpa)), DeterministicVisiblyPushdownAutomaton) diff --git a/sofic/exceptions.py b/sofic/exceptions.py index b8a7658..e9d39ff 100644 --- a/sofic/exceptions.py +++ b/sofic/exceptions.py @@ -43,3 +43,7 @@ class LumpabilityError(SoficValidationError): class InfiniteTransductionError(SoficError): """Raised when a finite input has infinitely many transducer outputs.""" + + +class NonWellMatchedLanguageError(SoficValidationError): + """Raised when a construction defined for well-matched languages meets a pending call or return.""" diff --git a/sofic/serialization.py b/sofic/serialization.py index dac0024..c660409 100644 --- a/sofic/serialization.py +++ b/sofic/serialization.py @@ -106,8 +106,6 @@ def _metadata_for(model: StateMachine, spec: _ModelSpec) -> dict[str, Any]: def _build_model(spec: _ModelSpec, graph: TransitionGraph, metadata: dict[str, Any]) -> StateMachine: - if spec.builder == "composite_vpa": - return spec.cls(operation=metadata["operation"], operands=metadata["operands"]) if spec.builder == "hidden_hmm": return spec.cls(graph=graph, observation_alphabet=_observation_alphabet(graph), **metadata) if spec.builder == "pfa": @@ -312,7 +310,6 @@ def _specs() -> tuple[_ModelSpec, ...]: from sofic.automata.vpa import ( CallDrivenAutomaton, CanonicalVisiblyPushdownAutomaton, - CompositeVisiblyPushdownAutomaton, DeterministicVisiblyPushdownAutomaton, MultipleEntryVisiblyPushdownAutomaton, SingleEntryVisiblyPushdownAutomaton, @@ -409,7 +406,6 @@ def _specs() -> tuple[_ModelSpec, ...]: _spec(MultipleEntryVisiblyPushdownAutomaton, (*cda, "entry_states")), _spec(SingleEntryVisiblyPushdownAutomaton, (*cda, "entry_states")), _spec(CanonicalVisiblyPushdownAutomaton, (*vpa, "summary_representatives")), - _spec(CompositeVisiblyPushdownAutomaton, ("operation", "operands"), builder="composite_vpa"), _spec(StochasticModel, stochastic), _spec(HiddenMarkovModel, hidden, builder="hidden_hmm"), _spec(MarkovChain, stochastic), diff --git a/tests/test_canonical_rfsa.py b/tests/test_canonical_rfsa.py new file mode 100644 index 0000000..8f167b9 --- /dev/null +++ b/tests/test_canonical_rfsa.py @@ -0,0 +1,125 @@ +"""Exact canonical RFSA, maximized prime átomaton, residuals, and atoms.""" + +import pytest +from hypothesis import given, settings + +from sofic.automata.algorithms import equivalent +from sofic.automata.atomaton import Atomaton, MaximizedPrimeAtomaton, is_atomic +from sofic.automata.dfa import DFA +from sofic.automata.languages.atoms import atoms, prime_atoms +from sofic.automata.languages.base import AutomatonLanguage +from sofic.automata.languages.residuals import prime_residuals +from sofic.automata.nfa import NFA +from sofic.automata.rfsa import CanonicalRFSA, ResidualFiniteStateAutomaton +from sofic.exceptions import SoficValidationError +from sofic.testing.strategies import dfas + + +def _a_then_n_symbols(n: int) -> NFA: + """NFA for Sigma* a Sigma^n: minimal DFA 2^(n+1) states, canonical RFSA n+2 (Denis et al. 2002).""" + nfa = NFA(input_alphabet=frozenset("ab"), initial_states=frozenset({0}), accepting_states=frozenset({n + 1})) + for state in range(n + 2): + nfa.graph.add_state(state) + nfa.add_transition(0, 0, "a") + nfa.add_transition(0, 0, "b") + nfa.add_transition(0, 1, "a") + for state in range(1, n + 1): + nfa.add_transition(state, state + 1, "a") + nfa.add_transition(state, state + 1, "b") + return nfa + + +@pytest.mark.parametrize("n", [1, 2, 3]) +def test_canonical_rfsa_is_exponentially_smaller_than_minimal_dfa(n): + nfa = _a_then_n_symbols(n) + assert len(list(nfa.minimize().states())) == 2 ** (n + 1) + rfsa = CanonicalRFSA.from_language(nfa) + assert len(list(rfsa.states())) == n + 2 + assert equivalent(rfsa, nfa) + rfsa.validate() + + +def test_prime_residual_count_matches_canonical_rfsa(): + assert len(prime_residuals(AutomatonLanguage(_a_then_n_symbols(1)))) == 3 + + +def test_canonical_rfsa_is_invariant_under_state_renaming(): + dfa = _a_then_n_symbols(2).minimize() + renamed = DFA( + input_alphabet=dfa.input_alphabet, + initial_states=frozenset(("r", s) for s in dfa.initial_states), + accepting_states=frozenset(("r", s) for s in dfa.accepting_states), + ) + for state in dfa.states(): + renamed.graph.add_state(("r", state)) + for t in dfa.transitions(): + renamed.add_transition(("r", t.source), ("r", t.target), t.data["symbol"]) + original = CanonicalRFSA.from_language(dfa) + other = CanonicalRFSA.from_language(renamed) + assert len(list(original.states())) == len(list(other.states())) + assert len(list(original.transitions())) == len(list(other.transitions())) + + +def test_residual_validate_rejects_non_residual_state(): + # State "x" accepts {b}, which is not a residual of L = {ab}. + nfa = ResidualFiniteStateAutomaton( + input_alphabet=frozenset("ab"), initial_states=frozenset({"s"}), accepting_states=frozenset({"f"}) + ) + for state in ("s", "m", "f", "x"): + nfa.graph.add_state(state) + nfa.add_transition("s", "m", "a") + nfa.add_transition("m", "f", "b") + nfa.add_transition("x", "f", "b") + nfa.add_transition("x", "f", "a") + with pytest.raises(SoficValidationError, match="not a residual"): + nfa.validate() + + +@settings(max_examples=60, deadline=None) +@given(dfas(min_states=1, max_states=4)) +def test_canonical_rfsa_and_prime_atomaton_recognize_the_language(dfa): + rfsa = CanonicalRFSA.from_language(dfa) + assert equivalent(rfsa, dfa) + rfsa.validate() + assert len(list(rfsa.states())) <= len(list(dfa.minimize().states())) + + mpa = MaximizedPrimeAtomaton.from_language(dfa) + assert equivalent(mpa, dfa) + mpa.validate() + + +def test_maximized_prime_atomaton_need_not_be_atomic(): + """Its right languages lie between an atom and a maximized atom (Tamm 2015), not on atoms.""" + from sofic.automata.icdfa import icdfa_string_to_dfa + + dfa = icdfa_string_to_dfa((0, 1, 0, 2, 0, 1), ("0", "1"), n=3, k=2, final_states=frozenset({0, 1})) + mpa = MaximizedPrimeAtomaton.from_language(dfa) + assert equivalent(mpa, dfa) + assert not is_atomic(mpa) + assert is_atomic(Atomaton.from_language(dfa)) + + +@settings(max_examples=40, deadline=None) +@given(dfas(min_states=1, max_states=4)) +def test_dual_round_trip(dfa): + rfsa = CanonicalRFSA.from_language(dfa) + mpa_of_reverse = rfsa.dual() + assert isinstance(mpa_of_reverse, MaximizedPrimeAtomaton) + assert equivalent(mpa_of_reverse, dfa.reverse()) + back = mpa_of_reverse.dual() + assert isinstance(back, CanonicalRFSA) + assert equivalent(back, dfa) + expected = MaximizedPrimeAtomaton.from_language(dfa.reverse()) + assert len(list(mpa_of_reverse.states())) == len(list(expected.states())) + + +@settings(max_examples=40, deadline=None) +@given(dfas(min_states=1, max_states=4)) +def test_atoms_label_atomaton_states_and_prime_atoms_label_prime_atomaton(dfa): + if not dfa.minimize().accepting_states: + return + language = AutomatonLanguage(dfa) + assert len(atoms(language)) == len(list(Atomaton.from_language(dfa).states())) + assert len(prime_atoms(language)) == len(list(MaximizedPrimeAtomaton.from_language(dfa).states())) + for atom in atoms(language): + assert atom.automaton.minimize().accepting_states diff --git a/tests/test_learning.py b/tests/test_learning.py index 7bfe988..98d56bd 100644 --- a/tests/test_learning.py +++ b/tests/test_learning.py @@ -1,27 +1,80 @@ -"""Tests for NL* learning.""" +"""Tests for NL* learning of canonical RFSAs and maximized prime átomata.""" +import pytest +from hypothesis import given, settings + +from sofic.automata.active import AutomatonEquivalenceOracle, LanguageMembershipOracle +from sofic.automata.algorithms import equivalent +from sofic.automata.atomaton import MaximizedPrimeAtomaton from sofic.automata.dfa import DFA -from sofic.automata.languages.base import AutomatonLanguage -from sofic.automata.learning import learn_maximized_prime_atomaton +from sofic.automata.learning import learn_prime_atomaton_nlstar, learn_rfsa_from_language, learn_rfsa_nlstar +from sofic.automata.nfa import NFA +from sofic.automata.rfsa import CanonicalRFSA +from sofic.testing.strategies import dfas + + +def _a_then_n_symbols(n: int) -> NFA: + nfa = NFA(input_alphabet=frozenset("ab"), initial_states=frozenset({0}), accepting_states=frozenset({n + 1})) + for state in range(n + 2): + nfa.graph.add_state(state) + nfa.add_transition(0, 0, "a") + nfa.add_transition(0, 0, "b") + nfa.add_transition(0, 1, "a") + for state in range(1, n + 1): + nfa.add_transition(state, state + 1, "a") + nfa.add_transition(state, state + 1, "b") + return nfa + + +def _signature(aut) -> tuple[int, int, int, int]: + return ( + len(list(aut.states())), + len(list(aut.transitions())), + len(aut.initial_states), + len(aut.accepting_states), + ) -def _teacher_dfa() -> AutomatonLanguage: +@pytest.mark.parametrize("n", [1, 2, 3]) +def test_nlstar_learns_small_canonical_rfsa_of_exponential_dfa(n): + target = _a_then_n_symbols(n) + learned = learn_rfsa_from_language(target, frozenset("ab")) + assert isinstance(learned, CanonicalRFSA) + assert equivalent(learned, target) + assert len(list(learned.states())) == n + 2 + assert _signature(learned) == _signature(CanonicalRFSA.from_language(target)) + + +def test_nlstar_with_bounded_oracle_on_explicit_membership(): dfa = DFA( - input_alphabet=frozenset({"a", "b"}), - initial_states=frozenset({"q0"}), - accepting_states=frozenset({"q1"}), + input_alphabet=frozenset({"a", "b"}), initial_states=frozenset({"q0"}), accepting_states=frozenset({"q1"}) ) dfa.graph.add_state("q0") dfa.graph.add_state("q1") dfa.add_transition("q0", "q1", "a") - dfa.add_transition("q0", "q0", "b") dfa.add_transition("q1", "q1", "a") - dfa.add_transition("q1", "q0", "b") - return AutomatonLanguage(dfa) + dfa.add_transition("q1", "q1", "b") + learned = learn_rfsa_from_language(set(dfa.iter_language(max_length=6)), frozenset("ab"), max_length=5) + for word in dfa.iter_language(max_length=5): + assert learned.recognizes(word) + +@settings(max_examples=40, deadline=None) +@given(dfas(min_states=1, max_states=4)) +def test_nlstar_matches_canonical_rfsa_construction(dfa): + alphabet = frozenset({"0", "1"}) + learned = learn_rfsa_nlstar(alphabet, LanguageMembershipOracle(dfa), AutomatonEquivalenceOracle(dfa, alphabet)) + assert equivalent(learned, dfa) + assert _signature(learned) == _signature(CanonicalRFSA.from_language(dfa)) -def test_learn_mpa(): - teacher = _teacher_dfa() - learned = learn_maximized_prime_atomaton(teacher, frozenset({"a", "b"})) - learned.validate() - assert learned.recognizes(("a",)) == (("a",) in teacher) + +@settings(max_examples=25, deadline=None) +@given(dfas(min_states=1, max_states=4)) +def test_prime_atomaton_learner_matches_construction(dfa): + alphabet = frozenset({"0", "1"}) + learned = learn_prime_atomaton_nlstar( + alphabet, LanguageMembershipOracle(dfa), AutomatonEquivalenceOracle(dfa, alphabet) + ) + assert isinstance(learned, MaximizedPrimeAtomaton) + assert equivalent(learned, dfa) + assert _signature(learned) == _signature(MaximizedPrimeAtomaton.from_language(dfa)) diff --git a/tests/test_vpa_constructions.py b/tests/test_vpa_constructions.py new file mode 100644 index 0000000..cb34e65 --- /dev/null +++ b/tests/test_vpa_constructions.py @@ -0,0 +1,210 @@ +"""Concrete VPA constructions checked against brute-force reference semantics.""" + +from itertools import product + +import pytest +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from sofic.automata.nwa import NestedWordAutomaton +from sofic.automata.vpa import ( + CanonicalVisiblyPushdownAutomaton, + DeterministicVisiblyPushdownAutomaton, + MultipleEntryVisiblyPushdownAutomaton, + SingleEntryVisiblyPushdownAutomaton, + VisiblyPushdownAutomaton, +) +from sofic.automata.vpa_constructions import to_multiple_entry, to_single_entry +from sofic.exceptions import NonWellMatchedLanguageError + +CALLS, RETURNS, INTERNALS = frozenset({"c"}), frozenset({"r"}), frozenset({"i"}) +SYMBOLS = ("c", "r", "i") +MAX_LENGTH = 6 +WORDS = [word for n in range(MAX_LENGTH + 1) for word in product(SYMBOLS, repeat=n)] + + +@st.composite +def vpas(draw, max_states: int = 3): + """Small, possibly nondeterministic VPAs over one call, return, and internal symbol.""" + n = draw(st.integers(1, max_states)) + states = list(range(n)) + with_bottom = draw(st.booleans()) + vpa = VisiblyPushdownAutomaton( + call_alphabet=CALLS, + return_alphabet=RETURNS, + internal_alphabet=INTERNALS, + stack_alphabet=frozenset({"A", "B"} | ({"Z"} if with_bottom else set())), + bottom_stack_symbol="Z" if with_bottom else None, + initial_state=0, + accepting_states=frozenset(draw(st.sets(st.sampled_from(states)))), + ) + for state in states: + vpa.graph.add_state(state) + state = st.sampled_from(states) + for source, target in draw(st.lists(st.tuples(state, state), max_size=4)): + vpa.add_internal_transition(source, target, "i") + for source, target, push in draw(st.lists(st.tuples(state, state, st.sampled_from("AB")), max_size=3)): + vpa.add_call_transition(source, target, "c", push) + guards = ["A", "B", None] + (["Z"] if with_bottom else []) + for source, target, guard in draw(st.lists(st.tuples(state, state, st.sampled_from(guards)), max_size=3)): + vpa.add_return_transition(source, target, "r", guard) + return vpa + + +def _language(recognize) -> frozenset[tuple[str, ...]]: + return frozenset(word for word in WORDS if recognize(word)) + + +def _concat_reference(left, right): + return lambda w: any(left(w[:k]) and right(w[k:]) for k in range(len(w) + 1)) + + +def _star_reference(inner): + def recognize(word): + accepted = [True] + [False] * len(word) + for end in range(1, len(word) + 1): + accepted[end] = any(accepted[start] and inner(word[start:end]) for start in range(end)) + return accepted[-1] + + return recognize + + +SETTINGS = settings(max_examples=40, deadline=None, suppress_health_check=[HealthCheck.too_slow]) + + +@SETTINGS +@given(vpas(), vpas()) +def test_boolean_operations_match_reference(left, right): + a, b = _language(left.recognizes), _language(right.recognizes) + assert _language(left.union(right).recognizes) == a | b + assert _language(left.intersection(right).recognizes) == a & b + assert _language(left.difference(right).recognizes) == a - b + assert _language(left.complement().recognizes) == frozenset(WORDS) - a + + +@SETTINGS +@given(vpas(), vpas()) +def test_concatenation_matches_reference(left, right): + reference = _concat_reference(left.recognizes, right.recognizes) + assert _language(left.concat(right).recognizes) == _language(reference) + + +@SETTINGS +@given(vpas(max_states=2)) +def test_kleene_star_matches_reference(vpa): + assert _language(vpa.kleene_star().recognizes) == _language(_star_reference(vpa.recognizes)) + + +@SETTINGS +@given(vpas()) +def test_determinization_preserves_language_and_is_deterministic(vpa): + deterministic = vpa.determinize() + assert isinstance(deterministic, DeterministicVisiblyPushdownAutomaton) + deterministic.validate() + assert _language(deterministic.recognizes) == _language(vpa.recognizes) + assert DeterministicVisiblyPushdownAutomaton.from_vpa(vpa).equivalent(vpa) + + +@SETTINGS +@given(vpas()) +def test_double_complement_and_decision_procedures(vpa): + assert vpa.complement().complement().equivalent(vpa) + assert vpa.equivalent(vpa) + assert vpa.includes(vpa.intersection(vpa.complement())) + sample_accepts = any(vpa.recognizes(word) for word in WORDS) + witness = vpa.accepted_word() + assert vpa.is_empty() == (witness is None) + if witness is not None: + assert vpa.recognizes(witness) + if sample_accepts: + assert not vpa.is_empty() + assert vpa.union(vpa.complement()).is_universal() + + +@SETTINGS +@given(vpas()) +def test_has_unmatched_word_matches_reference(vpa): + def unmatched(word): + depth = 0 + for symbol in word: + if symbol == "c": + depth += 1 + elif symbol == "r": + if depth == 0: + return True + depth -= 1 + return depth > 0 + + if any(vpa.recognizes(w) and unmatched(w) for w in WORDS): + assert vpa.has_unmatched_word() + if not vpa.has_unmatched_word(): + assert not any(vpa.recognizes(w) and unmatched(w) for w in WORDS) + + +@SETTINGS +@given(vpas()) +def test_modular_conversions_preserve_well_matched_languages(vpa): + if vpa.has_unmatched_word(): + with pytest.raises(NonWellMatchedLanguageError): + to_single_entry(vpa) + return + reference = _language(vpa.recognizes) + single = to_single_entry(vpa) + multiple = to_multiple_entry(vpa) + assert _language(single.recognizes) == reference + assert _language(multiple.recognizes) == reference + assert _language(SingleEntryVisiblyPushdownAutomaton.minimize(vpa).recognizes) == reference + assert _language(MultipleEntryVisiblyPushdownAutomaton.minimize(vpa).recognizes) == reference + canonical = CanonicalVisiblyPushdownAutomaton.from_vpa(vpa) + assert _language(canonical.recognizes) == reference + + +def _pending_call_vpa() -> VisiblyPushdownAutomaton: + vpa = VisiblyPushdownAutomaton( + call_alphabet=CALLS, + stack_alphabet=frozenset({"S"}), + initial_state=0, + accepting_states=frozenset({1}), + ) + vpa.graph.add_state(0) + vpa.graph.add_state(1) + vpa.add_call_transition(0, 1, "c", "S") + return vpa + + +def test_canonical_vpa_rejects_languages_with_pending_calls(): + vpa = _pending_call_vpa() + assert vpa.recognizes(("c",)) + with pytest.raises(NonWellMatchedLanguageError): + CanonicalVisiblyPushdownAutomaton.from_vpa(vpa) + + +def test_nondeterministic_from_vpa_determinizes(): + vpa = VisiblyPushdownAutomaton(internal_alphabet=INTERNALS, initial_state=0, accepting_states=frozenset({1})) + vpa.graph.add_state(0) + vpa.graph.add_state(1) + vpa.add_internal_transition(0, 0, "i") + vpa.add_internal_transition(0, 1, "i") + deterministic = DeterministicVisiblyPushdownAutomaton.from_vpa(vpa) + deterministic.validate() + assert deterministic.recognizes(("i", "i")) + assert not deterministic.recognizes(()) + + +def test_nwa_operations_delegate_to_vpas(): + nwa = NestedWordAutomaton( + call_alphabet=frozenset({"x"}), + return_alphabet=frozenset({"x"}), + internal_alphabet=frozenset(), + hier_alphabet=frozenset({"H"}), + initial_state=0, + accepting_states=frozenset({0}), + ) + nwa.graph.add_state(0) + nwa.add_call_transition(0, 0, "x", "H") + nwa.add_return_transition(0, 0, "x", "H") + complement = nwa.complement() + assert not nwa.is_empty() + assert nwa.union(complement).is_universal() + assert nwa.intersection(complement).is_empty() + assert nwa.kleene_star().equivalent(nwa) diff --git a/tests/test_yaml.py b/tests/test_yaml.py index b5d8237..0ce8009 100644 --- a/tests/test_yaml.py +++ b/tests/test_yaml.py @@ -13,7 +13,6 @@ from sofic.automata.vpa import ( CallDrivenAutomaton, CanonicalVisiblyPushdownAutomaton, - CompositeVisiblyPushdownAutomaton, DeterministicVisiblyPushdownAutomaton, MultipleEntryVisiblyPushdownAutomaton, SingleEntryVisiblyPushdownAutomaton, @@ -221,11 +220,10 @@ def test_vpa_variants_round_trip(): _round_trip(canonical) -def test_composite_vpa_round_trip(): - union = CompositeVisiblyPushdownAutomaton(operation="union", operands=(_base_vpa(), _base_vpa())) +def test_constructed_vpa_round_trip(): + union = _base_vpa().union(_base_vpa()) restored = _round_trip(union) - assert restored.operation == "union" - assert len(restored.operands) == 2 + assert restored.equivalent(union) def test_stochastic_generators_round_trip():