Matiq/soothe2-re
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

step7: live BandConfig capture infra + FUN_180563a60 decode — GUI-only, audio path refutes premise (22b)

Browse Source
This commit is contained in:
Matiq
2026-08-22 15:28:01 +03:00
parent c222e054ca
commit 7fac8b2a4f
5 changed files with 304 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/step7_capture.py
+242
View File
@@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""step7_capture.py — Step 7: live capture of level-path BandConfig A/B/gamma per RPP.
Method (NOTES_CAPTURE 2026-08-20c):
1. spawn reaper with <rpp> (+ play.lua realtime transport by default),
2. wait for yabridge-host (soothe2 mapped, not reaper), sleep for init,
3. chunked full-heap snapshot (8MB pread chunks; large regions EIO otherwise),
4. fingerprint scan: per-band level_gain pair buffers = 0x400 [level,gain]
f32 pairs with level[j] == j/1024 EXACTLY (j<1024 -> exact in fp32),
5. u64 refs to those buffers land at base+0xe0+band*0x18 (stride 0x18)
-> majority-vote the level-path object base,
6. decode every qword ptr at base+0x170..0x1a8 as BandConfig:
A@+0x0 B@+0x4 gamma@+0xc flag@+0x10 shaper@+0x18.. callback@+0x90,
plus first mask doubles at base+0x4198+band*0x2000.
Usage:
python3 scripts/step7_capture.py <file.rpp> [--offline] [--json PATH] [--snap PATH]
Default mode is realtime playback (play.lua) so short projects keep the DSP
host alive during capture. --offline uses -renderproject instead.
"""
import subprocess, os, glob, time, struct, sys, json, collections
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PLAY_LUA = os.path.join(REPO, 'play.lua')
def find_host():
for p in glob.glob('/proc/[0-9]*'):
pid = int(os.path.basename(p))
try:
cmd = open(f'/proc/{pid}/cmdline', 'rb').read().replace(b'\0', b' ').decode('utf8', 'replace')
maps = open(f'/proc/{pid}/maps').read()
except Exception:
continue
if 'soothe2' in maps and 'reaper' not in cmd:
return pid
return None
def snapshot(host, path):
fd = os.open(f'/proc/{host}/mem', os.O_RDONLY)
out = open(path, 'wb')
nreg = 0; nbytes = 0
for line in open(f'/proc/{host}/maps').read().splitlines():
p = line.split()
if len(p) < 2: continue
lo, hi = (int(x, 16) for x in p[0].split('-'))
if 'r' not in p[1]: continue
a = lo
while a < hi:
n = min(hi - a, 8 * 1024 * 1024)
try:
d = os.pread(fd, n, a)
except Exception:
a += n; continue
if not d:
a += n; continue
out.write(struct.pack('<QQ', a, len(d))); out.write(d)
nreg += 1; nbytes += len(d)
a += n
out.close(); os.close(fd)
return nreg, nbytes
def parse_snap(path):
regs = []
data = open(path, 'rb')
while True:
hdr = data.read(16)
if len(hdr) < 16: break
lo, sz = struct.unpack('<QQ', hdr)
body = data.read(sz)
if len(body) < sz: break
regs.append((lo, body))
return regs
def readabs(regs, addr, n):
for lo, body in regs:
if lo <= addr < lo + len(body) and addr - lo + n <= len(body):
return body[addr - lo:addr - lo + n]
return None
import numpy as np
def find_levelpair_buffers(regs):
"""Buffers where f32[2j]==j/1024 exactly for j=0..1023 (level axis)."""
expect = (np.arange(1024, dtype=np.float64) / 1024).astype(np.float32)
anchors = np.nonzero(expect == np.float32(1))[0] # sanity of construction
found = []
for lo, body in regs:
if len(body) < 8192: continue
a = np.frombuffer(body[:len(body) // 4 * 4], dtype='<f4')
# anchor: level[1] == 1/1024 at slot 2
cand = np.nonzero(a == np.float32(1.0 / 1024.0))[0]
for i in cand:
i = int(i)
if i < 2 or i % 2: continue
s = i - 2 # slot of level[0]
if s + 2048 > len(a): continue
if np.array_equal(a[s:s + 2048:2], expect):
found.append(lo + 4 * s)
found = sorted(set(found))
dedup = []
for h in found:
if not dedup or h - dedup[-1] > 0x100:
dedup.append(h)
return dedup
def find_object_base(regs, bufs):
"""u64 refs to band buffers sit at base+0xe0+band*0x18."""
votes = collections.Counter()
detail = []
for k, buf in enumerate(bufs):
pat = struct.pack('<Q', buf)
for lo, body in regs:
j = 0
while True:
j = body.find(pat, j)
if j < 0: break
addr = lo + j
base = addr - 0xe0 - k * 0x18
votes[base] += 1
detail.append((k, addr, base))
j += 1
if not votes:
return None, detail
base, cnt = votes.most_common(1)[0]
return (base, cnt) if cnt >= 2 else (None, detail)
def looks_like_bandconfig(regs, p):
b = readabs(regs, p, 0x98)
if not b: return False
A, B = struct.unpack_from('<ff', b, 0)
g, = struct.unpack_from('<f', b, 0xc)
if not (-96.0 <= A <= 96.0): return False
if not (1.0 <= B <= 100000.0): return False
if not (0.05 <= abs(g) <= 8.0): return False
return True
def dump_bandconfigs(regs, base):
def f32(addr):
v = readabs(regs, addr, 4)
return round(struct.unpack('<f', v)[0], 6) if v else None
def u64(addr):
v = readabs(regs, addr, 8)
return struct.unpack('<Q', v)[0] if v else None
out = {}
for off in range(0x80, 0x400, 8):
p = u64(base + off)
if not p or p < 0x10000: continue
if p & 7 or not looks_like_bandconfig(regs, p): continue
out['+0x%x' % off] = {
'ptr': '0x%x' % p,
'A': f32(p), 'B': f32(p + 4),
'f08': f32(p + 8), 'gamma': f32(p + 0xc),
'flag': (readabs(regs, p + 0x10, 1) or b'\xff')[0],
'raw_f32': [f32(p + x) for x in range(0x14, 0x30, 4)],
'cb': ('set' if u64(p + 0x90) else 'none'),
}
masks = {}
for band in range(6):
b = readabs(regs, base + 0x4198 + band * 0x2000, 32)
if b:
masks['band%d' % band] = [round(v, 4) for v in struct.unpack('<4d', b)]
return out, masks
def main():
args = sys.argv[1:]
rpp = args[0]
offline = '--offline' in args
jpath = None
spath = '/tmp/opencode/step7_snap.bin'
if '--json' in args: jpath = args[args.index('--json') + 1]
if '--snap' in args: spath = args[args.index('--snap') + 1]
name = os.path.splitext(os.path.basename(rpp))[0]
if not jpath:
jpath = f'/tmp/opencode/step7_{name}.json'
os.makedirs(os.path.dirname(jpath), exist_ok=True)
subprocess.run('pkill -9 -x reaper 2>/dev/null; pkill -9 -f "[y]abridge" 2>/dev/null; sleep 1', shell=True)
if offline:
cmd = ['/usr/bin/reaper', '-nosplash', '-ignoreerrors', '-renderproject', rpp]
else:
cmd = ['/usr/bin/reaper', '-nosplash', '-ignoreerrors', rpp, PLAY_LUA]
t0 = time.time()
proc = subprocess.Popen(cmd, stdout=open('/dev/null', 'w'), stderr=subprocess.STDOUT)
host = None
while time.time() - t0 < 60 and not host:
host = find_host(); time.sleep(0.2)
if not host:
print('NO HOST'); return 1
print('host %d at %.1fs' % (host, time.time() - t0))
time.sleep(8)
for attempt in range(3):
try:
nreg, nbytes = snapshot(host, spath)
except Exception as e:
print('snapshot failed:', e); break
print('snapshot #%d: %d regs %.1f MB' % (attempt, nreg, nbytes / 1e6))
if nbytes > 50e6: break
time.sleep(3)
if proc.poll() is None: proc.kill()
regs = parse_snap(spath)
bufs = find_levelpair_buffers(regs)
print('level-pair buffers:', ['0x%x' % b for b in bufs])
res = {'rpp': rpp, 'buffers': ['0x%x' % b for b in bufs]}
if not bufs:
json.dump(res, open(jpath, 'w'), indent=1); print('saved', jpath); return 2
# group into bands: cluster addrs with stride ~0x2000
groups = [[bufs[0]]]
for b in bufs[1:]:
if b - groups[-1][-1] <= 0x3000: groups[-1].append(b)
else: groups.append([b])
best = None
for g in groups:
base, cnt = find_object_base(regs, g)
print('group n=%d -> base %s (votes=%s)' % (len(g), ('0x%x' % base) if base else None, cnt if isinstance(cnt, int) else '-'))
if base and (best is None or cnt > best[1]):
best = (base, cnt)
if not best:
json.dump(res, open(jpath, 'w'), indent=1); print('saved', jpath); return 3
base = best[0]
cfgs, masks = dump_bandconfigs(regs, base)
res.update(levelpath_base='0x%x' % base, configs=cfgs, masks_head=masks)
json.dump(res, open(jpath, 'w'), indent=1)
print(json.dumps(cfgs, indent=1))
print('masks head:', json.dumps(masks))
print('saved', jpath)
return 0
if __name__ == '__main__':
sys.exit(main())