feat: BLENDER_CINEMATIC workflow graph node (M1)
Add StepName.BLENDER_CINEMATIC and full graph runtime support so cinematic output types can be promoted from legacy_only to graph/shadow rollout mode. - process_steps.py: add BLENDER_CINEMATIC = "blender_cinematic" enum value - workflow_executor.py: map to render_cinematic_task in STEP_TASK_MAP - workflow_node_registry.py: node definition with render/scene/camera fields (no animation params — cinematic is fixed at 250 frames @ 25fps) - workflow_graph_runtime.py: _ORDER_LINE_RENDER_STEPS, _CINEMATIC_TASK_KEYS, shadow queue routing, predict_render_output_artifact (mp4), _build_task_kwargs, _artifact_kind_override_for_step - tasks.py: _normalize_cinematic_params + render_cinematic_task Celery task (calls render_cinematic_to_file, publishes as turntable asset type since mp4) No DB migration needed: admins can now manually set cinematic output types to graph rollout mode via the admin panel and assign a workflow definition. docs: learnings erfasst — BLENDER_CINEMATIC workflow graph node M1 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
@@ -1631,6 +1631,251 @@ def render_order_line_still_task(self, order_line_id: str, **params) -> dict:
|
||||
raise self.retry(exc=exc, countdown=30)
|
||||
|
||||
|
||||
def _normalize_cinematic_params(params: dict) -> dict:
|
||||
"""Map graph/editor params onto render_cinematic_to_file kwargs."""
|
||||
normalized = dict(params)
|
||||
normalized.pop("use_custom_render_settings", None)
|
||||
|
||||
legacy_engine = normalized.pop("render_engine", None)
|
||||
if legacy_engine is not None and normalized.get("engine") is None:
|
||||
normalized["engine"] = legacy_engine
|
||||
|
||||
usd_path = normalized.get("usd_path")
|
||||
if isinstance(usd_path, str) and usd_path.strip():
|
||||
normalized["usd_path"] = Path(usd_path)
|
||||
elif usd_path == "":
|
||||
normalized.pop("usd_path", None)
|
||||
|
||||
for key in _RENDER_STILL_CONTROL_PARAM_KEYS:
|
||||
normalized.pop(key, None)
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
@celery_app.task(
|
||||
bind=True,
|
||||
name="app.domains.rendering.tasks.render_cinematic_task",
|
||||
queue="asset_pipeline",
|
||||
max_retries=2,
|
||||
)
|
||||
def render_cinematic_task(self, order_line_id: str, **params) -> dict:
|
||||
"""Render a cinematic highlight animation for an order line (250 frames @ 25 fps).
|
||||
|
||||
Accepts order_line_id as context, resolves the STEP path from DB, runs
|
||||
render_cinematic_to_file, and publishes the result as a media asset.
|
||||
"""
|
||||
from app.domains.rendering.job_document import RenderJobDocument, JobState
|
||||
from app.core.process_steps import StepName
|
||||
|
||||
workflow_run_id = params.pop("workflow_run_id", None)
|
||||
workflow_node_id = params.pop("workflow_node_id", None)
|
||||
publish_asset_enabled = bool(params.pop("publish_asset_enabled", True))
|
||||
observer_output_enabled = bool(params.pop("observer_output_enabled", False))
|
||||
graph_authoritative_output_enabled = bool(params.pop("graph_authoritative_output_enabled", False))
|
||||
graph_output_node_ids = list(params.pop("graph_output_node_ids", []) or [])
|
||||
graph_notify_node_ids = list(params.pop("graph_notify_node_ids", []) or [])
|
||||
emit_events = bool(params.pop("emit_events", True))
|
||||
job_document_enabled = bool(params.pop("job_document_enabled", True))
|
||||
emit_legacy_notifications = bool(params.pop("emit_legacy_notifications", False))
|
||||
output_name_suffix = params.pop("output_name_suffix", None)
|
||||
|
||||
log_task_event(self.request.id, f"Starting render_cinematic_task: order_line={order_line_id}", "info")
|
||||
_mark_workflow_node_running(
|
||||
order_line_id,
|
||||
workflow_run_id=workflow_run_id,
|
||||
workflow_node_id=workflow_node_id,
|
||||
task_id=self.request.id,
|
||||
)
|
||||
|
||||
job_doc = RenderJobDocument.new(order_line_id=order_line_id, celery_task_id=self.request.id)
|
||||
job_doc.set_state(JobState.RUNNING)
|
||||
|
||||
def _save_job_doc():
|
||||
if not job_document_enabled:
|
||||
return
|
||||
try:
|
||||
from sqlalchemy import update as _upd
|
||||
from app.core.db_utils import get_sync_session
|
||||
from app.domains.orders.models import OrderLine
|
||||
with get_sync_session() as db:
|
||||
db.execute(
|
||||
_upd(OrderLine)
|
||||
.where(OrderLine.id == order_line_id)
|
||||
.values(render_job_doc=job_doc.to_dict())
|
||||
)
|
||||
except Exception as _exc:
|
||||
logger.debug("_save_job_doc failed: %s", _exc)
|
||||
|
||||
_save_job_doc()
|
||||
|
||||
job_doc.begin_step(StepName.RESOLVE_STEP_PATH)
|
||||
step_path_str, _cad_file_id = _resolve_step_path_for_order_line(order_line_id)
|
||||
if not step_path_str:
|
||||
job_doc.fail_step(StepName.RESOLVE_STEP_PATH, "product missing or has no linked CAD file")
|
||||
job_doc.set_state(JobState.FAILED, error="Cannot resolve STEP path")
|
||||
_save_job_doc()
|
||||
log_task_event(self.request.id, f"Failed: cannot resolve STEP path for order_line {order_line_id}", "error")
|
||||
raise RuntimeError(
|
||||
f"Cannot resolve STEP path for order_line {order_line_id}: "
|
||||
"product missing or has no linked CAD file"
|
||||
)
|
||||
job_doc.finish_step(StepName.RESOLVE_STEP_PATH, output={"step_path": step_path_str})
|
||||
|
||||
step = Path(step_path_str)
|
||||
cinematic_filename = f"line_{order_line_id}_cinematic.mp4"
|
||||
if output_name_suffix:
|
||||
cinematic_filename = f"line_{order_line_id}_cinematic_{output_name_suffix}.mp4"
|
||||
output_path = build_order_line_step_render_path(
|
||||
step,
|
||||
order_line_id,
|
||||
cinematic_filename,
|
||||
ensure_exists=True,
|
||||
)
|
||||
|
||||
render_params = _normalize_cinematic_params(params)
|
||||
|
||||
try:
|
||||
job_doc.begin_step(StepName.BLENDER_CINEMATIC)
|
||||
from app.services.render_blender import render_cinematic_to_file
|
||||
|
||||
result = render_cinematic_to_file(
|
||||
step_path=step,
|
||||
output_path=output_path,
|
||||
**render_params,
|
||||
)
|
||||
job_doc.finish_step(
|
||||
StepName.BLENDER_CINEMATIC,
|
||||
output={"output_path": str(output_path), "duration_s": result.get("total_duration_s")},
|
||||
)
|
||||
job_doc.set_state(JobState.COMPLETED, result={
|
||||
"output_path": str(output_path),
|
||||
"duration_s": result.get("total_duration_s"),
|
||||
"engine_used": result.get("engine_used"),
|
||||
})
|
||||
_save_job_doc()
|
||||
|
||||
if graph_authoritative_output_enabled:
|
||||
_finalize_graph_turntable_output(
|
||||
order_line_id,
|
||||
success=True,
|
||||
output_path=str(output_path),
|
||||
render_log=result,
|
||||
workflow_run_id=workflow_run_id,
|
||||
output_node_ids=graph_output_node_ids,
|
||||
render_node_id=workflow_node_id,
|
||||
)
|
||||
elif observer_output_enabled:
|
||||
_finalize_shadow_turntable_output(
|
||||
order_line_id,
|
||||
success=True,
|
||||
output_path=str(output_path),
|
||||
render_log=result,
|
||||
workflow_run_id=workflow_run_id,
|
||||
output_node_ids=graph_output_node_ids,
|
||||
render_node_id=workflow_node_id,
|
||||
)
|
||||
elif publish_asset_enabled:
|
||||
publish_asset.delay(
|
||||
order_line_id,
|
||||
"turntable",
|
||||
str(output_path),
|
||||
render_config=result,
|
||||
workflow_run_id=workflow_run_id,
|
||||
)
|
||||
log_task_event(self.request.id, f"Completed successfully in {result.get('total_duration_s', 0):.1f}s", "done")
|
||||
logger.info(
|
||||
"render_cinematic_task completed for line %s in %.1fs",
|
||||
order_line_id, result.get("total_duration_s", 0),
|
||||
)
|
||||
try:
|
||||
from app.core.websocket import publish_event_sync
|
||||
if emit_events:
|
||||
publish_event_sync(None, {
|
||||
"type": "render.order_line.completed",
|
||||
"order_line_id": order_line_id,
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
if emit_legacy_notifications:
|
||||
_emit_graph_render_notifications(
|
||||
order_line_id,
|
||||
success=True,
|
||||
render_log=result,
|
||||
)
|
||||
_finalize_graph_notify_nodes(
|
||||
workflow_run_id=workflow_run_id,
|
||||
notify_node_ids=graph_notify_node_ids,
|
||||
success=True,
|
||||
render_node_id=workflow_node_id,
|
||||
)
|
||||
_update_workflow_run_status(
|
||||
order_line_id,
|
||||
"completed",
|
||||
workflow_run_id=workflow_run_id,
|
||||
workflow_node_id=workflow_node_id,
|
||||
)
|
||||
return result
|
||||
except Exception as exc:
|
||||
job_doc.fail_step(StepName.BLENDER_CINEMATIC, str(exc))
|
||||
job_doc.set_state(JobState.FAILED, error=str(exc))
|
||||
_save_job_doc()
|
||||
log_task_event(self.request.id, f"Failed: {exc}", "error")
|
||||
logger.error("render_cinematic_task failed for %s: %s", order_line_id, exc)
|
||||
try:
|
||||
from app.core.websocket import publish_event_sync
|
||||
if emit_events:
|
||||
publish_event_sync(None, {
|
||||
"type": "render.order_line.failed",
|
||||
"order_line_id": order_line_id,
|
||||
"error": str(exc),
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
if graph_authoritative_output_enabled:
|
||||
_finalize_graph_turntable_output(
|
||||
order_line_id,
|
||||
success=False,
|
||||
output_path=str(output_path),
|
||||
render_log={"error": str(exc)},
|
||||
workflow_run_id=workflow_run_id,
|
||||
output_node_ids=graph_output_node_ids,
|
||||
render_node_id=workflow_node_id,
|
||||
error=str(exc),
|
||||
)
|
||||
elif observer_output_enabled:
|
||||
_finalize_shadow_turntable_output(
|
||||
order_line_id,
|
||||
success=False,
|
||||
output_path=str(output_path),
|
||||
render_log={"error": str(exc)},
|
||||
workflow_run_id=workflow_run_id,
|
||||
output_node_ids=graph_output_node_ids,
|
||||
render_node_id=workflow_node_id,
|
||||
error=str(exc),
|
||||
)
|
||||
if emit_legacy_notifications:
|
||||
_emit_graph_render_notifications(
|
||||
order_line_id,
|
||||
success=False,
|
||||
render_log={"error": str(exc)},
|
||||
)
|
||||
_finalize_graph_notify_nodes(
|
||||
workflow_run_id=workflow_run_id,
|
||||
notify_node_ids=graph_notify_node_ids,
|
||||
success=False,
|
||||
render_node_id=workflow_node_id,
|
||||
error=str(exc),
|
||||
)
|
||||
_update_workflow_run_status(
|
||||
order_line_id,
|
||||
"failed",
|
||||
str(exc),
|
||||
workflow_run_id=workflow_run_id,
|
||||
workflow_node_id=workflow_node_id,
|
||||
)
|
||||
raise self.retry(exc=exc, countdown=60)
|
||||
|
||||
|
||||
@celery_app.task(
|
||||
bind=True,
|
||||
name="app.domains.rendering.tasks.export_blend_for_order_line_task",
|
||||
|
||||
Reference in New Issue
Block a user