OverviewFilesCommitsIssuesDocsReleasesPeers

blender-bridge / blender_addon / procedural_baker.py

215 lines · python ref:
"""Non-blocking procedural texture baker using background Blender subprocess.

Bakes procedural materials (Noise, Voronoi, etc.) to images by spawning
`blender --background --python bake_script.py`, keeping the main UI responsive.
"""

import os
import shutil
import subprocess
import tempfile

import bpy


class ProceduralBaker:
    """Non-blocking procedural texture baker using background subprocess."""

    def __init__(self):
        self._queue = []           # list of {obj_name, mat_name, mat_key, size}
        self._process = None       # Active subprocess.Popen
        self._current = None       # Currently baking item dict
        self._results = {}         # mat_key -> baked image name
        self._blend_path = None    # Path to cached .blend save
        self._temp_dir = None      # Temp directory for bake files
        self._started = False      # Whether baking has been triggered

    @property
    def pending_count(self):
        """Number of materials waiting to be baked."""
        return len(self._queue)

    @property
    def is_baking(self):
        """Whether a bake subprocess is currently running."""
        return self._process is not None

    def queue(self, obj_name, mat_name, size=1024):
        """Add a material to the bake queue. Non-blocking."""
        mat_key = f"{obj_name}:{mat_name}"
        if mat_key in self._results:
            return  # Already baked
        for item in self._queue:
            if item["mat_key"] == mat_key:
                return  # Already queued
        if self._current and self._current["mat_key"] == mat_key:
            return  # Currently baking
        self._queue.append({
            "obj_name": obj_name,
            "mat_name": mat_name,
            "mat_key": mat_key,
            "size": size,
        })

    def invalidate(self, mat_key=None):
        """Mark material(s) for re-baking (parameter changed)."""
        if mat_key:
            self._results.pop(mat_key, None)
        else:
            self._results.clear()

    def start(self):
        """Begin processing the bake queue. Saves .blend then starts first subprocess."""
        if not self._queue:
            return
        self._started = True
        self._save_blend()
        self._start_next()

    def poll(self):
        """Check for completed bakes. Returns list of (mat_key, image_name)."""
        completed = []
        if self._process is not None:
            ret = self._process.poll()
            if ret is not None:
                if ret == 0 and self._current:
                    img_name = self._load_result(self._current)
                    if img_name:
                        completed.append((self._current["mat_key"], img_name))
                elif self._current:
                    print(f"BlenderBridge: Bake failed for '{self._current['mat_name']}' (exit {ret})")
                    if self._process.stderr:
                        err = self._process.stderr.read()
                        if err:
                            print(f"  stderr: {err.decode('utf-8', errors='replace')[:500]}")
                self._process = None
                self._current = None

        # Start next if idle and started
        if self._process is None and self._queue and self._started:
            self._start_next()

        return completed

    def cleanup(self):
        """Kill subprocess and clean temp files."""
        if self._process:
            self._process.kill()
            self._process = None
        if self._temp_dir and os.path.exists(self._temp_dir):
            shutil.rmtree(self._temp_dir, ignore_errors=True)
        self._temp_dir = None
        self._blend_path = None

    def reset(self):
        """Full reset (disconnect)."""
        self.cleanup()
        self._queue.clear()
        self._results.clear()
        self._current = None
        self._started = False

    # -- Internal --

    def _save_blend(self):
        """Save current .blend to temp for subprocess use."""
        if self._temp_dir is None:
            self._temp_dir = tempfile.mkdtemp(prefix="bb_bake_")
        self._blend_path = os.path.join(self._temp_dir, "source.blend")
        bpy.ops.wm.save_as_mainfile(filepath=self._blend_path, copy=True)

    def _start_next(self):
        """Start baking next queued material."""
        if not self._queue:
            self._started = False
            return
        item = self._queue.pop(0)

        if self._blend_path is None or not os.path.exists(self._blend_path):
            self._save_blend()

        output_path = os.path.join(self._temp_dir, f"bake_{item['mat_name']}.png")
        item["output_path"] = output_path

        script = _BAKE_SCRIPT.format(
            obj_name=item["obj_name"],
            mat_name=item["mat_name"],
            size=item["size"],
            output=output_path.replace("\\", "\\\\"),
        )
        script_path = os.path.join(self._temp_dir, "bake_script.py")
        with open(script_path, "w") as f:
            f.write(script)

        self._process = subprocess.Popen(
            [bpy.app.binary_path, "--background", self._blend_path,
             "--python", script_path],
            stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )
        self._current = item
        print(f"BlenderBridge: Baking '{item['mat_name']}' in background ({item['size']}x{item['size']})")

    def _load_result(self, item):
        """Load baked image from disk into bpy.data.images."""
        output = item["output_path"]
        if not os.path.exists(output):
            print(f"BlenderBridge: Bake output not found: {output}")
            return None
        img_name = f"_bb_bake_{item['mat_name']}"
        if img_name in bpy.data.images:
            bpy.data.images.remove(bpy.data.images[img_name])
        img = bpy.data.images.load(output)
        img.name = img_name
        self._results[item["mat_key"]] = img_name
        print(f"BlenderBridge: Loaded baked texture '{img_name}'")
        return img_name


_BAKE_SCRIPT = '''
import bpy, sys

obj_name = "{obj_name}"
mat_name = "{mat_name}"
size = {size}
output = "{output}"

scene = bpy.context.scene
scene.render.engine = "CYCLES"
scene.cycles.samples = 1
scene.cycles.device = "CPU"

obj = bpy.data.objects.get(obj_name)
mat = bpy.data.materials.get(mat_name)

if obj is None or mat is None:
    print(f"Object '{{obj_name}}' or material '{{mat_name}}' not found")
    sys.exit(1)

if not obj.data.uv_layers:
    print(f"No UV layers on '{{obj_name}}'")
    sys.exit(1)

# Create bake target image
bake_img = bpy.data.images.new("_bake_target", size, size, alpha=True)

# Add temp bake node
tree = mat.node_tree
bake_node = tree.nodes.new("ShaderNodeTexImage")
bake_node.image = bake_img
tree.nodes.active = bake_node

bpy.ops.object.select_all(action="DESELECT")
obj.select_set(True)
bpy.context.view_layer.objects.active = obj

try:
    bpy.ops.object.bake(type="DIFFUSE", pass_filter={{"COLOR"}}, margin=2)
    bake_img.filepath_raw = output
    bake_img.file_format = "PNG"
    bake_img.save()
    print(f"Bake complete: {{output}}")
except Exception as e:
    print(f"Bake error: {{e}}")
    sys.exit(1)
'''