| Name | py-aiowialon JSON |
| Version |
1.3.3
JSON |
| download |
| home_page | None |
| Summary | Async Wialon Remote API wrapper for Python 3 |
| upload_time | 2024-08-17 19:43:53 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.8 |
| license | MIT License Copyright (c) 2013-2016 Gurtam Alex Chernetsky Copyright (c) 2022 Dmytro Yaroshenko (https://github.com/o-murphy) Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
asyncio
wialon
wialon remote api
gurtam
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
[![SWUbanner]][SWUdocs]
[SWUbanner]:
https://raw.githubusercontent.com/vshymanskyy/StandWithUkraine/main/banner-direct-single.svg
[SWUdocs]:
https://github.com/vshymanskyy/StandWithUkraine/blob/main/docs/README.md
# AIO Wialon
[](https://opensource.org/licenses/MIT)
[](https://pypi.org/project/py-aiowialon/)
`AIO Wialon` is an async implementation of Python wrapper for Wialon Remote API,
### Table of
* [Installation](#installation)
* [Start Polling](#start-polling)
* [Wialon API Call](#wialon-api-call)
* [API Call Example](#api-call-example)
* [Batch requests](#batch-requests)
* [Multipart requests](#multipart-requests)
* [Shortcuts](#shortcuts)
* [Wialon Events](#wialon-events)
* [Register AVL Events](#register-avl-events)
* [On login/logout](#on-loginlogout)
* [AVL Events Handling](#avl-events-handling)
* [Register AVL Events handler](#register-avl-events-handlers)
* [Remove AVL Events handler](#remove-avl-events-handlers)
* [Disposable handlers](#disposable-handlers)
* [Exceptions Handling](#exceptions-handling)
* [Get exception results, batch exceptions](#exceptions-handling-batch)
* [Quick API Help](#quick-api-help)
* [Advanced](#advanced-usage)
* [Limitations](#limitations)
* [Prevent polling auto logout](#prevent-polling-logout)
* [Critical requests execution (Render, Reports, Messages)](#critical-requests-execution)
* [Async session lock](#async-session-lock)
* [Timeout for API call](#timeout-for-api-call)
* [Extending AIO Wialon](#extending-aio-wialon)
* [Debugging](#debugging)
* [Wialon Remote Api documentation](http://sdk.wialon.com/wiki/en/sidebar/remoteapi/apiref/apiref)
## Installation
```bash
pip install py-aiowialon
```
## Start Polling
Open session and start poll AVL events immediately.
[Look the Wialon Events section](#wialon-events) to see how we can handle AVL Events on polling
```python
import asyncio
from aiowialon import Wialon
TOKEN = '<Your Wialon API access token>'
HOST = '<API host IP or url>'
wialon = Wialon(host=HOST, token=TOKEN)
if __name__ == "__main__":
asyncio.run(wialon.start_polling())
```
> [!TIP]
> `Wialon.start_polling()` is not require a manual Wialon.login() call
## Wialon API Call
API Call is function that returns `Wialon.call()` instance
Almost all Wialon Remote API `services/actions` available
through dot syntax: `wialon.<service>_<action_name>(**params)`
To make API call use method of Wialon instance with same name as API endpoint
replace `/` with underscore.
#### API Call Example
```python
import asyncio
from aiowialon import Wialon, flags
TOKEN = '<Your Wialon API access token>'
wialon = Wialon(token=TOKEN)
async def main():
await wialon.login()
# The example of core/search_item API call:
result = await wialon.core_search_item(id=12345, flags=flags.UnitsDataFlag.ALL)
print(result)
await wialon.logout()
asyncio.run(main())
```
> [!WARNING]
> Some Wialon Remote API methods requires a lock of asynchronous context
> (execution of reports, loading messages, etc).
> If you need these methods,
> it's highly recommended to get acquainted with [**Critical requests execution**](#critical-requests-execution) section
### Batch requests
Use `Wialon.batch` instead of `asyncio.gather` to make multiple API calls in a same time.
It allows to make just one request to server with few calls.
This avoids reaching the server's request limits.
And transfers the overhead of processing asynchronous context to the server side.
Few Wialon.call() coroutines would be batched to single 'core/batch' request.
```python
# put few calls to a batch method
from aiowialon import Wialon, flags
wialon = Wialon(token=TOKEN)
async def some_func(params1, params2):
api_calls = [
wialon.core_search_item(**params1),
wialon.unit_get_fuel_settings(**params2),
...
]
return await wialon.batch(*api_calls, flags_=flags.BatchFlag.EXECUTE_ALL)
```
> [!TIP]
> * You can combine different API services and actions in single batch call
> * [How to handle batch exceptions](#exceptions-handling-batch)
> [!WARNING]
> * Some requests don't support batch!
> * Don't try to put batch into other batch, it can raise unexpected behaviour
> * Go to the [Wialon Remote Api documentation](http://sdk.wialon.com/wiki/en/sidebar/remoteapi/apiref/apiref) to get details
### Multipart requests
Use `Wialon.multipart` method and `MultipartField` with API call to but multipart data to request,
Put call coroutine and required MultipartField instances to the `Wialon.multipart()`
```python
from aiowialon import Wialon, MultipartField
wialon = Wialon(token=TOKEN)
async def upload_driver_image():
event_hash = 'aiowialon_drv_upd' # custom event hash
params = {"itemId": 717351, "driverId": 38, "eventHash": event_hash}
file_path = "driver_img.jpg"
with open(file_path, 'rb') as f:
file_data = f.read()
await wialon.multipart(
wialon.resource_upload_driver_image(**params),
*[
MultipartField(
name='drivers_dlg_props_upload_image',
value=file_data,
filename="image.jpg",
content_type='image/jpeg'
)
]
)
```
> [!WARNING]
> * Don't try to put multipart requests to batch!
> * Some requests don't support multipart
> * Don't try to put multipart request into batch, it can raise unexpected behaviour
> * Go to the [Wialon Remote Api documentation](http://sdk.wialon.com/wiki/en/sidebar/remoteapi/apiref/apiref) to get details
### Shortcuts
Shortcuts are available as efficient solutions for some common actions, like .wlp export
```python
from aiowialon import Wialon
from aiowialon.shortcuts import WLP
wialon = Wialon(token=TOKEN)
async def dump_unit(item_id):
await wialon.login()
wlp = await WLP.export_item(wialon, item_id)
with open(f"{id}.wlp", 'wb') as fp:
fp.write(wlp)
```
## Wialon Events
The library propose using the polling to handle AVL Events.
AVL events is the events that happens on the server and returns to us if we registered it in current session
This section references to [Wialon AVL Events Docs](https://sdk.wialon.com/wiki/en/sidebar/remoteapi/apiref/requests/avl_evts)
### Register AVL Events
Firstly we have to register items for AVL events handling in current session.
_[(api reference here)](https://sdk.wialon.com/wiki/en/sidebar/remoteapi/codesamples/update_datafalags)_
Bellow is example how to add all AVL Units (vehicles) to handle AVL events of this units in current session,
We use there just simple Wialon API Call
```python
from aiowialon import Wialon, flags
wialon = Wialon(token=TOKEN)
async def register_avl_events():
spec = [
{
"type_": "type",
"data": "avl_unit",
"flags": flags.UnitsDataFlag.BASE | flags.UnitsDataFlag.POS,
"mode": 0
}
]
return await wialon.core_update_data_flags(spec=spec)
```
### On login/logout
We can automate this logic for each session opening by registering `on_session_open` callback,
Use `@wialon.on_session_open` decorator for this
So wialon will login and register avl items to polling before polling start
```python
@wialon.on_session_open
async def register_avl_events(session_login):
print("Session eid:", session_login['eid'])
spec = [
{
"type_": "type",
"data": "avl_unit",
"flags": flags.UnitsDataFlag.BASE | flags.UnitsDataFlag.POS,
"mode": 0
}
]
return await wialon.core_update_data_flags(spec=spec)
if __name__ == "__main__":
asyncio.run(wialon.start_polling())
```
Also we can add callback on session logout. Use `@wialon.on_session_close` decorator for this
```python
@wialon.on_session_close
async def on_session_close(session_logout):
print("Logout event:", session_logout)
```
> [!NOTE]
> * You can register just one `on_session_open` callback for Wialon instance
> * You can register just one `on_session_close` callback for Wialon instance
### AVL Events Handling
After polling start and AVL Items registered for polling we can handle the AVL Events.
Use `@wialon.avl_event_handler()` decorator
#### Register AVL Events handlers
```python
from aiowialon import AvlEvent
@wialon.avl_event_handler()
async def unit_event(event: AvlEvent):
print("Handler got event:", event)
```
Put the filter function to the decorator to apply filtering of AVL events
```python
from aiowialon import AvlEvent
@wialon.avl_event_handler(lambda event: event.data.i == 734455)
async def unit_734455_event(event: AvlEvent):
print("Handler got event from item 734455:", event)
```
> [!NOTE]
> Register handlers in an order in which filters have to be applied. If some handler catched the event, next handler in order will never do.
#### Remove AVL Events handlers
```python
# use them as you need
wialon.remove_avl_event_handler('handler_name')
wialon.remove_avl_event_handler(handler_func)
```
#### Disposable handlers
Use `@wialon.avl_event_once` to be certain that handler will be removed immediately after single execution
```python
@wialon.avl_event_handler()
@wialon.avl_event_once
async def unit_event(event: AvlEvent):
print("Handler got event:", event)
```
## Exceptions Handling
The avl_event_handler suppress the callback's WialonError exceptions to protect app to be closed on unexpected behaviour
So if u want to handle some specific WialonError, do it in handler's callback scope
> [!NOTE]
> You still can get access to response data even if WialonError exception was raised, [see next section](#exceptions-handling-batch)
```python
from aiowialon import WialonError, WialonAccessDenied
@wialon.avl_event_handler()
async def unit_event(event: AvlEvent):
try:
raise WialonAccessDenied # example of wialon exception raised in callback scope
except WialonError as err:
# do something
pass
```
### Exceptions Handling (Batch)
You still can get access to response data even if WialonError exception was raised
It can be usefull for debug or for the batch requests
`WialonError.reason` returns string for single call or `list[WialonError]` for batch call
```python
async def some_func():
result = None
try:
result = await wialon.batch(*calls, flags_=flags.BatchFlag.STOP_ON_ERROR)
except WialonError as err:
print("Errors", err.reason) # returns a list of WialonErrors for each call in batch
result = err.result
finally:
print("Result", result)
```
## Quick API Help
Use `Wialon.help(service_name, action_name)` to open Wialon Remote API docs in your system browser
```python
from aiowialon import Wialon
Wialon.help('core', 'search_item')
```
## Advanced usage
### Limitations
Adjusting to the Wialon API limitations the Wialon API client limited to 10 connections maximum per session wia `asyncio.semaphore`
Also it limited to 10 requests per second for the session with `aiolimiter`
You can set custom limit of requests per second for your requirements
```python
from aiowialon import Wialon
wialon = Wialon(rps=15) # set custom requests per second limit
```
### Prevent polling logout
By default `start_polling` autologout on `Exception` or on manual `stop_polling`. You can adjust it to your requirements
```python
from aiowialon import Wialon
wialon = Wialon() # set custom requests per second limit
wialon.start_polling(token=TOKEN, logout_finally=False)
```
### Critical requests execution
#### Async session lock
Some requests to services like `Render`, `Reports`, `Messages` requires blocking other requests to be executed together per single session.
* Use the `@Wialon.lock_session` decorator to block async loop till your operation done
* You can apply `@Wialon.session_lock` also for handlers
* You can use `@Wialon.session_lock` inside the methods when [inheriting Wialon](#extending-aio-wialon)
```python
import asyncio
from functools import wraps
from aiowialon import Wialon
wialon = Wialon(token=TOKEN)
@wialon.session_lock
async def critical_method(self, params1, params2):
# For example: execute and export report
previous_request_timeout = self.timeout # Store current timeout
try:
self.timeout = 600 # Setup request timeout up to 10 minutes
await self.report_exec_report(**params1)
self.timeout = previous_request_timeout # Return previous timeout
report_result = await self.export_result(**params2)
return report_result
finally:
self.timeout = previous_request_timeout # Return previous timeout
await self.report_cleanup_result()
```
With handlers:
```python
@wialon.avl_event_handler(lambda event: event.data.i == 734455)
@wialon.session_lock
async def unit_event(event: AvlEvent):
print("Handler got event:", event)
# simulating long operation
for i in range(5):
print("Waiting lock release", i)
await asyncio.sleep(1)
```
#### Timeout for API call
Some API calls requires special timeouts, cause them are processing long.
Default timeout for aiohttp request is 5 seconds.
You can set custom timeout on some call executing.
It mostly usefull with `@Wialon.session_lock`
```python
@wialon.avl_event_handler()
@wialon.session_lock
async def unit_event(event: AvlEvent):
try:
await wialon.wait(wialon.messages_load_last(
itemId=event.data.i,
lastTime=event.tm,
lastCount=10000,
flags=0x0000,
flagsMask=0xFF00,
loadCount=10000
), 10)
except (TimeoutError, WialonError) as err:
print(err)
for i in range(5):
print("Waiting exclusive operation", i, "item:", event.data.i)
await asyncio.sleep(1)
```
### Extending AIO Wialon
Inherit from `Wialon` class to add your custom logic and behaviour
* You can directly use `Wialon.request` to make requests to special endpoints
* You can use `@wialon.session_lock` inside the methods when inheriting Wialon
```python
import json
import asyncio
from aiowialon import Wialon
class CustomWialon(Wialon):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.__geocode_url = f"{kwargs.get('scheme', 'https')}://geocode-maps.wialon.com/{self.__base_url}/gis_geocode"
async def geocode_address(self, coords, city_radius, dist_from_unit, txt_dist, flags):
payload = {
'coords': coords,
... # other fields
}
return await self.request('geocode_fetch', self.__geocode_url, payload=json.dumps(payload))
async def critical_method(self):
@self.session_lock
async def locked_task():
# simulating long operation
for i in range(5):
print("Waiting lock release", i)
await asyncio.sleep(1)
return await locked_task()
```
### Debugging
Enable debug messages for `aiowialon` and `aiohttp`
```python
import logging
from aiowialon import Wialon, WialonError, flags, AvlEvent
logging.basicConfig(level=logging.DEBUG)
```
> [!WARNING]
> ### RISK NOTICE
> THE CODE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE MATERIALS OR THE USE OR OTHER DEALINGS IN THE MATERIALS.
*Copyright 2023 Yaroshenko Dmytro (https://github.com/o-murphy)*
Raw data
{
"_id": null,
"home_page": null,
"name": "py-aiowialon",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "asyncio, wialon, wialon remote api, gurtam",
"author": null,
"author_email": "o-murphy <thehelixpg@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/57/ec/d07efa735556f8a06c1f70fdb68976e0143119f3ff85d0bf372d3db836cc/py_aiowialon-1.3.3.tar.gz",
"platform": null,
"description": "[![SWUbanner]][SWUdocs]\n\n[SWUbanner]:\nhttps://raw.githubusercontent.com/vshymanskyy/StandWithUkraine/main/banner-direct-single.svg\n[SWUdocs]:\nhttps://github.com/vshymanskyy/StandWithUkraine/blob/main/docs/README.md\n\n\n\n# AIO Wialon\n[](https://opensource.org/licenses/MIT)\n[](https://pypi.org/project/py-aiowialon/)\n\n`AIO Wialon` is an async implementation of Python wrapper for Wialon Remote API, \n\n### Table of \n* [Installation](#installation)\n* [Start Polling](#start-polling)\n* [Wialon API Call](#wialon-api-call)\n * [API Call Example](#api-call-example)\n * [Batch requests](#batch-requests)\n * [Multipart requests](#multipart-requests)\n * [Shortcuts](#shortcuts)\n* [Wialon Events](#wialon-events)\n * [Register AVL Events](#register-avl-events)\n * [On login/logout](#on-loginlogout)\n * [AVL Events Handling](#avl-events-handling)\n * [Register AVL Events handler](#register-avl-events-handlers)\n * [Remove AVL Events handler](#remove-avl-events-handlers)\n * [Disposable handlers](#disposable-handlers)\n* [Exceptions Handling](#exceptions-handling)\n * [Get exception results, batch exceptions](#exceptions-handling-batch)\n* [Quick API Help](#quick-api-help)\n* [Advanced](#advanced-usage)\n * [Limitations](#limitations)\n * [Prevent polling auto logout](#prevent-polling-logout)\n * [Critical requests execution (Render, Reports, Messages)](#critical-requests-execution)\n * [Async session lock](#async-session-lock)\n * [Timeout for API call](#timeout-for-api-call)\n * [Extending AIO Wialon](#extending-aio-wialon)\n * [Debugging](#debugging)\n \n* [Wialon Remote Api documentation](http://sdk.wialon.com/wiki/en/sidebar/remoteapi/apiref/apiref)\n\n## Installation\n```bash\npip install py-aiowialon\n```\n\n## Start Polling\nOpen session and start poll AVL events immediately.\n[Look the Wialon Events section](#wialon-events) to see how we can handle AVL Events on polling\n```python\nimport asyncio\nfrom aiowialon import Wialon\n\nTOKEN = '<Your Wialon API access token>'\nHOST = '<API host IP or url>'\nwialon = Wialon(host=HOST, token=TOKEN)\n\nif __name__ == \"__main__\":\n asyncio.run(wialon.start_polling())\n```\n> [!TIP]\n> `Wialon.start_polling()` is not require a manual Wialon.login() call \n\n## Wialon API Call\nAPI Call is function that returns `Wialon.call()` instance\nAlmost all Wialon Remote API `services/actions` available \nthrough dot syntax: `wialon.<service>_<action_name>(**params)`\n\nTo make API call use method of Wialon instance with same name as API endpoint\nreplace `/` with underscore.\n\n#### API Call Example\n```python\nimport asyncio\nfrom aiowialon import Wialon, flags\n\nTOKEN = '<Your Wialon API access token>'\nwialon = Wialon(token=TOKEN)\n\nasync def main():\n await wialon.login()\n # The example of core/search_item API call:\n result = await wialon.core_search_item(id=12345, flags=flags.UnitsDataFlag.ALL)\n print(result)\n await wialon.logout()\n\nasyncio.run(main())\n```\n\n> [!WARNING]\n> Some Wialon Remote API methods requires a lock of asynchronous context \n> (execution of reports, loading messages, etc). \n> If you need these methods, \n> it's highly recommended to get acquainted with [**Critical requests execution**](#critical-requests-execution) section\n\n### Batch requests\nUse `Wialon.batch` instead of `asyncio.gather` to make multiple API calls in a same time.\nIt allows to make just one request to server with few calls.\nThis avoids reaching the server's request limits. \nAnd transfers the overhead of processing asynchronous context to the server side.\nFew Wialon.call() coroutines would be batched to single 'core/batch' request.\n```python\n# put few calls to a batch method\nfrom aiowialon import Wialon, flags\n\nwialon = Wialon(token=TOKEN)\n\nasync def some_func(params1, params2):\n api_calls = [\n wialon.core_search_item(**params1),\n wialon.unit_get_fuel_settings(**params2),\n ...\n ]\n return await wialon.batch(*api_calls, flags_=flags.BatchFlag.EXECUTE_ALL)\n```\n> [!TIP]\n> * You can combine different API services and actions in single batch call\n> * [How to handle batch exceptions](#exceptions-handling-batch)\n\n> [!WARNING]\n> * Some requests don't support batch!\n> * Don't try to put batch into other batch, it can raise unexpected behaviour\n> * Go to the [Wialon Remote Api documentation](http://sdk.wialon.com/wiki/en/sidebar/remoteapi/apiref/apiref) to get details\n\n### Multipart requests\nUse `Wialon.multipart` method and `MultipartField` with API call to but multipart data to request,\nPut call coroutine and required MultipartField instances to the `Wialon.multipart()`\n```python\nfrom aiowialon import Wialon, MultipartField\n\nwialon = Wialon(token=TOKEN)\n\nasync def upload_driver_image():\n event_hash = 'aiowialon_drv_upd' # custom event hash\n params = {\"itemId\": 717351, \"driverId\": 38, \"eventHash\": event_hash}\n file_path = \"driver_img.jpg\"\n with open(file_path, 'rb') as f:\n file_data = f.read()\n\n await wialon.multipart(\n wialon.resource_upload_driver_image(**params),\n *[\n MultipartField(\n name='drivers_dlg_props_upload_image',\n value=file_data,\n filename=\"image.jpg\",\n content_type='image/jpeg'\n )\n ]\n )\n```\n> [!WARNING]\n> * Don't try to put multipart requests to batch!\n> * Some requests don't support multipart\n> * Don't try to put multipart request into batch, it can raise unexpected behaviour\n> * Go to the [Wialon Remote Api documentation](http://sdk.wialon.com/wiki/en/sidebar/remoteapi/apiref/apiref) to get details\n\n### Shortcuts\nShortcuts are available as efficient solutions for some common actions, like .wlp export\n\n```python\nfrom aiowialon import Wialon\nfrom aiowialon.shortcuts import WLP\n\nwialon = Wialon(token=TOKEN)\n\n\nasync def dump_unit(item_id):\n await wialon.login()\n wlp = await WLP.export_item(wialon, item_id)\n with open(f\"{id}.wlp\", 'wb') as fp:\n fp.write(wlp)\n```\n\n## Wialon Events\nThe library propose using the polling to handle AVL Events.\nAVL events is the events that happens on the server and returns to us if we registered it in current session\nThis section references to [Wialon AVL Events Docs](https://sdk.wialon.com/wiki/en/sidebar/remoteapi/apiref/requests/avl_evts)\n\n### Register AVL Events\nFirstly we have to register items for AVL events handling in current session.\n_[(api reference here)](https://sdk.wialon.com/wiki/en/sidebar/remoteapi/codesamples/update_datafalags)_\n\nBellow is example how to add all AVL Units (vehicles) to handle AVL events of this units in current session,\nWe use there just simple Wialon API Call\n```python\nfrom aiowialon import Wialon, flags\n\nwialon = Wialon(token=TOKEN)\n\nasync def register_avl_events():\n spec = [\n {\n \"type_\": \"type\",\n \"data\": \"avl_unit\",\n \"flags\": flags.UnitsDataFlag.BASE | flags.UnitsDataFlag.POS,\n \"mode\": 0\n }\n ]\n return await wialon.core_update_data_flags(spec=spec)\n```\n\n### On login/logout\nWe can automate this logic for each session opening by registering `on_session_open` callback,\nUse `@wialon.on_session_open` decorator for this\nSo wialon will login and register avl items to polling before polling start\n```python\n@wialon.on_session_open\nasync def register_avl_events(session_login):\n print(\"Session eid:\", session_login['eid'])\n spec = [\n {\n \"type_\": \"type\",\n \"data\": \"avl_unit\",\n \"flags\": flags.UnitsDataFlag.BASE | flags.UnitsDataFlag.POS,\n \"mode\": 0\n }\n ]\n return await wialon.core_update_data_flags(spec=spec)\n\nif __name__ == \"__main__\":\n asyncio.run(wialon.start_polling())\n```\n\nAlso we can add callback on session logout. Use `@wialon.on_session_close` decorator for this\n\n```python\n@wialon.on_session_close\nasync def on_session_close(session_logout):\n print(\"Logout event:\", session_logout)\n```\n> [!NOTE]\n> * You can register just one `on_session_open` callback for Wialon instance\n> * You can register just one `on_session_close` callback for Wialon instance\n\n\n### AVL Events Handling\nAfter polling start and AVL Items registered for polling we can handle the AVL Events.\nUse `@wialon.avl_event_handler()` decorator\n\n#### Register AVL Events handlers\n```python\nfrom aiowialon import AvlEvent\n\n\n@wialon.avl_event_handler()\nasync def unit_event(event: AvlEvent):\n print(\"Handler got event:\", event)\n```\n\nPut the filter function to the decorator to apply filtering of AVL events\n\n```python\nfrom aiowialon import AvlEvent\n\n\n@wialon.avl_event_handler(lambda event: event.data.i == 734455)\nasync def unit_734455_event(event: AvlEvent):\n print(\"Handler got event from item 734455:\", event)\n```\n> [!NOTE]\n> Register handlers in an order in which filters have to be applied. If some handler catched the event, next handler in order will never do.\n\n#### Remove AVL Events handlers\n```python\n# use them as you need\nwialon.remove_avl_event_handler('handler_name')\nwialon.remove_avl_event_handler(handler_func)\n```\n\n#### Disposable handlers\nUse `@wialon.avl_event_once` to be certain that handler will be removed immediately after single execution\n```python\n@wialon.avl_event_handler()\n@wialon.avl_event_once\nasync def unit_event(event: AvlEvent):\n print(\"Handler got event:\", event)\n```\n\n\n## Exceptions Handling\nThe avl_event_handler suppress the callback's WialonError exceptions to protect app to be closed on unexpected behaviour\nSo if u want to handle some specific WialonError, do it in handler's callback scope\n\n> [!NOTE]\n> You still can get access to response data even if WialonError exception was raised, [see next section](#exceptions-handling-batch)\n\n```python\nfrom aiowialon import WialonError, WialonAccessDenied\n\n\n@wialon.avl_event_handler()\nasync def unit_event(event: AvlEvent):\n try:\n raise WialonAccessDenied # example of wialon exception raised in callback scope\n except WialonError as err:\n # do something\n pass\n\n```\n\n### Exceptions Handling (Batch)\nYou still can get access to response data even if WialonError exception was raised\nIt can be usefull for debug or for the batch requests\n`WialonError.reason` returns string for single call or `list[WialonError]` for batch call\n```python\nasync def some_func():\n result = None \n try:\n result = await wialon.batch(*calls, flags_=flags.BatchFlag.STOP_ON_ERROR)\n except WialonError as err:\n print(\"Errors\", err.reason) # returns a list of WialonErrors for each call in batch\n result = err.result\n finally:\n print(\"Result\", result)\n```\n\n\n## Quick API Help\nUse `Wialon.help(service_name, action_name)` to open Wialon Remote API docs in your system browser\n```python\nfrom aiowialon import Wialon\n\nWialon.help('core', 'search_item')\n```\n\n\n## Advanced usage\n\n### Limitations\nAdjusting to the Wialon API limitations the Wialon API client limited to 10 connections maximum per session wia `asyncio.semaphore`\nAlso it limited to 10 requests per second for the session with `aiolimiter`\nYou can set custom limit of requests per second for your requirements\n```python\nfrom aiowialon import Wialon\nwialon = Wialon(rps=15) # set custom requests per second limit\n```\n\n### Prevent polling logout\nBy default `start_polling` autologout on `Exception` or on manual `stop_polling`. You can adjust it to your requirements\n```python\nfrom aiowialon import Wialon\nwialon = Wialon() # set custom requests per second limit\nwialon.start_polling(token=TOKEN, logout_finally=False)\n```\n\n### Critical requests execution\n\n#### Async session lock\nSome requests to services like `Render`, `Reports`, `Messages` requires blocking other requests to be executed together per single session.\n* Use the `@Wialon.lock_session` decorator to block async loop till your operation done\n* You can apply `@Wialon.session_lock` also for handlers\n* You can use `@Wialon.session_lock` inside the methods when [inheriting Wialon](#extending-aio-wialon)\n\n```python\nimport asyncio\nfrom functools import wraps\n\nfrom aiowialon import Wialon\n\nwialon = Wialon(token=TOKEN)\n\n@wialon.session_lock\nasync def critical_method(self, params1, params2):\n # For example: execute and export report\n previous_request_timeout = self.timeout # Store current timeout\n try:\n self.timeout = 600 # Setup request timeout up to 10 minutes\n await self.report_exec_report(**params1)\n self.timeout = previous_request_timeout # Return previous timeout\n report_result = await self.export_result(**params2)\n return report_result\n finally:\n self.timeout = previous_request_timeout # Return previous timeout\n await self.report_cleanup_result()\n```\n\nWith handlers:\n\n```python\n@wialon.avl_event_handler(lambda event: event.data.i == 734455)\n@wialon.session_lock\nasync def unit_event(event: AvlEvent):\n print(\"Handler got event:\", event)\n # simulating long operation\n for i in range(5):\n print(\"Waiting lock release\", i)\n await asyncio.sleep(1)\n```\n\n\n#### Timeout for API call\nSome API calls requires special timeouts, cause them are processing long. \nDefault timeout for aiohttp request is 5 seconds.\nYou can set custom timeout on some call executing.\nIt mostly usefull with `@Wialon.session_lock`\n```python\n@wialon.avl_event_handler()\n@wialon.session_lock\nasync def unit_event(event: AvlEvent):\n try:\n await wialon.wait(wialon.messages_load_last(\n itemId=event.data.i,\n lastTime=event.tm,\n lastCount=10000,\n flags=0x0000,\n flagsMask=0xFF00,\n loadCount=10000\n ), 10)\n except (TimeoutError, WialonError) as err:\n print(err)\n for i in range(5):\n print(\"Waiting exclusive operation\", i, \"item:\", event.data.i)\n await asyncio.sleep(1)\n```\n\n\n### Extending AIO Wialon\nInherit from `Wialon` class to add your custom logic and behaviour\n* You can directly use `Wialon.request` to make requests to special endpoints\n* You can use `@wialon.session_lock` inside the methods when inheriting Wialon\n\n\n```python\nimport json\nimport asyncio\nfrom aiowialon import Wialon\n\nclass CustomWialon(Wialon):\n def __init__(self, **kwargs):\n super().__init__(**kwargs)\n self.__geocode_url = f\"{kwargs.get('scheme', 'https')}://geocode-maps.wialon.com/{self.__base_url}/gis_geocode\"\n\n async def geocode_address(self, coords, city_radius, dist_from_unit, txt_dist, flags):\n payload = {\n 'coords': coords,\n ... # other fields\n }\n return await self.request('geocode_fetch', self.__geocode_url, payload=json.dumps(payload))\n\n async def critical_method(self):\n @self.session_lock\n async def locked_task():\n # simulating long operation\n for i in range(5):\n print(\"Waiting lock release\", i)\n await asyncio.sleep(1)\n return await locked_task()\n```\n\n### Debugging\nEnable debug messages for `aiowialon` and `aiohttp`\n```python\nimport logging\nfrom aiowialon import Wialon, WialonError, flags, AvlEvent\nlogging.basicConfig(level=logging.DEBUG)\n```\n\n> [!WARNING]\n> ### RISK NOTICE\n> THE CODE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE MATERIALS OR THE USE OR OTHER DEALINGS IN THE MATERIALS.\n\n*Copyright 2023 Yaroshenko Dmytro (https://github.com/o-murphy)*\n",
"bugtrack_url": null,
"license": "MIT License Copyright (c) 2013-2016 Gurtam Alex Chernetsky Copyright (c) 2022 Dmytro Yaroshenko (https://github.com/o-murphy) Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ",
"summary": "Async Wialon Remote API wrapper for Python 3",
"version": "1.3.3",
"project_urls": {
"Bug Reports": "https://github.com/o-murphy/py-aiowialon/issues",
"Homepage": "https://github.com/o-murphy/py-aiowialon",
"Source": "https://github.com/o-murphy/py-aiowialon"
},
"split_keywords": [
"asyncio",
" wialon",
" wialon remote api",
" gurtam"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "3fc2a1d51515757f164081997cd956eeef21959b4c23bf419c534d08ec373927",
"md5": "e10aa3bd4bde5abaadedfe0af92b3267",
"sha256": "5050ce8f7488e1579a1925d6ac5a3b6e4e156f38c8a9a41a01c1ef8d0fe909d2"
},
"downloads": -1,
"filename": "py_aiowialon-1.3.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "e10aa3bd4bde5abaadedfe0af92b3267",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 34951,
"upload_time": "2024-08-17T19:43:52",
"upload_time_iso_8601": "2024-08-17T19:43:52.179131Z",
"url": "https://files.pythonhosted.org/packages/3f/c2/a1d51515757f164081997cd956eeef21959b4c23bf419c534d08ec373927/py_aiowialon-1.3.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "57ecd07efa735556f8a06c1f70fdb68976e0143119f3ff85d0bf372d3db836cc",
"md5": "450c1a2e64ce25d714fe8746b18faaf1",
"sha256": "dd229e2dfd2d035a7f31a360c89ea0417ab0e589f1ce77bc5a7f3ca1225c9593"
},
"downloads": -1,
"filename": "py_aiowialon-1.3.3.tar.gz",
"has_sig": false,
"md5_digest": "450c1a2e64ce25d714fe8746b18faaf1",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 36014,
"upload_time": "2024-08-17T19:43:53",
"upload_time_iso_8601": "2024-08-17T19:43:53.821590Z",
"url": "https://files.pythonhosted.org/packages/57/ec/d07efa735556f8a06c1f70fdb68976e0143119f3ff85d0bf372d3db836cc/py_aiowialon-1.3.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-08-17 19:43:53",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "o-murphy",
"github_project": "py-aiowialon",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "py-aiowialon"
}
