160 lines
6.1 KiB
Python
160 lines
6.1 KiB
Python
# G1.5 regression stress driver: N iterations of LOAD/OPEN_GUI/PLAY/STOP/WM_CLOSE
|
|
# on Nexus + EZkeys. Watch: bridge process alive, no new EXCEPTION/FAILED, writeIndex advancing.
|
|
import ctypes, ctypes.wintypes as w, struct, subprocess, sys, os, time, re
|
|
|
|
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
|
|
|
|
APP_PID = int(sys.argv[1])
|
|
ITERS = int(sys.argv[2]) if len(sys.argv) > 2 else 5
|
|
SHM_NAME = f"SonicForge_DAW_IPC_{APP_PID}".encode()
|
|
APP_LOG = os.path.join(os.environ.get('APPDATA', '.'), 'SonicForgeDAW', 'logs', 'bridge.log')
|
|
NEXUS = r"C:\Program Files\Common Files\VST3\Nexus.vst3"
|
|
EZKEYS = r"C:\Program Files\Common Files\VST3\Toontrack\EZkeys 2.vst3"
|
|
|
|
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
|
|
kernel32.OpenFileMappingA.argtypes = [ctypes.wintypes.DWORD, w.BOOL, ctypes.c_char_p]
|
|
kernel32.OpenFileMappingA.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
|
|
hMap = kernel32.OpenFileMappingA(0xF001F, False, SHM_NAME)
|
|
if not hMap:
|
|
print(f"OpenFileMapping failed err={ctypes.get_last_error()}"); sys.exit(1)
|
|
ptr = kernel32.MapViewOfFile(hMap, 0xF001F, 0, 0, 0)
|
|
print(f"SHM mapped ({SHM_NAME}) iters={ITERS}")
|
|
|
|
def w32(off, fmt, *args):
|
|
ctypes.memmove(ptr + off, struct.pack(fmt, *args), struct.calcsize(fmt))
|
|
def rd32(off):
|
|
return ctypes.c_uint32.from_address(ptr + off).value
|
|
|
|
def write_control(ctype, arg0, arg1, ch, path):
|
|
for i in range(8):
|
|
off = 2836 + i * 1040
|
|
if rd32(off) == 0:
|
|
w32(off, 'IIII', ctype, 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 push_controls(events):
|
|
for i in range(8):
|
|
w32(2836 + i * 1040, 'I', 0)
|
|
for (ctype, arg0, arg1, ch, path) in events:
|
|
write_control(ctype, arg0, arg1, ch, path)
|
|
w32(11156, 'I', len(events))
|
|
|
|
def log_lines():
|
|
try:
|
|
with open(APP_LOG, 'r', encoding='utf-8', errors='replace') as f:
|
|
return f.readlines()
|
|
except FileNotFoundError:
|
|
return []
|
|
|
|
def wait_log(pattern, timeout, since):
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
for i in range(since, len(log_lines())):
|
|
if pattern in log_lines()[i]:
|
|
return log_lines()[i]
|
|
time.sleep(0.2)
|
|
return None
|
|
|
|
def wait_gui(ch, timeout, since):
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
for i in range(since, len(log_lines())):
|
|
l = log_lines()[i]
|
|
if "GUI attached hwnd=" in l and f"ch={ch}" in l:
|
|
return l
|
|
time.sleep(0.2)
|
|
return None
|
|
|
|
def bridge_pid():
|
|
out = subprocess.run(['powershell', '-Command',
|
|
'(Get-Process daw_vst_bridge -ErrorAction SilentlyContinue).Id'],
|
|
capture_output=True, text=True).stdout.strip().splitlines()
|
|
return [int(x) for x in out if x.strip().isdigit()]
|
|
|
|
def bad_since(since):
|
|
return [l for l in log_lines()[since:] if 'EXCEPTION' in l or 'FAILED' in l]
|
|
|
|
def find_hwnd(ch):
|
|
last = None
|
|
for l in log_lines():
|
|
if "GUI attached hwnd=" in l and f"ch={ch}" in l:
|
|
m = re.search(r'hwnd=([0-9A-Fa-f]+)', l)
|
|
if m: last = int(m.group(1), 16)
|
|
return last
|
|
|
|
fails = 0
|
|
for it in range(ITERS):
|
|
print(f"--- iter {it} ---")
|
|
since = len(log_lines())
|
|
# LOAD Nexus ch5
|
|
push_controls([(2, 0, 0, 5, NEXUS)])
|
|
if not wait_log("instrument loaded ch=5 type=0", 30, since):
|
|
print(f"iter{it} LOAD Nexus TIMEOUT"); fails += 1; continue
|
|
# OPEN_GUI ch5
|
|
push_controls([(4, 0, 0, 5, NEXUS)])
|
|
if not wait_gui(5, 30, since):
|
|
print(f"iter{it} OPEN_GUI Nexus TIMEOUT"); fails += 1; continue
|
|
# PLAY -> writeIndex advancing while editor open (G1.4 core)
|
|
wi0 = rd32(4)
|
|
push_controls([(3, 1, 0, 0, "")])
|
|
if not wait_log("transport PLAY", 10, since):
|
|
print(f"iter{it} PLAY TIMEOUT"); fails += 1; continue
|
|
time.sleep(1.5)
|
|
wi1 = rd32(4)
|
|
ok = (wi1 - wi0) > 50
|
|
print(f"iter{it} PLAY editor-open writeIndex {wi0}->{wi1} delta={wi1-wi0} {'OK' if ok else 'STALLED'}")
|
|
if not ok: fails += 1
|
|
# STOP
|
|
push_controls([(3, 0, 0, 0, "")])
|
|
if not wait_log("transport STOP", 10, since):
|
|
print(f"iter{it} STOP TIMEOUT"); fails += 1; continue
|
|
# WM_CLOSE ch5 -> state restored
|
|
hwnd = find_hwnd(5)
|
|
if hwnd:
|
|
ctypes.windll.user32.PostMessageW(hwnd, 0x0010, 0, 0)
|
|
r = wait_log("state restored ch=5", 30, since)
|
|
print(f"iter{it} WM_CLOSE Nexus hwnd={hex(hwnd)}: {'OK' if r else 'TIMEOUT'}")
|
|
if not r: fails += 1
|
|
else:
|
|
print(f"iter{it} hwnd Nexus not found"); fails += 1
|
|
# LOAD EZkeys ch5 + OPEN_GUI + PLAY + STOP + CLOSE
|
|
since = len(log_lines())
|
|
push_controls([(2, 0, 0, 5, EZKEYS)])
|
|
if not wait_log("instrument loaded ch=5 type=0", 30, since):
|
|
print(f"iter{it} LOAD EZkeys TIMEOUT"); fails += 1; continue
|
|
push_controls([(4, 0, 0, 5, EZKEYS)])
|
|
if not wait_gui(5, 30, since):
|
|
print(f"iter{it} OPEN_GUI EZkeys TIMEOUT"); fails += 1; continue
|
|
push_controls([(3, 1, 0, 0, "")])
|
|
wait_log("transport PLAY", 10, since)
|
|
time.sleep(1.5)
|
|
push_controls([(3, 0, 0, 0, "")])
|
|
wait_log("transport STOP", 10, since)
|
|
hwnd = find_hwnd(5)
|
|
if hwnd:
|
|
ctypes.windll.user32.PostMessageW(hwnd, 0x0010, 0, 0)
|
|
r = wait_log("state restored ch=5", 30, since)
|
|
print(f"iter{it} WM_CLOSE EZkeys hwnd={hex(hwnd)}: {'OK' if r else 'TIMEOUT'}")
|
|
if not r: fails += 1
|
|
else:
|
|
print(f"iter{it} hwnd EZkeys not found"); fails += 1
|
|
bad = bad_since(since)
|
|
if bad:
|
|
print(f"iter{it} EXCEPTION/FAILED: {bad[-2:]}"); fails += 1
|
|
|
|
# final watch: bridge alive + no new bad
|
|
since = len(log_lines())
|
|
time.sleep(3)
|
|
bad = bad_since(since)
|
|
alive = bridge_pid()
|
|
print(f"bridge alive: {alive} (expect non-empty)") if alive else print("BRIDGE DEAD")
|
|
if not alive: fails += 1
|
|
print(f"G1.5 RESULT: {'PASS' if fails == 0 else f'FAIL ({fails} issues)'}")
|
|
sys.exit(0 if fails == 0 else 1)
|