ulogger-cloud 0.3.0

uLogger cloud communication library – MQTT session management, binary log upload, and data validation.

ulogger-cloud

A Python library for communicating with the uLogger cloud platform. It handles:

• Device registration – MQTT boot handshake to obtain a session token.
• Binary log upload – publish raw log data to the cloud for server-side parsing and visualisation.
• Header checksum validation – verify the integrity of a firmware log buffer before upload.
• Session token caching – persist tokens to disk so re-registration is skipped on subsequent runs (or across firmware upgrades).

Installation

pip install ulogger-cloud

Requires Python 3.9 or later. The only runtime dependency is paho-mqtt (>=1.6), which is installed automatically.

Quick start

High-level API (recommended)

upload_log handles everything in one call: checksum validation, session-token retrieval (with caching), header patching, and MQTT publish.

from pathlib import Path
from ulogger_cloud import DeviceInfo, MqttConfig, upload_log

# Device identification – normally parsed from a BLE characteristic
device = DeviceInfo(
    application_id=42,
    device_serial="ABC123",
    device_type="my_board",
    git_version="v1.0.0",
    git_hash="abcdef0",
)

# MQTT broker configuration
mqtt_cfg = MqttConfig(
    cert_file=Path("certificate.pem.crt"),
    key_file=Path("private.pem.key"),
    customer_id=975773647,
)

# buf is a bytearray received over BLE
success = upload_log(device, buf, mqtt_cfg)
print("Upload OK" if success else "Upload failed")

Persistent session-token cache

By default upload_log uses an in-memory token store that is discarded when the process exits. To avoid a new MQTT boot handshake on every run, pass a file-backed SessionStore:

from ulogger_cloud import (
    DeviceInfo, MqttConfig, SessionStore,
    DEFAULT_FILE_STORE_PATH, upload_log,
)

store = SessionStore(path=DEFAULT_FILE_STORE_PATH)  # ~/.ulogger/session_tokens.json

success = upload_log(device, buf, mqtt_cfg, store=store)

Tokens are automatically invalidated whenever the device firmware's git_hash changes, triggering a fresh registration transparently.

Low-level building blocks

If you need finer control you can call each step individually:

from pathlib import Path
from ulogger_cloud import (
    DeviceInfo,
    MqttConfig,
    get_session_token,
    patch_session_token,
    publish_binary_log,
    validate_checksum,
)

device = DeviceInfo(
    application_id=42,
    device_serial="ABC123",
    device_type="my_board",
    git_version="v1.0.0",
    git_hash="abcdef0",
)

mqtt_cfg = MqttConfig(
    cert_file=Path("certificate.pem.crt"),
    key_file=Path("private.pem.key"),
    customer_id=975773647,
)

buf = bytearray(raw_ble_transfer)
if validate_checksum(bytes(buf)):
    token = get_session_token(device, mqtt_cfg)
    if token is not None:
        patch_session_token(buf, token)
    publish_binary_log(device, bytes(buf), mqtt_cfg)

API reference

DeviceInfo

Dataclass holding device identification read from the firmware:

Field	Type	Description
application_id	int	Application ID (uint32)
device_serial	str	Unique device serial string
device_type	str	Board / product type string
git_version	str	Human-readable firmware version tag
git_hash	str	Short git commit hash of the firmware

MqttConfig

Dataclass for MQTT broker connection parameters:

Field	Type	Default	Description
broker	str	"mqtt.ulogger.ai"	Broker hostname
port	int	8883	Broker port (TLS)
cert_file	Path | None	None	Path to client certificate (.pem.crt)
key_file	Path | None	None	Path to private key (.pem.key)
customer_id	int	0	Customer account ID
token_timeout	float	15.0	Seconds to wait for session-token response

SessionStore

Thread-safe mapping of device serial numbers to session tokens.

from ulogger_cloud import SessionStore, DEFAULT_FILE_STORE_PATH

# In-memory only (default)
store = SessionStore()

# File-backed persistence
store = SessionStore(path=DEFAULT_FILE_STORE_PATH)

# Memory-capped + file-backed (keeps the 1 000 most-recently-used tokens)
store = SessionStore(path=DEFAULT_FILE_STORE_PATH, max_entries=1000)

Functions

Function	Description
upload_log(device, buf, cfg, store=None)	Validate, patch, and publish a binary log buffer. Returns True on success.
get_or_fetch_token(device, cfg, store=None)	Return a cached token or perform a fresh MQTT boot registration.
get_session_token(device, cfg)	Perform an MQTT boot registration and return the server-issued token.
validate_checksum(buf)	Return True if the binary log header checksum is valid.
patch_session_token(buf, token)	Write a session token into the binary log header in-place.
publish_binary_log(device, buf, cfg)	Publish a raw binary log buffer to the MQTT broker.

Development

git clone https://github.com/your-org/ulogger-cloud.git
cd ulogger-cloud
pip install -e ".[dev]"
pytest