modelverse/services/DEVS/pypdevs/examples/client.py

print("SCCD PythonPDEVS CLI interface")
import sys
import time
from pprint import pprint

sys.path.append("../src/")
from sccd.runtime.statecharts_core import Event

import services.DEVS.pypdevs.src.simulator as simulator
from services.DEVS.pypdevs.examples.ps_model import Root
controller = simulator.Controller(Root())
reply_port = controller.addOutputListener('reply')

def set_defaults(inp, defaultlist):
    for i, v in enumerate(defaultlist):
        if len(inp) == i + 1:
            inp.append(v)

def get_bool(val):
    if val.lower() in ["true", "yes", "1"]:
        return True
    else:
        return False

def poll_port(port):
    while 1:
        try:
            event = port.fetch(-1)
            print("%s" % event.getName())
            pprint(event.getParameters())
        except:
            pass

def generate_input():
    print("READY for input")
    print("Type 'help' for information")
    while 1:
        if sys.version_info[0] < 3:
            inp = raw_input().split(" ")
        else:
            inp = input().split(" ")

        action = inp[0]
        if inp[0] == "simulate":
            set_defaults(inp, ['inf'])
            params = [{"termination_time": float(inp[1])}]
        elif inp[0] == "big_step":
            set_defaults(inp, ['inf'])
            params = [{"termination_time": float(inp[1])}]
        elif inp[0] == "realtime":
            set_defaults(inp, ['inf', 1.0])
            params = [{"termination_time": float(inp[1]), "realtime_scale": float(inp[2])}]
        elif inp[0] == "small_step":
            set_defaults(inp, ['inf'])
            params = [{"termination_time": float(inp[1])}]
        elif inp[0] == "god_event":
            if len(inp) != 4:
                print("God Events require 3 parameters!")
                continue
            params = [{"model": inp[1], "attribute": inp[2], "value": inp[3]}]
        elif inp[0] == "inject":
            if len(inp) != 4:
                print("Injects require 3 parameters!")
                continue
            params = [{"time": float(inp[1]), "port": inp[2], "event": inp[3]}]
        elif inp[0] == "trace":
            set_defaults(inp, [None])
            params = [inp[1]]
        elif inp[0] == "pause":
            params = []
        elif inp[0] == "exit" or inp[0] == "quit":
            break
        elif inp[0] == "add_breakpoint":
            if len(inp) < 5:
                print("Breakpoint removal requires 2 parameters!")
                continue
            # Merge together the final part again
            inp = [inp[0], inp[1], " ".join(inp[2:-2]).replace("\\n", "\n"), get_bool(inp[-2]), get_bool(inp[-1])]
            print("Generated parameters: " + str(inp))
            params = inp[1:]
        elif inp[0] == "del_breakpoint":
            if len(inp) < 2:
                print("Breakpoint removal requires 1 parameter!")
                continue
            params = [inp[1]]
        elif inp[0] == "enable_breakpoint":
            action = "toggle_breakpoint"
            params = [inp[1], True]
        elif inp[0] == "disable_breakpoint":
            action = "toggle_breakpoint"
            params = [inp[1], False]
        elif inp[0] == "reset":
            params = []
        elif inp[0] == "backwards_step":
            params = []
        elif inp[0] == "help":
            print("Supported operations:")
            print("  simulate [termination_time]")
            print("   --> Simulate until termination time is reached")
            print("  big_step [termination_time]")
            print("   --> Simulate a single step, unless termination time is reached")
            print("  small_step [termination_time]")
            print("   --> Simulate a single internal simulation step")
            print("       Termination time is ONLY checked at the")
            print("       check_termination phase")
            print("  backwards_step")
            print("   --> Step back to the previous time")
            print("  realtime [termination_time [realtime_scale]]")
            print("   --> Simulate in realtime until simulation time is reached")
            print("       realtime_scale can speed up or slow down the pace")
            print("  god_event model_name attribute_name new_value")
            print("   --> Modify the internal state of an arbitrary model")
            print("       model_name should be the fully qualified name")
            print("       attribute_name is the name of the attribute to alter")
            print("       new_value is the value to assign")
            print("       new_value can only be a string due to string-only input")
            print("  inject time port_name event")
            print("   --> Put a user-defined event on an input port")
            print("       port_name should be a fully qualified port name")
            print("       event should be the event to put on it, string only")
            print("  trace [filename]")
            print("   --> Write out trace information to the specified file.")
            print("       Leave empty to disable tracing.")
            print("  add_breakpoint id function enabled disable_on_trigger")
            print("   --> Add a breakpoint that will pause simulation when function returns True")
            print("       the function should contain a definition of the 'breakpoint' function")
            print("       and must take 3 parameters: time, model and transitioned")
            print("       enabled indicates whether or not the breakpoint should be checked")
            print("       disable_on_trigger determines if the breakpoint should auto-disable")
            print("       after being triggered")
            print("  del_breakpoint id")
            print("   --> Remove a breakpoint")
            print("  enable_breakpoint id")
            print("   --> Enable the provided breakpoint")
            print("  disable_breakpoint id")
            print("   --> Disable the provided breakpoint")
            print("  reset")
            print("   --> Reset the simulation")
            print("  exit")
            print("   --> Stop the client")
            print("  pause")
            print("   --> Pause the simulation")
            print("  quit")
            print("   --> Stop the client")
            print("")
            print("Defaults: ")
            print("  termination_time --> 'inf'")
            print("  realtime_scale   --> 1.0")
            continue
        else:
            print("Command not understood: " + str(inp))
            continue
        controller.addInput(Event(action, "request", params))
      
import threading
poll_thrd = threading.Thread(target=poll_port,args=[reply_port])
poll_thrd.daemon = True
poll_thrd.start()
inp_thrd = threading.Thread(target=generate_input)
inp_thrd.daemon = True
inp_thrd.start()
          
try:
    controller.start()
finally:
    controller.stop()