""" vsi_wizard.py - a guided way to make a VSI digital twin, with Python blocks whose behaviour is one plain function per component. python vsi_wizard.py ask the questions, write the files python vsi_wizard.py --answers my_twin.json --run replay saved answers and run it all python vsi_wizard.py --answers my_twin.json --run --stimulus behaviour The wizard asks for: the twin's name and folder, the simulation step and length, the components (name, C++ or Python), their ports (name, protocol), the signals on each port, and the links between them (signal to signal over GenericPayload, a shared CAN frame, or an Ethernet socket pair). It then: 1. writes TWIN.vsi.cmd in the grammar vsiBuild 2026.2 accepts (same emitter conventions as icd_to_vsi.py and sysml2vsi.py: one-token bus comments, in_out on frame signals, opposite directions on connect signals, UART and GPIO folded onto GenericPayload, 8 bits per CAN signal) and TWIN.answers.json so the run is repeatable and editable; 2. with --run: generates with vsiBuild, applies the stimulus, compiles with mingw32-make, runs vsiSim in batch mode, and reads the result from the traces. Two stimulus modes: generic every link carries a counter and the checker verifies every link (default; this is twin_glue/apply_glue_generic.py and check_traces.py) behaviour every PYTHON component gets src/COMP/COMP_behaviour.py with on_step(step, time_ns, inputs) returning a dict of outputs; C++ components keep the generic counters; every component writes a trace CSV per step. Edit the behaviour file and re-run: no rebuild is needed for Python. Nothing here talks to Teamcenter or SysML v2; for those, icd_to_vsi.py and sysml2vsi.py produce the same command file and the same --run steps apply. """ import argparse import importlib.util import json import os import re import shutil import subprocess import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_VSI_HOME = r"D:\InnexisVSI\innexis_home\vsi_2026.2" PROTOCOLS = ["GenericPayload", "Can", "Ethernet", "Uart", "Gpio", "Lin", "Spi"] FOLD = {"Uart": "GenericPayload", "Gpio": "GenericPayload"} # no C++/Python peer gateway for these GATEWAY_PREFIX = {"C++": "c++2Dt", "Python": "python2Dt"} CAN_BITS = 8 IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") # --------------------------------------------------------------------------- questions def ask(prompt, default=None, choices=None, validate=None): while True: tail = "" if choices: tail = " [" + "/".join(choices) + "]" if default is not None: tail += " (" + str(default) + ")" raw = input(prompt + tail + ": ").strip() if not raw and default is not None: raw = str(default) if choices and raw not in choices: print(" choose one of: " + ", ".join(choices)); continue if validate and not validate(raw): print(" not valid, try again"); continue return raw def ask_int(prompt, default): return int(ask(prompt, default, validate=lambda s: s.isdigit())) def collect(): print("VSI twin wizard. Enter accepts the default in parentheses. Names are identifiers.") a = {} a["twin"] = ask("Twin name", "MyTwinDT", validate=IDENT.match) a["workspace"] = ask("Workspace root (short path if any block is Python)", r"D:\vw\%s" % a["twin"].lower()) a["step_ns"] = ask_int("Simulation step in ns", 100000) a["total_ns"] = ask_int("Total simulation time in ns", 10000000) a["components"] = [] n = ask_int("How many components", 2) for i in range(n): c = {"name": ask(" Component %d name" % (i + 1), "Comp%d" % (i + 1), validate=IDENT.match)} c["lang"] = ask(" Language", "Python" if i % 2 == 0 else "C++", choices=["C++", "Python"]) c["ports"] = [] np_ = ask_int(" How many ports on %s" % c["name"], 1) for j in range(np_): p = {"name": ask(" Port %d name" % (j + 1), "%s_p%d" % (c["name"].lower(), j + 1), validate=IDENT.match)} p["protocol"] = ask(" Protocol", "GenericPayload", choices=PROTOCOLS) p["signals"] = [] ns = ask_int(" How many signals on %s" % p["name"], 1) for k in range(ns): p["signals"].append({"name": ask(" Signal %d name" % (k + 1), "%s_sig%d" % (p["name"], k + 1), validate=IDENT.match)}) c["ports"].append(p) a["components"].append(c) a["links"] = [] print("Links. A signal link is SOURCE.signal to DEST.signal over GenericPayload ports.") print("A CAN link names the ports that share one frame; the first named port owns the frame.") print("An Ethernet link is one server port and one client port.") while True: kind = ask(" Add a link", "done", choices=["signal", "can", "ethernet", "done"]) if kind == "done": break if kind == "signal": a["links"].append({"type": "signal", "src": ask(" Source COMPONENT.signal"), "dst": ask(" Destination COMPONENT.signal"), "bus": ask(" Bus name (one word)", "Signals")}) elif kind == "can": members = ask(" Member COMPONENT.port list, comma separated").replace(" ", "").split(",") a["links"].append({"type": "can", "id": ask(" Frame id", "0x100"), "members": members, "bus": ask(" Bus name (one word)", "CanBus")}) else: a["links"].append({"type": "ethernet", "server": ask(" Server COMPONENT.port"), "client": ask(" Client COMPONENT.port"), "tcp": ask_int(" TCP port number", 8800), "bus": ask(" Bus name (one word)", "Ethernet")}) return a # --------------------------------------------------------------------------- emitter class WizardError(Exception): pass def index(a): comps = {c["name"]: c for c in a["components"]} ports = {} signals = {} for c in a["components"]: for p in c["ports"]: ports[(c["name"], p["name"])] = p for s in p["signals"]: key = (c["name"], s["name"]) if key in signals: raise WizardError("signal %s.%s declared twice" % key) signals[key] = (p, s) return comps, ports, signals def split_ref(ref, what): if "." not in ref: raise WizardError("%s must be COMPONENT.NAME, got %r" % (what, ref)) c, n = ref.split(".", 1) return c, n def emit(a): comps, ports, signals = index(a) problems, notes = [], [] direction = {} # (comp, sig) -> input | output | in_out lines = [] w = lines.append # resolve links first so signal directions are known before they are declared signal_links, frames, sockets = [], [], [] for l in a["links"]: if l["type"] == "signal": sc, ss = split_ref(l["src"], "signal source"); dc, ds = split_ref(l["dst"], "signal destination") for key, what in (((sc, ss), "source"), ((dc, ds), "destination")): if key not in signals: problems.append("link %s: %s %s.%s is not a declared signal" % (l["src"], what, key[0], key[1])); continue if problems: continue sp, dp = signals[(sc, ss)][0], signals[(dc, ds)][0] for p in (sp, dp): if FOLD.get(p["protocol"], p["protocol"]) != "GenericPayload": problems.append("link %s to %s: signal links need GenericPayload (or Uart/Gpio) ports, %s is %s" % (l["src"], l["dst"], p["name"], p["protocol"])) if direction.get((sc, ss), "output") != "output": problems.append("%s.%s is both a source and a destination" % (sc, ss)) if direction.get((dc, ds), "input") != "input": problems.append("%s.%s is both a destination and a source" % (dc, ds)) direction[(sc, ss)] = "output"; direction[(dc, ds)] = "input" signal_links.append((sc, ss, dc, ds, sp["name"], dp["name"], l.get("bus", "Signals"))) elif l["type"] == "can": members = [] for m in l["members"]: c, pn = split_ref(m, "CAN member") if (c, pn) not in ports: problems.append("CAN member %s is not a declared port" % m); continue if ports[(c, pn)]["protocol"] != "Can": problems.append("CAN member %s is a %s port" % (m, ports[(c, pn)]["protocol"])); continue members.append((c, pn)) for s in ports[(c, pn)]["signals"]: direction[(c, s["name"])] = "in_out" frames.append((l.get("id", "0x100"), members, l.get("bus", "CanBus"))) elif l["type"] == "ethernet": sc, sp = split_ref(l["server"], "Ethernet server"); cc, cp = split_ref(l["client"], "Ethernet client") for key in ((sc, sp), (cc, cp)): if key not in ports or ports[key]["protocol"] != "Ethernet": problems.append("Ethernet end %s.%s is not a declared Ethernet port" % key) sockets.append((sc, sp, cc, cp, int(l.get("tcp", 8800)), l.get("bus", "Ethernet"))) else: problems.append("unknown link type %r" % l["type"]) if problems: raise WizardError("\n".join(problems)) # language rules that vsiBuild will not tell you about until much later lang = {} for c in a["components"]: lang[c["name"]] = c["lang"] seen_protocols = {} for p in c["ports"]: proto = FOLD.get(p["protocol"], p["protocol"]) if proto in seen_protocols and proto != "Gpio": raise WizardError("%s has two %s ports (%s, %s); vsiBuild allows one port per protocol per component" % (c["name"], proto, seen_protocols[proto], p["name"])) seen_protocols[proto] = p["name"] if c["lang"] == "Python" and proto == "Ethernet": # vsiBuild 2026.2 emits an empty establishTcpUdpConnection for a Python component whose # socket carries no connected signals (measured 2026-09-08); the wizard never connects # signals over sockets, so keep such a block C++ and say so. lang[c["name"]] = "C++" notes.append("%s kept C++: a Python component with an Ethernet port generates invalid Python in vsiBuild 2026.2" % c["name"]) if not p["signals"]: raise WizardError("port %s.%s has no signals; vsiBuild refuses a component without signals" % (c["name"], p["name"])) for p in c["ports"]: for s in p["signals"]: key = (c["name"], s["name"]) if key not in direction: direction[key] = "output" if p["protocol"] != "Can" else "in_out" if p["protocol"] != "Can": notes.append("%s.%s is not linked; declared output" % key) w("# Innexis VSI Builder command file") w("# GENERATED by vsi_wizard.py - edit the .answers.json and regenerate.") w("#") w("set digitalTwinName %s" % a["twin"]) w("set workspaceDir %s" % a["workspace"].replace("\\", "/")) w("set digitalTwinPlatform mingw64") w("set simulationStep %d" % int(a["step_ns"])) w("set totalSimTime %d" % int(a["total_ns"])) w("") w("# --- components and ports ---------------------------------------------") for c in a["components"]: w("add component -type %s -name %s" % (lang[c["name"]], c["name"])) for p in c["ports"]: proto = FOLD.get(p["protocol"], p["protocol"]) extra = "" if p["protocol"] == "Can" and lang[c["name"]] == "Python": extra = " -canVersion can" if p["protocol"] in ("Spi", "Lin"): extra = " -nodeType master" notes.append("%s.%s declared master; change to slave by hand if needed" % (c["name"], p["name"])) w("add port -componentName %s -name %s -gateway %s%s%s" % (c["name"], p["name"], GATEWAY_PREFIX[lang[c["name"]]], proto, extra)) if proto != p["protocol"]: notes.append("%s.%s: %s rides GenericPayload between generated components" % (c["name"], p["name"], p["protocol"])) w("") w("# --- signals -------------------------------------------------------------") for c in a["components"]: sigs = ["%s:int:%s" % (s["name"], direction[(c["name"], s["name"])]) for p in c["ports"] for s in p["signals"]] w("define componentSignals -componentName %s -signals [%s]" % (c["name"], ",".join(sigs))) w("") w("# --- connectivity ------------------------------------------------------") for sc, sp, cc, cp, tcp, bus in sockets: w("# bus %s [Ethernet] tcp %d" % (re.sub(r"\s+", "_", bus), tcp)) w("add portSocket -componentName %s -portName %s -ipPortNum %d -socketType server -protocol tcpIp" % (sc, sp, tcp)) w("add portSocket -componentName %s -portName %s -ipPortNum %d -socketType client -protocol tcpIp" % (cc, cp, tcp)) for fid, members, bus in frames: w("# bus %s [Can] frame %s on %d port(s)" % (re.sub(r"\s+", "_", bus), fid, len(members))) for c, pn in members: layout = ",".join("%s:%d:%d" % (s["name"], k * CAN_BITS, CAN_BITS) for k, s in enumerate(ports[(c, pn)]["signals"])) w("define frame -componentName %s -portName %s -id %s -signals [%s] -idType standard" % (c, pn, fid, layout)) notes.append("frame %s: %d bits per signal in declaration order; owner is %s (first member)" % (fid, CAN_BITS, members[0][0] if members else "?")) by_bus = {} for sc, ss, dc, ds, spn, dpn, bus in signal_links: by_bus.setdefault(re.sub(r"\s+", "_", bus), []).append((sc, ss, dc, ds, spn, dpn)) for bus, items in by_bus.items(): w("# bus %s [GenericPayload] %d link(s)" % (bus, len(items))) for sc, ss, dc, ds, spn, dpn in items: w("connect signals -sourceSignal %s.%s -destSignal %s.%s -sourcePortName %s -destPortName %s" % (sc, ss, dc, ds, spn, dpn)) w("") w("generate -overwrite -no-build") w("quit") return "\n".join(lines) + "\n", notes, lang # --------------------------------------------------------------------------- running def load_module(name, path): spec = importlib.util.spec_from_file_location(name, path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod def vsi_env(): """make the VSI environment available to subprocesses without a PowerShell script""" home = os.environ.get("INNEXIS_VSI_HOME") or DEFAULT_VSI_HOME env = dict(os.environ) cfg = os.path.join(home, "env_vsi.win.txt") if os.path.isfile(cfg): for line in open(cfg, encoding="utf-8", errors="replace"): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) env.setdefault(k.strip(), v.strip()) env["INNEXIS_VSI_HOME"] = home # vsiBuild needs the derived variables the vendor's env_vsi.ps1 computes (MINGW_PATH, # SYSTEMC, XL_VIP_HOME, UVMC_HOME, ...; measured: "MINGW_PATH must be set on Windows"). # Run that script once in a child PowerShell and harvest its environment. if "MINGW_PATH" not in env: script = os.path.join(home, "env_vsi.ps1") for shell in ("pwsh", "powershell"): try: p = subprocess.run([shell, "-NoProfile", "-Command", ". '%s' | Out-Null; Get-ChildItem env: | ForEach-Object { $_.Name + '=' + $_.Value }" % script], capture_output=True, text=True, timeout=120) except (OSError, subprocess.TimeoutExpired): continue if p.returncode == 0: for line in p.stdout.splitlines(): if "=" in line: k, v = line.split("=", 1) env[k] = v break mingw = os.path.join(home, "common", "mingw64", "mingw64-11.2.0", "bin") # The generated Makefiles use cmd.exe syntax ("if not exist X (mkdir X)"). GNU make picks # Git's sh.exe as its shell when it is on PATH (launching from Git Bash puts it there) and # then fails on the first rule. Keep Git's usr/bin off the child PATH and name cmd.exe. keep = [p for p in env.get("PATH", "").split(os.pathsep) if not re.search(r"[\\/]Git[\\/](usr[\\/])?bin", p, re.I) and not re.search(r"[\\/]Git[\\/]mingw64[\\/]bin", p, re.I)] env["PATH"] = mingw + os.pathsep + os.path.join(home, "bin") + os.pathsep + os.pathsep.join(keep) env["SHELL"] = env.get("ComSpec", r"C:\Windows\System32\cmd.exe") for k in ("MAKESHELL", "MSYSTEM", "MSYS"): env.pop(k, None) if "MINGW_PATH" not in env: raise WizardError("could not establish the VSI environment (MINGW_PATH); source env_vsi.ps1 first") return env, home def kill_twin_processes(twin_dir): """stop everything a stalled run left behind: vsiSim's children are not in our process tree, so match on the command line (they all carry the twin folder or its .dt name)""" needle = twin_dir.replace("/", "\\") ps = ("Get-CimInstance Win32_Process | Where-Object { $_.CommandLine -and $_.CommandLine.Replace('/', '\\') -like '*%s*' " "-and $_.Name -match 'vsiSim|FabricServer|VsiClient|python|mingw32-make|cmd' } | " "ForEach-Object { Stop-Process -Id $_.ProcessId -Force -ErrorAction SilentlyContinue; $_.Name }") % needle for shell in ("pwsh", "powershell"): try: r = subprocess.run([shell, "-NoProfile", "-Command", ps], capture_output=True, text=True, timeout=60) if r.returncode == 0: names = [n for n in r.stdout.split() if n] if names: print(" stopped leftover process(es): " + ", ".join(names)) return except (OSError, subprocess.TimeoutExpired): continue def run_step(title, cmd, cwd, env, log_path, timeout, stdin_path=None): print("== %s" % title) print(" " + " ".join(cmd)) t0 = time.time() with open(log_path, "w", encoding="utf-8", errors="replace") as log: stdin = open(stdin_path, "rb") if stdin_path else subprocess.DEVNULL try: p = subprocess.run(cmd, cwd=cwd, env=env, stdin=stdin, stdout=log, stderr=subprocess.STDOUT, timeout=timeout) code = p.returncode except subprocess.TimeoutExpired: code = "timeout" finally: if stdin_path: stdin.close() print(" exit %s in %d s, log %s" % (code, int(time.time() - t0), log_path)) return code def behaviour_stub(comp, inputs, outputs): body = ['"""Behaviour of %s. Edit freely; the twin does not need rebuilding for Python.' % comp, "", "on_step is called once per simulation step with the values received so far.", "inputs : dict, input signal name to int (what other components sent)", "return : dict, output signal name to int (what this component sends this step)", "Values are ints; keep CAN frame signals within 0 to 255 (8 bits each).", '"""', "", "", "def on_step(step, time_ns, inputs):", " outs = {}"] for i, o in enumerate(outputs): # a saw-tooth distinct from the generic counters, so a trace shows which glue produced it body.append(" outs[%r] = (7 * step + %d) %% 256" % (o, i * 16)) if inputs: body.append(" # example reaction: echo the first input into the log every 10 steps") body.append(" if step % 10 == 0:") body.append(" print('%s step', step, 'sees', {k: inputs[k] for k in inputs})" % comp) body.append(" return outs") return "\n".join(body) + "\n" def behaviour_regions(comp, inputs, outputs, muted): cols = inputs + outputs glob = ["import os as _os, sys as _sys", "_sys.path.insert(0, _os.path.dirname(_os.path.abspath(__file__)))", "_sys.path.insert(0, _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), '..'))", "import generic_stim as twin", "import %s_behaviour as behaviour" % comp, 'g_glue = twin.Glue("%s", [%s], 0)' % (comp, ", ".join('"%s"' % c for c in cols))] if muted: glob.append('vsiCanPythonGateway = twin.MutedSend(vsiCanPythonGateway, "sendCanPacket") # consumer of a frame it does not own') before = ["_ins = {%s}" % ", ".join('"%s": self.mySignals.%s' % (s, s) for s in inputs), "_outs = behaviour.on_step(g_glue.step, vsiCommonPythonApi.getSimulationTimeInNs(), _ins)", "for _k, _v in _outs.items():", "\tsetattr(self.mySignals, _k, int(_v))"] after = ["g_glue.trace.row(g_glue.step, vsiCommonPythonApi.getSimulationTimeInNs(), [%s])" % ", ".join("self.mySignals.%s" % c for c in cols), "g_glue.step += 1"] return {"Global Variables & Definitions": glob, "Before sending the packet": before, "After sending the packet": after} def apply_behaviour(cmd_path, twin_dir, answers, lang): glue = load_module("apply_glue_generic", os.path.join(HERE, "twin_glue", "apply_glue_generic.py")) wiring = json.load(open(os.path.join(twin_dir, "glue_wiring.json"), encoding="utf-8")) muted = set(wiring.get("silent_can", [])) written = [] for c in answers["components"]: if lang[c["name"]] != "Python": continue # the generic glue's wiring already decided roles: a CAN frame signal is an output on # the frame owner and an input everywhere else, a linked signal is what its link says produced = {e[0] for e in wiring["components"][c["name"]]["produce"]} inputs, outputs = [], [] for p in c["ports"]: for s in p["signals"]: (outputs if s["name"] in produced else inputs).append(s["name"]) stub = os.path.join(twin_dir, "src", c["name"], "%s_behaviour.py" % c["name"]) if not os.path.isfile(stub): with open(stub, "w", encoding="utf-8", newline="\n") as fh: fh.write(behaviour_stub(c["name"], inputs, outputs)) written.append(stub) err = glue.patch(os.path.join(twin_dir, "src", c["name"], c["name"] + ".py"), behaviour_regions(c["name"], inputs, outputs, c["name"] in muted)) if err: raise WizardError("%s: %s" % (c["name"], err)) return written def summarize_traces(twin_dir): logs = os.path.join(twin_dir, "vsi.sim", "_logs") print("== traces in %s" % logs) for f in sorted(os.listdir(logs)) if os.path.isdir(logs) else []: if f.endswith("_trace.csv"): rows = open(os.path.join(logs, f), encoding="utf-8").read().strip().split("\n") print(" %-40s %3d rows first %s last %s" % (f, len(rows) - 1, rows[1] if len(rows) > 1 else "-", rows[-1])) def run_all(cmd_path, answers, lang, stimulus, skip_build=False): env, home = vsi_env() twin_dir = os.path.join(answers["workspace"], answers["twin"]) base = os.path.dirname(os.path.abspath(cmd_path)) vsibuild = os.path.join(home, "bin", "vsiBuild.exe") vsisim = os.path.join(home, "bin", "vsiSim.exe") if not os.path.isfile(vsibuild): raise WizardError("vsiBuild not found at %s; set INNEXIS_VSI_HOME" % vsibuild) os.makedirs(answers["workspace"], exist_ok=True) code = run_step("generate", [vsibuild, "-c", "-f", os.path.abspath(cmd_path)], base, env, os.path.join(base, answers["twin"] + ".generate.log"), 300, stdin_path=cmd_path) if code != 0 or not os.path.isdir(twin_dir): raise WizardError("generate failed (exit %s); read the log" % code) code = run_step("stimulus: generic glue", [sys.executable, os.path.join(HERE, "twin_glue", "apply_glue_generic.py"), os.path.abspath(cmd_path), twin_dir], base, env, os.path.join(base, answers["twin"] + ".glue.log"), 120) if code != 0: raise WizardError("glue failed; read the log") if stimulus == "behaviour": written = apply_behaviour(cmd_path, twin_dir, answers, lang) for wpath in written: print(" behaviour stub written: %s" % wpath) # compile the skeleton AND the behaviour stub before anything is launched: a syntax # error in either makes that client exit at start, and the fabric then waits for it # forever (measured: a stray doubled percent sign in a stub hung a run past 10 min) for c in answers["components"]: if lang[c["name"]] == "Python": for f in (c["name"] + ".py", c["name"] + "_behaviour.py"): src = os.path.join(twin_dir, "src", c["name"], f) r = subprocess.run([sys.executable, "-m", "py_compile", src], capture_output=True, text=True) if r.returncode != 0: raise WizardError("%s does not compile:\n%s" % (src, (r.stderr or r.stdout).strip())) if not skip_build: # full path: CreateProcess resolves a bare name against the PARENT's PATH, not the env passed make = os.path.join(home, "common", "mingw64", "mingw64-11.2.0", "bin", "mingw32-make.exe") code = run_step("compile and build", [make, "compile", "build"], twin_dir, env, os.path.join(base, answers["twin"] + ".build.log"), 1800) if code != 0: raise WizardError("build failed (exit %s); read the log" % code) # bound the run at three times the wall clock a 63-component twin needs; a client that # died at start leaves the fabric waiting for it forever, and only a kill ends that code = run_step("run (vsiSim batch)", [vsisim, answers["twin"] + ".dt", "--batch", "--run"], twin_dir, env, os.path.join(base, answers["twin"] + ".sim.log"), 300) kill_twin_processes(twin_dir) if code == "timeout": print(" the run did not finish in 300 s; read vsi.sim/_logs/check.COMPONENT.log for a client that exited at start") # The verdict on "did it run to the end" is the fabric log's stop line plus a trace row per # step from every glued component. Not the exit code: the fabric server exits 0xC0000374 # at teardown after a complete run (measured on 3- and 63-component twins). fabric_log = os.path.join(twin_dir, "vsi.sim", "_logs", "check.%s.log" % answers["twin"]) text = open(fabric_log, encoding="utf-8", errors="replace").read() if os.path.isfile(fabric_log) else "" stopped = ("Simulation stopped by user" in text) or ("Total SystemC time" in text) expected = int(answers["total_ns"]) // int(answers["step_ns"]) logs = os.path.join(twin_dir, "vsi.sim", "_logs") traces = [f for f in os.listdir(logs) if f.endswith("_trace.csv")] if os.path.isdir(logs) else [] rows = {f: len(open(os.path.join(logs, f), encoding="utf-8").read().strip().split("\n")) - 1 for f in traces} complete = stopped and traces and all(n == expected for n in rows.values()) print(" run complete: %s (fabric stop line %s; %d trace(s), %s of %d steps each; exit code is not the verdict)" % ( complete, stopped, len(traces), "all" if complete else str(sorted(set(rows.values()))), expected)) done = complete if stimulus == "generic": code = run_step("check", [sys.executable, os.path.join(HERE, "twin_glue", "check_traces.py"), twin_dir, "--traces", os.path.join(twin_dir, "vsi.sim", "_logs")], base, env, os.path.join(base, answers["twin"] + ".check.log"), 300) last = open(os.path.join(base, answers["twin"] + ".check.log"), encoding="utf-8", errors="replace").read().strip().split("\n")[-1] print(" " + last) summarize_traces(twin_dir) return done # --------------------------------------------------------------------------- main def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--answers", help="JSON answers file to replay instead of asking") ap.add_argument("--out", help="command file to write (default TWIN.vsi.cmd beside the answers file or here)") ap.add_argument("--run", action="store_true", help="generate, glue, build, run and check") ap.add_argument("--stimulus", choices=["generic", "behaviour"], default="generic") ap.add_argument("--skip-build", action="store_true", help="with --run: reuse the compiled twin (Python behaviour edits need no rebuild)") args = ap.parse_args() if args.answers: answers = json.load(open(args.answers, encoding="utf-8")) base = os.path.dirname(os.path.abspath(args.answers)) else: answers = collect() base = HERE try: text, notes, lang = emit(answers) except WizardError as e: print("vsi_wizard: cannot write a valid twin:\n " + str(e).replace("\n", "\n ")) sys.exit(1) out = args.out or os.path.join(base, answers["twin"] + ".vsi.cmd") with open(out, "w", encoding="utf-8", newline="\n") as fh: fh.write(text) ans_path = os.path.join(base, answers["twin"] + ".answers.json") if not args.answers or os.path.abspath(args.answers) != os.path.abspath(ans_path): with open(ans_path, "w", encoding="utf-8", newline="\n") as fh: json.dump(answers, fh, indent=1) print("wrote %s and %s" % (out, ans_path)) for n in notes: print(" note: " + n) if args.run: try: ok = run_all(out, answers, lang, args.stimulus, args.skip_build) except WizardError as e: print("vsi_wizard: " + str(e)); sys.exit(1) sys.exit(0 if ok else 1) if __name__ == "__main__": main()