pontem 0.8.0

Pontem edge SDK — logs, metrics, and config for edge devices

Pontem Python SDK

Logs, metrics, and config for processes running on Pontem-managed edge devices. Zero runtime dependencies — stdlib only.

Requirements

• Python 3.9+.

Install

pip install pontem

Quick start

import pontem

pontem.init(service_name="my-service")

pontem.logger.info("model loaded", model="scoring_v3")

pontem.metrics.count("frames_processed")
with pontem.metrics.timer("model.inference"):
    result = model.predict(frame)

pontem.shutdown()  # also runs at process exit

That's it. Logs land as JSONL under $PONTEM_EMIT_DIR for the agent to ship; metrics POST to a local OpenTelemetry collector. See Metrics → Setup for the collector requirement.

Already using stdlib logging? Pass stdlib_logging=True to init and your existing logging.getLogger(...).info(...) calls produce Pontem records — no call-site changes. See Use with stdlib logging below.

---

Logging

OTel-aligned severity levels:

Method	OTel SeverityNumber	When to use
trace	1	Fine-grained debugging
debug	5	Diagnostic information
info	9	Normal operational events
warn	13	Unexpected but recoverable
error	17	Errors that need attention
fatal	21	Unrecoverable failures

The enum is at pontem.log.Level (Level.TRACE, Level.INFO, …).

Direct API

pontem.logger.info("model loaded", model="scoring_v3")
pontem.logger.warn("high latency", latency_ms=120, endpoint="/predict")
pontem.logger.error("inference failed", error=str(e), frame_id=frame.id)

Keyword arguments become structured attributes. Calls are non-blocking and queued; serialization and disk I/O run on the background thread.

This is what you want on hot paths.

Use with stdlib logging

If your code already uses logging.getLogger(...).info(...), enable the drop-in path:

import logging
import pontem

logging.basicConfig(level=logging.INFO)            # 1. set up your handlers first
pontem.init(service_name="my-service",             # 2. then init with the flag
            stdlib_logging=True)

logging.getLogger(__name__).info("model loaded", extra={"model": "v3"})

What it does: installs PontemFormatter on every handler currently on the root logger. Destinations, rotation policies, and filters stay intact — only the on-the-wire format changes.

Call order matters. The flag swaps formatters on handlers attached to root at the time of init. Run basicConfig / dictConfig / addHandler first; otherwise init raises. Handlers added after init are not picked up automatically — call PontemFormatter.install() again to apply to them.

You can mix paths freely: pontem.logger.* on hot inference loops, stdlib elsewhere. Both produce the same wire shape.

The SDK's own logs (under the pontem logger namespace) propagate to your chain too. Quiet them with stdlib mechanisms:

logging.getLogger("pontem").setLevel(logging.WARNING)  # WARN+ only
logging.getLogger("pontem").propagate = False          # drop entirely

Custom formatter setup

When you need finer control than the stdlib_logging=True flag — e.g. attaching the formatter to specific handlers, or wiring it through dictConfig:

from logging.handlers import RotatingFileHandler
from pontem.log import PontemFormatter

handler = RotatingFileHandler("/var/log/myapp/app.log", maxBytes=10_000_000)
handler.setFormatter(PontemFormatter(service_name="my-service"))
logging.getLogger().addHandler(handler)

dictConfig (YAML / JSON):

formatters:
  pontem:
    (): pontem.log.PontemFormatter
    service_name: my-service
handlers:
  console:
    class: logging.StreamHandler
    formatter: pontem
root:
  handlers: [console]
  level: INFO

Resource attributes (service.name, service.version) come from constructor kwargs first, falling back to whatever pontem.init() populated. service.name is required from at least one source.

For both formatter paths:

• Set the root level. Stdlib defaults to WARNING; info/debug records are filtered before reaching any handler. basicConfig(level=logging.INFO) is the standard fix.
• No emit pipeline. Records flow through your handler's I/O (sync FileHandler, etc.), not through Pontem's bounded queue + background writer. For non-blocking, queued, rotation-and-gzip behavior on hot paths, use the direct API.

journald (host systemd services)

For a host-native systemd service (not a container), send logs to the local systemd journal in the Pontem field shape:

import logging
import pontem

logging.basicConfig(level=logging.INFO)
pontem.init(service_name="my-service", emit_target="journald")
pontem.log.JournaldHandler.install()   # journald becomes root's sole destination
logging.getLogger(__name__).warning("disk almost full", extra={"free_mb": 12})

install() removes the default basicConfig handler so records aren't sent twice. The direct API (pontem.logger.*) also reaches the journal when emit_target="journald".

On a host without systemd (local dev, CI, containers — no /run/systemd/journal/socket), the handler writes JSON to stderr instead; no code change.

Attribute keys are uppercased in the journal (frame_id → FRAME_ID, which journald requires). The stdout and file targets keep the original case, so the same key can differ in case between a host-systemd deployment and a container deployment of the same service.

These logs reach Cloud Logging only when pontem-log-collector on the device is collecting this service's journal.

---

Metrics

Aggregated in memory; the background thread POSTs OTLP/HTTP/JSON batches to a local OpenTelemetry collector every second. All public methods return in under 1µs.

Setup

Metrics need a collector listening on the device. The collector accepts OTLP/HTTP on port 4318 by default.

The SDK targets a local collector by default so compose packages just work. Override for host-native or K3s deployments:

export PONTEM_OTLP_ENDPOINT=http://127.0.0.1:4318                  # host-native
export PONTEM_OTLP_ENDPOINT=http://<collector-service>:4318        # K3s

The SDK attaches service to every datapoint. The collector adds device_id on the way through, so each datapoint carries both labels. The metric API doesn't accept caller-supplied labels.

Counter

Cumulative — each flush reports the running total since process start.

pontem.metrics.count("frames_processed")
pontem.metrics.count("bytes_sent", len(payload))

metrics.count(name, amount=1)

Histogram

Cumulative count, sum, min, max since process start.

pontem.metrics.record("payload_size", len(data), unit="bytes")

metrics.record(name, value, *, unit="")

Gauge

Point-in-time — last write wins per flush interval.

pontem.metrics.set_gauge("gpu_temp", 72.0, unit="celsius")
pontem.metrics.set_gauge("queue_depth", len(queue))

metrics.set_gauge(name, value, *, unit="")

Timer

Context manager or decorator. Records elapsed time to a histogram.

with pontem.metrics.timer("model.inference"):
    result = model.predict(frame)

@pontem.metrics.timer("preprocessing")
def preprocess(frame):
    ...

metrics.timer(name, *, unit="s")

Reliability

OTLP POSTs that fail (collector down, network blip) buffer to a bounded in-memory retry queue (configurable via metric_otlp_queue_size=, drop-oldest on overflow). Counter/histogram resets across process restarts are reported with a fresh start_time so the segments render correctly downstream.

Cardinality

The SDK caps distinct metric names per process (configurable via metric_name_limit=). Names past the cap are dropped with a one-time warning. Don't generate metric names from user input.

---

Config

Reads agent-managed values from $PONTEM_CONFIG_DIR/<namespace>.json. Each namespace is its own JSON file: a top-level {key: value} map whose values are any JSON type — strings, numbers, booleans, or nested objects and arrays.

Both access modes return an immutable Snapshot — read it with .get(key, default), .require(key) (raises if absent), or .as_dict(). A snapshot never changes once handed out, so reading several keys off one snapshot can't tear across a mid-read update.

install(ns) — read once at startup, fixed for the process lifetime. The default, cheapest mode; use it for values that need a restart to change.

cfg = pontem.config.install("scoring")
threshold = cfg.get("confidence_threshold", 0.85)   # scalar
engine = cfg.require("engine")            # raises ConfigError if missing
cameras = cfg.get("cameras", {})          # object-typed key → a dict
transports = cfg.get("transport", [])     # array-typed key → a list

reloadable(ns) — a live handle whose .current() returns the latest snapshot, re-reading when the file changes (one stat() per call; re-reads only on change, keeps last-good on a malformed write). Use it for values the agent rewrites at runtime. Grab .current() once per work unit and pass the snapshot down:

live = pontem.config.reloadable("scoring")
cfg = live.current()                      # fresh as of this call
threshold = cfg.get("confidence_threshold", 0.85)

default is returned when either the namespace file or the key is absent.

---

Deployment

Docker / compose (stdout emit)

For containerized deployments where a sidecar log collector tails stdout, switch the direct API to stdout mode:

pontem.init(service_name="my-service", emit_target="stdout")

Or via env var (lets the same image run on edge devices and compose hosts):

PONTEM_EMIT_TARGET=stdout

Precedence: init(emit_target=...) > PONTEM_EMIT_TARGET > default "file".

In stdout mode, emit_dir, file rotation, and gzip compression are no-ops — the docker daemon's json-file driver handles container log rotation. This setting affects the direct API only; formatter paths always go through your handler chain.

File output and rotation

In file mode (default), only logs are written to disk:

$PONTEM_EMIT_DIR/
    logs.jsonl                # active (SDK writes)
    logs.jsonl.1713100000.gz  # rotated + compressed (agent picks up + deletes)

Files rotate at 10 MB, are gzip-compressed on the background thread, and up to 5 rotated files are kept. Metrics go over HTTP to the collector — see Metrics → Setup.

---

Reference

init()

pontem.init(
    service_name="my-service",      # required — identifies the service in all telemetry
    service_version="1.2.0",        # auto-detected from package metadata if omitted
    emit_dir="/custom/path",        # overrides PONTEM_EMIT_DIR (logs)
    emit_target="file",             # "file" (default), "stdout", or "journald"
    stdlib_logging=False,           # True → install PontemFormatter on root handlers
    otlp_endpoint=None,             # overrides PONTEM_OTLP_ENDPOINT (metrics)
    metric_name_limit=...,          # max distinct metric names per process
    metric_otlp_queue_size=...,     # max buffered failed POSTs (drop-oldest)
)

init() kwargs take precedence over environment variables. Call once at startup. Device identity (device_id) is assigned on the device by the platform, not by the SDK; it isn't a kwarg.

shutdown()

Flushes aggregated metrics, drains the log queue, and closes files. Registered automatically via atexit; call explicitly if you need a deterministic flush.

Environment variables

Variable	Purpose	Default
PONTEM_EMIT_DIR	Log JSONL output directory	platform-managed directory
PONTEM_EMIT_TARGET	"file", "stdout", or "journald" (logs only)	"file"
PONTEM_OTLP_ENDPOINT	Collector endpoint for metrics	local collector
PONTEM_CONFIG_DIR	Directory containing per-namespace <namespace>.json files	platform-managed directory

Wire format

Log record:

{
  "timestamp": "2025-01-15T10:30:00.123456Z",
  "severityNumber": 9,
  "severityText": "INFO",
  "body": "model loaded",
  "attributes": {"model": "scoring_v3"},
  "resource": {"service.name": "my-service"}
}

Metrics post as OTLP/JSON ExportMetricsServiceRequest bodies — resourceMetrics → scopeMetrics → metrics[] with cumulative counters/histograms and point-in-time gauges. Every datapoint carries a service attribute; the collector adds device_id on the way through. The full wire shape and proto3 canonical JSON conventions live in SCHEMA.md.