Skip to content

redsun.ports

Slots

slot 

slot(fn: F) -> F

slot(
    *,
    name: str | None = ...,
    thread: SlotThread = ...,
    signal: str | Sequence[str] = ...,
) -> Callable[[F], F]

slot(
    fn: F | None = None,
    /,
    *,
    name: str | None = None,
    thread: SlotThread = None,
    signal: str | Sequence[str] = (),
) -> F | Callable[[F], F]

Mark a method as connectable to a signal.

A marked method is public API: its name and signature are what other components are connected against, and an unmarked method cannot be connected at all. async def methods may be marked too.

Parameters:

Name Type Description Default
fn F | None

The method, when the decorator is written bare. None when it is written with arguments, which returns the decorator itself.

 None
name str | None

Port name a configuration file addresses the method by. Defaults to the method name without leading underscores.

 None
thread SlotThread

Delivery thread, overriding the affinity the class declares.

 None
signal str | Sequence[str]

The signal, or the signals, that reach this slot when a session pairs its component with another: each is the attribute name of a signal of the other component. Without it, only the links a session lists reach the slot.

 ()

SlotThread module-attribute 

SlotThread: TypeAlias = (
    "Literal['main', 'current'] | Thread | None"
)

Thread a slot is delivered on, as accepted by psygnal.

Connections

Link: TypeAlias = (
    "tuple[SignalInstance | SignalR[Any], SlotCallable]"
)

A signal and the slot it reaches, as a session's wire yields it.

links_between(a: object, b: object) -> list[Link]

Return the links pairing two built components makes, both ways.

Each signal of a reaches each slot of b naming it in its signal, then each signal of b reaches each slot of a naming it, in the order ports lists them. Only those signals are matched, so a device signal never is. An empty list means nothing matched.

def wire(self) -> Iterator[Link]:
    yield from links_between(self.motor_widget, self.motor_ctrl)

Raises:

Type Description
ValueError

If a and b are one object.
WiringError

If either exposes two signals under one port name.

Connection dataclass 

Connection(
    *,
    publisher: str,
    publisher_port: str,
    consumer: str,
    consumer_port: str,
    thread: SlotThread = None,
)

A recorded link between a signal and a slot.

The signal is a psygnal signal of a component, or a signal of a device, whose publisher is the device and whose port is the signal's name within it.

publisher instance-attribute 

publisher: str

Name of the component or device that sends.

publisher_port instance-attribute 

publisher_port: str

Name of the signal within the publisher.

consumer instance-attribute 

consumer: str

Name of the component that receives.

consumer_port instance-attribute 

consumer_port: str

Port name of the slot within the consumer.

thread class-attribute instance-attribute 

thread: SlotThread = None

Thread the slot runs on. None is the thread that emits.

Unconnected dataclass 

Unconnected(
    *,
    signals: list[str] = list(),
    slots: list[str] = list(),
)

Ports of the built components that no connection reaches.

Each entry is a component.port path. A signal listed here emits into nothing; a slot listed here is never called.

signals class-attribute instance-attribute 

signals: list[str] = field(default_factory=list)

Paths of the signals nothing listens to.

slots class-attribute instance-attribute 

slots: list[str] = field(default_factory=list)

Paths of the slots nothing reaches.

ports 

ports(component: object) -> Ports

Return the signals and slots component exposes, by port name.

A signal is a public Signal attribute, or a member of a SignalGroup the component holds, in which case the member name is the port name. A slot is a method marked with slot.

Parameters:

Name Type Description Default
component object

The built component to inspect.

 required

Raises:

Type Description
WiringError

If two signals claim the same port name, which would leave the component unaddressable.

Ports dataclass 

Ports(
    signals: dict[str, SignalInstance] = dict(),
    slots: dict[str, Callable[..., Any]] = dict(),
)

The connectable surface of a component.

signals class-attribute instance-attribute 

signals: dict[str, SignalInstance] = field(
    default_factory=dict
)

Signals, by port name.

slots class-attribute instance-attribute 

slots: dict[str, Callable[..., Any]] = field(
    default_factory=dict
)

Slots, by port name.

Errors

WiringError

Bases: RuntimeError

Raised when a connection between two components cannot be made.

ComponentNotBuilt 

ComponentNotBuilt(component: str, message: str)

Bases: WiringError

Raised when a port path names a component that is not there.

component is the name the path used, so a caller that knows which components failed to build can tell one of those from a name that was never declared.