magscope.ipc
============

.. py:module:: magscope.ipc


Exceptions
----------

.. autoapisummary::

   magscope.ipc.CommandRegistrationError
   magscope.ipc.CommandConflictError
   magscope.ipc.MissingCommandHandlerError
   magscope.ipc.UnknownCommandError


Classes
-------

.. autoapisummary::

   magscope.ipc.Delivery
   magscope.ipc.CommandSpec
   magscope.ipc.HandlerRegistration
   magscope.ipc.CommandRegistry


Functions
---------

.. autoapisummary::

   magscope.ipc.register_ipc_command
   magscope.ipc._collect_handler_registrations
   magscope.ipc.command_kwargs
   magscope.ipc.create_pipes
   magscope.ipc.broadcast_command
   magscope.ipc.drain_pipe_until_quit


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

.. py:class:: Delivery

   Bases: :py:obj:`enum.StrEnum`


   Enum where members are also (and must be) strings


   .. py:attribute:: DIRECT
      :value: 'direct'



   .. py:attribute:: BROADCAST
      :value: 'broadcast'



   .. py:attribute:: MAG_SCOPE
      :value: 'mag_scope'



.. py:exception:: CommandRegistrationError

   Bases: :py:obj:`RuntimeError`


   Base error for command registration problems.


.. py:exception:: CommandConflictError

   Bases: :py:obj:`CommandRegistrationError`


   Raised when a command is registered more than once with incompatible metadata.


.. py:exception:: MissingCommandHandlerError

   Bases: :py:obj:`CommandRegistrationError`


   Raised when a handler is missing for a registered command.


.. py:exception:: UnknownCommandError

   Bases: :py:obj:`RuntimeError`


   Raised when dispatch is attempted for an unknown command.


.. py:class:: CommandSpec

   .. py:attribute:: command_type
      :type:  type[magscope.ipc_commands.Command]


   .. py:attribute:: handler
      :type:  str


   .. py:attribute:: target
      :type:  str


   .. py:attribute:: delivery
      :type:  Delivery


.. py:class:: HandlerRegistration

   .. py:attribute:: command_type
      :type:  type[magscope.ipc_commands.Command]


   .. py:attribute:: handler
      :type:  str


   .. py:attribute:: delivery
      :type:  Delivery


   .. py:attribute:: target_override
      :type:  str | None
      :value: None



.. py:function:: register_ipc_command(command_type: type[magscope.ipc_commands.Command], *, delivery: Delivery = Delivery.DIRECT, target: str | None = None)

   Decorator to associate an IPC command type with a handler method.


.. py:function:: _collect_handler_registrations(cls: Type) -> Iterable[HandlerRegistration]

   Yield command registrations declared on ``cls`` and its bases.


.. py:function:: command_kwargs(command: magscope.ipc_commands.Command) -> dict[str, object]

   Return the payload of a command as keyword arguments.


.. py:class:: CommandRegistry

   Registry mapping IPC command types to their handlers and destinations.


   .. py:attribute:: _specs
      :type:  dict[type[magscope.ipc_commands.Command], CommandSpec]


   .. py:attribute:: _handler_index
      :type:  dict[tuple[str, str], type[magscope.ipc_commands.Command]]


   .. py:method:: register(*, command_type: type[magscope.ipc_commands.Command], handler: str, owner: Type, delivery: Delivery, target: str) -> None

      Register a command handler.



   .. py:method:: register_manager(manager: magscope.processes.ManagerProcessBase) -> None

      Register all decorated command handlers on ``manager``.



   .. py:method:: register_object(obj: object, *, target: str | None = None) -> None

      Register decorated handlers on arbitrary objects (e.g., MagScope).



   .. py:method:: route_for(command: magscope.ipc_commands.Command) -> CommandSpec

      Return the route information for ``command``.



   .. py:method:: handlers_for_target(target: str) -> dict[type[magscope.ipc_commands.Command], CommandSpec]

      Return handler specs applicable to ``target``.



   .. py:method:: validate_targets(processes: Mapping[str, magscope.processes.ManagerProcessBase]) -> None

      Ensure every command has a reachable target and handler.



   .. py:method:: command_for_handler(owner: str, handler: str) -> type[magscope.ipc_commands.Command]

      Return the command type bound to ``owner.handler``.



.. py:function:: create_pipes(processes: Mapping[str, magscope.processes.ManagerProcessBase]) -> tuple[dict[str, multiprocessing.connection.Connection], dict[str, multiprocessing.connection.Connection]]

   Create duplex pipes for each managed process.

   Returns a pair of dictionaries mapping process names to the parent and child
   pipe ends, respectively. The parent ends are intended to be owned by the
   coordinating ``MagScope`` instance while the child ends are passed to
   individual manager processes.


.. py:function:: broadcast_command(command: magscope.ipc_commands.Command, *, pipes: Mapping[str, multiprocessing.connection.Connection], processes: Mapping[str, magscope.processes.ManagerProcessBase], quitting_events: Mapping[str, multiprocessing.synchronize.Event]) -> None

   Send a command to all running, non-quitting processes.


.. py:function:: drain_pipe_until_quit(pipe: multiprocessing.connection.Connection, quitting_event: multiprocessing.synchronize.Event, *, poll_interval: float | None = 0.0) -> None

   Drain a pipe until the paired quit event is set.