| Name | siobrultech-protocols JSON |
| Version |
0.14.0
JSON |
| download |
| home_page | |
| Summary | A Sans-I/O Python client library for Brultech Devices |
| upload_time | 2023-12-08 04:44:28 |
| maintainer | |
| docs_url | None |
| author | Shawn Wilsher |
| requires_python | <4,>3.8 |
| license | |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# What is siobrultech-protocols?
This library is a collection of protcols that decode various packet formats from
[Brultech Research](https://www.brultech.com/).
# 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](https://sans-io.readthedocs.io/) for more information.
## Installation
```
pip install siobrultech-protocols
```
## Usage
### Receiving data packets
```python
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:
```python
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`:
```python
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:
```python
# 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:
```python
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](https://github.com/sdwilsh/siobrultech-protocols/network/dependents).
### 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](https://earthly.dev/) with `./earthly.sh +lint`
Raw data
{
"_id": null,
"home_page": "",
"name": "siobrultech-protocols",
"maintainer": "",
"docs_url": null,
"requires_python": "<4,>3.8",
"maintainer_email": "",
"keywords": "",
"author": "Shawn Wilsher",
"author_email": "me@shawnwilsher.com",
"download_url": "https://files.pythonhosted.org/packages/0e/f6/49bfa852212a79f8c616f075c82abbf02a205d99cb984c28336f53ffc989/siobrultech-protocols-0.14.0.tar.gz",
"platform": null,
"description": "\n\n\n# What is siobrultech-protocols?\n\nThis library is a collection of protcols that decode various packet formats from\n[Brultech Research](https://www.brultech.com/).\n\n# What is Sans-I/O?\n\nSans-I/O is a philosophy for developing protocol processing libraries in which\nthe library does not do any I/O. Instead, a user of the library is responsible\nfor transferring blocks of bytes between the socket or pipe and the protocol\nlibrary, and for receiving application-level protocol items from and sending\nthem to the library. This obviously makes a sans-I/O library a little more\ndifficult to use, but comes with the advantage that the same library can be\nused with any I/O and concurrency mechanism: the same library should be usable\nin a single-request-at-a-time server, a process-per-request or\nthread-per-request blocking server, a server using select/poll and\ncontinuations, or a server using asyncio, Twisted, or any other asynchronous\nframework.\n\nSee [SansIO](https://sans-io.readthedocs.io/) for more information.\n\n## Installation\n\n```\npip install siobrultech-protocols\n```\n\n## Usage\n\n### Receiving data packets\n\n```python\nimport functools\nfrom siobrultech_protocols.gem.protocol import PacketProtocol, PacketReceivedMessage\n\n# Queue to get received packets from.\nqueue = asyncio.Queue()\n\n# Pass this Protocol to whatever receives data from the device.\nprotocol_factory = functools.partial(PacketProtocol, queue=queue)\n\n# Dequeue and look for packet received messages. (Typically do this in a loop.)\nmessage = await queue.get()\nif isinstance(message, PacketReceivedMessage):\n packet = message.packet\nqueue.task_done()\n```\n\n### Receiving data packets AND sending API commands\n\nIf 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:\n\n```python\nfrom siobrultech_protocols.gem.api import get_serial_number\n\nserial = await get_serial_number(protocol)\n```\n\n`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.\n\n#### Methods to Get Information from a Device\n\n| Method | GEM | ECM | Description |\n| ------------------- | --- | --- | ---------------------------------------- |\n| `get_serial_number` | \u2705\ufe0e | \u2705\ufe0e | Obtains the serial number of the device. |\n\n#### Methods to Setup a Device\n\n| Method | GEM | ECM | Description |\n| ----------------------------- | --- | --- | --------------------------------------------------------------------------- |\n| `set_date_and_time` | \u2705\ufe0e | \u274c | Sets the GEM's clock to the specified `datetime`. |\n| `set_packet_format` | \u2705\ufe0e | \u274c | Sets the GEM's packet format to the specified `PacketFormatType`. |\n| `set_packet_send_interval` | \u2705\ufe0e | \u2705\ufe0e | Sets the frequency (seconds) that the monitor should send packets. |\n| `set_secondary_packet_format` | \u2705\ufe0e | \u274c | Sets the GEM's secondary packet format to the specified `PacketFormatType`. |\n| `synchronize_time` | \u2705\ufe0e | \u274c | Synchronizes the GEM's clock to the time of the local device. |\n\n### Calling API endpoints that aren't supported by this library\n\n`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`:\n\n```python\nfrom siobrultech_protocols.gem import api\n\n# Define a Python data type for the response. It can be whatever you want; a simple Dict, a custom dataclass, etc.\nAllSettings = Dict[str, Any]\n\ndef _parse_all_gem_settings(response: str) -> AllSettings:\n # Here you would parse the GEM response into the python type you defined above\n\ndef _parse_all_ecm_settings(response: bytes) -> AllSettings:\n # Here you would parse the ECM response into the python type you defined above\n\nGET_ALL_SETTINGS = api.ApiCall[None, AllSettings](\n gem_formatter=lambda _: \"^^^RQSALL\", gem_parser=_parse_all_gem_settings,\n ecm_formatter=lambda _: [b\"\\xfc\", b\"SET\", b\"RCV\"], ecm_parser=_parse_all_ecm_settings,\n)\n```\n\nGiven an `ApiCall` (whether one of those in `siobrultech_protocols.api` or defined yourself as above), you can\nmake the request by working with the protocol directly as follows:\n\n```python\n# Start the API request; do this once for each API call. Each protocol instance can only support one\n# API call at a time.\ndelay = protocol.begin_api_request()\nsleep(delay) # Wait for the specified delay, using whatever mechanism is appropriate for your environment\n\n# Send the API request. We use asyncio's implementation of Future, but you can use whatever you like.\nresult = asyncio.get_event_loop().create_future()\nprotocol.invoke_api(GET_ALL_SETTINGS, None, result)\nsettings = await asyncio.wait_for(result, timeout=5)\n\n# End the API request\nprotocol.end_api_request()\n```\n\nAlternatively, we also provide a context wrapper that works with `asyncio` as well:\n\n```python\nfrom siobrultech_protocols.gem import api\n\nasync with api.call_api(GET_ALL_SETTINGS, protocol) as f:\n settings = await f(None)\n```\n\nTake a look at some usage examples from [libraries that use this](https://github.com/sdwilsh/siobrultech-protocols/network/dependents).\n\n### Calling API endpoints when multiple devices share a connection\n\nAll 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.\n\n## Development\n\n### Setup\n\n```\npython3.11 -m venv .venv\nsource .venv/bin/activate\n\n# Install Requirements\npip install -r requirements.txt\n\n# Install Dev Requirements\npip install -r requirements-dev.txt\n```\n\n### Testing\n\nTests are run with `pytest`.\n\n### Linting\n\nLint can be run with [Earthly](https://earthly.dev/) with `./earthly.sh +lint`\n",
"bugtrack_url": null,
"license": "",
"summary": "A Sans-I/O Python client library for Brultech Devices",
"version": "0.14.0",
"project_urls": {
"Bug Reports": "https://github.com/sdwilsh/siobrultech-protocols/issues",
"Release Notes": "https://github.com/sdwilsh/siobrultech-protocols/releases/",
"Source": "https://github.com/sdwilsh/siobrultech-protocols"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b9504c04f7a900fd5828f0a4f34bb9c291c60e5695c3b0c304247173f975e56a",
"md5": "f7a5d1bd1729ec039434d5cdb7d1f4cc",
"sha256": "f66e33a3b8a469c5f3ec123b7a48bcc62df62950120d46c38e388d27cb12a794"
},
"downloads": -1,
"filename": "siobrultech_protocols-0.14.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "f7a5d1bd1729ec039434d5cdb7d1f4cc",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4,>3.8",
"size": 19190,
"upload_time": "2023-12-08T04:44:27",
"upload_time_iso_8601": "2023-12-08T04:44:27.663206Z",
"url": "https://files.pythonhosted.org/packages/b9/50/4c04f7a900fd5828f0a4f34bb9c291c60e5695c3b0c304247173f975e56a/siobrultech_protocols-0.14.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "0ef649bfa852212a79f8c616f075c82abbf02a205d99cb984c28336f53ffc989",
"md5": "946c26bdb6c4b7090edae0fb253faff0",
"sha256": "6364422e6a7dbb3230f768374fa6f858ad03f39c806cbea766eb6fc135077819"
},
"downloads": -1,
"filename": "siobrultech-protocols-0.14.0.tar.gz",
"has_sig": false,
"md5_digest": "946c26bdb6c4b7090edae0fb253faff0",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4,>3.8",
"size": 19637,
"upload_time": "2023-12-08T04:44:28",
"upload_time_iso_8601": "2023-12-08T04:44:28.873895Z",
"url": "https://files.pythonhosted.org/packages/0e/f6/49bfa852212a79f8c616f075c82abbf02a205d99cb984c28336f53ffc989/siobrultech-protocols-0.14.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-12-08 04:44:28",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "sdwilsh",
"github_project": "siobrultech-protocols",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "siobrultech-protocols"
}
