Data collectors

A collector turns what the equipment produces (registers, files, HTTP responses, frames a model has run on) into records (structured values) or blobs (files) and puts them into the local queue. The queue and the uploader send them to Central, so collectors keep running while the link is down.

You declare them under collectors: in edge.yaml. No code needed.

Common fields#

collectors:
  - type: http          # type (required)
    name: gateway-1     # name. The collector list on screen and the record's collector value
    enabled: true       # false skips it
    priority: 50        # 0-100. Default 50
    options: { ... }    # type-specific options

priority is read by both retention and upload. When the size or free-disk limit is exceeded, the lowest are dropped first (the age limit drops items regardless of priority); when sending, the highest go first. At sync.urgent_priority (default 90) and above, the upload window restriction is ignored too. Unless something is especially important, leave it at 50.

A single misconfigured collector is logged and skipped. The other collectors and communication with Central keep running. A skipped collector does not appear in the collector list, so look for collector '<name>' could not be built in the log. Collector state (running, disconnected, stopped) and the last error appear in the Collectors (수집기) card on the device detail page and in geo-mlops-edge status.

You can write options under options: or directly on the same level.

http — poll another server#

- 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 <your-secret>" }   # optional
    auth: { username: edge, password: <your-secret> }    # optional, HTTP Basic
    ts_field: measured_at      # the field holding the observation time, if the response has one
    max_body_bytes: 1MiB       # response size limit. 0 disables the check

The record content is {"collector": "gateway-1", "body": <response JSON>}. With items_path, each item becomes {"collector": …, "item": <item>}, and the record id is <collector name>:<id_field value>.

The same value still makes one record every time. "It was still 21.5 at 12:00:01" is data too. So volume is predictable as interval × response size — reading 2 KiB every second is about 177 MB a day, and with the default retention limits (50 GiB / 30 days) the age limit is hit first.

Endpoints that return a list of past entries#

An endpoint that returns a list of past entries, such as "the last N alarms", sends rows you already have on every poll. Tell it the array location and the key, and each item is queued only once.

- type: http
  name: alarms
  priority: 70
  options:
    url: http://10.0.0.7/api/alarms
    interval_ms: 5000
    items_path: data.items
    id_field: id

If the other server goes down, slows or returns odd values, the collector logs the error and marks itself disconnected. It then retries at intervals growing from 1 second to 30 seconds and recovers on its own.

push — an entry point for pushed records#

Use it when a robot controller, a vision PC or a service written in another language pushes records to the agent over the LAN.

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

The sender PUTs with an id it chooses.

curl -X PUT http://edge-pc: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 = stored, 200 + "duplicate": true = the same id is still in the queue. If that id has already been sent to Central, sending it again returns 201, but Central stores the id only once. If the response is lost and you send again, only one record remains.
• ts is optional. Without it, the receive time is used.
• The YAML decides kind (push if omitted) and priority. A sender cannot put data in under a kind nobody looks at.
• A declared push collector shows in the collector list with its last receive time. If the sender dies, not even an error arrives, so this time stopping is the only signal.

Pushing without a declaration#

For processes not in the config, POST /api/v1/records and POST /api/v1/blobs are also open. But the agent creates the id, so a retry makes two records, and they do not appear in the collector list.

modbus — read PLC registers#

Needs pip install 'geo-mlops-sdk[edge,modbus]'.

- type: modbus
  name: line-1
  priority: 50
  options:
    host: 10.0.0.5
    port: 502
    unit_id: 1
    interval_ms: 1000
    schema: /etc/geo-mlops/plc.yaml   # register map file (can also be written inline in the YAML)

The register map is plain YAML.

# /etc/geo-mlops/plc.yaml
version: "1"
unit_id: 1

fields:
  # type: holding | input | coil | discrete
  # dtype: bool | uint16 | int16 | uint32 | int32 | float32
  - { name: temperature.zone1, type: holding, address: 100, dtype: uint16, scale: 0.1 }
  - { name: temperature.zone2, type: holding, address: 101, dtype: uint16, scale: 0.1 }

  # 32-bit values take two registers. A wrong word_order gives plausible garbage instead of an error
  - { name: flow_rate, type: holding, address: 110, dtype: float32, word_order: big }
  - { name: cycle_count, type: holding, address: 112, dtype: uint32 }

  - { name: running, type: coil, address: 5 }
  - { name: fault, type: discrete, address: 12 }

Contiguous addresses are read in one go. The record content is {"collector": "line-1", "version": "1", "fields": {"temperature.zone1": 21.3, ...}}.

If the PLC is off, the collector is marked disconnected and reconnects at intervals from 1 second up to 30 seconds.

watchdir — files dropped in a folder#

Picks up files that cameras, lidars or legacy tools write into a folder. The files go to the Files (파일) tab.

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

So as not to pick up a file that is still being written, it takes only files whose size stayed the same for one interval. The agent's account also needs write permission on that folder. If it cannot delete the original, it does not pick up the same file repeatedly, but the collector shows an error, and after an agent restart it uploads that file once more.

frames — frames the model has judged#

For an application on the device that runs a model on a camera or a video and wants the judged frames as training data. Unlike the other collectors, it is fed by your application code, not by equipment. Every every_n_frames it queues one image and one LabelMe JSON with that frame's detections. The JSON holds pre-labels (boxes the model drew) for a person to review; train on them unreviewed and the model learns its own mistakes back.

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

The application finds the collector in the same process as the agent and hands it frames. The image goes in as encoded bytes (JPEG and so on): the SDK does not depend on OpenCV, so encoding is the application's job.

from geo_mlops_sdk.contracts.inference import ModelRef
from geo_mlops_sdk.edge.collectors.frames import FrameCollector

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

# the video loop, every frame
if frames.wants(source, index):                  # synchronous; ask before encoding
    ok, jpeg = cv2.imencode(".jpg", frame)
    await frames.collect(source, index, jpeg.tobytes(), output,
                         model=ModelRef(name="helmet-detector", version="3"))

• output is the same InferenceOutput as an inference result. bbox is normalized 0–1; the collector scales it by width and height into a two-point rectangle in pixels.
• wants() only checks the stride and the dedup ledger. It is synchronous and safe to call from the video thread. collect() is a coroutine for the agent's event loop (from another thread, use asyncio.run_coroutine_threadsafe). It returns False for a frame that was not wanted.
• Files are named <source name>_<frame number, 6 digits>.jpg and .json with the same stem. Central pairs the image and label by stem, and the YOLO training runtime turns the rectangles into boxes.
• Frames already queued are recorded in <data_dir>/collectors/<name>.seen, so a looping video or a restart does not upload the same frame again. Delete that file and restart to collect from the start.
• For a live stream (RTSP and the like) pass seekable=False. Frames are then named by time instead of frame number, with no dedup check.
• The device token needs the data:write (files) scope.
• The metadata collector, source (video name), frame (number), model (name:version) and dataset_id show in the metadata of the collected-data detail.

Custom collectors#

For equipment the five built-in types cannot handle (serial ports, vendor SDKs and so on), write your own collector and register it. register_collector must be called in the same process before the agent starts, so use a small launcher script instead of geo-mlops-edge run.

# my_edge.py
import asyncio
import sys
from datetime import datetime, timezone
from pathlib import Path

from geo_mlops_sdk.edge.collectors import CollectorBase, register_collector
from geo_mlops_sdk.edge.daemon import run
from geo_mlops_sdk.edge.settings import EdgeSettings


class CounterCollector(CollectorBase):
    type_name = "counter"          # type shown on screen

    def __init__(self, name, *, priority=50, interval_s=5.0):
        super().__init__(name, priority=priority)
        self.interval_s = interval_s
        self._task = None

    @classmethod
    def from_options(cls, *, name, priority, options):
        return cls(name, priority=priority,
                   interval_s=float(options.get("interval_s", 5)))

    async def start(self, sink):
        self.state = "running"
        self._task = asyncio.create_task(self._loop(sink))

    async def stop(self):
        if self._task:
            self._task.cancel()
        self.state = "stopped"

    async def _loop(self, sink):
        value = 0
        while True:
            value += 1
            now = datetime.now(timezone.utc)
            await sink.record("counter", {"value": value},
                              priority=self.priority, ts=now)
            self.note_emit(now)        # updates 'last received' on the collector card
            await asyncio.sleep(self.interval_s)


register_collector("counter", CounterCollector.from_options)

if __name__ == "__main__":
    config = Path(sys.argv[1]) if len(sys.argv) > 1 else None
    sys.exit(run(EdgeSettings.load(config)))

collectors:
  - type: counter
    name: counter-1
    options:
      interval_s: 15

python my_edge.py /etc/geo-mlops/edge.yaml

• The factory is called as factory(name=..., priority=..., options=...).
• sink.record(kind, payload, priority=, ts=, meta=, record_id=) queues a record, and sink.blob(kind, path|bytes, filename=, priority=, move=) queues a file. If you pass record_id, the same id goes in only once.
• sink.state_dir is a folder for state that must survive a restart (<data_dir>/collectors). Put the collector's name in the file name.
• You do not have to subclass CollectorBase; name, start(sink), stop() and status() are enough. If you do subclass it, note_emit() and note_error() fill in the collector state on screen correctly.
• Pass the exit code returned by run() (3 for a restart request) straight to sys.exit so systemd restarts it correctly.

When it runs, it shows up as counter-1 (counter) running in the Collectors (수집기) card on the device detail page, and as records like {"value": 8} in the Records (레코드) tab.