데이터 수집기

수집기는 장비가 생성하는 데이터(레지스터, 파일, HTTP 응답, 모델을 실행한 프레임)를 레코드(구조화된 값)나 블롭(파일)으로 변환해 로컬 큐에 저장. 중앙 전송은 큐와 업로더가 담당하므로 수집기는 링크가 끊어져도 계속 동작.

edge.yaml 의 collectors: 에 선언. 코드 작성 불필요.

공통 필드#

collectors:
  - type: http          # 종류 (필수)
    name: gateway-1     # 이름. 화면의 수집기 목록과 레코드의 collector 값
    enabled: true       # false 면 건너뜀
    priority: 50        # 0~100. 기본 50
    options: { ... }    # 종류별 옵션

priority 는 보존과 업로드에 모두 적용. 용량·디스크 여유 한도를 넘으면 낮은 것부터 삭제(기간 한도는 우선순위와 무관하게 삭제), 전송 시에는 높은 것부터 전송. sync.urgent_priority(기본 90) 이상은 전송 시간대 제한도 무시. 특별히 중요한 데이터가 아니면 기본값 50 유지.

설정이 잘못된 수집기는 로그에 기록하고 건너뜀. 나머지 수집기와 중앙 통신은 계속 동작. 건너뛴 수집기는 수집기 목록에 표시되지 않으므로 로그의 collector '<이름>' could not be built 로 확인. 수집기 상태(running·disconnected·stopped)와 마지막 오류는 디바이스 상세의 수집기 카드와 geo-mlops-edge status 에 표시.

옵션은 options: 아래에도, 같은 줄에도 작성 가능.

http: 다른 서버를 폴링#

- 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>" }   # 선택
    auth: { username: edge, password: <your-secret> }    # 선택, HTTP Basic
    ts_field: measured_at      # 응답에 관측 시각이 있으면 그 필드
    max_body_bytes: 1MiB       # 응답 크기 상한. 0 이면 검사 안 함

레코드 내용은 {"collector": "gateway-1", "body": <응답 JSON>}. items_path 를 사용하면 항목마다 {"collector": …, "item": <항목>}, 레코드 id 는 <수집기 이름>:<id_field 값>.

같은 값을 수신해도 매번 한 건으로 저장. "12:00:01 에도 21.5 였다"는 것도 데이터이기 때문. 데이터 양은 주기 × 응답 크기 로 추정. 2 KiB 를 1초마다 읽으면 하루 약 177 MB. 기본 보존 한도(50 GiB / 30일)에서는 기간 한도에 먼저 도달.

지난 기록 목록을 반환하는 엔드포인트#

"최근 알람 N건"처럼 지난 기록 목록을 반환하는 엔드포인트는 폴링 시마다 이미 수신한 행을 다시 반환. 배열 위치와 키를 지정하면 항목마다 한 번씩만 큐에 저장.

- 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

대상 서버가 응답하지 않거나 느리거나 비정상적인 값을 반환하면 수집기는 오류를 기록하고 disconnected 로 표시. 이후 간격을 1초에서 30초까지 늘리며 재시도해 자동 복구.

push: 외부에서 레코드 수신#

로봇 컨트롤러·비전 PC·다른 언어로 된 서비스가 LAN 에서 에이전트로 레코드를 푸시할 때 사용.

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

송신 측은 직접 정한 id 로 PUT 요청 전송.

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 은 저장 완료, 200 과 "duplicate": true 는 같은 id 가 아직 큐에 있음을 의미. 이미 중앙으로 전송한 id 를 다시 전송하면 201 반환, 단 중앙은 같은 id 를 한 번만 저장. 응답을 수신하지 못해 재전송해도 한 건만 저장.
• ts 는 선택 사항. 없으면 수신 시각 사용.
• kind(생략하면 push)와 priority 는 YAML 에서 지정. 송신 측이 정의되지 않은 종류 이름으로 데이터를 입력하는 것을 방지.
• 선언된 push 수집기는 수집기 목록에 마지막 수신 시각과 함께 표시. 송신 측이 중단되어도 오류는 발생하지 않음. 이 시각이 더 이상 갱신되지 않는 것이 유일한 신호.

선언 없이 전송#

설정에 없는 프로세스용으로 POST /api/v1/records, POST /api/v1/blobs 도 제공. 대신 id 를 에이전트가 생성하므로 재시도하면 두 건이 되고, 수집기 목록에도 표시되지 않음.

modbus: PLC 레지스터 읽기#

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   # 레지스터 맵 파일 (YAML 안에 바로 적어도 됨)

레지스터 맵은 일반 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비트 값은 2개의 레지스터 점유. word_order 가 틀리면 오류 대신 그럴듯한 엉터리 값이 나온다
  - { 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 }

연속된 주소는 한 번의 요청으로 읽음. 레코드 내용은 {"collector": "line-1", "version": "1", "fields": {"temperature.zone1": 21.3, ...}}.

PLC 가 꺼져 있으면 disconnected 로 표시, 1초부터 30초까지 간격을 늘리며 재접속 시도.

watchdir: 폴더에 새로 생긴 파일#

카메라, 라이다, 구형 장비 프로그램이 폴더에 저장하는 파일을 가져와 업로드. 파일은 파일 탭에 표시.

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

쓰기 중인 파일을 수집하지 않도록, 크기가 한 주기 동안 변하지 않은 파일만 가져옴. 에이전트 계정에는 그 폴더의 쓰기 권한도 필요. 원본을 삭제하지 못하면 같은 파일을 반복 수집하지는 않지만 수집기에 오류로 표시되고, 에이전트 재시작 시 그 파일을 한 번 더 업로드.

frames: 모델이 판정한 프레임#

장비에서 카메라·영상에 모델을 실행하는 애플리케이션이, 판정한 프레임을 학습 데이터로 보내는 수집기. 다른 수집기와 달리 장비가 아니라 애플리케이션 코드가 프레임을 전달. every_n_frames 마다 이미지 1장과, 그 프레임의 검출 결과를 담은 LabelMe JSON 1개를 큐에 추가. JSON 은 사람이 검수할 사전 라벨(모델이 미리 그린 박스). 그대로 학습하면 모델이 자기 오류를 다시 학습하므로 학습 전 검수 필요.

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

애플리케이션은 에이전트와 같은 프로세스에서 수집기를 찾아 프레임을 전달. 이미지는 인코딩된 bytes(JPEG 등)로 전달. SDK 는 OpenCV 에 의존하지 않으므로 인코딩은 애플리케이션 몫.

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

# 비디오 루프, 매 프레임
if frames.wants(source, index):                  # 동기 함수. 인코딩 전에 호출
    ok, jpeg = cv2.imencode(".jpg", frame)
    await frames.collect(source, index, jpeg.tobytes(), output,
                         model=ModelRef(name="helmet-detector", version="3"))

• output 은 추론 결과와 같은 InferenceOutput. bbox 는 0~1 정규화 값이고, 수집기가 width·height 로 픽셀 좌표의 2점 rectangle 로 변환.
• wants() 는 수집 간격과 중복 여부만 확인하는 동기 함수. 비디오 스레드에서 호출 가능. collect() 는 코루틴이므로 에이전트의 이벤트 루프에서 실행(다른 스레드면 asyncio.run_coroutine_threadsafe). 수집 대상이 아닌 프레임이면 False 반환.
• 파일 이름은 <영상 이름>_<프레임 번호 6자리>.jpg 와 같은 이름의 .json. 중앙은 같은 이름의 이미지와 라벨을 짝짓고, YOLO 학습 런타임은 사각형을 박스로 변환.
• 큐에 넣은 프레임은 <data_dir>/collectors/<이름>.seen 에 기록. 반복 재생하거나 재시작해도 같은 프레임은 다시 업로드하지 않음. 처음부터 다시 수집하려면 이 파일을 삭제하고 재시작.
• 실시간 스트림(RTSP 등)은 seekable=False 로 전달. 프레임 번호 대신 시각으로 이름을 짓고 중복 검사 없음.
• 장비 토큰에 data:write(파일) 스코프 필요.
• 메타데이터 collector·source(영상 이름)·frame(번호)·model(이름:버전)·dataset_id 는 수집 데이터 상세의 메타데이터에 표시.

커스텀 수집기#

기본 다섯 가지로 지원되지 않는 장비(시리얼 포트, 전용 SDK 등)는 수집기를 직접 만들어 등록. register_collector 는 같은 프로세스 내부에서 에이전트 기동 전에 호출해야 하므로 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"          # 화면에 보이는 종류

    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)        # 수집기 카드의 '마지막 수신' 갱신
            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

• 팩토리는 factory(name=..., priority=..., options=...) 로 호출.
• sink.record(kind, payload, priority=, ts=, meta=, record_id=) 는 레코드를, sink.blob(kind, 경로|bytes, filename=, priority=, move=) 는 파일을 큐에 추가. record_id 를 지정하면 같은 id 는 한 번만 저장.
• sink.state_dir 는 재시작 후에도 기억할 상태를 두는 폴더(<data_dir>/collectors). 파일 이름에 수집기 이름 포함 권장.
• CollectorBase 를 상속하지 않아도 name, start(sink), stop(), status() 만 있으면 동작. 상속하면 note_emit()·note_error() 로 화면의 수집기 상태가 올바르게 표시.
• run() 이 반환하는 종료 코드(재시작 요청이면 3)를 그대로 sys.exit 에 전달해야 systemd 가 올바르게 재시작.

실행하면 디바이스 상세 수집기 카드에 counter-1 (counter) running 으로, 레코드 탭에 {"value": 8} 같은 레코드로 표시.