Forge Asset Pipeline

forge-asset-pipeline

Scaffold a Python CLI pipeline that discovers plugins, scans source files, fingerprints them by content hash, and reports what needs processing.

When to use this skill

• Setting up an asset processing pipeline for a game or graphics project
• Building any file-processing CLI that needs incremental builds
• Creating a plugin-based tool where new file types are added without modifying core code
• When someone asks for "a build system", "asset pipeline", "file processor", or "incremental build tool"

Key concepts

Architecture

pipeline.toml ──> CLI (__main__.py)
                   │
                   ├──> Plugin Discovery (plugins/*.py)
                   │        register by name + extension
                   │
                   └──> Scanner (scanner.py)
                            │
                            ├── walk source directory
                            ├── fingerprint (SHA-256)
                            ├── compare against cache
                            └── classify: NEW / CHANGED / UNCHANGED

Correct order

1. Create the project structure (pyproject.toml, pipeline/ package)
2. Implement config.py — TOML loader with typed dataclass
3. Implement plugin.py — base class, registry, file-based discovery
4. Implement scanner.py — fingerprinting, cache, scan function
5. Implement __main__.py — CLI that ties them together
6. Create example plugins in plugins/
7. Create pipeline.toml with sensible defaults

Project structure template

<project>/
  pipeline.toml              # TOML configuration
  pyproject.toml             # Python package definition
  pipeline/
    __init__.py
    __main__.py              # CLI entry point (argparse)
    config.py                # TOML config loader
    plugin.py                # AssetPlugin base, PluginRegistry, discovery
    scanner.py               # File scanning and fingerprinting
  plugins/
    <type>.py                # One file per asset type
  tests/
    test_config.py
    test_plugin.py
    test_scanner.py

Ready-to-use templates

config.py — core pattern

import sys
from dataclasses import dataclass, field
from pathlib import Path

if sys.version_info >= (3, 11):
    import tomllib
else:
    try:
        import tomllib
    except ModuleNotFoundError:
        import tomli as tomllib

@dataclass
class PipelineConfig:
    source_dir: Path          # where raw source files live
    output_dir: Path          # where processed files go
    cache_dir: Path           # fingerprint cache location
    plugin_settings: dict[str, dict] = field(default_factory=dict)
    raw: dict = field(default_factory=dict, repr=False)

class ConfigError(Exception):
    pass

def load_config(path: Path) -> PipelineConfig:
    if not path.exists():
        raise ConfigError(f"Configuration file not found: {path}")
    try:
        with open(path, "rb") as f:
            raw = tomllib.load(f)
    except tomllib.TOMLDecodeError as exc:
        raise ConfigError(f"Invalid TOML in {path}: {exc}") from exc

    pipeline = raw.get("pipeline", {})
    plugin_settings = {k: v for k, v in raw.items()
                       if k != "pipeline" and isinstance(v, dict)}
    return PipelineConfig(
        source_dir=Path(pipeline.get("source_dir", "assets/raw")),
        output_dir=Path(pipeline.get("output_dir", "assets/processed")),
        cache_dir=Path(pipeline.get("cache_dir", ".forge-cache")),
        plugin_settings=plugin_settings,
        raw=raw,
    )

plugin.py — core pattern

import importlib
import importlib.util
from dataclasses import dataclass, field
from pathlib import Path

@dataclass
class AssetResult:
    source: Path              # original source file
    output: Path              # processed output file
    metadata: dict = field(default_factory=dict)

class AssetPlugin:
    name: str = ""            # human-readable identifier
    extensions: list[str] = []  # e.g. [".png", ".jpg"]

    def process(self, source: Path, output_dir: Path, settings: dict) -> AssetResult:
        raise NotImplementedError

class PluginRegistry:
    def __init__(self):
        self._plugins: dict[str, AssetPlugin] = {}
        self._ext_map: dict[str, AssetPlugin] = {}

    def register(self, plugin: AssetPlugin) -> None:
        # Validate: non-empty name, no duplicate name, no duplicate extensions
        self._plugins[plugin.name] = plugin
        for ext in plugin.extensions:
            self._ext_map[ext.lower()] = plugin

    def get_by_extension(self, ext: str) -> AssetPlugin | None:
        return self._ext_map.get(ext.lower())

    def discover(self, plugins_dir: Path) -> int:
        count = 0
        for py_file in sorted(plugins_dir.glob("*.py")):
            if py_file.name.startswith("_"):
                continue
            spec = importlib.util.spec_from_file_location(
                f"plugins.{py_file.stem}", py_file)
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            for name in dir(module):
                attr = getattr(module, name)
                if (isinstance(attr, type) and issubclass(attr, AssetPlugin)
                        and attr is not AssetPlugin and attr.name):
                    self.register(attr())
                    count += 1
        return count

scanner.py — core pattern

import hashlib
import json
from dataclasses import dataclass
from enum import Enum, auto
from pathlib import Path

HASH_CHUNK_SIZE = 65536  # 64 KiB read chunks

class FileStatus(Enum):
    NEW = auto()        # not in cache
    CHANGED = auto()    # hash differs from cache
    UNCHANGED = auto()  # hash matches cache

@dataclass
class ScannedFile:
    path: Path            # absolute path
    relative: Path        # relative to source_dir
    extension: str        # lowercase, with dot
    fingerprint: str      # SHA-256 hex digest
    status: FileStatus

def fingerprint_file(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        while chunk := f.read(HASH_CHUNK_SIZE):
            h.update(chunk)
    return h.hexdigest()

class FingerprintCache:
    def __init__(self, path: Path):
        self._path = path
        self._data = json.loads(path.read_text()) if path.exists() else {}

    def get(self, rel: Path) -> str | None:
        return self._data.get(rel.as_posix())

    def set(self, rel: Path, fp: str) -> None:
        self._data[rel.as_posix()] = fp

    def save(self) -> None:
        self._path.parent.mkdir(parents=True, exist_ok=True)
        self._path.write_text(json.dumps(self._data, indent=2, sort_keys=True))

def scan(source_dir, extensions, cache):
    results = []
    for path in sorted(source_dir.rglob("*")):
        if not path.is_file() or path.suffix.lower() not in extensions:
            continue
        rel = path.relative_to(source_dir)
        fp = fingerprint_file(path)
        cached = cache.get(rel)
        status = (FileStatus.NEW if cached is None
                  else FileStatus.CHANGED if cached != fp
                  else FileStatus.UNCHANGED)
        results.append(ScannedFile(path, rel, path.suffix.lower(), fp, status))
    return results

Writing a plugin

# plugins/texture.py
from pipeline.plugin import AssetPlugin, AssetResult
from pathlib import Path

class TexturePlugin(AssetPlugin):
    name = "texture"
    extensions = [".png", ".jpg", ".tga", ".bmp"]

    def process(self, source: Path, output_dir: Path, settings: dict) -> AssetResult:
        output = output_dir / source.name
        # ... actual processing here ...
        return AssetResult(source=source, output=output)

pipeline.toml

[pipeline]
source_dir = "assets/raw"
output_dir = "assets/processed"
cache_dir  = ".forge-cache"

[texture]
max_size = 2048
generate_mipmaps = true

[mesh]
deduplicate = true

bundler.py — bundle packing and reading

from pathlib import Path
from pipeline.bundler import BundleWriter, BundleReader, create_bundle

# Write a bundle
writer = BundleWriter(Path("game.forgepak"))
writer.add("textures/brick.png", brick_bytes)
writer.add_file(Path("hero.fmesh"), "meshes/hero.fmesh",
                dependencies=["textures/brick.png"])
manifest = writer.finalize()

# Read a bundle (random access)
with BundleReader(Path("game.forgepak")) as reader:
    data = reader.read("textures/brick.png")

# Bundle an entire directory
manifest = create_bundle(
    output_dir=Path("assets/processed"),
    bundle_path=Path("assets/bundles/game.forgepak"),
)

CLI bundling

forge-pipeline bundle                              # bundle all processed assets
forge-pipeline bundle -o game.forgepak --level 9   # custom output, higher compression
forge-pipeline info game.forgepak                  # inspect bundle contents

Common mistakes

Reference

Based on Asset Lesson 01 — Pipeline Scaffold and Asset Lesson 05 — Asset Bundles.