177 lines
6.0 KiB
Python
177 lines
6.0 KiB
Python
# Driver verify: SHM + spawn real bridge, reproduce user flow:
|
|
# load 6x Nexus -> open GUI ch=5 -> play notes -> switch instrument ch=5 (crash point)
|
|
# -> reopen GUI -> verify alive + writeIndex advancing + no crash.
|
|
import ctypes, ctypes.wintypes as w, struct, subprocess, sys, os, time
|
|
|
|
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
|
|
|
|
SHM_NAME = "SonicForge_DAW_IPC_VERIFY"
|
|
SIZE = 32768 # bigger than struct, safe
|
|
BRIDGE = r"C:\Users\locpham\SonicForgeStudio\native_bridge\build\Release\daw_vst_bridge.exe"
|
|
PROCDUMP = r"C:\Users\locpham\AppData\Local\Temp\procdump\procdump64.exe"
|
|
DUMPDIR = r"C:\Users\locpham\AppData\Local\Temp\procdump\dumps"
|
|
LOG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "driver_procdump.log")
|
|
NEXUS = r"C:\Program Files\Common Files\VST3\Nexus.vst3"
|
|
AMPLE = r"C:\Program Files\Common Files\VST3\Ample Sound\ACDD.vst3"
|
|
|
|
kernel32 = ctypes.windll.kernel32
|
|
kernel32.CreateFileMappingA.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.c_char_p]
|
|
kernel32.CreateFileMappingA.restype = ctypes.c_void_p
|
|
kernel32.MapViewOfFile.argtypes = [ctypes.c_void_p, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.c_size_t]
|
|
kernel32.MapViewOfFile.restype = ctypes.c_void_p
|
|
INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value
|
|
PAGE_READWRITE = 0x04
|
|
FILE_MAP_ALL_ACCESS = 0xF001F
|
|
FILE_MAP_WRITE = 0x0002
|
|
FILE_MAP_READ = 0x0004
|
|
|
|
hMap = kernel32.CreateFileMappingA(INVALID_HANDLE_VALUE, None, PAGE_READWRITE, 0, SIZE, SHM_NAME.encode())
|
|
if not hMap:
|
|
print("CreateFileMapping failed", ctypes.get_last_error()); sys.exit(1)
|
|
ptr = kernel32.MapViewOfFile(hMap, FILE_MAP_ALL_ACCESS, 0, 0, 0)
|
|
if not ptr:
|
|
print("MapViewOfFile failed", ctypes.get_last_error()); sys.exit(1)
|
|
|
|
def w32(off, fmt, *args):
|
|
ctypes.memmove(ptr + off, struct.pack(fmt, *args), struct.calcsize(fmt))
|
|
|
|
def write_control(type_, arg0, arg1, ch, path):
|
|
# find first free slot
|
|
for i in range(8):
|
|
off = 2836 + i * 1040
|
|
t = ctypes.c_uint32.from_address(ptr + off).value
|
|
if t == 0:
|
|
w32(off, 'IIII', type_, arg0, arg1, ch)
|
|
if path:
|
|
b = path.encode('utf-8')[:1023] + b'\x00'
|
|
ctypes.memmove(ptr + off + 16, b, len(b))
|
|
return i
|
|
raise RuntimeError("control queue full")
|
|
|
|
def submit_controls():
|
|
w32(11156, 'I', 0) # will set below
|
|
# reset type of all slots first
|
|
for i in range(8):
|
|
w32(2836 + i * 1040, 'I', 0)
|
|
|
|
def push_controls(events):
|
|
submit_controls()
|
|
n = 0
|
|
for (type_, arg0, arg1, ch, path) in events:
|
|
write_control(type_, arg0, arg1, ch, path)
|
|
n += 1
|
|
w32(11156, 'I', n)
|
|
|
|
def midi(cmd, ch, pitch, vel, data2=0, data3=0):
|
|
# bridge does not zero midiQueue after drain - always write slot 0, count 1
|
|
w32(2064, 'BBBBBB', cmd, ch, pitch, vel, data2, data3)
|
|
w32(2064 + 8, 'I', 0)
|
|
w32(2832, 'I', 1)
|
|
return True
|
|
|
|
def midi_count():
|
|
return ctypes.c_uint32.from_address(ptr + 2832).value
|
|
|
|
def write_index():
|
|
return ctypes.c_uint32.from_address(ptr + 4).value
|
|
|
|
def wait_log(pattern, timeout, since=0):
|
|
end = time.time() + timeout
|
|
lines = []
|
|
try:
|
|
with open(LOG, 'r', encoding='utf-8', errors='replace') as f:
|
|
lines = f.readlines()
|
|
except FileNotFoundError:
|
|
pass
|
|
last = max(0, len(lines) - since)
|
|
while time.time() < end:
|
|
try:
|
|
with open(LOG, 'r', encoding='utf-8', errors='replace') as f:
|
|
lines = f.readlines()
|
|
except FileNotFoundError:
|
|
lines = []
|
|
for i in range(last, len(lines)):
|
|
if pattern in lines[i]:
|
|
return lines[i].strip()
|
|
last = max(0, len(lines))
|
|
time.sleep(0.2)
|
|
return None
|
|
|
|
proc = subprocess.Popen(
|
|
[PROCDUMP, "-accepteula", "-x", DUMPDIR, BRIDGE, "--shm", SHM_NAME], # second chance = real crash (first-chance caught a benign Nexus json throw)
|
|
env={**os.environ, "SF_SHM_NAME": SHM_NAME, "SF_PARENT_PID": str(os.getpid()),
|
|
"SF_SAMPLE_RATE": "48000", "SF_BLOCK_SIZE": "256"},
|
|
stdout=open(LOG, 'wb'), stderr=subprocess.STDOUT,
|
|
)
|
|
time.sleep(1.5)
|
|
print("bridge pid", proc.pid, "poll", proc.poll())
|
|
|
|
# 1. Load 6 channels Nexus
|
|
print("[1] load 6x Nexus...")
|
|
push_controls([(2, 0, 0, ch, NEXUS) for ch in range(6)])
|
|
print(" wait ch=5 loaded...")
|
|
r = wait_log("instrument loaded ch=5", 90)
|
|
print(" ", r)
|
|
|
|
# 2. Open GUI ch=5
|
|
print("[2] open GUI ch=5...")
|
|
push_controls([(4, 0, 0, 5, "nexus")])
|
|
r = wait_log("GUI attached", 30)
|
|
print(" ", r)
|
|
|
|
# 3. Play notes on ch=5 and ch=2 while editor open
|
|
print("[3] play notes...")
|
|
for k in range(40):
|
|
midi(0x9, 5, 60 + (k % 12), 100)
|
|
midi(0x9, 2, 48 + (k % 7), 90)
|
|
if k % 2 == 0:
|
|
midi(0x8, 5, 60 + ((k - 1) % 12), 0)
|
|
time.sleep(0.05)
|
|
|
|
# 4. Switch instrument ch=5 while GUI open (OLD CRASH POINT)
|
|
print("[4] switch instrument ch=5 -> Ample ACDD while GUI open...")
|
|
t0 = time.time()
|
|
push_controls([(2, 0, 0, 5, AMPLE)])
|
|
r = wait_log("instrument loaded ch=5", 40)
|
|
print(" ", r, "in", round(time.time() - t0, 1), "s")
|
|
alive = proc.poll() is None
|
|
wi = write_index()
|
|
time.sleep(2)
|
|
wi2 = write_index()
|
|
print(" alive:", alive, "writeIndex", wi, "->", wi2, "advancing:", wi2 > wi)
|
|
|
|
# 5. Reopen GUI ch=5 (after switch) - attach path again
|
|
print("[5] reopen GUI ch=5...")
|
|
push_controls([(4, 0, 0, 5, "ample")])
|
|
r = wait_log("GUI attached", 40)
|
|
print(" ", r)
|
|
|
|
# 6. Final liveness check + crash check
|
|
time.sleep(3)
|
|
alive = proc.poll() is None
|
|
wi3 = write_index()
|
|
time.sleep(2)
|
|
wi4 = write_index()
|
|
print("FINAL alive:", alive, "writeIndex", wi3, "->", wi4, "advancing:", wi4 > wi3)
|
|
if not alive:
|
|
print("!!! BRIDGE CRASHED exit", proc.returncode)
|
|
|
|
print("--- last 30 log lines ---")
|
|
try:
|
|
with open(LOG, 'r', encoding='utf-8', errors='replace') as f:
|
|
lines = f.readlines()
|
|
for l in lines[-30:]:
|
|
print(l.rstrip())
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
# cleanup
|
|
kernel32.UnmapViewOfFile(ptr)
|
|
kernel32.CloseHandle(hMap)
|
|
try:
|
|
proc.terminate()
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.5)
|
|
print("done (procdump)")
|