422 lines
16 KiB
Python
422 lines
16 KiB
Python
"""Blender-owned bpy worker for declarative scene authoring."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import bpy
|
|
from mathutils import Vector
|
|
|
|
ADAPTER_VERSION = "0.1.0"
|
|
|
|
|
|
def atomic_json(path: Path, value: Any) -> None:
|
|
temporary = path.with_name(path.name + ".tmp-" + os.urandom(8).hex())
|
|
try:
|
|
with temporary.open("w", encoding="utf-8", newline="\n") as handle:
|
|
json.dump(value, handle, ensure_ascii=False, indent=2)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def terminal(
|
|
record: dict[str, Any],
|
|
status: str,
|
|
artifacts: list[dict[str, Any]],
|
|
error: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"job_id": record.get("job_id", ""),
|
|
"lease_id": record.get("lease_id", ""),
|
|
"request_digest": record.get("request_digest", ""),
|
|
"status": status,
|
|
"error": error,
|
|
"artifact_manifest": artifacts,
|
|
"terminal_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
def rgba(value: str) -> tuple[float, float, float, float]:
|
|
return tuple(int(value[index:index + 2], 16) / 255 for index in (1, 3, 5)) + (1.0,)
|
|
|
|
|
|
def material(name: str, color: str, metallic: float, roughness: float) -> Any:
|
|
result = bpy.data.materials.new(name)
|
|
result.diffuse_color = rgba(color)
|
|
result.use_nodes = True
|
|
shader = result.node_tree.nodes.get("Principled BSDF")
|
|
if shader is not None:
|
|
shader.inputs["Base Color"].default_value = rgba(color)
|
|
shader.inputs["Metallic"].default_value = metallic
|
|
shader.inputs["Roughness"].default_value = roughness
|
|
return result
|
|
|
|
|
|
def assign(obj: Any, value: Any) -> None:
|
|
if getattr(obj.data, "materials", None) is not None:
|
|
obj.data.materials.append(value)
|
|
|
|
|
|
def cube(name: str, location: Vector, dimensions: tuple[float, float, float], mat: Any) -> Any:
|
|
bpy.ops.mesh.primitive_cube_add(location=location)
|
|
obj = bpy.context.object
|
|
obj.name = name
|
|
obj.dimensions = dimensions
|
|
bpy.ops.object.transform_apply(location=False, rotation=False, scale=True)
|
|
assign(obj, mat)
|
|
return obj
|
|
|
|
|
|
def cylinder(
|
|
name: str,
|
|
location: Vector,
|
|
direction: Vector,
|
|
radius: float,
|
|
length: float,
|
|
mat: Any,
|
|
vertices: int = 64,
|
|
) -> Any:
|
|
bpy.ops.mesh.primitive_cylinder_add(vertices=vertices, radius=radius, depth=length, location=location)
|
|
obj = bpy.context.object
|
|
obj.name = name
|
|
obj.rotation_mode = "QUATERNION"
|
|
obj.rotation_quaternion = direction.normalized().to_track_quat("Z", "Y")
|
|
assign(obj, mat)
|
|
return obj
|
|
|
|
|
|
def torus(name: str, location: Vector, direction: Vector, major: float, minor: float, mat: Any) -> Any:
|
|
bpy.ops.mesh.primitive_torus_add(
|
|
major_radius=major,
|
|
minor_radius=minor,
|
|
major_segments=64,
|
|
minor_segments=12,
|
|
location=location,
|
|
)
|
|
obj = bpy.context.object
|
|
obj.name = name
|
|
obj.rotation_mode = "QUATERNION"
|
|
obj.rotation_quaternion = direction.normalized().to_track_quat("Z", "Y")
|
|
assign(obj, mat)
|
|
return obj
|
|
|
|
|
|
def look_at(obj: Any, target: Vector) -> None:
|
|
obj.rotation_euler = (target - obj.location).to_track_quat("-Z", "Y").to_euler()
|
|
|
|
|
|
def clear_scene(base: Path | None) -> None:
|
|
if base is not None:
|
|
source = base / "scene.blend"
|
|
if not source.is_file():
|
|
raise FileNotFoundError("WORKSPACE_SCENE_MISSING")
|
|
bpy.ops.wm.open_mainfile(filepath=str(source), load_ui=False)
|
|
else:
|
|
bpy.ops.wm.read_factory_settings(use_empty=True)
|
|
bpy.ops.object.select_all(action="SELECT")
|
|
bpy.ops.object.delete(use_global=False)
|
|
for datablocks in (bpy.data.materials, bpy.data.cameras, bpy.data.lights):
|
|
for datablock in list(datablocks):
|
|
if datablock.users == 0:
|
|
datablocks.remove(datablock)
|
|
|
|
|
|
def build_rotary_kiln(spec: dict[str, Any]) -> dict[str, Any]:
|
|
kiln = spec["kiln"]
|
|
components = spec.get("components") or {}
|
|
appearance = spec.get("appearance") or {}
|
|
length = float(kiln["length_m"])
|
|
diameter = float(kiln["outer_diameter_m"])
|
|
radius = diameter / 2
|
|
slope = math.atan(float(kiln["slope_percent"]) / 100)
|
|
stations = int(kiln["support_stations"])
|
|
direction = Vector((math.cos(slope), 0, -math.sin(slope)))
|
|
low_height = radius + max(1.2, radius * 0.55)
|
|
center = Vector((0, 0, low_height + length * math.sin(slope) / 2))
|
|
|
|
def axis_at(distance: float) -> Vector:
|
|
return center + direction * distance
|
|
|
|
shell_mat = material("ShellMaterial", appearance.get("shell_color", "#707780"), 0.65, 0.3)
|
|
metal_mat = material("EquipmentMetal", appearance.get("metal_color", "#303840"), 0.75, 0.25)
|
|
lining_mat = material("RefractoryLining", appearance.get("lining_color", "#B56A3B"), 0.05, 0.85)
|
|
hot_mat = material("HotMaterial", appearance.get("material_color", "#E35D20"), 0.0, 0.55)
|
|
concrete_mat = material("Concrete", "#77736B", 0.0, 0.9)
|
|
|
|
shell = cylinder("KilnShell", center, direction, radius, length, shell_mat, 96)
|
|
bevel = shell.modifiers.new("ShellEdgeBevel", "BEVEL")
|
|
bevel.width = min(0.08, diameter * 0.015)
|
|
bevel.segments = 3
|
|
|
|
inner_radius = max(0.05, radius - float(kiln["shell_thickness_m"]) - float(kiln["lining_thickness_m"]))
|
|
for side, distance in (("Tail", -length / 2 - 0.015), ("Head", length / 2 + 0.015)):
|
|
cylinder(f"{side}LiningFace", axis_at(distance), direction, inner_radius, 0.04, lining_mat, 96)
|
|
if components.get("material_bed", True):
|
|
cylinder("KilnMaterial", center, direction, inner_radius * 0.72, length * 0.985, hot_mat, 64)
|
|
|
|
station_positions: list[float] = []
|
|
for index in range(stations):
|
|
distance = -length * 0.38 + index * (length * 0.76 / (stations - 1))
|
|
station_positions.append(distance)
|
|
point = axis_at(distance)
|
|
torus(f"Tyre_{index + 1:02d}", point, direction, radius + 0.12, max(0.1, diameter * 0.045), metal_mat)
|
|
roller_radius = max(0.22, diameter * 0.09)
|
|
roller_length = max(0.7, diameter * 0.32)
|
|
for side_index, y_sign in enumerate((-1, 1), start=1):
|
|
roller_point = point + Vector((0, y_sign * radius * 0.72, -radius * 0.78))
|
|
cylinder(
|
|
f"SupportRoller_{index + 1:02d}_{side_index}",
|
|
roller_point,
|
|
Vector((0, 1, 0)),
|
|
roller_radius,
|
|
roller_length,
|
|
metal_mat,
|
|
48,
|
|
)
|
|
foundation_top = max(0.3, point.z - radius * 0.78 - roller_radius + 0.05)
|
|
base_point = Vector((point.x, 0, foundation_top / 2))
|
|
cube(
|
|
f"Foundation_{index + 1:02d}",
|
|
base_point,
|
|
(diameter * 0.7, diameter * 2.0, foundation_top),
|
|
concrete_mat,
|
|
)
|
|
|
|
if components.get("drive_system", True):
|
|
drive_distance = station_positions[min(1, len(station_positions) - 1)] + length * 0.06
|
|
drive_point = axis_at(drive_distance)
|
|
torus("GirthGear", drive_point, direction, radius + 0.22, max(0.12, diameter * 0.06), metal_mat)
|
|
motor_center = drive_point + Vector((0, -radius * 1.35, -radius * 0.55))
|
|
cube("Gearbox", motor_center, (diameter * 0.45, diameter * 0.5, diameter * 0.42), metal_mat)
|
|
cube("DriveMotor", motor_center + Vector((-diameter * 0.48, 0, 0)), (diameter * 0.5, diameter * 0.32, diameter * 0.32), shell_mat)
|
|
|
|
if components.get("hoods", True):
|
|
tail = axis_at(-length / 2 - radius * 0.35)
|
|
head = axis_at(length / 2 + radius * 0.35)
|
|
cube("KilnTailHood", tail, (diameter * 0.75, diameter * 1.35, diameter * 1.5), metal_mat)
|
|
cube("KilnHeadHood", head, (diameter * 0.85, diameter * 1.55, diameter * 1.65), metal_mat)
|
|
|
|
if components.get("burner", True):
|
|
head = axis_at(length / 2)
|
|
burner_center = head + direction * radius * 1.0
|
|
cylinder("Burner", burner_center, direction, max(0.08, radius * 0.1), radius * 2.4, metal_mat, 32)
|
|
|
|
if components.get("inspection_platform", True):
|
|
head = axis_at(length / 2 - radius)
|
|
platform = head + Vector((0, -radius * 1.2, -radius * 0.15))
|
|
cube("InspectionPlatform", platform, (diameter * 1.5, diameter * 0.85, 0.12), metal_mat)
|
|
for offset_x in (-diameter * 0.65, diameter * 0.65):
|
|
cube("PlatformLeg", platform + Vector((offset_x, 0, -radius)), (0.12, 0.12, radius * 2), metal_mat)
|
|
|
|
cube(
|
|
"Ground",
|
|
Vector((0, 0, -0.1)),
|
|
(length * 1.25, max(length * 0.38, diameter * 5), 0.2),
|
|
concrete_mat,
|
|
)
|
|
return {
|
|
"length_m": length,
|
|
"outer_diameter_m": diameter,
|
|
"slope_percent": float(kiln["slope_percent"]),
|
|
"support_stations": stations,
|
|
"rotational_speed_rpm": float(kiln.get("rotational_speed_rpm", 3.0)),
|
|
"center": list(center),
|
|
}
|
|
|
|
|
|
def configure_scene(spec: dict[str, Any], dimensions: dict[str, Any]) -> None:
|
|
scene = bpy.context.scene
|
|
scene.unit_settings.system = "METRIC"
|
|
scene.unit_settings.length_unit = "METERS"
|
|
scene.unit_settings.scale_length = 1.0
|
|
render = spec.get("render") or {}
|
|
scene.render.resolution_x = int(render.get("width", 1280))
|
|
scene.render.resolution_y = int(render.get("height", 720))
|
|
scene.render.resolution_percentage = 100
|
|
scene.render.image_settings.file_format = "PNG"
|
|
scene.render.film_transparent = bool(render.get("transparent_background", False))
|
|
for engine in ("BLENDER_EEVEE_NEXT", "BLENDER_EEVEE"):
|
|
try:
|
|
scene.render.engine = engine
|
|
break
|
|
except TypeError:
|
|
continue
|
|
background = (spec.get("appearance") or {}).get("background_color", "#20252B")
|
|
if scene.world is None:
|
|
scene.world = bpy.data.worlds.new("SceneWorld")
|
|
scene.world.use_nodes = True
|
|
scene.world.color = rgba(background)[:3]
|
|
node = scene.world.node_tree.nodes.get("Background")
|
|
if node is not None:
|
|
node.inputs["Color"].default_value = rgba(background)
|
|
node.inputs["Strength"].default_value = 0.28
|
|
|
|
length = dimensions["length_m"]
|
|
diameter = dimensions["outer_diameter_m"]
|
|
target = Vector(dimensions["center"])
|
|
view = render.get("view", "isometric")
|
|
offsets = {
|
|
"isometric": Vector((length * 0.62, -length * 0.68, length * 0.34)),
|
|
"side": Vector((0, -length * 0.82, length * 0.12)),
|
|
"kiln_head": Vector((length * 0.68, -length * 0.18, length * 0.18)),
|
|
"kiln_tail": Vector((-length * 0.68, -length * 0.18, length * 0.18)),
|
|
}
|
|
bpy.ops.object.camera_add(location=target + offsets[view])
|
|
camera = bpy.context.object
|
|
camera.name = "SceneCamera"
|
|
camera.data.type = "ORTHO"
|
|
aspect = scene.render.resolution_x / scene.render.resolution_y
|
|
camera.data.ortho_scale = max(diameter * 5.5, length / aspect * 1.45)
|
|
camera.data.clip_end = max(1000, length * 10)
|
|
look_at(camera, target)
|
|
scene.camera = camera
|
|
|
|
bpy.ops.object.light_add(type="SUN", location=target + Vector((-length, -length, length)))
|
|
sun = bpy.context.object
|
|
sun.name = "KeySun"
|
|
sun.data.energy = 2.2
|
|
sun.rotation_euler = (math.radians(28), math.radians(-22), math.radians(-35))
|
|
bpy.ops.object.light_add(type="AREA", location=target + Vector((0, -diameter * 5, diameter * 6)))
|
|
area = bpy.context.object
|
|
area.name = "FillArea"
|
|
area.data.energy = max(1200, length * 80)
|
|
area.data.shape = "DISK"
|
|
area.data.size = max(10, length * 0.45)
|
|
look_at(area, target)
|
|
|
|
|
|
def artifact(path: Path, artifact_id: str, media_type: str) -> dict[str, Any]:
|
|
if not path.is_file() or path.stat().st_size == 0:
|
|
raise RuntimeError(f"OUTPUT_EXPORT_EMPTY:{path.name}")
|
|
if path.suffix.lower() == ".png" and path.read_bytes()[:8] != b"\x89PNG\r\n\x1a\n":
|
|
raise RuntimeError("PNG_EXPORT_INVALID")
|
|
return {
|
|
"artifact_id": artifact_id,
|
|
"filename": path.name,
|
|
"media_type": media_type,
|
|
"size_bytes": path.stat().st_size,
|
|
"sha256": sha256(path),
|
|
}
|
|
|
|
|
|
def execute(job_dir: Path, record: dict[str, Any]) -> list[dict[str, Any]]:
|
|
request = record["request"]
|
|
spec = request["operation"]["scene"]
|
|
if spec["type"] != "rotary_kiln" or request["inputs"]:
|
|
raise ValueError("BLENDER_REQUEST_SEMANTICS_INVALID")
|
|
workspace_path = job_dir / "workspace.json"
|
|
workspace = json.loads(workspace_path.read_text(encoding="utf-8"))
|
|
base_value = workspace.get("local_base_path")
|
|
base = Path(base_value).resolve(strict=True) if base_value else None
|
|
clear_scene(base)
|
|
dimensions = build_rotary_kiln(spec)
|
|
configure_scene(spec, dimensions)
|
|
|
|
output = job_dir / "output"
|
|
output.mkdir(exist_ok=True)
|
|
project = output / "scene.blend"
|
|
preview = output / "preview.png"
|
|
manifest = output / "scene-manifest.json"
|
|
provenance = output / "provenance.json"
|
|
bpy.ops.wm.save_as_mainfile(filepath=str(project), check_existing=False)
|
|
bpy.context.scene.render.filepath = str(preview)
|
|
bpy.ops.render.render(write_still=True)
|
|
|
|
meshes = [obj for obj in bpy.context.scene.objects if obj.type == "MESH"]
|
|
atomic_json(
|
|
manifest,
|
|
{
|
|
"title": spec["title"],
|
|
"scene_type": "rotary_kiln",
|
|
"dimensions": dimensions,
|
|
"objects": [
|
|
{
|
|
"name": obj.name,
|
|
"type": obj.type,
|
|
"vertices": len(obj.data.vertices),
|
|
"polygons": len(obj.data.polygons),
|
|
}
|
|
for obj in meshes
|
|
],
|
|
"object_count": len(bpy.context.scene.objects),
|
|
"mesh_count": len(meshes),
|
|
"vertex_count": sum(len(obj.data.vertices) for obj in meshes),
|
|
"polygon_count": sum(len(obj.data.polygons) for obj in meshes),
|
|
},
|
|
)
|
|
atomic_json(
|
|
provenance,
|
|
{
|
|
"adapter_version": ADAPTER_VERSION,
|
|
"blender_version": bpy.app.version_string,
|
|
"request_digest": record["request_digest"],
|
|
"workspace_mode": workspace.get("mode"),
|
|
"generator": "rotary_kiln@v1",
|
|
},
|
|
)
|
|
|
|
artifacts = [
|
|
artifact(project, "project", "application/x-blender"),
|
|
artifact(preview, "preview", "image/png"),
|
|
artifact(manifest, "scene_manifest", "application/json"),
|
|
]
|
|
if any(item.get("key") == "scene_glb" for item in request["outputs"]):
|
|
glb = output / "scene.glb"
|
|
bpy.ops.export_scene.gltf(filepath=str(glb), export_format="GLB", export_apply=True)
|
|
artifacts.append(artifact(glb, "scene_glb", "model/gltf-binary"))
|
|
artifacts.append(artifact(provenance, "provenance", "application/json"))
|
|
return artifacts
|
|
|
|
|
|
def job_directory() -> Path:
|
|
arguments = sys.argv[sys.argv.index("--") + 1:] if "--" in sys.argv else []
|
|
if len(arguments) != 1:
|
|
raise ValueError("BLENDER_JOB_DIRECTORY_MISSING")
|
|
return Path(arguments[0]).resolve(strict=True)
|
|
|
|
|
|
def main() -> int:
|
|
job_dir = job_directory()
|
|
record: dict[str, Any] = {}
|
|
try:
|
|
record = json.loads((job_dir / "request" / "request.json").read_text(encoding="utf-8"))
|
|
artifacts = execute(job_dir, record)
|
|
atomic_json(job_dir / "artifacts.json", artifacts)
|
|
atomic_json(job_dir / "terminal.json", terminal(record, "succeeded", artifacts, {}))
|
|
print("[OK] Blender rotary kiln scene completed.")
|
|
return 0
|
|
except Exception as exc: # terminal.json must capture Blender-side failures
|
|
atomic_json(
|
|
job_dir / "terminal.json",
|
|
terminal(
|
|
record,
|
|
"failed",
|
|
[],
|
|
{"code": type(exc).__name__, "detail": str(exc)[:500]},
|
|
),
|
|
)
|
|
print(f"[ERR] {type(exc).__name__}: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|