diff --git a/CLAUDE.md b/CLAUDE.md index 4273b5c..0cbdc4c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -86,7 +86,7 @@ From building the first real look (Que Calor V2), and independently reproduced h ## The types, in one paragraph -`Effect` is what the caller **asks for** and carries **no tier** — if it could, the refusal would be theatre. `Look` is an ordered stack plus a `Policy`; `a + b` takes the **stricter** of the two, per field. `Ref` is a parameter that must be measured, and one with no default that the probe does not answer **raises**. `ImplRef` declares `Terms`, never a tier. A tier appears for the first time in a compiled `Step`, *read* off the selected implementation. Three identity levels: `look_hash` (intent — policy-independent), `plan_hash` (pixels — includes the clip's colour state and the env fingerprint), `output_key` (content-addressed — takes a digest of **bytes** and refuses a path). +`Effect` is what the caller **asks for** and carries **no tier** — if it could, the refusal would be theatre. `Look` is an ordered stack plus a `Policy`; `a + b` takes the **stricter** of the two, per field. `Ref` is a parameter that must be measured, and one with no default that the probe does not answer **raises**. `ImplRef` declares `Terms`, never a tier, and also `timeline` (can it be gated to an `Effect.at`?) and `time_varying` (does its own fragment read the clock regardless of any span?) — close to inverses in practice (issue #17), and `time_varying` is **tri-state**: `None` means undeclared, never "does not read the clock". A tier appears for the first time in a compiled `Step`, *read* off the selected implementation; `Step.time_varying` and `LookPlan.time_varying` are tri-state too, ORing/aggregating `ImplRef.time_varying` with whether an `Effect.at` Span actually bounds something. Three identity levels: `look_hash` (intent — policy-independent), `plan_hash` (pixels — includes the clip's colour state and the env fingerprint), `output_key` (content-addressed — takes a digest of **bytes** and refuses a path). `_admits` **delegates** to `licence.check` rather than comparing tiers itself, and a test asserts its body has no comparison operator. The one place a tier meets a ceiling must stay one place. diff --git a/docs/decisions_and_rationale.md b/docs/decisions_and_rationale.md index 3ebffef..9ae5ca4 100644 --- a/docs/decisions_and_rationale.md +++ b/docs/decisions_and_rationale.md @@ -179,6 +179,7 @@ class ImplRef: requires_filters: tuple[str, ...] = () # ffmpeg filters this impl emits impl_version: str = "1" # behaviour lock; enters plan_hash UNCONDITIONALLY timeline: bool = True # can this be gated to an Effect.at? + time_varying: bool | None = None # does its OWN fragment read the clock? None = undeclared preference: int = 0 # explicit tiebreak within one tier; lower wins lossy_substitute_for: tuple[str, ...] = () # see rule 12 ``` @@ -189,6 +190,8 @@ class ImplRef: **`timeline` is a candidate FILTER, not a post-check.** Verified on ffmpeg 8.1: `lut3d`, `lutrgb`, `curves`, `unsharp`, `gblur`, `colorchannelmixer`, `colorlevels`, `geq` carry the `T` flag; `scale`, `crop`, `elbg`, `palettegen`, `zoompan` do not, and `xfade` does not. If `Effect.at` is set, ungateable candidates are dropped in step 1 alongside the `impl`/`backend` pins — checking it *after* tier selection discards a perfectly good gateable candidate and then refuses. [R3][2] +**`time_varying` answers a different question from `timeline`, and the two are close to inverses in practice (issue #17).** `timeline` means "can be gated to an `Effect.at`"; `time_varying` means "does the fragment read the clock regardless of any span". `motion.ffmpeg.crop` compiles to `zoompan`/`crop` expressions that reference `in_time` — the one implementation registered here that reads the clock — and it has `timeline=False`, while every static grade has `timeline=True` and never reads the clock. **Tri-state, and `None` is not `False`**, for the same reason `Step.cpu_seconds` and `DependencyReport.can_flicker` are tri-state: an implementation nobody declared this for is *unknown*, not "does not read the clock" — a silent `False` would let a consumer like muvid#73 ship an unwarned ramp restart across a transitioned boundary. `register_defaults` therefore requires every built-in implementation to declare it explicitly, with no default. `Step.time_varying` ORs it with "this step carries an `at` Span that actually bounds something" (`Span.is_whole`, shared with `looks.ffmpeg.gated`'s own emission check, so the two decisions cannot drift apart) — a certain span-gate is `True` even over an undeclared impl. `LookPlan.time_varying` aggregates: one certain `True` step wins outright; otherwise any unknown step poisons the answer to `None`, mirroring `has_unknown_costs`/`total_cpu_seconds`. Measured: `frame_dependency.classify`'s own TIME_VARYING probe is a spatially uniform still, so it cannot see a moving crop window — `motion.ffmpeg.crop` reads `time_delta=0.0` against it and is classified `CONTENT_ADAPTIVE` instead, a blind spot in that probe's still rather than a wrong declaration (confirmed independently by compiling the effect and reading `LookPlan.time_varying`). + ### 4.6 The compiled form: `Step` and `LookPlan` ```python diff --git a/looks/ffmpeg.py b/looks/ffmpeg.py index 37356c9..4412538 100644 --- a/looks/ffmpeg.py +++ b/looks/ffmpeg.py @@ -136,10 +136,12 @@ def gated(fragment: str, at: Optional[Span]) -> str: >>> gated("scale=2:2,crop=1:1", Span(0.0, 1.0)) "scale=2:2:enable='between(t,0,1)',crop=1:1:enable='between(t,0,1)'" """ - if at is None or (at.start is None and at.end is None): - # A Span open at BOTH ends bounds nothing, so it is not a gate. Emitting - # `enable=` for it would be a filter option that always evaluates true — - # noise in the string and a lie in a diff. + if at is None or at.is_whole: + # A Span open at BOTH ends (`Span.is_whole`) bounds nothing, so it is + # not a gate. Emitting `enable=` for it would be a filter option that + # always evaluates true — noise in the string and a lie in a diff. + # `looks.spec.Step.time_varying` asks this exact question through the + # same predicate, on purpose: one decision, shared rather than copied. return fragment if at.start is None: window = f"enable='lte(t,{_num(at.end)})'" @@ -388,7 +390,21 @@ def register_defaults(registry: Optional[EffectRegistry] = None) -> EffectRegist return reg terms = _ffmpeg_terms() - def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): + def add( + key, + filters, + compiler, + *, + time_varying, + preference=0, + timeline=True, + tags=(), + ): + # `time_varying` has no default, deliberately: `ImplRef.time_varying` + # is unknown (`None`) unless declared, and unknown is not the same + # claim as "does not read the clock". Forcing every built-in + # registration to say which is how this module avoids shipping a + # silent `False` for something nobody checked. effect, backend, _variant = key.split(".") reg.register( ImplRef( @@ -399,6 +415,7 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): requires_filters=filters, preference=preference, timeline=timeline, + time_varying=time_varying, ), compiler, tags=tags, @@ -409,6 +426,7 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): "gradient_map.ffmpeg.lut3d", ("lut3d",), _gradient_map_compiler, + time_varying=False, ) add( "lut3d.ffmpeg.default", @@ -417,11 +435,13 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): "lut3d", lambda p: {"file": p["cube"], "interp": p.get("interp", "tetrahedral")}, ), + time_varying=False, ) add( "saturation.ffmpeg.colorchannelmixer", ("colorchannelmixer",), _simple("colorchannelmixer", _saturation_matrix), + time_varying=False, ) add( "saturation.ffmpeg.eq", @@ -429,11 +449,13 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): _simple("eq", lambda p: {"saturation": p.get("amount", 1.0)}), preference=1, tags=("gpl-gated",), + time_varying=False, ) add( "contrast.ffmpeg.lutyuv", ("lutyuv",), _simple("lutyuv", _contrast_luma), + time_varying=False, ) add( "contrast.ffmpeg.eq", @@ -441,8 +463,14 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): _simple("eq", lambda p: {"contrast": p.get("amount", 1.0)}), preference=1, tags=("gpl-gated",), + time_varying=False, + ) + add( + "gamma.ffmpeg.lutrgb", + ("lutrgb",), + _simple("lutrgb", _gamma_expression), + time_varying=False, ) - add("gamma.ffmpeg.lutrgb", ("lutrgb",), _simple("lutrgb", _gamma_expression)) add( "levels.ffmpeg.colorlevels", ("colorlevels",), @@ -457,9 +485,13 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): "bimax": p.get("white", 1), }, ), + time_varying=False, ) add( - "posterize.ffmpeg.lutrgb", ("lutrgb",), _simple("lutrgb", _posterize_expression) + "posterize.ffmpeg.lutrgb", + ("lutrgb",), + _simple("lutrgb", _posterize_expression), + time_varying=False, ) add( "flatten.ffmpeg.bilateral", @@ -475,6 +507,7 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): "planes": p.get("planes", 7), }, ), + time_varying=False, ) # --- spatial ------------------------------------------------------------ @@ -482,6 +515,7 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): "blur.ffmpeg.gblur", ("gblur",), _simple("gblur", lambda p: {"sigma": p.get("sigma", 2)}), + time_varying=False, ) add( "blur.ffmpeg.boxblur", @@ -489,6 +523,7 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): _simple("boxblur", lambda p: {"luma_radius": p.get("radius", 2)}), preference=1, tags=("gpl-gated",), + time_varying=False, ) add( "sharpen.ffmpeg.unsharp", @@ -500,6 +535,7 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): # the registry sweep, which is what that sweep is for. lambda p: {"luma_amount": p.get("amount", 1.0)}, ), + time_varying=False, ) # --- geometry, wrapping looks.geometry ---------------------------------- @@ -510,7 +546,13 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): # its own registry data — the refusal existed and never fired, and the # binary rejected the command instead. A gated geometry step is now # refused at SELECTION, which is where a caller can act on it. - add(f"{mode}.ffmpeg.scale", ("scale",), _geometry(mode), timeline=False) + add( + f"{mode}.ffmpeg.scale", + ("scale",), + _geometry(mode), + timeline=False, + time_varying=False, + ) # --- motion, wrapping looks.motion -------------------------------------- add( @@ -518,6 +560,12 @@ def add(key, filters, compiler, *, preference=0, timeline=True, tags=()): ("crop", "setpts"), _motion_compiler, timeline=False, + # `zoompan`/`crop` compile to expressions that reference `in_time` — + # the one implementation registered here that reads the clock + # regardless of any Effect.at span. See issue #17: `timeline=False` + # (above) answers "can it be gated to a span", which is a different + # and close-to-inverse question. + time_varying=True, ) return reg diff --git a/looks/frame_dependency.py b/looks/frame_dependency.py index 80e01c0..fc43db8 100644 --- a/looks/frame_dependency.py +++ b/looks/frame_dependency.py @@ -71,6 +71,20 @@ discontinuity at every cut. Do not read that as making the gap harmless — it means the gap is not currently load-bearing, not that it is closed. +**A purely positional clock-reader can be misclassified `CONTENT_ADAPTIVE`.** +The TIME_VARYING probe's still (`_still_source`) is spatially UNIFORM +(`color=c=gray`), so an effect whose output depends on the clock only through +*where* it samples — a moving crop window, say — produces zero luma +difference against it: there is nothing for a diff to see when every pixel is +the same colour regardless of position. Measured on +``looks.ffmpeg``'s ``motion.ffmpeg.crop`` (a `crop` whose x/y expressions +read `t`): `time_delta == 0.0` against this probe. The asymmetric content +probe incidentally reacts to the moving window (`content_delta` nonzero, +because the window drifts into the varying half), so the effect is reported +`CONTENT_ADAPTIVE` rather than `TIME_VARYING` — the wrong class for the right +reason. See `looks` issue #20 and `looks/tests/test_time_varying.py`'s +``TestTheStillProbeCannotSeeAMovingCropWindow`` for the full measurement. + **And an `INDEPENDENT` verdict is evidence, not proof.** It says: none of three probes could make this effect change a pixel it should not have touched. An effect adapting to a statistic none of them moves would pass, and so would diff --git a/looks/spec.py b/looks/spec.py index 7e4800c..b5d8425 100644 --- a/looks/spec.py +++ b/looks/spec.py @@ -724,7 +724,30 @@ class ImplRef: impl_version: A behaviour lock, not a receipt — "same interface, changed behaviour" bumps it, and it enters :func:`plan_hash` **unconditionally**. - timeline: Whether it can be gated to an :attr:`Effect.at`. + timeline: Whether it can be gated to an :attr:`Effect.at`. Answers a + different question from :attr:`time_varying` and the two are close + to inverses in practice: ``motion.ffmpeg.crop`` compiles to + ``zoompan``/``crop`` expressions that read ``in_time`` and has + **no** timeline support (``timeline=False``), while every static + grade has timeline support (``timeline=True``) and never reads the + clock. See issue #17. + time_varying: Whether this implementation's own fragment reads the + clock **regardless of any span** — true only for a camera move + whose expressions reference ``in_time``. Distinct from + :attr:`Step.time_varying`, which also accounts for an + :attr:`Effect.at` Span gating an otherwise-static filter with + ``enable='between(t,...)'``. + + **Tri-state, and ``None`` is not ``False``.** The same rule as + :attr:`Step.cpu_seconds` and + :attr:`~looks.frame_dependency.DependencyReport.can_flicker`: + "we were not told" and "it does not" are different claims, and only + one of them is a guarantee a caller may act on. An implementation + registered without declaring this reads as **unknown**, never as + "does not read the clock" — a silent ``False`` here is exactly the + false permission that would let muvid#73's ramp restart ship + unwarned. :func:`looks.ffmpeg.register_defaults` requires every + built-in implementation to declare it explicitly for this reason. preference: Explicit tiebreak within one tier; lower wins. Examples: @@ -736,6 +759,8 @@ class ImplRef: '1' >>> impl.timeline True + >>> impl.time_varying is None # undeclared is unknown, not False + True """ effect: str @@ -745,6 +770,7 @@ class ImplRef: requires_filters: tuple[str, ...] = () impl_version: str = "1" timeline: bool = True + time_varying: Optional[bool] = None preference: int = 0 def __post_init__(self) -> None: @@ -895,6 +921,22 @@ class Step: 'lut3d=look.cube' >>> s.cpu_seconds is None # unknown, not free True + + An impl that declares it does not read the clock stays that way — gate + it to a span and it does, regardless (issue #17, the question + ``ImplRef.timeline`` looks like but is not): + + >>> from dataclasses import replace + >>> static = replace(s, impl=replace(impl, time_varying=False)) + >>> static.time_varying + False + >>> replace(static, at=Span(1.0, 2.0)).time_varying + True + + An impl that never declared it reads as **unknown**, not ``False``: + + >>> s.time_varying is None + True """ effect: str @@ -922,6 +964,30 @@ def __post_init__(self) -> None: object.__setattr__(self, "payload", _freeze(self.payload)) object.__setattr__(self, "metadata", _freeze(self.metadata)) + @property + def time_varying(self) -> Optional[bool]: + """Whether this compiled fragment reads the clock (issue #17). + + Not ``not self.impl.timeline`` — that field means "can be gated to a + span", which is close to the *inverse* of this question: the one + genuinely clock-reading implementation registered today + (``motion.ffmpeg.crop``, whose ``zoompan``/``crop`` expressions + reference ``in_time``) has **no** timeline support, while every static + grade has it and never reads the clock. + + **Tri-state.** A span that actually bounds something is a *certain* + ``True`` regardless of what the implementation declares — it compiles + with ``enable='between(t,...)'`` no matter what — using the same + predicate :func:`looks.ffmpeg.gated` uses to decide whether to emit + one (:attr:`Span.is_whole`, so the two never drift apart). Absent + that, the answer is exactly :attr:`ImplRef.time_varying`, unknown + included: an implementation that never declared it does not become + "does not read the clock" just because no span was given either. + """ + if self.at is not None and not self.at.is_whole: + return True + return self.impl.time_varying + @dataclass(frozen=True, slots=True, kw_only=True) class LookPlan: @@ -1002,6 +1068,33 @@ def has_unknown_costs(self) -> bool: """Whether anything is unpriced. A gate reads this, not the sum.""" return self.unknown_step_count > 0 + @property + def time_varying(self) -> Optional[bool]: + """Whether any compiled step reads the clock — see :attr:`Step.time_varying`. + + The question issue #17 was filed for: a consumer that renders a blend + as a separate input-seeked invocation (input-side ``-ss`` rebases the + filter clock to 0) needs to know, per compiled fragment, whether that + rebasing can affect it. ``not impl.timeline`` answers a different + question and gets it wrong on three of the four filters that matter — + this property is the one to ask instead. + + **Tri-state, and a known ``True`` wins over an unknown step.** One + step that certainly reads the clock makes the plan ``True`` even if + another step's implementation never declared itself — a plan is not + "unknown" about the one thing it is sure of. Absent any certain + ``True``, an unknown step poisons the answer to ``None``: "every step + we could classify does not read the clock" is not "nothing here + reads the clock", the same asymmetry :attr:`has_unknown_costs` keeps + separate from :attr:`total_cpu_seconds`. + """ + values = [s.time_varying for s in self.steps] + if any(v is True for v in values): + return True + if any(v is None for v in values): + return None + return False + @property def realtime_factor(self) -> Optional[float]: """CPU-seconds per second of output, or ``None`` when anything is unknown. diff --git a/looks/tests/test_time_varying.py b/looks/tests/test_time_varying.py new file mode 100644 index 0000000..b550f94 --- /dev/null +++ b/looks/tests/test_time_varying.py @@ -0,0 +1,328 @@ +"""Whether a compiled fragment reads the clock — issue #17. + +`ImplRef.timeline` looks like the answer and is close to the exact inverse: +`motion.ffmpeg.crop` (`zoompan`/`crop`, which reference `in_time`) has no +timeline support, and every static grade — which never reads the clock — does. +`Step.time_varying` and `LookPlan.time_varying` are the fields a consumer +should ask instead, folding in both ways a fragment ends up clock-driven: the +implementation itself, or an `Effect.at` Span gating an otherwise-static +filter with `enable=`. + +**Tri-state throughout, on purpose.** `None` means "nobody told us", never +"does not read the clock" — the same posture as +`frame_dependency.DependencyReport.can_flicker`. A declaration is an +assertion; `TestDeclarationAgreesWithMeasurement` is the check, sweeping every +registered ffmpeg implementation and asking `frame_dependency.classify` to +confirm it against the same looped-still probe the flicker tests already use. + +Offline and free: uses the real default registry (`looks.ffmpeg.register_defaults`, +run on `import looks`), synthesises every source with `lavfi`, and nothing +here starts a process outside of `ffmpeg`/`ffprobe` analysis runs. +""" + +import dataclasses +import shutil + +import pytest + +import looks +from looks.cache import materialize +from looks.compile import compile_look +from looks.environment import probe +from looks.ffmpeg import vf +from looks.frame_dependency import NOISE_FLOOR, Dependency, classify +from looks.licence import terms_for +from looks.registry import REGISTRY +from looks.spec import ClipSpec, Effect, ImplRef, Look, LookPlan, Span, Step + +CLIP = ClipSpec(width=320, height=240, fps=10) + +#: The minimal params each registered effect needs to compile, mirroring +#: test_ffmpeg.py's own sweep fixture — kept independent so this file does not +#: need to import that module's private table. +PARAMS = { + "lut3d": {"cube": None}, # filled in by the fixture with a real file + "gradient_map": { + "stops": [[8.2, "#2E0C18"], [46.8, "#D5254A"], [100.0, "#FEF0DC"]], + "size": 9, + }, + "saturation": {"amount": 1.4}, + "contrast": {"amount": 1.2}, + "gamma": {"gamma": 1.2}, + "levels": {"black": 0.02, "white": 0.98}, + "posterize": {"levels": 6}, + "flatten": {"spatial": 20, "range": 0.05}, + "blur": {"sigma": 2}, + "sharpen": {"amount": 1.0}, + "fit": {"target": "320x240"}, + "fill": {"target": "shorts"}, + "stretch": {"target": "square"}, + "motion": {"keyframes": [(0.0, (0.0, 0.0, 0.5, 0.5)), (0.5, (0.5, 0.5, 0.5, 0.5))]}, +} + +ALL_IMPLS = tuple( + impl for effect in REGISTRY.effects() for impl in REGISTRY.implementations(effect) +) + + +def _ffmpeg_or_skip() -> None: + if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None: + pytest.skip("ffmpeg/ffprobe not on PATH") + + +def _impl(effect="lut3d", *, timeline=True, time_varying=None): + return ImplRef( + effect=effect, + impl=f"{effect}.ffmpeg.default", + backend="ffmpeg", + terms=terms_for("ffmpeg")[0], + timeline=timeline, + time_varying=time_varying, + ) + + +@pytest.fixture(scope="module") +def env(): + e = probe() + if not e.available: + pytest.skip("ffmpeg not usable") + return e + + +@pytest.fixture(scope="module") +def cube(tmp_path_factory): + ramp = looks.Ramp.from_hex( + [(8.2, "#2E0C18"), (46.8, "#D5254A"), (100.0, "#FEF0DC")] + ) + path = tmp_path_factory.mktemp("cubes") / "look.cube" + return looks.write_cube(looks.gradient_map(ramp), path, size=17) + + +class TestImplRefTimeVaryingIsUnknownByDefault: + """`None` is the default, and it is not a spelling of `False`.""" + + def test_undeclared_reads_as_unknown_not_false(self): + assert _impl().time_varying is None + + def test_timeline_and_time_varying_are_independent_fields(self): + # A camera move: no timeline support, and it DOES read the clock. + motion = _impl("motion", timeline=False, time_varying=True) + assert motion.timeline is False + assert motion.time_varying is True + + +class TestStepTimeVarying: + """A Step reads the clock via its impl, or via a bounding `at` Span — + and a certain span-gate wins even over an impl that never declared itself.""" + + def test_a_static_step_with_no_span_reflects_its_impls_declaration(self): + step = Step( + effect="gamma", + impl=_impl("gamma", time_varying=False), + tier=_impl().tier, + ) + assert step.time_varying is False + + def test_an_undeclared_impl_with_no_span_is_unknown(self): + step = Step(effect="gamma", impl=_impl("gamma"), tier=_impl().tier) + assert step.time_varying is None + + def test_a_span_open_at_both_ends_bounds_nothing_and_does_not_count(self): + step = Step( + effect="gamma", + impl=_impl("gamma", time_varying=False), + tier=_impl().tier, + at=Span(), + ) + assert step.time_varying is False + + def test_a_real_span_wins_even_over_a_declared_false(self): + static = Step( + effect="blur", impl=_impl("blur", time_varying=False), tier=_impl().tier + ) + gated = dataclasses.replace(static, at=Span(1.0, 2.0)) + assert static.time_varying is False + assert gated.time_varying is True + + def test_a_real_span_wins_even_over_an_undeclared_impl(self): + step = Step( + effect="blur", impl=_impl("blur"), tier=_impl().tier, at=Span(1.0, 2.0) + ) + assert step.time_varying is True + + def test_a_clock_reading_impl_is_time_varying_even_with_no_span(self): + motion_impl = _impl("motion", timeline=False, time_varying=True) + step = Step(effect="motion", impl=motion_impl, tier=motion_impl.tier) + assert step.time_varying is True + + +class TestLookPlanTimeVarying: + """The aggregate a consumer actually asks, tri-state and all.""" + + def test_an_empty_plan_does_not_read_the_clock(self): + assert LookPlan().time_varying is False + + def test_one_certain_true_step_wins_even_beside_an_unknown_one(self): + unknown = Step(effect="gamma", impl=_impl("gamma"), tier=_impl().tier) + motion_impl = _impl("motion", timeline=False, time_varying=True) + moving = Step(effect="motion", impl=motion_impl, tier=motion_impl.tier) + assert LookPlan(steps=(unknown, moving)).time_varying is True + + def test_an_unknown_step_poisons_an_otherwise_false_plan(self): + known_false = Step( + effect="gamma", + impl=_impl("gamma", time_varying=False), + tier=_impl().tier, + ) + unknown = Step(effect="blur", impl=_impl("blur"), tier=_impl().tier) + assert LookPlan(steps=(known_false,)).time_varying is False + assert LookPlan(steps=(known_false, unknown)).time_varying is None + + +class TestEveryBuiltInImplementationDeclaresItExplicitly: + """`register_defaults` requires the kwarg — this is the drift guard.""" + + def test_no_built_in_implementation_ships_unknown(self): + undeclared = sorted(i.impl for i in ALL_IMPLS if i.time_varying is None) + assert not undeclared, ( + f"{undeclared} are registered with no time_varying declaration. " + "`None` means unknown, and a shipped implementation should not " + "ship an unanswered question about whether it reads the clock." + ) + + +#: `frame_dependency.classify`'s TIME_VARYING probe (`_still_source`) is a +#: spatially UNIFORM still (`color=c=gray`). Measured here: a `crop` whose +#: window position reads `t` produces `time_delta == 0.0` against it, because +#: cropping a flat colour is invariant to where the window sits — there is +#: nothing for a luma diff to see. The classifier's decision order checks +#: `time_delta` first, so a purely positional clock-reader that the time probe +#: cannot see falls through to whichever of `temporal_delta`/`content_delta` +#: its *content* probe (a spatially split source) happens to catch instead — +#: measured as `CONTENT_ADAPTIVE` (`content_delta=83.0`) for +#: `motion.ffmpeg.crop`, not `TIME_VARYING`. That is a blind spot in the +#: probe's still, not a wrong declaration: `TestAgainstTheRealRegistry` below +#: confirms `time_varying=True` for this impl a different way, by compiling it +#: and reading `LookPlan.time_varying` directly. Excluded from the +#: measurement sweep for that reason, with the finding recorded rather than +#: silently skipped. +_BLIND_TO_THE_STILL_PROBE = frozenset({"motion.ffmpeg.crop"}) + + +class TestDeclarationAgreesWithMeasurement: + """The declaration is an assertion; frame_dependency.classify is the check. + + Mirrors test_ffmpeg.py's `TestEveryRegisteredEffectConfigures` sweep, one + level deeper: not just "does ffmpeg accept this fragment" but "does its + measured dependency class agree with what this PR just declared about it". + """ + + def test_every_effect_has_parameters_here(self): + """So the sweep cannot quietly stop covering a new effect.""" + assert set(looks.effects()) == set(PARAMS), ( + "an effect was registered without being added to this sweep's PARAMS" + ) + + @pytest.mark.parametrize( + "impl", + [i for i in ALL_IMPLS if i.impl not in _BLIND_TO_THE_STILL_PROBE], + ids=lambda i: i.impl, + ) + def test_declared_time_varying_matches_the_probed_dependency( + self, impl, env, cube, tmp_path_factory + ): + _ffmpeg_or_skip() + # No `pytest.skip` for a missing PARAMS entry here: `PARAMS` is + # asserted complete above, so a gap would silently drop an impl out of + # this sweep rather than reddening. + params = dict(PARAMS[impl.effect]) + if impl.effect == "lut3d": + params["cube"] = str(cube) + look = Look(steps=(Effect(name=impl.effect, params=params, impl=impl.impl),)) + plan = compile_look(look, clip=CLIP, env=env) + plan = materialize(plan, into=tmp_path_factory.mktemp("materialize")) + fragment = vf(plan) + report = classify(fragment) + if report.dependency is Dependency.UNDETERMINED: + pytest.skip(f"a probe did not run for {impl.impl!r}: {report.note}") + measured = report.dependency is Dependency.TIME_VARYING + assert impl.time_varying is measured, ( + f"{impl.impl!r} declares time_varying={impl.time_varying!r}, but " + f"frame_dependency.classify measured {report.dependency.value!r} " + f"(time_delta={report.time_delta}) against {fragment!r}" + ) + + +class TestTheStillProbeCannotSeeAMovingCropWindow: + """The excluded case, measured and named rather than silently skipped. + + `motion.ffmpeg.crop`'s window position is a function of `t`, but + `frame_dependency.classify`'s time probe runs it over a spatially uniform + still, where no crop window produces a different pixel. The mismatch is in + the probe's still, not in this PR's `time_varying=True` declaration — + confirmed independently in `TestAgainstTheRealRegistry`, which reads + `LookPlan.time_varying` off a real compiled plan rather than inferring it + from `frame_dependency.classify`. + """ + + def test_a_flat_still_cannot_reveal_a_moving_crop_window(self, env, tmp_path): + _ffmpeg_or_skip() + look = Look( + steps=( + Effect( + name="motion", + params={ + "keyframes": [ + (0.0, (0.0, 0.0, 0.5, 0.5)), + (0.5, (0.5, 0.5, 0.5, 0.5)), + ] + }, + ), + ) + ) + plan = compile_look(look, clip=CLIP, env=env) + report = classify(vf(plan)) + # <= NOISE_FLOOR, not exact equality: this asserts the probe still + # cannot see the moving window, and reddens the day it can (rather + # than the day a measured value moves by rounding noise). + assert report.time_delta <= NOISE_FLOOR + assert report.dependency is not Dependency.TIME_VARYING + # The impl still declares (and LookPlan.time_varying still reports) + # the true answer — the probe's blind spot does not leak into it. + assert plan.time_varying is True + + +class TestAgainstTheRealRegistry: + """The registered fact the issue was filed about, end to end.""" + + def test_motion_reads_the_clock_and_a_static_grade_does_not(self): + env_ = probe() + if not env_.available: + pytest.skip("ffmpeg not usable") + motion = Look( + steps=( + Effect( + name="motion", + params={"keyframes": [(0.0, (0.0, 0.0, 1.0, 1.0))]}, + ), + ) + ) + grade = Look(steps=(Effect(name="gamma", params={"gamma": 1.2}),)) + motion_plan = compile_look(motion, clip=CLIP, env=env_) + grade_plan = compile_look(grade, clip=CLIP, env=env_) + assert motion_plan.time_varying is True + assert grade_plan.time_varying is False + # The obvious-but-wrong field, named so a future reader sees why this + # test does not use it: `timeline` is inverted for exactly this case. + assert motion_plan.steps[0].impl.timeline is False + assert grade_plan.steps[0].impl.timeline is True + + def test_the_same_static_grade_reads_the_clock_once_gated_to_a_span(self): + env_ = probe() + if not env_.available: + pytest.skip("ffmpeg not usable") + gated = Look( + steps=(Effect(name="gamma", params={"gamma": 1.2}, at=Span(1.0, 2.0)),) + ) + plan = compile_look(gated, clip=CLIP, env=env_) + assert plan.time_varying is True