nodes/malkuthro-ComfyUI-Koolook
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(nodes): remove superseded LTX Director extra nodes + harden registry (#198) (#199)

Browse Source 
* refactor(nodes): remove superseded LTX Director extra nodes + harden registry (#198)

Remove the two experimental nodes layered on the WhatDreamsCost fork — KoolookAudioTranscriptTimeline (k_audio_timeline.py) and KoolookTimelineEditor (k_timeline_editor.py) — plus the transcribe_audio_timeline.py CLI and its test. The audio-transcript to Director-timing job now lives inside the fork via the Director's audio_transcript_json input, so the standalone nodes are redundant. Also drops the orphaned [audio]/faster-whisper extra and updates packaging, dev-sync, web, and docs references.

Harden the node registry against the #183 failure: __init__.py now imports each node group independently and installs the /koolook/presets/* routes regardless of node-import results, so one missing or broken module degrades to 'that node absent' instead of zeroing every node and 404'ing the sidebar disk routes. A new test keeps every root k_*.py listed in dev-sync's RUNTIME_PATHS.

Closes #198.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(nodes): ship loader gates via dev-sync-audio; generalize RUNTIME_PATHS guard (#198)

Address the K-Flame review on #199:

MEDIUM — the #198 refactor made koolook_versioning.py a single registration gate, but dev-sync-audio shipped the new __init__.py without it (and the k_*.py-glob guard test couldn't see it). Add koolook_versioning.py AND koolook_install_guard.py (whose absolute-import fallback raises uncaught) to AUDIO_PATHS so a scoped sync can never leave the loader pointing at a module the target lacks.

Generalize the guard test: parse __init__.py's actual load-time import targets (relative imports + _merge_node_group calls) and assert each is in RUNTIME_PATHS, instead of globbing k_*.py which missed the non-k_ gates. Add an AUDIO_PATHS gate-coverage test.

LOW — CHANGELOG now records the [audio]/faster-whisper + transcribe_audio_timeline.py removal as an external-contract break.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malkuthro
2026-06-02 19:47:25 +02:00
committed by GitHub
co-authored by Claude Opus 4.8
parent 8058e17456
commit 447465ecd8
13 changed files with 223 additions and 1541 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+20 -20
View File
@@ -14,7 +14,23 @@ The format is inspired by Keep a Changelog and SemVer.
(Koolook)` for a second workflow stage without parsing VHS's mixed
JSON/video list.
### Removed
- **Standalone audio-transcript tooling.** Removed the experimental
`KoolookAudioTranscriptTimeline` / `KoolookTimelineEditor` nodes, the
`scripts/transcribe_audio_timeline.py` CLI, and the `[audio]`
(`faster-whisper`) optional dependency — superseded by the Koolook
Director's built-in `audio_transcript_json` input (#198). Anyone using
`pip install -e '.[audio]'` or that CLI out-of-tree should pin a
pre-removal commit.
### Fixed
- **Node registry survives one broken or missing module.** The root
`__init__.py` now imports each node group independently and installs the
Kforge Labs snapshot/preset routes (`/koolook/presets/*`) regardless of
node-import results. Previously a single missing or failing node module
raised at import time and zeroed *every* Koolook node **and** the snapshot
routes at once (sidebar disk browse/save returning 404). A new guard test
keeps every root `k_*.py` listed in dev-sync's `RUNTIME_PATHS`.
- **`loop-audio`: Director card reads execution-active values.** The
audio-lipsync card now reads Director widgets from the saved Director
widget order before falling back to generic input-name lookup, fixing
@@ -39,14 +55,6 @@ The format is inspired by Keep a Changelog and SemVer.
audio sync now removes the old `web/whatdreamscost_koolook_v1_3_2/`
folder after the stable v1.3.9 web-extension rename, preventing legacy
workflows from loading two identical timeline editors in dev installs.
- **Audio Transcript Timeline now follows every timeline audio clip.** When
the node receives `timeline_data` from Koolook Timeline Editor it now
transcribes all `audioSegments`, applies each clip's timeline start,
trim start, and visible length, then merges the phrases into one ordered
Prompt Relay timing sequence. This fixes separated audio clips where only
the first file affected `local_prompts` / `segment_lengths`. Generated
speech/pause prompts now also include the active image segment's prompt,
with a first-non-empty timeline prompt fallback for empty image segments.
- **LTX Director timeline images persist after canvas/setup switches.** The
timeline editor keeps saved workflow payloads light by omitting preview-only
image blobs, but now rebuilds image previews from the persisted `imageFile`
@@ -224,18 +232,10 @@ The format is inspired by Keep a Changelog and SemVer.
*Loop* and *Run* entries are rewritten to point at the new structure.
### Added
- **Koolook Audio Transcript Timeline node for custom-audio sync tests.**
New optional `.[audio]` dependency group installs `faster-whisper`, and
the new `Koolook Audio Transcript Timeline` Comfy node converts a speech
file into Director-shaped `timeline_data`, `local_prompts`, and
`segment_lengths` with explicit closed-mouth pause segments. The
Koolook Director now also accepts the node's `transcript_json` directly
via `audio_transcript_json` and builds those Director fields internally
before Prompt Relay conditioning runs. The companion
`scripts/transcribe_audio_timeline.py` produces the same data from the
command line. This gives the audio-lipsync loop a repeatable way to test
whether LTX needs semantic speech timing in addition to the raw
custom-audio latent.
- **Koolook Director accepts an audio transcript for speech timing.** The
Koolook LTX Director gained an `audio_transcript_json` input that converts
phrase-timed transcript JSON into Director `timeline_data`, `local_prompts`,
and `segment_lengths` internally before Prompt Relay conditioning runs.
- **`forks/whatdreamscost_koolook/v1_3_9/` — updated Koolook LTX Director
fork.** Upgrades the Koolook Director to upstream WhatDreamsCost-ComfyUI `3b65410`
(pyproject/README version `1.3.9`; no `v1.3.9` git tag at fork time).
__init__.py
+82 -110
View File
@@ -1,3 +1,21 @@
"""ComfyUI-Koolook node registry.
Two layers of resilience guard the import:
1. Duplicate-install guard (#162) — if two Koolook folders exist in
``custom_nodes/``, only the alphabetical winner registers; the loser
short-circuits before any heavy imports or route install.
2. Per-group registration (#198 / #183) — each node group is imported
independently so a single missing or broken module degrades to "that
node absent" instead of taking down the whole pack, and the Kforge Labs
snapshot/preset routes install regardless of node-import results. The
failure this guards against: a module that ``__init__`` imported but
``dev-sync`` never copied raised ``ImportError`` at startup, which zeroed
every node *and* 404'd every snapshot route.
"""
from __future__ import annotations
import importlib
from pathlib import Path
# Duplicate-install guard (#162). Detection + winner selection live in the
@@ -23,6 +41,29 @@ except ImportError: # pragma: no cover - exercised only outside package context
detect_duplicate_koolook_installs,
)
NODE_CLASS_MAPPINGS: dict = {}
NODE_DISPLAY_NAME_MAPPINGS: dict = {}
def _merge_node_group(label: str, module_name: str) -> None:
"""Import one node group's mappings, isolating any import failure.
A missing module, a broken import inside it, or an absent
``NODE_CLASS_MAPPINGS`` / ``NODE_DISPLAY_NAME_MAPPINGS`` attribute is
caught and logged; the rest of the registry (and the route install
below) is unaffected.
"""
try:
module = importlib.import_module(module_name, __package__)
class_mappings = module.NODE_CLASS_MAPPINGS
display_mappings = module.NODE_DISPLAY_NAME_MAPPINGS
except (ImportError, AttributeError) as exc:
print(f"[Koolook] {label} registry skipped: {exc}")
return
NODE_CLASS_MAPPINGS.update(class_mappings)
NODE_DISPLAY_NAME_MAPPINGS.update(display_mappings)
_here = Path(__file__).resolve().parent
# The guard must never take down the plugin: any failure to scan or parse
# siblings degrades to "register normally" (fail-safe) instead of aborting
@@ -44,91 +85,51 @@ except Exception as _guard_err: # pragma: no cover - defensive backstop
if not _is_winning_install:
# Non-winning duplicate (#162). Short-circuit BEFORE any of the
# heavy node imports (torch, PIL, diffusers via the radiance/
# whatdreamscost forks) or the aiohttp route install — the winner
# owns those for this session and there's no point paying their
# import cost just to discard the mappings afterward. Critical log
# was already printed by ``build_duplicate_report`` above.
NODE_CLASS_MAPPINGS = {}
NODE_DISPLAY_NAME_MAPPINGS = {}
# Non-winning duplicate (#162). Short-circuit BEFORE any of the heavy
# node imports (torch, PIL, diffusers via the radiance/whatdreamscost
# forks) or the aiohttp route install — the winner owns those for this
# session. No WEB_DIRECTORY either, so the loser doesn't double-register
# the sidebar web assets. Critical log already printed above.
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
else:
# Probe for a real package context. ComfyUI's custom-node loader provides
# one; pytest's package collector and static-analysis tools import this
# file as a standalone module, where relative imports raise. In that case
# we register nothing (a single message) and let tooling import the
# package directory without crashing. ``koolook_versioning`` is a
# pure-stdlib helper, so the probe can only fail on the context, never on
# its body.
try:
from .k_easy_wan22_prompt import NODE_CLASS_MAPPINGS as wan_mappings
from .k_easy_wan22_prompt import NODE_DISPLAY_NAME_MAPPINGS as wan_display
from .k_easy_resize import NODE_CLASS_MAPPINGS as resize_mappings
from .k_easy_resize import NODE_DISPLAY_NAME_MAPPINGS as resize_display
from .k_ai_pipeline import NODE_CLASS_MAPPINGS as pipeline_mappings
from .k_ai_pipeline import NODE_DISPLAY_NAME_MAPPINGS as pipeline_display
from .k_easy_image_batch import NODE_CLASS_MAPPINGS as batch_mappings # Added for new VFX batch node
from .k_easy_image_batch import NODE_DISPLAY_NAME_MAPPINGS as batch_display # Added for new VFX batch node
from .k_easy_track import NODE_CLASS_MAPPINGS as cam_loader_mappings
from .k_easy_track import NODE_DISPLAY_NAME_MAPPINGS as cam_loader_display
from .k_easy_pattern import NODE_CLASS_MAPPINGS as pattern_mappings
from .k_easy_pattern import NODE_DISPLAY_NAME_MAPPINGS as pattern_display
from .k_easy_utility import NODE_CLASS_MAPPINGS as utility_mappings
from .k_easy_utility import NODE_DISPLAY_NAME_MAPPINGS as utility_display
from .k_audio_timeline import NODE_CLASS_MAPPINGS as audio_timeline_mappings
from .k_audio_timeline import NODE_DISPLAY_NAME_MAPPINGS as audio_timeline_display
from .k_timeline_editor import NODE_CLASS_MAPPINGS as timeline_editor_mappings
from .k_timeline_editor import NODE_DISPLAY_NAME_MAPPINGS as timeline_editor_display
from .forks.radiance_koolook import NODE_CLASS_MAPPINGS as radiance_koolook_mappings
from .forks.radiance_koolook import NODE_DISPLAY_NAME_MAPPINGS as radiance_koolook_display
from .forks.whatdreamscost_koolook import NODE_CLASS_MAPPINGS as whatdreamscost_koolook_mappings
from .forks.whatdreamscost_koolook import NODE_DISPLAY_NAME_MAPPINGS as whatdreamscost_koolook_display
except ImportError as _node_import_exc:
# The relative imports above require a parent-package context that
# only ComfyUI's custom-node loader provides. Pytest's package
# collector (which walks up from test files looking for ``__init__.py``)
# and static-analysis tools load this file as a standalone module,
# where the imports raise. Falling back to empty registries lets
# tooling import the package directory without crashing — tests
# access sibling modules (e.g. ``koolook_routes``) via absolute
# imports that don't need the package context.
from . import koolook_versioning as _context_probe # noqa: F401
except ImportError as _context_exc:
print(
f"[Koolook] node registry skipped (non-Comfy import context): "
f"{_node_import_exc}"
f"{_context_exc}"
)
NODE_CLASS_MAPPINGS = {}
NODE_DISPLAY_NAME_MAPPINGS = {}
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
else:
# VHS-dependent wrappers self-skip when VideoHelperSuite is absent, and
# these guards keep a local import surprise from hiding the core nodes.
video_load_mappings = {}
video_load_display = {}
try:
from .k_video_load import NODE_CLASS_MAPPINGS as video_load_mappings
from .k_video_load import NODE_DISPLAY_NAME_MAPPINGS as video_load_display
except ImportError as _exc:
print(f"[Koolook] Easy_LoadVideo registry skipped: {_exc}")
# Order is cosmetic (menu grouping); each group is independent.
_merge_node_group("Wan 2.2 Easy Prompt", ".k_easy_wan22_prompt")
_merge_node_group("Easy Resize", ".k_easy_resize")
_merge_node_group("Easy AI Pipeline", ".k_ai_pipeline")
_merge_node_group("Easy Image Batch", ".k_easy_image_batch")
_merge_node_group("Koolook Camera Loader", ".k_easy_track")
_merge_node_group("Easy Pattern", ".k_easy_pattern")
_merge_node_group("Easy Utility", ".k_easy_utility")
# The VHS-dependent video wrappers degrade to empty mappings when
# VideoHelperSuite is absent (they catch the import internally); the
# guard also covers a hard ImportError should one ever slip through.
_merge_node_group("Easy Load Video", ".k_video_load")
_merge_node_group("Easy Video Combine", ".k_video_combine")
_merge_node_group("Radiance Koolook VAE", ".forks.radiance_koolook")
_merge_node_group("WhatDreamsCost LTX Director", ".forks.whatdreamscost_koolook")
video_combine_mappings = {}
video_combine_display = {}
try:
from .k_video_combine import NODE_CLASS_MAPPINGS as video_combine_mappings
from .k_video_combine import NODE_DISPLAY_NAME_MAPPINGS as video_combine_display
except ImportError as _exc:
print(f"[Koolook] Easy_VideoCombine registry skipped: {_exc}")
# Register the Kforge Labs snapshot/preset endpoints. Failure here is
# non-fatal — the node mappings still load and the rest of the plugin
# works; only the snapshot feature in the sidebar is unavailable. Both
# the import itself AND the `install()` call are guarded: a missing
# `aiohttp` (extremely unusual since it's a ComfyUI hard-dep) or any
# other module-load exception falls through to the print without
# crashing the node registry.
# Register the Kforge Labs snapshot/preset endpoints. Installed
# regardless of the node-import results above, so a broken node
# module can no longer take the /koolook/presets/* routes down with
# it. Both the import and the install() call are guarded: a missing
# aiohttp (a ComfyUI hard-dep) or any other load error degrades to
# "snapshot feature unavailable", never a crash.
try:
from . import koolook_routes # noqa: E402
@@ -137,40 +138,11 @@ else:
"[Koolook] PromptServer unavailable at import time; preset routes "
"(/koolook/presets/*) not registered for this session."
)
except Exception as _exc: # pragma: no cover
print(f"[Koolook] preset routes import failed; node mappings unaffected: {_exc}")
NODE_CLASS_MAPPINGS = {
**wan_mappings,
**resize_mappings,
**pipeline_mappings,
**batch_mappings,
**cam_loader_mappings,
**pattern_mappings,
**utility_mappings,
**audio_timeline_mappings,
**timeline_editor_mappings,
**video_load_mappings,
**video_combine_mappings,
**radiance_koolook_mappings,
**whatdreamscost_koolook_mappings,
}
NODE_DISPLAY_NAME_MAPPINGS = {
**wan_display,
**resize_display,
**pipeline_display,
**batch_display,
**cam_loader_display,
**pattern_display,
**utility_display,
**audio_timeline_display,
**timeline_editor_display,
**video_load_display,
**video_combine_display,
**radiance_koolook_display,
**whatdreamscost_koolook_display,
}
except Exception as _routes_exc: # pragma: no cover
print(
f"[Koolook] preset routes import failed; node mappings unaffected: "
f"{_routes_exc}"
)
WEB_DIRECTORY = "./web"
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
docs/automations/LTX-2.3/audio-lipsync/CHEATSHEET.md
+7 -50
View File
@@ -19,7 +19,6 @@ look as clean as baseline.
| Phrase | Script | What it does |
|---|---|---|
| `loop-audio` | [`scripts/loop_audio.py`](../../../../scripts/loop_audio.py) | Snapshots the most recent render into `runs/run-NNN_<label>/` (redacted runNNN_workflow.json + relay_overrides.txt + patch_state.txt + metadata.json + notes.md + card.png), delivers the external card copy, and appends a row to [`runs/log.md`](runs/log.md). |
| `transcribe-audio` | [`scripts/transcribe_audio_timeline.py`](../../../../scripts/transcribe_audio_timeline.py) or the `Koolook Audio Transcript Timeline` Comfy node | Uses the optional Whisper helper to turn the current speech file into timed Director prompts (`timeline_data`, `local_prompts`, `segment_lengths`). Install with `.[audio]` first. |
| `dev-sync-audio` | [`scripts/sync_to_dev_audio.py`](../../../../scripts/sync_to_dev_audio.py) | Copies `forks/whatdreamscost_koolook/`, `web/whatdreamscost_koolook/`, and root `__init__.py` into `$KOLOOK_COMFYUI_DEV_PATH`; removes stale pre-v1.3.9 web extension folder. Restart ComfyUI manually after Python changes need to be re-imported. Widget-only changes on the canvas don't need it. |
Config lives in `<script>.config.json` next to each script — edit the
@@ -168,56 +167,14 @@ runs**:
3. Remove the corresponding hypothesis from [`README.md`](README.md) or
[`backstory/audio-lipsync-rationale.md`](backstory/audio-lipsync-rationale.md).
## Timed transcript experiment
## Timed transcript experiment (removed)
When raw custom audio is not enough to drive mouth timing, generate timed
speech prompts from the audio and feed those back into the Director:
**Inside ComfyUI:** add `Koolook Audio Transcript Timeline`, set the same
`audio_file`, `image_file`, duration, and FPS as the Director, then link:
- `transcript_json` -> Director `audio_transcript_json`
When its `timeline_data` input is linked from `Koolook Timeline Editor`,
the node uses all audio clips on the timeline. Each clip is transcribed
separately, shifted by its timeline start, clipped by its trim/length, and
then merged into one ordered prompt-timing sequence. This is the preferred
ComfyUI path for testing separated dialogue clips. The generated speech
and pause prompts include the active image segment prompt, so multi-image
timelines can keep visual directions on the image clips while the transcript
node adds exact mouth timing.
Keep `use_custom_audio=True`. The node emits `transcript_json` so the
recognized phrase timing can also be inspected with a text-preview node.
The Koolook Director converts that JSON into its own `timeline_data`,
`local_prompts`, and `segment_lengths` immediately before Prompt Relay
conditioning runs.
The older manual hook still works for debugging:
- `timeline_data` -> Director `timeline_data`
- `local_prompts` -> Director `local_prompts`
- `segment_lengths` -> Director `segment_lengths`
**Script/export path:**
```powershell
.\.venv\Scripts\python -m pip install -e ".[audio]"
.\.venv\Scripts\python scripts\transcribe_audio_timeline.py <audio.mp3> --workflow <workflow.json> --out timed-prompts.json --patched-workflow timed-workflow.json
```
The output JSON contains:
- `phrases` - timestamped transcript chunks for review.
- `timeline_data` - Director timeline JSON with one prompt segment per
phrase, preserving the source image and audio segments from the workflow.
- `local_prompts` and `segment_lengths` - the matching Director fields.
- `--patched-workflow` - optional loadable workflow JSON with those fields
already written onto the Koolook Director node.
Paste the generated prompt/timeline fields into the Director, keep
`use_custom_audio=True`, and render. This tests whether the model needs
semantic speech timing in addition to the raw audio latent.
The standalone `Koolook Audio Transcript Timeline` node and the
`scripts/transcribe_audio_timeline.py` Whisper helper were removed in #198.
The capability was experimental and superseded by the Koolook Director's
built-in `audio_transcript_json` input — supply it a transcript JSON from
your own tooling if you still want semantic speech timing on top of the raw
audio latent. Nothing in the bundled loop generates that JSON anymore.
## Where settings live (no code edits for the common cases)
docs/automations/LTX-2.3/audio-lipsync/README.md
+7 -38
View File
@@ -120,45 +120,14 @@ in `metadata.json` and the PNG `koolook_audio_loop` text metadata.
See [`runs/LOOP.md`](runs/LOOP.md) for the full per-render protocol and the retention rationale.
## Timed transcript experiment
## Timed transcript experiment (removed)
Raw custom audio alone can fail to produce usable mouth timing even when
the same words improve the render when typed into the prompt. The loop now
has an optional helper for that gap.
Inside ComfyUI, add `Koolook Audio Transcript Timeline`, set the same
`audio_file`, `image_file`, duration, and FPS as the Director, then link
its `transcript_json` output into the Koolook Director's
`audio_transcript_json` input. Keep `use_custom_audio=True`; use the same
`transcript_json` output with a text preview node to inspect what Whisper
recognized. The Director converts the phrase timings into its internal
`timeline_data`, `local_prompts`, and `segment_lengths` immediately before
Prompt Relay conditioning runs.
For Timeline Editor workflows, link `Koolook Timeline Editor.timeline_data`
into `Koolook Audio Transcript Timeline.timeline_data`. The transcript node
then reads every `audioSegments` clip, including separated clips, and applies
that clip's start frame, trim start, and visible length before merging the
phrases into the final Prompt Relay timing. The node also composes each
speech/pause timing instruction with the active image segment prompt, so
visual directions stay on the timeline image clips while speech timing is
generated automatically.
The same helper can also run from a script when an export file is useful:
```powershell
.\.venv\Scripts\python -m pip install -e ".[audio]"
.\.venv\Scripts\python scripts\transcribe_audio_timeline.py <audio.mp3> --workflow <workflow.json> --out timed-prompts.json --patched-workflow timed-workflow.json
```
The helper uses `faster-whisper` to transcribe speech into timestamped
phrases, inserts closed-mouth pause segments for silence, and emits
Director-shaped `timeline_data`, `local_prompts`, and `segment_lengths`.
Those fields remain useful for debugging or patched-workflow exports. Use
`--patched-workflow` to write a loadable workflow JSON with those fields
already applied to the Koolook Director. Render it with
`use_custom_audio=True` to test whether LTX needs semantic speech timing
in addition to the raw audio latent.
The standalone `Koolook Audio Transcript Timeline` node and the
`scripts/transcribe_audio_timeline.py` Whisper helper were removed in #198 —
the capability was experimental and not worth maintaining as bundled nodes.
The Koolook Director keeps its optional `audio_transcript_json` input, so a
transcript JSON produced by external tooling can still drive semantic speech
timing; the bundled loop no longer generates that JSON.
## Workflow JSON — Koolook node ID
k_audio_timeline.py
-588
View File
@@ -1,588 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# ComfyUI-Koolook - Audio transcript timeline
# Copyright (C) 2026 ComfyUI-Koolook contributors (kforgelabs).
"""Comfy node that turns speech audio into timed LTXDirector prompts."""
from __future__ import annotations
import json
import os
import re
from dataclasses import dataclass
from pathlib import Path
try:
import folder_paths # ComfyUI core; present at runtime.
except ImportError: # pragma: no cover - tests run outside ComfyUI.
folder_paths = None # type: ignore[assignment]
DEFAULT_PROMPT_TEMPLATE = (
'[Audio timing, frames {start_frame:03d}-{end_frame:03d}, '
'{start_seconds:.2f}-{end_seconds:.2f}s]: '
'The bear says exactly "{text}". '
"The mouth, jaw, and lips form the spoken words within this frame range."
)
DEFAULT_PAUSE_TEMPLATE = (
"[Audio timing, frames {start_frame:03d}-{end_frame:03d}, "
"{start_seconds:.2f}-{end_seconds:.2f}s]: "
"Silence/reaction pause. The bear does not speak; mouth closes naturally "
"while the face stays alive and attentive."
)
@dataclass(frozen=True)
class Word:
start: float
end: float
text: str
@dataclass(frozen=True)
class Phrase:
start: float
end: float
text: str
@dataclass(frozen=True)
class TimelineAudioSegment:
audio_file: str
start_frames: int
trim_start_frames: int
length_frames: int
audio_duration_frames: int
segment_id: str = ""
file_name: str = ""
@dataclass(frozen=True)
class TimelineImageSegment:
image_file: str
start_frames: int
length_frames: int
prompt: str = ""
segment_id: str = ""
def _input_root() -> Path:
if folder_paths is not None:
return Path(folder_paths.get_input_directory())
return Path.cwd()
def _resolve_input_file(name_or_path: str) -> Path:
text = str(name_or_path or "").strip().strip('"').strip("'")
path = Path(text)
if path.is_absolute():
return path
return _input_root() / text
def _clean_word(text: str) -> str:
return re.sub(r"\s+", " ", text).strip()
def _phrase_text(words: list[Word]) -> str:
text = " ".join(w.text for w in words)
text = re.sub(r"\s+([,.!?;:])", r"\1", text)
return re.sub(r"\s+", " ", text).strip()
def _frame(value: float, fps: float) -> int:
return max(0, int(round(value * fps)))
def _timeline_frame(value: object) -> int:
return max(0, int(round(float(value or 0))))
def _timeline_defaults(timeline_data: str) -> dict:
if not str(timeline_data or "").strip():
return {}
data = json.loads(timeline_data)
if not isinstance(data, dict):
return {}
image_segments = _parse_timeline_image_segments(timeline_data)
first_image = image_segments[0] if image_segments else None
audio_segments = _parse_timeline_audio_segments(timeline_data)
first_audio = audio_segments[0] if audio_segments else None
return {
"image_file": first_image.image_file if first_image else "",
"image_segments": image_segments,
"audio_file": first_audio.audio_file if first_audio else "",
"audio_start_frames": first_audio.start_frames if first_audio else 0,
"audio_trim_start_frames": first_audio.trim_start_frames if first_audio else 0,
"audio_length_frames": first_audio.length_frames if first_audio else 0,
"audio_segments": audio_segments,
}
def _parse_timeline_image_segments(timeline_data: str) -> list[TimelineImageSegment]:
if not str(timeline_data or "").strip():
return []
data = json.loads(timeline_data)
if not isinstance(data, dict):
return []
segments: list[TimelineImageSegment] = []
for idx, raw in enumerate(data.get("segments", [])):
if not isinstance(raw, dict) or not raw.get("imageFile"):
continue
segments.append(
TimelineImageSegment(
image_file=str(raw.get("imageFile", "")),
start_frames=_timeline_frame(raw.get("start", 0)),
length_frames=max(0, _timeline_frame(raw.get("length", 0))),
prompt=str(raw.get("prompt") or "").strip(),
segment_id=str(raw.get("id") or f"image_{idx + 1:03d}"),
)
)
return sorted(segments, key=lambda item: (item.start_frames, item.segment_id))
def _parse_timeline_audio_segments(timeline_data: str) -> list[TimelineAudioSegment]:
if not str(timeline_data or "").strip():
return []
data = json.loads(timeline_data)
if not isinstance(data, dict):
return []
segments: list[TimelineAudioSegment] = []
for idx, raw in enumerate(data.get("audioSegments", [])):
if not isinstance(raw, dict) or not raw.get("audioFile"):
continue
length = _timeline_frame(raw.get("length", 0))
duration = _timeline_frame(raw.get("audioDurationFrames", 0))
if length <= 0:
length = duration
segments.append(
TimelineAudioSegment(
audio_file=str(raw.get("audioFile", "")),
start_frames=_timeline_frame(raw.get("start", 0)),
trim_start_frames=_timeline_frame(raw.get("trimStart", 0)),
length_frames=max(0, length),
audio_duration_frames=max(0, duration or length),
segment_id=str(raw.get("id") or f"audio_{idx + 1:03d}"),
file_name=str(raw.get("fileName") or os.path.basename(str(raw.get("audioFile", "")))),
)
)
return sorted(segments, key=lambda item: (item.start_frames, item.segment_id))
def _fallback_audio_segment(audio_file: str, audio_length_frames: int) -> TimelineAudioSegment:
audio_len = max(0, int(audio_length_frames or 0))
return TimelineAudioSegment(
audio_file=audio_file,
start_frames=0,
trim_start_frames=0,
length_frames=audio_len,
audio_duration_frames=audio_len,
segment_id="audio_001",
file_name=os.path.basename(audio_file),
)
def _fallback_image_segment(image_file: str, duration_frames: int) -> TimelineImageSegment:
return TimelineImageSegment(
image_file=image_file,
start_frames=0,
length_frames=max(1, int(duration_frames or 1)),
)
def shift_phrases_to_timeline(
phrases: list[Phrase],
*,
frame_rate: float,
audio_start_frames: int,
audio_trim_start_frames: int,
duration_frames: int,
audio_length_frames: int = 0,
) -> list[Phrase]:
trim_start_seconds = audio_trim_start_frames / frame_rate
start_seconds = audio_start_frames / frame_rate
trim_end_seconds = None
if audio_length_frames > 0:
trim_end_seconds = trim_start_seconds + (audio_length_frames / frame_rate)
duration_seconds = duration_frames / frame_rate
shifted: list[Phrase] = []
for phrase in phrases:
if phrase.end <= trim_start_seconds:
continue
if trim_end_seconds is not None and phrase.start >= trim_end_seconds:
continue
source_start = max(phrase.start, trim_start_seconds)
source_end = min(phrase.end, trim_end_seconds) if trim_end_seconds is not None else phrase.end
start = source_start - trim_start_seconds + start_seconds
end = source_end - trim_start_seconds + start_seconds
if end <= 0 or start >= duration_seconds:
continue
shifted.append(Phrase(max(0.0, start), min(duration_seconds, max(0.0, end)), phrase.text))
return shifted
def _format_prompt(
template: str,
*,
text: str = "",
start: int,
end: int,
fps: float,
) -> str:
return template.format(
text=text,
start_frame=start,
end_frame=max(start, end - 1),
end_exclusive_frame=end,
length_frames=max(1, end - start),
start_seconds=start / fps,
end_seconds=end / fps,
)
def _compose_prompt(scene_prompt: str, timing_prompt: str) -> str:
scene = re.sub(r"\s+", " ", scene_prompt or "").strip()
timing = re.sub(r"\s+", " ", timing_prompt or "").strip()
if scene and timing:
return f"{scene} {timing}"
return timing or scene
def _active_image_segment(
frame: int,
image_segments: list[TimelineImageSegment],
) -> TimelineImageSegment | None:
if not image_segments:
return None
active = [
segment
for segment in image_segments
if segment.start_frames <= frame < segment.start_frames + max(1, segment.length_frames)
]
if active:
return active[-1]
previous = [segment for segment in image_segments if segment.start_frames <= frame]
if previous:
return previous[-1]
return image_segments[0]
def _active_scene_prompt(
frame: int,
image_segments: list[TimelineImageSegment],
) -> str:
active = _active_image_segment(frame, image_segments)
if active and active.prompt:
return active.prompt
first_prompt = next((segment.prompt for segment in image_segments if segment.prompt), "")
return first_prompt
def transcribe_words(audio_path: Path, model_size_or_path: str, device: str) -> list[Word]:
try:
from faster_whisper import WhisperModel
except ImportError as exc:
raise RuntimeError(
"faster-whisper is not installed. Install Koolook's optional audio "
'tools with: python -m pip install -e ".[audio]"'
) from exc
compute_type = "float16" if device == "cuda" else "int8"
model = WhisperModel(model_size_or_path, device=device, compute_type=compute_type)
segments, _info = model.transcribe(
str(audio_path),
word_timestamps=True,
vad_filter=False,
)
words: list[Word] = []
for seg in segments:
for raw in seg.words or []:
text = _clean_word(raw.word)
if text:
words.append(Word(float(raw.start), float(raw.end), text))
return words
def group_words(
words: list[Word],
*,
max_phrase_seconds: float,
max_words: int,
gap_seconds: float,
) -> list[Phrase]:
phrases: list[Phrase] = []
current: list[Word] = []
def flush() -> None:
nonlocal current
if current:
phrases.append(Phrase(current[0].start, current[-1].end, _phrase_text(current)))
current = []
for word in words:
if not current:
current = [word]
continue
gap = word.start - current[-1].end
span = word.end - current[0].start
punctuation_break = current[-1].text.endswith((".", "!", "?"))
if (
gap >= gap_seconds
or span > max_phrase_seconds
or len(current) >= max_words
or punctuation_break
):
flush()
current.append(word)
flush()
return phrases
def build_prompt_spans(
phrases: list[Phrase],
*,
fps: float,
duration_frames: int,
prompt_template: str,
pause_template: str,
pause_threshold_seconds: float,
image_segments: list[TimelineImageSegment] | None = None,
) -> list[tuple[int, int, str]]:
visual_segments = image_segments or []
spans: list[tuple[int, int, str]] = []
cursor = 0
for phrase in phrases:
start = _frame(phrase.start, fps)
end = max(start + 1, _frame(phrase.end, fps))
if start - cursor >= _frame(pause_threshold_seconds, fps):
pause_prompt = _format_prompt(pause_template, start=cursor, end=start, fps=fps)
spans.append((
cursor,
start,
_compose_prompt(_active_scene_prompt(cursor, visual_segments), pause_prompt),
))
elif start > cursor and spans:
prev_start, _prev_end, prev_prompt = spans[-1]
spans[-1] = (prev_start, start, prev_prompt)
speech_prompt = _format_prompt(
prompt_template,
text=phrase.text,
start=start,
end=end,
fps=fps,
)
spans.append((
start,
end,
_compose_prompt(_active_scene_prompt(start, visual_segments), speech_prompt),
))
cursor = end
if duration_frames > cursor:
pause_prompt = _format_prompt(
pause_template,
start=cursor,
end=duration_frames,
fps=fps,
)
spans.append((
cursor,
duration_frames,
_compose_prompt(_active_scene_prompt(cursor, visual_segments), pause_prompt),
))
return [(start, max(start + 1, end), prompt) for start, end, prompt in spans]
def build_timeline_data(
phrases: list[Phrase],
*,
image_file: str,
audio_file: str,
fps: float,
duration_frames: int,
audio_length_frames: int,
prompt_template: str,
pause_template: str,
pause_threshold_seconds: float,
audio_segments: list[TimelineAudioSegment] | None = None,
image_segments: list[TimelineImageSegment] | None = None,
) -> dict:
timeline_images = image_segments or [_fallback_image_segment(image_file, duration_frames)]
spans = build_prompt_spans(
phrases,
fps=fps,
duration_frames=duration_frames,
prompt_template=prompt_template,
pause_template=pause_template,
pause_threshold_seconds=pause_threshold_seconds,
image_segments=timeline_images,
)
segments = [
{
"id": f"speech_{idx + 1:03d}",
"start": start,
"length": end - start,
"prompt": prompt,
"type": "image",
"imageFile": (
_active_image_segment(start, timeline_images).image_file
if _active_image_segment(start, timeline_images)
else image_file
),
}
for idx, (start, end, prompt) in enumerate(spans)
]
timeline_audio = audio_segments or [_fallback_audio_segment(audio_file, audio_length_frames or duration_frames)]
audio_payload = [
{
"id": segment.segment_id or f"audio_{idx + 1:03d}",
"type": "audio",
"start": segment.start_frames,
"length": segment.length_frames,
"trimStart": segment.trim_start_frames,
"audioDurationFrames": segment.audio_duration_frames,
"audioFile": segment.audio_file,
"fileName": segment.file_name or os.path.basename(segment.audio_file),
}
for idx, segment in enumerate(timeline_audio)
]
return {"segments": segments, "audioSegments": audio_payload}
class KoolookAudioTranscriptTimeline:
"""Generate LTXDirector timeline strings from a speech audio file."""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio_file": ("STRING", {"default": "", "multiline": True, "tooltip": "Audio filename in ComfyUI input/, or an absolute path."}),
"image_file": ("STRING", {"default": "", "tooltip": "Source image filename in ComfyUI input/."}),
"duration_frames": ("INT", {"default": 120, "min": 1, "max": 10000, "step": 1}),
"frame_rate": ("FLOAT", {"default": 24.0, "min": 1.0, "max": 240.0, "step": 1.0}),
"audio_length_frames": ("INT", {"default": 0, "min": 0, "max": 10000, "step": 1}),
"model_size_or_path": ("STRING", {"default": "base.en", "multiline": True, "tooltip": "faster-whisper model name, or a local model folder."}),
"device": (["cpu", "cuda"], {"default": "cpu"}),
"prompt_template": ("STRING", {"default": DEFAULT_PROMPT_TEMPLATE, "multiline": True}),
"pause_template": ("STRING", {"default": DEFAULT_PAUSE_TEMPLATE, "multiline": True}),
"max_phrase_seconds": ("FLOAT", {"default": 1.2, "min": 0.2, "max": 10.0, "step": 0.1}),
"max_words": ("INT", {"default": 6, "min": 1, "max": 30, "step": 1}),
"gap_seconds": ("FLOAT", {"default": 0.45, "min": 0.0, "max": 5.0, "step": 0.05}),
"pause_threshold_seconds": ("FLOAT", {"default": 0.2, "min": 0.0, "max": 5.0, "step": 0.05}),
},
"optional": {
"timeline_data": ("STRING", {"default": "", "multiline": True, "tooltip": "Optional timeline JSON from Koolook Timeline Editor. When set, all audio segments are transcribed and offset by their timeline start, trim, and clip length."}),
},
}
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = ("timeline_data", "local_prompts", "segment_lengths", "transcript_json")
CATEGORY = "Koolook/Audio"
FUNCTION = "run"
def run(
self,
audio_file: str,
image_file: str,
duration_frames: int,
frame_rate: float,
audio_length_frames: int,
model_size_or_path: str,
device: str,
prompt_template: str,
pause_template: str,
max_phrase_seconds: float,
max_words: int,
gap_seconds: float,
pause_threshold_seconds: float,
timeline_data: str = "",
):
timeline = _timeline_defaults(timeline_data)
if timeline.get("audio_file"):
audio_file = timeline["audio_file"]
if timeline.get("image_file"):
image_file = timeline["image_file"]
if timeline.get("audio_length_frames"):
audio_length_frames = timeline["audio_length_frames"]
audio_segments = list(timeline.get("audio_segments") or [])
if not audio_segments:
audio_segments = [_fallback_audio_segment(audio_file, int(audio_length_frames))]
image_segments = list(timeline.get("image_segments") or [])
if not image_segments:
image_segments = [_fallback_image_segment(image_file, int(duration_frames))]
phrases: list[Phrase] = []
for segment in audio_segments:
audio_path = _resolve_input_file(segment.audio_file)
if not audio_path.is_file():
raise FileNotFoundError(f"Audio file not found: {audio_path}")
words = transcribe_words(audio_path, model_size_or_path, device)
source_phrases = group_words(
words,
max_phrase_seconds=float(max_phrase_seconds),
max_words=int(max_words),
gap_seconds=float(gap_seconds),
)
phrases.extend(
shift_phrases_to_timeline(
source_phrases,
frame_rate=float(frame_rate),
audio_start_frames=segment.start_frames,
audio_trim_start_frames=segment.trim_start_frames,
audio_length_frames=segment.length_frames,
duration_frames=int(duration_frames),
)
)
phrases = sorted(phrases, key=lambda item: (item.start, item.end, item.text))
output_timeline = build_timeline_data(
phrases,
image_file=image_file,
audio_file=audio_file,
fps=float(frame_rate),
duration_frames=int(duration_frames),
audio_length_frames=int(audio_length_frames),
audio_segments=audio_segments,
image_segments=image_segments,
prompt_template=prompt_template,
pause_template=pause_template,
pause_threshold_seconds=float(pause_threshold_seconds),
)
local_prompts = " | ".join(seg["prompt"] for seg in output_timeline["segments"])
segment_lengths = ",".join(str(seg["length"]) for seg in output_timeline["segments"])
transcript = {
"phrases": [
{"start": p.start, "end": p.end, "text": p.text}
for p in phrases
],
}
return (
json.dumps(output_timeline, ensure_ascii=False, separators=(",", ":")),
local_prompts,
segment_lengths,
json.dumps(transcript, ensure_ascii=False, indent=2),
)
NODE_CLASS_MAPPINGS = {"KoolookAudioTranscriptTimeline": KoolookAudioTranscriptTimeline}
NODE_DISPLAY_NAME_MAPPINGS = {
"KoolookAudioTranscriptTimeline": "Koolook Audio Transcript Timeline"
}
__all__ = [
"NODE_CLASS_MAPPINGS",
"NODE_DISPLAY_NAME_MAPPINGS",
"Word",
"Phrase",
"TimelineAudioSegment",
"TimelineImageSegment",
"build_prompt_spans",
"build_timeline_data",
"group_words",
]
k_timeline_editor.py
-61
View File
@@ -1,61 +0,0 @@
# SPDX-License-Identifier: GPL-3.0-or-later
#
# ComfyUI-Koolook - standalone timeline editor
"""Lightweight timeline-data node that reuses the Koolook Director UI."""
from __future__ import annotations
class KoolookTimelineEditor:
"""Expose the Director timeline editor without applying LTX conditioning."""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"duration_frames": ("INT", {"default": 120, "min": 1, "max": 10000, "step": 1}),
"duration_seconds": ("FLOAT", {"default": 5.0, "min": 0.1, "max": 1000.0, "step": 0.01}),
"frame_rate": ("FLOAT", {"default": 24.0, "min": 1.0, "max": 240.0, "step": 1.0}),
"display_mode": (["frames", "seconds"], {"default": "seconds"}),
"timeline_data": ("STRING", {"default": "{}", "multiline": True}),
"local_prompts": ("STRING", {"default": "", "multiline": True}),
"segment_lengths": ("STRING", {"default": ""}),
},
}
RETURN_TYPES = ("STRING", "STRING", "STRING", "FLOAT", "INT")
RETURN_NAMES = (
"timeline_data",
"local_prompts",
"segment_lengths",
"frame_rate",
"duration_frames",
)
CATEGORY = "Koolook/Timeline"
FUNCTION = "run"
def run(
self,
duration_frames: int,
duration_seconds: float,
frame_rate: float,
display_mode: str,
timeline_data: str,
local_prompts: str,
segment_lengths: str,
):
_ = duration_seconds, display_mode
return (
timeline_data or "{}",
local_prompts or "",
segment_lengths or "",
float(frame_rate),
int(duration_frames),
)
NODE_CLASS_MAPPINGS = {"KoolookTimelineEditor": KoolookTimelineEditor}
NODE_DISPLAY_NAME_MAPPINGS = {
"KoolookTimelineEditor": "Koolook Timeline Editor"
}
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
pyproject.toml
-3
View File
@@ -45,8 +45,6 @@ py-modules = [
"k_easy_resize",
"k_easy_track",
"k_easy_wan22_prompt",
"k_audio_timeline",
"k_timeline_editor",
"k_video_combine",
"k_video_load",
"koolook_install_guard",
@@ -73,4 +71,3 @@ addopts = ["--import-mode=importlib"]
# ``aiohttp`` is a runtime dep of ``koolook_routes`` and must be importable
# even when only a pure helper is under test.
test = ["pytest>=8", "aiohttp>=3.9", "pip-audit>=2.7", "Pillow>=10"]
audio = ["faster-whisper>=1.1,<2"]
scripts/sync_to_dev_audio.py
+12 -4
View File
@@ -11,10 +11,12 @@ Copies only what that automation's iteration loop touches:
node mappings; needs to be re-synced
when the fork's NODE_CLASS_MAPPINGS
set changes)
- k_audio_timeline.py transcript-to-Director prompt node
- k_timeline_editor.py standalone Director timeline UI node
- web/whatdreamscost_koolook/ Director timeline-editor extension
It also ships ``koolook_install_guard.py`` + ``koolook_versioning.py`` — the
loader gates ``__init__.py`` imports at load — so a scoped sync never leaves
the loader pointing at a module the target lacks (#198 / #183).
Everything else in the live install - ``forks/radiance_koolook/``, the
root ``k_*.py`` nodes, unrelated ``web/`` assets, ``video_formats/`` -
is left untouched.
@@ -68,8 +70,14 @@ import sync_to_dev as _dev # noqa: E402
AUDIO_PATHS: tuple[str, ...] = (
"__init__.py",
"k_audio_timeline.py",
"k_timeline_editor.py",
# __init__.py imports these at load, before any node group, and a missing
# gate is catastrophic (not per-group-guarded): the install guard's
# absolute-import fallback raises uncaught, and the koolook_versioning
# context probe mislabels its ImportError as a non-Comfy context and
# registers nothing. Ship them with __init__.py so a scoped sync never
# leaves the loader pointing at a module the target lacks (#198 / #183).
"koolook_install_guard.py",
"koolook_versioning.py",
"forks/whatdreamscost_koolook",
"web/whatdreamscost_koolook",
)
scripts/transcribe_audio_timeline.py
-247
View File
@@ -1,247 +0,0 @@
#!/usr/bin/env python3
"""Build timed LTXDirector prompt segments from a speech audio file.
The audio-lipsync loop uses this as a bridge between raw custom audio and
Prompt Relay: transcribe the line, group timestamped words into short phrases,
and emit Director-shaped ``timeline_data`` / ``local_prompts`` /
``segment_lengths`` fields that can be pasted into, or later applied to, a
workflow JSON.
Usage:
python scripts/transcribe_audio_timeline.py <audio.mp3>
python scripts/transcribe_audio_timeline.py <audio.mp3> --workflow workflow.json
python scripts/transcribe_audio_timeline.py <audio.mp3> --out timed.json
Requires the optional audio extra:
python -m pip install -e ".[audio]"
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from k_audio_timeline import ( # noqa: E402
DEFAULT_PAUSE_TEMPLATE,
DEFAULT_PROMPT_TEMPLATE,
Phrase,
group_words,
transcribe_words,
)
def _frame(value: float, fps: float) -> int:
return max(0, int(round(value * fps)))
def load_director_timeline(workflow_path: Path) -> tuple[dict[str, Any], int, float]:
data = json.loads(workflow_path.read_text(encoding="utf-8"))
for node in data.get("nodes", []):
if str(node.get("type", "")).startswith("LTXDirector__koolook"):
values = node.get("widgets_values") or []
timeline = json.loads(values[3]) if len(values) > 3 and values[3] else {}
duration_frames = int(values[5]) if len(values) > 5 else 0
fps = float(values[9]) if len(values) > 9 else 24.0
return timeline, duration_frames, fps
raise SystemExit(f"No Koolook LTXDirector node found in {workflow_path}")
def patch_workflow(
workflow_path: Path,
out_path: Path,
*,
timeline_data: dict[str, Any],
local_prompts: str,
segment_lengths: str,
) -> None:
data = json.loads(workflow_path.read_text(encoding="utf-8"))
for node in data.get("nodes", []):
if str(node.get("type", "")).startswith("LTXDirector__koolook"):
values = node.setdefault("widgets_values", [])
if len(values) < 6:
values.extend([""] * (6 - len(values)))
values[3] = json.dumps(timeline_data, ensure_ascii=False, separators=(",", ":"))
values[4] = local_prompts
values[5] = segment_lengths
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(json.dumps(data, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
return
raise SystemExit(f"No Koolook LTXDirector node found in {workflow_path}")
def build_prompt_spans(
phrases: list[Phrase],
*,
fps: float,
duration_frames: int,
prompt_template: str,
pause_template: str,
pause_threshold_seconds: float,
) -> list[tuple[int, int, str]]:
spans: list[tuple[int, int, str]] = []
cursor = 0
for phrase in phrases:
start = _frame(phrase.start, fps)
end = max(start + 1, _frame(phrase.end, fps))
if start - cursor >= _frame(pause_threshold_seconds, fps):
spans.append((cursor, start, pause_template))
elif start > cursor and spans:
prev_start, _prev_end, prev_prompt = spans[-1]
spans[-1] = (prev_start, start, prev_prompt)
spans.append((start, end, prompt_template.format(text=phrase.text)))
cursor = end
if duration_frames > cursor:
spans.append((cursor, duration_frames, pause_template))
return [(start, max(start + 1, end), prompt) for start, end, prompt in spans]
def build_timeline(
phrases: list[Phrase],
*,
fps: float,
duration_frames: int,
base_timeline: dict[str, Any] | None,
prompt_template: str,
pause_template: str,
pause_threshold_seconds: float,
) -> dict[str, Any]:
base_timeline = base_timeline or {}
source_segments = base_timeline.get("segments") or []
source = source_segments[0] if source_segments else {}
audio_segments = base_timeline.get("audioSegments") or []
spans = build_prompt_spans(
phrases,
fps=fps,
duration_frames=duration_frames,
prompt_template=prompt_template,
pause_template=pause_template,
pause_threshold_seconds=pause_threshold_seconds,
)
segments: list[dict[str, Any]] = []
for idx, (start, end, prompt) in enumerate(spans):
seg = {
"id": f"speech_{idx + 1:03d}",
"start": start,
"length": end - start,
"prompt": prompt,
"type": source.get("type", "image"),
}
for key in ("imageFile", "imageB64"):
if key in source:
seg[key] = source[key]
segments.append(seg)
return {
**base_timeline,
"segments": segments,
"audioSegments": audio_segments,
}
def output_payload(
phrases: list[Phrase],
*,
fps: float,
timeline_data: dict[str, Any],
) -> dict[str, Any]:
local_prompts = [seg["prompt"] for seg in timeline_data["segments"]]
segment_lengths = [str(seg["length"]) for seg in timeline_data["segments"]]
return {
"fps": fps,
"phrases": [
{"start": p.start, "end": p.end, "text": p.text}
for p in phrases
],
"local_prompts": " | ".join(local_prompts),
"segment_lengths": ",".join(segment_lengths),
"timeline_data": timeline_data,
}
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("audio", type=Path)
parser.add_argument("--workflow", type=Path)
parser.add_argument("--out", type=Path)
parser.add_argument("--patched-workflow", type=Path)
parser.add_argument("--model", default="base.en")
parser.add_argument("--device", default="cpu", choices=("cpu", "cuda", "auto"))
parser.add_argument("--fps", type=float, default=24.0)
parser.add_argument("--max-phrase-seconds", type=float, default=1.2)
parser.add_argument("--max-words", type=int, default=6)
parser.add_argument("--gap-seconds", type=float, default=0.45)
parser.add_argument("--prompt-template", default=DEFAULT_PROMPT_TEMPLATE)
parser.add_argument("--pause-threshold-seconds", type=float, default=0.2)
parser.add_argument("--pause-template", default=DEFAULT_PAUSE_TEMPLATE)
args = parser.parse_args(argv)
if not args.audio.is_file():
print(f"Audio file not found: {args.audio}", file=sys.stderr)
return 2
base_timeline = None
duration_frames = 0
fps = args.fps
if args.workflow:
base_timeline, duration_frames, fps = load_director_timeline(args.workflow)
device = args.device
if device == "auto":
try:
import torch
device = "cuda" if torch.cuda.is_available() else "cpu"
except ImportError:
device = "cpu"
words = transcribe_words(args.audio, args.model, device)
phrases = group_words(
words,
max_phrase_seconds=args.max_phrase_seconds,
max_words=args.max_words,
gap_seconds=args.gap_seconds,
)
timeline_data = build_timeline(
phrases,
fps=fps,
duration_frames=duration_frames or _frame(phrases[-1].end, fps),
base_timeline=base_timeline,
prompt_template=args.prompt_template,
pause_template=args.pause_template,
pause_threshold_seconds=args.pause_threshold_seconds,
)
payload = output_payload(phrases, fps=fps, timeline_data=timeline_data)
text = json.dumps(payload, indent=2, ensure_ascii=False)
if args.out:
args.out.parent.mkdir(parents=True, exist_ok=True)
args.out.write_text(text + "\n", encoding="utf-8")
print(f"wrote: {args.out}")
else:
print(text)
if args.patched_workflow:
if not args.workflow:
print("--patched-workflow requires --workflow", file=sys.stderr)
return 2
patch_workflow(
args.workflow,
args.patched_workflow,
timeline_data=timeline_data,
local_prompts=payload["local_prompts"],
segment_lengths=payload["segment_lengths"],
)
print(f"patched workflow: {args.patched_workflow}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
tests/scripts/test_sync_to_dev.py
+78
View File
@@ -0,0 +1,78 @@
"""Guards on which modules ``dev-sync`` ships, so the loader can never point
at a file the install lacks.
The #183 incident (root cause behind #198): a module that ``__init__.py``
imported but ``scripts/sync_to_dev.py``'s ``RUNTIME_PATHS`` never copied raised
``ImportError`` at startup. After the #198 refactor the blast radius of such a
gap depends on *where* the import sits — a per-group node import degrades to
"that node absent", but the context probe (``koolook_versioning``) and the
install guard gate *all* registration. So the real invariant is: every module
``__init__.py`` imports at load must be shipped by ``RUNTIME_PATHS``. This test
derives those imports from ``__init__.py`` itself (rather than a ``k_*.py``
glob, which misses the non-``k_`` gates) so it can't drift.
"""
from __future__ import annotations
import ast
import importlib.util
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[2]
_SCRIPT = REPO_ROOT / "scripts" / "sync_to_dev.py"
_spec = importlib.util.spec_from_file_location("koolook_sync_to_dev_under_test", _SCRIPT)
assert _spec is not None and _spec.loader is not None
sync_to_dev = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(sync_to_dev)
def _init_load_time_module_deps() -> set[str]:
"""Top-level package modules/packages ``__init__.py`` imports at load.
Covers relative ``from .X import ...`` / ``from . import X`` statements and
the ``_merge_node_group(label, ".X")`` node-group imports. Returns the
first path segment of each (e.g. ``forks.radiance_koolook`` -> ``forks``),
the granularity ``RUNTIME_PATHS`` ships at. Absolute imports (the install
guard's out-of-package fallback) are level 0 and intentionally ignored.
"""
tree = ast.parse((REPO_ROOT / "__init__.py").read_text(encoding="utf-8"))
deps: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and (node.level or 0) >= 1:
if node.module: # from .pkg import ...
deps.add(node.module.split(".")[0])
else: # from . import a, b
deps.update(alias.name.split(".")[0] for alias in node.names)
elif (
isinstance(node, ast.Call)
and getattr(node.func, "id", None) == "_merge_node_group"
):
for arg in node.args:
if (
isinstance(arg, ast.Constant)
and isinstance(arg.value, str)
and arg.value.startswith(".")
):
deps.add(arg.value.lstrip(".").split(".")[0])
return deps
def test_init_load_deps_are_shipped_by_runtime_paths():
runtime_paths = set(sync_to_dev.RUNTIME_PATHS)
deps = _init_load_time_module_deps()
# Sanity: the parse actually found the imports (guards against a refactor
# that renames _merge_node_group and silently makes this test vacuous).
assert "koolook_versioning" in deps and "k_ai_pipeline" in deps, (
f"__init__.py import parse looks broken — only found: {sorted(deps)}"
)
# A single-file module ships as "<dep>.py"; a package (e.g. forks) ships as
# its top-level dir entry.
missing = sorted(
dep
for dep in deps
if f"{dep}.py" not in runtime_paths and dep not in runtime_paths
)
assert not missing, (
"__init__.py imports these at load but dev-sync RUNTIME_PATHS does not "
f"ship them: {missing}. Add them to RUNTIME_PATHS in "
"scripts/sync_to_dev.py so a plain dev-sync can't 404 the loader."
)
tests/scripts/test_sync_to_dev_audio.py
+16
View File
@@ -81,3 +81,19 @@ def test_remove_stale_paths_unlinks_symlink_without_touching_target(tmp_path: Pa
def test_target_is_repo_root_detects_source_repo() -> None:
assert sync_to_dev_audio.target_is_repo_root(sync_to_dev_audio._dev.REPO_ROOT)
def test_audio_paths_ship_init_load_gates() -> None:
"""dev-sync-audio ships ``__init__.py``, so it must also ship the modules
``__init__`` imports at load before any node group. Those gates are not
per-group-guarded: a missing ``koolook_install_guard`` makes the absolute
import fallback raise uncaught (plugin dead), and a missing
``koolook_versioning`` makes the context probe mislabel its ImportError as
a non-Comfy context and register nothing (#198 / #183).
"""
audio_paths = set(sync_to_dev_audio.AUDIO_PATHS)
required = {"__init__.py", "koolook_install_guard.py", "koolook_versioning.py"}
missing = sorted(required - audio_paths)
assert not missing, (
f"dev-sync-audio ships __init__.py but not its load-time gate(s): {missing}"
)
tests/scripts/test_transcribe_audio_timeline.py
-418
View File
@@ -1,418 +0,0 @@
"""Pure helper coverage for audio transcript timeline tooling."""
from __future__ import annotations
import importlib.util
import json
import sys
from pathlib import Path
import k_audio_timeline
_SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "transcribe_audio_timeline.py"
_spec = importlib.util.spec_from_file_location("transcribe_audio_timeline", _SCRIPT)
transcribe_audio_timeline = importlib.util.module_from_spec(_spec)
assert _spec and _spec.loader
sys.modules[_spec.name] = transcribe_audio_timeline
_spec.loader.exec_module(transcribe_audio_timeline)
Word = k_audio_timeline.Word
Phrase = k_audio_timeline.Phrase
def test_group_words_splits_on_gap_and_punctuation():
words = [
Word(0.0, 0.2, "This"),
Word(0.2, 0.4, "works."),
Word(0.5, 0.7, "Next"),
Word(1.4, 1.6, "line"),
]
phrases = k_audio_timeline.group_words(
words,
max_phrase_seconds=2.0,
max_words=6,
gap_seconds=0.45,
)
assert [p.text for p in phrases] == ["This works.", "Next", "line"]
def test_build_prompt_spans_includes_silence_gaps_and_tail():
phrases = [
Phrase(0.0, 0.9, "This can't be real."),
Phrase(1.44, 2.06, "It's made by AI."),
]
spans = k_audio_timeline.build_prompt_spans(
phrases,
fps=24.0,
duration_frames=120,
prompt_template='says "{text}"',
pause_template="mouth closed",
pause_threshold_seconds=0.2,
)
assert spans == [
(0, 22, 'says "This can\'t be real."'),
(22, 35, "mouth closed"),
(35, 49, 'says "It\'s made by AI."'),
(49, 120, "mouth closed"),
]
def test_build_prompt_spans_supports_precise_timing_template_fields():
phrases = [Phrase(1.0, 1.5, "What?")]
spans = k_audio_timeline.build_prompt_spans(
phrases,
fps=24.0,
duration_frames=48,
prompt_template='frames {start_frame:03d}-{end_frame:03d}: "{text}"',
pause_template="pause {start_seconds:.2f}-{end_seconds:.2f}s",
pause_threshold_seconds=0.2,
)
assert spans == [
(0, 24, "pause 0.00-1.00s"),
(24, 36, 'frames 024-035: "What?"'),
(36, 48, "pause 1.50-2.00s"),
]
def test_timeline_data_offsets_transcript_timing(monkeypatch, tmp_path: Path):
audio = tmp_path / "line.mp3"
audio.write_bytes(b"fake")
monkeypatch.setattr(k_audio_timeline, "_resolve_input_file", lambda _name: audio)
monkeypatch.setattr(
k_audio_timeline,
"transcribe_words",
lambda *_args, **_kwargs: [Word(0.0, 0.5, "What?")],
)
timeline_data = json.dumps(
{
"segments": [{"imageFile": "bear.png", "start": 0, "length": 120}],
"audioSegments": [
{
"audioFile": "line.mp3",
"start": 24,
"trimStart": 0,
"length": 65,
}
],
}
)
node = k_audio_timeline.KoolookAudioTranscriptTimeline()
_timeline_data, local_prompts, segment_lengths, transcript_json = node.run(
audio_file="ignored.mp3",
image_file="ignored.png",
duration_frames=120,
frame_rate=24.0,
audio_length_frames=0,
model_size_or_path="base.en",
device="cpu",
prompt_template='frames {start_frame:03d}-{end_frame:03d}: "{text}"',
pause_template="pause {start_frame:03d}-{end_frame:03d}",
max_phrase_seconds=1.2,
max_words=6,
gap_seconds=0.45,
pause_threshold_seconds=0.2,
timeline_data=timeline_data,
)
assert 'frames 024-035: "What?"' in local_prompts
assert segment_lengths == "24,12,84"
assert json.loads(transcript_json)["phrases"][0]["start"] == 1.0
def test_timeline_data_transcribes_all_separated_audio_segments(monkeypatch, tmp_path: Path):
for name in ("line_a.mp3", "line_b.mp3"):
(tmp_path / name).write_bytes(b"fake")
monkeypatch.setattr(k_audio_timeline, "_resolve_input_file", lambda name: tmp_path / name)
def fake_transcribe(audio_path: Path, *_args, **_kwargs):
if audio_path.name == "line_a.mp3":
return [Word(0.0, 0.5, "First.")]
return [Word(0.0, 0.5, "Second.")]
monkeypatch.setattr(k_audio_timeline, "transcribe_words", fake_transcribe)
timeline_data = json.dumps(
{
"segments": [{"imageFile": "bear.png", "start": 0, "length": 120}],
"audioSegments": [
{"id": "a", "audioFile": "line_a.mp3", "start": 0, "trimStart": 0, "length": 12},
{"id": "b", "audioFile": "line_b.mp3", "start": 48, "trimStart": 0, "length": 12},
],
}
)
node = k_audio_timeline.KoolookAudioTranscriptTimeline()
output_timeline, local_prompts, segment_lengths, transcript_json = node.run(
audio_file="ignored.mp3",
image_file="ignored.png",
duration_frames=120,
frame_rate=24.0,
audio_length_frames=0,
model_size_or_path="base.en",
device="cpu",
prompt_template='frames {start_frame:03d}-{end_frame:03d}: "{text}"',
pause_template="pause {start_frame:03d}-{end_frame:03d}",
max_phrase_seconds=1.2,
max_words=6,
gap_seconds=0.45,
pause_threshold_seconds=0.2,
timeline_data=timeline_data,
)
assert 'frames 000-011: "First."' in local_prompts
assert 'frames 048-059: "Second."' in local_prompts
assert segment_lengths == "12,36,12,60"
assert [item["audioFile"] for item in json.loads(output_timeline)["audioSegments"]] == [
"line_a.mp3",
"line_b.mp3",
]
transcript = json.loads(transcript_json)
assert [phrase["text"] for phrase in transcript["phrases"]] == ["First.", "Second."]
assert transcript["phrases"][1]["start"] == 2.0
def test_timeline_data_respects_audio_segment_trim_and_length(monkeypatch, tmp_path: Path):
audio = tmp_path / "line.mp3"
audio.write_bytes(b"fake")
monkeypatch.setattr(k_audio_timeline, "_resolve_input_file", lambda _name: audio)
monkeypatch.setattr(
k_audio_timeline,
"transcribe_words",
lambda *_args, **_kwargs: [
Word(0.0, 0.4, "Skipped."),
Word(1.0, 1.5, "Kept."),
Word(2.1, 2.5, "Past clip."),
],
)
timeline_data = json.dumps(
{
"segments": [{"imageFile": "bear.png", "start": 0, "length": 120}],
"audioSegments": [
{
"audioFile": "line.mp3",
"start": 24,
"trimStart": 24,
"length": 24,
}
],
}
)
node = k_audio_timeline.KoolookAudioTranscriptTimeline()
_timeline_data, local_prompts, segment_lengths, transcript_json = node.run(
audio_file="ignored.mp3",
image_file="ignored.png",
duration_frames=120,
frame_rate=24.0,
audio_length_frames=0,
model_size_or_path="base.en",
device="cpu",
prompt_template='frames {start_frame:03d}-{end_frame:03d}: "{text}"',
pause_template="pause {start_frame:03d}-{end_frame:03d}",
max_phrase_seconds=1.2,
max_words=6,
gap_seconds=0.45,
pause_threshold_seconds=0.2,
timeline_data=timeline_data,
)
assert 'frames 024-035: "Kept."' in local_prompts
assert "Skipped." not in local_prompts
assert "Past clip." not in local_prompts
assert segment_lengths == "24,12,84"
assert json.loads(transcript_json)["phrases"] == [
{"start": 1.0, "end": 1.5, "text": "Kept."}
]
def test_timeline_data_combines_active_image_prompt_with_audio_timing(monkeypatch, tmp_path: Path):
audio = tmp_path / "line.mp3"
audio.write_bytes(b"fake")
monkeypatch.setattr(k_audio_timeline, "_resolve_input_file", lambda _name: audio)
monkeypatch.setattr(
k_audio_timeline,
"transcribe_words",
lambda *_args, **_kwargs: [Word(2.0, 2.5, "Hello.")],
)
timeline_data = json.dumps(
{
"segments": [
{
"imageFile": "bear_a.png",
"start": 0,
"length": 48,
"prompt": "The bear waits in a forest.",
},
{
"imageFile": "bear_b.png",
"start": 48,
"length": 72,
"prompt": "The bear leans closer to camera.",
},
],
"audioSegments": [
{"audioFile": "line.mp3", "start": 0, "trimStart": 0, "length": 120},
],
}
)
node = k_audio_timeline.KoolookAudioTranscriptTimeline()
output_timeline, local_prompts, _segment_lengths, _transcript_json = node.run(
audio_file="ignored.mp3",
image_file="ignored.png",
duration_frames=120,
frame_rate=24.0,
audio_length_frames=0,
model_size_or_path="base.en",
device="cpu",
prompt_template='frames {start_frame:03d}-{end_frame:03d}: "{text}"',
pause_template="pause {start_frame:03d}-{end_frame:03d}",
max_phrase_seconds=1.2,
max_words=6,
gap_seconds=0.45,
pause_threshold_seconds=0.2,
timeline_data=timeline_data,
)
assert 'The bear waits in a forest. pause 000-047' in local_prompts
assert 'The bear leans closer to camera. frames 048-059: "Hello."' in local_prompts
timeline = json.loads(output_timeline)
assert timeline["segments"][0]["imageFile"] == "bear_a.png"
assert timeline["segments"][1]["imageFile"] == "bear_b.png"
def test_timeline_data_uses_first_non_empty_image_prompt_as_scene_fallback(monkeypatch, tmp_path: Path):
audio = tmp_path / "line.mp3"
audio.write_bytes(b"fake")
monkeypatch.setattr(k_audio_timeline, "_resolve_input_file", lambda _name: audio)
monkeypatch.setattr(
k_audio_timeline,
"transcribe_words",
lambda *_args, **_kwargs: [Word(0.0, 0.5, "Hello.")],
)
timeline_data = json.dumps(
{
"segments": [
{"imageFile": "bear_a.png", "start": 0, "length": 100, "prompt": ""},
{"imageFile": "bear_b.png", "start": 100, "length": 20, "prompt": "The bear faces camera."},
],
"audioSegments": [
{"audioFile": "line.mp3", "start": 0, "trimStart": 0, "length": 120},
],
}
)
node = k_audio_timeline.KoolookAudioTranscriptTimeline()
_output_timeline, local_prompts, _segment_lengths, _transcript_json = node.run(
audio_file="ignored.mp3",
image_file="ignored.png",
duration_frames=120,
frame_rate=24.0,
audio_length_frames=0,
model_size_or_path="base.en",
device="cpu",
prompt_template='frames {start_frame:03d}-{end_frame:03d}: "{text}"',
pause_template="pause {start_frame:03d}-{end_frame:03d}",
max_phrase_seconds=1.2,
max_words=6,
gap_seconds=0.45,
pause_threshold_seconds=0.2,
timeline_data=timeline_data,
)
assert 'The bear faces camera. frames 000-011: "Hello."' in local_prompts
def test_build_timeline_data_builds_comfy_director_fields():
phrases = [Phrase(0.0, 0.5, "Hello.")]
timeline = k_audio_timeline.build_timeline_data(
phrases,
image_file="bear.png",
audio_file="line.mp3",
fps=24.0,
duration_frames=24,
audio_length_frames=13,
prompt_template='says "{text}"',
pause_template="mouth closed",
pause_threshold_seconds=0.2,
)
assert timeline["audioSegments"][0]["audioFile"] == "line.mp3"
assert timeline["audioSegments"][0]["length"] == 13
assert timeline["segments"][0]["imageFile"] == "bear.png"
assert timeline["segments"][0]["prompt"] == 'says "Hello."'
assert timeline["segments"][1]["prompt"] == "mouth closed"
def test_comfy_node_formats_outputs(monkeypatch, tmp_path: Path):
audio = tmp_path / "line.mp3"
audio.write_bytes(b"fake")
monkeypatch.setattr(k_audio_timeline, "_resolve_input_file", lambda _name: audio)
monkeypatch.setattr(
k_audio_timeline,
"transcribe_words",
lambda *_args, **_kwargs: [
Word(0.0, 0.9, "This"),
Word(1.44, 2.06, "works."),
],
)
node = k_audio_timeline.KoolookAudioTranscriptTimeline()
timeline_data, local_prompts, segment_lengths, transcript_json = node.run(
audio_file="line.mp3",
image_file="bear.png",
duration_frames=120,
frame_rate=24.0,
audio_length_frames=65,
model_size_or_path="base.en",
device="cpu",
prompt_template='bear says "{text}"',
pause_template="mouth closed",
max_phrase_seconds=1.2,
max_words=6,
gap_seconds=0.45,
pause_threshold_seconds=0.2,
)
timeline = json.loads(timeline_data)
assert len(timeline["segments"]) == 4
assert timeline["audioSegments"][0]["length"] == 65
assert timeline["segments"][0]["prompt"] == 'bear says "This"'
assert "mouth closed" in local_prompts
assert segment_lengths == "22,13,14,71"
assert json.loads(transcript_json)["phrases"][1]["text"] == "works."
def test_patch_workflow_updates_director_fields(tmp_path: Path):
workflow = tmp_path / "workflow.json"
patched = tmp_path / "patched.json"
workflow.write_text(
'{"nodes":[{"type":"LTXDirector__koolook","widgets_values":["g",1,1,"{}","old","1"]}]}',
encoding="utf-8",
)
transcribe_audio_timeline.patch_workflow(
workflow,
patched,
timeline_data={"segments": [{"prompt": "new"}], "audioSegments": []},
local_prompts="new",
segment_lengths="24",
)
data = __import__("json").loads(patched.read_text(encoding="utf-8"))
values = data["nodes"][0]["widgets_values"]
assert __import__("json").loads(values[3]) == {
"segments": [{"prompt": "new"}],
"audioSegments": [],
}
assert values[4] == "new"
assert values[5] == "24"
web/whatdreamscost_koolook/ltx_director.js
+1 -2
View File
@@ -3980,8 +3980,7 @@ app.registerExtension({
async beforeRegisterNodeDef(nodeType, nodeData, app) {
if (
nodeData.name === "LTXDirector__koolook" ||
nodeData.name === "LTXDirector__koolook_v1_3_2" ||
nodeData.name === "KoolookTimelineEditor"
nodeData.name === "LTXDirector__koolook_v1_3_2"
) {
const onNodeCreated = nodeType.prototype.onNodeCreated;