task_registry

orchard.core.task_registry

Task Component Registry.

Maps task_type strings to frozen bundles of task-specific strategy implementations. Each bundle (:class:TaskComponents) carries the three pluggable dimensions defined in :mod:orchard.core.task_protocols:

• criterion factory
• validation metrics
• eval pipeline (inference + visualization + reporting)

The registry is populated at import time by orchard.tasks.<task_type> packages and exposed as an immutable :class:~types.MappingProxyType via :func:get_registry.

Usage::

from orchard.core.task_registry import get_task

task = get_task("classification")
criterion = task.criterion_factory.get_criterion(training_cfg)

TaskComponents(criterion_factory, validation_metrics, eval_pipeline) dataclass

Immutable bundle of task-specific strategy implementations.

Attributes:

Name	Type	Description
criterion_factory	TaskCriterionFactory	Builds the loss function for this task.
validation_metrics	TaskValidationMetrics	Computes per-epoch validation metrics.
eval_pipeline	TaskEvalPipeline	Orchestrates inference, visualization, and reporting.

register_task(task_type, components)

Register a task-specific component bundle.

Parameters:

Name	Type	Description	Default
task_type	str	Identifier matching a TaskType literal (e.g. "classification").	required
components	TaskComponents	Frozen bundle of strategy implementations.	required

Raises:

Type	Description
OrchardConfigError	If task_type is already registered.

Source code in orchard/core/task_registry.py

def register_task(task_type: str, components: TaskComponents) -> None:
    """
    Register a task-specific component bundle.

    Args:
        task_type: Identifier matching a ``TaskType`` literal
            (e.g. ``"classification"``).
        components: Frozen bundle of strategy implementations.

    Raises:
        OrchardConfigError: If ``task_type`` is already registered.
    """
    if task_type in _TASK_REGISTRY:
        raise OrchardConfigError(
            f"Task '{task_type}' is already registered. "
            "Duplicate registration indicates a plugin conflict."
        )
    _TASK_REGISTRY[task_type] = components

get_task(task_type)

Retrieve the component bundle for a given task type.

Parameters:

Name	Type	Description	Default
task_type	str	Registered task identifier.	required

Returns:

Type	Description
TaskComponents	The corresponding :class:TaskComponents bundle.

Raises:

Type	Description
OrchardConfigError	If task_type has not been registered.

Source code in orchard/core/task_registry.py

def get_task(task_type: str) -> TaskComponents:
    """
    Retrieve the component bundle for a given task type.

    Args:
        task_type: Registered task identifier.

    Returns:
        The corresponding :class:`TaskComponents` bundle.

    Raises:
        OrchardConfigError: If ``task_type`` has not been registered.
    """
    if task_type not in _TASK_REGISTRY:
        available = sorted(_TASK_REGISTRY.keys())
        raise OrchardConfigError(f"Unknown task_type '{task_type}'. Registered: {available}")
    return _TASK_REGISTRY[task_type]

get_registry()

Return an immutable view of the full task registry.

Returns:

Type	Description
MappingProxyType[str, TaskComponents]	Read-only mapping of task type strings to component bundles.

Source code in orchard/core/task_registry.py

def get_registry() -> MappingProxyType[str, TaskComponents]:
    """
    Return an immutable view of the full task registry.

    Returns:
        Read-only mapping of task type strings to component bundles.
    """
    return MappingProxyType(_TASK_REGISTRY)