Serial

Prerequisites

If you are communicating with a Serial device on Windows, you already have permission (from the operating system) to access the serial port (or USB-to-Serial adaptor). If you receive an Access is denied error message when connecting to a serial port, that indicates the serial port is already open in another application.

If you are using Linux or macOS and receive a Device or resource busy error message when connecting to a serial port, that indicates the serial port is already open in another application. A Permission denied error can be solved by either running your application as the root user or by adding the current user to the same group that the serial port belongs to.

To determine the group that a serial port (e.g., /dev/ttyUSB0) belongs to, run the following.

ls -l /dev/ttyUSB0

The output will be similar to

crw-rw---- 1 root dialout 4, 64 Feb 20 17:23 /dev/ttyUSB0

where it is shown that this serial port belongs to the dialout group.

Add the current user to the appropriate group (e.g., dialout for the above case)

sudo usermod -a -G dialout $USER

and then log out and in again (or restart the computer if you still encounter permission issues after re-logging in).

Serial (MessageBased)

Serial(equipment: Equipment)

Base class for equipment that is connected through a Serial port (or a USB-to-Serial adaptor).

Parameters:

Name	Type	Description	Default
equipment	Equipment	An Equipment instance.	required

A Connection instance supports the following properties for the serial communication protocol, as well as the properties defined in MessageBased. The DataBits, Parity and StopBits enumeration names and values may also be used. For properties that specify an alias, you may also use the alternative name as the property name. See serial.Serial for more details.

Connection Properties:

Name	Type	Description
baud_rate	int	The baud rate (alias: baudrate). Default: 9600
buffer_size	int	The maximum number of bytes to read at a time. Default: 1024
data_bits	DataBits | str | int	The number of data bits: 5, 6, 7 or 8 (alias: bytesize). Default: 8
dsr_dtr	bool	Whether to enable hardware (DSR/DTR) flow control (alias: dsrdtr). Default: False
inter_byte_timeout	float | None	The maximum duration, in seconds, that is allowed between two consecutive bytes in a read operation. A value of zero (or None) indicates that the inter-byte timeout condition is not used. On Windows, the minimum supported value is 0.001 seconds, on POSIX it is 0.1 seconds. A value less than the minimum will disable the inter-byte timeout. Default: None
parity	Parity | str	Parity checking: NONE, ODD, EVEN, MARK or SPACE. Default: NONE
rts_cts	bool	Whether to enable hardware (RTS/CTS) flow control (alias: rtscts). Default: False
stop_bits	StopBits | str | float	The number of stop bits: 1, 1.5 or 2 (alias: stopbits). Default: 1
xon_xoff	bool	Whether to enable software flow control (alias: xonxoff). Default: False

Source code in src/msl/equipment/interfaces/serial.py

def __init__(self, equipment: Equipment) -> None:
    """Base class for equipment that is connected through a Serial port (or a USB-to-Serial adaptor).

    Args:
        equipment: An [Equipment][] instance.

    A [Connection][msl.equipment.schema.Connection] instance supports the following _properties_
    for the serial communication protocol, as well as the _properties_ defined in
    [MessageBased][msl.equipment.interfaces.message_based.MessageBased]. The
    [DataBits][msl.equipment.enumerations.DataBits], [Parity][msl.equipment.enumerations.Parity]
    and [StopBits][msl.equipment.enumerations.StopBits] enumeration names and values may also
    be used. For properties that specify an _alias_, you may also use the alternative name as
    the property name. See [serial.Serial][] for more details.

    Attributes: Connection Properties:
        baud_rate (int): The baud rate (_alias:_ baudrate). _Default: `9600`_
        buffer_size (int): The maximum number of bytes to read at a time. _Default: `1024`_
        data_bits (DataBits | str | int): The number of data bits: 5, 6, 7 or 8 (_alias:_ bytesize). _Default: `8`_
        dsr_dtr (bool): Whether to enable hardware (DSR/DTR) flow control (_alias:_ dsrdtr). _Default: `False`_
        inter_byte_timeout (float | None): The maximum duration, in seconds, that is allowed between two
            consecutive bytes in a read operation. A value of zero (or `None`) indicates that the inter-byte timeout
            condition is not used. On Windows, the minimum supported value is 0.001 seconds, on POSIX it is 0.1
            seconds. A value less than the minimum will disable the inter-byte timeout. _Default: `None`_
        parity (Parity | str): Parity checking: NONE, ODD, EVEN, MARK or SPACE. _Default: `NONE`_
        rts_cts (bool): Whether to enable hardware (RTS/CTS) flow control (_alias:_ rtscts). _Default: `False`_
        stop_bits (StopBits | str | float): The number of stop bits: 1, 1.5 or 2 (_alias:_ stopbits). _Default: `1`_
        xon_xoff (bool): Whether to enable software flow control (_alias:_ xonxoff). _Default: `False`_
    """
    super().__init__(equipment)

    assert equipment.connection is not None  # noqa: S101
    info = parse_serial_address(equipment.connection.address)
    if info is None:
        msg = f"Invalid Serial address {equipment.connection.address!r}"
        raise ValueError(msg)

    url = find_port(info.url[3:], equipment.connection.address) if info.url.startswith("?::") else info.url
    self._serial: serial.Serial = _init_serial(url, equipment.connection.properties)
    self._set_interface_timeout()

    self._buffer_size: int = equipment.connection.properties.get("buffer_size", 1024)
    self._buffer: bytearray = bytearray()

    try:
        self._serial.open()
    except serial.SerialException as e:
        raise MSLConnectionError(self, str(e)) from None

encoding property writable

encoding: str

The encoding that is used for read and write operations.

equipment property

equipment: Equipment

The Equipment associated with the interface.

max_read_size property writable

max_read_size: int

The maximum number of bytes that can be read.

read_termination property writable

read_termination: bytes | None

The termination character sequence that is used for a read operation.

Reading stops when the equipment stops sending data or the read_termination character sequence is detected. If you set the read_termination to be equal to a variable of type str, it will be encoded as bytes.

rstrip property writable

rstrip: bool

Whether to remove trailing whitespace from read messages.

serial property

serial: Serial

Returns the reference to the pySerial.Serial instance.

timeout property writable

timeout: float | None

The timeout, in seconds, for read and write operations.

A value <0 will set the timeout to be None (blocking mode).

write_termination property writable

write_termination: bytes | None

The termination character sequence that is appended to write messages.

If you set the write_termination to be equal to a variable of type str, it will be encoded as bytes.

disconnect

disconnect() -> None

Close the serial port.

Source code in src/msl/equipment/interfaces/serial.py

def disconnect(self) -> None:  # pyright: ignore[reportImplicitOverride]
    """Close the serial port."""
    if hasattr(self, "_serial") and self._serial.is_open:
        self._serial.close()
        super().disconnect()

query

query(
    message: bytes | str,
    *,
    delay: float = 0.0,
    decode: Literal[True] = True,
    dtype: None = None,
    fmt: MessageFormat = ...,
    size: int | None = ...
) -> str

query(
    message: bytes | str,
    *,
    delay: float = 0.0,
    decode: Literal[False] = False,
    dtype: None = None,
    fmt: MessageFormat = ...,
    size: int | None = ...
) -> bytes

query(
    message: bytes | str,
    *,
    delay: float = 0.0,
    decode: bool = ...,
    dtype: MessageDataType = ...,
    fmt: MessageFormat = ...,
    size: int | None = ...
) -> NumpyArray1D

query(
    message: bytes | str,
    *,
    delay: float = 0.0,
    decode: bool = True,
    dtype: MessageDataType | None = None,
    fmt: MessageFormat = None,
    size: int | None = None
) -> bytes | str | NumpyArray1D

Convenience method for performing a write followed by a read.

Parameters:

Name	Type	Description	Default
message	bytes | str	The message to write to the equipment.	required
delay	float	Time delay, in seconds, to wait between the write and read operations.	0.0
decode	bool	Whether to decode the returned message (i.e., convert the message to a str) or keep the message as bytes. Ignored if dtype is not None.	True
dtype	MessageDataType | None	The data type of the elements in the returned message. Can be any object that numpy dtype supports. For messages that are of scalar type (i.e., a single number) it is more efficient to not specify dtype but to pass the returned message to the int or float class to convert the message to the appropriate numeric type. See MessageDataType for more details.	None
fmt	MessageFormat	The format that the returned message data is in. Ignored if dtype is None. See MessageFormat for more details.	None
size	int | None	The number of bytes to read. Ignored if the value is None.	None

Returns:

Type	Description
bytes | str | NumpyArray1D	The message from the equipment. If dtype is specified, then the message is returned as a numpy ndarray, if decode is True then the message is returned as a str, otherwise the message is returned as bytes.

Source code in src/msl/equipment/interfaces/message_based.py

def query(
    self,
    message: bytes | str,
    *,
    delay: float = 0.0,
    decode: bool = True,
    dtype: MessageDataType | None = None,
    fmt: MessageFormat = None,
    size: int | None = None,
) -> bytes | str | NumpyArray1D:
    """Convenience method for performing a [write][msl.equipment.interfaces.message_based.MessageBased.write]
    followed by a [read][msl.equipment.interfaces.message_based.MessageBased.read].

    Args:
        message: The message to write to the equipment.
        delay: Time delay, in seconds, to wait between the _write_ and _read_ operations.
        decode: Whether to decode the returned message (i.e., convert the message to a [str][])
            or keep the message as [bytes][]. Ignored if `dtype` is not `None`.
        dtype: The data type of the elements in the returned message. Can be any object that numpy
            [dtype][numpy.dtype] supports. For messages that are of scalar type (i.e., a single number)
            it is more efficient to not specify `dtype` but to pass the returned message to the
            [int][] or [float][] class to convert the message to the appropriate numeric type.
            See [MessageDataType][msl.equipment._types.MessageDataType] for more details.
        fmt: The format that the returned message data is in. Ignored if `dtype` is `None`.
            See [MessageFormat][msl.equipment._types.MessageFormat] for more details.
        size: The number of bytes to read. Ignored if the value is `None`.

    Returns:
        The message from the equipment. If `dtype` is specified, then the message is
            returned as a numpy [ndarray][numpy.ndarray], if `decode` is `True` then the message
            is returned as a [str][], otherwise the message is returned as [bytes][].
    """  # noqa: D205
    _ = self.write(message)
    if delay > 0:
        time.sleep(delay)
    if dtype:
        return self.read(dtype=dtype, fmt=fmt, size=size)
    return self.read(decode=decode, size=size)

read

read(
    *,
    decode: Literal[True] = True,
    dtype: None = None,
    fmt: MessageFormat = ...,
    size: int | None = ...
) -> str

read(
    *,
    decode: Literal[False] = False,
    dtype: None = None,
    fmt: MessageFormat = ...,
    size: int | None = ...
) -> bytes

read(
    *,
    decode: bool = ...,
    dtype: MessageDataType = ...,
    fmt: MessageFormat = ...,
    size: int | None = ...
) -> NumpyArray1D

read(
    *,
    decode: bool = True,
    dtype: MessageDataType | None = None,
    fmt: MessageFormat = None,
    size: int | None = None
) -> bytes | str | NumpyArray1D

Read a message from the equipment.

This method will block until one of the following conditions is fulfilled:

1. size bytes have been received — only if size is not None.
2. the read_termination byte(s) is(are) received — only if read_termination is not None.
3. a timeout occurs — only if timeout is not None. If a timeout occurs, an MSLTimeoutError is raised.
4. max_read_size bytes have been received. If the maximum number of bytes have been read, an MSLConnectionError is raised.

Tip

You may also want to set the rstrip value for the class instance.

Parameters:

Name	Type	Description	Default
decode	bool	Whether to decode the message (i.e., convert the message to a str) or keep the message as bytes. Ignored if dtype is not None.	True
dtype	MessageDataType | None	The data type of the elements in the returned message. Can be any object that numpy dtype supports. For messages that are of scalar type (i.e., a single number) it is more efficient to not specify dtype but to pass the returned message to the int or float class to convert the message to the appropriate numeric type. See MessageDataType for more details.	None
fmt	MessageFormat	The format that the returned message data is in. Ignored if dtype is None. See MessageFormat for more details.	None
size	int | None	The number of bytes to read. Ignored if the value is None.	None

Returns:

Type	Description
bytes | str | NumpyArray1D	The message from the equipment. If dtype is specified, then the message is returned as a numpy ndarray, if decode is True then the message is returned as a str, otherwise the message is returned as bytes.

Source code in src/msl/equipment/interfaces/message_based.py

def read(
    self,
    *,
    decode: bool = True,
    dtype: MessageDataType | None = None,
    fmt: MessageFormat = None,
    size: int | None = None,
) -> bytes | str | NumpyArray1D:
    """Read a message from the equipment.

    This method will block until one of the following conditions is fulfilled:

    1. `size` bytes have been received — only if `size` is not `None`.
    2. the [read_termination][msl.equipment.interfaces.message_based.MessageBased.read_termination]
       byte(s) is(are) received — only if
       [read_termination][msl.equipment.interfaces.message_based.MessageBased.read_termination]
       is not `None`.
    3. a timeout occurs — only if [timeout][msl.equipment.interfaces.message_based.MessageBased.timeout]
       is not `None`. If a timeout occurs, an
       [MSLTimeoutError][msl.equipment.interfaces.message_based.MSLTimeoutError] is raised.
    4. [max_read_size][msl.equipment.interfaces.message_based.MessageBased.max_read_size]
       bytes have been received. If the maximum number of bytes have been read, an
       [MSLConnectionError][msl.equipment.interfaces.message_based.MSLConnectionError] is raised.

    !!! tip
        You may also want to set the [rstrip][msl.equipment.interfaces.message_based.MessageBased.rstrip]
        value for the class instance.

    Args:
        decode: Whether to decode the message (i.e., convert the message to a [str][])
            or keep the message as [bytes][]. Ignored if `dtype` is not `None`.
        dtype: The data type of the elements in the returned message. Can be any object that numpy
            [dtype][numpy.dtype] supports. For messages that are of scalar type (i.e., a single number)
            it is more efficient to not specify `dtype` but to pass the returned message to the
            [int][] or [float][] class to convert the message to the appropriate numeric type.
            See [MessageDataType][msl.equipment._types.MessageDataType] for more details.
        fmt: The format that the returned message data is in. Ignored if `dtype` is `None`.
            See [MessageFormat][msl.equipment._types.MessageFormat] for more details.
        size: The number of bytes to read. Ignored if the value is `None`.

    Returns:
        The message from the equipment. If `dtype` is specified, then the message is returned
            as a numpy [ndarray][numpy.ndarray], if `decode` is `True` then the message
            is returned as a [str][], otherwise the message is returned as [bytes][].
    """
    if size is not None and size > self._max_read_size:
        msg = f"max_read_size is {self._max_read_size} bytes, requesting {size} bytes"
        raise MSLConnectionError(self, msg)

    try:
        message = self._read(size)
    except (serial.SerialTimeoutException, socket.timeout, TimeoutError, USBTimeoutError):
        raise MSLTimeoutError(self) from None
    except Exception as e:  # noqa: BLE001
        msg = f"{e.__class__.__name__}: {e}"
        raise MSLConnectionError(self, msg) from None

    if size is None:
        if dtype:
            logger.debug("%s.read(dtype=%r, fmt=%r) -> %r", self, dtype, fmt, message)
        else:
            logger.debug("%s.read() -> %r", self, message)
    else:
        if len(message) != size:
            msg = f"received {len(message)} bytes, requested {size} bytes"
            raise MSLConnectionError(self, msg)
        logger.debug("%s.read(size=%s) -> %r", self, size, message)

    if self._rstrip:
        message = message.rstrip()

    if dtype:
        return from_bytes(message, fmt=fmt, dtype=dtype)

    if decode:
        return message.decode(encoding=self._encoding)

    return message

write

write(
    message: bytes | str,
    *,
    data: Sequence1D | None = None,
    dtype: MessageDataType = "<f",
    fmt: MessageFormat = "ieee"
) -> int

Write a message to the equipment.

Parameters:

Name	Type	Description	Default
message	bytes | str	The message to write to the equipment.	required
data	Sequence1D | None	The data to append to message.	None
dtype	MessageDataType	The data type to use to convert each element in data to bytes. Ignored if data is None. See MessageDataType for more details.	'<f'
fmt	MessageFormat	The format to use to convert data to bytes. Ignored if data is None. See MessageFormat for more details.	'ieee'

Returns:

Type	Description
int	The number of bytes written.

Source code in src/msl/equipment/interfaces/message_based.py

def write(
    self,
    message: bytes | str,
    *,
    data: Sequence1D | None = None,
    dtype: MessageDataType = "<f",
    fmt: MessageFormat = "ieee",
) -> int:
    """Write a message to the equipment.

    Args:
        message: The message to write to the equipment.
        data: The data to append to `message`.
        dtype: The data type to use to convert each element in `data` to bytes. Ignored
            if `data` is `None`. See [MessageDataType][msl.equipment._types.MessageDataType]
            for more details.
        fmt: The format to use to convert `data` to bytes. Ignored if `data` is `None`.
            See [MessageFormat][msl.equipment._types.MessageFormat] for more details.

    Returns:
        The number of bytes written.
    """
    if not isinstance(message, bytes):
        message = message.encode(encoding=self._encoding)

    if data is not None:
        message += to_bytes(data, fmt=fmt, dtype=dtype)

    if self._write_termination and not message.endswith(self._write_termination):
        message += self._write_termination

    logger.debug("%s.write(%r)", self, message)

    try:
        return self._write(message)
    except (serial.SerialTimeoutException, socket.timeout, TimeoutError, USBTimeoutError):
        raise MSLTimeoutError(self) from None
    except Exception as e:  # noqa: BLE001
        raise MSLConnectionError(self, str(e)) from None