Assignments 7 and 8

Assignment 7 consists of Tasks 1–3, and Assignment 8 consists of Tasks 4–5.

In these assignments, you will implement finite state automata for generating, recognizing, and parsing English syllables and words (Assignment 7), as well as finite state transducers for recognizing, parsing, and transducing English sentences (Assignment 8). We will use the FiniteStateAutomaton class developed in class and a FiniteStateTransducer class that builds on it.

Finite State Automata

We begin with two utilities used for determinization and epsilon closure.

from collections.abc import Iterable, Iterator
from itertools import chain, combinations


def powerset[T](iterable: Iterable[T]) -> Iterator[tuple[T, ...]]:
    """https://docs.python.org/3/library/itertools.html#recipes"""
    s = list(iterable)
    return chain.from_iterable(combinations(s, r) for r in range(len(s) + 1))


def transitive_closure[T](edges: set[tuple[T, T]]) -> set[tuple[T, T]]:
    """Compute the transitive closure of a graph.

    Parameters
    ----------
    edges : set[tuple[T, T]]
        The graph to compute the closure of, represented as
        a set of directed edges.

    Returns
    -------
    set[tuple[T, T]]
        The transitive closure of the input graph.
    """
    while True:
        new_edges = {(x, w) for x, y in edges for q, w in edges if q == y}

        all_edges = edges | new_edges

        if all_edges == edges:
            return edges

        edges = all_edges

We will use transitive_closure when computing epsilon closure. The full epsilon-closure operation requires a few more steps, which we will work through below.

transitive_closure({(0, 1), (1, 2), (2, 3), (3, 4)})

Now consider the base FiniteStateAutomaton class. It uses a TransitionFunction class to keep the bookkeeping for transition functions in one place.

from copy import copy, deepcopy
from functools import lru_cache
from collections.abc import Sequence
from typing import Literal, Never

type TransitionGraph = dict[tuple[str, str], str | set[str]]
type FSAParse = tuple[tuple[str, str], ...]
type FSAMode = Literal["recognize", "parse"]


class FiniteStateAutomaton:
    """A finite state automaton.

    Parameters
    ----------
    alphabet : set[str]
        The alphabet of the automaton.
    states : set[str]
        The set of states.
    initial_state : str
        The initial state.
    final_states : set[str]
        The set of accepting states.
    transition_graph : TransitionGraph
        The transition graph mapping (state, symbol) pairs to output states.
    """

    def __init__(
        self,
        alphabet: set[str],
        states: set[str],
        initial_state: str,
        final_states: set[str],
        transition_graph: TransitionGraph,
    ) -> None:
        self._alphabet = alphabet | {""}
        self._states = states
        self._initial_state = initial_state
        self._final_states = final_states
        self._transition_function = TransitionFunction(
            self._alphabet, states, transition_graph
        )

        self._validate_initial_state()
        self._validate_final_states()

        self._generator = self._build_generator()

    def __contains__(self, string: str | Sequence[str]) -> bool:
        normalized = string if isinstance(string, str) else tuple(string)
        return self._recognize(normalized)

    def __add__(self, other: FiniteStateAutomaton) -> FiniteStateAutomaton:
        return self.concatenate(other)

    def __or__(self, other: FiniteStateAutomaton) -> FiniteStateAutomaton:
        return self.union(other)

    def __and__(self, other: FiniteStateAutomaton) -> FiniteStateAutomaton:
        return self.intersect(other)

    def __pow__(self, k: int) -> FiniteStateAutomaton:
        return self.exponentiate(k)

    def __neg__(self) -> FiniteStateAutomaton:
        return self.complement()

    def __iter__(self) -> FiniteStateAutomaton:
        return self

    def __next__(self) -> str:
        return next(self._generator)

    def _build_generator(self) -> Iterator[str]:
        string_buffer = [(self._initial_state, "")]

        if self._initial_state in self._final_states:
            stack = [""]
        else:
            stack = []

        while string_buffer:
            if stack:
                yield stack.pop()

            else:
                # very inefficient with total transition functions
                # that have many transitions to the sink state
                new_buffer = []
                for symb in self._alphabet:
                    for old_state, string in string_buffer:
                        new_states = self._transition_function(old_state, symb)
                        for st in new_states:
                            new_elem = (st, string + symb)
                            new_buffer.append(new_elem)

                stack += [
                    string
                    for state, string in new_buffer
                    if state in self._final_states
                ]

                string_buffer = new_buffer

    def __call__(
        self, string: str | Sequence[str], mode: FSAMode = "recognize"
    ) -> bool | set[FSAParse]:
        """Determine whether/how a string is accepted/parsed by the FSA.

        Parameters
        ----------
        string : str | Sequence[str]
            The string to recognize or parse.
        mode : FSAMode
            Whether to run in "recognize" or "parse" mode.

        Returns
        -------
        bool | set[FSAParse]
            Boolean for recognize mode, set of parses for parse mode.
        """

        normalized = string if isinstance(string, str) else tuple(string)
        if mode == "recognize":
            return self._recognize(normalized)
        elif mode == "parse":
            return self._parse(normalized)
        else:
            msg = 'mode must be "recognize" or "parse"'
            raise ValueError(msg)

    def _add_epsilon_extensions(
        self,
        paths: set[tuple[str, ...]],
    ) -> set[tuple[str, ...]]:
        paths_extended = {
            s1 + (s2,) for s1 in paths for s2 in self._transition_function(s1[-1], "")
        }

        while new_paths := paths_extended - paths:
            paths |= new_paths
            paths_extended = {
                path + (next_state,)
                for path in new_paths
                for next_state in self._transition_function(path[-1], "")
                if next_state not in path
            }

        return paths

    @lru_cache(512)
    def _recognize(
        self, string: str | tuple[str, ...], prev_state: str | None = None
    ) -> bool:
        """Whether a string is accepted by the FSA.

        Parameters
        ----------
        string : str | Sequence[str]
            The string to recognize.
        prev_state : str | None
            The state to start from, or None for the initial state.

        Returns
        -------
        bool
            Whether the string is accepted.
        """
        paths = {(self._initial_state,)} if prev_state is None else {(prev_state,)}
        paths = self._add_epsilon_extensions(paths)

        if string:
            return any(
                self._recognize(string[1:], state)
                for p in paths
                for state in self._transition_function(p[-1], string[0])
            )

        else:
            return any(s[-1] in self._final_states for s in paths)

    @lru_cache(512)
    def _parse(
        self, string: str | tuple[str, ...], prev_state: str | None = None
    ) -> set[FSAParse]:
        """How a string is parsed by the FSA.

        Returns the list of transitions that the machine could
        go through to parse a string.

        Parameters
        ----------
        string : str | Sequence[str]
            The string to parse.
        prev_state : str | None
            The state to start from, or None for the initial state.

        Returns
        -------
        set[FSAParse]
            Set of parse tuples representing transition sequences.
        """
        paths = {(self._initial_state,)} if prev_state is None else {(prev_state,)}
        paths = self._add_epsilon_extensions(paths)

        if string:
            return {
                tuple((s, "") for s in p[1:]) + ((state, string[0]),) + parse
                for p in paths
                for state in self._transition_function(p[-1], string[0])
                for parse in self._parse(string[1:], state)
            }

        else:
            return {
                tuple((s, "") for s in p[1:])
                for p in paths
                if p[-1] in self._final_states
            }

    def _validate_initial_state(self) -> None:
        if self._initial_state not in self._states:
            msg = "initial state must be in set of states"
            raise ValueError(msg)

    def _validate_final_states(self) -> None:
        for state in self._final_states:
            if state not in self._states:
                msg = "final states must be in set of states"
                raise ValueError(msg)

    def _deepcopy(self) -> FiniteStateAutomaton:
        """Create a deep copy of this FSA.

        Returns
        -------
        FiniteStateAutomaton
            A deep copy of this automaton.
        """
        return FiniteStateAutomaton(
            copy(self._alphabet),
            copy(self._states),
            self._initial_state,
            copy(self._final_states),
            deepcopy(self._transition_function._transition_graph),
        )

    def _relabel_states(self, tag: str) -> FiniteStateAutomaton:
        """Append tag to the input/output states throughout the FSA.

        Parameters
        ----------
        tag : str
            The tag to append to state names.
        """

        state_map = {s: s + "_" + tag for s in self._states}

        self._initial_state = state_map[self._initial_state]
        self._final_states = {state_map[s] for s in self._final_states}
        self._states = {state_map[s] for s in self._states}
        self._transition_function.relabel_states(state_map)

        return self

    def concatenate(self, other: FiniteStateAutomaton) -> FiniteStateAutomaton:
        """Concatenate this FSA with another.

        Parameters
        ----------
        other : FiniteStateAutomaton
            The automaton to concatenate with.

        Returns
        -------
        FiniteStateAutomaton
            The concatenated automaton.
        """
        msg = "you still need to implement FiniteStateAutomaton.concatenate"
        raise NotImplementedError(msg)

    def exponentiate(self, k: int) -> FiniteStateAutomaton:
        """Concatenate this FSA k times.

        Parameters
        ----------
        k : int
            The number of times to repeat; must be >1.

        Returns
        -------
        FiniteStateAutomaton
            The exponentiated automaton.
        """
        if k <= 1:
            raise ValueError("must be >1")

        new = self

        for i in range(1, k):
            new += self

        return new

    def union(self, other: FiniteStateAutomaton) -> FiniteStateAutomaton:
        """Union this FSA with another.

        Parameters
        ----------
        other : FiniteStateAutomaton
            The automaton to union with.

        Returns
        -------
        FiniteStateAutomaton
            The unioned automaton.
        """
        msg = "you still need to implement FiniteStateAutomaton.union"
        raise NotImplementedError(msg)

    def complement(self) -> FiniteStateAutomaton:
        """Complement this FSA.

        Returns
        -------
        FiniteStateAutomaton
            The complemented automaton.
        """
        fsa = self._deepcopy()
        fsa = fsa.determinize()
        fsa._transition_function.totalize()
        fsa._final_states = fsa._states - fsa._final_states

        return fsa

    def intersect(self, other: FiniteStateAutomaton) -> FiniteStateAutomaton:
        """Intersect this FSA with another.

        Parameters
        ----------
        other : FiniteStateAutomaton
            The automaton to intersect with.

        Returns
        -------
        FiniteStateAutomaton
            The intersected automaton.
        """
        fsa1 = self.complement()
        fsa2 = other.complement()

        return fsa1.union(fsa2).complement()

    def determinize(self) -> FiniteStateAutomaton:
        """Determinize the FSA using the standard subset construction.

        Returns
        -------
        FiniteStateAutomaton
            The determinized automaton.
        """
        new_init, new_trans = self._transition_function.determinize(self._initial_state)

        # the alphabet without epsilon
        alphabet = self._alphabet - {""}

        # build the DFA by exploring reachable states from the
        # initial state; each DFA state is a frozenset of NFA states
        new_init_frozen = frozenset(new_init)
        worklist = [new_init_frozen]
        visited = set()
        new_transition = {}

        while worklist:
            current = worklist.pop()
            if current in visited:
                continue
            visited.add(current)

            for a in alphabet:
                # compute the set of NFA states reachable from
                # current on symbol a (epsilon closure is already
                # baked into new_trans by _add_epsilon_transitive_closure)
                target = set()
                for q in current:
                    target |= new_trans(q, a)

                if target:
                    target_frozen = frozenset(target)
                    src_label = "_".join(sorted(current))
                    tgt_label = "_".join(sorted(target_frozen))
                    new_transition[(src_label, a)] = {tgt_label}

                    if target_frozen not in visited:
                        worklist.append(target_frozen)

        new_states = {"_".join(sorted(s)) for s in visited}
        new_final = {
            "_".join(sorted(s))
            for s in visited
            if any(t in s for t in self._final_states)
        }

        return FiniteStateAutomaton(
            alphabet,
            new_states,
            "_".join(sorted(new_init_frozen)),
            new_final,
            new_transition,
        )

    @property
    def isdeterministic(self) -> bool:
        """Whether this automaton is deterministic."""
        return self._transition_function.isdeterministic


class TransitionFunction:
    """A finite state machine transition function.

    Parameters
    ----------
    alphabet : set[str]
        The alphabet of the automaton.
    states : set[str]
        The set of states.
    transition_graph : TransitionGraph
        The transition graph.

    Attributes
    ----------
    isdeterministic : bool
        Whether the transition function is deterministic.
    istotalfunction : bool
        Whether the transition function is total.
    transition_graph : dict
        The underlying transition graph.
    """

    def __init__(
        self, alphabet: set[str], states: set[str], transition_graph: TransitionGraph
    ) -> None:
        self._alphabet = alphabet
        self._states = states
        self._transition_graph = transition_graph

        self._validate()

    def __call__(self, state: str, symbol: str) -> set[str]:
        """Apply the transition function.

        Parameters
        ----------
        state : str
            The current state.
        symbol : str
            The input symbol.

        Returns
        -------
        set[str]
            The set of reachable states.
        """
        try:
            return self._transition_graph[(state, symbol)]
        except KeyError:
            return set({})

    def __or__(self, other: TransitionFunction) -> TransitionFunction:
        alphabet = self._alphabet | other._alphabet
        states = self._states | other._states
        graph = {key: set(value) for key, value in self._transition_graph.items()}
        for key, value in other._transition_graph.items():
            graph.setdefault(key, set()).update(value)
        return TransitionFunction(alphabet, states, graph)

    def _add_epsilon_transitive_closure(self) -> TransitionGraph:
        # get the state graph of all epsilon transitions
        transitions = {
            (instate, outstate)
            for (instate, insymb), outs in self._transition_graph.items()
            for outstate in outs
            if not insymb
        }

        # compute the transitive closure of the epsilon transition
        # state graph; requires homogenization beforehand
        for instate, outstate in transitive_closure(transitions):
            self._transition_graph[(instate, "")] |= {outstate}

        new_graph = dict(self._transition_graph)

        for (instate1, insymb1), outs1 in self._transition_graph.items():
            for (instate2, insymb2), outs2 in self._transition_graph.items():
                # Case 1: epsilon-after-symbol
                # if (s1, a) -> s2 and (s2, epsilon) -> s3,
                # then add s3 to (s1, a)
                if instate2 in outs1 and not insymb2:
                    try:
                        new_graph[(instate1, insymb1)] |= outs2
                    except KeyError:
                        new_graph[(instate1, insymb1)] = set(outs2)

                # Case 2: symbol-after-epsilon
                # if (s1, epsilon) -> s2 and (s2, a) -> s3,
                # then add s3 to (s1, a)
                if instate2 in outs1 and not insymb1 and insymb2:
                    try:
                        new_graph[(instate1, insymb2)] |= outs2
                    except KeyError:
                        new_graph[(instate1, insymb2)] = set(outs2)

        return new_graph

    def determinize(self, initial_state: str) -> tuple[set[str], TransitionFunction]:
        """Determinize using epsilon transitive closure.

        Parameters
        ----------
        initial_state : str
            The initial state of the NFA.

        Returns
        -------
        tuple[set[str], TransitionFunction]
            The new initial state set and determinized transition function.
        """
        # add epsilon transitive closure
        epsilon_closed_graph = self._add_epsilon_transitive_closure()

        # the new initial state is the set containing the old initial state
        # along with any state reachable by epsilon transitions
        if (initial_state, "") in epsilon_closed_graph:
            new_initial_state = {initial_state} | epsilon_closed_graph[
                initial_state, ""
            ]
        else:
            new_initial_state = {initial_state}

        # filter epsilon transitions
        filtered_transition_graph = {
            (instate, insymb): outstates
            for (instate, insymb), outstates in epsilon_closed_graph.items()
            if insymb
        }

        new_trans = TransitionFunction(
            self._alphabet, self._states, filtered_transition_graph
        )

        return new_initial_state, new_trans

    def add_transitions(self, transition_graph: TransitionGraph) -> None:
        """Add transitions to the graph.

        Parameters
        ----------
        transition_graph : TransitionGraph
            New transitions to add.
        """
        self._transition_graph.update(transition_graph)

    def _validate(self) -> None:
        self._validate_input_values()
        self._validate_output_types()
        self._homogenize_output_types()
        self._validate_output_values()

    def _validate_input_values(self) -> None:
        for state, symbol in self._transition_graph.keys():
            if symbol not in self._alphabet:
                msg = (
                    "all input symbols in transition function " + "must be in alphabet"
                )
                raise ValueError(msg)

            if state not in self._states:
                msg = (
                    "all input states in transition function "
                    + "must be in set of states"
                )
                raise ValueError(msg)

    def _validate_output_types(self) -> None:
        for states in self._transition_graph.values():
            if type(states) is not str and type(states) is not set:
                msg = (
                    "all outputs in transition function "
                    + "must be specified via str or set"
                )
                raise ValueError(msg)

    def _homogenize_output_types(self) -> None:
        for inp, out in self._transition_graph.items():
            if type(out) is str:
                self._transition_graph[inp] = {out}

    def _validate_output_values(self) -> None:
        for out in self._transition_graph.values():
            if not all(state in self._states for state in out):
                msg = "all output symbols in transition function " + "must be in states"
                raise ValueError(msg)

    @property
    def isdeterministic(self) -> bool:
        """Whether the transition function is deterministic."""
        has_epsilon = any(symb == "" for _, symb in self._transition_graph.keys())
        all_singleton = all(len(v) < 2 for v in self._transition_graph.values())
        return all_singleton and not has_epsilon

    @property
    def istotalfunction(self) -> bool:
        """Whether the transition function is total."""
        return all(
            (state, symbol) in self._transition_graph
            for state in self._states
            for symbol in self._alphabet - {""}
        )

    def relabel_states(self, state_map: dict[str, str]) -> None:
        """Relabel states according to the given mapping.

        Parameters
        ----------
        state_map : dict[str, str]
            A mapping from old states to new states.
        """

        new_transition_graph = {}

        for (instate, insymb), outs in self._transition_graph.items():
            new_inp = (state_map[instate], insymb)
            new_outs = {state_map[o] for o in outs}
            new_transition_graph[new_inp] = new_outs

        self._transition_graph = new_transition_graph

    def totalize(self) -> None:
        """Make the transition function total by adding a sink state."""
        if not self.istotalfunction:
            symbols = self._alphabet - {""}
            sink_state = "qsink"

            while sink_state in self._states:
                sink_state += "sink"

            old_states = set(self._states)
            self._states.add(sink_state)
            for state in old_states:
                for symbol in symbols:
                    self._transition_graph.setdefault(
                        (state, symbol), {sink_state}
                    )
            for symbol in symbols:
                self._transition_graph[(sink_state, symbol)] = {sink_state}

    @property
    def transition_graph(self) -> TransitionGraph:
        """The underlying transition graph."""
        return self._transition_graph

We can now define an FSA in a way that closely matches the formal definition.

fsa1 = FiniteStateAutomaton(
    alphabet={"a", "b", ""},
    states={"q0", "q1", "q2"},
    initial_state="q0",
    final_states={"q0", "q1"},
    transition_graph={
        ("q0", "a"): {"q0", "q1"},
        ("q0", ""): {"q1"},
        ("q1", ""): {"q2"},
        ("q1", "a"): {"q0"},
        ("q1", "b"): {"q0"},
    },
)

The transition function normalizes every target to a set, so it need not be identical to the dictionary passed to the constructor. Epsilon closure is computed when the automaton is determinized or when recognition follows epsilon paths.

fsa1._transition_function._transition_graph

You will not need to determinize the FSA directly. Still, it is useful to see what the implemented operation returns:

fsa1_det = fsa1.determinize()

fsa1_det._initial_state
fsa1_det._transition_function._transition_graph

So the original FSA is nondeterministic, while the determinized version is deterministic.

fsa1.isdeterministic, fsa1_det.isdeterministic

You will not need to compute the complement directly. The intersection operation uses it internally, so it is implemented here as well.

fsa1_comp = fsa1.complement()

print(fsa1_det._states)
print(fsa1_comp._states)
print(fsa1_det._final_states)
print(fsa1_comp._final_states)

Notice that, because taking the complement requires a totalized transition function, we have exactly the deterministic machine from above, except with explicit sink states.

fsa1_comp._transition_function._transition_graph

We don’t need explicit determinization with the power set construction to have a deterministic machine. It’s also possible to define a DFA directly. This machine gets implicitly converted to the strongly equivalent NFA.

fsa2 = FiniteStateAutomaton(
    alphabet={"c", "d"},
    states={"q0"},
    initial_state="q0",
    final_states={"q0"},
    transition_graph={("q0", "c"): "q0", ("q0", "d"): "q0"},
)

fsa2.isdeterministic
fsa2._transition_function._transition_graph

How can we inspect strings in the language of an FSA? We can iterate through the FiniteStateAutomaton object. The methods FiniteStateAutomaton.__iter__ and FiniteStateAutomaton.__next__ implement this behavior using FiniteStateAutomaton._build_generator.

for i, string in enumerate(fsa1):
    if i < 50:
        print(string)
    else:
        break

Note that some strings are repeated. Different paths through an FSA can produce the same string, which is also why one string can have multiple parses. The iterator retains those repetitions so that the implementation used in Task 1 remains easy to follow.

Compare the determinized machine.

for i, string in enumerate(fsa1_det):
    if i < 50:
        print(string)
    else:
        break

The iterator uses a deliberately simple search. It may run for a long time, or forever, when an automaton has many sink states or computes the empty set. Avoiding that behavior would require extra bookkeeping that is not needed for the assignment.

# this will hang
for i, string in enumerate(fsa1_comp):
    if i < 50:
        print(string)
    else:
        break

Now check that the DFA recognizes the expected strings.

for i, string in enumerate(fsa2):
    if i < 50:
        print(string)
    else:
        break

Now we can use the recognition and parsing algorithms. The FiniteStateAutomaton.__call__ method provides the interface to both. Set mode="recognize" to use the recognizer; it returns a Boolean.

fsa1("ab", mode="recognize"), fsa1("ac", mode="recognize")

To use the parser, set mode="parse". If the string is recognized, a set of parses will be output.

fsa1("ab", mode="parse")

If the string is not recognized, an empty set will be returned.

fsa1("ac", mode="parse")

Study these two algorithms before beginning Task 4. You will implement the analogous operations for FSTs there, and much of the control logic can be reused.

Task 1

Implement FiniteStateAutomaton.union and FiniteStateAutomaton.concatenate.

Test your implementation.

Task 2

Define an FSA that can generate English noun phrases built from the following vocabulary.

\[\Sigma = \{\text{the}, \text{that}, \text{those}, \text{lazy}, \text{greyhound}, \text{greyhounds}, \text{human}, \text{humans}, \text{love}, \text{loves}\}\]

You may find it useful to define a few different machines and then union or concatenate them.

Your FSA should recognize both simple noun phrases—such as greyhounds, the greyhound, that human, those lazy greyhounds, and the humans—and noun phrases with relative clauses—such as the greyhound that the lazy human loves and the humans that love the lazy greyhound. Make sure, however, that your machine cannot generate a string that is not a noun phrase of English. For instance, it should not generate noun phrases with agreement errors in the relative clause—such as the humans that loves the greyhound.

To make a parse—i.e. the sequence of states used to recognize the string—interpretable, use state labels that track at least part-of-speech information. For instance, the labels might record that the is a determiner and that that is a determiner in some contexts but a complementizer in others. You may want to track additional information to handle agreement correctly.

An FSA can generate an infinite regular set of noun phrases, including some patterns with an unbounded number of relative clauses. It cannot generate every structurally possible noun phrase while enforcing dependencies that require unbounded memory. Strive to capture as many noun phrases as possible—including strings of the form the humans that love the lazy greyhound that loves the humans that… with an unbounded number of right-branching relative clauses—and then identify a dependency that your FSA cannot enforce.

Test this FSA by attempting to parse and recognize five possible noun phrases, which should be recognizable and have at least one parse, and five possible noun phrases, which should not be recognizable and should have no parses.

What kind of noun phrase cannot be generated by an FSA at all? Why?

Task 3

Define an FSA that can generate English sentences built from the vocabulary from Task 2. You are encouraged to use the noun phrase machine you developed in that task to generate the noun phrases within these sentences. As in that task, you may find it useful to define a few different machines (in addition to the noun phrase machine) and then union and/or concatenate them.

Your FSA should recognize both simple transitive sentences—such as the greyhound loves the humans—and sentences with clause embedding—such as the greyhound loves that the humans love the greyhound. As before, handle agreement correctly. Where possible, the machine should also recognize sentences with unbounded embedding—such as the humans love that the greyhound loves that the humans love that….

Test this FSA by attempting to parse and recognize five possible sentences, which should be recognizable and have at least one parse, and five possible sentences, which should not be recognizable and should have no parses.

Finite State Transducers

type FSTMode = Literal["recognize", "parse", "transduce"]
type FSTParse = tuple[tuple[str, str, str], ...]

type TransductionGraph = dict[tuple[str, str, str], str]


class FiniteStateTransducer(FiniteStateAutomaton):
    """A finite state transducer.

    Parameters
    ----------
    alphabet1 : set[str]
        The input alphabet.
    alphabet2 : set[str]
        The output alphabet.
    states : set[str]
        The set of states.
    initial_state : str
        The initial state.
    final_states : set[str]
        The set of accepting states.
    transition_graph : TransitionGraph
        The transition graph.
    transduction_graph : TransductionGraph
        The transduction graph mapping (state1, state2, symbol) to output symbols.
    """

    def __init__(
        self,
        alphabet1: set[str],
        alphabet2: set[str],
        states: set[str],
        initial_state: str,
        final_states: set[str],
        transition_graph: TransitionGraph,
        transduction_graph: TransductionGraph,
    ) -> None:
        self._transduction_function = TransductionFunction(transduction_graph)
        self._alphabet2 = alphabet2
        self._transduction_function.validate(alphabet1, alphabet2, states)

        super().__init__(
            alphabet1, states, initial_state, final_states, transition_graph
        )

    def __next__(self) -> Never:
        msg = "we do not need this for the current assignment"
        raise NotImplementedError(msg)

    def __call__(
        self,
        string1: str | Sequence[str],
        string2: str | Sequence[str] | None = None,
        mode: FSTMode = "recognize",
    ) -> bool | set[FSTParse] | set[str]:
        """Determine whether/how/what a string is accepted/parsed/transduced to by the FST.

        Parameters
        ----------
        string1 : str | Sequence[str]
            The input string.
        string2 : str | Sequence[str] | None
            The second string; only needed for "recognize" or "parse" mode.
        mode : FSTMode
            Whether to run in "recognize", "parse", or "transduce" mode.

        Returns
        -------
        bool | set[FSTParse] | set[str]
            Depends on mode: bool for recognize, set of parses for parse,
            set of strings for transduce.
        """

        normalized1 = string1 if isinstance(string1, str) else tuple(string1)
        normalized2 = (
            string2
            if isinstance(string2, str) or string2 is None
            else tuple(string2)
        )
        if mode in ("recognize", "parse") and normalized2 is None:
            raise ValueError(f"string2 is required in {mode} mode")
        if mode == "recognize":
            return self._recognize(normalized1, normalized2)
        elif mode == "parse":
            return self._parse(normalized1, normalized2)
        elif mode == "transduce":
            return self._transduce(normalized1)
        else:
            msg = 'mode must be "recognize", "parse", or "transduce"'
            raise ValueError(msg)

    @lru_cache(65536)
    def _recognize(
        self,
        string1: str | Sequence[str],
        string2: str | Sequence[str],
        state: str | None = None,
    ) -> bool:
        """Whether a pair of strings is accepted by the FST.

        Parameters
        ----------
        string1 : str | Sequence[str]
            The first input string.
        string2 : str | Sequence[str]
            The second input string.
        state : str | None
            The current state, or None for the initial state.

        Returns
        -------
        bool
            Whether the pair is accepted.
        """
        msg = "you still need to implement FiniteStateTransducer._recognize"
        raise NotImplementedError(msg)

    @lru_cache(65536)
    def _parse(
        self,
        string1: str | Sequence[str],
        string2: str | Sequence[str],
        prev_state: str | None = None,
    ) -> set[FSTParse]:
        """How a pair of strings is parsed by the FST.

        Parameters
        ----------
        string1 : str | Sequence[str]
            The first input string.
        string2 : str | Sequence[str]
            The second input string.
        prev_state : str | None
            The current state, or None for the initial state.

        Returns
        -------
        set[FSTParse]
            Set of parse tuples representing transition sequences.
        """
        msg = "you still need to implement FiniteStateTransducer._parse"
        raise NotImplementedError(msg)

    @lru_cache(65536)
    def _transduce(
        self, string: str | Sequence[str], state: str | None = None, new_string: str = ""
    ) -> set[str]:
        """Compute the strings transduced from the input by the FST.

        Parameters
        ----------
        string : str | Sequence[str]
            The input string to transduce.
        state : str | None
            The current state, or None for the initial state.
        new_string : str
            Accumulator for the output string being built.

        Returns
        -------
        set[str]
            The set of transduced output strings.
        """
        msg = "you still need to implement FiniteStateTransducer._transduce"
        raise NotImplementedError(msg)

    def _relabel_states(self, tag: str) -> FiniteStateTransducer:
        """Append tag to the input/output states throughout the FST.

        Parameters
        ----------
        tag : str
            The tag to append to state names.
        """

        state_map = {state: state + "_" + tag for state in self._states}
        self._transduction_function.relabel_states(state_map)
        super()._relabel_states(tag)

        return self

    def concatenate(self, other: FiniteStateAutomaton) -> Never:
        msg = "we do not need this for the current assignment"
        raise NotImplementedError(msg)

    def exponentiate(self, k: int) -> Never:
        msg = "we do not need this for the current assignment"
        raise NotImplementedError(msg)

    def intersect(self, other: FiniteStateAutomaton) -> Never:
        msg = "we do not need this for the current assignment"
        raise NotImplementedError(msg)

    def complement(self) -> Never:
        msg = "we do not need this for the current assignment"
        raise NotImplementedError(msg)

    def union(self, other: FiniteStateAutomaton) -> Never:
        msg = "we do not need this for the current assignment"
        raise NotImplementedError(msg)

    def determinize(self) -> Never:
        msg = "we do not need this for the current assignment"
        raise NotImplementedError(msg)


class TransductionFunction:
    """A finite state transduction function.

    Parameters
    ----------
    transduction_graph : TransductionGraph
        The transduction graph mapping (state1, state2, symbol) to output symbols.
    """

    def __init__(self, transduction_graph: TransductionGraph) -> None:
        self._transduction_graph = transduction_graph

    def __call__(self, state1: str, state2: str, symbol: str) -> str | None:
        """Apply the transduction function.

        Parameters
        ----------
        state1 : str
            The source state.
        state2 : str
            The target state.
        symbol : str
            The input symbol.

        Returns
        -------
        str | set
            The output symbol(s).
        """
        try:
            return self._transduction_graph[(state1, state2, symbol)]
        except KeyError:
            return None

    def add_transductions(self, transduction_graph: TransductionGraph) -> None:
        """Add transductions to the graph.

        Parameters
        ----------
        transduction_graph : TransductionGraph
            New transductions to add.
        """
        self._transduction_graph.update(transduction_graph)

    def validate(
        self, alphabet1: set[str], alphabet2: set[str], states: set[str]
    ) -> None:
        """Validate the transduction function against the given alphabets and states.

        Parameters
        ----------
        alphabet1 : set[str]
            The input alphabet.
        alphabet2 : set[str]
            The output alphabet.
        states : set[str]
            The set of states.
        """
        self._validate_input_values(alphabet1, states)
        self._validate_output_values(alphabet2)

    def _validate_input_values(self, alphabet: set[str], states: set[str]) -> None:
        for state1, state2, symbol in self._transduction_graph.keys():
            if symbol not in alphabet:
                msg = (
                    "all input symbols in transduction function "
                    + "must be in alphabet1"
                )
                raise ValueError(msg)

            if state1 not in states or state2 not in states:
                msg = (
                    "all input states in transduction function "
                    + "must be in set of states"
                )
                raise ValueError(msg)

            if symbol == "":
                msg = "epsilon transduction not currently supported"
                raise ValueError(msg)

        # Note: total function check is deferred to avoid
        #        O(|states|^2 * |alphabet|) on construction
        self._states_ref = states
        self._alphabet_ref = alphabet

    def _validate_output_values(self, alphabet: set[str]) -> None:
        for symb in self._transduction_graph.values():
            if symb not in alphabet:
                msg = (
                    "all output symbols in transduction function "
                    + "must be in alphabet2"
                )
                raise ValueError(msg)

    def relabel_states(self, state_map: dict[str, str]) -> None:
        """Relabel states in the transduction function.

        Parameters
        ----------
        state_map : dict[str, str]
            A mapping from old states to new states.
        """

        new_transduction_graph = {}

        for (instate, outstate, insymb), outs in self._transduction_graph.items():
            new_inp = (state_map[instate], state_map[outstate], insymb)
            new_transduction_graph[new_inp] = outs

        self._transduction_graph = new_transduction_graph

    @property
    def transduction_graph(self) -> TransductionGraph:
        """The underlying transduction graph."""
        return self._transduction_graph
fst = FiniteStateTransducer(
    alphabet1={"a", "b"},
    alphabet2={"c", "d"},
    states={"q0", "q1"},
    initial_state="q0",
    final_states={"q0", "q1"},
    transition_graph={
        ("q0", "a"): {"q0", "q1"},
        ("q0", "b"): {"q0"},
        ("q1", "a"): {"q0"},
        ("q1", "b"): {"q0"},
    },
    transduction_graph={
        ("q0", "q0", "a"): "d",
        ("q0", "q0", "b"): "d",
        ("q0", "q1", "a"): "c",
        ("q0", "q1", "b"): "d",
        ("q1", "q0", "a"): "c",
        ("q1", "q0", "b"): "d",
    },
)

Task 4

Implement FiniteStateTransducer._recognize, FiniteStateTransducer._parse, and FiniteStateTransducer._transduce. If you wish, you may alter the typing of the transduction_graph so that it outputs sets of characters in the output alphabet rather the single character.

Task 5

For this task, we’ll be working with the MegaAcceptability dataset, which you can read about in this paper.

# pandas is installed by the course requirements
import pandas as pd

mega_acceptability = pd.read_csv(
    "http://megaattitude.io/projects/mega-acceptability/mega-acceptability-v1/mega-acceptability-v1-normalized.tsv",
    sep="\t",
)

mega_acceptability

We will use three columns from the dataset. First, consider the frames.

frames = set(mega_acceptability.frame.unique())

frames

Now consider the sentences associated with those frames.

sentences = {
    frame: {
        s.lower()
        for s in mega_acceptability.query(f'frame=="{frame}"').sentence.unique()
    }
    for frame in frames
}

sentences["NP Ved NP"]

We will also need the verb forms to construct the transductions.

verbs = {
    "root": set(mega_acceptability.verb.unique()),
    "past": set(mega_acceptability.query('frame=="NP Ved"').verbform.unique()),
    "past_participle": set(
        mega_acceptability.query('frame=="NP was Ved"').verbform.unique()
    ),
}

verbs["past"]

The provided FST interface emits exactly one output symbol for each non-epsilon input transition. Use a length-expanded frame encoding in which a frame token is repeated once for each output word it realizes. Restrict attention to rows whose verb form contains no whitespace. Construct an FST that maps each supported expanded frame encoding to the corresponding MegaAcceptability sentences. For instance, the ordinary three-token encoding NP Ved NP maps to strings such as someone liked something, while a clausal token such as S may need to be repeated to align with something happened.

Test whether your FST recognizes every supported one-token-verb sentence under this expanded encoding. Report how many dataset rows the restriction excludes.

Now choose one frame and test whether the transducer outputs all of its corresponding sentences.

MegaAcceptability contains many items that are unacceptable, denoted by having a more negative responsenorm.

mega_acceptability.sort_values("responsenorm")

Suppose we wanted the FST to output only the instances of a frame whose responsenorm exceeds a particular threshold—i.e. only the acceptable instances. You do not need to implement this restriction, but describe how you might do so.