muMLE/util/simulator.py

import abc
import random
import math
import functools
import sys
from typing import Callable, Generator

from framework.conformance import Conformance, render_conformance_check_result
from concrete_syntax.common import indent
from concrete_syntax.textual_od.renderer import render_od
from transformation.cloner import clone_od
from api.od import ODAPI


class DecisionMaker:
    @abc.abstractmethod
    def __call__(self, actions):
        pass

class RandomDecisionMaker(DecisionMaker):
    def __init__(self, seed=0, verbose=True):
        self.seed = seed
        self.r = random.Random(seed)

    def __str__(self):
        return f"RandomDecisionMaker(seed={self.seed})"

    def __call__(self, actions):
        arr = [action for descr, action in actions]
        if len(arr) == 0:
            return
        i = math.floor(self.r.random()*len(arr))
        return arr[i]

class InteractiveDecisionMaker(DecisionMaker):
    # auto_proceed: whether to prompt if there is only one enabled action
    def __init__(self, msg="Select action:", auto_proceed=False):
        self.msg = msg
        self.auto_proceed = auto_proceed

    def __str__(self):
        return f"InteractiveDecisionMaker()"

    def __call__(self, actions):
        arr = []
        for i, (key, result) in enumerate(actions):
           print(f"  {chr(97+i)}. {key}")
           arr.append(result)
        if len(arr) == 0:
           return
        if len(arr) == 1 and self.auto_proceed:
            return arr[0]

        def __choose():
           sys.stdout.write(f"{self.msg} ")
           try:
              raw = input()
              choice = ord(raw)-97 # may raise ValueError
              if choice >= 0 and choice < len(arr):
                 return arr[choice]
           except (ValueError, TypeError):
              pass
           print("Invalid option")
           return __choose()

        return __choose()



class MinimalSimulator:
    def __init__(self,
        action_generator: Callable[[any], Generator[any, None, None]],
        decision_maker: DecisionMaker = RandomDecisionMaker(seed=0),

        # Returns 'None' to keep running, or a string to end simulation
        # Can also have side effects, such as rendering the model, and performing a conformance check.
        # BTW, Simulation will always end when there are no more enabled actions.
        termination_condition=lambda model: None,

        verbose=True,
    ):
        self.action_generator = action_generator
        self.decision_maker = decision_maker
        self.termination_condition = termination_condition
        self.verbose = verbose

    def _print(self, *args):
        if self.verbose:
            print(*args)

    # Run simulation until termination condition satisfied
    def run(self, model):
        self._print("Start simulation")
        self._print(f"Decision maker: {self.decision_maker}")
        step_counter = 0
        while step_counter < 10:
            termination_reason = self.termination_condition(model)
            if termination_reason != None:
                self._print(f"Termination condition satisfied.\nReason: {termination_reason}.")
                break

            chosen_action = self.decision_maker(self.action_generator(model))

            if chosen_action == None:
                self._print(f"No enabled actions.")
                break

            (model, msgs) = chosen_action()
            self._print(indent('\n'.join(f"▸ {msg}" for msg in msgs), 4))

            step_counter += 1

        self._print(f"Executed {step_counter} steps.")
        return model