"""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)
'''