Core I/O 与媒体

逻辑路径、本地和远程 URI、序列化、图像视频规范化以及分块媒体处理。

最后更新: 2026年7月22日

本页内容

Core I/O 为模型接入提供统一的“数据在哪里、以什么形态存在”契约。路径 helper 解析 WorldFoundry 逻辑位置；storage helper 操作本地路径和受支持的 URI scheme；序列化根据显式格式或后缀选择实现；媒体 helper 在进入模型专属预处理之前规范化常见图像和视频输入。

解析逻辑路径，不要写死机器路径

worldfoundry_path_tokens 会计算 checkpoint、dataset、model、artifact、cache、源码仓库和 Conda 环境根目录。显式环境变量优先，否则使用可预测的 WorldFoundry cache 或仓库相邻目录。

from worldfoundry.core.io.paths import (
    checkpoint_root_path,
    resolve_worldfoundry_path,
    worldfoundry_path_tokens,
)

env = {
    "WORLDFOUNDRY_HOME": "/srv/wf",
    "WORLDFOUNDRY_CKPT_DIR": "/models/checkpoints",
}

tokens = worldfoundry_path_tokens(env)
assert tokens["WORLDFOUNDRY_CKPT_DIR"] == "/models/checkpoints"
assert checkpoint_root_path("matrix-game-2", env=env) == \
    resolve_worldfoundry_path("${WORLDFOUNDRY_CKPT_DIR}/matrix-game-2", env)

传入 env mapping 可以在不修改进程环境的情况下测试路径解析。resolve_data_path 的职责不同：它指向 package 自带的静态数据，不应该用于下载的数据集。

序列化往返示例

from pathlib import Path
from tempfile import TemporaryDirectory

from worldfoundry.core import dump_serialized, load_serialized

with TemporaryDirectory() as directory:
    path = Path(directory) / "request.yaml"
    dump_serialized({"seed": 42, "actions": ["forward", "left"]}, path)
    payload = load_serialized(path)
    assert payload["seed"] == 42

没有传 file_format 时，文件后缀会选择 JSON、YAML、JSONL、pickle/gzip、NumPy、Torch、图像、视频、CSV/Pandas 或 tar。没有传文件时，文本和二进制格式会直接返回序列化结果。Pickle 和不受限制的 Torch checkpoint 是可执行格式，不要从不可信来源加载。

coerce_video_frames 是路径、Torch tensor、NumPy array、PIL frame list 和 tensor frame list 的统一规范化入口，返回 T × H × W × C 布局的 uint8 NumPy array。video_tensor_to_uint8_frames 处理规范化的 C × T × H × W 或单 batch tensor，并明确 value range。read_video 还会返回 decoder metadata，load_frames_from_video 则适合只读取指定 frame index。

当 subprocess 或外部 runtime 必须得到本地文件名时，使用 materialize_video_input。只有 codec 或模型函数明确支持重叠时空 tile 时才使用 TileProcessor；它会改变执行布局，但最后把重叠区融合回单一输出。

完整参考

以下为该类别的生成签名。可用本页符号索引跳转；源码链接指向各惰性导出背后的具体实现。

56 个公开符号

`checkpoint_root_path`

func

def checkpoint_root_path(*parts: str | Path,specific_env: str | None = None,env: Mapping[str, str] | None = None) -> Path

worldfoundry.core.io.paths.checkpoint_root_pathfrom worldfoundry.core.io.paths import checkpoint_root_path

源码

简介

checkpoint_root_path — Resolves a target model checkpoint directory path with nested subfolders. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

源码 docstring

Resolves a target model checkpoint directory path with nested subfolders.

Avoids host-specific hardcoding by querying general and model-specific variables.

参数

partsstr | Path: Child path components appended to the resolved checkpoint root.
specific_envstr | None: Optional model-specific environment variable that takes precedence over `WORLDFOUNDRY_CKPT_DIR`.默认值: None
envMapping[str, str] | None: Environment mapping used instead of `os.environ`.默认值: None

返回值: Path — Resolved checkpoint path without creating it.

`coerce_video_frames`

func

def coerce_video_frames(video_input)

worldfoundry.core.coerce_video_framesfrom worldfoundry.core import coerce_video_frames

源码

简介

coerce_video_frames — Normalize common video inputs into a uint8 THWC numpy array. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

video_input

`Color`

cls

class Color()

worldfoundry.core.Colorfrom worldfoundry.core import Color

源码

简介

Color — Small color helper compatible with legacy model-runtime config printers. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

value

`copy_uri`

func

def copy_uri(src: str | os.PathLike[str],dst: str | os.PathLike[str],**storage_options) -> str

worldfoundry.core.copy_urifrom worldfoundry.core import copy_uri

源码

简介

copy_uri — Copy one URI to another using storage-aware byte streams. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str。

参数

srcstr | os.PathLike[str]
dststr | os.PathLike[str]
storage_options

返回值: str

`crop_and_resize`

func

def crop_and_resize(image: Image.Image,height: int,width: int) -> Image.Image

worldfoundry.core.crop_and_resizefrom worldfoundry.core import crop_and_resize

源码

简介

crop_and_resize — Center-crop and resize a PIL image to `(width, height). 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Image.Image`。

参数

imageImage.Image
heightint
widthint

返回值: Image.Image

`dump_serialized`

func

def dump_serialized(obj: Any,file: str | Path | IO[Any] | None = None,file_format: str | None = None,encoding: str = 'utf-8',**kwargs: Any) -> str | bytes | None

worldfoundry.core.dump_serializedfrom worldfoundry.core import dump_serialized

源码

简介

dump_serialized — Dump an object to a URI, file object, or string when `file is None. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str | bytes | None`。

参数

objAny
filestr | Path | IO[Any] | None: 默认值: None
file_formatstr | None: 默认值: None
encodingstr: 默认值: 'utf-8'
kwargsAny

返回值: str | bytes | None

`exists_uri`

func

def exists_uri(uri: str | os.PathLike[str], **storage_options) -> bool

worldfoundry.core.exists_urifrom worldfoundry.core import exists_uri

源码

简介

判断 WorldFoundry URI（本地或远程协议）当前是否存在。

参数

uristr | os.PathLike[str]
storage_options

返回值: bool

`extract_frames_from_video_url`

func

def extract_frames_from_video_url(video_url: str)

worldfoundry.core.extract_frames_from_video_urlfrom worldfoundry.core import extract_frames_from_video_url

源码

简介

extract_frames_from_video_url — Decode a remote video URL and return RGB PIL frames. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

video_urlstr

`file_sha256`

func

def file_sha256(path: str | os.PathLike[str], , chunk_size: int = 1024  1024) -> str

worldfoundry.core.file_sha256from worldfoundry.core import file_sha256

源码

简介

file_sha256 — Compute a file's SHA-256 digest without loading the whole artifact into memory. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str。

参数

pathstr | os.PathLike[str]
chunk_sizeint: 默认值: 1024 * 1024

返回值: str

`get_video_details`

func

def get_video_details(video_path: str | Path) -> tuple[int, float, float]

worldfoundry.core.get_video_detailsfrom worldfoundry.core import get_video_details

源码

简介

get_video_details — Return `(total_frames, fps, duration_seconds) for a local video. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：tuple[int, float, float]`。

参数

video_pathstr | Path

返回值: tuple[int, float, float]

`hf_download_or_fpath`

func

def hf_download_or_fpath(path: str | PathLike[str] | None) -> str | Any

worldfoundry.core.hf_download_or_fpathfrom worldfoundry.core import hf_download_or_fpath

源码

简介

hf_download_or_fpath — Backwards-compatible alias for :func:resolve_hf_path. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str | Any。

参数

pathstr | PathLike[str] | None

返回值: str | Any

`hfd_root_path`

func

def hfd_root_path(parts: str | Path, env: Mapping[str, str] | None = None) -> Path

worldfoundry.core.io.paths.hfd_root_pathfrom worldfoundry.core.io.paths import hfd_root_path

源码

简介

hfd_root_path — Resolves the hfd-style local downloader checkpoint directory. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

参数

partsstr | Path
envMapping[str, str] | None: 默认值: None

返回值: Path

`infer_serialization_format`

func

def infer_serialization_format(file: str | Path | IO[Any] | None,file_format: str | None = None) -> str

worldfoundry.core.infer_serialization_formatfrom worldfoundry.core import infer_serialization_format

源码

简介

infer_serialization_format — Infer a normalized serialization format key. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str。

参数

filestr | Path | IO[Any] | None
file_formatstr | None: 默认值: None

返回值: str

`is_remote_uri`

func

def is_remote_uri(uri: str | os.PathLike[str]) -> bool

worldfoundry.core.is_remote_urifrom worldfoundry.core import is_remote_uri

源码

简介

is_remote_uri — Return whether a path needs a non-local storage backend. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：bool。

参数

uristr | os.PathLike[str]

返回值: bool

`join_uri`

func

def join_uri(base: str | os.PathLike[str], parts: str | os.PathLike[str]) -> str

worldfoundry.core.join_urifrom worldfoundry.core import join_uri

源码

简介

join_uri — Join path components without losing remote URI schemes. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str。

参数

basestr | os.PathLike[str]
partsstr | os.PathLike[str]

返回值: str

`jsonable`

func

def jsonable(value: Any) -> Any

worldfoundry.core.jsonablefrom worldfoundry.core import jsonable

源码

简介

jsonable — Recursively convert common runtime objects into JSON-safe values. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Any。

参数

valueAny: Dataclass, `to_dict object, path, mapping, sequence, tensor/ array-like object with tolist`, callable, primitive, or fallback object.

说明

Mapping keys are stringified, sets become lists, callables become a module/qualified-name record, and otherwise unsupported objects fall back to `repr`. This is evidence serialization, not a reversible object codec.

返回值: Any — A tree containing only JSON-compatible primitives and containers.

`load_frames_from_video`

func

def load_frames_from_video(video_path: str | Path,indices: Iterable[int],video_decode_backend: str = 'decord',eval_: bool = True)

worldfoundry.core.load_frames_from_videofrom worldfoundry.core import load_frames_from_video

源码

简介

load_frames_from_video — Load selected RGB frames into a torch tensor using decord or OpenCV. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

video_pathstr | Path
indicesIterable[int]
video_decode_backendstr: 默认值: 'decord'
eval_bool: 默认值: True

`load_serialized`

func

def load_serialized(file: str | Path | IO[Any],file_format: str | None = None,encoding: str = 'utf-8',**kwargs: Any) -> Any

worldfoundry.core.load_serializedfrom worldfoundry.core import load_serialized

源码

简介

load_serialized — Load a structured object from a URI or file object. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Any。

参数

filestr | Path | IO[Any]
file_formatstr | None: 默认值: None
encodingstr: 默认值: 'utf-8'
kwargsAny

返回值: Any

`load_video_frames`

func

def load_video_frames(video_path: str | Path)

worldfoundry.core.load_video_framesfrom worldfoundry.core import load_video_frames

源码

简介

load_video_frames — Decode a local or remote video into a uint8 THWC numpy array. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

video_pathstr | Path

`local_model_root_path`

func

def local_model_root_path(env: Mapping[str, str] | None = None) -> Path

worldfoundry.core.io.paths.local_model_root_pathfrom worldfoundry.core.io.paths import local_model_root_path

源码

简介

local_model_root_path — Resolves the root directory containing local model weights, configs, and adapters. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

参数

envMapping[str, str] | None: 默认值: None

返回值: Path

`LowMemoryImageFolder`

cls

class LowMemoryImageFolder(folder, file_list = None)

worldfoundry.core.LowMemoryImageFolderfrom worldfoundry.core import LowMemoryImageFolder

源码

简介

LowMemoryImageFolder — Lazy image-folder reader that loads frames on demand. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

folder
file_list: 默认值: None

`LowMemoryVideo`

cls

class LowMemoryVideo(file_name: str)

worldfoundry.core.LowMemoryVideofrom worldfoundry.core import LowMemoryVideo

源码

简介

LowMemoryVideo — Lazy video reader that loads frames on demand. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

file_namestr

`materialize_hf_snapshot`

func

def materialize_hf_snapshot(repo_id_or_path: str | os.PathLike[str],revision: str | None = None,cache_dir: str | os.PathLike[str] | None = None,allow_patterns: str | Sequence[str] | None = None,ignore_patterns: str | Sequence[str] | None = None,required_files: Sequence[str] = (),local_files_only: bool | None = None,token: str | bool | None = None) -> Path

worldfoundry.core.materialize_hf_snapshotfrom worldfoundry.core import materialize_hf_snapshot

源码

简介

materialize_hf_snapshot — Return a local snapshot for either a path or Hugging Face repo id. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

源码 docstring

Return a local snapshot for either a path or Hugging Face repo id.

Remote downloads are serialized on rank zero and then reopened in local-only mode, avoiding partial-cache races in multi-GPU jobs.

参数

repo_id_or_pathstr | os.PathLike[str]
revisionstr | None: 默认值: None
cache_dirstr | os.PathLike[str] | None: 默认值: None
allow_patternsstr | Sequence[str] | None: 默认值: None
ignore_patternsstr | Sequence[str] | None: 默认值: None
required_filesSequence[str]: 默认值: ()
local_files_onlybool | None: 默认值: None
tokenstr | bool | None: 默认值: None

返回值: Path

`materialize_video_input`

func

def materialize_video_input(video_input,output_dir: Optional[str] = None,filename: str = 'input.mp4',fps: int = 24) -> str

worldfoundry.core.materialize_video_inputfrom worldfoundry.core import materialize_video_input

源码

简介

materialize_video_input — Return a local video path, materializing in-memory inputs when needed. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str。

参数

video_input
output_dirOptional[str]: 默认值: None
filenamestr: 默认值: 'input.mp4'
fpsint: 默认值: 24

返回值: str

`maybe_download_hf_repo_on_rank0`

func

def maybe_download_hf_repo_on_rank0(repo_id_or_path: str,revision: str | None = None,cache_dir: str | os.PathLike[str] | None = None,allow_patterns: str | Sequence[str] | None = None,ignore_patterns: str | Sequence[str] | None = None,token: str | bool | None = None) -> None

worldfoundry.core.maybe_download_hf_repo_on_rank0from worldfoundry.core import maybe_download_hf_repo_on_rank0

源码

简介

maybe_download_hf_repo_on_rank0 — Download a remote HF repo snapshot from rank 0 when downloads are allowed. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：None。

源码 docstring

Download a remote HF repo snapshot from rank 0 when downloads are allowed.

Local paths and explicit offline/local-only modes are no-ops. For remote repositories, rank 0 preloads the snapshot while other distributed ranks wait for its success/failure signal. A filesystem lock serializes independent processes that share the same HF cache directory.

参数

repo_id_or_pathstr
revisionstr | None: 默认值: None
cache_dirstr | os.PathLike[str] | None: 默认值: None
allow_patternsstr | Sequence[str] | None: 默认值: None
ignore_patternsstr | Sequence[str] | None: 默认值: None
tokenstr | bool | None: 默认值: None

返回值: None

`merge_video_audio`

func

def merge_video_audio(video_path: str, audio_path: str) -> None

worldfoundry.core.merge_video_audiofrom worldfoundry.core import merge_video_audio

源码

简介

merge_video_audio — Merge video and audio with ffmpeg; overwrite `video_path on success. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：None`。

参数

video_pathstr
audio_pathstr

返回值: None

`open_uri`

func

def open_uri(uri: str | os.PathLike[str],mode: str = 'rb',encoding: str = 'utf-8',**storage_options) -> Generator[BinaryIO | TextIO, None, None]

worldfoundry.core.open_urifrom worldfoundry.core import open_uri

源码

简介

open_uri — Open a local or remote URI. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Generator[BinaryIO | TextIO, None, None]。

源码 docstring

Open a local or remote URI.

`fsspec` is used when available. Without it, local files are supported for all modes and HTTP(S) URLs are supported for reads.

参数

uristr | os.PathLike[str]
modestr: 默认值: 'rb'
encodingstr: 默认值: 'utf-8'
storage_options

返回值: Generator[BinaryIO | TextIO, None, None]

`package_module_root`

func

def package_module_root(package: str) -> Path

worldfoundry.core.package_module_rootfrom worldfoundry.core import package_module_root

源码

简介

package_module_root — Resolve the source directory for an importable package. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

参数

packagestr

返回值: Path

`ParallelHelper`

cls

class ParallelHelper()

worldfoundry.core.ParallelHelperfrom worldfoundry.core import ParallelHelper

源码

简介

ParallelHelper — Distribute video tiles by cost and reconstruct their global ordering. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

源码 docstring

Distribute video tiles by cost and reconstruct their global ordering.

In non-distributed execution every tile remains local. With a process group, larger tiles are round-robin balanced across ranks and decoded frame tensors are gathered with explicit count, dtype, and shape metadata.

方法

smeth

split_tile_list(tile_numel_dict: OrderedDict[int, int],parallel_group: torch.distributed.ProcessGroup = None) -> List[int]

源码

简介

split_tile_list — Splits the given tile size into a list of sizes that each rank should handle.

源码 docstring

Splits the given tile size into a list of sizes that each rank should handle.

This method takes into account the number of ranks in a distributed setting. If the distributed environment is not initialized, it returns a list of integers from 0 to tile_size - 1, representing each tile index.

If the distributed environment is initialized, it calculates the base tile size for each rank and distributes any remaining tiles among the ranks.

参数

tile_numel_dictOrderedDict[int, int]: Dict of index and numel of tiles.
parallel_grouptorch.distributed.ProcessGroup: Distributed decoding group. Defaults to None.默认值: None

返回值: List[int] — List[int]: A list of tile indices assigned to the current rank. List[int]: A list of global tile indices.

smeth

gather_frames(frames: List[torch.Tensor],global_tile_idxs: List[int],parallel_group: torch.distributed.ProcessGroup = None) -> List[torch.Tensor]

源码

简介

gather_frames — Gathers frame data from all ranks in a distributed environment.

源码 docstring

Gathers frame data from all ranks in a distributed environment.

This method collects frames from all ranks and combines them into a single list. If the distributed environment is not initialized, it simply returns the input frames.

参数

framesList[torch.Tensor]: A list of frames (tensors) from the current rank.
global_tile_idxsList[int]: A list of global tile indices.
parallel_grouptorch.distributed.ProcessGroup: Distributed decoding group. Defaults to None.默认值: None

返回值: List[torch.Tensor] — List[torch.Tensor]: A list of frames (tensors) from all ranks.

smethindex_undot(index: int, loop_size: List[int]) -> List[int]源码

简介

index_undot — Converts a single index into a list of indices, representing the position in a multi-dimensional space.

源码 docstring

Converts a single index into a list of indices, representing the position in a multi-dimensional space.

This method takes an integer index and a list of loop sizes, and converts the index into a list of indices that correspond to the position in a multi-dimensional space.

参数

indexint: The single index to be converted.
loop_sizeList[int]: A list of integers representing the size of each dimension in the multi-dimensional space.

返回值: List[int] — List[int]: A list of integers representing the position in the multi-dimensional space.

smethindex_dot(index: List[int], loop_size: List[int]) -> int源码

简介

index_dot — Converts a list of indices into a single index, representing the position in a multi-dimensional space.

源码 docstring

Converts a list of indices into a single index, representing the position in a multi-dimensional space.

This method takes a list of indices and a list of loop sizes, and converts the list of indices into a single index that corresponds to the position in a multi-dimensional space.

参数

indexList[int]: A list of integers representing the position in the multi-dimensional space.
loop_sizeList[int]: A list of integers representing the size of each dimension in the multi-dimensional space.

返回值: int — int: A single integer representing the position in the multi-dimensional space.

`parse_uri_scheme`

func

def parse_uri_scheme(uri: str | os.PathLike[str]) -> str

worldfoundry.core.parse_uri_schemefrom worldfoundry.core import parse_uri_scheme

源码

简介

parse_uri_scheme — Return the lowercase URI scheme, using `file for local paths. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str`。

参数

uristr | os.PathLike[str]

返回值: str

`project_root`

func

def project_root(start: str | Path | None = None) -> Path

worldfoundry.core.io.paths.project_rootfrom worldfoundry.core.io.paths import project_root

源码

简介

project_root — Walks upward from a starting path to locate the root repository containing pyproject.toml. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

源码 docstring

Walks upward from a starting path to locate the root repository containing pyproject.toml.

This helper provides robust local development support, falling back to a package-relative root if executed from a system-wide python site-packages deployment.

参数

startstr | Path | None: File or directory from which to search upward. `None` starts from this module's installed source path.默认值: None

返回值: Path — First ancestor containing `pyproject.toml`, or the package-relative fallback when no repository marker is found.

`read_binary_uri`

func

def read_binary_uri(uri: str | os.PathLike[str], **storage_options) -> bytes

worldfoundry.core.read_binary_urifrom worldfoundry.core import read_binary_uri

源码

简介

通过 Core 存储辅助，从 URI 读取全部字节。

参数

uristr | os.PathLike[str]
storage_options

返回值: bytes

`read_image_as_video_tensor`

func

def read_image_as_video_tensor(image_path: str | Path,resolution: Iterable[int],num_video_frames: int,resize: bool = True)

worldfoundry.core.read_image_as_video_tensorfrom worldfoundry.core import read_image_as_video_tensor

源码

简介

read_image_as_video_tensor — Load an image and materialize a `[1,C,T,H,W]` uint8 conditioning video tensor. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

image_pathstr | Path
resolutionIterable[int]
num_video_framesint
resizebool: 默认值: True

`read_text_uri`

func

def read_text_uri(uri: str | os.PathLike[str],encoding: str = 'utf-8',**storage_options) -> str

worldfoundry.core.read_text_urifrom worldfoundry.core import read_text_uri

源码

简介

通过 Core 存储辅助，从 URI 读取全部文本。

参数

uristr | os.PathLike[str]
encodingstr: 默认值: 'utf-8'
storage_options

返回值: str

`read_video`

func

def read_video(video_path: str | Path, , return_metadata: bool = True)

worldfoundry.core.read_videofrom worldfoundry.core import read_video

源码

简介

read_video — Decode a video into frames and optional metadata. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

video_pathstr | Path
return_metadatabool: 默认值: True

`resize_video_tensor_to_resolution`

func

def resize_video_tensor_to_resolution(video_tensor, resolution: Iterable[int])

worldfoundry.core.resize_video_tensor_to_resolutionfrom worldfoundry.core import resize_video_tensor_to_resolution

源码

简介

resize_video_tensor_to_resolution — Resize and center-crop a `[T,C,H,W] tensor to (target_h, target_w)`. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

video_tensor
resolutionIterable[int]

`resolve_data_path`

func

def resolve_data_path(parts: str | Path) -> Path

worldfoundry.core.io.paths.resolve_data_pathfrom worldfoundry.core.io.paths import resolve_data_path

源码

简介

resolve_data_path — Resolves a subpath under the internal worldfoundry/data asset directory. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

参数

partsstr | Path

返回值: Path

`resolve_hf_path`

func

def resolve_hf_path(path: str | PathLike[str] | None) -> str | Any

worldfoundry.core.resolve_hf_pathfrom worldfoundry.core import resolve_hf_path

源码

简介

resolve_hf_path — Resolve a possibly `hf://-prefixed path to a local filesystem path. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str | Any`。

源码 docstring

Resolve a possibly `hf://`-prefixed path to a local filesystem path.

Accepts either:

* a local path (returned unchanged if it exists), or * `hf://<owner>/<repo>[/<subpath>]` — resolves an already materialized WorldFoundry-local snapshot and returns the requested file or directory.

Runtime I/O is deliberately offline. Repository acquisition belongs to the explicit preparation workflow, never to model inference.

参数

pathstr | PathLike[str] | None

返回值: str | Any

`resolve_hf_snapshot_path`

func

def resolve_hf_snapshot_path(value: str | os.PathLike[str],required_files: Sequence[str] = (),local_files_only_env: str = 'WORLDFOUNDRY_HF_LOCAL_FILES_ONLY',local_files_only: bool | None = None) -> Path

worldfoundry.core.resolve_hf_snapshot_pathfrom worldfoundry.core import resolve_hf_snapshot_path

源码

简介

resolve_hf_snapshot_path — Resolve a repo id, HF cache repo root, or local path to a usable snapshot. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

参数

valuestr | os.PathLike[str]
required_filesSequence[str]: 默认值: ()
local_files_only_envstr: 默认值: 'WORLDFOUNDRY_HF_LOCAL_FILES_ONLY'
local_files_onlybool | None: 默认值: None

返回值: Path

`resolve_local_checkpoint_file`

func

def resolve_local_checkpoint_file(model_id_or_path: str | Path,filename: str,env: Mapping[str, str] | None = None) -> Path

worldfoundry.core.resolve_local_checkpoint_filefrom worldfoundry.core import resolve_local_checkpoint_file

源码

简介

resolve_local_checkpoint_file — Resolve one checkpoint file from explicit or WorldFoundry-local storage. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

源码 docstring

Resolve one checkpoint file from explicit or WorldFoundry-local storage.

An explicit file path is accepted directly. Directory paths and repository identifiers are resolved through :func:resolve_local_hf_model_path, so incomplete hfd transfers are rejected and this function never contacts a model hub.

参数

model_id_or_pathstr | Path
filenamestr
envMapping[str, str] | None: 默认值: None

返回值: Path

`resolve_local_hf_model_path`

func

def resolve_local_hf_model_path(model_id_or_path: str | Path,required_files: Sequence[str] = (),env: Mapping[str, str] | None = None) -> Path

worldfoundry.core.resolve_local_hf_model_pathfrom worldfoundry.core import resolve_local_hf_model_path

源码

简介

resolve_local_hf_model_path — Resolve a Hugging Face model strictly from WorldFoundry-local storage. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

源码 docstring

Resolve a Hugging Face model strictly from WorldFoundry-local storage.

The resolver understands direct export directories, hfd-style names such as `owner--repo, and the Hub cache snapshots/<revision>` layout. It never contacts the network, which keeps model initialization deterministic and works in inference environments that do not install PyTorch.

参数

model_id_or_pathstr | Path
required_filesSequence[str]: 默认值: ()
envMapping[str, str] | None: 默认值: None

返回值: Path

`resolve_worldfoundry_path`

func

def resolve_worldfoundry_path(value: str | Path,env: Mapping[str, str] | None = None) -> Path

worldfoundry.core.io.paths.resolve_worldfoundry_pathfrom worldfoundry.core.io.paths import resolve_worldfoundry_path

源码

简介

resolve_worldfoundry_path — Expands structural WorldFoundry path tokens (e.g. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：Path。

源码 docstring

Expands structural WorldFoundry path tokens (e.g. ${WORLDFOUNDRY_CKPT_DIR}) and home markers (~).

Performs precise regex-free variable mapping replacement while preserving subfolder hierarchies.

参数

valuestr | Path: Path containing optional `$NAME or ${NAME}` tokens.
envMapping[str, str] | None: Environment mapping used to build/override WorldFoundry tokens.默认值: None

返回值: Path — Expanded `Path`. The target is not created or required to exist.

`save_frames`

func

def save_frames(frames, save_path)

worldfoundry.core.save_framesfrom worldfoundry.core import save_frames

源码

简介

save_frames — Write a sequence of PIL frames to numbered PNG files. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

frames
save_path

`save_image_or_video_tensor`

func

def save_image_or_video_tensor(tensor,save_path,fps: int = 24,quality: int | None = None,ffmpeg_params: list[str] | None = None,value_range: str | tuple[float, float] = 'auto',image_format: str = 'JPEG',video_format: str = 'mp4',**kwargs) -> str | None

worldfoundry.core.save_image_or_video_tensorfrom worldfoundry.core import save_image_or_video_tensor

源码

简介

save_image_or_video_tensor — Save a normalized `[C,T,H,W] or [B,C,T,H,W] tensor as image/video. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：str | None`。

源码 docstring

Save a normalized `[C,T,H,W] or [B,C,T,H,W]` tensor as image/video.

A single-frame tensor is saved as an image; multi-frame tensors are saved as videos. Local paths and URI-like targets supported by `worldfoundry.core.io` storage helpers are both accepted.

参数

tensor
save_path
fpsint: 默认值: 24
qualityint | None: 默认值: None
ffmpeg_paramslist[str] | None: 默认值: None
value_rangestr | tuple[float, float]: 默认值: 'auto'
image_formatstr: 默认值: 'JPEG'
video_formatstr: 默认值: 'mp4'
kwargs

返回值: str | None

`save_video`

func

def save_video(frames,save_path,fps,quality = 9,ffmpeg_params = None)

worldfoundry.core.save_videofrom worldfoundry.core import save_video

源码

简介

save_video — Write a sequence of PIL frames to a video file. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

frames
save_path
fps
quality: 默认值: 9
ffmpeg_params: 默认值: None

`save_video_frames`

func

def save_video_frames(video_frames,output_path: str | Path,fps: int = 16,**kwargs) -> None

worldfoundry.core.save_video_framesfrom worldfoundry.core import save_video_frames

源码

简介

save_video_frames — Write a THWC uint8 frame array/list to a video path or URI. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：None。

参数

video_frames
output_pathstr | Path
fpsint: 默认值: 16
kwargs

返回值: None

`save_video_with_audio`

func

def save_video_with_audio(frames,save_path,audio_path,fps = 16,quality = 9,ffmpeg_params = None)

worldfoundry.core.save_video_with_audiofrom worldfoundry.core import save_video_with_audio

源码

简介

save_video_with_audio — Write PIL frames, then mux an external audio track with ffmpeg. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

frames: Iterable of PIL images.
save_path: Destination video path, replaced after muxing.
audio_path: Existing audio file.
fps: Output frame rate.默认值: 16
quality: ImageIO encoder quality.默认值: 9
ffmpeg_params: Optional ImageIO ffmpeg arguments.默认值: None

说明

`merge_video_audio` reports ffmpeg failures and removes its temporary file. Use lower-level helpers when the caller needs structured process error handling.

`search_for_images`

func

def search_for_images(folder)

worldfoundry.core.search_for_imagesfrom worldfoundry.core import search_for_images

源码

简介

search_for_images — Return naturally sorted JPG/PNG paths from one image-sequence folder. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

源码 docstring

Return naturally sorted JPG/PNG paths from one image-sequence folder.

Embedded digit runs are compared numerically, so `2.png sorts before 10.png`. The search is non-recursive.

参数

folder

`TileProcessor`

cls

class TileProcessor(encode_fn,decode_fn,tile_sample_min_height: int = 256,tile_sample_min_width: int = 256,tile_sample_min_length: int = 16,spatial_downsample_factor: int = 8,temporal_downsample_factor: int = 1,spatial_tile_overlap_factor: float = 0.25,temporal_tile_overlap_factor: float = 0,sr_ratio = 1,first_frame_as_image: bool = False,parallel_group: torch.distributed.ProcessGroup = None)

worldfoundry.core.TileProcessorfrom worldfoundry.core import TileProcessor

源码

简介

TileProcessor — Encode or decode large videos through overlapping spatiotemporal tiles. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

源码 docstring

Encode or decode large videos through overlapping spatiotemporal tiles.

The processor derives latent tile sizes from codec downsample factors, schedules tiles locally or across `parallel_group, and linearly blends temporal/vertical/horizontal overlaps to suppress seams. encode_fn and decode_fn` remain model-owned callables.

参数

encode_fn: The encoding function used for tile sampling.
decode_fn: The decoding function used for tile reconstruction.
tile_sample_min_heightint: 默认值: 256
tile_sample_min_widthint: 默认值: 256
tile_sample_min_lengthint: The minimum length of the sampled tiles. Defaults to 16.默认值: 16
spatial_downsample_factorint: The actual spataial downsample factor of given encode_fn. Defaults to 8.默认值: 8
temporal_downsample_factorint: The actual temporal downsample factor of the latent space tiles. Defaults to 1.默认值: 1
spatial_tile_overlap_factorfloat: 默认值: 0.25
temporal_tile_overlap_factorfloat: 默认值: 0
sr_ratio: 默认值: 1
first_frame_as_imagebool: 默认值: False
parallel_grouptorch.distributed.ProcessGroup: Distributed decoding group. Defaults to None.默认值: None

方法

methblend_t(a: torch.Tensor,b: torch.Tensor,blend_extent: int) -> torch.Tensor源码

简介

该类型上的公开 method。

参数

atorch.Tensor
btorch.Tensor
blend_extentint

返回值: torch.Tensor

methblend_v(a: torch.Tensor,b: torch.Tensor,blend_extent: int) -> torch.Tensor源码

简介

该类型上的公开 method。

参数

atorch.Tensor
btorch.Tensor
blend_extentint

返回值: torch.Tensor

methblend_h(a: torch.Tensor,b: torch.Tensor,blend_extent: int) -> torch.Tensor源码

简介

该类型上的公开 method。

参数

atorch.Tensor
btorch.Tensor
blend_extentint

返回值: torch.Tensor

methtiled_encode(x: torch.FloatTensor, verbose: bool = False)源码

简介

该类型上的公开 method。

参数

xtorch.FloatTensor
verbosebool: 默认值: False

methtiled_decode(z: torch.FloatTensor, verbose: bool = False)源码

简介

该类型上的公开 method。

参数

ztorch.FloatTensor
verbosebool: 默认值: False

`video_tensor_to_uint8_frames`

func

def video_tensor_to_uint8_frames(video_tensor,value_range: str | tuple[float, float] = 'auto') -> 'object'

worldfoundry.core.video_tensor_to_uint8_framesfrom worldfoundry.core import video_tensor_to_uint8_frames

源码

简介

video_tensor_to_uint8_frames — Convert a normalized torch video tensor to uint8 THWC frames. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：'object'。

源码 docstring

Convert a normalized torch video tensor to uint8 THWC frames.

`value_range="auto" treats tensors with negative values as [-1, 1] and non-negative floating tensors as [0, 1]`.

参数

video_tensor
value_rangestr | tuple[float, float]: 默认值: 'auto'

返回值: 'object'

`VideoData`

cls

class VideoData(video_file = None,image_folder = None,height = None,width = None,**kwargs)

worldfoundry.core.VideoDatafrom worldfoundry.core import VideoData

源码

简介

VideoData — Lazy video or image-folder dataset with optional resize. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。

参数

video_file: 默认值: None
image_folder: 默认值: None
height: 默认值: None
width: 默认值: None
kwargs

简介

该类型上的公开 method。

methsave_images(folder)源码

简介

该类型上的公开 method。

参数

folder

`worldfoundry_path_tokens`

func

def worldfoundry_path_tokens(env: Mapping[str, str] | None = None) -> dict[str, str]

worldfoundry.core.io.paths.worldfoundry_path_tokensfrom worldfoundry.core.io.paths import worldfoundry_path_tokens

源码

简介

worldfoundry_path_tokens — Generates the dictionary of logical path-token replacements used across the system. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：dict[str, str]。

源码 docstring

Generates the dictionary of logical path-token replacements used across the system.

Builds dynamic mappings for artifact, checkpoint, data, conda, and repo paths. Prioritizes explicit environment overrides (such as WORLDFOUNDRY_HOME or WORLDFOUNDRY_CACHE_DIR) and falls back to user-home cache directories when variables are unset.

参数

envMapping[str, str] | None: Environment mapping to resolve instead of `os.environ`. Passing a mapping makes resolution deterministic in tests.默认值: None

返回值: dict[str, str] — Token name to expanded path-string mapping.

`write_binary_uri`

func

def write_binary_uri(uri: str | os.PathLike[str],data: bytes | bytearray | memoryview | io.BytesIO,**storage_options) -> None

worldfoundry.core.write_binary_urifrom worldfoundry.core import write_binary_uri

源码

简介

write_binary_uri — Write bytes to a URI, creating local parent directories as needed. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：None。

参数

uristr | os.PathLike[str]
databytes | bytearray | memoryview | io.BytesIO
storage_options

返回值: None

`write_text_uri`

func

def write_text_uri(uri: str | os.PathLike[str],data: str,encoding: str = 'utf-8',**storage_options) -> None

worldfoundry.core.write_text_urifrom worldfoundry.core import write_text_uri

源码

简介

write_text_uri — Write text to a URI, creating local parent directories as needed. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：None。

参数

uristr | os.PathLike[str]
datastr
encodingstr: 默认值: 'utf-8'
storage_options

返回值: None

`write_video`

func

def write_video(video_frames,output_path: str | Path,fps: int = 16,quality: int | None = None,format: str | None = None,**kwargs) -> None

worldfoundry.core.write_videofrom worldfoundry.core import write_video

源码

简介

write_video — Write a THWC video array/list to a local path or remote URI. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：None。

参数

video_frames
output_pathstr | Path
fpsint: 默认值: 16
qualityint | None: 默认值: None
formatstr | None: 默认值: None
kwargs

返回值: None

`write_video_torchvision`

func

def write_video_torchvision(filename: str | Path,video_array: Any,fps: float,*args: Any,**kwargs: Any) -> None

worldfoundry.core.write_video_torchvisionfrom worldfoundry.core import write_video_torchvision

源码

简介

write_video_torchvision — Write RGB video frames with a `torchvision.io.write_video-compatible signature. 属于 Core I/O 与媒体（路径、URI、图像、视频、序列化）。标注返回类型：None`。

参数

filenamestr | Path
video_arrayAny
fpsfloat
argsAny
kwargsAny

返回值: None