Metadata-Version: 2.4
Name: ulogger-cloud
Version: 1.2.1
Summary: uLogger cloud communication library – MQTT session management, binary log upload, and data validation.
Author-email: Eric Ibarra <support@ulogger.ai>
Maintainer-email: Eric Ibarra <support@ulogger.ai>
License: MIT
Project-URL: Homepage, https://ulogger.ai
Project-URL: Documentation, https://ulogger.ai/documentation
Project-URL: Release Notes, https://github.com/ulogger-ai/py-ulogger-cloud/releases
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Logging
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: paho-mqtt<3,>=1.6
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: requests; extra == "dev"
Dynamic: license-file

# ulogger-cloud

A Python library for communicating with the [uLogger](https://ulogger.ai) 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](https://ulogger.ai) 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](https://ulogger.ai)** — [Documentation](https://ulogger.ai/documentation.html) · [Solutions](https://ulogger.ai/solutions.html) · [Pricing](https://ulogger.ai/pricing.html)

## Installation

```bash
pip install ulogger-cloud
```

Requires Python 3.9 or later.  The only runtime dependency is
[paho-mqtt](https://pypi.org/project/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.

```python
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`:

```python
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:

```python
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:

```python
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.

```python
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:

```python
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).

```python
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:

```python
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

```python
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:

```python
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.

```python
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

```python
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:

```python
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.

```python
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 |

