Skip to main content

A Sans-I/O Python client library for Brultech Devices

Project description

Lint Build

What is siobrultech-protocols?

This library is a collection of protcols that decode various packet formats from Brultech Research.

What is Sans-I/O?

Sans-I/O is a philosophy for developing protocol processing libraries in which the library does not do any I/O. Instead, a user of the library is responsible for transferring blocks of bytes between the socket or pipe and the protocol library, and for receiving application-level protocol items from and sending them to the library. This obviously makes a sans-I/O library a little more difficult to use, but comes with the advantage that the same library can be used with any I/O and concurrency mechanism: the same library should be usable in a single-request-at-a-time server, a process-per-request or thread-per-request blocking server, a server using select/poll and continuations, or a server using asyncio, Twisted, or any other asynchronous framework.

See SansIO for more information.

Installation

pip install siobrultech-protocols

Usage

Receiving data packets

import functools
from siobrultech_protocols.gem.protocol import PacketProtocol, PacketReceivedMessage

# Queue to get received packets from.
queue = asyncio.Queue()

# Pass this Protocol to whatever receives data from the device.
protocol_factory = functools.partial(PacketProtocol, queue=queue)

# Dequeue and look for packet received messages. (Typically do this in a loop.)
message = await queue.get()
if isinstance(message, PacketReceivedMessage):
    packet = message.packet
queue.task_done()

Receiving data packets AND sending API commands

If you want to send API commands as well, use a BidirectionalProtocol instead of a PacketProtocol. Then given the protocol instance for a given connection, do the API call as follows:

from siobrultech_protocols.gem.api import get_serial_number

serial = await get_serial_number(protocol)

siobrultech_protocols provides direct support for a small set of API calls. Some of these calls work for both GEM and ECM monitors, others are GEM-only.

Methods to Get Information from a Device

Method GEM ECM Description
get_serial_number ✅︎ ✅︎ Obtains the serial number of the device.

Methods to Setup a Device

Method GEM ECM Description
set_date_and_time ✅︎ Sets the GEM's clock to the specified datetime.
set_packet_format ✅︎ Sets the GEM's packet format to the specified PacketFormatType.
set_packet_send_interval ✅︎ ✅︎ Sets the frequency (seconds) that the monitor should send packets.
set_secondary_packet_format ✅︎ Sets the GEM's secondary packet format to the specified PacketFormatType.
synchronize_time ✅︎ Synchronizes the GEM's clock to the time of the local device.

Calling API endpoints that aren't supported by this library

siobrultech_protocols has built-in support for just a tiny subset of the full API exposed by GEM and ECM. If you want to call an API endpoint for which this library doesn't provide a helper, you can make your own. For example, the following outline could be filled in to support the "get all settings" endpoint; you could define GET_ALL_SETTINGS:

from siobrultech_protocols.gem import api

# Define a Python data type for the response. It can be whatever you want; a simple Dict, a custom dataclass, etc.
AllSettings = Dict[str, Any]

def _parse_all_gem_settings(response: str) -> AllSettings:
    # Here you would parse the GEM response into the python type you defined above

def _parse_all_ecm_settings(response: bytes) -> AllSettings:
    # Here you would parse the ECM response into the python type you defined above

GET_ALL_SETTINGS = api.ApiCall[None, AllSettings](
    gem_formatter=lambda _: "^^^RQSALL", gem_parser=_parse_all_gem_settings,
    ecm_formatter=lambda _: [b"\xfc", b"SET", b"RCV"], ecm_parser=_parse_all_ecm_settings,
)

Given an ApiCall (whether one of those in siobrultech_protocols.api or defined yourself as above), you can make the request by working with the protocol directly as follows:

# Start the API request; do this once for each API call. Each protocol instance can only support one
# API call at a time.
delay = protocol.begin_api_request()
sleep(delay)  # Wait for the specified delay, using whatever mechanism is appropriate for your environment

# Send the API request. We use asyncio's implementation of Future, but you can use whatever you like.
result = asyncio.get_event_loop().create_future()
protocol.invoke_api(GET_ALL_SETTINGS, None, result)
settings = await asyncio.wait_for(result, timeout=5)

# End the API request
protocol.end_api_request()

Alternatively, we also provide a context wrapper that works with asyncio as well:

from siobrultech_protocols.gem import api

async with api.call_api(GET_ALL_SETTINGS, protocol) as f:
    settings = await f(None)

Take a look at some usage examples from libraries that use this.

Calling API endpoints when multiple devices share a connection

All of the API helper methods take an optional serial_number parameter to target a specific device if there are multiple devices on the same connection. This has no effect for ECM devices.

Development

Setup

python3.11 -m venv .venv
source .venv/bin/activate

# Install Requirements
pip install -r requirements.txt

# Install Dev Requirements
pip install -r requirements-dev.txt

Testing

Tests are run with pytest.

Linting

Lint can be run with Earthly with ./earthly.sh +lint

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

siobrultech-protocols-0.14.0.tar.gz (19.6 kB view details)

Uploaded Source

Built Distribution

siobrultech_protocols-0.14.0-py3-none-any.whl (19.2 kB view details)

Uploaded Python 3

File details

Details for the file siobrultech-protocols-0.14.0.tar.gz.

File metadata

File hashes

Hashes for siobrultech-protocols-0.14.0.tar.gz
Algorithm Hash digest
SHA256 6364422e6a7dbb3230f768374fa6f858ad03f39c806cbea766eb6fc135077819
MD5 946c26bdb6c4b7090edae0fb253faff0
BLAKE2b-256 0ef649bfa852212a79f8c616f075c82abbf02a205d99cb984c28336f53ffc989

See more details on using hashes here.

File details

Details for the file siobrultech_protocols-0.14.0-py3-none-any.whl.

File metadata

File hashes

Hashes for siobrultech_protocols-0.14.0-py3-none-any.whl
Algorithm Hash digest
SHA256 f66e33a3b8a469c5f3ec123b7a48bcc62df62950120d46c38e388d27cb12a794
MD5 f7a5d1bd1729ec039434d5cdb7d1f4cc
BLAKE2b-256 b9504c04f7a900fd5828f0a4f34bb9c291c60e5695c3b0c304247173f975e56a

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page