ulogger-cloud 1.1.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).

About uLogger

uLogger is an embedded observability platform purpose-built for debugging embedded systems in production. It gives firmware and IoT teams the visibility they need to understand what actually happened in the field — eliminating blind debugging and "cannot reproduce" dead ends.

Core capabilities:

• Pre-trigger logging — a circular ring buffer continuously captures events so the full context before a crash is always preserved, enabling real embedded core dump analysis and stack trace inspection.
• Automatic crash capture — hard faults, watchdog resets, and stack overflows are intercepted automatically with register state, call stack, and buffered log history queued for transmission.
• Compressed logging — logs are compressed up to 90%, making uLogger a true low-bandwidth logging solution for constrained IoT device monitoring environments.
• AI-powered bug resolution — structured crash data is fed directly into AI tooling for automated root cause analysis, turning a fault into a proposed fix without manual triage.
• Remote log capture — capture and debug IoT devices remotely across your entire fleet, whether you have ten devices or ten thousand.

uLogger addresses the observability gaps that traditional logging tools leave behind: missing logs from field devices, intermittent firmware bugs, and the lack of visibility into embedded systems that leads to costly field failure escalations. It is designed for industrial IoT monitoring, predictive maintenance, edge device observability, and any environment where debugging firmware in production is a daily reality.

Learn more at ulogger.ai — Documentation · Solutions · Pricing

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.