step.resource
=============

.. py:module:: step.resource

.. autoapi-nested-parse::

   Hardware resource tracking and conflict detection for parallel step execution.

   Every step declares which hardware resources it requires exclusive access to via
   ``required_resources()``.  Composite steps (Parallel, DoWhileActive) validate at
   construction time that no two concurrent branches claim the same resource.  A
   lightweight runtime ``ResourceManager`` acts as a safety net for dynamically
   constructed steps (e.g. ``Defer``) that cannot be checked statically.

   Resource identifiers are plain strings with the format ``"<type>"`` or
   ``"<type>:<qualifier>"``:

       "drive"       — chassis drive system
       "motor:0"     — motor on port 0
       "servo:3"     — servo on port 3
       "servo:*"     — all servo ports (used by FullyDisableServos)



Exceptions
----------

.. autoapisummary::

   step.resource.ResourceConflictError


Classes
-------

.. autoapisummary::

   step.resource.ResourceManager


Functions
---------

.. autoapisummary::

   step.resource.validate_no_overlap
   step.resource.get_resource_manager


Module Contents
---------------

.. py:exception:: ResourceConflictError(resource: str, holder: str, requester: str)

   Bases: :py:obj:`Exception`


   Raised when two concurrent steps require the same hardware resource.


   .. py:attribute:: resource


   .. py:attribute:: holder


   .. py:attribute:: requester


.. py:function:: validate_no_overlap(branches: Sequence[_HasResources], context: str = 'Parallel') -> None

   Check pairwise that no two branches claim the same resource.

   :param branches: Sequence of objects with a ``required_resources()`` method.
   :param context: Name used in the error message (e.g. ``"Parallel"``
                   or ``"DoWhileActive"``).

   :raises ResourceConflictError: If any two branches share a resource.


.. py:class:: ResourceManager

   Non-blocking, fail-fast resource lock attached to the robot instance.

   Since all steps run in a single asyncio event loop (cooperative
   multitasking), no real lock is needed — a simple dict suffices.


   .. py:method:: acquire(resources: frozenset[str], holder: str) -> None

      Claim *resources* for *holder*.  Raises on conflict.



   .. py:method:: release(resources: frozenset[str]) -> None

      Release previously acquired resources.



.. py:function:: get_resource_manager(robot: object) -> ResourceManager

   Get or create the ``ResourceManager`` attached to *robot*.