Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 21 additions & 2 deletions cleave/config_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -1825,7 +1825,19 @@ def parse_timeline_section(data: dict[str, Any], ctx: ParseCtx) -> Any | None:
+ ", ".join(unknown)
)
layers = {stem: bool(layers_raw[stem]) for stem in layers_raw}
cues.append(TimelineCue(t=t, layers=layers))
no_tick_raw = cue_map.get("no_tick", [])
if no_tick_raw is None:
no_tick_raw = []
if not isinstance(no_tick_raw, list):
raise ValueError(f"timeline.cues[{index}].no_tick must be a list")
no_tick = frozenset(str(stem) for stem in no_tick_raw)
unknown_no_tick = sorted(no_tick - set(layers))
if unknown_no_tick:
raise ValueError(
f"timeline.cues[{index}].no_tick must reference layers in the cue: "
+ ", ".join(unknown_no_tick)
)
cues.append(TimelineCue(t=t, layers=layers, no_tick_slots=no_tick))
cues.sort(key=lambda cue: cue.t)
return TimelineConfig(enabled=enabled, cues=tuple(cues))

Expand All @@ -1835,12 +1847,19 @@ def persist_timeline(ctx: PersistCtx) -> dict[str, Any]:
out: dict[str, Any] = {"enabled": runtime.enabled}
if runtime.cues:
out["cues"] = [
{"t": cue.t, "layers": dict(cue.layers)}
_persist_timeline_cue(cue)
for cue in sorted(runtime.cues, key=lambda cue: cue.t)
]
return out


def _persist_timeline_cue(cue: TimelineCue) -> dict[str, Any]:
entry: dict[str, Any] = {"t": cue.t, "layers": dict(cue.layers)}
if cue.no_tick_slots:
entry["no_tick"] = sorted(cue.no_tick_slots)
return entry


def persisted_session_payload(cfg: Any, session: Any) -> dict[str, Any]:
cfg_dir = getattr(cfg, "config_path", None)
cfg_dir = cfg_dir.parent if cfg_dir is not None else None
Expand Down
67 changes: 53 additions & 14 deletions cleave/timeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,18 @@
class TimelineCue:
t: float
layers: dict[str, bool]
show_tick: bool = True
no_tick_slots: frozenset[str] = frozenset()

def shows_tick(self, slot: str) -> bool:
"""True when *slot* changes here and marks a real (ticked) transition.

Tick-ness is per slot so cues that merge several slots at the same time
keep each slot's own transition semantics. A slot in ``no_tick_slots`` is
a synthetic anchor/restore (baseline capture, committed restore, seek
fill) that neither draws a tick nor counts as a transition for
pre-first-cue anchor inference.
"""
return slot in self.layers and slot not in self.no_tick_slots


def stem_abbreviation(stem: StemSource) -> str:
Expand Down Expand Up @@ -73,7 +84,7 @@ def _merge_cues_at_same_t(cues: list[TimelineCue]) -> list[TimelineCue]:
merged: list[TimelineCue] = []
current_t: float | None = None
current_layers: dict[str, bool] = {}
current_show_tick = True
current_no_tick: set[str] = set()
for cue in sorted(cues, key=lambda c: c.t):
if current_t is None:
current_t = cue.t
Expand All @@ -82,20 +93,24 @@ def _merge_cues_at_same_t(cues: list[TimelineCue]) -> list[TimelineCue]:
TimelineCue(
t=current_t,
layers=dict(current_layers),
show_tick=current_show_tick,
no_tick_slots=frozenset(current_no_tick),
)
)
current_t = cue.t
current_layers = {}
current_show_tick = True
current_layers.update(cue.layers)
current_show_tick = current_show_tick and cue.show_tick
current_no_tick = set()
for slot, value in cue.layers.items():
current_layers[slot] = value
if slot in cue.no_tick_slots:
current_no_tick.add(slot)
else:
current_no_tick.discard(slot)
if current_t is not None:
merged.append(
TimelineCue(
t=current_t,
layers=dict(current_layers),
show_tick=current_show_tick,
no_tick_slots=frozenset(current_no_tick),
)
)
return merged
Expand All @@ -108,14 +123,34 @@ def punch_replace(
stop_sec: float,
new_cues: list[TimelineCue],
) -> list[TimelineCue]:
kept = [
cue
for cue in cues
"""Overwrite armed slots in ``[start_sec, stop_sec]``; leave unarmed slots intact.

Cues are often multi-slot (timeline presets put every slot on the t=0 cue,
and simultaneous transitions share one object). Deleting a whole cue when it
mentions an armed stem would erase unarmed slots that share that object —
strip armed keys only, then merge in ``new_cues``.
"""
kept: list[TimelineCue] = []
for cue in cues:
if not (
start_sec <= cue.t <= stop_sec
and _cue_modifies_armed_stem(cue, armed_stems)
)
]
):
kept.append(cue)
continue
remaining = {
slot: visible
for slot, visible in cue.layers.items()
if slot not in armed_stems
}
if remaining:
kept.append(
TimelineCue(
t=cue.t,
layers=remaining,
no_tick_slots=cue.no_tick_slots.intersection(remaining),
)
)
combined = kept + list(new_cues)
return _merge_cues_at_same_t(combined)

Expand All @@ -142,7 +177,11 @@ def snap_cues_to_beats(
if beats.size == 1:
sole = float(beats[0])
snapped = [
TimelineCue(t=sole, layers=dict(cue.layers), show_tick=cue.show_tick)
TimelineCue(
t=sole,
layers=dict(cue.layers),
no_tick_slots=cue.no_tick_slots,
)
for cue in cues
]
return _merge_cues_at_same_t(snapped)
Expand Down Expand Up @@ -171,7 +210,7 @@ def snap_t(t: float) -> float:
TimelineCue(
t=snap_t(cue.t),
layers=dict(cue.layers),
show_tick=cue.show_tick,
no_tick_slots=cue.no_tick_slots,
)
for cue in cues
]
Expand Down
8 changes: 4 additions & 4 deletions cleave/viz/layer_visibility.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ def _anchor_visibility_for_slot(
if not slot_cues:
return layer_enabled_fallback
earliest = min(slot_cues, key=lambda cue: cue.t)
if earliest.t > 0.0 and earliest.show_tick:
if earliest.t > 0.0 and earliest.shows_tick(slot):
return not earliest.layers[slot]
return layer_enabled_fallback

Expand Down Expand Up @@ -149,7 +149,7 @@ def build_record_punch_cues(
layers={
slot: timeline_committed_visible(session, slot, 0.0),
},
show_tick=False,
no_tick_slots=frozenset({slot}),
)
)
for slot, baseline in tl.record_baseline.items():
Expand All @@ -160,7 +160,7 @@ def build_record_punch_cues(
TimelineCue(
t=record_start,
layers={slot: baseline},
show_tick=False,
no_tick_slots=frozenset({slot}),
)
)
punch.extend(
Expand All @@ -178,7 +178,7 @@ def build_record_punch_cues(
TimelineCue(
t=record_stop,
layers={slot: committed_at_stop},
show_tick=False,
no_tick_slots=frozenset({slot}),
)
)
return punch
Expand Down
12 changes: 10 additions & 2 deletions cleave/viz/timeline_controls.py
Original file line number Diff line number Diff line change
Expand Up @@ -346,13 +346,21 @@ def _fill_record_at_seek(self, old_t: float, new_t: float) -> None:
remaining = {k: val for k, val in cue.layers.items() if k != slot}
if remaining:
cleaned.append(
TimelineCue(t=cue.t, layers=remaining, show_tick=cue.show_tick)
TimelineCue(
t=cue.t,
layers=remaining,
no_tick_slots=cue.no_tick_slots.intersection(remaining),
)
)
else:
cleaned.append(cue)
tl.record_buffer = cleaned
tl.record_buffer.append(
TimelineCue(t=skip_start, layers={slot: v}, show_tick=False)
TimelineCue(
t=skip_start,
layers={slot: v},
no_tick_slots=frozenset({slot}),
)
)
self._last_toggle_t.pop(slot, None)
tl.record_buffer.sort(key=lambda c: c.t)
Expand Down
6 changes: 1 addition & 5 deletions cleave/viz/timeline_overlay.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,11 +125,7 @@ def cue_times_for_stem(
{
cue.t
for cue in cues
if (
slot in cue.layers
and cue.show_tick
and 0.0 <= cue.t <= duration_sec
)
if cue.shows_tick(slot) and 0.0 <= cue.t <= duration_sec
}
)

Expand Down
7 changes: 5 additions & 2 deletions cleave/viz/timeline_panel_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,9 +65,12 @@ def visibility_bucket(visibility: float) -> int:
return min(255, int(visibility * 255))


def _cue_fingerprint(cues: list) -> tuple[tuple[float, tuple[tuple[str, bool], ...], bool], ...]:
def _cue_fingerprint(
cues: list,
) -> tuple[tuple[float, tuple[tuple[str, bool], ...], tuple[str, ...]], ...]:
return tuple(
(cue.t, tuple(sorted(cue.layers.items())), cue.show_tick) for cue in cues
(cue.t, tuple(sorted(cue.layers.items())), tuple(sorted(cue.no_tick_slots)))
for cue in cues
)


Expand Down
39 changes: 39 additions & 0 deletions tests/cleave/test_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -1035,6 +1035,45 @@ def test_parse_timeline_reads_cues_sorted_by_t() -> None:
)


def test_parse_timeline_reads_no_tick_slots() -> None:
timeline = parse_timeline_section(
{
"timeline": {
"cues": [
{
"t": 12.0,
"layers": {"layer_1": True, "layer_2": False},
"no_tick": ["layer_1"],
},
],
}
},
_timeline_parse_ctx(),
)
assert timeline is not None
assert timeline.cues == (
TimelineCue(
t=12.0,
layers={"layer_1": True, "layer_2": False},
no_tick_slots=frozenset({"layer_1"}),
),
)


def test_parse_timeline_rejects_no_tick_not_in_layers() -> None:
with pytest.raises(ValueError, match="no_tick must reference layers in the cue"):
parse_timeline_section(
{
"timeline": {
"cues": [
{"t": 1.0, "layers": {"layer_1": True}, "no_tick": ["layer_2"]},
],
}
},
_timeline_parse_ctx(),
)


def test_parse_timeline_rejects_unknown_stem() -> None:
with pytest.raises(ValueError, match="unknown layer keys in timeline.cues"):
parse_timeline_section(
Expand Down
34 changes: 34 additions & 0 deletions tests/cleave/test_config_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -966,6 +966,40 @@ def test_write_session_snapshot_persists_timeline_at_bottom(tmp_path: Path) -> N
assert session2.timeline.cues == list(timeline.cues)


def test_write_session_snapshot_round_trips_no_tick_slots(tmp_path: Path) -> None:
cfg, session, out_path = _snapshot_fixture(tmp_path)
session.timeline.enabled = True
session.timeline.cues = [
TimelineCue(t=0.0, layers={"layer_1": True}, no_tick_slots=frozenset({"layer_1"})),
TimelineCue(t=12.0, layers={"layer_1": False}),
]
write_session_snapshot(out_path, cfg=cfg, session=session)

data = yaml.safe_load(out_path.read_text(encoding="utf-8"))
assert data["timeline"]["cues"] == [
{"t": 0.0, "layers": {"layer_1": True}, "no_tick": ["layer_1"]},
{"t": 12.0, "layers": {"layer_1": False}},
]

timeline = parse_timeline_section(
data,
ParseCtx(layer_slots=tuple(cfg.layer_z_order)),
)
assert timeline is not None
playlists = _round_trip_playlists(cfg.paths.preset_root)
cfg_with_timeline = CleaveConfig(
paths=cfg.paths,
layers=cfg.layers,
editor=cfg.editor,
config_path=out_path,
user_config_path=cfg.user_config_path,
render=cfg.render,
timeline=timeline,
)
session2 = session_from_cfg(cfg_with_timeline, playlists)
assert session2.timeline.cues == session.timeline.cues


def test_write_session_snapshot_persists_timeline_disabled_without_cues(
tmp_path: Path,
) -> None:
Expand Down
Loading
Loading