"""Blender-Godot Bridge addon.
Provides real-time bidirectional scene sync between Blender and Godot
via POSIX shared memory and Unix Domain Socket signaling.
Usage:
1. In Godot: add a BlenderBridge node, call start_server() then export_scene()
2. In Blender: enable this addon, click "Connect" in the sidebar panel
3. Edits in either application stream to the other in real-time
"""
import bpy
import numpy as np
def _read_manifest_version() -> str:
"""Read version from blender_manifest.toml (single source of truth)."""
import os
manifest = os.path.join(os.path.dirname(__file__), "blender_manifest.toml")
try:
with open(manifest) as f:
for line in f:
if line.startswith("version"):
return line.split("=", 1)[1].strip().strip('"')
except OSError:
pass
return "unknown"
ADDON_VERSION = _read_manifest_version()
# Support hot-reload (F3 > Reload Scripts)
if "shm_protocol" in locals():
import importlib
importlib.reload(shm_protocol) # noqa: F821
importlib.reload(connection) # noqa: F821
importlib.reload(scene_import) # noqa: F821
importlib.reload(change_detector) # noqa: F821
importlib.reload(tile_baker) # noqa: F821
importlib.reload(procedural_baker) # noqa: F821
from . import change_detector, connection, procedural_baker, scene_import, shm_protocol, tile_baker
from .change_detector import ChangeDetector
from .connection import BridgeConnection
from .procedural_baker import ProceduralBaker
from .scene_import import SceneImporter
from .shm_protocol import ShmRegion
from .tile_baker import TileBaker
# ── Global state ─────────────────────────────────────────────────────────────
_shm: ShmRegion | None = None
_conn: BridgeConnection | None = None
_importer: SceneImporter | None = None
_detector: ChangeDetector | None = None
_baker: TileBaker | None = None
_proc_baker: ProceduralBaker | None = None
_timer_running = False
_timer_generation = 0 # Incremented on each connect to invalidate stale timers
SHM_NAME = "/blender_bridge"
SHM_SIZE = 24 * 1024 * 1024 # 24 MB (8 MB region 0 + 8 MB region 1 + 8 MB tile ring)
MAX_OBJECTS = 256
SOCKET_PATH = "/tmp/blender_bridge.sock"
POLL_INTERVAL = 0.05 # 50ms = 20 Hz polling
# ── Timer callback ───────────────────────────────────────────────────────────
def _make_poll_timer(generation: int):
"""Create a poll timer closure bound to a specific generation."""
def _poll_timer():
return _poll_timer_impl(generation)
return _poll_timer
def _scan_material_textures():
"""Scan Blender scene for materials with textures and request SVT registration.
Handles image textures (existing pipeline), procedural textures (bake first),
and ColorRamp chains (trace upstream recursively).
"""
global _conn, _baker, _proc_baker
if _conn is None or _baker is None:
return
import bpy
image_requested = set()
for obj in bpy.data.objects:
if obj.type != "MESH" or obj.active_material is None:
continue
mat = obj.active_material
if not mat.use_nodes or not mat.node_tree:
continue
for node in mat.node_tree.nodes:
if node.type != "BSDF_PRINCIPLED":
continue
base_color = node.inputs.get("Base Color")
if base_color is None or not base_color.is_linked:
continue
for link in base_color.links:
_handle_texture_node(obj, mat, link.from_node, image_requested)
if image_requested:
print(f"BlenderBridge: Requested {len(image_requested)} texture(s)")
_PROCEDURAL_TYPES = {
"TEX_NOISE", "TEX_VORONOI", "TEX_GRADIENT", "TEX_CHECKER",
"TEX_WAVE", "TEX_BRICK", "TEX_MAGIC", "TEX_MUSGRAVE",
"TEX_WHITE_NOISE",
}
def _handle_texture_node(obj, mat, from_node, image_requested):
"""Route a texture node to the appropriate handler."""
global _conn, _baker, _proc_baker
# Image textures → existing SVT pipeline (no bake)
if from_node.type == "TEX_IMAGE" and from_node.image is not None:
img = from_node.image
img_name = img.name
if img_name not in image_requested and img_name not in _baker._texture_ids:
image_requested.add(img_name)
img_w, img_h = img.size[0], img.size[1]
msg = f"request_texture:{img_name},{img_w},{img_h}"
_conn.send_message(msg.encode())
print(f"BlenderBridge: Requesting image texture '{img_name}' ({img_w}x{img_h})")
return
# ColorRamp → trace upstream for the actual texture node
if from_node.type == "VALTORGB":
fac_input = from_node.inputs.get("Fac")
if fac_input and fac_input.is_linked:
for inner_link in fac_input.links:
_handle_texture_node(obj, mat, inner_link.from_node, image_requested)
return
# Procedural textures → queue for non-blocking background bake
if from_node.type in _PROCEDURAL_TYPES:
if _proc_baker is not None:
_proc_baker.queue(obj.name, mat.name)
TILE_RING_SIZE = 8 * 1024 * 1024 # 8 MB tile ring (fixed, does not grow)
def _on_shm_resized(new_size: int, new_region_size: int):
"""Called by ChangeDetector after SHM resize. Notifies Godot and updates tile baker."""
global _conn, _baker, _shm
if _conn is not None:
msg = f"shm_resized:{new_size},{new_region_size}"
_conn.send_message(msg.encode())
# Re-initialize tile baker with new ring buffer offset
if _baker is not None and _shm is not None:
tile_ring_offset = new_region_size * 2
tile_ring_size = new_size - tile_ring_offset
if tile_ring_size > 0:
_baker.init_ring_buffer(tile_ring_offset, tile_ring_size)
print(f"BlenderBridge: Tile ring moved to offset {tile_ring_offset}")
def _send_image_file(img_name: str, tex_id: int):
"""Save Blender image as raw RGBA8 file and notify Godot of the path."""
global _conn
if _conn is None or bpy is None:
return
image = bpy.data.images.get(img_name)
if image is None:
return
w, h = image.size[0], image.size[1]
if w == 0 or h == 0:
return
# Extract pixels as raw RGBA8. Blender stores bottom-to-top and Godot stores
# top-to-bottom, but UV conventions are also opposite (Blender V=0 at bottom,
# Godot V=0 at top). These cancel out, so NO flip is needed.
pixels = np.empty(w * h * 4, dtype=np.float32)
image.pixels.foreach_get(pixels)
uint8 = (np.clip(pixels, 0.0, 1.0) * 255.0 + 0.5).astype(np.uint8)
path = f"/tmp/bridge_tex_{tex_id}.rgba"
with open(path, "wb") as f:
f.write(uint8.tobytes())
msg = f"image_file:{tex_id},{w},{h},{path}"
_conn.send_message(msg.encode())
print(f"BlenderBridge: Sent image '{img_name}' ({w}x{h}) as {path}")
def _poll_timer_impl(generation: int):
"""Called by Blender's timer to poll for updates from Godot."""
global _conn, _shm, _importer, _detector, _baker, _timer_running
# Stop if this timer belongs to a stale generation (disconnect+reconnect happened)
if generation != _timer_generation:
return None
if _conn is None:
_timer_running = False
return None # Stop timer
# Auto-reconnect on unexpected disconnect
if not _conn.connected and _conn.should_reconnect:
return _attempt_reconnect()
# Manual disconnect — stop timer
if not _conn.connected:
_timer_running = False
return None
# Tick suppression counter (feedback loop prevention)
if _detector is not None:
_detector.unsuppress()
# Poll UDS messages
got_incoming = False
messages = _conn.poll_messages()
# Check if polling caused an unexpected disconnect
if not _conn.connected and _conn.should_reconnect:
print("BlenderBridge: Connection lost, will attempt reconnect...")
return 2.0 # Retry in 2 seconds
for msg in messages:
text = msg.decode("utf-8", errors="replace")
if text == "scene_exported":
# Full scene sync from Godot
if _importer is not None:
_importer.full_import()
if _detector is not None:
_detector.set_id_map(_importer._id_to_name)
# Suppress echoes for all imported objects
_detector.suppress(set(_importer._id_to_name.values()), 5)
got_incoming = True
print(f"BlenderBridge: Synced scene from Godot")
# Scan Blender scene for textures (image + procedural) and request SVT registration
_scan_material_textures()
elif text in ("updated", "transform_update"):
# Incremental update from Godot (region 0)
if _importer is not None:
updated_names = _importer.poll_updates()
if updated_names and _detector is not None:
# Only suppress echoes for the objects Godot just sent.
# Other objects (e.g. modifier dependents) still update.
_detector.suppress(updated_names, 5)
got_incoming = True
elif text == "finalized":
print("BlenderBridge: Godot finalized scene successfully")
elif text == "finalize_error":
print("BlenderBridge: Godot finalize failed")
elif text.startswith("texture_registered:"):
# Godot registered a texture: texture_registered:name,id
parts = text.split(":", 1)[1].split(",")
if len(parts) == 2 and _baker is not None:
img_name = parts[0]
tex_id = int(parts[1])
_baker.register_texture(img_name, tex_id)
print(f"BlenderBridge: Registered texture '{img_name}' id={tex_id}")
# Send full image file to Godot (replaces VT tile pipeline)
_send_image_file(img_name, tex_id)
# Force re-flush of meshes using this texture so tex_id is written
# to the material blob (first flush had tex_id=0 because registration
# hadn't completed yet)
if _detector is not None:
_detector.invalidate_meshes_with_image(img_name)
# Flush accumulated Blender changes to SHM and notify Godot
if _detector is not None:
flushed = _detector.flush()
if flushed > 0 and _conn is not None:
_conn.send_message(b"updated")
# Sync detector's ID map to importer so Godot transforms for
# Blender-created objects can be applied on incoming updates
if _importer is not None:
_importer._id_to_name.update(_detector.get_id_map())
# Re-send any images that were edited (texture paint, image editor)
if _detector is not None and _baker is not None:
# Keep detector's tracked images in sync with baker's registrations
_detector.set_tracked_images(set(_baker._texture_ids.keys()))
dirty_images = _detector.drain_dirty_images()
for img_name in dirty_images:
tex_id = _baker._texture_ids.get(img_name)
if tex_id is not None:
_send_image_file(img_name, tex_id)
# Poll non-blocking procedural baker for completed bakes
if _proc_baker is not None and _conn is not None and _baker is not None:
for mat_key, img_name in _proc_baker.poll():
img = bpy.data.images.get(img_name)
if img is not None:
img_w, img_h = img.size[0], img.size[1]
if img_name not in _baker._texture_ids:
msg = f"request_texture:{img_name},{img_w},{img_h}"
_conn.send_message(msg.encode())
else:
tex_id = _baker._texture_ids[img_name]
_send_image_file(img_name, tex_id)
# Re-bake procedural textures if shading changed (parameter edits)
if _detector is not None and _detector.had_shading_update():
if _proc_baker is not None:
for mat_key in list(_proc_baker._results.keys()):
_proc_baker.invalidate(mat_key)
obj_name, mat_name = mat_key.split(":", 1)
_proc_baker.queue(obj_name, mat_name)
if _proc_baker.pending_count > 0:
_proc_baker._save_blend()
_proc_baker.start()
# Flush any pending tile_ready messages from the baker
if _baker is not None:
for tile_msg in _baker.get_pending_messages():
_conn.send_message(tile_msg.encode())
return POLL_INTERVAL # Continue polling
def _attempt_reconnect():
"""Try to reconnect after unexpected disconnect. Returns timer interval."""
global _shm, _conn, _importer, _detector, _baker
# Re-open SHM (Godot may have restarted with fresh SHM)
if _shm is not None:
_shm.close()
_shm = ShmRegion()
if not _shm.open(SHM_NAME, SHM_SIZE):
return 2.0 # Retry in 2 seconds
# Reconnect UDS
if not _conn.connect():
_shm.close()
_shm = None
return 2.0 # Retry in 2 seconds
# Re-initialize importer; preserve detector across reconnects to keep
# _name_to_id and _sequence (prevents ID loss and sequence mismatch)
_importer = SceneImporter(_shm, MAX_OBJECTS)
if _detector is not None:
_detector.unregister()
_detector._shm = _shm
_detector._pending.clear()
_detector._last_depsgraph = None
# Keep _transform_hashes and _geometry_hashes intact — they reflect
# the current Blender scene state and are still valid. Clearing them
# causes ALL objects to be falsely re-detected as changed, triggering
# an echo loop that duplicates objects in Godot.
_detector.register()
_detector.set_resize_callback(_on_shm_resized)
else:
_detector = ChangeDetector(_shm, MAX_OBJECTS)
_detector.register()
_detector.set_resize_callback(_on_shm_resized)
# Reuse existing baker but clear stale _texture_ids (Godot restarted;
# its SVT has no registrations, so textures must be re-requested).
tile_ring_offset = _shm._region_size * 2
tile_ring_size = _shm.size - tile_ring_offset
if _baker is not None:
_baker._shm = _shm
_baker.init_ring_buffer(tile_ring_offset, tile_ring_size)
_baker._texture_ids.clear() # Godot restarted; force re-registration
else:
_baker = TileBaker(_shm, tile_size=128, max_mips=8)
_baker.init_ring_buffer(tile_ring_offset, tile_ring_size)
# Share texture ID map with change_detector
if _detector is not None and _baker is not None:
_detector._texture_ids = _baker._texture_ids
_conn.clear_reconnect()
print("BlenderBridge: Reconnected successfully")
return POLL_INTERVAL # Resume normal polling
# ── Addon Preferences ───────────────────────────────────────────────────────
class BridgeAddonPreferences(bpy.types.AddonPreferences):
bl_idname = __package__
source_path: bpy.props.StringProperty(
name="Source Path",
subtype="DIR_PATH",
default="/storage/projects/code/godot/blender-bridge/blender_addon",
description="Local path to blender_bridge addon source directory",
)
def draw(self, context):
layout = self.layout
layout.prop(self, "source_path")
layout.operator("bridge.update_addon", icon="FILE_REFRESH")
# ── Operators ────────────────────────────────────────────────────────────────
class BRIDGE_OT_update_addon(bpy.types.Operator):
"""Update addon files from local source and hot-reload"""
bl_idname = "bridge.update_addon"
bl_label = "Update Addon"
def execute(self, context):
import os
import shutil
prefs = context.preferences.addons[__package__].preferences
source = bpy.path.abspath(prefs.source_path)
if not os.path.isdir(source):
self.report({"ERROR"}, f"Source directory not found: {source}")
return {"CANCELLED"}
addon_dir = os.path.dirname(os.path.realpath(__file__))
is_symlink = os.path.islink(os.path.dirname(__file__))
copied = 0
if not is_symlink:
for fname in os.listdir(source):
if fname.endswith((".py", ".toml")):
src = os.path.join(source, fname)
dst = os.path.join(addon_dir, fname)
shutil.copy2(src, dst)
copied += 1
# Hot-reload: unregister, purge our modules, reimport, register
pkg = __package__
def _deferred_reload():
import importlib
import sys
# Disconnect bridge if active
try:
unregister()
except Exception as e:
print(f"BlenderBridge: unregister warning: {e}")
# Purge all our modules from sys.modules so Python reimports fresh
to_remove = [
name
for name in sys.modules
if name == pkg or name.startswith(pkg + ".")
]
for name in to_remove:
del sys.modules[name]
# Reimport and re-register
try:
mod = importlib.import_module(pkg)
mod.register()
print("BlenderBridge: Addon reloaded successfully")
except Exception as e:
print(f"BlenderBridge: Reload failed: {e}")
return None
bpy.app.timers.register(_deferred_reload, first_interval=0.1)
if is_symlink:
self.report({"INFO"}, "Reloading addon...")
else:
self.report({"INFO"}, f"Copied {copied} files, reloading...")
return {"FINISHED"}
class BRIDGE_OT_connect(bpy.types.Operator):
"""Connect to Godot BlenderBridge server"""
bl_idname = "bridge.connect"
bl_label = "Connect to Godot"
def execute(self, context):
global _shm, _conn, _importer, _detector, _baker, _proc_baker
global _timer_running, _timer_generation
# Clean up any stale state from previous connection
if _proc_baker is not None:
_proc_baker.reset()
_proc_baker = None
if _detector is not None:
_detector.unregister()
_detector = None
_baker = None
_importer = None
if _conn is not None:
_conn.close()
_conn = None
if _shm is not None:
_shm.close()
_shm = None
# Open shared memory
_shm = ShmRegion()
if not _shm.open(SHM_NAME, SHM_SIZE):
self.report({"ERROR"}, f"Failed to open shared memory: {SHM_NAME}")
return {"CANCELLED"}
# Connect UDS
_conn = BridgeConnection(SOCKET_PATH)
if not _conn.connect():
_shm.close()
_shm = None
self.report({"ERROR"}, f"Failed to connect to: {SOCKET_PATH}")
return {"CANCELLED"}
# Set up importer and change detector
_importer = SceneImporter(_shm, MAX_OBJECTS)
_detector = ChangeDetector(_shm, MAX_OBJECTS)
_detector.register()
_detector.set_resize_callback(_on_shm_resized)
# Set up tile baker (ring buffer after both regions)
from .shm_protocol import REGION_SIZE
tile_ring_offset = REGION_SIZE * 2 # 16 MB offset
tile_ring_size = TILE_RING_SIZE
_baker = TileBaker(_shm, tile_size=128, max_mips=8)
_baker.init_ring_buffer(tile_ring_offset, tile_ring_size)
# Share texture ID map so change_detector can write tex_id in material blob
_detector._texture_ids = _baker._texture_ids
# Set up non-blocking procedural baker
_proc_baker = ProceduralBaker()
# Start polling timer with new generation (old timers auto-stop
# when they detect their generation is stale)
_timer_generation += 1
bpy.app.timers.register(
_make_poll_timer(_timer_generation), first_interval=POLL_INTERVAL
)
_timer_running = True
self.report({"INFO"}, "Connected to Godot BlenderBridge")
return {"FINISHED"}
class BRIDGE_OT_disconnect(bpy.types.Operator):
"""Disconnect from Godot BlenderBridge server"""
bl_idname = "bridge.disconnect"
bl_label = "Disconnect"
def execute(self, context):
global _shm, _conn, _importer, _detector, _baker, _proc_baker, _timer_running
if _detector is not None:
_detector.unregister()
_detector = None
_baker = None
if _proc_baker is not None:
_proc_baker.reset()
_proc_baker = None
_importer = None
if _conn is not None:
_conn.close()
_conn = None
if _shm is not None:
_shm.close()
_shm = None
_timer_running = False
self.report({"INFO"}, "Disconnected from Godot BlenderBridge")
return {"FINISHED"}
class BRIDGE_OT_import_scene(bpy.types.Operator):
"""Sync Blender scene to match Godot scene"""
bl_idname = "bridge.import_scene"
bl_label = "Sync Scene"
def execute(self, context):
if _importer is None:
self.report({"ERROR"}, "Not connected")
return {"CANCELLED"}
_importer.full_import()
if _detector is not None:
_detector.set_id_map(_importer._id_to_name)
header = _shm.read_header(region=0)
self.report(
{"INFO"},
f"Synced {header['object_count']} objects from Godot",
)
return {"FINISHED"}
class BRIDGE_OT_finalize(bpy.types.Operator):
"""Save .blend and tell Godot to save .tscn + textures"""
bl_idname = "bridge.finalize"
bl_label = "Finalize Bridge"
def execute(self, context):
global _conn
if _conn is None or not _conn.connected:
self.report({"ERROR"}, "Not connected to Godot")
return {"CANCELLED"}
# Save the .blend file
bpy.ops.wm.save_mainfile()
# Tell Godot to finalize
_conn.send_message(b"finalize")
self.report({"INFO"}, "Finalize sent — saving .blend and Godot scene")
return {"FINISHED"}
class BRIDGE_OT_bake_textures(bpy.types.Operator):
"""Bake procedural textures in background (non-blocking)"""
bl_idname = "bridge.bake_textures"
bl_label = "Bake Textures"
def execute(self, context):
if _proc_baker is None or _proc_baker.pending_count == 0:
self.report({"INFO"}, "No procedural textures to bake")
return {"CANCELLED"}
_proc_baker.start()
self.report({"INFO"}, f"Baking {_proc_baker.pending_count} texture(s) in background...")
return {"FINISHED"}
# ── UI Panel ─────────────────────────────────────────────────────────────────
class BRIDGE_PT_panel(bpy.types.Panel):
"""Godot Bridge panel in the 3D viewport sidebar"""
bl_label = "Godot Bridge"
bl_idname = "BRIDGE_PT_panel"
bl_space_type = "VIEW_3D"
bl_region_type = "UI"
bl_category = "Godot"
def draw(self, context):
layout = self.layout
connected = _conn is not None and _conn.connected
if not connected:
layout.operator("bridge.connect", icon="LINKED")
else:
layout.operator("bridge.disconnect", icon="UNLINKED")
layout.separator()
layout.operator("bridge.import_scene", icon="FILE_REFRESH")
layout.operator("bridge.finalize", icon="EXPORT")
# Procedural bake button
if _proc_baker is not None:
pending = _proc_baker.pending_count
if pending > 0:
layout.operator("bridge.bake_textures",
text=f"Bake Textures ({pending})",
icon="RENDER_STILL")
elif _proc_baker.is_baking:
layout.label(text="Baking...", icon="TIME")
# Status info
box = layout.box()
box.label(text="Status: Connected", icon="CHECKMARK")
if _shm is not None and _shm.valid:
hdr0 = _shm.read_header(region=0)
hdr1 = _shm.read_header(region=1)
box.label(text=f"Godot→Blender: {hdr0['object_count']} obj, seq={hdr0['sequence']}")
box.label(text=f"Blender→Godot: {hdr1['object_count']} obj, seq={hdr1['sequence']}")
# Update button (always visible)
layout.separator()
layout.operator("bridge.update_addon", icon="IMPORT")
layout.label(text=f"v{ADDON_VERSION}")
# ── Registration ─────────────────────────────────────────────────────────────
_classes = (
BridgeAddonPreferences,
BRIDGE_OT_update_addon,
BRIDGE_OT_connect,
BRIDGE_OT_disconnect,
BRIDGE_OT_import_scene,
BRIDGE_OT_finalize,
BRIDGE_OT_bake_textures,
BRIDGE_PT_panel,
)
def register():
for cls in _classes:
bpy.utils.register_class(cls)
def unregister():
global _shm, _conn, _importer, _detector, _baker, _proc_baker, _timer_running
# Clean up connection
if _detector is not None:
_detector.unregister()
_detector = None
_baker = None
if _proc_baker is not None:
_proc_baker.reset()
_proc_baker = None
_importer = None
if _conn is not None:
_conn.close()
_conn = None
if _shm is not None:
_shm.close()
_shm = None
_timer_running = False
for cls in reversed(_classes):
bpy.utils.unregister_class(cls)
if __name__ == "__main__":
register()