Audio Wave: fixed 721-frame subsegment grid + per-segment select/crop + notes box auto-fill
Pivot to the user's model: segments are a fixed grid of subsegment_frames frames (default 721 @ 24fps = one LTX clip), not arbitrary clicks. New inputs: subsegment_frames (grid size) and segment_select (0=all, N=output ONLY chunk N — crops waveform_image, AUDIO, and summary so you can generate/skip one beat at a time). _render gains window-crop + frame markers + per-segment time/frame labels. JS rewritten: draws the fixed grid, auto-fills the notes box with one segN: line per chunk (type or dblclick to note), click-to-seek, playhead + time readout. Workflows updated for the new widgets. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+30
-13
@@ -130,27 +130,44 @@ def _segments(rms_n, times, duration, fps, max_segments, beats):
|
||||
return segs, bounds
|
||||
|
||||
|
||||
def _render(rms_n, times, duration, beats, bounds, segs):
|
||||
"""Render the envelope + beats + segment boundaries to a ComfyUI IMAGE tensor."""
|
||||
def _render(rms_n, times, duration, beats, bounds, segs,
|
||||
fps=None, frames_marker=0, window=None):
|
||||
"""Render the envelope + beats + segment boundaries + time labels to a ComfyUI IMAGE.
|
||||
window=(t0,t1) crops to that time span (for a selected segment). frames_marker draws
|
||||
a green line every N frames (N/fps seconds) — LTX clip-length grid."""
|
||||
W, H = 1024, 256
|
||||
img = Image.new("RGB", (W, H), (18, 18, 22))
|
||||
d = ImageDraw.Draw(img)
|
||||
dur = max(duration, 1e-6)
|
||||
t0, t1 = window if window else (0.0, duration)
|
||||
span = max(t1 - t0, 1e-6)
|
||||
|
||||
def X(t):
|
||||
return int(max(0, min(W - 1, t / dur * (W - 1))))
|
||||
return int(max(0, min(W - 1, (t - t0) / span * (W - 1))))
|
||||
|
||||
pts = [(0, H)] + [(X(times[i]), H - int(rms_n[i] * (H - 26))) for i in range(len(rms_n))] + [(W - 1, H)]
|
||||
d.polygon(pts, fill=(60, 140, 220))
|
||||
pts = [(0, H)] + [(X(times[i]), H - int(rms_n[i] * (H - 30)))
|
||||
for i in range(len(rms_n)) if t0 <= times[i] <= t1] + [(W - 1, H)]
|
||||
if len(pts) > 2:
|
||||
d.polygon(pts, fill=(60, 140, 220))
|
||||
if fps and frames_marker: # frame grid (green)
|
||||
step = frames_marker / fps
|
||||
k = 1
|
||||
while k * step < duration + 1e-6:
|
||||
mt = k * step
|
||||
if t0 <= mt <= t1:
|
||||
d.line([(X(mt), 0), (X(mt), H)], fill=(70, 200, 120), width=1)
|
||||
d.text((X(mt) + 2, H - 13), f"{frames_marker * k}f", fill=(70, 200, 120))
|
||||
k += 1
|
||||
for b in beats: # beat markers (orange)
|
||||
d.line([(X(b), 0), (X(b), H)], fill=(230, 110, 60), width=1)
|
||||
for bd in bounds: # segment boundaries (white)
|
||||
d.line([(X(bd), 0), (X(bd), H)], fill=(240, 240, 240), width=1)
|
||||
for s in segs:
|
||||
x = X(s["start_s"]) + 4
|
||||
d.text((x, 4), f"S{s['segment']} {s['energy']}", fill=(255, 255, 255))
|
||||
if t0 <= b <= t1:
|
||||
d.line([(X(b), 0), (X(b), H)], fill=(230, 110, 60), width=1)
|
||||
for s in segs: # boundaries + label + time + note
|
||||
if not (t0 <= s["start_s"] <= t1):
|
||||
continue
|
||||
x = X(s["start_s"])
|
||||
d.line([(x, 0), (x, H)], fill=(240, 240, 240), width=1)
|
||||
d.text((x + 4, 4), f"S{s['segment']} {s['energy']} {s['start_s']}s/{s['frames']}f", fill=(255, 255, 255))
|
||||
if s.get("note"): # per-segment note (amber)
|
||||
d.text((x, 18), s["note"][:30], fill=(255, 210, 110))
|
||||
d.text((x + 4, 18), s["note"][:30], fill=(255, 210, 110))
|
||||
arr = np.asarray(img, dtype=np.float32) / 255.0
|
||||
return torch.from_numpy(arr)[None, ...] # [1, H, W, 3]
|
||||
|
||||
|
||||
@@ -88,6 +88,33 @@ def _audio_files():
|
||||
return files or [_NO_AUDIO]
|
||||
|
||||
|
||||
def _segments_fixed(rms_n, times, duration, fps, sub_frames, beats):
|
||||
"""Split into fixed chunks of `sub_frames` frames (the LTX clip length). The last
|
||||
chunk gets whatever's left (snapped to 8n+1)."""
|
||||
step = max(sub_frames, 1) / max(fps, 1) # seconds per chunk
|
||||
n = max(1, int(np.ceil(duration / step - 1e-6)))
|
||||
stages = ["establish", "build", "peak", "settle"]
|
||||
segs, bounds = [], [0.0]
|
||||
for i in range(n):
|
||||
t0 = i * step
|
||||
t1 = min(duration, (i + 1) * step)
|
||||
bounds.append(round(t1, 3))
|
||||
mask = (times >= t0) & (times < t1)
|
||||
e = float(rms_n[mask].mean()) if mask.any() else 0.0
|
||||
peak_t = float(times[mask][np.argmax(rms_n[mask])]) if mask.any() else t0
|
||||
dur = round(t1 - t0, 2)
|
||||
frames = sub_frames if i < n - 1 else _snap8(round(dur * fps))
|
||||
label = "high" if e > 0.66 else ("medium" if e > 0.33 else "low")
|
||||
stage = stages[i] if i < len(stages) else ("peak" if e > 0.6 else "settle")
|
||||
segs.append({
|
||||
"segment": i + 1, "start_s": round(t0, 2), "duration_s": dur, "frames": frames,
|
||||
"energy": label, "energy_val": round(e, 3), "peak_s": round(peak_t, 2),
|
||||
"stage_hint": stage, "note": "",
|
||||
"beats_in": [round(b, 2) for b in beats if t0 <= b < t1],
|
||||
})
|
||||
return segs, np.array(bounds)
|
||||
|
||||
|
||||
class AudioWaveSegments:
|
||||
CATEGORY = "prompt_calibrator"
|
||||
FUNCTION = "run"
|
||||
@@ -101,25 +128,29 @@ class AudioWaveSegments:
|
||||
# Pick a file from ComfyUI/input, or use the widget's "upload" button (JS).
|
||||
"audio": (_audio_files(),),
|
||||
"fps": ("INT", {"default": 24, "min": 1, "max": 120}),
|
||||
"max_segments": ("INT", {"default": 6, "min": 3, "max": 12}),
|
||||
# Fixed LTX clip length. Segments = chunks of this many frames (721@24fps
|
||||
# ~= 30.04s). This is also the waveform grid. Set 0 to use segments_json.
|
||||
"subsegment_frames": ("INT", {"default": 721, "min": 0, "max": 100000}),
|
||||
# 0 = whole clip. N = output ONLY subsegment N (crops image + audio + summary),
|
||||
# so you can generate/skip one beat at a time.
|
||||
"segment_select": ("INT", {"default": 0, "min": 0, "max": 999}),
|
||||
"notes": ("STRING", {"default": "", "multiline": True}),
|
||||
# Written by the JS waveform widget: [{"start_s": 0.0, "note": "..."}, ...].
|
||||
# Leave as "[]" to auto-split. Also editable by hand.
|
||||
# Machine field: per-chunk notes/boundaries from the JS widget (hidden).
|
||||
"segments_json": ("STRING", {"default": "[]"}),
|
||||
},
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def IS_CHANGED(cls, audio, fps, max_segments, notes, segments_json):
|
||||
def IS_CHANGED(cls, audio, fps, subsegment_frames, segment_select, notes, segments_json):
|
||||
try:
|
||||
import folder_paths
|
||||
p = folder_paths.get_annotated_filepath(audio)
|
||||
mt = os.path.getmtime(p) if os.path.isfile(p) else ""
|
||||
except Exception:
|
||||
mt = ""
|
||||
return f"{audio}|{fps}|{max_segments}|{notes}|{segments_json}|{mt}"
|
||||
return f"{audio}|{fps}|{subsegment_frames}|{segment_select}|{notes}|{segments_json}|{mt}"
|
||||
|
||||
def run(self, audio, fps, max_segments, notes, segments_json):
|
||||
def run(self, audio, fps, subsegment_frames, segment_select, notes, segments_json):
|
||||
try:
|
||||
import folder_paths
|
||||
path = folder_paths.get_annotated_filepath(audio)
|
||||
@@ -135,28 +166,44 @@ class AudioWaveSegments:
|
||||
rms_n, times = _rms_envelope(y, sr)
|
||||
bpm, beats = _tempo_beats(y, sr)
|
||||
|
||||
# User boundaries from the JS widget, else auto-split.
|
||||
starts, seg_notes = [], {}
|
||||
try:
|
||||
data = json.loads(segments_json) if segments_json.strip() else []
|
||||
for i, seg in enumerate(data):
|
||||
starts.append(float(seg.get("start_s", 0.0)))
|
||||
seg_notes[i + 1] = str(seg.get("note", "") or "")
|
||||
except Exception as e:
|
||||
print(f"[AudioWaveSegments] bad segments_json ({e}); auto-splitting.")
|
||||
data = []
|
||||
|
||||
if data:
|
||||
if subsegment_frames > 0: # fixed LTX-clip grid
|
||||
segs, bounds = _segments_fixed(rms_n, times, duration, fps, subsegment_frames, beats)
|
||||
else: # manual boundaries from the JS widget
|
||||
starts = []
|
||||
try:
|
||||
for seg in (json.loads(segments_json) if segments_json.strip() else []):
|
||||
starts.append(float(seg.get("start_s", 0.0)))
|
||||
except Exception as e:
|
||||
print(f"[AudioWaveSegments] bad segments_json ({e})")
|
||||
segs, bounds = _segments_from_boundaries(rms_n, times, duration, fps, starts, beats)
|
||||
for s in segs: # notes placed on the waveform
|
||||
s["note"] = seg_notes.get(s["segment"], "")
|
||||
else:
|
||||
segs, bounds = _segments(rms_n, times, duration, fps, max_segments, beats)
|
||||
|
||||
global_notes = _attach_notes(segs, notes) # merge the text-box notes on top
|
||||
image = _render(rms_n, times, duration, beats, bounds, segs)
|
||||
summary = _summary(duration, sr, bpm, beats, segs, global_notes)
|
||||
audio_out = {"waveform": wav.unsqueeze(0), "sample_rate": sr} # [1, C, N]
|
||||
# per-chunk notes from segments_json (JS) then the notes-box (segN: syntax) on top.
|
||||
try:
|
||||
for i, seg in enumerate(json.loads(segments_json) if segments_json.strip() else []):
|
||||
if i < len(segs) and seg.get("note"):
|
||||
segs[i]["note"] = str(seg["note"])
|
||||
except Exception:
|
||||
pass
|
||||
global_notes = _attach_notes(segs, notes)
|
||||
|
||||
window, render_segs = None, segs
|
||||
audio_out = {"waveform": wav.unsqueeze(0), "sample_rate": sr}
|
||||
sel = int(segment_select)
|
||||
if 1 <= sel <= len(segs): # crop to one subsegment
|
||||
seg = segs[sel - 1]
|
||||
s0 = seg["start_s"]
|
||||
s1 = min(duration, s0 + seg["duration_s"])
|
||||
a, b = int(s0 * sr), int(s1 * sr)
|
||||
if b > a:
|
||||
audio_out = {"waveform": wav[:, a:b].unsqueeze(0), "sample_rate": sr}
|
||||
window, render_segs = (s0, s1), [seg]
|
||||
summary = (f"SELECTED SUBSEGMENT {sel} of {len(segs)} — generate ONLY this beat.\n"
|
||||
+ _summary(duration, sr, bpm, beats, [seg], global_notes))
|
||||
else:
|
||||
summary = _summary(duration, sr, bpm, beats, segs, global_notes)
|
||||
|
||||
image = _render(rms_n, times, duration, beats, bounds, render_segs,
|
||||
fps=fps, frames_marker=subsegment_frames, window=window)
|
||||
return (image, summary, audio_out)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user