nedo-worker-common 0.1.0

Shared library for Nedo Vision worker, annotator and related services (system metrics, gRPC, messaging, hardware identity).

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
• Optional dependencies
• Package layout
• Module guide
  • system — cross-platform metrics
  • logging — one-call logger setup
  • identity — hardware UUID & service identity
  • grpc — channel & client base
  • messaging — RabbitMQ pub/sub
  • config — configuration sources
  • db — SQLAlchemy plumbing
  • shutdown — coordinated graceful shutdown
  • health — health checks & registry
  • network — IP & reachability helpers
  • aio — async variants
  • clock — time & duration helpers
  • diagnostics — setup checks & nedo-doctor CLI
• Putting it together: a typical service
• Migrating from the per-service implementations
• Extending the library
• Testing

---

Install

# 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:

# 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.

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):

{
  "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 —

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.

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):

{"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.

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.

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:

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):

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.

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.

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.

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.

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.

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.

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.

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."

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.

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:

@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:

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:

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:

# 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

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

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

# 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.