Modules

Module slip

The slip module contains configuration settings and lower-level functions and classes that are mainly useful for extending the sliplib package.

The configuration settings, functions, classes, and exception in this module can also be imported directly from the sliplib package.

Configuration

config

The configuration settings are stored in the config configuration object. The config object provides constants, configuration settings, and a context manager for sliplib.

Constants

config.END = b'\xc0'

Type:    bytes

The SLIP END byte. Used to delimit SLIP packages.

config.ESC = b'\xdb'

Type:    bytes

The SLIP ESC byte. Used to escape ESC and END bytes in the message.

config.ESC_END = b'\xdc'

Type:    bytes

The escaped value of an END byte.

config.ESC_ESC = b'\xdd'

Type:    bytes

The escaped value of an ESC byte.

For backwards compatibility reasons these constants can also be imported directly from the top-level sliplib module.

Settings

config.USE_LEADING_END_BYTE = False

Type:    bool

Indicates if a leading END byte must be sent.

Context Manager

config.use_leading_end_byte(value)

Temporarily modify the value of USE_LEADING_END_BYTE.

This context manager ensures that any Driver and SlipWrapper instances that are defined in its body use a specific value for USE_LEADING_END_BYTE. This is particularly useful when the application interacts with different endpoints, some that require a leading END byte, and others that cannot handle multiple subsequent END bytes. By using this context manager, the order of the creation of the instances can be modified without having to worry about the current value of USE_LEADING_END_BYTE. Example:

with use_leading_end_byte(True):
    slip_socket = SlipSocket(sock)  # Where sock is a previously created socket.
# Calling `slip_socket.send_msg(message)` will send the encoded message
# with both a leading and trailing END byte.

Added in version 0.7.0.

Note

The temporary value of USE_LEADING_END_BYTE does not propagate to processes that are created or started in the body of the context manager.

Warning

USE_LEADING_END_BYTE is a global setting. For that reason, the temporary value of USE_LEADING_END_BYTE is protected from modification by other threads. This means that inside the body of this context manager, you should e.g. not wait for notifications from another thread when that other thread also uses this context manager. As an example, the following will deadlock because the spawned thread will never enter the body of the context manager.

def make_server(address, handler_class, event):
    with config.use_leading_end_byte(False):
        slip_server = SlipServer(address, handler_class)
        slip_server.handle_request()
    event.set()


with config.use_leading_end_byte(True):
    event = threading.Event()
    server_thread = threading.Thread(
        target=self.server,
        args=(address, SlipRequestHandler, event),
    )
    server_thread.start()
    event.wait()
    client = SlipSocket.make_client(address)

As a general rule, the body of the context manager should only contain statements that directly create Driver and/or SlipWrapper instances. The above example should be rewritten to something like:

def make_server(address, handler_class, event):
    with config.use_leading_end_byte(False):
        slip_server = SlipServer(address, handler_class)
    slip_server.handle_request()
    event.set()


event = threading.Event()
server_thread = threading.Thread(
    target=self.server,
    args=(address, SlipRequestHandler, event),
)
server_thread.start()
event.wait()
with config.use_leading_end_byte(True):
    client = SlipSocket.make_client(address)

Parameters:

value (bool) – The temporary value of USE_LEADING_END_BYTE.

Return type:

ContextManager [None]

The config configuration object and its constants END, ESC, ESC_END, and ESC_ESC, as well as the use_leading_end_byte() context manager can also be imported directly from the sliplib module.

Functions

The following are lower-level functions that should normally not be used directly.

encode(msg)

Encode a message (a byte sequence) into a SLIP-encoded packet.

This function replaces any END or ESC byte in the message with their SLIP-escaped value.

Parameters:

msg (bytes) – The message that must be encoded

Return type:

bytes

Returns:

The SLIP-encoded message

Changed in version 0.7.0: Leading and/or trailing END bytes are no longer included in the return value. As of version 0.7.1, the original version of the encode() function is available in the legacy module.

decode(packet)

Retrieve the message from the SLIP-encoded packet.

This function replaces any escaped SLIP bytes with their original values

Parameters:

packet (bytes) – The SLIP-encoded message. Note that this must be exactly one complete packet, without any leading and/or trailing END bytes. The decode() function does not provide any buffering for incomplete packages, nor does it provide support for decoding data with multiple packets.

Return type:

bytes

Returns:

The decoded message

Raises:

ProtocolError – if the packet contains an invalid byte sequence.

Changed in version 0.7.0: Leading and/or trailing END bytes are no longer allowed. As of version 0.7.1, the original version of the decode() function is available in the legacy module.

is_valid(packet)

Indicate if the packet’s contents conform to the SLIP specification.

A packet is valid if:

• It contains no END bytes, and

• Each ESC byte is followed by either an ESC_END or an ESC_ESC byte.

Parameters:

packet (bytes) – The packet to inspect.

Return type:

bool

Returns:

True if the packet is valid, False otherwise

Changed in version 0.7.0: Leading and/or trailing END bytes are no longer allowed.

Classes

class Driver

Handle the SLIP-encoding and decoding of messages.

Class Driver offers the following methods:

send(message)

Encode a message into a SLIP-encoded packet.

The message can be any arbitrary byte sequence.

Parameters:

message (bytes) – The message that must be encoded.

Return type:

bytes

Returns:

A packet with the SLIP-encoded message, including any required leading and trailing END bytes.

receive(data)

Extract and buffer SLIP packets.

Processes data, which must be a bytes-like object, and extracts and buffers the SLIP packets contained therein.

A non-terminated SLIP packet in data is also buffered, and extended with the next call to receive().

Parameters:

data (Union[bytes, int]) –

A bytes-like object to be processed.

An empty data parameter indicates that no more data will follow.

To accommodate iteration over byte sequences, an integer in the range(0, 256) is also accepted.

Return type:

None

Changed in version 0.7: receive() no longer returns a list of decoded messages.

get(*, block=True, timeout=None)

Get the next decoded message.

Remove and decode a SLIP packet from the internal buffer, and return the resulting message. If block is True and timeout is None (the default), then this method blocks until a message is available. If timeout is a positive number, the blocking will last for at most timeout seconds, and the method will return None if no message became available in that time. If block is False, the method returns immediately with either a message or None.

Note

block and timeout are keyword-only parameters.

Parameters:

• block (bool) – If True, then block for at most timeout seconds. Otherwise, return immediately.

• timeout (Optional[float]) – The number of seconds to wait for a message to become available.

Return type:

Optional[bytes]

Returns:

• None if no message is available,

• a decoded SLIP message, or

• an empty bytestring b"" if no further messages will come available.

Raises:

ProtocolError – When the packet that contained the message had an invalid byte sequence.

Added in version 0.7.

property sends_leading_end_byte

Indicates whether this Driver instance sends a leading END byte.

Added in version 0.7.0.

Return type:

bool

Exceptions

exception ProtocolError

Exception to indicate that a SLIP protocol error has occurred.

This exception is raised when an attempt is made to decode a packet with invalid bytes or byte sequences. Invalid bytes are END bytes or a trailing ESC byte. An invalid byte sequence is an ESC byte followed by any byte that is not an ESC_ESC or ESC_END byte.

The ProtocolError carries the invalid packet as the first (and only) element in its args tuple.

Module slipwrapper

The slipwrapper module contains the abstract base class SlipWrapper that is the basis for concrete implementations that combine the SLIP protocol with a byte stream.

The SlipWrapper class can also be imported directly from the sliplib package.

ByteStream = TypeVar(ByteStream)

Invariant TypeVar.

ByteStream represents a generic byte stream.

class SlipWrapper(stream)

Bases: ABC, Generic[ByteStream]

Abstract base class that provides a message based interface to a byte stream.

SlipWrapper combines a Driver instance with a (generic) byte stream. The SlipWrapper class is an abstract base class. It offers the methods send_msg() and recv_msg() to send and receive single messages over the byte stream, but it does not of itself provide the means to interact with the stream.

To interact with a concrete stream, a derived class must implement the methods send_bytes() and recv_bytes() to write to and read from the stream.

A SlipWrapper instance can be iterated over. Each iteration will provide the next message that is received from the byte stream.

Changed in version 0.5: Allow iteration over a SlipWrapper instance.

To instantiate a SlipWrapper, the user must provide an existing byte stream.

Parameters:

stream (ByteStream) – The byte stream that will be wrapped.

Class SlipWrapper offers the following methods and attributes:

recv_msg()

Receive a single message from the stream.

Return type:

bytes

Returns:

A SLIP-decoded message

Raises:

ProtocolError – when a SLIP protocol error has been encountered. A subsequent call to recv_msg() (after handling the exception) will return the message from the next packet.

send_msg(message)

Send a SLIP-encoded message over the stream.

Parameters:

message (bytes) – The message to encode and send

Return type:

None

driver

The SlipWrapper’s Driver instance.

stream

The wrapped ByteStream.

In addition, SlipWrapper requires that derived classes implement the following methods:

abstractmethod recv_bytes()

Receive data from the stream.

Derived classes must implement this method.

Note

The convention used within the SlipWrapper class is that recv_bytes() returns an empty bytes object b"" to indicate that the end of the byte stream has been reached and no further data will follow. Derived implementations must ensure that this convention is followed.

Return type:

bytes

Returns:

The bytes received from the stream

abstractmethod send_bytes(packet)

Send a packet over the stream.

Derived classes must implement this method.

Parameters:

packet (bytes) – the packet to send over the stream

Return type:

None

Module slipstream

The slipstream module contains the class SlipStream. The SlipStream class can also be imported directly from the sliplib package.

protocol IOStream

Bases: Protocol

Protocol class for wrappable byte streams.

Any object that produces and consumes a byte stream and contains the two required methods can be used. Typically, an IOStream is a subclass of io.RawIOBase, io.BufferedIOBase, io.FileIO, or similar classes, but this is not required.

Classes that implement this protocol must have the following methods / attributes:

read(chunksize)

Read chunksize bytes from the stream.

Parameters:

chunksize (int) – The number of bytes to read from the IOStream.

Return type:

bytes

Returns:

The bytes read from the IOStream. The number of bytes read may be less than the number specified by chunksize.

write(data)

Write data to the stream.

Parameters:

data (bytes) – The bytes to write on to IOStream.

Return type:

int

Returns:

The number of bytes actually written. This may be less than the number of bytes contained in data.

class SlipStream(stream, chunk_size=8192)

Bases: SlipWrapper[IOStream]

Class that wraps an IO stream with a Driver.

SlipStream combines a Driver instance with a concrete byte stream. The byte stream must support the methods read() and write(). To avoid conflicts and ambiguities caused by different newline conventions, streams that have an encoding attribute (such as io.StringIO objects, or text files that are not opened in binary mode) are not accepted as a byte stream.

The SlipStream class has all the methods and attributes from its base class SlipWrapper. In addition, it directly exposes all methods and attributes of the contained stream, except for the following:

• read*() and write*(). These methods are not supported, because byte-oriented read and write operations would invalidate the internal state maintained by SlipStream.

• Similarly, seek(), tell(), and truncate() are not supported, because repositioning or truncating the stream would invalidate the internal state.

• raw(), detach() and other methods that provide access to or manipulate the stream’s internal data.

Instead of the read*() and write*() methods a SlipStream object provides the method recv_msg() and send_msg() to read and write SLIP-encoded messages.

Deprecated since version 0.6: Direct access to the methods and attributes of the contained stream will be removed in version 1.0.

To instantiate a SlipStream object, the user must provide a pre-constructed open byte stream that is ready for reading and/or writing.

Parameters:

• stream (IOStream) – The byte stream that will be wrapped.

• chunk_size (int) –

  The number of bytes to read per read operation. The default value for chunck_size is io.DEFAULT_BUFFER_SIZE.

  Setting chunk_size is useful when the stream has a low bandwidth and/or bursty data (e.g. a serial port interface). In such cases it is useful to have a chunk_size value of 1, to avoid that the application hangs or becomes unresponsive.

Added in version 0.6: The chunk_size parameter.

A SlipStream instance can e.g. be useful to read slip-encoded messages from a file:

with open('/path/to/a/slip/encoded/file', mode='rb') as f:
    slip_file = SlipStream(f)
    for msg in slip_file:
        # Do something with the message

A SlipStream instance has the following attributes and read-only properties in addition to the attributes offered by its base class SlipWrapper:

chunk_size

The number of bytes to read during each read operation.

property readable

Indicates if the wrapped stream is readable.

The value is True if the readability of the wrapped stream cannot be determined.

Return type:

bool

property writable

Indicates if the wrapped stream is writable.

The value is True if the writabilty of the wrapped stream cannot be determined.

Return type:

bool

Module slipsocket

The slipsocket module contains the class SlipSocket. The SlipSocket class can also be imported directly from the sliplib package.

TCPAddress

TCPAddress stands for either an IPv4 address, a (host: str, port: int) tuple, or an IPv6 address, a (host: str, port: int, flowinfo: int, scope_id: int) tuple.

class SlipSocket(sock)

Bases: SlipWrapper[socket]

Class that wraps a TCP socket with a Driver.

SlipSocket combines a Driver instance with a socket. The SlipSocket class has all the methods from its base class SlipWrapper. In addition, it directly exposes all methods and attributes of the contained socket, except for the following:

• send*() and recv*(). These methods are not supported, because byte-oriented send and receive operations would invalidate the internal state maintained by SlipSocket.

• Similarly, makefile() is not supported, because byte- or line-oriented read and write operations would invalidate the internal state.

• share() (Windows only) and dup(). The internal state of the SlipSocket would have to be duplicated and shared to make these methods meaningful. Because of the lack of a convincing use case for this, sharing and duplication is not supported.

• The accept() method is delegated to the contained socket.socket. The socket that is returned by the socket’s accept() method is automatically wrapped in a SlipSocket object.

Instead of the socket’s send*() and recv*() methods a SlipSocket provides the method send_msg() and recv_msg() to send and receive SLIP-encoded messages.

Deprecated since version 0.6: Direct access to the methods and attributes of the contained socket.socket other than family, type, and proto will be removed in version 1.0

Only TCP sockets are supported. Using the SLIP protocol on UDP sockets is not supported for the following reasons:

• UDP is datagram-based. Using SLIP with UDP therefore introduces ambiguity: should SLIP packets be allowed to span multiple UDP datagrams or not?

• UDP does not guarantee delivery, and does not guarantee that datagrams are delivered in the correct order.

To instantiate a SlipSocket, the user must provide a pre-constructed TCP socket. An alternative way to instantiate s SlipSocket is to use the class method create_connection().

Parameters:

sock (socket.socket) – An existing TCP socket, i.e. a socket with type socket.SOCK_STREAM

Class SlipSocket offers the following methods in addition to the methods offered by its base class SlipWrapper:

accept()

Accept an incoming connection.

Returns a SlipSocket and a remote TCPAddress. The returned SlipSocket inherits the leading-end-byte behavior of the SlipSocket instance on which accept() was called.

Return type:

(SlipSocket, TCPAddress):

Returns:

A tuple with a SlipSocket object and the remote IP address.

classmethod create_connection(address, timeout=None, source_address=None)

Create a SlipSocket connection.

This convenience method creates a connection to a socket at the specified address using the socket.create_connection() function. The socket that is returned from that call is wrapped in a SlipSocket object.

Parameters:

• address (TCPAddress) – The remote address.

• timeout (Optional[float]) – Optional timeout value.

• source_address (Optional [TCPAddress]) – Optional local address for the near socket.

Return type:

SlipSocket

Returns:

A SlipSocket instance that is connected to the socket at the remote address.

See also

socket.create_connection()

Note

The accept() and create_connection() methods do not magically turn the socket at the remote address into a socket that uses the SLIP protocol. For the connection to work properly, the remote socket must already have been configured to use the SLIP protocol.

The following commonly used socket.socket methods are exposed through a SlipSocket object. These methods are simply delegated to the wrapped socket.socket instance. See the documentation for socket.socket for more information on these methods.

bind(address)

Bind the SlipSocket instance to address.

Parameters:

address (TCPAddress) – The address to bind to.

Return type:

None

close()

Close the SlipSocket instance.

Return type:

None

connect(address)

Connect the SlipSocket instance to a remote socket at address.

Parameters:

address (TCPAddress) – The IP address of the remote socket.

Return type:

None

connect_ex(address)

Connect the SlipSocket instance to a remote socket at address.

Parameters:

address (TCPAddress) – The IP address of the remote socket.

Return type:

None

fileno()

Get the socket’s file descriptor.

Return type:

int

Returns:

The wrapped socket’s file descriptor, or -1 on failure.

getpeername()

Get the IP address of the remote socket to which the SlipSocket instance is connected.

Return type:

TCPAddress

Returns:

The remote IP address.

getsockname()

Get the SlipSocket instance’s own address.

Return type:

TCPAddress

Returns:

The local IP address.

getsockopt(level, optname, ...)

Get the socket option from the embedded socket. See the documentation for socket.socket.getsockopt() for more information about the parameters.

Parameters:

• level (int) – The socket option level.

• optname (int) – The socket option name.

• ... – Other valid argument(s) depending on the option.

Return type:

Union [int, bytes]

Returns:

The integer or bytes representing the value of the socket option.

gettimeout()

Get the timeout in seconds.

Return type:

Optional[float]

Returns:

The timeout value, or None if no timeout is set.

listen(backlog=None)

Enable a SlipSocket server to accept connections.

Parameters:

backlog (int) – The maximum number of waiting connections.

Return type:

None

setsockopt(level, optname, ...)

Set the socket option of the embedded socket. See the documentation for socket.socket.setsockopt() for more information about the parameters.

Parameters:

• level (int) – The socket option level.

• optname (int) – The socket option name.

• ... – Other valid argument(s) depending on the option.

Return type:

None

shutdown(how)

Shutdown the connection.

Parameters:

how (int) – Flag to indicate which halves of the connection must be shut down.

Return type:

None

Since the wrapped socket is available as the socket attribute, any other socket.socket method can be invoked through that attribute.

Warning

Avoid using socket.socket methods that affect the bytes that are sent or received through the socket. Doing so will invalidate the internal state of the enclosed Driver instance, resulting in corrupted SLIP messages. In particular, do not use any of the recv*() or send*() methods on the socket attribute.

A SlipSocket instance has the following attributes and read-only properties in addition to the attributes and properties offered by its base class SlipWrapper:

socket

The wrapped socket. This is actually just an alias for the stream attribute in the base class.

property family

The wrapped socket’s address family.

Usually socket.AF_INET (IPv4) or socket.AF_INET6 (IPv6).

Return type:

int

property type

The wrapped socket’s type.

Always socket.SOCK_STREAM.

Return type:

int

property proto

The wrapped socket’s protocol number.

Usually 0.

Return type:

int

Module slipserver

The slipserver module contains the classes SlipRequestHandler and SlipServer, that offer building blocks to create a SLIP server.

The SlipRequestHandler and SlipServer classes can also be imported directly from the sliplib package.

class SlipRequestHandler(request, client_address, server)

Bases: BaseRequestHandler

Base class for request handlers for SLIP-based communication.

This class is derived from socketserver.BaseRequestHandler for the purpose of creating TCP server instances that can handle incoming SLIP-based connections.

To do anything useful, a derived class must define its own handle() method that uses self.request to send and receive SLIP-encoded messages.

Other methods can of course also be overridden if necessary.

To initialize the request handler, a request, client address, and server instance must be provided.

The type of the request parameter depends on the type of server that instantiates the request handler. If the server is a SlipServer, then request is a SlipSocket instance. Otherwise, it is a regular socket, and the request handler wraps it in a SlipSocket instance.

Note

If request is not a SlipSocket instance (as will happen when server is not a SlipServer instance), the behavior of the wrapped SlipSocket instance with respect to sending a leading END byte is determined by the value of USE_LEADING_END_BYTE at the time the request handler is initialized.

Parameters:

• request (Union[socket, SlipSocket]) – The socket that is connected to the remote party.

• client_address (TCPAddress) – The remote TCP address.

• server (TCPServer) – The server instance that instantiated this handler object.

handle()

Service the request. Must be defined by a derived class.

Note that in general it does not make sense to use a SlipRequestHandler object to handle a single transmission, as is e.g. common with HTTP. The purpose of the SLIP protocol is to allow separation of messages in a continuous byte stream. As such, it is expected that the handle() method of a derived class is capable of handling multiple SLIP messages, for example:

def handle(self):
    while True:
        msg = self.request.recv_msg()
        if msg == b"":
            break
        # Do something with the message

Return type:

None

finish()

Perform any cleanup actions.

The default implementation does nothing.

Return type:

None

class SlipServer(server_address, handler_class, bind_and_activate=True)

Bases: TCPServer

Base class for SlipSocket based server classes.

This is a convenience class, that offers a few enhancements over the regular TCPServer from the standard library.

• It uses the supplied server_address to determine if it must use an IPv4 or IPv6 socket. The class TCPServer on the other hand is hardcoded to use only IPv4 addresses, and must be subclassed in order to use IPv6 addresses.

• The socket that is passed to the handler_class instance is a SlipSocket instance with the same leading-end-byte behavior as the server socket. That avoids ambiguity when config.USE_LEADING_END_BYTE may have been modified.

Parameters:

• server_address (TCPAddress) – The address on which the server listens.

• handler_class (type[SlipRequestHandler]) – The class that will be instantiated to handle an incoming request.

• bind_and_activate (bool) – Flag to indicate if the server must be bound and activated at creation time.

Module legacy

This module contains the legacy versions of the encode() and decode() functions, as they were defined prior to version 0.7.0. Note that usage of these functions is discouraged. When using these functions, the responsibility to properly split incoming data into SLIP packets, and buffer incomplete SLIP packets resides completely with the user.

encode(message)

Encode a message into a SLIP packet.

Note that the SLIP packet will start with a leading END byte if the current value of USE_LEADING_END_BYTE is True. A leading END byte will be absent if the current value of USE_LEADING_END_BYTE is False.

Parameters:

message (bytes) – The message to encode.

Return type:

bytes

Returns:

The encoded message.

decode(packet)

Extract the original message from a SLIP packet.

Parameters:

packet (bytes) – the SLIP packet to decode.

Return type:

bytes

Returns:

The decoded message.

Raises:

ProtocolError – if the packet cannot be decoded.