Metadata-Version: 2.4
Name: nedo-worker-common
Version: 0.1.0
Summary: Shared library for Nedo Vision worker, annotator and related services (system metrics, gRPC, messaging, hardware identity).
Author-email: Willy Achmat Fauzi <willy.achmat@gmail.com>
Maintainer-email: Willy Achmat Fauzi <willy.achmat@gmail.com>
License-Expression: MIT
Project-URL: Homepage, https://gitlab.com/sindika/research/nedo-vision/nedo-worker-common
Project-URL: Repository, https://gitlab.com/sindika/research/nedo-vision/nedo-worker-common
Project-URL: Bug Reports, https://gitlab.com/sindika/research/nedo-vision/nedo-worker-common/-/issues
Keywords: nedo-vision,system-metrics,monitoring,jetson,gpu,cross-platform
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Operating System :: POSIX :: Linux
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: MacOS
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.8
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 :: Python Modules
Classifier: Topic :: System :: Monitoring
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: psutil>=5.9.0
Provides-Extra: gpu
Requires-Dist: nvidia-ml-py>=12.535.0; extra == "gpu"
Provides-Extra: windows
Requires-Dist: WMI>=1.5.1; extra == "windows"
Provides-Extra: jetson
Requires-Dist: jetson-stats>=4.2.0; extra == "jetson"
Provides-Extra: grpc
Requires-Dist: grpcio>=1.50.0; extra == "grpc"
Requires-Dist: protobuf>=4.21.0; extra == "grpc"
Provides-Extra: messaging
Requires-Dist: pika>=1.3.0; extra == "messaging"
Provides-Extra: messaging-async
Requires-Dist: aio-pika>=9.0.0; extra == "messaging-async"
Provides-Extra: mqtt
Requires-Dist: paho-mqtt>=1.6.0; extra == "mqtt"
Provides-Extra: db
Requires-Dist: SQLAlchemy>=2.0.0; extra == "db"
Provides-Extra: all
Requires-Dist: nvidia-ml-py>=12.535.0; extra == "all"
Requires-Dist: WMI>=1.5.1; sys_platform == "win32" and extra == "all"
Requires-Dist: grpcio>=1.50.0; extra == "all"
Requires-Dist: protobuf>=4.21.0; extra == "all"
Requires-Dist: pika>=1.3.0; extra == "all"
Requires-Dist: aio-pika>=9.0.0; extra == "all"
Requires-Dist: paho-mqtt>=1.6.0; extra == "all"
Requires-Dist: SQLAlchemy>=2.0.0; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: black>=22.0.0; extra == "dev"
Requires-Dist: isort>=5.10.0; extra == "dev"
Requires-Dist: mypy>=0.950; extra == "dev"
Requires-Dist: flake8>=4.0.0; extra == "dev"

# nedo-worker-common

Shared Python library for Nedo Vision services (worker, annotator, manager).
Owns code that previously lived twice — system metrics, hardware identity,
gRPC base, messaging, configuration, database plumbing, graceful shutdown,
health checks, network helpers, async variants, and time helpers.

Designed so that adopting a single module is a one-line import change. No
service has to take the whole library at once.

---

## Contents

- [Install](#install)
- [Optional dependencies](#optional-dependencies)
- [Package layout](#package-layout)
- [Module guide](#module-guide)
  - [`system` — cross-platform metrics](#system--cross-platform-metrics)
  - [`logging` — one-call logger setup](#logging--one-call-logger-setup)
  - [`identity` — hardware UUID & service identity](#identity--hardware-uuid--service-identity)
  - [`grpc` — channel & client base](#grpc--channel--client-base)
  - [`messaging` — RabbitMQ pub/sub](#messaging--rabbitmq-pubsub)
  - [`config` — configuration sources](#config--configuration-sources)
  - [`db` — SQLAlchemy plumbing](#db--sqlalchemy-plumbing)
  - [`shutdown` — coordinated graceful shutdown](#shutdown--coordinated-graceful-shutdown)
  - [`health` — health checks & registry](#health--health-checks--registry)
  - [`network` — IP & reachability helpers](#network--ip--reachability-helpers)
  - [`aio` — async variants](#aio--async-variants)
  - [`clock` — time & duration helpers](#clock--time--duration-helpers)
  - [`diagnostics` — setup checks & `nedo-doctor` CLI](#diagnostics--setup-checks--nedo-doctor-cli)
- [Putting it together: a typical service](#putting-it-together-a-typical-service)
- [Migrating from the per-service implementations](#migrating-from-the-per-service-implementations)
- [Extending the library](#extending-the-library)
- [Testing](#testing)

---

## Install

```bash
# Editable install for local development (recommended during the transition):
pip install -e ../nedo-worker-common

# Pick extras based on what your service actually uses:
pip install -e "../nedo-worker-common[gpu,windows]"           # system metrics, full
pip install -e "../nedo-worker-common[grpc,messaging,db]"     # service primitives
pip install -e "../nedo-worker-common[all,dev]"               # everything + dev tools
```

For production wheels, publish to the GitLab Package Registry alongside the
other Nedo Vision libraries and pin a version:

```toml
# in nedo-vision-worker-service/pyproject.toml
dependencies = [
    "nedo-worker-common>=0.2,<0.3",
]
```

## Optional dependencies

| Extra              | Pulls in              | Needed for                                        |
| ------------------ | --------------------- | ------------------------------------------------- |
| `gpu`              | `nvidia-ml-py`        | NVENC/NVDEC/power/throttle reporting              |
| `windows`          | `WMI`                 | Windows CPU-temperature sources (LHM/OHM/MSAcpi)  |
| `jetson`           | `jetson-stats`        | Optional richer Jetson telemetry                  |
| `grpc`             | `grpcio`, `protobuf`  | `nedo_worker_common.grpc` and async variant       |
| `messaging`        | `pika`                | `nedo_worker_common.messaging` (sync)             |
| `messaging-async`  | `aio-pika`            | `nedo_worker_common.aio.rabbitmq`                 |
| `db`               | `SQLAlchemy`          | `nedo_worker_common.db`                           |
| `all`              | union of the above    | "I want everything"                               |
| `dev`              | `pytest`, etc.        | running the test suite & linters                  |

The library imports cleanly without any extras — every optional dependency is
imported lazily inside the module that uses it, and that module raises a
descriptive `ImportError` pointing at the right extra if the dep is missing.

## Package layout

```
nedo_worker_common/
├── system/        CPU / RAM / GPU / disk metrics, TTL temperature cache
├── logging/       PrettyFormatter, JsonFormatter, setup_logging
├── identity/      HardwareID provider pattern, ServiceIdentity dataclass
├── grpc/          GrpcConnection (per-target), GrpcClientBase, retry, interceptors
├── messaging/     RabbitMQPublisher, RabbitMQConsumer, BrokerConfig
├── config/        ConfigurationManager + Env / Dict / Chained
├── db/            SessionFactory, declarative_base, TimestampMixin, SoftDeleteMixin
├── shutdown/      ShutdownManager with signal handlers & LIFO cleanup
├── health/        HealthCheck + HealthRegistry + built-in checks
├── network/       local/public IP, port reachability, HTTP/TCP latency
├── aio/           Async gRPC & async RabbitMQ (aio-pika)
├── clock/         utc_now, RFC 3339 round-trip, parse/format_duration
└── diagnostics/   DiagnosticCheck + DiagnosticsRunner + `nedo-doctor` CLI
```

Each module exposes its public API through its `__init__.py`. Internal helpers
are prefixed with `_` and not re-exported.

---

## Module guide

Every section follows the same shape: what it's for, the smallest useful
example, the public API, and the design tradeoffs worth knowing.

### `system` — cross-platform metrics

Sampling CPU / RAM / GPU / disk on Windows, Linux, macOS, and Jetson. Drop-in
for the old per-service `SystemMonitor` classes.

```python
from nedo_worker_common.system import (
    SystemMonitor, SystemMetricsReporter, get_program_disk,
)

monitor = SystemMonitor(
    disk_paths=[get_program_disk().mountpoint, "/data"],
    temperature_cache_ttl=1.0,     # CPU-temp reads cached this many seconds (0 = disabled)
)
sample = monitor.get_system_usage()
print(sample["cpu"]["temperature_celsius"], sample["gpu"][0]["power_watts"])

reporter = SystemMetricsReporter(interval_seconds=5)
reporter.start()
# ...
print(reporter.latest())   # ~0ms cached snapshot
reporter.stop()
```

**Output shape** (legacy-compatible dict — same keys the old code emitted, plus
new GPU and disk fields):

```jsonc
{
  "cpu":  { "usage_percent": 18.7, "temperature_celsius": 27.85 },
  "ram":  { "total": ..., "used": ..., "free": ..., "percent": 82.5 },
  "gpu":  [{
    "gpu_index": 0,
    "name": "NVIDIA GeForce RTX 5060 Ti",
    "gpu_usage_percent": 4.0,
    "memory_usage_percent": 52.8,
    "temperature_celsius": 39.0,
    "total_memory": 17103323136, "used_memory": ..., "free_memory": ...,
    "encoder_usage_percent": 0.0,   // NVENC
    "decoder_usage_percent": 0.0,   // NVDEC
    "power_watts": 11.98,
    "power_limit_watts": 180.0,
    "throttle_reasons": []          // e.g. ["sw_power_cap", "hw_thermal_slowdown"]
  }],
  "disk": [{ "mountpoint": "D:\\", "device": "D:\\", "fstype": "NTFS",
             "total": ..., "used": ..., "free": ..., "percent": 56.3 }]
}
```

**Disk-path helpers** —

```python
from nedo_worker_common.system import get_disk_for_path, get_program_disk

prog = get_program_disk()           # disk the running script lives on
target = get_disk_for_path("/var/lib/nedo-vision")
```

**CPU-temperature cascade** — the cascade falls through to the next source if
one returns `None`. Each is tested independently.

| Platform | Source order                                                                            | Needs admin? | Works in container?                    |
| -------- | --------------------------------------------------------------------------------------- | ------------ | -------------------------------------- |
| Windows  | LibreHardwareMonitor → OpenHardwareMonitor → `Win32_PerfFormattedData_*` → MSAcpi       | last only    | Win32_Perf works in Windows containers |
| Linux    | `psutil.sensors_temperatures()` → `/sys/class/thermal/thermal_zone*` → `/sys/class/hwmon/*` | no        | sysfs paths work in Docker by default  |
| Jetson   | `thermal_zone1` / `thermal_zone0` + generic Linux scan                                  | no           | yes                                    |
| macOS    | `osx-cpu-temp` CLI (`brew install osx-cpu-temp`)                                        | no           | n/a                                    |

**Performance** — `get_system_usage()` is ~8 ms steady-state with the default
1 s temperature cache enabled, vs. ~32 ms with `temperature_cache_ttl=0`. For
hot paths, prefer `SystemMetricsReporter` and read `.latest()`.

### `logging` — one-call logger setup

Unifies formatting across services. Pretty colored output on a TTY, JSON lines
under systemd/Docker/k8s, and a `LOG_LEVEL` env override at runtime.

```python
from nedo_worker_common.logging import setup_logging
import logging

setup_logging("worker", level="INFO")     # auto-detect TTY → pretty/JSON
log = logging.getLogger("nedo.worker.boot")
log.info("starting up", extra={"hardware_id": "abc-...", "role": "edge"})
```

**Pretty output** (dev terminal):

```
14:23:01.482 INFO  worker.nedo.worker.boot - starting up
```

**JSON output** (Docker/k8s — `extra=` fields are merged into the record):

```json
{"ts":"2026-05-12T14:23:01.482Z","level":"INFO","logger":"nedo.worker.boot","service":"worker","message":"starting up","hardware_id":"abc-...","role":"edge"}
```

**Idempotent** — calling `setup_logging` multiple times replaces the previous
handler instead of stacking. Safe to call from tests, notebooks, and dynamic
reloads.

**Toggle level at runtime** — export `LOG_LEVEL=DEBUG` before starting the
service to override whatever `level=` was passed. No code change, no redeploy.

### `identity` — hardware UUID & service identity

Two things: a stable per-host UUID, and the dataclass services use when they
register with the manager.

```python
from nedo_worker_common.identity import get_hardware_id, build_service_identity

hw_id = get_hardware_id()       # cached for the process lifetime
print(hw_id)                    # e.g. "7eb5228f-312d-fb99-f9e4-50ebf627479a"

identity = build_service_identity(
    service_name="nedo-vision-worker",
    service_version="2.3.1",
    role="edge",
    capabilities=["ppe", "human"],
    metadata={"region": "id-jkt"},
)
manager_client.register(identity.to_dict())
```

**Sources per platform**, in priority order:

| Platform | Source                                                              |
| -------- | ------------------------------------------------------------------- |
| Windows  | `wmic csproduct UUID` → PowerShell `Win32_ComputerSystemProduct` → registry `MachineGuid` |
| Linux    | `/etc/machine-id` → `/var/lib/dbus/machine-id` → DMI `product_uuid` |
| macOS    | `ioreg` `IOPlatformUUID`                                            |
| Jetson   | `/sys/devices/soc0/serial_number` → `/proc/device-tree/serial-number` → generic Linux |

When **every** source fails (locked-down container, no DMI), a MAC-derived
UUIDv5 fallback fires with a logged warning. That ID is less stable across
NIC swaps but better than crashing the registration heartbeat.

**`coerce_to_uuid(raw)`** is exposed for callers that already have a raw ID
from somewhere else and want the same canonical formatting.

### `grpc` — channel & client base

Per-target channel manager (no singleton) plus a client base with structured
error mapping, single-flight reconnect, and an auth-failure callback hook.

```python
from nedo_worker_common.grpc import GrpcClientBase, get_or_create_connection
from nedo_vision_worker.protos.VisionWorkerService_pb2_grpc import VisionWorkerServiceStub

conn = get_or_create_connection("manager.internal", 50051)
client = GrpcClientBase(conn, stub_class=VisionWorkerServiceStub)
result = client.handle_rpc(client.stub.Heartbeat, request=hb_request)
```

**What `handle_rpc` does for you:**

- Maps every `grpc.StatusCode` to a log level + label (no spam from harmless
  `CANCELLED` warnings, no swallowed `INTERNAL` errors).
- On `UNAVAILABLE` — triggers a single-flight reconnect on the underlying
  `GrpcConnection`, re-binds the stub, retries the call once.
- On `UNAUTHENTICATED` / `PERMISSION_DENIED` — invokes the auth-failure
  callback you registered with the constructor (typically: refresh the token,
  then re-issue the RPC at the application layer).
- On any unexpected `Exception` — if the message contains "closed channel",
  re-bind the stub once before returning `None`.

**Channels are insecure by default**; `port=443` opts into the default SSL
credentials. Pass `credentials=` to use mTLS or anything else.

**Retry / backoff helpers** ship separately so anything else (e.g. the
RabbitMQ reconnect loop) can reuse them:

```python
from nedo_worker_common.grpc import exponential_backoff, sleep_with_stop

for delay in exponential_backoff(base_seconds=1, max_seconds=30, jitter=0.1):
    if try_something():
        break
    if not sleep_with_stop(delay, stop_event=shutdown.stop_event):
        break  # service is shutting down, abandon the retry loop
```

**`get_or_create_connection`** is a deliberate replacement for the old
process-wide singleton: each `(host, port)` gets one shared channel without
making host/port immutable for the lifetime of the process. Use it when you
want shared connection state; instantiate `GrpcConnection` directly when you
want isolation.

**Optional auth interceptor** (`nedo_worker_common.grpc.interceptors`):

```python
from nedo_worker_common.grpc.interceptors import make_auth_interceptor
import grpc

interceptor = make_auth_interceptor(token_provider=lambda: store.current_token())
channel = grpc.intercept_channel(conn.get_channel(), interceptor)
stub = MyServiceStub(channel)
```

`token_provider` is invoked on every call so token rotation just works.

### `messaging` — RabbitMQ pub/sub

Sync publisher + consumer over `pika`. The publisher runs its I/O loop on a
dedicated thread and uses `add_callback_threadsafe` so callers from any thread
can publish without serialising on a single mutex.

```python
from nedo_worker_common.messaging import (
    BrokerConfig, ExchangeConfig, QueueConfig,
    RabbitMQPublisher, RabbitMQConsumer,
)

broker = BrokerConfig(host="rabbit.local", username="nedo", password="secret")

# Publisher
with RabbitMQPublisher(broker) as pub:
    pub.declare_exchange(ExchangeConfig(name="events"))
    pub.publish_message({"type": "frame.ready", "id": 42}, "events", "frame.ready")

# Consumer (background thread; handler is a normal callable)
def handle(msg):
    print("got:", msg)
    # Return normally to ack; raise to nack-and-requeue.

consumer = RabbitMQConsumer(
    config=broker,
    queue=QueueConfig(name="work.frames"),
    exchange=ExchangeConfig(name="events"),
    routing_keys=["frame.ready"],
    handler=handle,
    prefetch_count=4,
)
consumer.start()
# ...
consumer.stop()
```

The consumer reconnects automatically on transient broker drops, with a 5 s
back-off between attempts. If the handler raises a non-`Exception`-class
error (e.g. `KeyboardInterrupt`), the message is *not* requeued — that's
treated as a shutdown signal.

### `config` — configuration sources

A small ABC plus three implementations. The Chained variant lets you say "env
wins, file is the default" without writing the merging code yourself.

```python
from nedo_worker_common.config import (
    EnvConfigurationManager, DictConfigurationManager, ChainedConfigurationManager,
)

cfg = ChainedConfigurationManager([
    EnvConfigurationManager(prefix="NEDO_"),   # highest priority
    DictConfigurationManager(yaml_loaded),     # defaults from file
])

host = cfg.get_config("MANAGER_HOST", default="localhost")
port = cfg.get_int("MANAGER_PORT", default=50051)
flag = cfg.get_bool("ENABLE_TRACE", default=False)
tags = cfg.get_list("FEATURE_FLAGS")           # parses "a,b,c"
secret = cfg.require("API_SECRET")             # raises KeyError if missing
```

**`MISSING` sentinel** — `get_config(key)` without a default raises
`KeyError`. Pass `default=...` (any value, including `None`) to make it
return that instead. The `require()` shortcut is just `get_config(key)` with
intent.

**Typed accessors** (`get_int`, `get_float`, `get_bool`, `get_list`) all
respect the underlying source's storage type — if the value is already a
list/bool/etc the conversion is a no-op.

### `db` — SQLAlchemy plumbing

Engine + session factory tuned for the worker/annotator usage pattern (long
idle stretches between RPCs, occasional bursts). Entities and migrations
stay in each service.

```python
from sqlalchemy import Column, Integer, String

from nedo_worker_common.db import (
    SessionFactory, TimestampMixin, SoftDeleteMixin, declarative_base,
)

Base = declarative_base()

class Event(Base, TimestampMixin):
    __tablename__ = "events"
    id = Column(Integer, primary_key=True)
    payload = Column(String)

factory = SessionFactory("postgresql+psycopg2://nedo:secret@db/nedo")
Base.metadata.create_all(factory.engine)

with factory.session() as s:
    s.add(Event(payload="hello"))
# commits on success; rolls back + re-raises on exception; closes either way
```

**Tuning defaults** — `pool_size=5, max_overflow=10, pool_pre_ping=True,
pool_recycle=1800`. `pool_pre_ping` is on because both services keep
connections idle for minutes — we'd rather pay one extra ping than fail
mid-transaction on a server-side TCP RST.

**Mixins** — `TimestampMixin` adds `created_at` + `updated_at` populated by
the DB; `SoftDeleteMixin` adds an `is_deleted` boolean.

### `shutdown` — coordinated graceful shutdown

LIFO cleanup chain, signal-driven trigger, total-deadline enforcement,
per-callback timeout abandonment. Required on every service with more than
one background thread.

```python
from nedo_worker_common.shutdown import ShutdownManager

sm = ShutdownManager(deadline_seconds=30)
sm.install_signal_handlers()                  # SIGTERM + SIGINT

# Register in the order things are *started*. Cleanup runs in reverse.
sm.register(reporter.stop,   name="metrics",         timeout=2)
sm.register(consumer.stop,   name="rabbit-consumer", timeout=10)
sm.register(publisher.stop,  name="rabbit-publisher",timeout=5)
sm.register(grpc_conn.close, name="grpc-channel",    timeout=2)

# Block until SIGTERM / SIGINT. While we sleep, background threads can poll
# sm.stop_event so they wake immediately rather than ticking past their
# timeout.
sm.wait_for_shutdown()
```

**LIFO?** — yes. The pattern is "set up bottom-up, tear down top-down":
network channels open before consumers that use them, and close after the
consumers stop publishing on them.

**Timeouts** — every callback gets a budget (per-callback `timeout=` or
"whatever's left of the deadline"). When a callback exceeds its budget,
the supervising thread is *abandoned* (it leaks until process exit) and the
chain continues. That's intentional: hung sockets shouldn't take the rest
of the shutdown down with them.

**SIGINT (Ctrl-C)** exits with code 130 (the conventional value). SIGTERM
exits with 0. Process supervisors can distinguish "asked to stop" from
"crashed" without trickery.

**Default singleton** — `get_default()` returns a lazily-constructed shared
manager. Use it for services that only have one; construct
`ShutdownManager` directly for multi-tenant apps and tests.

### `health` — health checks & registry

A registry pattern: register named checks, run them all in parallel, get a
single aggregated status.

```python
from nedo_worker_common.health import (
    HealthRegistry, DiskUsageCheck, MemoryUsageCheck,
    BrokerConnectionCheck, GrpcConnectionCheck, CallableCheck,
)

registry = HealthRegistry()
registry.register(DiskUsageCheck(warn_percent=80, critical_percent=95, paths=["/data"]))
registry.register(MemoryUsageCheck(threshold_percent=95))
registry.register(GrpcConnectionCheck(manager_conn))
registry.register(BrokerConnectionCheck(rabbit_pub))
registry.register(CallableCheck("model_loaded", lambda: model.is_loaded()))

report = registry.aggregate()
# {
#   "status": "healthy" | "degraded" | "unhealthy" | "unknown",
#   "checks": [{ "name": "disk", "status": "healthy", "message": "...",
#                "metadata": {...}, "elapsed_ms": 1.2 }, ...]
# }
```

**Statuses** are ordered `HEALTHY < UNKNOWN < DEGRADED < UNHEALTHY`. The
aggregator returns the worst status across all checks. `UNKNOWN` is treated
as worse than `HEALTHY` because "I can't tell" is closer to a problem than
"definitely fine."

**Parallel by default** — checks run in a small thread pool (`max_workers=8`)
so a slow broker probe doesn't gate the whole report. Set `parallel=False`
when you want deterministic ordering or are running on a hostile
single-threaded sandbox.

**Built-ins** — `DiskUsageCheck`, `MemoryUsageCheck`, `GpuAvailableCheck`,
`GrpcConnectionCheck`, `BrokerConnectionCheck`, `CallableCheck` (escape hatch
for one-off checks: pass a callable returning `bool`, a string, or a
`HealthResult`).

**Exposing it** — wire `registry.aggregate()` to whatever your manager or
k8s probe expects. A 3-line HTTP handler with `json.dumps` works; a gRPC
`HealthCheckResponse` is equivalent.

### `network` — IP & reachability helpers

Stdlib-only (no `requests` dep), so it's safe to import early in startup
before the rest of your dependency graph is loaded.

```python
from nedo_worker_common.network import (
    get_local_ip, get_public_ip, get_hostname, get_fqdn,
    is_port_reachable, measure_tcp_latency, measure_http_latency,
)

local = get_local_ip()                        # "192.168.1.42" or "127.0.0.1"
public = get_public_ip(timeout=2)              # str or None — tries multiple providers
ok = is_port_reachable("manager.internal", 50051, timeout=2)
ms = measure_tcp_latency("manager.internal", 50051)
ms_http = measure_http_latency("https://manager.internal/healthz")
```

`get_local_ip` uses the "open a UDP socket and ask for its source address"
trick — no packets are sent, but the kernel reveals which interface it would
route through. Falls back to `127.0.0.1` for air-gapped hosts.

`get_public_ip` rotates through `api.ipify.org`, `ifconfig.me`, and
`ipinfo.io` so a single provider going down doesn't take you down. Pass
`services=...` to use a private STUN endpoint instead.

### `aio` — async variants

For services migrating to asyncio. The async modules mirror the sync APIs as
closely as possible so the migration is mechanical.

```python
import asyncio
from nedo_worker_common.aio import (
    AsyncGrpcConnection, AsyncGrpcClientBase,
    AsyncRabbitMQPublisher, AsyncRabbitMQConsumer,
)
from nedo_worker_common.messaging import BrokerConfig, ExchangeConfig, QueueConfig

async def main():
    # gRPC
    async with AsyncGrpcConnection("manager.internal", 50051) as conn:
        client = AsyncGrpcClientBase(conn, stub_class=MyServiceStub)
        result = await client.handle_rpc(client.stub.Heartbeat, request=hb)

    # RabbitMQ
    broker = BrokerConfig(host="rabbit.local", username="nedo", password="secret")
    async with AsyncRabbitMQPublisher(broker) as pub:
        await pub.publish_message({"id": 42}, "events", "frame.ready")

    # Consumer
    async def handle(msg):
        print("got:", msg)

    consumer = AsyncRabbitMQConsumer(
        config=broker,
        queue=QueueConfig(name="work.frames"),
        exchange=ExchangeConfig(name="events"),
        routing_keys=["frame.ready"],
        handler=handle,
    )
    await consumer.start()
    await asyncio.sleep(60)
    await consumer.stop()

asyncio.run(main())
```

The async gRPC client reuses the same backoff helpers and error-mapping
strategy as the sync variant; only the underlying coroutine plumbing
differs. Reconnect is single-flight via an `asyncio.Lock`.

The async RabbitMQ pieces are built on `aio-pika` (separate optional dep:
`pip install nedo-worker-common[messaging-async]`) and use
`connect_robust()` so the underlying connection auto-recovers across broker
restarts.

### `clock` — time & duration helpers

Force everyone through timezone-aware UTC and a single timestamp format. The
class of bug this prevents: "the manager sends `2026-05-12T09:00:00` and the
worker interprets it as local time."

```python
from nedo_worker_common.clock import (
    utc_now, utc_now_ms, monotonic_now,
    to_rfc3339, parse_rfc3339,
    parse_duration, format_duration,
)

now = utc_now()                          # timezone-aware UTC datetime
text = to_rfc3339(now)                   # "2026-05-12T09:00:00.123456Z"
parsed = parse_rfc3339("2026-05-12T16:00:00+07:00")  # → UTC

secs = parse_duration("1h 30m")          # 5400.0  (also accepts: "500ms", "1.5h", "2d", "1w")
secs = parse_duration(30)                # 30.0    (numeric inputs pass through)
human = format_duration(5400)            # "1h 30m"
```

`utc_now()` returns a *timezone-aware* datetime — use it everywhere instead
of the naive `datetime.utcnow()` (which Python 3.12 has deprecated).

`to_rfc3339` quietly assumes naive datetimes are UTC and annotates them. The
pattern of mixing naive utcnow() into a serialiser is so common we'd rather
fix it transparently than emit a string the peer rejects.

`parse_duration` is convenient for configs that may carry either
`timeout: 30` or `timeout: "30s"` — both work.

### `diagnostics` — setup checks & `nedo-doctor` CLI

A framework for setup-time validation: "is this host configured to run me?".
Companion to `health` but with a different audience — operators and support,
not k8s probes.

```python
from nedo_worker_common.diagnostics import DiagnosticsRunner
from nedo_worker_common.diagnostics.checks import (
    PythonEnvironmentCheck, SystemResourcesCheck, StorageCheck,
    HardwareIdentityCheck, GpuCapabilityCheck,
    FFmpegCheck, OpenCVCheck, NetworkReachabilityCheck,
)

runner = DiagnosticsRunner(service_name="nedo-vision-worker", service_version="2.3.1")
runner.register_all([
    PythonEnvironmentCheck(min_version=(3, 10)),
    SystemResourcesCheck(min_ram_gb=8),
    StorageCheck(write_test_dirs=["data", "models", "logs"]),
    HardwareIdentityCheck(),
    GpuCapabilityCheck(required=True, min_vram_gb=4),
    FFmpegCheck(require_hwaccel=False),
    OpenCVCheck(required=True),
    NetworkReachabilityCheck(host="manager.internal", port=50051),
])

report = runner.run()
print(report.to_text())          # colored terminal output
print(report.to_json(indent=2))  # machine-readable
sys.exit(report.exit_code())     # 0 = OK, 1 = at least one blocking issue
```

**Built-in checks**:

| Check                          | What it asserts                                      |
| ------------------------------ | ---------------------------------------------------- |
| `PythonEnvironmentCheck`       | Interpreter version + venv presence                  |
| `SystemResourcesCheck`         | RAM total / CPU cores against minima                 |
| `StorageCheck`                 | Disk space + write permission on listed directories  |
| `HardwareIdentityCheck`        | SMBIOS / machine-id source is working (not fallback) |
| `LocalNetworkCheck`            | Local IP is routable, not loopback                   |
| `NetworkReachabilityCheck`     | TCP probe to a specific host:port (manager, broker)  |
| `GpuCapabilityCheck`           | NVIDIA GPU enumerable, VRAM above threshold          |
| `FFmpegCheck`                  | `ffmpeg` on PATH, optional hwaccel requirement       |
| `OpenCVCheck`                  | `cv2` importable, optimisations available            |

**Result anatomy** — every check returns a `DiagnosticResult`:

```python
@dataclass
class DiagnosticResult:
    name: str
    status: DiagnosticStatus     # EXCELLENT / GOOD / INFO / WARNING / CRITICAL
    message: str                 # one-line headline
    details: list[str]           # context for the operator
    recommendations: list[str]   # copy-pasteable fix steps
    is_blocking: bool            # if True, exit_code() returns 1
    performance_impact: PerformanceImpact
    elapsed_ms: float
```

**Custom checks** — subclass `DiagnosticCheck`:

```python
from nedo_worker_common.diagnostics import (
    DiagnosticCheck, DiagnosticResult, DiagnosticStatus, PerformanceImpact,
)

class ModelFileCheck(DiagnosticCheck):
    name = "model_file"

    def __init__(self, path: str):
        self.path = path

    def run(self) -> DiagnosticResult:
        from pathlib import Path
        if not Path(self.path).is_file():
            return DiagnosticResult(
                name=self.name,
                status=DiagnosticStatus.CRITICAL,
                message=f"model not found at {self.path}",
                recommendations=[f"Download the model to {self.path}"],
                is_blocking=True,
                performance_impact=PerformanceImpact.HIGH,
            )
        return DiagnosticResult(name=self.name, status=DiagnosticStatus.GOOD,
                                message=f"found {self.path}")
```

**CLI** — installed as `nedo-doctor` and runnable via the module:

```bash
nedo-doctor                                # text report, exit 0/1
nedo-doctor --json                          # machine-readable
nedo-doctor --checks python_environment,storage   # subset
nedo-doctor --no-color                      # CI-friendly output

python -m nedo_worker_common.diagnostics --json   # equivalent module form
```

The default CLI runs only platform-agnostic checks
(`python_environment`, `system_resources`, `storage`, `hardware_identity`,
`local_network`). For service-specific diagnostics, define a small wrapper:

```python
# nedo-vision-worker-service/nedo_vision_worker/doctor.py
from nedo_worker_common.diagnostics import DiagnosticsRunner
from nedo_worker_common.diagnostics.checks import (
    PythonEnvironmentCheck, SystemResourcesCheck, StorageCheck,
    HardwareIdentityCheck, NetworkReachabilityCheck,
    GpuCapabilityCheck, FFmpegCheck, OpenCVCheck,
)

def main():
    runner = DiagnosticsRunner("nedo-vision-worker", "2.3.1")
    runner.register_all([
        PythonEnvironmentCheck(min_version=(3, 10)),
        SystemResourcesCheck(min_ram_gb=8),
        StorageCheck(write_test_dirs=["data", "models", "logs"]),
        HardwareIdentityCheck(),
        GpuCapabilityCheck(required=True),
        FFmpegCheck(require_hwaccel=True),
        OpenCVCheck(required=True),
        NetworkReachabilityCheck("manager.internal", 50051),
    ])
    runner.run_and_exit()
```

That replaces the old ~1000-line `doctor.py` with ~20 lines of composition.

**Why a separate framework from `health`?** The audiences are different:

| | `health/` | `diagnostics/` |
|---|---|---|
| Audience | k8s probes, manager heartbeat | Operators, support engineers |
| Output | binary status + JSON | rich text + JSON |
| Fields | name, status, message, metadata | + recommendations, is_blocking, perf_impact, details |
| Frequency | every few seconds | once at startup / on demand |
| Style | tight, machine-readable | verbose, human-readable |

Both can coexist — the worker has both wired in [main.py](#putting-it-together-a-typical-service).

---

## Putting it together: a typical service

What a worker `main.py` looks like once everything is wired through this lib:

```python
import logging

from nedo_worker_common.logging import setup_logging
from nedo_worker_common.config import (
    EnvConfigurationManager, DictConfigurationManager, ChainedConfigurationManager,
)
from nedo_worker_common.identity import build_service_identity
from nedo_worker_common.system import SystemMonitor, SystemMetricsReporter, get_program_disk
from nedo_worker_common.grpc import get_or_create_connection, GrpcClientBase
from nedo_worker_common.messaging import (
    BrokerConfig, ExchangeConfig, QueueConfig,
    RabbitMQPublisher, RabbitMQConsumer,
)
from nedo_worker_common.shutdown import ShutdownManager
from nedo_worker_common.health import (
    HealthRegistry, DiskUsageCheck, MemoryUsageCheck,
    BrokerConnectionCheck, GrpcConnectionCheck,
)
from nedo_worker_common.network import get_local_ip


def main():
    setup_logging("worker", level="INFO")
    log = logging.getLogger("worker")

    # ---- config ----
    cfg = ChainedConfigurationManager([
        EnvConfigurationManager(prefix="NEDO_"),
        DictConfigurationManager(load_yaml("/etc/nedo/worker.yaml")),
    ])

    # ---- identity & manager ----
    identity = build_service_identity(
        service_name="nedo-vision-worker",
        service_version=__version__,
        role=cfg.get_config("ROLE", default="edge"),
        capabilities=cfg.get_list("CAPABILITIES"),
        metadata={"local_ip": get_local_ip()},
    )
    log.info("identity: %s", identity.to_dict())

    grpc_conn = get_or_create_connection(
        host=cfg.get_config("MANAGER_HOST"),
        port=cfg.get_int("MANAGER_PORT", default=50051),
    )
    grpc_conn.connect()
    manager = GrpcClientBase(grpc_conn, stub_class=VisionWorkerServiceStub)

    # ---- messaging ----
    broker = BrokerConfig(
        host=cfg.get_config("RABBIT_HOST"),
        username=cfg.get_config("RABBIT_USER"),
        password=cfg.get_config("RABBIT_PASS"),
    )
    publisher = RabbitMQPublisher(broker)
    publisher.connect()

    consumer = RabbitMQConsumer(
        config=broker,
        queue=QueueConfig(name="work.frames"),
        exchange=ExchangeConfig(name="events"),
        routing_keys=["frame.ready"],
        handler=process_frame,
    )
    consumer.start()

    # ---- telemetry ----
    reporter = SystemMetricsReporter(interval_seconds=5)
    reporter.start()

    # ---- health ----
    health = HealthRegistry()
    health.register(DiskUsageCheck(paths=[get_program_disk().mountpoint, "/data"]))
    health.register(MemoryUsageCheck())
    health.register(GrpcConnectionCheck(grpc_conn))
    health.register(BrokerConnectionCheck(publisher))
    expose_health_endpoint(health)   # service-specific HTTP/gRPC handler

    # ---- shutdown ----
    sm = ShutdownManager(deadline_seconds=30)
    sm.install_signal_handlers()
    sm.register(reporter.stop,    name="metrics",     timeout=2)
    sm.register(consumer.stop,    name="consumer",    timeout=10)
    sm.register(publisher.stop,   name="publisher",   timeout=5)
    sm.register(grpc_conn.close,  name="grpc",        timeout=2)

    log.info("worker ready")
    sm.wait_for_shutdown()
    log.info("clean exit")


if __name__ == "__main__":
    main()
```

That's the whole skeleton. Service-specific behaviour (`process_frame`,
`expose_health_endpoint`, the actual AI work) plugs in via the registered
callbacks and consumers.

---

## Migrating from the per-service implementations

Drop-in replacements for the duplicated files in the worker and annotator:

| Old import                                                       | New import                                          |
| ---------------------------------------------------------------- | --------------------------------------------------- |
| `nedo_vision_worker.util.HardwareID.HardwareID.get_unique_id()`  | `nedo_worker_common.identity.get_hardware_id()`     |
| `nedo_vision_worker.util.SystemMonitor.SystemMonitor`            | `nedo_worker_common.system.SystemMonitor`           |
| `nedo_vision_worker.util.Networking.Networking.get_local_ip()`   | `nedo_worker_common.network.get_local_ip()`         |
| `nedo_vision_worker.services.GrpcClientBase.GrpcClientBase`      | `nedo_worker_common.grpc.GrpcClientBase`            |
| `nedo_vision_worker.services.GrpcConnection.GrpcConnection`      | `nedo_worker_common.grpc.GrpcConnection`            |
| `nedo_vision_worker.config.ConfigurationManagerInterface`        | `nedo_worker_common.config.ConfigurationManager`    |
| `nedo_vision_annotator.services.RabbitMQPublisher`               | `nedo_worker_common.messaging.RabbitMQPublisher`    |
| `nedo_vision_annotator.services.RabbitMQConsumer`                | `nedo_worker_common.messaging.RabbitMQConsumer`     |
| `nedo_vision_annotator.util.SystemMetricsReporter`               | `nedo_worker_common.system.SystemMetricsReporter`   |

Most signatures are preserved. Notable changes:

- **`GrpcConnection` is per-instance, not a singleton.** If you previously
  relied on the singleton, use `get_or_create_connection(host, port)`.
- **`RabbitMQPublisher` takes a `BrokerConfig` dataclass** instead of four
  positional args. The constructor signature change is the only breakage.
- **`SystemMonitor` returns a dict whose `gpu` entries have new fields**
  (`encoder_usage_percent`, `decoder_usage_percent`, `power_watts`,
  `power_limit_watts`, `throttle_reasons`). Existing readers that pluck out
  specific keys will keep working; readers that iterate over keys will see
  the new ones. There is also a new top-level `disk` array.

---

## Extending the library

The pattern that recurs across `system/`, `identity/`, and (in spirit)
`messaging/`:

1. Define an ABC in `module/base.py` (or `module/providers/base.py`).
2. One file per platform / backend.
3. A small `__init__.py` registry + factory function (`detect_*` →
   `get_*_provider`).
4. Optional deps go in `pyproject.toml` under
   `[project.optional-dependencies]` and are imported lazily inside the
   module that needs them — never at package import time.
5. Tests use `pytest.importorskip("<dep>")` at the top of any file that
   needs an optional dep so the suite stays green on minimal installs.

When adding a new top-level module:

- Put the public surface in `__init__.py`. Anything not in `__all__` is
  internal.
- Add a section here in the README with: the smallest useful example, what
  it's for, and one or two design tradeoffs worth knowing.
- Add tests with descriptive names — the names are documentation when a
  future maintainer skims the suite.

## Testing

```bash
# Run everything (auto-skips tests for optional deps that aren't installed):
pip install -e ".[all,dev]"
pytest tests/

# Run a single module's tests:
pytest tests/test_system_smoke.py -v
pytest tests/test_grpc.py -v
```

The suite currently has **155 tests** across 17 files. Each module's tests
live in `tests/test_<module>.py` and only import the module they're testing
plus stdlib + pytest. Anything that needs an optional dep (`grpc`, `pika`,
`sqlalchemy`, `aio_pika`) gates the whole file with
`pytest.importorskip()`.

Smoke tests of the host environment (real disk usage, real psutil, etc.)
live in `tests/test_system_smoke.py` and `tests/test_disk.py`. Pure unit
tests for logic that doesn't touch the host live everywhere else.
