ulogger-cloud 1.2.1

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)

Dynamic log configuration

The cloud dashboard can push log configuration changes to a connected device at any time. The library provides two functions for receiving these messages, depending on whether you need a one-shot fetch or a long-running listener.

Message format

Config messages are published by the dashboard to:

config/v0/{customer_id}/{application_id}/{device_serial}

The payload is a 9-byte binary header optionally followed by a UTF-8 JSON array of module name strings:

[level:u8][module_flags:u32le][timeout_secs:u32le][json_names:utf-8]

Bytes	Field	Description
0	level	Log level: 0=DEBUG 1=INFO 2=WARNING 3=ERROR 4=CRITICAL
1–4	module_flags	32-bit bitfield — which modules are active
5–8	timeout_secs	How long the new config stays active before the device reverts
9+	json_names	UTF-8 JSON array of enabled module name strings, e.g. ["COMM","SENSOR"]

The parsed payload is returned as a LogConfig dataclass:

from ulogger_cloud import LogConfig

# LogConfig fields:
cfg.log_level       # str:  "DEBUG", "INFO", "WARNING", "ERROR", or "CRITICAL"
cfg.module_flags    # int:  raw 32-bit bitfield
cfg.log_modules     # list: decoded module name strings, e.g. ["COMM", "SENSOR"]
cfg.timeout_seconds # int:  revert-to-default timeout in seconds

One-shot: wait_for_log_config

Blocks until one config message is received (or timeout seconds elapse) and returns it. Pass timeout=None to wait indefinitely.

import threading
from ulogger_cloud import DeviceInfo, MqttConfig, wait_for_log_config

config = wait_for_log_config(device, mqtt_cfg, timeout=60.0)
if config:
    print(f"Level: {config.log_level}, modules: {config.log_modules}")

Use the optional on_config callback to also receive the result asynchronously:

def handle_config(cfg):
    apply_to_device(cfg.log_level, cfg.module_flags, cfg.timeout_seconds)

config = wait_for_log_config(device, mqtt_cfg, timeout=60.0, on_config=handle_config)

Continuous: subscribe_log_config_loop

Stays connected and calls on_config every time a config message arrives. Blocks until stop_event is set — designed to be run in a thread-pool executor so it overlaps with other work (e.g. a BLE transfer).

import threading
from ulogger_cloud import DeviceInfo, MqttConfig, subscribe_log_config_loop

stop_event = threading.Event()

def on_config(cfg):
    print(f"New config: level={cfg.log_level} timeout={cfg.timeout_seconds}s")
    apply_to_device(cfg)

# Run in a background thread
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as pool:
    future = pool.submit(subscribe_log_config_loop, device, mqtt_cfg, on_config, stop_event)
    # ... do other work ...
    stop_event.set()   # clean shutdown
    future.result()

In an asyncio application, run it in the default executor so it doesn't block the event loop:

import asyncio
import threading
from ulogger_cloud import subscribe_log_config_loop

async def listen_for_config(device, mqtt_cfg):
    stop_event = threading.Event()
    loop = asyncio.get_running_loop()

    def on_config(cfg):
        # called from MQTT network thread — schedule work on the event loop
        asyncio.run_coroutine_threadsafe(apply_async(cfg), loop)

    try:
        await loop.run_in_executor(
            None,
            lambda: subscribe_log_config_loop(device, mqtt_cfg, on_config, stop_event),
        )
    finally:
        stop_event.set()

The example-bluetooth-demo uses exactly this pattern: the listener task starts immediately after BLE device info is read, receives configs in the background while the log transfer and upload proceed, and writes each config to the device over BLE as soon as it arrives. The task is cancelled (and stop_event set) when the BLE session ends.

Metrics and events

The library can publish device metrics (numeric measurements) and events (discrete occurrences) to the uLogger cloud. Both functions support two transports: MQTT (preferred, when certificates are configured) and HTTP POST (fallback, using an API token).

Publishing metrics

import time
from ulogger_cloud import MqttConfig, publish_metrics

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

metrics = [
    {"name": "cpu_temp", "value": 42.5, "timestamp": time.time()},
    {"name": "free_heap", "value": 81920},
]

success = publish_metrics(
    device_serial="ABC123",
    app_id="42",
    metrics=metrics,
    cfg=mqtt_cfg,
)

When MQTT certificates are not available, pass an API token instead:

success = publish_metrics(
    device_serial="ABC123",
    app_id="42",
    metrics=metrics,
    cfg=mqtt_cfg,
    api_token="ulg_your_api_token",
)

Publishing a heartbeat on behalf of a device

For devices that don't talk to MQTT directly — typically BLE devices that tether through a host application — publish_heartbeat lets the host send the required heartbeat metric without forcing the device-side ulogger_heartbeat() macro to burn NV memory waiting for the next sync.

from ulogger_cloud import MqttConfig, publish_heartbeat

# Call from a gateway/host whenever you've confirmed the device is alive
# (e.g. right after a successful BLE log transfer).
publish_heartbeat(
    device_serial="ABC123",
    app_id="42",
    cfg=mqtt_cfg,
)

The metric lands in device_metrics as name="heartbeat", value=1 — the same record the embedded ulogger_heartbeat() macro produces. Both paths satisfy the platform's "device is active" check.

Publishing events

import time
from ulogger_cloud import MqttConfig, publish_events

events = [
    {"type": "boot", "timestamp": time.time()},
    {"type": "ota_update_started"},
]

success = publish_events(
    device_serial="ABC123",
    app_id="42",
    events=events,
    cfg=mqtt_cfg,
)

The HTTP-token fallback works the same as for metrics:

success = publish_events(
    device_serial="ABC123",
    app_id="42",
    events=events,
    cfg=mqtt_cfg,
    api_token="ulg_your_api_token",
)

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.
wait_for_log_config(device, cfg, timeout=30.0, on_config=None)	Block until one log-config message is received; returns a LogConfig or None. Pass timeout=None to wait indefinitely.
subscribe_log_config_loop(device, cfg, on_config, stop_event)	Stay connected and call on_config for every config message until stop_event is set. Blocking — run in an executor.
publish_metrics(device_serial, app_id, metrics, cfg, api_token=None)	Publish a list of metric dicts (each with name, value, and optional timestamp). Uses MQTT when certs are available, otherwise HTTP with api_token.
publish_events(device_serial, app_id, events, cfg, api_token=None)	Publish a list of event dicts (each with type and optional timestamp). Uses MQTT when certs are available, otherwise HTTP with api_token.
publish_heartbeat(device_serial, app_id, cfg, api_token=None, timestamp=None)	Publish a single heartbeat metric (value=1) on behalf of a device that isn't directly connected to MQTT. Mirror of the embedded ulogger_heartbeat() macro.

LogConfig

Dataclass representing a log configuration pushed from the cloud dashboard:

Field	Type	Description
log_level	str	Minimum log level: "DEBUG", "INFO", "WARNING", "ERROR", or "CRITICAL"
module_flags	int	Raw 32-bit bitfield of enabled modules
log_modules	list[str]	Decoded module name strings, e.g. ["COMM", "SENSOR"]
timeout_seconds	int	Seconds before the device reverts to its default configuration