Core foundations

Foundational Core APIs are small, but they define conventions shared by many larger systems: normalized identifiers, deterministic registries, environment flags, exact division, common list conversion, image materialization/composition, random seeding, and safety interfaces.

Typed registry example

from worldfoundry.core import TypedRegistry

metrics = TypedRegistry()
metrics.register(
    "temporal-consistency",
    object(),
    aliases=("temporal", "tc"),
    metadata={"direction": "higher-is-better"},
)

assert metrics.get("TC") is metrics.get("temporal-consistency")
assert metrics.keys() == ("temporal-consistency",)

Keys and aliases use stripped, case-folded lookup, while the original public key remains in RegistryItem. Registration rejects collisions instead of silently overwriting them. Enumeration is deterministic, which makes generated manifests and tests stable.

Guardrail composition example

from worldfoundry.core.safety import GuardrailRunner

class RejectEmptyPrompt:
    def is_safe(self, value):
        prompt = str(value).strip()
        return (bool(prompt), "prompt is empty" if not prompt else "")

runner = GuardrailRunner(
    safety_models=[RejectEmptyPrompt()],
    generic_block_msg="Request rejected",
)

assert runner.run_safety_check("walk through a snowy forest")[0] is True
assert runner.run_safety_check("") == (False, "Request rejected")

ContentSafetyGuardrail and PostprocessingGuardrail are structural protocols: implementations only need the documented method. GuardrailRunner stops at the first unsafe classifier, then applies postprocessors sequentially when requested. An empty classifier list returns safe with a warning; production policy should decide explicitly whether that fail-open behavior is acceptable.

The safety package imports NumPy and rank-aware Loguru logging from the model runtime environment. The registry example itself uses only the import-light top-level Core facade.

Utility boundaries

env_is_true recognizes a bounded truthy vocabulary; it is not a general parser. divide asserts exact divisibility and is intended for shape/group invariants. as_list normalizes a scalar or sequence at API boundaries. set_random_seed coordinates common random backends but does not by itself guarantee deterministic CUDA kernels.

Image helpers accept paths, PIL images, arrays, or tensors where documented. compose_horizontal_views and split_horizontal_views preserve a named view ordering; keep that ordering with the resulting artifact instead of relying on a later reader to guess it.

Complete reference

The blocks below are the generated signatures for this category. Use the on-page symbol index to jump; source links open the defining implementation behind each lazy export.

def as_list(value,none_as_empty: bool = True) -> list

def batched_image_features(images: torch.Tensor,encoder: Callable[[torch.Tensor], torch.Tensor],device: torch.device | str | int,batch_size: int | None = None,default_batch_size: int = 32,scope: str = 'image_features',normalize: bool = False,output_device: torch.device | str | int | None = 'cpu',batch_size_buckets: Sequence[int] | None = None) -> torch.Tensor

def compose_horizontal_views(images,target_size: tuple[int, int] | None = None) -> Image.Image

class ContentSafetyGuardrail(Protocol)

def divide(numerator: int,denominator: int) -> int

class DuplicateRegistryKeyError()

def env_is_true(env_name: str) -> bool

class GuardrailRunner(safety_models: Sequence[ContentSafetyGuardrail] | None = None,generic_block_msg: str = '',generic_safe_msg: str = '',postprocessors: Sequence[PostprocessingGuardrail] | None = None)

def load_pil_image(image_input,first_sequence_item: bool = True) -> Image.Image

def materialize_image_input(image_input,output_dir: str | os.PathLike[str],filename: str = 'input.png') -> str

def mean_flat(tensor: torch.Tensor) -> torch.Tensor

def mean_pairwise_cosine_distance(features: Sequence[torch.Tensor]) -> float

def normalize_action_values(values: Any,statistics: Mapping[str, Any],mode: str = 'min_max',clip: float | None = None) -> np.ndarray

def normalize_registry_key(value: str,field_name: str = 'registry key') -> str

class PostprocessingGuardrail(Protocol)

class PromptProcessor()

class RegistryError()

class RegistryItem(key: str,value: ItemT,aliases: tuple[str, ...] = (),metadata: Mapping[str, object] = <dict factory>)

def resize_and_letterbox(image,target_width: int,target_height: int,fill: tuple[int, int, int] = (255, 255, 255)) -> Image.Image

def select_modality_statistics(dataset_statistics: Mapping[str, Any],modality: str = 'action',key: str | None = None) -> tuple[str | None, Mapping[str, Any]]

def set_random_seed(seed: int,by_rank: bool = False) -> int

def split_horizontal_views(image,num_views: int,target_size: tuple[int, int] | None = None) -> list[Image.Image]

def stack_or_pad_tensors(values: Sequence[Tensor],padding_value: float | int | bool | None = None) -> Tensor

class TypedRegistry(items: Iterable[RegistryItem[ItemT]] = ())

class UnknownRegistryKeyError()

def unnormalize_action_values(normalized_values: Any,statistics: Mapping[str, Any],mode: str = 'min_max') -> np.ndarray