AIO Wialon

AIO Wialon is an async Python wrapper for the Wialon Remote API.

Table of Contents

• AIO Wialon
  • Table of Contents
  • Installation
    • With uv
    • With pip
  • Start Polling
  • Wialon API Call
    • API Call Example
    • Batch requests
    • Multipart requests
    • Shortcuts
  • Wialon Events
    • Register AVL Events
    • On login/logout
    • AVL Events Handling
      • Register AVL Events handlers
      • Remove AVL Events handlers
      • Disposable handlers
  • Exceptions Handling
    • Exceptions Handling (Batch)
  • Quick API Help
  • Advanced usage
    • Limitations
    • Context manager
    • Prevent polling logout
    • Critical requests execution
      • Async session lock
      • Timeout for API call
    • Extending AIO Wialon
    • Debugging

Installation

With uv

uv add py-aiowialon

With pip

pip install py-aiowialon

Start Polling

Open session and start polling AVL events immediately. Look the Wialon Events section to see how to handle AVL Events during polling.

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())

Wialon API Call

Almost all Wialon Remote API services/actions are available through dot syntax: wialon.<service>_<action_name>(**params)

Replace / with _ to map an API endpoint to a method name.

API Call Example

import asyncio
from aiowialon import Wialon, flags

TOKEN = '<Your Wialon API access token>'
wialon = Wialon(token=TOKEN)

async def main():
    await wialon.login()
    # 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())

Batch requests

Use Wialon.batch instead of asyncio.gather to make multiple API calls in one request. This avoids hitting server request limits and moves async overhead to the server side.

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)

Multipart requests

Use Wialon.multipart with MultipartField to send multipart data.

from aiowialon import Wialon, MultipartField

wialon = Wialon(token=TOKEN)

async def upload_driver_image():
    event_hash = 'aiowialon_drv_upd'
    params = {"itemId": 717351, "driverId": 38, "eventHash": event_hash}
    with open("driver_img.jpg", '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'
        )
    )

Shortcuts

Shortcuts are pre-built helpers for common operations, such as .wlp export.

from aiowialon import Wialon, 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"{item_id}.wlp", 'wb') as fp:
        fp.write(wlp)

Wialon Events

The library uses polling to handle AVL Events — events that occur on the server and are returned when registered for the current session. See Wialon AVL Events Docs.

Register AVL Events

Register items for AVL event handling in the current session. (api reference)

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

Use @wialon.on_session_open to run logic automatically after each login.

@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())

Use @wialon.on_session_close to run logic after logout:

@wialon.on_session_close
async def on_session_close(session_logout):
    print("Logout event:", session_logout)

AVL Events Handling

After polling starts and AVL items are registered, use @wialon.avl_event_handler() to handle events.

Each handler runs its callback in a sequential queue — events for a given handler are processed one at a time, in order, without unbounded parallelism.

Register AVL Events handlers

from aiowialon import AvlEvent


@wialon.avl_event_handler()
async def unit_event(event: AvlEvent):
    print("Handler got event:", event)

Apply a filter to handle only matching events:

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)

Remove AVL Events handlers

wialon.remove_avl_event_handler('handler_name')
wialon.remove_avl_event_handler(handler_func)

Disposable handlers

Use @wialon.avl_event_once to automatically remove a handler after its first execution.

@wialon.avl_event_handler()
@wialon.avl_event_once
async def unit_event(event: AvlEvent):
    print("Handler got event:", event)

Exceptions Handling

avl_event_handler catches and logs WialonError and aiohttp.ClientError from callbacks to prevent a single failure from stopping the polling loop. Handle specific errors inside the callback scope if needed.

from aiowialon import WialonError, WialonAccessDenied


@wialon.avl_event_handler()
async def unit_event(event: AvlEvent):
    try:
        await wialon.core_search_item(id=event.data.i, flags=1)
    except WialonAccessDenied as err:
        print("Access denied:", err)
    except WialonError as err:
        print("Wialon error:", err)

Exceptions Handling (Batch)

WialonError.reason returns a string for single calls or list[WialonError] for batch calls. WialonError.result carries the raw response data.

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)  # list[WialonError] for batch
        result = err.result
    finally:
        print("Result", result)

Quick API Help

Open Wialon Remote API docs in your browser:

from aiowialon import Wialon

Wialon.help('core', 'search_item')

Advanced usage

Limitations

The client is limited to 10 concurrent connections via asyncio.Semaphore and to 10 requests per second via aiolimiter, matching Wialon API server limits. Adjust rps to match your account tier:

from aiowialon import Wialon
wialon = Wialon(rps=15)

Context manager

Wialon supports use as an async context manager, which ensures the underlying HTTP session is closed cleanly even if an exception occurs:

async def main():
    async with Wialon(token=TOKEN) as wialon:
        await wialon.login()
        result = await wialon.core_search_item(id=12345, flags=1)
        print(result)
        await wialon.logout()

Prevent polling logout

By default start_polling logs out when stopped or on exception. Disable with logout_finally=False:

from aiowialon import Wialon
wialon = Wialon(token=TOKEN)
asyncio.run(wialon.start_polling(logout_finally=False))

Critical requests execution

Async session lock

Some API calls (reports, messages, render) must run exclusively — no other requests should be sent to the same session concurrently. Use @wialon.session_lock to block the session until the decorated coroutine completes.

import asyncio
from aiowialon import Wialon

wialon = Wialon(token=TOKEN)


@wialon.session_lock
async def run_report(params1, params2):
    try:
        wialon.timeout = 600  # reports can take a long time
        await wialon.report_exec_report(**params1)
        return await wialon.report_export_result(**params2)
    finally:
        wialon.timeout = 5
        await wialon.report_cleanup_result()

With event handlers:

@wialon.avl_event_handler(lambda event: event.data.i == 734455)
@wialon.session_lock
async def unit_event(event: AvlEvent):
    print("Handler got event:", event)
    for i in range(5):
        print("Exclusive operation in progress:", i)
        await asyncio.sleep(1)

Timeout for API call

Use Wialon.wait() to set a per-call timeout without changing the global default:

@wialon.avl_event_handler()
@wialon.session_lock
async def unit_event(event: AvlEvent):
    try:
        messages = await wialon.wait(
            wialon.messages_load_last(
                itemId=event.data.i,
                lastTime=event.tm,
                lastCount=10000,
                flags=0x0000,
                flagsMask=0xFF00,
                loadCount=10000,
            ),
            timeout=30,
        )
    except (TimeoutError, WialonError) as err:
        print(err)

Extending AIO Wialon

Inherit from Wialon to add custom logic. Use self.request() for direct HTTP calls and @self.session_lock for exclusive access:

import json
import asyncio
from aiowialon import Wialon


class CustomWialon(Wialon):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        scheme = kwargs.get('scheme', 'https')
        host = kwargs.get('host', 'hst-api.wialon.com')
        self._geocode_url = f"{scheme}://geocode-maps.wialon.com/{host}/gis_geocode"

    async def geocode_address(self, coords, city_radius, dist_from_unit, txt_dist, flags):
        payload = json.dumps({
            'coords': coords,
            'cityRadius': city_radius,
            'distFromUnit': dist_from_unit,
            'txtDist': txt_dist,
            'flags': flags,
        })
        return await self.request('gis_geocode', self._geocode_url, payload)

    async def critical_method(self):
        @self.session_lock
        async def locked_task():
            for i in range(5):
                print("Exclusive operation:", i)
                await asyncio.sleep(1)
        return await locked_task()

Debugging

Enable debug logging for aiowialon and aiohttp.client:

import logging
logging.basicConfig(level=logging.DEBUG)

Copyright 2023 Yaroshenko Dmytro (https://github.com/o-murphy)