magscope.ipc

magscope.ipc#

Exceptions#

`CommandRegistrationError`	Base error for command registration problems.
`CommandConflictError`	Raised when a command is registered more than once with incompatible metadata.
`MissingCommandHandlerError`	Raised when a handler is missing for a registered command.
`UnknownCommandError`	Raised when dispatch is attempted for an unknown command.

Classes#

`Delivery`	Enum where members are also (and must be) strings
`CommandSpec`
`HandlerRegistration`
`CommandRegistry`	Registry mapping IPC command types to their handlers and destinations.

Functions#

`register_ipc_command`(command_type, *[, delivery, target])	Decorator to associate an IPC command type with a handler method.
`_collect_handler_registrations`(...)	Yield command registrations declared on `cls` and its bases.
`command_kwargs`(→ dict[str, object])	Return the payload of a command as keyword arguments.
`create_pipes`(→ tuple[dict[str, ...)	Create duplex pipes for each managed process.
`broadcast_command`(→ None)	Send a command to all running, non-quitting processes.
`drain_pipe_until_quit`(→ None)	Drain a pipe until the paired quit event is set.

Module Contents#

class magscope.ipc.Delivery[source]#

Bases: enum.StrEnum

Enum where members are also (and must be) strings

DIRECT = 'direct'[source]#

BROADCAST = 'broadcast'[source]#

MAG_SCOPE = 'mag_scope'[source]#

exception magscope.ipc.CommandRegistrationError[source]#

Bases: RuntimeError

Base error for command registration problems.

exception magscope.ipc.CommandConflictError[source]#

Bases: CommandRegistrationError

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

exception magscope.ipc.MissingCommandHandlerError[source]#

Bases: CommandRegistrationError

Raised when a handler is missing for a registered command.

exception magscope.ipc.UnknownCommandError[source]#

Bases: RuntimeError

Raised when dispatch is attempted for an unknown command.

class magscope.ipc.CommandSpec[source]#

command_type: type[magscope.ipc_commands.Command][source]#

handler: str[source]#

target: str[source]#

delivery: Delivery[source]#

class magscope.ipc.HandlerRegistration[source]#

command_type: type[magscope.ipc_commands.Command][source]#

handler: str[source]#

delivery: Delivery[source]#

target_override: str | None = None[source]#

magscope.ipc.register_ipc_command(command_type: type[magscope.ipc_commands.Command], *, delivery: Delivery = Delivery.DIRECT, target: str | None = None)[source]#: Decorator to associate an IPC command type with a handler method.

magscope.ipc._collect_handler_registrations(cls: Type) → Iterable[HandlerRegistration][source]#: Yield command registrations declared on cls and its bases.

magscope.ipc.command_kwargs(command: magscope.ipc_commands.Command) → dict[str, object][source]#: Return the payload of a command as keyword arguments.

class magscope.ipc.CommandRegistry[source]#

Registry mapping IPC command types to their handlers and destinations.

_specs: dict[type[magscope.ipc_commands.Command], CommandSpec][source]#

_handler_index: dict[tuple[str, str], type[magscope.ipc_commands.Command]][source]#

register(*, command_type: type[magscope.ipc_commands.Command], handler: str, owner: Type, delivery: Delivery, target: str) → None[source]#: Register a command handler.

register_manager(manager: magscope.processes.ManagerProcessBase) → None[source]#: Register all decorated command handlers on manager.

register_object(obj: object, *, target: str | None = None) → None[source]#: Register decorated handlers on arbitrary objects (e.g., MagScope).

route_for(command: magscope.ipc_commands.Command) → CommandSpec[source]#: Return the route information for command.

handlers_for_target(target: str) → dict[type[magscope.ipc_commands.Command], CommandSpec][source]#: Return handler specs applicable to target.

validate_targets(processes: Mapping[str, magscope.processes.ManagerProcessBase]) → None[source]#: Ensure every command has a reachable target and handler.

command_for_handler(owner: str, handler: str) → type[magscope.ipc_commands.Command][source]#: Return the command type bound to owner.handler.

magscope.ipc.create_pipes(processes: Mapping[str, magscope.processes.ManagerProcessBase]) → tuple[dict[str, multiprocessing.connection.Connection], dict[str, multiprocessing.connection.Connection]][source]#

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.

magscope.ipc.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[source]#: Send a command to all running, non-quitting processes.

magscope.ipc.drain_pipe_until_quit(pipe: multiprocessing.connection.Connection, quitting_event: multiprocessing.synchronize.Event, *, poll_interval: float | None = 0.0) → None[source]#: Drain a pipe until the paired quit event is set.