diff --git a/fastvideo/api/compat.py b/fastvideo/api/compat.py index 647b71934..cae812f7a 100644 --- a/fastvideo/api/compat.py +++ b/fastvideo/api/compat.py @@ -7,12 +7,13 @@ from dataclasses import fields, is_dataclass from pathlib import Path from typing import Any -from fastvideo.api.overrides import apply_overrides, parse_cli_overrides +from fastvideo.api.overrides import apply_overrides, normalize_overrides from fastvideo.api.parser import config_to_dict, load_raw_config, parse_config from fastvideo.api.request_metadata import ( - EXPLICIT_REQUEST_ATTR, + EXPLICIT_PATHS_ATTR, bind_generation_request_raw, - refresh_generation_request_raw, + get_explicit_paths, + reset_tracking_roots, ) from fastvideo.api.schema import ( GenerationRequest, @@ -50,7 +51,7 @@ def load_generator_config_from_file( overrides: list[str] | Mapping[str, Any] | None = None, ) -> GeneratorConfig: raw = load_raw_config(path) - normalized_overrides = _normalize_overrides(overrides) + normalized_overrides = normalize_overrides(overrides) if _looks_like_run_or_serve_config(raw): if normalized_overrides: @@ -228,9 +229,9 @@ def generator_config_to_fastvideo_args(config: GeneratorConfig | Mapping[str, An def normalize_generation_request(request: GenerationRequest | Mapping[str, Any], ) -> GenerationRequest: normalized = (request if isinstance(request, GenerationRequest) else parse_config(GenerationRequest, request)) - if hasattr(normalized, EXPLICIT_REQUEST_ATTR): - refresh_generation_request_raw(normalized) - else: + if not hasattr(normalized, EXPLICIT_PATHS_ATTR): + # Request wasn't bound through the parser (e.g. constructed + # directly). Treat every currently-set field as explicit. bind_generation_request_raw(normalized, _serialize_generation_request(normalized)) return normalized @@ -279,7 +280,12 @@ def request_to_sampling_param( for key, value in updates.items(): if hasattr(sampling_param, key): setattr(sampling_param, key, deepcopy(value)) - elif key in _REQUEST_PIPELINE_OVERRIDE_FIELDS or _is_supported_as_default_only(key, value): + elif key in _REQUEST_PIPELINE_OVERRIDE_FIELDS: + continue + elif value == _SCHEMA_DEFAULT_UPDATES.get(key, _MISSING): + # Schema-default field that isn't on SamplingParam; tolerated + # because direct GenerationRequest(...) construction has no + # way to distinguish "user set" from "schema default". continue else: raise ValueError(f"Request field {key!r} is not supported by sampling params for {model_path}") @@ -296,10 +302,12 @@ def expand_request_prompt_batch(request: GenerationRequest, ) -> list[Generation requests: list[GenerationRequest] = [] for index, prompt in enumerate(request.prompt): single_request = deepcopy(request) + # deepcopy preserves the tracking-root cycle, but re-pin roots + # defensively so that subsequent setattrs record on the copy. + reset_tracking_roots(single_request) single_request.prompt = prompt _fan_out_batched_input_value(request, single_request, "image_path", index) _fan_out_batched_input_value(request, single_request, "video_path", index) - _fan_out_explicit_request_metadata(request, single_request, index, prompt) requests.append(single_request) return requests @@ -308,14 +316,6 @@ def _looks_like_run_or_serve_config(raw: Mapping[str, Any]) -> bool: return isinstance(raw.get("generator"), Mapping) -def _normalize_overrides(overrides: list[str] | Mapping[str, Any] | None, ) -> dict[str, Any] | None: - if not overrides: - return None - if isinstance(overrides, list): - return parse_cli_overrides(overrides) - return dict(overrides) - - def _sampling_param_to_request_raw(sampling_param: SamplingParam | None, ) -> dict[str, Any]: if sampling_param is None: return {} @@ -361,13 +361,59 @@ def request_to_pipeline_overrides(request: GenerationRequest) -> dict[str, Any]: def _explicit_request_updates(request: GenerationRequest) -> dict[str, Any]: - raw = getattr(request, EXPLICIT_REQUEST_ATTR, None) - if raw is None: - raw = _serialize_generation_request(request) - + assert hasattr(request, + EXPLICIT_PATHS_ATTR), ("GenerationRequest reached _explicit_request_updates without tracking; " + "every entry point must route through normalize_generation_request " + "or parse_config first") + paths = get_explicit_paths(request) + raw = _build_sparse_raw_from_paths(request, paths) return _extract_request_updates(raw) +def _build_sparse_raw_from_paths( + request: GenerationRequest, + paths: frozenset[str], +) -> dict[str, Any]: + result: dict[str, Any] = {} + for path in paths: + parts = path.split(".") + value = _read_dotted_path(request, parts) + if value is _MISSING: + continue + _set_dotted_path(result, parts, deepcopy(value)) + return result + + +def _read_dotted_path(obj: Any, parts: list[str]) -> Any: + for part in parts: + if is_dataclass(obj) and not isinstance(obj, type): + if not hasattr(obj, part): + return _MISSING + obj = getattr(obj, part) + elif isinstance(obj, Mapping): + if part not in obj: + return _MISSING + obj = obj[part] + else: + return _MISSING + return obj + + +def _set_dotted_path( + target: dict[str, Any], + parts: list[str], + value: Any, +) -> None: + cursor = target + for part in parts[:-1]: + nxt = cursor.get(part) + if not isinstance(nxt, dict): + nxt = {} + cursor[part] = nxt + cursor = nxt + cursor[parts[-1]] = value + + def _extract_request_updates(raw: Mapping[str, Any]) -> dict[str, Any]: updates: dict[str, Any] = {} if "negative_prompt" in raw: @@ -411,6 +457,9 @@ def _serialize_generation_request(request: GenerationRequest) -> dict[str, Any]: return deepcopy(config_to_dict(request)) +_SCHEMA_DEFAULT_UPDATES = _extract_request_updates(config_to_dict(GenerationRequest())) + + def _fan_out_batched_input_value( source_request: GenerationRequest, target_request: GenerationRequest, @@ -424,29 +473,6 @@ def _fan_out_batched_input_value( setattr(target_request.inputs, field_name, deepcopy(value[index])) -def _fan_out_explicit_request_metadata( - source_request: GenerationRequest, - target_request: GenerationRequest, - index: int, - prompt: str, -) -> None: - raw = getattr(source_request, EXPLICIT_REQUEST_ATTR, None) - if raw is None: - return - - raw = deepcopy(raw) - raw["prompt"] = prompt - inputs = raw.get("inputs") - if isinstance(inputs, dict): - for field_name in ("image_path", "video_path"): - value = inputs.get(field_name) - if isinstance(value, list): - _validate_batched_input_length(source_request.prompt, value, field_name) - inputs[field_name] = deepcopy(value[index]) - - setattr(target_request, EXPLICIT_REQUEST_ATTR, raw) - - def _validate_batched_input_length( prompts: str | list[str] | None, values: list[Any], @@ -458,43 +484,6 @@ def _validate_batched_input_length( raise ValueError(f"GenerationRequest.inputs.{field_name} must have the same length as request.prompt") -def _is_supported_as_default_only(key: str, value: Any) -> bool: - default_value = _DEFAULT_REQUEST_UPDATES.get(key, _MISSING) - return default_value is not _MISSING and _values_equal(value, default_value) - - -def _collect_non_default_fields( - value: Any, - default: Any, -) -> dict[str, Any]: - if not (is_dataclass(value) and is_dataclass(default)): - return {} - - result: dict[str, Any] = {} - for field in fields(value): - current = getattr(value, field.name) - default_value = getattr(default, field.name) - if is_dataclass(current) and is_dataclass(default_value): - nested = _collect_non_default_fields(current, default_value) - if nested: - result[field.name] = nested - continue - if not _values_equal(current, default_value): - result[field.name] = deepcopy(current) - return result - - -def _values_equal(left: Any, right: Any) -> bool: - if left is right: - return True - try: - return bool(left == right) - except Exception: - return False - - -_DEFAULT_REQUEST_UPDATES = _extract_request_updates(config_to_dict(GenerationRequest())) - __all__ = [ "generator_config_to_fastvideo_args", "legacy_from_pretrained_to_config", diff --git a/fastvideo/api/overrides.py b/fastvideo/api/overrides.py index 3eaca5832..fafac7e38 100644 --- a/fastvideo/api/overrides.py +++ b/fastvideo/api/overrides.py @@ -45,6 +45,15 @@ def apply_overrides(config: Mapping[str, Any], overrides: Mapping[str, Any]) -> return merged +def normalize_overrides(overrides: list[str] | Mapping[str, Any] | None, ) -> dict[str, Any] | None: + """Normalize a CLI list or mapping of overrides into a flat dict.""" + if not overrides: + return None + if isinstance(overrides, list): + return parse_cli_overrides(overrides) + return dict(overrides) + + def _apply_single_override(config: dict[str, Any], dotted_key: str, value: Any) -> None: parts = dotted_key.split(".") if not all(parts): @@ -98,4 +107,4 @@ def _normalize_override_key(key: str) -> str: return key.replace("-", "_") -__all__ = ["apply_overrides", "parse_cli_overrides"] +__all__ = ["apply_overrides", "normalize_overrides", "parse_cli_overrides"] diff --git a/fastvideo/api/parser.py b/fastvideo/api/parser.py index 8da994665..969dc8576 100644 --- a/fastvideo/api/parser.py +++ b/fastvideo/api/parser.py @@ -11,7 +11,7 @@ from typing import Any, Literal, TypeVar, Union, get_args, get_origin, get_type_ import yaml from fastvideo.api.errors import ConfigValidationError -from fastvideo.api.overrides import apply_overrides, parse_cli_overrides +from fastvideo.api.overrides import apply_overrides, normalize_overrides from fastvideo.api.request_metadata import ( bind_generation_request_raw, bind_run_config_raw, @@ -64,7 +64,7 @@ def load_config( ) -> T: """Load a typed config object from YAML or JSON.""" raw = load_raw_config(path) - normalized_overrides = _normalize_overrides(overrides) + normalized_overrides = normalize_overrides(overrides) if normalized_overrides: raw = apply_overrides(raw, normalized_overrides) return parse_config(config_type, raw) @@ -108,14 +108,6 @@ def _load_raw_mapping(handle: Any, config_path: Path) -> Any: raise ValueError(f"Unsupported config file format: {config_path}") -def _normalize_overrides(overrides: list[str] | Mapping[str, Any] | None, ) -> dict[str, Any] | None: - if not overrides: - return None - if isinstance(overrides, list): - return parse_cli_overrides(overrides) - return dict(overrides) - - class _SchemaParser: def parse_dataclass( diff --git a/fastvideo/api/request_metadata.py b/fastvideo/api/request_metadata.py index eb241bf6c..394453e4c 100644 --- a/fastvideo/api/request_metadata.py +++ b/fastvideo/api/request_metadata.py @@ -1,24 +1,25 @@ # SPDX-License-Identifier: Apache-2.0 """Track which GenerationRequest fields the user explicitly provided. -This module solves a specific problem: when translating a GenerationRequest into -a legacy SamplingParam, we need to distinguish user-provided values (which -should override model defaults) from schema defaults (which should NOT override -model defaults). +When translating a GenerationRequest into a legacy SamplingParam we must +distinguish user-provided values (which should override model defaults) +from schema defaults (which should NOT override model defaults). -The approach: -1. At bind time, store the original raw dict and a baseline snapshot. -2. Patch __setattr__ on tracked dataclass types to record dirty field paths. -3. At access time, do a lazy 3-way merge: raw + baseline + current state, - with dirty paths forcing inclusion even when current == baseline. +The mechanism: a single ``_fastvideo_explicit_paths`` set stored on the +root ``GenerationRequest``. It holds dotted leaf paths (e.g. +``"sampling.guidance_scale"``) the user has touched, either via raw +config at bind time or via attribute assignment at runtime. A patched +``__setattr__`` on the request dataclass types records assignments into +this set. + +The set holds leaf paths only. Nested dataclass or mapping assignments +are flattened to their leaves at record time. """ from __future__ import annotations -from collections.abc import Mapping -from copy import deepcopy +from collections.abc import Callable, Mapping import dataclasses from typing import Any, cast -from collections.abc import Callable from fastvideo.api.schema import ( ContinuationState, @@ -33,12 +34,12 @@ from fastvideo.api.schema import ( ServeConfig, ) -EXPLICIT_REQUEST_ATTR = "_fastvideo_explicit_request" -ORIGINAL_REQUEST_STATE_ATTR = "_fastvideo_original_request_state" +EXPLICIT_PATHS_ATTR = "_fastvideo_explicit_paths" + _TRACKING_ROOT_ATTR = "_fastvideo_request_tracking_root" _TRACKING_PATH_ATTR = "_fastvideo_request_tracking_path" _TRACKING_PATCHED_ATTR = "_fastvideo_request_tracking_patched" -_DIRTY_PATHS_ATTR = "_fastvideo_dirty_paths" + _TRACKED_REQUEST_TYPES = ( GenerationRequest, InputConfig, @@ -55,14 +56,20 @@ def bind_generation_request_raw( request: GenerationRequest, raw: Mapping[str, Any] | None, ) -> GenerationRequest: + """Install explicit-path tracking on *request*. + + *raw* is the parsed config dict (YAML/JSON/kwargs); every leaf key + in it becomes an explicit path. Subsequent attribute assignments on + *request* or its nested dataclasses are recorded automatically via a + patched ``__setattr__``. + """ _ensure_request_tracking() - # Disable dirty tracking during bind so tree walk doesn't record paths. - object.__setattr__(request, _DIRTY_PATHS_ATTR, None) - object.__setattr__(request, EXPLICIT_REQUEST_ATTR, deepcopy(dict(raw or {}))) - object.__setattr__(request, ORIGINAL_REQUEST_STATE_ATTR, _serialize_config(request)) + # Disable recording while we walk the tree to install roots. + object.__setattr__(request, EXPLICIT_PATHS_ATTR, None) _set_tracking_roots(request, request, "") - # Enable dirty tracking. - object.__setattr__(request, _DIRTY_PATHS_ATTR, set()) + paths: set[str] = set() + _record_value_paths(raw or {}, "", paths) + object.__setattr__(request, EXPLICIT_PATHS_ATTR, paths) return request @@ -73,6 +80,8 @@ def bind_run_config_raw( request_raw = raw.get("request") if isinstance(request_raw, Mapping): bind_generation_request_raw(config.request, request_raw) + else: + bind_generation_request_raw(config.request, {}) return config @@ -83,87 +92,72 @@ def bind_serve_config_raw( default_request_raw = raw.get("default_request") if isinstance(default_request_raw, Mapping): bind_generation_request_raw(config.default_request, default_request_raw) - elif "default_request" not in raw: + else: bind_generation_request_raw(config.default_request, {}) return config -def refresh_generation_request_raw(request: GenerationRequest, ) -> dict[str, Any] | None: - raw = getattr(request, EXPLICIT_REQUEST_ATTR, None) - baseline = getattr(request, ORIGINAL_REQUEST_STATE_ATTR, None) - if not isinstance(raw, Mapping) or not isinstance(baseline, Mapping): - return None +def get_explicit_paths(request: GenerationRequest) -> frozenset[str]: + """Return a snapshot of the explicit paths set on *request*.""" + paths = getattr(request, EXPLICIT_PATHS_ATTR, None) + if isinstance(paths, set | frozenset): + return frozenset(paths) + return frozenset() - dirty = getattr(request, _DIRTY_PATHS_ATTR, None) or frozenset() - current = _serialize_config(request) - merged = deepcopy(dict(raw)) - _merge_request_mutations(merged, dict(baseline), current, dirty) - object.__setattr__(request, EXPLICIT_REQUEST_ATTR, merged) - object.__setattr__(request, ORIGINAL_REQUEST_STATE_ATTR, current) - object.__setattr__(request, _DIRTY_PATHS_ATTR, set()) - return merged +def reset_tracking_roots(request: GenerationRequest) -> None: + """Re-install tracking roots after a deepcopy or manual clone. + + The paths set itself deepcopies correctly; we only need to repoint + the tracking root on nested dataclasses at the new root. + """ + _ensure_request_tracking() + _set_tracking_roots(request, request, "") # --------------------------------------------------------------------------- -# 3-way merge: raw + baseline + current, with dirty-path forcing +# Path recording # --------------------------------------------------------------------------- -_MISSING = object() - -def _merge_request_mutations( - merged: dict[str, Any], - baseline: Mapping[str, Any], - current: Mapping[str, Any], - dirty: frozenset[str] | set[str], - path_prefix: str = "", - force_dirty: bool = False, +def _record_value_paths( + value: Any, + prefix: str, + out: set[str], ) -> None: - # Remove keys that were deleted from the current state. - for key in set(merged) | set(baseline): - if key not in current: - merged.pop(key, None) + """Add every leaf path under *value* to *out*. - for key in current: - current_path = f"{path_prefix}.{key}" if path_prefix else key - current_value = current[key] - baseline_value = baseline.get(key, _MISSING) - merged_value = merged.get(key, _MISSING) - - # If this exact path was dirtied (e.g. whole section replaced), - # propagate to all children. - child_force = force_dirty or current_path in dirty - - # Recurse into nested mappings. - if isinstance(current_value, Mapping) and isinstance(baseline_value, Mapping): - nested = (deepcopy(dict(merged_value)) if isinstance(merged_value, Mapping) else {}) - _merge_request_mutations( - nested, - baseline_value, - current_value, - dirty, - current_path, - child_force, - ) - if nested: - merged[key] = nested - else: - merged.pop(key, None) - continue - - # A field is explicitly set if: - # - it's new (not in baseline), - # - it changed from baseline, - # - its path was touched by __setattr__ (dirty), or - # - an ancestor path was dirty (whole section replaced). - is_dirty = child_force or current_path in dirty - if baseline_value is _MISSING or current_value != baseline_value or is_dirty: - merged[key] = deepcopy(current_value) + A leaf is any terminal value (non-dataclass, non-mapping, or empty + mapping/dataclass). ``prefix`` is the dotted path at which *value* + sits. When called with an empty ``prefix`` (the root), leaves are + recorded at their own key. + """ + if dataclasses.is_dataclass(value) and not isinstance(value, type): + dc_fields = dataclasses.fields(value) + if not dc_fields: + if prefix: + out.add(prefix) + return + for field in dc_fields: + child = getattr(value, field.name) + path = f"{prefix}.{field.name}" if prefix else field.name + _record_value_paths(child, path, out) + return + if isinstance(value, Mapping): + if not value: + if prefix: + out.add(prefix) + return + for key, child in value.items(): + path = f"{prefix}.{key}" if prefix else key + _record_value_paths(child, path, out) + return + if prefix: + out.add(prefix) # --------------------------------------------------------------------------- -# __setattr__ patching for dirty-path recording +# __setattr__ patching # --------------------------------------------------------------------------- @@ -187,16 +181,23 @@ def _patch_tracking_setattr(config_type: type[Any]) -> None: original_setattr(self, name, value) return - root = getattr(self, _TRACKING_ROOT_ATTR, None) - if root is not None: - dirty = getattr(root, _DIRTY_PATHS_ATTR, None) - if isinstance(dirty, set): - prefix = getattr(self, _TRACKING_PATH_ATTR, "") - path = f"{prefix}.{name}" if prefix else name - dirty.add(path) - original_setattr(self, name, value) + root = getattr(self, _TRACKING_ROOT_ATTR, None) + if root is None: + return + paths = getattr(root, EXPLICIT_PATHS_ATTR, None) + if not isinstance(paths, set): + return + + prefix = getattr(self, _TRACKING_PATH_ATTR, "") + path = f"{prefix}.{name}" if prefix else name + # Wholesale dataclass replacement: install roots on the new + # instance so its future mutations are tracked too. + if dataclasses.is_dataclass(value) and not isinstance(value, type): + _set_tracking_roots(root, value, path) + _record_value_paths(value, path, paths) + type.__setattr__(config_type, "__setattr__", _tracking_setattr) setattr(config_type, _TRACKING_PATCHED_ATTR, True) @@ -222,26 +223,11 @@ def _set_tracking_roots( _set_tracking_roots(root, child, child_path) -# --------------------------------------------------------------------------- -# Serialization helper -# --------------------------------------------------------------------------- - - -def _serialize_config(config: Any) -> Any: - if dataclasses.is_dataclass(config) and not isinstance(config, type): - return {field.name: _serialize_config(getattr(config, field.name)) for field in dataclasses.fields(config)} - if isinstance(config, list): - return [_serialize_config(item) for item in config] - if isinstance(config, dict): - return {key: _serialize_config(value) for key, value in config.items()} - return deepcopy(config) - - __all__ = [ - "EXPLICIT_REQUEST_ATTR", - "ORIGINAL_REQUEST_STATE_ATTR", + "EXPLICIT_PATHS_ATTR", "bind_generation_request_raw", "bind_run_config_raw", "bind_serve_config_raw", - "refresh_generation_request_raw", + "get_explicit_paths", + "reset_tracking_roots", ] diff --git a/fastvideo/entrypoints/cli/serve.py b/fastvideo/entrypoints/cli/serve.py index 0e52fa572..100b795f0 100644 --- a/fastvideo/entrypoints/cli/serve.py +++ b/fastvideo/entrypoints/cli/serve.py @@ -6,7 +6,7 @@ import os from typing import cast from fastvideo.api.compat import generator_config_to_fastvideo_args -from fastvideo.api.request_metadata import EXPLICIT_REQUEST_ATTR +from fastvideo.api.request_metadata import get_explicit_paths from fastvideo.entrypoints.cli.cli_types import CLISubcommand from fastvideo.entrypoints.cli.inference_config import build_serve_config from fastvideo.logger import init_logger @@ -30,12 +30,7 @@ class ServeSubcommand(CLISubcommand): args, overrides=getattr(args, "_unknown", None), ) - explicit_raw = getattr( - serve_config.default_request, - EXPLICIT_REQUEST_ATTR, - None, - ) - if explicit_raw: + if get_explicit_paths(serve_config.default_request): raise NotImplementedError("ServeConfig.default_request is not wired into the OpenAI " "server yet")