shyft.dashboard.base.ports

This module contains all the ports and connection functions

Functions

`check_port_compatibility`(*, port_sender, ...)	function to check if two ports can be connected
`connect_ports`(port_sender, port_receiver)	Function to connect a sender and a receiver port
`connect_ports_and_state_ports`(port_sender, ...)	Convenience wrapper function to connect a Sender and Receiver, in addition connect the state ports of the sender.parent and receiver.parent.
`connect_state_ports`(sender_state_port, ...)	Function to connect state ports
`disconnect_ports`(port_sender, port_receiver)	Function to disconnect a sender and a receiver port
`grep_ports`(*, cls, port_type)	find port of port_type in a given cls

Classes

`PortBase`(*, parent, name, func, p_type)	Base port function Both Sender and Receiver are derived from this class which makes the implementation of both port types symmetric.
`PortTypes`(value[, names, module, qualname, ...])	Defines the type of a Port
`Receiver`(*, parent, name, func, signal_type)
`ReceiverFunc`(*, parent, callback, signal_type)	Class for an receiver port function which buffers the received object (self.obj), where the obj type annotation can be set in runtime.
`Sender`(*, parent, name, signal_type)
`SenderFunc`(*, parent, signal_type)	Class for a Sender port function, where the type annotation of the send object can be set in runtime.
`StatePorts`(*, parent, _receive_state)
`States`(value[, names, module, qualname, ...])	States used as signal type by StatePorts

Exceptions

`PortConnectionError`	PortConnectionError derived from RuntimeError
`PortConnectionWarning`
`PortError`	PortError derived from RuntimeError

exception shyft.dashboard.base.ports.PortError[source]

Bases: RuntimeError

PortError derived from RuntimeError

exception shyft.dashboard.base.ports.PortConnectionError[source]

Bases: RuntimeError

PortConnectionError derived from RuntimeError

exception shyft.dashboard.base.ports.PortConnectionWarning[source]: Bases: RuntimeWarning

class shyft.dashboard.base.ports.States(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

States used as signal type by StatePorts

LOADING = 1

INVALID = 2

READY = 3

ACTIVE = 4

DEACTIVE = 5

PROCESSING = 6

EXPECTING_DATA = 7

class shyft.dashboard.base.ports.ReceiverFunc(*, parent: portBase, callback: Callable[[Any], None], signal_type: str | Any)[source]

Bases: object

Class for an receiver port function which buffers the received object (self.obj), where the obj type annotation can be set in runtime. The sender function also saves the timestamp whenever the last obj was send.

__init__(*, parent: portBase, callback: Callable[[Any], None], signal_type: str | Any) → None[source]

self.timestamp: timestamp since epoch of last received object

Parameters:

parent – Port class it belongs to
callback – callback function which call on receiving an object
signal_type – type annotation of the object to receive

__call__(obj: Any) → None[source]

Buffers the object when receiving it

Parameters:: obj – Any signal to pass through

class shyft.dashboard.base.ports.SenderFunc(*, parent: portBase, signal_type: str | Any)[source]

Bases: object

Class for a Sender port function, where the type annotation of the send object can be set in runtime. The sender function also saves the timestamp whenever the last obj was send.

__init__(*, parent: portBase, signal_type: str | Any) → None[source]

self.timestamp: timestamp since epoch of last send object

Parameters:

parent – Port class it belongs to
signal_type – type annotation of the object to receive

__call__(signal: Any) → Any[source]: Call self as a function.

class shyft.dashboard.base.ports.PortTypes(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Defines the type of a Port

SENDER = 1

RECEIVER = 2

class shyft.dashboard.base.ports.PortBase(*, parent: Any, name: str, func: ReceiverFunc | SenderFunc, p_type: PortTypes)[source]

Bases: Hashable

Base port function Both Sender and Receiver are derived from this class which makes the implementation of both port types symmetric. The main difference between both are that they have a different func: ReceiverFunc, SenderFunc respectively. A Receiver will not have connected_ports and connected functions.

__init__(*, parent: Any, name: str, func: ReceiverFunc | SenderFunc, p_type: PortTypes) → None[source]

Parameters:

parent – instance in which the port resides
name – name of the sender port
func – function to be called to receive an the signal object
p_type – type of the port: sender or receiver

__call__(*args, **kwargs)[source]

The call function calls first its internal func representation (ReceiverFunc or SenderFunc), and in case the type is a Sender, all connected functions are called.

Parameters:

args – signal to send or receive
kwargs – signal to send or receive

Raises:

PortConnectionError: – if the Sender is connected in a closed loop, and while sending the information comes back!

property signal_type: Any

property connected: bool: Property returns if Port is connected

class shyft.dashboard.base.ports.Receiver(*, parent: Any, name: str, func: Callable, signal_type: str | Any)[source]

Bases: PortBase

__init__(*, parent: Any, name: str, func: Callable, signal_type: str | Any) → None[source]

Receiver, can be connected to a Sender with the connect_ports function.

NB: the callback ‘func’ must have 1 parameter type annotated with ‘signal_type’

Class instances of Receiver ports should be named with the following pattern:

self.receive_{what_will_be_received} = Receiver(…)

The receiver function of the Receiver port, should have the same name as it’s receiver starting with _:

def __init__(self):
self.receive_selections = Receiver(…, func=self._receive_selections, …)

def _receive_selections(self, ….):
…

Parameters:

parent – instance in which the port resides
name – name of the sender port
func – function to be called to receive an the signal object
signal_type – type annotation of the object to be received, if not provided type annotation of func is used

Examples

# receiver function for signal_type: int

def _receive_number(number: int) -> None: # create a simple receiving function

print(“received “, number)

receive_number = Receiver(parent=’main.py’, name=’receive number’, func=_receiver_func, signal_type=int)

class shyft.dashboard.base.ports.Sender(*, parent: Any, name: str, signal_type: str | Any)[source]

Bases: PortBase

__init__(*, parent: Any, name: str, signal_type: str | Any) → None[source]

Sender Port, can be connected to a receiver port. Signal_type is the type annotation of the object to be send. It must correspond with the receiver port function object type annotation.

Class instances of Receiver ports should be named with the following pattern:

self.send_{what_will_be_send} = Sender(…)

Parameters:

parent – instance in which the port resides
name – name of the sender port
signal_type – type annotation of the object to be send

Examples

# sender function for signal_type: int

send_number = Sender(parent=’main.py’, name=’receive number’,  signal_type=int)

connect(receiver: Receiver) → None[source]

Connect a Receiver to the Sender

Parameters:: receiver – Receiver port which should be connected to

disconnect(receiver: Receiver) → None[source]

Disconnect a Receiver from the Sender

Parameters:: receiver – Receiver port which should be connected to

shyft.dashboard.base.ports.check_port_compatibility(*, port_sender: Sender, port_receiver: Receiver) → bool[source]

function to check if two ports can be connected

Raises:

PortConnectionError – if ports are incompatible:

Parameters:

port_sender – sender port to verify
port_receiver – receiver port to verify

Return type:

returns True if compatible

shyft.dashboard.base.ports.connect_ports(port_sender: Sender, port_receiver: Receiver) → bool[source]

Function to connect a sender and a receiver port

Raises:

PortConnectionError: – if ports are incompatible
PortConnectionWarning: – if ports are already connected

Parameters:

port_sender – sender port to connect
port_receiver – receiver port to connect

Return type:

returns True if connected

shyft.dashboard.base.ports.disconnect_ports(port_sender: Sender, port_receiver: Receiver) → bool[source]

Function to disconnect a sender and a receiver port

Parameters:

port_sender – sender port to connect
port_receiver – receiver port to connect

Return type:

returns True if disconnected

class shyft.dashboard.base.ports.StatePorts(*, parent: Any, _receive_state: Callable[[States], None])[source]

Bases: object

__init__(*, parent: Any, _receive_state: Callable[[States], None]) → None[source]

State ports are 2 ports dedicated to send and receive States beweteen apps. For convenience sender and receiver ports are combined here.

Parameters:

parent – class/widget the port belongs to
_receive_state – Callable receiver function, with type annotation state: States

Examples

state_ports = StatePort(parent=parent, _receive_state=_receive_state)

def _receive_state(states: States) -> None:

if state == States.ACTIVE:

print(state)

shyft.dashboard.base.ports.grep_ports(*, cls: Any, port_type: Receiver | Sender | StatePorts) → Dict[str, Receiver | Sender | StatePorts][source]

find port of port_type in a given cls

Parameters:

cls – cls parse for ports
port_type – port type to search

Returns:

returns dict with Dict[name

Return type:

str, port:Union[Receiver, Sender, StatePorts]] with the found ports

shyft.dashboard.base.ports.connect_state_ports(sender_state_port: StatePorts, receiver_state_port: StatePorts) → bool[source]

Function to connect state ports

Raises:

PortConnectionError: – if not possible to connect

Parameters:

sender_state_port – sender state port
receiver_state_port – receiver state port

Return type:

True if connected successfully

shyft.dashboard.base.ports.connect_ports_and_state_ports(port_sender: Sender, port_receiver: Receiver) → Tuple[bool, bool][source]

Convenience wrapper function to connect a Sender and Receiver, in addition connect the state ports of the sender.parent and receiver.parent.

Both parents must have a StatePort, the Sender.parent.state_port is treat as sender_state_port!

Raises:

PortConnectionError: – if ports are incompatible
PortConnectionWarning: – if ports are already connected

Parameters:

port_sender – sender port to connect
port_receiver – receiver port to connect

Return type:

returns True if connected