Geo-MLOps-SDK

Edge SDK and runtime for the Geosoft MLOps Central Platform.

The package ships two things, and most people need exactly one of them:

• geo_mlops_sdk.client — an async HTTP client (httpx) whose methods map 1:1 onto Central's device API. Use it when your own application talks to the platform. See Using the client.
• geo_mlops_sdk.edge — a standalone agent you run as a service: local durable queue, connectivity monitor, throttled uploader, collectors, model cache and runners, a FastAPI surface for the on-site web UI. You configure it rather than import it. See Running the runtime.

An edge PC is assumed to be treated roughly — cables pulled, power cut, offline for days. The runtime accumulates locally under a size/age ceiling, watches for the link to come back, and pushes the backlog without starving the collection that keeps running.

Installation

pip install geo-mlops-sdk               # client only
pip install 'geo-mlops-sdk[edge]'       # + standalone runtime and daemon
pip install 'geo-mlops-sdk[edge,gpu,modbus,yolo]'

Extra	Adds
edge	runtime, FastAPI surface, daemon (geo-mlops-edge)
gpu	NVIDIA utilization in the heartbeat
modbus	Modbus TCP collector
yolo	local execution of YOLO models
rfdetr	local execution of RF-DETR models
all	everything above

PyPI also holds 0.0.2, a placeholder published before this SDK was rewritten. 0.1.0 was the first release of the rewritten package; anything below it predates the current API and shares nothing with it but the name.

---

Using the client

CentralClient is the whole surface. It needs httpx and pydantic and nothing else — no runtime, no queue, no SQLite, no edge extra.

from geo_mlops_sdk.client import CentralClient
from geo_mlops_sdk.contracts.edge import HeartbeatBody

async with CentralClient("https://central.example", "device-token") as client:
    await client.register(location="yard-1")
    await client.send_heartbeat(HeartbeatBody(cpu=12.5, mem=55.0, disk=61.0))

The methods follow the device API one for one: health, register, send_heartbeat, send_records, send_inference, upload_init / upload_chunk / upload_status / upload_complete, poll_commands / ack_command, fetch_policy, list_models / model_versions / download_model, resolve_container.

Request and response bodies are Pydantic models in geo_mlops_sdk.contracts (edge, records, uploads, inference, commands, models). They ignore fields they do not know, so a Central that grows a field does not break a deployed client.

Errors and retries

Status codes become exceptions rather than return values, so a caller cannot carry on with a body that never arrived:

Raised	When
OfflineError	no answer at all — DNS, refused, reset, timeout
AuthError (401)	the device token is unknown, expired or revoked
ForbiddenError (403)	the token is valid but lacks the scope for this call
NotFoundError (404), ConflictError (409)	—
PayloadTooLargeError (413)	a server or proxy ceiling; retrying cannot help
UnprocessableError (422)	the body did not validate
RateLimitedError (429)	back off; Retry-After is honoured
ServerError (5xx)	Central's problem, and worth retrying

All of them derive from SdkError, and ApiError carries status, code and detail.

Only 429, 5xx and transport failures are retried — never a 4xx, which would just fail again more slowly. RetryPolicy (3 attempts, 0.5 s initial, ×2 up to 30 s, ±25 % jitter) is a plain value object you can replace:

from geo_mlops_sdk.client import CentralClient, RetryPolicy

client = CentralClient(url, token, retry=RetryPolicy(attempts=5, jitter=0.5))

The jitter matters more than it looks: a fleet that lost the link together comes back together, and without it they reconnect in lockstep.

---

Running the runtime

Nothing here is imported. You write a configuration file, point a service unit at it, and talk to the result over HTTP.

Configuration

Precedence is environment > file > defaults. Anything secret belongs in the environment, not the file:

GEO_EDGE_CENTRAL__TOKEN=...      # nested keys use a double underscore
GEO_EDGE_DATA_DIR=/var/lib/geo-mlops-edge

Three annotated samples live in deploy/:

File	Describes
edge.example.yaml	a real site — Central, a PLC, a camera, a gateway, an ingress
edge.standalone.yaml	a laptop, offline — no central: block at all, everything under build/edge/
edge.dev.yaml	the same laptop against the development Central, under build/edge-dev/

Also there: a systemd unit, a Dockerfile and a Modbus register-map example. The top-level docker-compose.yml runs the agent from that Dockerfile, reading /etc/geo-mlops/edge.yaml and edge.env from the host; a push to main of the geo remote deploys it onto the self-hosted runner labelled edge (.github/workflows/deploy-edge.yml).

Running

geo-mlops-edge run --config /etc/geo-mlops/edge.yaml   # service entry point
geo-mlops-edge status                                  # ask a running agent
geo-mlops-edge queue --state failed                    # why nothing is arriving
geo-mlops-edge config                                  # the effective settings
geo-mlops-edge sync                                    # drain the queue now
geo-mlops-edge models --pull NAME                      # fetch a model version

The agent never restarts itself. A restart command from the fleet makes it exit with code 3, and the supervisor starts a clean process.

From the repository

./run starts the agent out of the working tree — no install, no service unit and no Central:

./run                      # same as ./run run
./run status               # from a second terminal
./run queue --limit 5

It uses deploy/edge.standalone.yaml. Everything the agent writes lands under build/edge/ (gitignored) — the queue database, the spool and the model cache — so a run leaves the machine as it found it. Arguments are passed through to geo-mlops-edge, and a --config of your own replaces the sample.

That sample has no central: block, which is the point: the link stays offline, the backlog grows and the local API keeps answering on http://127.0.0.1:8600. Offline accumulation is what an edge is built around and the hardest thing to try out against a real server.

To watch the other half — collect, upload, appear in the fleet screen — point it at the development Central instead:

export GEO_EDGE_CENTRAL__TOKEN=...            # from Central, never in the file
GEO_EDGE_DATA_DIR=build/edge-dev ./run run --config deploy/edge.dev.yaml

GEO_EDGE_DATA_DIR is not optional there. ./run exports it, and the environment beats the file, so without it the dev agent would share build/edge with the offline one — one device id's backlog pushed under another's registration.

Ctrl-C, or a SIGTERM to ./run, stops the agent the way the service unit does rather than killing the wrapper and leaving the agent holding the database and the port. The exit code is the agent's own, so 3 still means it asked to be restarted.

Getting data in

Five collectors ship with the runtime. A site declares them in collectors:; nothing is compiled in.

Type	Extra	Reads	Emits
modbus	modbus	a PLC's register map over Modbus TCP	one record per poll
watchdir	—	files another process drops in a directory	one blob per file
http	—	a JSON endpoint on another server	one record per poll (or per entry)
push	—	nothing; it is pushed to over the local API	one record per PUT
frames	—	frames your application has run a model on	two blobs per frame: the image and its LabelMe pre-labels

Where it lands on Central: records go to the device's records, and a blob goes to Central's collected data -- a space of its own, apart from datasets. A blob is filed straight into a dataset only when it carries a dataset_id: from POST /api/v1/blobs?dataset_id=... on the local API, or from a frames collector configured with one. The other collectors never set it.

collectors:
  - type: modbus
    name: line-1
    options: { host: 10.0.0.5, port: 502, interval_ms: 1000,
               schema: /etc/geo-mlops/plc.yaml }

  - type: watchdir
    name: cam-0
    priority: 20
    options: { path: /data/incoming, pattern: "*.jpg", delete_after: true }

  - type: http
    name: gateway-1
    options: { url: http://10.0.0.7/api/current, interval_ms: 1000 }

  - type: push
    name: robot-1
    priority: 60
    options: { kind: robot }

  - type: frames
    name: cam-0
    priority: 30
    options: { every_n_frames: 30, dataset_id: "", labels: true, label_conf: 0.5 }

A collector that will not build is logged and skipped — one bad address does not stop the others, or keep the agent off the network.

priority is what retention and the uploader both read: eviction takes the lowest first, the uploader ships the highest first. Leaving it at 50 is fine until something genuinely matters more than the rest.

Interpreting the data is not the SDK's job. A Modbus collector reports that register 5 went high; deciding that this means "job started" belongs to your application. The previous SDK baked one customer's process model into that layer and the next site could not use it at all.

http — polling another server

Every poll emits, exactly as the Modbus collector does. Two polls that return the same body are two samples, not a duplicate: "the gateway still read 21.5 at 12:00:01" is data, and dropping it leaves a hole nobody can reconstruct later.

That makes volume predictable rather than surprising: interval × body size. 2 KiB at 1 Hz is ~177 MB/day, so under the default ceiling (50 GiB / 30 days) the age bound is what trips first — which is what you want from a time series. Size starts winning above roughly 20 KB/s.

- type: http
  name: gateway-1
  options:
    url: http://10.0.0.7/api/current
    interval_ms: 1000
    timeout_s: 10
    verify_tls: true
    headers: { Authorization: "Bearer ..." }   # optional
    auth: { username: edge, password: secret } # optional, HTTP Basic
    ts_field: measured_at        # the server's own observation time, if it has one
    max_body_bytes: 1MiB         # 0 disables; a URL aimed at a firmware image
                                 # polled once a second is how memory runs out

The exception is an endpoint that returns a window of the past rather than the state now — the last N alarms, say. There every poll really does re-deliver rows that are already queued, so name the array and the key and each entry is queued once:

options:
  url: http://10.0.0.7/api/alarms
  items_path: data.items    # "." if the body is the array itself
  id_field: id              # required with items_path

The question to ask of an endpoint is simply: is this the state now, or a window of the past? The first is a sample; the second needs a key.

A server that is down, slow, or answering nonsense is normal: the collector records the error, reports disconnected, backs off 1 s → 30 s, and recovers on its own.

push — being pushed to

The other direction. A robot controller, a vision PC or a Go service on the LAN puts records in, and the runtime receives them:

curl -X PUT http://edge:8600/api/v1/collectors/robot-1/records/evt-1 \
     -H 'content-type: application/json' \
     -d '{"payload": {"step": 3}, "ts": "2026-09-16T01:02:03Z"}'

201 means it was stored, 200 with duplicate: true that it already was. The id is yours, so a client that sends, loses the answer and sends again has still queued one record — which is what it meant.

Declaring the ingress in the configuration does two further things. The kind and priority come from the YAML, so a pusher cannot file its data under something nothing is looking for. And it appears in the collector list beside the Modbus poller, which is the only way a pusher that died is visible at all: it sends no error, because it sends nothing.

frames — what the model just looked at

The one collector driven by your code rather than by a device. An application that runs a detector on a camera or a video has, every frame, the one thing Central cannot get on its own: an image from the site, with the model's own guess about it. Declare the collector, then offer it frames:

- type: frames
  name: cam-0
  priority: 30
  options:
    every_n_frames: 30     # at 30 fps, one frame a second
    dataset_id: ""         # set it and each pair is filed into that dataset
    labels: true           # a LabelMe JSON beside every image
    label_conf: 0.5        # detections below this are not pre-labels

from geo_mlops_sdk.edge.collectors.frames import FrameCollector

frames = next(c for c in runtime.collectors if isinstance(c, FrameCollector))

if frames.wants(source, index):                 # cheap; ask before encoding
    ok, jpeg = cv2.imencode(".jpg", frame)
    await frames.collect(source, index, jpeg.tobytes(), output, model=model_ref)

output is the InferenceOutput the model produced for that frame, the same contract the runners return. Every N-th frame is queued as <source stem>_<index>.jpg, and beside it <stem>.json: a LabelMe document whose two-point rectangles are the detections at or above label_conf, each with its confidence kept in description. Central pairs the two by stem, and the YOLO training runtime turns the rectangles into boxes. With dataset_id the pair is filed into that dataset and validated exactly like an upload from the UI; without it, into collected data. The device token needs the data:write scope either way.

Two things are deliberately left to you. The pixels: the collector takes the frame already encoded, so the SDK does not depend on OpenCV, and wants() is a synchronous check a capture loop makes before spending time on encoding. And the thread: collect() is a coroutine for the runtime's loop, so a capture loop on its own thread hands it over with asyncio.run_coroutine_threadsafe.

A file played in a loop sends frame 30 every round, and a restart would start over, so the frames already queued are written to <data_dir>/collectors/<name>.seen and a seekable source is never queued twice; delete that file to collect from the start again. A live stream has no positions worth remembering -- pass seekable=False -- so its frames are named by time and never deduplicated.

The pre-labels are the current model's guesses. Train on them unreviewed and the model learns its own mistakes back; have someone review them first.

Without declaring anything

POST /api/v1/records and POST /api/v1/blobs stay open for a process that has no entry in the configuration. They generate the id themselves, so a retry queues a second copy, and nothing about the sender shows up in the collector list.

Entrance	id	Retry-safe	kind decided by	Visible in the fleet
a collector	generated	n/a	the YAML	yes, as itself
POST /records, POST /blobs	generated	no	the caller	no
PUT /collectors/{name}/records/{id}	yours	yes	the YAML	yes, as itself

Local API

Served on api.port (8600 by default). Unauthenticated unless api.token is set — the edge PC is headless and the UI is reached from a tablet on the same LAN, so demanding a secret by default would mean distributing one to every shop-floor tablet before anyone could see whether the line is collecting. With a token set, send it as X-Edge-Api-Token.

Method	Path	Purpose
GET	/health	liveness — always reachable, never behind the token
GET	/api/v1/status	device, link, backlog, sync, models, collectors, resources
GET	/api/v1/resources	cpu / gpu / mem / disk
GET	/api/v1/queue	what is waiting, filterable by kind and state
DELETE	/api/v1/queue/{id}	drop one item
POST	/api/v1/records, /api/v1/blobs	queue from an undeclared process
PUT	/api/v1/collectors/{name}/records/{id}	queue at a declared ingress
GET	/api/v1/sync	uploader state
POST	/api/v1/sync:run, :retry-failed	drain now, or requeue failures
GET	/api/v1/models	cached models and the active one
POST	/api/v1/models/{name}:pull, :activate	fetch, or switch version
DELETE	/api/v1/models/{name}	remove a cached version
POST	/api/v1/inference	run the active model on an image
GET	/api/v1/settings	the effective configuration
GET	/api/v1/events	SSE: link, sync, queue, model, command

Rotating a device token

A device token is required to talk to Central, and rotating one is a site visit. There is no remote path back:

1. Rotate or revoke the token in Central (POST /edge/devices/{id}/token:rotate).
2. The agent's next call is refused, the link goes to AUTH_FAILED, and it stops — deliberately, rather than hammering Central with rejected requests.
3. On the machine, put the new token in /etc/geo-mlops/edge.env (GEO_EDGE_CENTRAL__TOKEN=...) and restart the service: sudo systemctl restart geo-mlops-edge.

Step 3 cannot be done from Central. The command poller stops along with everything else, so a restart command cannot reach the device either — which is why the fleet screen reports how many days a token has left (token_expires_in_days): tokens last a year, and that is exactly long enough for the expiry to arrive unplanned.

geo-mlops-edge status says the same thing in words, at the top of its output, so an engineer already standing at the machine does not have to infer it from a state name.

The operator's Offline switch

Edge Fleet has an Offline switch per device (POST /edge/devices/{id}/offline). While it is on, Central answers every token call with 503 device_blocked and a Retry-After — the health probe, which needs no token, keeps answering. The agent shows this as the link state BLOCKED rather than ONLINE:

• nothing is lost — records queue locally, and the refused batch goes back to the queue without counting as an attempt;
• the uploader and the command poller sit out;
• the heartbeat keeps trying at the Retry-After pace, and the first accepted beat puts the link back to ONLINE, registers, and drains the backlog.

Unlike a rotated token, there is nothing to do on the machine: flip the switch back in Central.

---

Development

This project uses uv for dependency management.

uv sync --group dev --all-extras

Quality tooling (wrappers run the tool from the project root):

./check.sh          # black --fix, isort --fix, flake8, mypy, pytest
./check.sh --ci     # formatters in check mode (what CI runs)
./check.sh --build  # the above plus `uv build`

Individual steps remain available: ./black.sh, ./isort.sh, ./flake8.sh, ./mypy.sh, ./pytest.sh.

To run the agent itself while working on it, see From the repository.

Tests that need something outside the process are opt-in markers and are excluded by default: live (a real Central server, see scripts/live/), weights (real framework packages and checkpoints), gpu (an NVIDIA device).

Plans live in docs/plans/; the current one is 06-frames-collector-plan.md.