embutils.serial.device module

Serial device implementation classes.

date: 2021
author: Christian Wiche
contact: cwichel@gmail.com
license: The MIT License (MIT)

class embutils.serial.device.Device(port: Optional[str] = None, looped: bool = False, settings: Optional[dict] = None)[source]

Bases: object

Serial device implementation wrapper. This class includes the USB ID information and allows easily use of looped serial ports for testing.

Device configuration. Applies the serial device settings to the selected port.

Parameters

port (str) – Port name.
looped (bool) – Enables the test mode (looped serial).
settings (dict) – Serial device configuration.

Raises

ValueError – Port is not provided, or it doesnt exist.

DEF_ID = 3735928559: Default device ID

DEF_SETTINGS = {'baudrate': 115200, 'bytesize': 8, 'parity': 'N', 'stopbits': 1, 'timeout': 0.1}: Default device settings

static _id_from_port(port: str) → Optional[int][source]

Retrieves the USB ID for the given port.

Parameters: port (str) – Port name.
Returns: USB ID
Return type: int

close() → None[source]: Closes the serial port.

flush() → None[source]: Flushes the serial buffer.

property id: int: Device USB ID.

property is_open: bool: Returns if the serial device is open.

open() → bool[source]

Tries to open the serial port.

Returns: True if open, false otherwise.
Return type: bool

property port: str: Device port name.

read(size: int = 1) → Optional[bytearray][source]

Reads a fixed number of bytes from the serial buffer. The process is stopped with error if a timeout is reached before completion.

Parameters: size (int) – Number of bytes to read.
Returns: None if empty or disconnected. Bytearray if bytes received.
Return type: Optional[bytearray]

read_until(expected: bytes = b'\n', size: int = None) → Optional[bytearray][source]

Reads bytes from the serial buffer until the expected sequence is found, the received bytes exceed the specified limit, or a timeout is reached.

Parameters

expected (bytes) – Stop read condition.
size (int) – Read array size limit.

Returns

None if empty or disconnected. Bytearray if bytes received.

Return type

Optional[bytearray]

property serial: serial.serialposix.Serial: Serial handler.

write(data: bytearray) → int[source]

Writes the given data through the serial port.

Parameters: data (bytearray) – Bytes to be sent through the serial port.
Returns: Number of bytes sent.
Return type: int

class embutils.serial.device.DeviceList(iterable=(), /)[source]

Bases: List[embutils.serial.device.Device]

Serial device list implementation. This class define mechanisms to scan, compare and filter lists of devices.

compare(other: embutils.serial.device.DeviceList) → embutils.serial.device.DeviceList[source]

Get the differences between two serial device lists.

Parameters: other (DeviceList) – List to compare with.
Returns: List containing the differences.
Return type: DeviceList

filter(port: Optional[str] = None, dev_id: Optional[int] = None) → embutils.serial.device.DeviceList[source]

Filter serial devices from a given list.

Parameters

port (str) – Port name to be filtered.
dev_id (int) – Device ID to be filtered.

Returns

List containing the filtered devices.

Return type

DeviceList

static scan() → embutils.serial.device.DeviceList[source]

Scan the system and return a list with the connected serial devices.

Returns: List with devices.
Return type: DeviceList

class embutils.serial.device.DeviceScanner(period: float = 0.5, **kwargs)[source]

Bases: embutils.utils.service.AbstractService

Serial device scanner implementation. This class define a thread that allows to check periodically for changes on the connected serial devices.

Available events:

on_scan_period: This event is emitted after the scan period is completed. Subscribe using callbacks with syntax:
```
def <callback>()
```
on_list_change: This event is emitted when a change is detected on the connected device list. Subscribe using callbacks with syntax:
```
def <callback>(event: SerialDeviceScanner.Event, changes: SerialDeviceList)
```

Class initialization.

Parameters: period (float) – Define the periodicity of the scanner executions in seconds.

class Event(value)[source]

Bases: embutils.utils.enum.IntEnum

Serial device scanner event definitions.

SD_LIST_CHANGED = 1

SD_NO_EVENT = 0

SD_PLUGGED_MULTI = 3

SD_PLUGGED_SINGLE = 2

SD_REMOVED_MULTI = 5

SD_REMOVED_SINGLE = 4

static get_event(old: embutils.serial.device.DeviceList, new: embutils.serial.device.DeviceList) → Tuple[embutils.serial.device.DeviceScanner.Event, embutils.serial.device.DeviceList][source]

Compares two serial device lists and return the differences.

Parameters

old (DeviceList) – Last scanned device list.
new (DeviceList) – New scanned device list.

Returns

Serial device scanner event and device difference list.

Return type

Tuple[DeviceScanner.Event, DeviceList]

TASK_PERIOD_S = 0.5: Task execution period.

_abc_impl = <_abc_data object>

_on_start() → None[source]: Run the first scan and start timer.

_scan() → None[source]

Scanner functionality:

Read the connected devices.
Check for changes on the connected devices.
Raise change event if required, updates current device list.

_task() → None[source]

Serial Scanner process:

Read the connected devices.
Check for changes on the connected devices.
Raise change event if required, updates current device list.

property devices: embutils.serial.device.DeviceList: Connected serial devices list.

property period: float: Scanner period in seconds.