182 lines
6.5 KiB
Python
182 lines
6.5 KiB
Python
# G1.2 verification against live app: single UiThread.
|
|
# Flow: PLAY(audio advancing) -> OPEN_GUI ch5 during PLAY (deferred, audio still alive)
|
|
# -> STOP (attach ch5) -> PLAY again -> OPEN_GUI ch6 (Option B closes ch5 editor
|
|
# synchronously, ch6 deferred) -> STOP (attach ch6) -> close ch6 editor via WM_CLOSE
|
|
# -> no EXCEPTION/FAILED/restart anywhere, writeIndex advancing while PLAY.
|
|
import ctypes, ctypes.wintypes as w, struct, subprocess, sys, os, time
|
|
|
|
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
|
|
|
|
APP_PID = int(sys.argv[1])
|
|
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"
|
|
|
|
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})")
|
|
|
|
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, verbose=True):
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
lines = log_lines()
|
|
for i in range(since, len(lines)):
|
|
if pattern in lines[i]:
|
|
if verbose: print(f" LOG: {lines[i].strip()}")
|
|
return lines[i].strip()
|
|
time.sleep(0.2)
|
|
return None
|
|
|
|
def write_index():
|
|
return rd32(4)
|
|
|
|
def sample_wi(seconds):
|
|
a = write_index(); time.sleep(seconds); b = write_index()
|
|
return a, b
|
|
|
|
def bridge_pid():
|
|
try:
|
|
out = subprocess.run(["cmd", "/c", "tasklist /FI \"IMAGENAME eq daw_vst_bridge.exe\" /FO CSV"],
|
|
capture_output=True, text=True).stdout
|
|
for line in out.splitlines()[1:]:
|
|
if "daw_vst_bridge" in line:
|
|
return line.split(',')[1].strip('"')
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
bad = []
|
|
def collect_bad(since):
|
|
for l in log_lines()[since:]:
|
|
if any(k in l for k in ("EXCEPTION", "FAILED", "restarted")):
|
|
bad.append(l.strip())
|
|
|
|
since = len(log_lines())
|
|
pid0 = bridge_pid()
|
|
print("bridge pid:", pid0)
|
|
|
|
# --- Phase A: PLAY, audio advancing ---
|
|
push_controls([(3, 1, 0, 0, "")])
|
|
r = wait_log("transport PLAY", 10, since)
|
|
print("A transport:", r or "TIMEOUT")
|
|
a, b = sample_wi(3)
|
|
print(f"A writeIndex {a}->{b} delta={b-a} ({'ADVANCING' if b > a else 'STALLED'})")
|
|
if b <= a: bad.append("A: writeIndex stalled during PLAY")
|
|
|
|
# --- Phase B: OPEN_GUI ch5 during PLAY (expect deferred) ---
|
|
since = len(log_lines())
|
|
push_controls([(4, 0, 0, 5, "Nexus")])
|
|
r = wait_log("transport PLAY) - queued", 10, since)
|
|
print("B defer ch5:", r or "TIMEOUT")
|
|
a, b = sample_wi(3)
|
|
print(f"B writeIndex {a}->{b} delta={b-a} ({'ADVANCING' if b > a else 'STALLED'})")
|
|
if b <= a: bad.append("B: writeIndex stalled while GUI deferred")
|
|
collect_bad(since)
|
|
|
|
# --- Phase C: STOP -> attach ch5 ---
|
|
since = len(log_lines())
|
|
push_controls([(3, 0, 0, 0, "")])
|
|
r = wait_log("transport STOP", 10, since)
|
|
print("C transport:", r or "TIMEOUT")
|
|
r = wait_log("GUI attached hwnd=", 30, since)
|
|
print("C attach ch5:", r or "TIMEOUT")
|
|
|
|
# --- Phase D: PLAY again -> OPEN_GUI ch6 (Option B synchronous close of ch5) ---
|
|
since = len(log_lines())
|
|
push_controls([(3, 1, 0, 0, "")])
|
|
r = wait_log("transport PLAY", 10, since)
|
|
print("D transport:", r or "TIMEOUT")
|
|
a, b = sample_wi(2)
|
|
print(f"D writeIndex {a}->{b} delta={b-a} ({'ADVANCING' if b > a else 'STALLED'})")
|
|
if b <= a: bad.append("D: writeIndex stalled during PLAY #2")
|
|
since = len(log_lines())
|
|
push_controls([(4, 0, 0, 6, "Nexus")])
|
|
r = wait_log("transport PLAY) - queued", 10, since)
|
|
print("D defer ch6:", r or "TIMEOUT (maybe attach directly?)")
|
|
time.sleep(1)
|
|
collect_bad(since)
|
|
# ch5 editor must be closed (Option B) — check log for close editor of ch5
|
|
closehit = wait_log("closeGUI", 15, since, verbose=False) or ""
|
|
print("D close ch5 editor:", closehit or "no closeGUI log (check)")
|
|
|
|
# --- Phase E: STOP -> attach ch6 ---
|
|
since = len(log_lines())
|
|
push_controls([(3, 0, 0, 0, "")])
|
|
r = wait_log("transport STOP", 10, since)
|
|
print("E transport:", r or "TIMEOUT")
|
|
r = wait_log("GUI attached hwnd=", 30, since)
|
|
print("E attach ch6:", r or "TIMEOUT")
|
|
collect_bad(since)
|
|
|
|
# --- Phase F: close ch6 editor via WM_CLOSE (post_close_gui path) ---
|
|
since = len(log_lines())
|
|
# find ch6 hwnd from log
|
|
hwnd6 = None
|
|
for l in log_lines():
|
|
if "ch=6" in l and "GUI attached" in l:
|
|
import re
|
|
m = re.search(r'hwnd=([0-9A-Fa-f]+)', l)
|
|
if m: hwnd6 = int(m.group(1), 16)
|
|
if hwnd6:
|
|
ctypes.windll.user32.PostMessageW(hwnd6, 0x0010, 0, 0) # WM_CLOSE
|
|
print("F posted WM_CLOSE to ch6 hwnd", hex(hwnd6))
|
|
else:
|
|
print("F hwnd6 not found")
|
|
r = wait_log("closeGUI reload", 30, since)
|
|
print("F close ch6:", r or "TIMEOUT")
|
|
collect_bad(since)
|
|
|
|
# --- watch 8s for crash/restart ---
|
|
since = len(log_lines())
|
|
end = time.time() + 8
|
|
while time.time() < end:
|
|
collect_bad(since)
|
|
pid = bridge_pid()
|
|
if pid != pid0:
|
|
bad.append(f"BRIDGE RESTART {pid0}->{pid}")
|
|
break
|
|
time.sleep(0.2)
|
|
print()
|
|
if bad:
|
|
print("G1.2 RESULT: FAIL")
|
|
for x in bad[:20]: print(" -", x)
|
|
else:
|
|
print("G1.2 RESULT: PASS - 1 UiThread, defer OK, Option B OK, close OK, audio advancing, no crash")
|