path: root/salis.py

#!/usr/bin/env -S PYTHONDONTWRITEBYTECODE=1 python

# file    : salis.py
# project : Salis-VM
# author  : Paul Oliver <contact@pauloliver.dev>

# index:
# [section] imports
# [section] argument parser
# [section] build class
# [section] logger
# [section] options dict formatter
# [section] entry-point; source options
# [section] load instruction set
# [section] assemble ancestor
# [section] defines
# [section] build and launch

# ------------------------------------------------------------------------------
# [section] imports
# ------------------------------------------------------------------------------
import ctypes
import json
import os
import random
import shutil
import signal
import subprocess
import sys

from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, ArgumentTypeError, RawTextHelpFormatter
from tempfile import TemporaryDirectory
from types import SimpleNamespace

# ------------------------------------------------------------------------------
# [section] argument parser
# ------------------------------------------------------------------------------
prog = sys.argv[0]
epilog = f"Use '-h' to show command arguments; e.g. '{prog} new -h'"

parser = ArgumentParser(description="Salis: Simple A-Life Simulator", epilog=epilog, formatter_class=RawTextHelpFormatter, prog=prog)
sub_parsers = parser.add_subparsers(dest="command", required=True)
formatter_class = lambda prog: ArgumentDefaultsHelpFormatter(max_help_position=48, prog=prog)

new = sub_parsers.add_parser("new", formatter_class=formatter_class, help="create new simulation")
load = sub_parsers.add_parser("load", formatter_class=formatter_class, help="load saved simulation")

def seed(i):
    ival = int(i, 0)
    if ival < -1: raise ArgumentTypeError("invalid seed value")
    return ival

def pos(i):
    ival = int(i, 0)
    if ival < 0: raise ArgumentTypeError("value must be positive integer")
    return ival

def nat(i):
    ival = int(i, 0)
    if ival < 1: raise ArgumentTypeError("value must be greater than zero")
    return ival

def port(i):
    ival = int(i, 0)
    if not 1024 <= ival <= 49151: raise ArgumentTypeError("value must be valid port number")
    return ival

fmt_id = lambda val: val
fmt_hex = lambda val: hex(int(val, 0)) if type(val) == str else hex(val)

options = (
    ("a", "anc", (new,), fmt_id, {"metavar": "ANC", "help": "ancestor file name without extension; to be compiled on all cores; ANC points to '{anc-path}/{arch}/{ANC}.asm'", "required": True, "type": str}),
    ("A", "anc-path", (new,), fmt_id, {"metavar": "PATH", "help": "path to search for ancestor files", "default": "anc", "required": False, "type": str}),
    ("b", "build-only", (new, load), fmt_id, {"action": "store_true", "help": "only build (do not run) executable", "required": False}),
    ("C", "clones", (new,), fmt_id, {"metavar": "N", "help": "number of ancestor clones on each core", "default": 1, "required": False, "type": nat}),
    ("c", "cores", (new,), fmt_id, {"metavar": "N", "help": "number of simulator cores", "default": 2, "required": False, "type": nat}),
    ("d", "data-push-pow", (new,), fmt_id, {"metavar": "POW", "help": "data aggregation interval exponent; interval = 2^{POW} >= {sync-pow}", "default": 28, "required": False, "type": nat}),
    ("f", "force", (new,), fmt_id, {"action": "store_true", "help": "overwrite existing simulation of given name", "required": False}),
    ("g", "c-compiler", (new, load), fmt_id, {"choices": ("clang", "gcc", "tcc"), "help": "C compiler to use", "default": "gcc", "required": False, "type": str}),
    ("H", "home", (new, load), fmt_id, {"metavar": "PATH", "help": "salis home directory", "default": f"{os.environ["HOME"]}/.salis", "required": False, "type": str}),
    ("l", "log-trace", (new, load), fmt_id, {"action": "store_true", "help": "log verbose trace messages", "required": False}),
    ("M", "muta-pow", (new,), fmt_id, {"metavar": "POW", "help": "mutator range exponent; each step a cosmic ray hits addr = rand_uint64() %% 2^{POW}; lower values of POW mean higher mutation rates", "default": 32, "required": False, "type": pos}),
    ("m", "mvec-pow", (new,), fmt_id, {"metavar": "POW", "help": "memory core size exponent; size = 2^{POW}", "default": 20, "required": False, "type": pos}),
    ("n", "name", (new, load), fmt_id, {"metavar": "NAME", "help": "name of new or loaded simulation", "default": "def.sim", "required": False, "type": str}),
    ("o", "optimized", (new, load), fmt_id, {"action": "store_true", "help": "build with optimizations", "required": False}),
    ("p", "pre-cmd", (new, load), fmt_id, {"metavar": "CMD", "help": "shell command with which to wrap call to executable; e.g. gdb, time, valgrind, etc.", "required": False, "type": str}),
    ("s", "seed", (new,), fmt_hex, {"metavar": "SEED", "help": "seed value for new simulation; a value of 0 disables cosmic rays; a value of -1 creates a random seed", "default": 0, "required": False, "type": seed}),
    ("T", "keep-temp-dir", (new, load), fmt_id, {"action": "store_true", "help": "keep temporary directory on exit", "required": False}),
    ("u", "ui", (new, load), fmt_id, {"metavar": "UI", "help": "user interface", "default": "curses", "required": False, "type": str}),
    ("U", "ui-path", (new, load), fmt_id, {"metavar": "PATH", "help": "path to search for UIs", "default": "ui", "required": False, "type": str}),
    ("v", "vm-arch", (new,), fmt_id, {"metavar": "ARCH", "help": "VM architecture", "default": "null", "required": False, "type": str}),
    ("y", "sync-pow", (new,), fmt_id, {"metavar": "POW", "help": "core sync interval exponent; sync events occur every N steps, where N = 2^{POW}", "default": 20, "required": False, "type": pos}),
    ("z", "auto-save-pow", (new,), fmt_id, {"metavar": "POW", "help": "auto-save interval exponent; interval = 2^{POW} >= {sync_pow}", "default": 36, "required": False, "type": pos}),
)

[sub_parser.add_argument(f"-{sopt}", f"--{lopt}", **kwargs) for sopt, lopt, sub_parsers, _, kwargs in options for sub_parser in sub_parsers]
args = parser.parse_args()

# ------------------------------------------------------------------------------
# [section] build class
# ------------------------------------------------------------------------------
class Build:
    def __init__(self, path, is_library=False, is_cpp=False, flags=None, defines=None, links=None):
        self.path = path
        self.is_library = is_library
        self.is_cpp = is_cpp

        self.flags = flags or []
        self.defines = defines or []
        self.links = links or []

        self.basename = os.path.splitext(os.path.basename(self.path))[0]
        self.extension = ".so" if self.is_library else ""
        self.tempdir = TemporaryDirectory(prefix="salis_", delete=not args.keep_temp_dir)
        self.binfile = f"{self.tempdir.name}/{self.basename}{self.extension}"

        self.flags += ["-Wall", "-Wextra", "-Werror", "-Wmissing-prototypes", "-Wmissing-declarations", "-Wno-overlength-strings", "-pedantic"]
        self.flags += ["-fPIC", "-shared"] if self.is_library else []
        self.flags += ["-O3", "-flto", "-march=native", "-mtune=native"] if args.optimized else ["-ggdb"]
        self.defines += ["-DNDEBUG"] if args.optimized else []

        self.compiler = args.cpp_compiler if self.is_cpp else args.c_compiler
        self.build_cmd = [self.compiler, *self.flags, *self.defines, *self.links, self.path, "-o", self.binfile]

    def build(self):
        subprocess.run(self.build_cmd, check=True)

        # Clear exec-stack on dynamic libraries when using TCC.
        # Otherwise python will complain when loading the library with CDLL.
        # NOTE: Using TCC thus requires `patchelf` to be accessible.
        if self.is_library and args.c_compiler == "tcc":
            subprocess.run(["patchelf", "--clear-execstack", self.binfile], check=True)

    def log(self, log):
        log.trace(f"Built source '{self.path}' to output: {self.binfile}")
        log.trace(f"Using build command: {self.build_cmd}")

    def run(self):
        run_cmd = args.pre_cmd.split() if args.pre_cmd else []
        run_cmd.append(self.binfile)
        proc = subprocess.Popen(run_cmd, stdout=sys.stdout, stderr=sys.stderr)

        # When a signal is received (e.g. SIGINT), is propagates through the entire process group.
        # Handle signal on the interpreter and allow the simulator to shut down gracefully.
        try:
            signal.pause()
        except KeyboardInterrupt:
            proc.wait()

        code = proc.returncode

        if code != 0: raise RuntimeError(f"Binary returned code: {code}")

# ------------------------------------------------------------------------------
# [section] logger
# ------------------------------------------------------------------------------
logger_defines = ["-DLOG_TRACE"] if args.log_trace else []
logger_shared = Build("core/logger.c", is_library=True, defines=logger_defines)
logger_shared.build()
logger_dll = ctypes.CDLL(logger_shared.binfile)

# Pull in logging functions from C
# This way there's just a single logging system to care about
log = SimpleNamespace()
log.trace = lambda msg: logger_dll.log_trace(msg.encode())
log.info = lambda msg: logger_dll.log_info(msg.encode())
log.attention = lambda msg: logger_dll.log_attention(msg.encode())
logger_shared.log(log)

# ------------------------------------------------------------------------------
# [section] options dict formatter
# ------------------------------------------------------------------------------
fmt_h2u = lambda field: field.replace("-", "_")
fmt_u2h = lambda field: field.replace("_", "-")

def fmt_field(field, val):
    for _, lopt, _, fmt, _ in options:
        if lopt == fmt_u2h(field):
            return fmt(val)

    return fmt_id(val)

def fmt_opts(opts):
    opts_out = {field: fmt_field(field, val) for field, val in opts.items()}
    return json.dumps(opts_out, indent=2)

# ------------------------------------------------------------------------------
# [section] entry-point; source options
# ------------------------------------------------------------------------------
log.info(f"Called '{prog} {args.command}' with the following options: {fmt_opts(vars(args))}")
paths = SimpleNamespace()

def gen_paths():
    paths.sim_dir = f"{args.home}/{args.name}"
    paths.sim_data = f"{paths.sim_dir}/{args.name}.sqlite3"
    paths.sim_evas = f"{paths.sim_dir}/evas"
    paths.sim_opts = f"{paths.sim_dir}/opts.json"
    paths.sim_path = f"{paths.sim_dir}/{args.name}"

def source_options():
    if not os.path.isdir(paths.sim_dir):
        raise RuntimeError(f"No simulation found named {args.name}")

    with open(paths.sim_opts, "r") as f:
        opts = json.loads(f.read())

    for key, val in opts.items():
        setattr(args, key, val)

    log.info(f"Sourced configuration from '{paths.sim_opts}': {fmt_opts(opts)}")

# Create new simulation
if args.command == "new":
    gen_paths()

    if 0 < args.data_push_pow < args.sync_pow:
        raise RuntimeError("Data push power must be equal or greater than core sync power")

    if 0 < args.auto_save_pow < args.sync_pow:
        raise RuntimeError("Autosave power must be equal or greater than core sync power")

    if os.path.isdir(paths.sim_dir) and args.force:
        log.attention(f"Force flag used! Wiping old simulation at: {paths.sim_dir}")
        shutil.rmtree(paths.sim_dir)

    if os.path.isdir(paths.sim_dir):
        raise RuntimeError(f"Simulation directory found at: {paths.sim_dir}; use --force flag to remove it")

    if args.seed == -1:
        args.seed = random.getrandbits(64)
        log.info(f"Using random seed: {hex(args.seed)}")

    log.info(f"Creating new simulation directory at: {paths.sim_dir}")
    log.info(f"Creating new event-array storage directory at: {paths.sim_evas}")
    log.info(f"Creating configuration file at: {paths.sim_opts}")
    os.mkdir(paths.sim_dir)
    os.mkdir(paths.sim_evas)

    opts = {fmt_h2u(lopt): getattr(args, fmt_h2u(lopt)) for _, lopt, sub_parsers, _, kwargs in options if new in sub_parsers and load not in sub_parsers}

    with open(paths.sim_opts, "w") as f:
        f.write(f"{fmt_opts(opts)}\n")

# Load saved simulation
if args.command == "load":
    gen_paths()
    source_options()

# ------------------------------------------------------------------------------
# [section] load instruction set
# ------------------------------------------------------------------------------
arch_inst_path = f"arch/{args.vm_arch}/arch_inst.c"
arch_build = Build(arch_inst_path, is_library=True)
arch_build.build()
arch_build.log(log)

arch_dll = ctypes.CDLL(arch_build.binfile)
arch_dll.arch_inst_cap.restype = ctypes.c_int
arch_dll.arch_inst_mnemonic.argtypes = [ctypes.c_uint8]
arch_dll.arch_inst_mnemonic.restype = ctypes.c_char_p

# ------------------------------------------------------------------------------
# [section] assemble ancestor
# ------------------------------------------------------------------------------
anc_path = f"{args.anc_path}/{args.vm_arch}/{args.anc}.asm"

if not os.path.isfile(anc_path):
    raise RuntimeError(f"Could not find ancestor file: {anc_path}")

with open(anc_path, "r") as f:
    lines = f.read().splitlines()

lines = filter(lambda line: line and not line.startswith(";") and not line.isspace(), lines)
lines = map(lambda line: tuple(line.split()), lines)

inst_cap = arch_dll.arch_inst_cap()
mnemo_map = {tuple(arch_dll.arch_inst_mnemonic(byte).decode().split()): byte for byte in range(inst_cap)}
anc_bytes = [mnemo_map[line] for line in lines]
anc_repr = f"{{{",".join(map(str, anc_bytes))}}}"

log.info(f"Compiled ancestor file '{anc_path}' into byte array: {anc_repr}")

# ------------------------------------------------------------------------------
# [section] defines
# ------------------------------------------------------------------------------
defines = [
    f"-DANC=\"{args.anc}\"",
    f"-DANC_BYTES={anc_repr}",
    f"-DANC_SIZE={len(anc_bytes)}",
    f"-DARCH=\"{args.vm_arch}\"",
    f"-DAUTOSAVE_INTERVAL={2 ** args.auto_save_pow}ul",
    f"-DAUTOSAVE_NAME_LEN={len(paths.sim_path) + 18}",
    f"-DCLONES={args.clones}",
    f"-DCOMMAND_{args.command.upper()}",
    f"-DCORES={args.cores}",
    f"-DDATA_PUSH_INTERVAL={2 ** args.data_push_pow}ul",
    f"-DDATA_PUSH_PATH=\"{paths.sim_data}\"",
    f"-DEVA_SAVE_NAME_LEN={len(paths.sim_evas) + 23}",
    f"-DFOR_CORES={" ".join(f"FOR_CORE({i})" for i in range(args.cores))}",
    f"-DMUTA_RANGE={2 ** args.muta_pow}ul",
    f"-DMVEC_SIZE={2 ** args.mvec_pow}ul",
    f"-DNAME=\"{args.name}\"",
    f"-DSEED={args.seed}ul",
    f"-DSIM_DATA=\"{paths.sim_data}\"",
    f"-DSIM_DIR=\"{paths.sim_dir}\"",
    f"-DSIM_EVAS=\"{paths.sim_evas}\"",
    f"-DSIM_OPTS=\"{paths.sim_opts}\"",
    f"-DSIM_PATH=\"{paths.sim_path}\"",
    f"-DSYNC_INTERVAL={2 ** args.sync_pow}ul",
]

# ------------------------------------------------------------------------------
# [section] build and launch
# ------------------------------------------------------------------------------
ui_path = f"{args.ui_path}/{args.ui}"
ui_flags = [f"-Iarch/{args.vm_arch}", "-Icore", f"arch/{args.vm_arch}/arch.c", arch_inst_path, "core/compress.c", "core/logger.c", "core/salis.c", "core/sql.c"]
ui_defines = defines + logger_defines

with open(f"{ui_path}/links.json", "r") as f:
    ui_links = json.load(f)

ui_links += ["-lsqlite3", "-lz"]
ui_build = Build(f"{ui_path}/ui.c", flags=ui_flags, defines=ui_defines, links=ui_links)
ui_build.build()
ui_build.log(log)

if not args.build_only:
    ui_build.run()