orchestrata_ultimate.cxp.priority

Reference › orchestrata_ultimate › cxp › priority

PriorityManager implements the priority rules of Spec 004 US-009 for CXPPacket objects, with the range taken from constants (0 to 10). sort_by_priority returns a new list ordered by descending priority and, within equal priority, by ascending timestamp, so older packets go first. get_priority_level maps a number to a level name (background, low, normal, high or urgent), using the closest lower named level for a value that has no name of its own, and raises InvalidPacketError for a value outside the range. get_default_priority returns the default for a packet type, falling back to normal. validate_priority checks the range and clamp_priority forces a value into it. See cxp and exceptions.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PriorityManager

Decorators: traces(spec='004', story='US-009', reqs=['REQ-004-US-009-005', 'REQ-004-US-009-006', 'REQ-004-US-009-004']) · Line 53

Manages CXP packet priority ordering and classification.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/cxp/priority.py

"""
CXP Priority Handling.

Implements priority ordering, level mapping, and default priority
policies for CXP packets per Spec 004 Section US-009.

Author: Mohamed
"""

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, PacketType, Priority
from orchestrata_ultimate.cxp.constants import PRIORITY_MAX, PRIORITY_MIN
from orchestrata_ultimate.cxp.exceptions import InvalidPacketError

# ============================================================================
# Priority Level Mapping
# ============================================================================

PRIORITY_NAMES: dict[int, str] = {
    Priority.BACKGROUND.value: "background",  # 0
    Priority.LOW.value: "low",  # 3
    Priority.NORMAL.value: "normal",  # 5
    Priority.HIGH.value: "high",  # 7
    Priority.URGENT.value: "urgent",  # 10
}

# ============================================================================
# Default Priorities by Packet Type
# ============================================================================

DEFAULT_PRIORITIES: dict[PacketType, int] = {
    PacketType.MESSAGE: Priority.NORMAL.value,
    PacketType.QUERY: Priority.NORMAL.value,
    PacketType.COMMAND: Priority.NORMAL.value,
    PacketType.RESPONSE: Priority.NORMAL.value,
    PacketType.ACKNOWLEDGEMENT: Priority.NORMAL.value,
    PacketType.ERROR: Priority.HIGH.value,
    PacketType.HEARTBEAT: Priority.URGENT.value,
    PacketType.NOTIFICATION: Priority.LOW.value,
}


# ============================================================================
# PriorityManager
# ============================================================================


@traces(
    spec="004",
    story="US-009",
    reqs=["REQ-004-US-009-005", "REQ-004-US-009-006", "REQ-004-US-009-004"],
)
class PriorityManager:
    """Manages CXP packet priority ordering and classification.

    Ordering rule (Spec 004 US-009):
        Higher priority first.  Among equal priorities, older packets
        (earlier timestamp) come first (FIFO within a priority band).
    """

    @traces(
        spec="004",
        story="US-009",
        reqs=[
            "REQ-004-US-009-005",
            "REQ-004-US-009-006",
        ],
    )
    def sort_by_priority(self, packets: list[CXPPacket]) -> list[CXPPacket]:
        """Sort packets by descending priority, then ascending timestamp.

        Args:
            packets: List of CXP packets to sort.

        Returns:
            New list sorted by priority (highest first), then by
            timestamp (oldest first) for packets at the same level.
        """
        return sorted(
            packets,
            key=lambda p: (-p.priority, p.timestamp),
        )

    @traces(spec="004", story="US-009", reqs=["REQ-004-US-009-002"])
    def get_priority_level(self, priority: int) -> str:
        """Return the human-readable level name for a numeric priority.

        If the value does not exactly match a named level, the closest
        lower named level is returned.

        Args:
            priority: Integer priority value (0-10).

        Returns:
            Level name string (background/low/normal/high/urgent).

        Raises:
            InvalidPacketError: If priority is outside the 0-10 range.
        """
        if not self.validate_priority(priority):
            raise InvalidPacketError(
                f"Priority {priority} is outside the valid range "
                f"[{PRIORITY_MIN}, {PRIORITY_MAX}]",
                field="priority",
            )

        # Exact match
        if priority in PRIORITY_NAMES:
            return PRIORITY_NAMES[priority]

        # Find closest lower named level
        thresholds = sorted(PRIORITY_NAMES.keys())
        level_name = PRIORITY_NAMES[thresholds[0]]
        for threshold in thresholds:
            if threshold <= priority:
                level_name = PRIORITY_NAMES[threshold]
            else:
                break
        return level_name

    @traces(
        spec="004",
        story="US-009",
        reqs=[
            "REQ-004-US-009-014",
            "REQ-004-US-009-015",
            "REQ-004-US-009-016",
            "REQ-004-US-009-017",
        ],
    )
    def get_default_priority(self, packet_type: PacketType) -> int:
        """Return the default priority for a given packet type.

        Args:
            packet_type: The CXP packet type.

        Returns:
            Default priority integer for that type.
        """
        return DEFAULT_PRIORITIES.get(packet_type, Priority.NORMAL.value)

    @traces(
        spec="004",
        story="US-009",
        reqs=[
            "REQ-004-US-009-004",
            "REQ-004-US-009-013",
        ],
    )
    def validate_priority(self, priority: int) -> bool:
        """Check whether a priority value is within the valid range.

        Args:
            priority: Integer to validate.

        Returns:
            True if within [PRIORITY_MIN, PRIORITY_MAX], False otherwise.
        """
        return PRIORITY_MIN <= priority <= PRIORITY_MAX

    @traces(spec="004", story="US-009", reqs=["REQ-004-US-009-013"])
    def clamp_priority(self, priority: int) -> int:
        """Clamp a priority value to the valid range [0, 10].

        Args:
            priority: Integer to clamp.

        Returns:
            Clamped integer within [PRIORITY_MIN, PRIORITY_MAX].
        """
        return max(PRIORITY_MIN, min(PRIORITY_MAX, priority))