diff --git a/__pycache__/bn_sim.cpython-314.pyc b/__pycache__/bn_sim.cpython-314.pyc deleted file mode 100644 index df020ed..0000000 Binary files a/__pycache__/bn_sim.cpython-314.pyc and /dev/null differ diff --git a/bn_sim.py b/bn_sim.py deleted file mode 100644 index c771fdd..0000000 --- a/bn_sim.py +++ /dev/null @@ -1,104 +0,0 @@ -from itertools import product -from random import random - -funcs = [ - lambda x1, x2, x3, x4: not x2, - lambda x1, x2, x3, x4: x1, - lambda x1, x2, x3, x4: x1 ^ x4, - lambda x1, x2, x3, x4: x3, -] - -sequential_sync_funcs = [ - lambda x1, x2, x3, x4: not x2, - lambda x1, x2, x3, x4: not x2, - lambda x1, x2, x3, x4: (not x2) ^ x4, - lambda x1, x2, x3, x4: (not x2) ^ x4, -] - -probabilistic_funcs = [ - (1, funcs[0], None), - (1, funcs[1], None), - (1, funcs[2], None), - (0.5, funcs[3], lambda x1, x2, x3, x4: not x3), -] - - -repressilator_funcs = [ - lambda x1, x2, x3: not x3, - lambda x1, x2, x3: not x1, - lambda x1, x2, x3: not x2, -] - -funcs = repressilator_funcs - -nodes = [] - - -def init(initial_state): - global nodes - nodes = [bool(i) for i in initial_state] - assert len(funcs) == len(nodes) - - -def synchronous_update(): - global nodes, funcs - temp = nodes.copy() - for i in range(len(funcs)): - temp[i] = funcs[i](*nodes) - nodes = temp - - -def sequential_update(order): - global nodes, funcs - assert len(order) == len(nodes) - for i in order: - nodes[i] = funcs[i](*nodes) - - -def probabilistic_update(): - global nodes, probabilistic_funcs - funcs = probabilistic_funcs - temp = nodes.copy() - for i in range(len(funcs)): - probability, *nodeFuncs = funcs[i] - func = nodeFuncs[0] if probability >= random() else nodeFuncs[1] - temp[i] = func(*nodes) - nodes = temp - - -def state_to_str(): - return "".join(str(int(i)) for i in nodes) - - -def simulate(initial_state): - print("press ENTER to simulate one step. press 'q' before ENTER to exit") - init(initial_state) - state = state_to_str() - while input() != "q": - update() - print(state + " -> " + state_to_str(), end="") - - -def get_table(): - for i in product((0, 1), repeat=3): - init(tuple(i)) - state = state_to_str() - update() - print(state + " -> " + state_to_str()) - - -def update(): - # return synchronous_update() - return sequential_update([0, 1, 2]) - # return probabilistic_update() - # return block_sequential_update(...) - # return asynchronous_deterministic_update(...) - - -def main(): - get_table() - # simulate([0,0,0,0]) - - -if __name__ == "__main__": - main() diff --git a/code/__pycache__/simulator.cpython-314.pyc b/code/__pycache__/simulator.cpython-314.pyc new file mode 100644 index 0000000..ccf907e Binary files /dev/null and b/code/__pycache__/simulator.cpython-314.pyc differ diff --git a/code/main.py b/code/main.py new file mode 100644 index 0000000..590d22f --- /dev/null +++ b/code/main.py @@ -0,0 +1,400 @@ +from functools import partial +from types import FunctionType +from typing import Callable, Optional + +from simple_term_menu import TerminalMenu +from simulator import BooleanNetwork + + +def ShowMenu(options: list[str], title: str = "", highlight_entry: int = 0) -> int: + menu = TerminalMenu( + menu_entries=options, + title=title, + multi_select=False, + cursor_index=highlight_entry, + ) + + selectedIndex = menu.show() + + if selectedIndex is None: + selectedIndex = len(options) - 1 + + return selectedIndex # pyright: ignore[reportReturnType] + + +def Continue() -> None: + ShowMenu(options=[None, "Continue"]) + + +# def MainMenu() -> None: +# global currentFunction +# options = ["Simulator", "Calculator", "Quit"] +# actions: list[Callable[[], None]] = [SimulatorMenu, StopApplication] + +# selectedIndex = ShowMenu( +# options, title="Boolean Network Simulator\nby Tom Zuidberg" +# ) + +# currentFunction = actions[selectedIndex] + + +def StopApplication() -> None: + quit() + + +def MainMenu() -> None: + global currentFunction + + def SelectExampleMenu() -> None: + def bn4nodeSync() -> None: + bn = ( + BooleanNetwork(4) + .UseSynchronousScheme() + .SetFunctions( + [ + lambda x1, x2, x3, x4: not x2, + lambda x1, x2, x3, x4: x1, + lambda x1, x2, x3, x4: x1 ^ x4, + lambda x1, x2, x3, x4: x3, + ] + ) + ) + global currentFunction + funcStrings = ["not x2", "x1", "x1 ^ x4", "x3"] + currentFunction = partial(BooleanNetworkMenu, bn, funcStrings) + + def bn4nodeSeq() -> None: + bn = ( + BooleanNetwork(4) + .UseSequentialScheme([1, 2, 3, 4]) + .SetFunctions( + [ + lambda x1, x2, x3, x4: not x2, + lambda x1, x2, x3, x4: x1, + lambda x1, x2, x3, x4: x1 ^ x4, + lambda x1, x2, x3, x4: x3, + ] + ) + ) + global currentFunction + funcStrings = ["not x2", "x1", "x1 ^ x4", "x3"] + currentFunction = partial(BooleanNetworkMenu, bn, funcStrings) + + def bn3nodeSync() -> None: + bn = ( + BooleanNetwork(3) + .UseSynchronousScheme() + .SetFunctions( + [ + lambda x1, x2, x3: not x3, + lambda x1, x2, x3: not x1, + lambda x1, x2, x3: not x2, + ] + ) + ) + global currentFunction + funcStrings = ["not x3", "not x1", "not x2"] + currentFunction = partial(BooleanNetworkMenu, bn, funcStrings) + + def bn3nodeSeq() -> None: + bn = ( + BooleanNetwork(3) + .UseSequentialScheme([1, 2, 3]) + .SetFunctions( + [ + lambda x1, x2, x3: not x3, + lambda x1, x2, x3: not x1, + lambda x1, x2, x3: not x2, + ] + ) + ) + global currentFunction + funcStrings = ["not x3", "not x1", "not x2"] + currentFunction = partial(BooleanNetworkMenu, bn, funcStrings) + + def ReturnHelper() -> None: + global currentFunction + currentFunction = MainMenu + + title = "Select an example boolean network:" + options = [ + "4-node network with synchronous update", + "4-node network with sequential update", + "3-node repressilator network with synchronous update", + "3-node repressilator network with sequential update", + "Return to main menu", + ] + actions = [bn4nodeSync, bn4nodeSeq, bn3nodeSync, bn3nodeSeq, ReturnHelper] + + actions[ShowMenu(options=options, title=title)]() + + options = [ + "Set up a new boolean network", + "Set up a boolean network from existing ones", + "Quit", + ] + + actions: list[Callable[[], None]] = [ + BooleanNetworkMenu, + SelectExampleMenu, + StopApplication, + ] + + selectedIndex = ShowMenu( + options, title="Boolean Network Simulator\nby Tom Zuidberg" + ) + + currentFunction = actions[selectedIndex] + + +def BooleanNetworkMenu( + bn: Optional[BooleanNetwork] = None, funcStrings: Optional[list[str]] = None +) -> None: + global currentFunction + + boolNetwork: BooleanNetwork = BooleanNetwork(3) + verbose: bool = False + done: bool = False + functions: list = [None for _ in range(3)] + functionStrings: list = [None for _ in range(3)] + functionsDirtyFlag = False + current_highlight = 0 + + if bn is not None and funcStrings is not None: + boolNetwork = bn + verbose = False + functions = [None for _ in range(3)] + functionStrings = funcStrings + + def Menu() -> None: + nonlocal current_highlight + title: str = f"Boolean Network: {boolNetwork} (timeStep | state)" + options = [ + f"Set size (current: {boolNetwork.size}) WARNING: this will reset all other options!", + f"Set state (current: {str(boolNetwork)[-boolNetwork.size :]})", + f"Set update scheme (current: {boolNetwork.updateScheme})", + *[ + f"Set update function of node x{i} (current: {functionStrings[i - 1]})" + for i in range(1, boolNetwork.size + 1) + ], + None, + f"Toggle verbose update: (current: {verbose})", + "Update once", + "Update multiple times", + "Get stationairy distribution", + None, + "Return to main menu", + ] + + actions = [ + SetSizeHelper, + SetStateHelper, + SetUpdateSchemeHelper, + *[partial(SetFunctionHelper, i) for i in range(boolNetwork.size)], + None, + ToggleVerboseHelper, + UpdateHelper, + MultiUpdateHelper, + GetStableDistrHelper, + None, + ReturnHelper, + ] + + selectedIndex = ShowMenu( + options=options, title=title, highlight_entry=current_highlight + ) + current_highlight = selectedIndex + actions[selectedIndex]() + + def SetSizeHelper() -> None: + nonlocal boolNetwork, functions, functionStrings + while True: + size = input("Set new size (Leave empty to cancel):\n").strip() + if size == "": + print("Cancelled") + return + try: + size = int(size) + except ValueError: + print("Please enter only integers.") + continue + + boolNetwork = BooleanNetwork(size) + functions = [None for _ in range(size)] + functionStrings = [None for _ in range(size)] + return + + def SetStateHelper() -> None: + nonlocal boolNetwork + while True: + state = input( + "Set new state (Leave empty to cancel).\nAccepted format example: '01001'\n" + ).strip() + if state == "": + print("Cancelled") + return + try: + boolNetwork.SetState(state) + return + except AssertionError as e: + print("Invalid input:", e) + + def SetUpdateSchemeHelper() -> None: + nonlocal boolNetwork + + def SetSequentialHelper() -> None: + while True: + seq = input( + "Set sequence. (Leave empty to cancel)\nFormat example: '1,4,3,2'\n" + ).strip() + if seq == "": + print("Cancelled") + return + + seq = seq.split(",") + try: + seq = [int(i) for i in seq] + boolNetwork.UseSequentialScheme(seq) + return + except Exception as e: + print("Invalid input:", e) + + def SetProbabilisticHelper() -> None: + while True: + chance = input( + "Set flip chance as float between 0.0 and 1.0. (Leave empty to cancel)\n" + ).strip() + if chance == "": + print("Cancelled") + return + + try: + chance = float(chance) + boolNetwork.UseProbabilisticScheme(chance) + return + + except Exception as e: + print("Invalid input:", e) + + title = "Select update scheme:" + options = [ + "Synchronous update scheme", + "Sequential update scheme", + "Probabilistic update scheme", + "Asynchronous random update scheme", + "Cancel", + ] + actions = [ + lambda: boolNetwork.UseSynchronousScheme(), + SetSequentialHelper, + SetProbabilisticHelper, + lambda: boolNetwork.UseAsynchronousRandomScheme(), + lambda: None, + ] + + actions[ShowMenu(options=options, title=title)]() + + def UpdateHelper() -> None: + nonlocal functions, boolNetwork, functionsDirtyFlag + if functionsDirtyFlag: + try: + boolNetwork.SetFunctions(functions) + functionsDirtyFlag = False + except Exception as e: + print("Error while updating the functions:", e) + return + + try: + boolNetwork.Update() + except Exception as e: + print("Error while simulating once:", e) + + def MultiUpdateHelper() -> None: + nonlocal boolNetwork, verbose, functions, functionsDirtyFlag + if functionsDirtyFlag: + try: + boolNetwork.SetFunctions(functions) + functionsDirtyFlag = False + except Exception as e: + print("Error while updating functions:", e) + + n = 0 + while True: + n = input( + "Enter amount of time steps to be simulated. (Leave empty to cancel)\n" + ).strip() + if n == "": + print("Cancelled") + return + + try: + n = int(n) + if n < 0: + raise ValueError + break + except ValueError: + print("Invalid input") + try: + boolNetwork.Update(n, verbose=verbose) + if verbose: + Continue() + print("\n") + except Exception as e: + print("Error while simulating:", e) + + def GetStableDistrHelper() -> None: + print(boolNetwork.GetStableProbabilityDistribution()) + Continue() + print("\n") + + def SetFunctionHelper(index: int) -> None: + nonlocal boolNetwork, functions, functionStrings, functionsDirtyFlag + while True: + funcString = input( + f"Set new function for node x{index + 1}. The function will receive all nodes in form of x1, x2, ..., x[size]. (Leave empty to cancel)\n" + ) + if funcString == "": + print("Cancelled") + return + try: + func = ( + "lambda " + + ",".join(f"x{i}" for i in range(1, boolNetwork.size + 1)) + + ":" + + funcString + ) + func = eval(func) + if not isinstance(func, FunctionType): + print("Please enter a valid function. Got: " + func) + continue + functions[index] = func + functionStrings[index] = funcString + functionsDirtyFlag = True + return + except Exception as e: + print("Error while parsing function:", e) + + def ToggleVerboseHelper(): + nonlocal verbose + verbose = not verbose + + def ReturnHelper(): + nonlocal done + done = True + + while not done: + Menu() + + currentFunction = MainMenu + + +def main() -> None: + global currentFunction + currentFunction = MainMenu + + while True: + currentFunction() + + +if __name__ == "__main__": + main() diff --git a/code/requirements.txt b/code/requirements.txt new file mode 100644 index 0000000..52158a0 --- /dev/null +++ b/code/requirements.txt @@ -0,0 +1,3 @@ +scipy +numpy +simple-term-menu diff --git a/code/simulator.py b/code/simulator.py new file mode 100644 index 0000000..a882027 --- /dev/null +++ b/code/simulator.py @@ -0,0 +1,299 @@ +import inspect +import random +from itertools import product +from typing import Callable, Concatenate, Iterable, Self + +import numpy as np +import scipy.linalg + + +class BooleanNetwork: + def __init__(self, size: int) -> None: + assert type(size) is int and size > 0, ( + f"Init error: Boolean Network must contain atleast one node. got {size=} nodes" + ) + self.size = size + self.__ready = False + self.__has_update_functions = False + self.__has_update_scheme = False + self.__has_sequence = False + self.__has_flip_chance = False + self.flip_chance: float = 0 + self.sequence: list[int] = list() + self.seed: int | None = None + self.time_step = 0 + + self.updateScheme: None | str = None + self.nodes: list[bool] = [False for _ in range(size)] + self.functions: list[Callable[Concatenate[bool, ...], bool]] = [ + lambda x: x for _ in range(size) + ] + + def SetFunctions( + self, functions: Iterable[Callable[Concatenate[bool, ...], bool]] + ) -> Self: + def wrapper( + function: Callable[Concatenate[bool, ...], bool], + ) -> Callable[Concatenate[bool, ...], bool]: + def wrap(*args, **kwargs) -> bool: + result = function(*args, **kwargs) + assert type(result) is bool, ( + f"Function error: Boolean network functions must always return a bool, however got type {type(result)}, {result=}" + ) + return result + + return wrap + + funcs: list[Callable[Concatenate[bool, ...], bool]] = list(functions) + + assert len(funcs) == self.size, ( + f"Function error: Function amount mismatch. got {len(funcs)} functions, expected {self.size}" + ) + + for i in range(self.size): + func = funcs[i] + assert len(inspect.signature(func).parameters) == self.size, ( + f"Function error: Function arg amount mismatch. Given function takes {len(inspect.signature(func).parameters)} arguments, expected {self.size}" + ) + + self.functions[i] = wrapper(func) + + self.__has_update_functions = True + return self + + def SetFunction( + self, index: int, function: Callable[Concatenate[bool, ...], bool] + ) -> Self: + def wrapper( + function: Callable[Concatenate[bool, ...], bool], + ) -> Callable[Concatenate[bool, ...], bool]: + def wrap(*args, **kwargs) -> bool: + result = function(*args, **kwargs) + assert type(result) is bool, ( + f"Function error: Boolean network functions must always return a bool, however got type {type(result)}, {result=}" + ) + return result + + return wrap + + assert 0 <= index < self.size, ( + f"Function error: cannot set function at index {index} - out of bound." + ) + self.functions[index] = wrapper(function) + return self + + def UseSynchronousScheme(self) -> Self: + self.updateScheme = "synchronous" + self.__has_update_scheme = True + + return self + + def UseSequentialScheme(self, sequence: Iterable[int]) -> Self: + sequence = list(sequence) + assert len(sequence) == self.size, ( + f"Sequence error: sequence must be the same size as the nodes of the network: sequence '{sequence}', #nodes={self.size}" + ) + assert all(type(i) is int for i in sequence), ( + f"Sequence error: sequence must only contain integers. sequence given: {sequence}" + ) + sorted_sequence = sorted(sequence) + compare_to = list(range(self.size + 1)) + assert ( + sorted_sequence == compare_to[:-1] or sorted_sequence == compare_to[1:] + ), ( + f"Sequence error: sequence doesn't contain the correct indices. It must contain all numbers from 0 to {self.size} (excluded) or from 1 to {self.size} (included)" + ) + + if sorted_sequence[0] == 1: + for i in range(self.size): + sequence[i] -= 1 + + self.sequence = sequence + self.updateScheme = "sequential" + self.__has_update_scheme = True + self.__has_sequence = True + + return self + + def UseAsynchronousRandomScheme(self, seed: int | None = None) -> Self: + if seed is not None: + assert type(seed) is int, ( + f"AsyncRandom error: wrong format for given seed. got {seed=}" + ) + self.seed = seed + random.seed(seed) + + self.updateScheme = "asynchronous_random" + self.__has_update_scheme = True + return self + + def UseProbabilisticScheme(self, flip_chance: float) -> Self: + if flip_chance is not None: + assert type(flip_chance) is float, ( + f"Probabilistic error: given flip_chance is not a float: got {flip_chance}" + ) + self.flip_chance = flip_chance + + self.updateScheme = "probabilistic" + self.__has_update_scheme = True + self.__has_flip_chance = True + return self + + def __synchronous_update(self) -> None: + temp = list() + for i in range(self.size): + temp.append(self.functions[i](*self.nodes)) + self.nodes = temp + + def __sequential_update(self) -> None: + for i in self.sequence: + self.nodes[i] = self.functions[i](*self.nodes) + + def __asynchronous_random_update(self) -> None: + index = random.randrange(0, self.size) + self.nodes[index] = self.functions[index](*self.nodes) + + def __probabilistic_update(self) -> None: + self.__synchronous_update() + for i in range(self.size): + rng = random.random() + if rng <= self.flip_chance: + self.nodes[i] = not self.nodes[i] + + def SetState(self, state: str | list[bool] | tuple[bool, ...]) -> Self: + assert isinstance(state, (str, list, tuple)), ( + f"SetState error: invalid type as state" + ) + assert len(state) == self.size, ( + f"SetState error: given state is not the same size as the boolean network. got size {len(state)}, expected {self.size}" + ) + + if type(state) is str: + for i in range(self.size): + self.nodes[i] = bool(int(state[i])) + return self + + if isinstance(state, (list, tuple)): + assert all(type(i) is bool for i in state), ( + f"SetState error: given state list contains elements of type different from bool. All elements must be bools. got {state}" + ) + self.nodes = list(state).copy() + return self + + raise Exception( + "SetState error: end of function reached. given state is not of type string nor list[bool]." + ) + + def Update(self, n: int = 1, /, verbose=False) -> None: + assert type(n) is int and n >= 0, ( + f"Update error: amount of updates must be an integer and positive. got {n=}" + ) + assert self.__has_update_functions, "Update error: no update functions defined" + assert self.__has_update_scheme, "Update error: no update scheme defined" + assert type(verbose) is bool, "Update error: verbose must be a bool" + + selected_update: Callable + + match self.updateScheme: + case "synchronous": + selected_update = self.__synchronous_update + case "sequential": + assert self.__has_sequence, "Update error: no sequence defined" + selected_update = self.__sequential_update + case "asynchronous_random": + selected_update = self.__asynchronous_random_update + case "probabilistic": + assert self.__has_flip_chance, "Update error: no flip_chance defined" + selected_update = self.__probabilistic_update + case _: + raise Exception("Update error: update scheme selection went wrong") + + for _ in range(n): + selected_update() + self.time_step += 1 + if verbose: + print(self) + + def __str__(self) -> str: + return f"{self.time_step:>5} | {''.join(str(int(node)) for node in self.nodes)}" + + @property + def state(self) -> str: + return "".join(str(int(node)) for node in self.nodes) + + def GetStableProbabilityDistribution(self) -> np.ndarray: + matrix: np.ndarray = self.__get_transition_matrix() + eigenvalues, eigenvectors = scipy.linalg.eig(matrix.T) + stationary_vector = np.real(eigenvectors[:, np.isclose(eigenvalues, 1)]) + stationary_distribution = stationary_vector / np.sum(stationary_vector) + return stationary_distribution.flatten() + + def __get_transition_matrix(self) -> np.ndarray: + dimension = 2**self.size + matrix: np.ndarray = np.zeros((dimension, dimension)) + + if self.updateScheme == "probabilistic": + flipChance = self.flip_chance + self.UseSynchronousScheme() + for i, state in enumerate(product((False, True), repeat=self.size)): + self.SetState(state) + self.Update() + for flips in product((False, True), repeat=self.size): + flipped = int( + "".join( + str( + int( + self.nodes[j] if not flips[j] else not self.nodes[j] + ) + ) + for j in range(self.size) + ), + 2, + ) + prob = np.float64(1) + for flip in flips: + prob *= flipChance if flip else 1 - flipChance + matrix[i][flipped] = prob + self.UseProbabilisticScheme(flipChance) + return matrix + + if self.updateScheme == "asynchronous_random": + for i, state in enumerate(product((False, True), repeat=self.size)): + for j in range(self.size): + self.SetState(state) + self.nodes[j] = self.functions[j](*self.nodes) + matrix[i][int(self.state, 2)] += np.float64(1) / self.size + return matrix + + for i, state in enumerate(product((False, True), repeat=self.size)): + self.SetState(state) + self.Update() + matrix[i][int(self.state, 2)] = 1 + return matrix + + +def main() -> None: + bn = ( + BooleanNetwork(4) + .SetState("0000") + .SetFunctions( + [ + lambda a, b, c, d: not b, + lambda a, b, c, d: a, + lambda a, b, c, d: a ^ d, + lambda a, b, c, d: c, + ] + ) + .UseSequentialScheme((1, 2, 3, 4)) + ) + + print(bn) + bn.Update() + print(bn) + + print("update 100 times verbose") + bn.Update(100, verbose=True) + + +if __name__ == "__main__": + main()