PyPDEVS-and-JADEVS-DEVStone-benchmarks/devstone/testPyPDEVS.py

#!/usr/bin/env python3
import os
import subprocess
import csv
import re
from argparse import ArgumentParser, Action
from pathlib import Path
from itertools import product
from datetime import datetime

from pandas.errors import ParserError

MODEL_TYPES = ("LI", "HI", "HO", "HOmod")
DS_MODEL_TYPES = ("LI2HI", "HI2LI", "LI-HI", "dLI", "dHI")
GEN_METHODS = ("start", "uniform", "end")

INT_CYCLES = 0
EXT_CYCLES = 0

PyPDEVS = "./pythonpdevs/main.py"
MinimalPyPDEVS = "./pythonpdevs-minimal/main.py"

# Regex to capture times from script output
TIME_PATTERN = re.compile(
    r"Model creation time: ([\d.eE+-]+).*"
    r"Engine setup time: ([\d.eE+-]+).*"
    r"Simulation time: ([\d.eE+-]+)",
    re.S
)

def run_devstone(model_type, depth, width, mode, out, memory, *, classic=False, number=1, gen_mode="uniform"):
    """Runs the devstone_driver.py script and returns timing results as floats."""
    SCRIPT_PATH = PyPDEVS
    if mode == "minimal":
        SCRIPT_PATH = MinimalPyPDEVS

    cmd = [
        "uv", "run", SCRIPT_PATH,
        "-m", model_type,
        "-d", str(depth),
        "-w", str(width),
        "-i", str(INT_CYCLES),
        "-e", str(EXT_CYCLES),
    ]
    if mode == "dynamic":
        cmd.extend(["-D"])
        if classic:
            cmd.extend(["-C"])
        cmd.extend(["-n", str(number)])
        cmd.extend(["-g", gen_mode])
    if mode == "classic":
        cmd.extend(["-C"])
    full_cmd = " ".join(cmd)

    if memory:
        Path("/".join(memory["file"].split("/")[0:-1])).mkdir(parents=True, exist_ok=True)
        full_cmd = (
            f"{full_cmd} & "
            f"PID=$!; "
            f"python -m procpath record -i {memory['interval']} -f stat -d {memory['file']} "
            f"--stop-without-result \"$..children[?(@.stat.pid == $PID)]\""
        )

    result = subprocess.run(full_cmd, shell=True, capture_output=True, text=True)
    cout = result.stdout.strip()
    if out:
        print(cout)

    cerr = result.stderr.strip()
    match = TIME_PATTERN.search(cout)

    if match:
        creation, setup, sim = map(float, match.groups())
        return creation, setup, sim

    print("Failed to run: ", cmd)
    print(cout)
    print(cerr)
    return None, None, None

def defaultArguments(parser):
    parser.add_argument("-p", "--path", type=str, default="./Results-PyPDEVS", help="Path to the results directory")
    parser.add_argument("-f", '--filename', type=str, default="", help="Name of the output file")
    parser.add_argument("-a", "--append", action="store_true", help="Append to the output file instead of overwriting.")

    parser.add_argument("-md", "--max-depth", default=8, type=int, help="Max depth for the generated models")
    parser.add_argument("-ds", "--depth-stepsize", type=int, help="Determines for which depths a model is generated.")

    parser.add_argument("-mw", "--max-width", default=8, type=int, help="Max width for the generated models")
    parser.add_argument("-ws", "--width-stepsize", type=int, help="Determines for which depths a model is generated.")

    parser.add_argument("-r", "--repetitions", default=1, type=int, help="The number of times a simulation is repeated")
    parser.add_argument("-O", "--output", action="store_true", help="Print Simulation Outputs")
    parser.add_argument("-M", "--memory", action="store_true", help="Measures the memory usage of a simulation.")
    parser.add_argument("-i", "--interval", default=1.0, type=float, help="The interval at which measurements are taken.")

def Minimal(subparsers):
    minimal = subparsers.add_parser("minimal", help="Minimal DEVS kernel")
    minimal.set_defaults(mode="minimal")
    defaultArguments(minimal)
    minimal.add_argument(
        "-ms", "--model-structure",
        nargs="+",
        choices=MODEL_TYPES,
        default=MODEL_TYPES
    )

def Classic(subparsers):
    classic = subparsers.add_parser("classic", help="Classic DEVS Simulator")
    classic.set_defaults(mode="classic")
    defaultArguments(classic)
    classic.add_argument(
        "-ms", "--model-structure",
        nargs="+",
        choices=MODEL_TYPES,
        default=MODEL_TYPES
    )

def Parallel(subparsers):
    parallel = subparsers.add_parser("parallel", help="Parallel DEVS")
    parallel.set_defaults(mode="parallel")
    defaultArguments(parallel)
    parallel.add_argument(
        "-ms", "--model-structure",
        nargs="+",
        choices=MODEL_TYPES,
        default=MODEL_TYPES
    )

def Dynamic(subparsers):
    dynamic = subparsers.add_parser("dynamic", help="Dynamic Structure DEVS")
    dynamic.set_defaults(mode="dynamic")
    defaultArguments(dynamic)
    dynamic.add_argument(
        "-ms", "--model-structure",
        nargs="+",
        choices=MODEL_TYPES + DS_MODEL_TYPES,
        default=MODEL_TYPES + DS_MODEL_TYPES
    )
    dynamic.add_argument(
        "-C", "--classic",
        action="store_true",
        default=False,
        help="Selects the Classic DEVS Simulator"
    )
    dynamic.add_argument(
        "-n", "--number",
        default=1,
        type=int,
        help="Number of models to be generated in a cycle per coupled model")
    dynamic.add_argument(
        "-g", "--gen-mode",
        choices=GEN_METHODS,
        default="uniform"
    )

def parse_args():
    parser = ArgumentParser()
    subparsers = parser.add_subparsers(dest="mode")
    Minimal(subparsers)
    Parallel(subparsers)
    Classic(subparsers)
    Dynamic(subparsers)

    args = parser.parse_args()
    if args.depth_stepsize is None:
        args.depth_stepsize = args.max_depth
    if args.width_stepsize is None:
        args.width_stepsize = args.max_width

    return args

def run(writer, csvfile, model_type, d, w, args, mem_cmd, extra):
    """Orchestrates the repeated simulation runs."""
    for depth in range(d[1], d[0] + 1, d[1]):
        for width in range(w[1], w[0] + 1, w[1]):
            for run in range(args.repetitions):
                creation, setup, sim = run_devstone(
                    model_type, depth, width,
                    args.mode, args.output,
                    mem_cmd,
                    **extra
                )

                # Only write to CSV if a writer was provided
                if writer and creation is not None:
                    row = {
                        "model_type": model_type, "depth": depth, "width": width, "run": run,
                        "creation_time": creation, "setup_time": setup, "simulation_time": sim,
                    }
                    if args.mode == "dynamic":
                        row["number"] = args.number
                        row["generator_mode"] = args.gen_mode
                    writer.writerow(row)
                    csvfile.flush()


if __name__ == "__main__":
    args = parse_args()
    print(args)

    d = (args.max_depth, args.depth_stepsize)
    w = (args.max_width, args.width_stepsize)
    if d[0] % d[1] != 0:
        raise RuntimeError("Invalid depth stepsize.")
    if w[0] % w[1] != 0:
        raise RuntimeError("Invalid width stepsize.")

    fieldnames = ["model_type", "depth", "width", "run",
                  "creation_time", "setup_time", "simulation_time"]
    mode = args.mode
    if mode == "dynamic":
        if args.classic:
            mode += "-classic"
        fieldnames += ["number", "generator_mode"]

    extra = {
        "classic": False if not hasattr(args, "classic") else args.classic,
        "gen_mode": "uniform" if not hasattr(args, "gen_mode") else args.gen_mode,
        "number": 1 if not hasattr(args, "number") else args.number,
    }

    for model_type in args.model_structure:
        model_time_file = f"{args.path}/Time/{mode}/{model_type}"
        model_memory_file = f"{args.path}/Memory/{mode}/{model_type}"

        if args.filename:
            suffix = f"_{args.filename}"
            model_time_file += suffix
            model_memory_file += suffix

        model_time_file += ".csv"
        model_memory_file += ".sqlite"

        mem_cmd = {"interval": args.interval, "file": model_memory_file} if args.memory else {}

        if not args.memory:
            file_mode = "a" if args.append else "w"
            write_header = not args.append or not os.path.exists(model_time_file)

            with open(model_time_file, file_mode, newline="") as csvfile:
                writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
                if write_header:
                    writer.writeheader()

                run(writer, csvfile, model_type, d, w, args, mem_cmd, extra)
            print(f"✅ Saved timing results to {model_time_file}")
        else:
            print("📊 Measuring memory (Timing results will not be saved)...")
            run(None, None, model_type, d, w, args, mem_cmd, extra)
            print(f"✅ Saved memory database to {model_memory_file}")