Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix(dfa|nfa): #47 fix bugs presents #48

Merged
merged 1 commit into from
Sep 26, 2024
Merged

fix(dfa|nfa): #47 fix bugs presents #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
249 changes: 77 additions & 172 deletions automathon/finite_automata/dfa.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,9 @@
from automathon.errors.errors import (
SigmaError,
)
from automathon.utils.utils import (
list_map,
)
from collections import (
deque,
)
Expand Down Expand Up @@ -117,12 +120,12 @@ def accept(self, string: str) -> bool:

if idx == len(string) and state in self.f:
ans = True
elif idx < len(string) and state in self.delta:
elif idx < len(string):
# Search through states
for transition in self.delta[state].items():
for a, state in self.delta[state].items():
# transition: ('1', 'q0')
if string[idx] == transition[0]:
q.append([idx + 1, transition[1]])
if string[idx] == a:
q.append([idx + 1, state])

return ans

Expand Down Expand Up @@ -305,7 +308,9 @@ def symmetric_difference(self, m: "DFA") -> "DFA":
)

def __binary_operation(
self, m: "DFA", operation: Callable[..., bool]
self,
m: "DFA",
operation: Callable[[str, set[str], str, set[str]], bool],
) -> "DFA":
new_q_list: list[tuple[str, str]] = []

Expand All @@ -320,7 +325,7 @@ def __binary_operation(
self.sigma, "Sigma from both DFAs must be the same"
)

queue = deque()
queue: deque[tuple[str, str]] = deque()
queue.append((self.initial_state, m.initial_state))

while queue:
Expand Down Expand Up @@ -349,195 +354,95 @@ def __binary_operation(

def minimize(self) -> "DFA":
"""Minimize the automata and return the minimized version"""
visited: dict[tuple[str, str], bool] = dict()
table, q = self.__initialize_table_and_queue()
p_k: set[frozenset[str]] = set(
[frozenset([*self.q.difference(self.f)]), frozenset([*self.f])]
)
p_prev: set[frozenset[str]] = set()

while q:
q_a, q_b = q.popleft()
if visited.get((q_a, q_b), False):
break
while p_k != p_prev:
states_idx: dict[str, int] = self.__states_idx_table(p_k)
new_p_k: list[set[str]] = []

visited[(q_a, q_b)] = True
for p_states in p_k:
p_states_lst = list(p_states)
new_p_k.append({p_states_lst[0]})

common_sigma = filter(
lambda s, q_a=q_a, q_b=q_b: s in self.delta[q_a]
and s in self.delta[q_b],
self.sigma,
)
for i in range(1, len(p_states_lst)):
was_added = False
p_i_sigma = set(self.delta[p_states_lst[i]].keys())

reachable_pairs = map(
lambda s, q_a=q_a, q_b=q_b: (
self.delta[q_a][s],
self.delta[q_b][s],
),
common_sigma,
)
new_p_k, was_added = self.__define_group_ith_element(
i, new_p_k, p_i_sigma, p_states_lst, states_idx
)

it_is_marked = any(
map(lambda p: table.get(p, False), reachable_pairs)
)
if not was_added:
new_p_k.append({p_states_lst[i]})

if it_is_marked:
table[(q_a, q_b)] = True
visited = dict()
else:
q.append((q_a, q_b))

# combine pending_states
unmarked_pairs: list[tuple[str, str]] = []
unmarked_states: set[str] = set()

while q:
q_a, q_b = q.popleft()
unmarked_pairs.append((q_a, q_b))
unmarked_states.add(q_a)
unmarked_states.add(q_b)

remaining_states = self.q.copy() - unmarked_states
new_final_states: set[str] = set()
groups: dict[str, list[str]] = dict()

for pair in unmarked_pairs:
groups[pair[0]] = groups.get(pair[0], []) + [pair[1]]
groups[pair[1]] = groups.get(pair[1], []) + [pair[0]]

# group states from groups (dfs)
grouped_states: dict[int, list[str]] = dict()
states_group: dict[str, int] = dict()
final_groups: list[int] = []

groups_count = self.__group_unmarked_states(
unmarked_states,
groups,
grouped_states,
final_groups,
states_group,
)
p_prev, p_k = p_k, set(map(lambda s: frozenset([*s]), new_p_k))

for state in remaining_states:
states_group[state] = groups_count
grouped_states[groups_count] = [state]
states_idx: dict[str, int] = self.__states_idx_table(p_k)
initial_state = f"q{states_idx[self.initial_state]}"
final_states = set(list_map(lambda s: f"q{states_idx[s]}", self.f))

if state in self.f:
final_groups.append(groups_count)
delta: dict[str, dict[str, str]] = dict()
states = set(list_map(lambda idx: f"q{idx}", states_idx.values()))

groups_count += 1
for state_group in p_k:
fst_state = list(state_group)[0]

new_delta, new_initial_state = self.__build_new_delta(
groups_count, grouped_states, states_group
)
delta[f"q{states_idx[fst_state]}"] = dict()

for i in final_groups:
new_final_states.add(f"q{i}")
for s in self.delta[fst_state]:
delta[f"q{states_idx[fst_state]}"][s] = (
f"q{states_idx[self.delta[fst_state][s]]}"
)

return DFA(
set(new_delta.keys()),
self.sigma.copy(),
new_delta,
new_initial_state,
new_final_states,
states, self.sigma.copy(), delta, initial_state, final_states
)

def __initialize_table_and_queue(
self,
) -> tuple[dict[tuple[str, str], bool], deque[tuple[str, str]]]:
table: dict[tuple[str, str], bool] = dict()
states = list(self.q.copy())
q: deque[tuple[str, str]] = deque()

for i in range(1, len(states)):
for j in range(0, i):
if states[i] in self.f and states[j] not in self.f:
table[(states[i], states[j])] = True
else:
q.append((states[i], states[j]))
def __states_idx_table(self, p_k: set[frozenset[str]]) -> dict[str, int]:
states_idx: dict[str, int] = dict()

return table, q
for i in range(len(p_k)):
for state in list(p_k)[i]:
states_idx[state] = i

def __group_unmarked_states(
return states_idx

def __define_group_ith_element(
self,
unmarked_states: set[str],
groups: dict[str, list[str]],
grouped_states: dict[int, list[str]],
final_groups: list[int],
states_group: dict[str, int],
) -> int:
visited: dict[str, bool] = dict()
groups_count = 0

for state in unmarked_states:
if visited.get(state, False):
i: int,
new_p_k: list[set[str]],
p_i_sigma: set[str],
p_states_lst: list[str],
states_idx: dict[str, int],
) -> tuple[list[set[str]], bool]:
was_added = False
for new_p_states in new_p_k:
new_p_states_lst = list(new_p_states)
new_p_sigma = set(self.delta[new_p_states_lst[0]].keys())

if p_i_sigma != new_p_sigma:
continue

final_states, non_final_states = (
self.__get_final_and_non_final_states(state, visited, groups)
are_equivalent = all(
list_map(
lambda s,
p=p_states_lst[i],
q=new_p_states_lst[0],
table=states_idx,
delta=self.delta: table[delta[p][s]] == table[delta[q][s]],
p_i_sigma,
)
)

if final_states:
grouped_states[groups_count] = final_states
final_groups.append(groups_count)

for f_state in final_states:
states_group[f_state] = groups_count

groups_count += 1

if non_final_states:
grouped_states[groups_count] = non_final_states

for nf_state in non_final_states:
states_group[nf_state] = groups_count

groups_count += 1

return groups_count

def __get_final_and_non_final_states(
self, state: str, visited: dict[str, bool], groups: dict[str, list[str]]
) -> tuple[list[str], list[str]]:
stack: deque[str] = deque()
stack.append(state)
visited[state] = True
final_states = []
non_final_states = []

while stack:
current_state = stack.pop()

if current_state in self.f:
final_states.append(current_state)
else:
non_final_states.append(current_state)

for next_state in groups[current_state]:
if not visited.get(next_state, False):
stack.append(next_state)
visited[next_state] = True

return final_states, non_final_states

def __build_new_delta(
self,
groups_count: int,
grouped_states: dict[int, list[str]],
states_group: dict[str, int],
) -> tuple[dict[str, dict[str, str]], str]:
new_delta: dict[str, dict[str, str]] = dict()
new_initial_state: str = ""

for i in range(groups_count):
new_delta[f"q{i}"] = dict()

for state in grouped_states[i]:
if state == self.initial_state:
new_initial_state = f"q{i}"

for s in self.delta[state].keys():
new_delta[f"q{i}"][s] = (
f"q{states_group[self.delta[state][s]]}"
)
if are_equivalent:
new_p_states.add(p_states_lst[i])
was_added = True
break

return new_delta, new_initial_state
return new_p_k, was_added

def view(
self,
Expand Down
5 changes: 1 addition & 4 deletions automathon/finite_automata/nfa.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
annotations,
)
from automathon.errors.errors import (
InputError,
SigmaError,
)
from automathon.finite_automata.dfa import (
Expand Down Expand Up @@ -169,9 +168,7 @@ def _add_pairs_to_queue(

if idx == len(string) and state in self.f:
ans = True
elif idx < len(string) and string[idx] not in self.sigma:
raise InputError(string[idx], "Is not declared in sigma")
elif idx < len(string) and state in self.delta:
elif idx < len(string):
# Search through states
epsilon_transitions = filter(
lambda x: x[0] == "", self.delta[state].items()
Expand Down
48 changes: 44 additions & 4 deletions tests/test_dfa.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,15 +154,55 @@ def test_minimize(self):
minimized_dfa = dfa.minimize()

self.assertTrue(minimized_dfa.is_valid())
self.assertGreaterEqual(len(dfa.q), len(minimized_dfa.q))

def test_minimize_2(self):
fa = DFA(
q={"q0", "q1", "q2", "q3", "q4", "q5"},
sigma={"0", "1"},
delta={
"q0": {"0": "q3", "1": "q1"},
"q1": {"0": "q2", "1": "q5"},
"q2": {"0": "q2", "1": "q5"},
"q3": {"0": "q0", "1": "q4"},
"q4": {"0": "q2", "1": "q5"},
"q5": {"0": "q5", "1": "q5"},
},
initial_state="q0",
f={"q1", "q2", "q4"},
)
minimized_fa = fa.minimize()

self.assertGreaterEqual(len(fa.q), len(minimized_fa.q))
self.assertEqual(fa.accept("1"), minimized_fa.accept("1"))
self.assertEqual(fa.accept("11"), minimized_fa.accept("11"))
self.assertEqual(fa.accept("110"), minimized_fa.accept("110"))
self.assertEqual(fa.accept("1101"), minimized_fa.accept("1101"))
self.assertEqual(fa.accept("01"), minimized_fa.accept("01"))
self.assertEqual(fa.accept("0101"), minimized_fa.accept("0101"))

def test_minimize_existing_1(self):
minimized_fa = self.fa.minimize()
fa = DFA(
q={"q0", "q1", "q2"},
sigma={"0", "1"},
delta={
"q0": {"0": "q0", "1": "q1"},
"q1": {"0": "q2", "1": "q0"},
"q2": {"0": "q1", "1": "q2"},
},
initial_state="q0",
f={"q0"},
)
minimized_fa = fa.minimize()
not_minimized_fa = minimized_fa.complement()

self.assertGreaterEqual(len(fa.q), len(minimized_fa.q))
self.assertTrue(minimized_fa.is_valid())
self.assertTrue(minimized_fa.accept(""))
self.assertTrue(minimized_fa.accept("001001"))
self.assertTrue(minimized_fa.accept("0101010101010"))
self.assertEqual(fa.accept(""), minimized_fa.accept(""))
self.assertEqual(fa.accept("001001"), minimized_fa.accept("001001"))
self.assertEqual(
fa.accept("0101010101010"), minimized_fa.accept("0101010101010")
)

self.assertTrue(not_minimized_fa.is_valid())
self.assertFalse(not_minimized_fa.accept(""))
Expand Down