"""Build context shared across the solution build pipeline.
Provides ``NamingStrategy`` (Protocol) and ``BuildContext``, which carry
the naming strategy, rigging conventions, and the in-memory outputs registry
for the duration of a rig build.
"""
from __future__ import annotations
import dataclasses
import json
import logging
from enum import Enum
from typing import Protocol
from openrig.conventions import RiggingConventions
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# NamingStrategy protocol
# ---------------------------------------------------------------------------
[docs]
class NamingStrategy(Protocol):
"""Interface every naming convention implementation must satisfy.
Solutions call ``ctx.namer.build_name(...)`` — they never import
``get_manager()`` directly, keeping them agnostic to which naming
convention is active.
"""
[docs]
def build_name(self, **tokens: object) -> str:
"""Build a validated name from token key/value pairs."""
...
[docs]
def parse_name(self, name: str) -> dict[str, str]:
"""Parse a name string into its constituent token values."""
...
# ---------------------------------------------------------------------------
# Internal protocol — avoids circular import with solution.py
# ---------------------------------------------------------------------------
class _SolutionLike(Protocol):
"""Minimal interface the context needs from a solution instance."""
solution_id: str
# ---------------------------------------------------------------------------
# JSON serialisation helper
# ---------------------------------------------------------------------------
def _settings_default(obj: object) -> object:
"""Serialise Enum members by their value for settings/outputs storage.
Passed as the ``default`` argument to ``json.dumps``. Raises
``TypeError`` for any non-Enum type, which is the standard contract.
Args:
obj: The object that the JSON encoder could not serialise.
Returns:
The enum's underlying value.
Raises:
TypeError: If ``obj`` is not an ``Enum`` instance.
"""
if isinstance(obj, Enum):
return obj.value
raise TypeError(
f"Object of type {type(obj).__name__!r} is not JSON serialisable."
)
# ---------------------------------------------------------------------------
# BuildContext
# ---------------------------------------------------------------------------
[docs]
class BuildContext:
"""Shared state and services for the solution build pipeline.
Carries the naming strategy, rigging conventions, and the in-memory
outputs registry. Solutions use this to register their outputs, resolve
inputs from other solutions, and query guide positions in the scene.
Args:
namer: Naming strategy implementation used by all solutions.
conventions: Active rigging conventions (axes, rotate order, etc.).
"""
def __init__(
self,
namer: NamingStrategy,
conventions: RiggingConventions,
) -> None:
self.namer = namer
self.conventions = conventions
# solution_id → {port_name → Maya DAG path}
self._outputs: dict[str, dict[str, str]] = {}
# solution_id → {port_name → "source_id.port_name"}
self._connections: dict[str, dict[str, str]] = {}
# solution_id → type string (e.g. "Arm")
self._solution_types: dict[str, str] = {}
# solution_id → settings dataclass instance
self._solution_settings: dict[str, object] = {}
# solution_id → Maya DAG path of the solution's root group
self._solution_roots: dict[str, str] = {}
# full rig description dict — stored for finalize()
self._description: dict[str, object] = {}
# --- output registry ---
[docs]
def get_output(self, solution_id: str, port_name: str) -> str:
"""Return the Maya DAG path registered for an output port.
Args:
solution_id: Unique identifier of the source solution.
port_name: Name of the output port.
Returns:
The registered Maya DAG path string.
Raises:
KeyError: If the solution or port has not been registered yet.
"""
try:
return self._outputs[solution_id][port_name]
except KeyError:
raise KeyError(
f"No output '{port_name}' registered for solution '{solution_id}'."
) from None
[docs]
def set_output(self, solution_id: str, port_name: str, value: str) -> None:
"""Register an output value produced by a built solution.
Called by solutions inside ``build()`` for each declared output port.
Args:
solution_id: Unique identifier of the solution setting the output.
port_name: Name of the output port being registered.
value: Maya DAG path of the output node.
"""
if solution_id not in self._outputs:
self._outputs[solution_id] = {}
self._outputs[solution_id][port_name] = value
# --- input resolution ---
[docs]
def resolve_input(self, solution: _SolutionLike, port_name: str) -> str | None:
"""Resolve a declared input connection to a Maya DAG path.
Looks up the connection declaration for the given port and retrieves
the corresponding output from the already-built outputs registry.
Returns ``None`` if no connection was declared for the port (optional
inputs).
Args:
solution: The solution requesting the input.
port_name: Name of the input port to resolve.
Returns:
The Maya DAG path of the connected output, or ``None`` if no
connection was declared for this port.
Raises:
KeyError: If the referenced source solution or port has not been
built yet.
ValueError: If the stored connection string has an invalid format.
"""
connections = self._connections.get(solution.solution_id, {})
connection = connections.get(port_name)
if connection is None:
return None
parts = connection.split(".", 1)
if len(parts) != 2:
raise ValueError(
f"Malformed connection '{connection}' on "
f"'{solution.solution_id}.{port_name}'. "
"Expected format: 'source_solution_id.port_name'."
)
src_solution_id, src_port_name = parts
return self.get_output(src_solution_id, src_port_name)
# --- guide system ---
[docs]
def get_guide(
self, solution: _SolutionLike, guide_name: str
) -> tuple[float, float, float] | None:
"""Return the world position of a guide object if it exists in the scene.
Looks for a node named ``{solution_id}_guides_grp`` in the Maya scene
and navigates to ``{guides_grp}|{guide_name}``. Returns ``None`` if
not found — the caller falls back to its settings values.
Args:
solution: The solution whose guides group is searched.
guide_name: Short name of the guide node within the guides group.
Returns:
World-space ``(x, y, z)`` position tuple, or ``None`` if no
guide was found.
"""
import maya.cmds as cmds # lazy — only needed inside Maya
guides_grp_name = f"{solution.solution_id}_guides"
candidates: list[str] = cmds.ls(guides_grp_name, long=True) or []
if not candidates:
logger.debug(
"Guide group '%s' not found — falling back to settings.",
guides_grp_name,
)
return None
if len(candidates) > 1:
logger.warning(
"Multiple nodes named '%s' found. Using '%s'.",
guides_grp_name,
candidates[0],
)
guide_path = f"{candidates[0]}|{guide_name}"
if not cmds.objExists(guide_path):
logger.debug(
"Guide '%s' not found in '%s' — falling back to settings.",
guide_name,
candidates[0],
)
return None
raw = cmds.xform(guide_path, query=True, worldSpace=True, translation=True)
if not isinstance(raw, list):
return None
return (float(raw[0]), float(raw[1]), float(raw[2]))
# --- pipeline helpers (called by the build pipeline, not by solutions) ---
def _register_connections(
self, solution_id: str, connections: dict[str, str]
) -> None:
"""Store connection declarations for a solution.
Called by the build pipeline before build() runs, not by solutions.
Args:
solution_id: Unique identifier of the solution.
connections: Mapping of port name to ``"source_id.port_name"``.
"""
self._connections[solution_id] = connections
def _register_solution_meta(
self,
solution_id: str,
solution_type: str,
settings: object,
root_node: str,
) -> None:
"""Store solution metadata for use by ``finalize()``.
Called by the build pipeline after the solution's root node is known.
Args:
solution_id: Unique identifier of the solution.
solution_type: Type string (e.g. ``"Arm"``).
settings: Settings dataclass instance for the solution.
root_node: Maya DAG path of the solution's root group node.
"""
self._solution_types[solution_id] = solution_type
self._solution_settings[solution_id] = settings
self._solution_roots[solution_id] = root_node
def _set_description(self, description: dict[str, object]) -> None:
"""Store the full rig description for use by ``finalize()``.
Called by the build pipeline at the start of a build.
Args:
description: The parsed rig description dict.
"""
self._description = description
# --- finalize ---
[docs]
def finalize(self, rig_root: str) -> None:
"""Write rig-level and solution-level metadata to Maya nodes.
Called automatically by the build pipeline after all solutions have
completed ``build()``. Writes custom string attributes to the rig
root and to each solution's root node so the scene is self-describing.
Rig-level attributes (written to ``rig_root``):
- ``openrig_version``
- ``description_json``
- ``build_date``
Solution-level attributes (written to each solution's root node):
- ``solution_type``
- ``solution_id``
- ``settings_json``
- ``outputs_json``
Args:
rig_root: Maya DAG path of the rig root node.
"""
import datetime
import maya.cmds as cmds # lazy — only needed inside Maya
import openrig
def _write_str_attr(node: str, attr: str, value: str) -> None:
if not cmds.attributeQuery(attr, node=node, exists=True):
cmds.addAttr(node, longName=attr, dataType="string")
cmds.setAttr(f"{node}.{attr}", value, type="string")
_write_str_attr(rig_root, "openrig_version", openrig.__version__)
_write_str_attr(rig_root, "description_json", json.dumps(self._description))
_write_str_attr(rig_root, "build_date", datetime.date.today().isoformat())
for solution_id, root_node in self._solution_roots.items():
settings_obj = self._solution_settings.get(solution_id)
settings_dict = (
dataclasses.asdict(settings_obj) # type: ignore[arg-type]
if dataclasses.is_dataclass(settings_obj)
else {}
)
outputs = self._outputs.get(solution_id, {})
_write_str_attr(
root_node,
"solution_type",
self._solution_types.get(solution_id, ""),
)
_write_str_attr(root_node, "solution_id", solution_id)
_write_str_attr(
root_node,
"settings_json",
json.dumps(settings_dict, default=_settings_default),
)
_write_str_attr(
root_node,
"outputs_json",
json.dumps(outputs, default=_settings_default),
)