CentralClient 로 직접 연결

장비에서 이미 실행 중인 애플리케이션이 중앙 API 를 직접 호출할 때 CentralClient 사용. 에이전트·로컬 큐·SQLite 없이 httpx 와 pydantic 만 필요하므로 엑스트라 없이 설치.

pip install geo-mlops-sdk

메서드#

메서드는 중앙의 장비 API(/api/v1/edge/…)와 일대일 대응. 모두 async, 토큰은 X-Edge-Token 헤더로 전송.

요청·응답 본문은 geo_mlops_sdk.contracts 의 Pydantic 모델(edge, records, uploads, inference, commands, models). 알 수 없는 필드는 무시하므로 중앙이 필드를 추가해도 이미 배포된 클라이언트에 오류가 발생하지 않음. 시그니처 전체는 Edge SDK API 레퍼런스 참고.

예시: 등록 → 하트비트 → 레코드 → 업로드 → 명령#

import asyncio
import hashlib
import os
import uuid
from datetime import datetime, timezone
from pathlib import Path

from geo_mlops_sdk.client import AuthError, CentralClient, OfflineError, RetryPolicy
from geo_mlops_sdk.contracts.commands import AckStatus, CommandAck
from geo_mlops_sdk.contracts.edge import HeartbeatBody
from geo_mlops_sdk.contracts.records import Record
from geo_mlops_sdk.contracts.uploads import UploadInit

BASE_URL = os.environ["GEO_EDGE_CENTRAL__BASE_URL"]
TOKEN = os.environ["GEO_EDGE_CENTRAL__TOKEN"]


async def main() -> None:
    retry = RetryPolicy(attempts=5, jitter=0.5)
    async with CentralClient(BASE_URL, TOKEN, retry=retry) as client:
        # 1. 등록. 부팅마다 호출해도 된다(멱등)
        device = await client.register(location="bench")
        print("registered:", device.id, device.status)

        # 2. 하트비트
        beat = await client.send_heartbeat(HeartbeatBody(cpu=12.5, mem=40.0, disk=55.0))
        print("heartbeat ok, policy_revision =", beat.policy_revision)

        # 3. 레코드. id 가 멱등 키다. 같은 배치를 다시 전송하면 duplicates 로 집계한다
        records = [
            Record(id=str(uuid.uuid4()), kind="sensor",
                   ts=datetime.now(timezone.utc), payload={"temp": 21.5})
        ]
        first = await client.send_records(records)
        again = await client.send_records(records)
        print("records:", first.accepted, "accepted /", again.duplicates, "duplicate")

        # 4. 파일 업로드: init → chunk → complete → status
        data = Path("frame-0001.jpg").read_bytes()
        created = await client.upload_init(UploadInit(
            filename="frame-0001.jpg", size=len(data),
            sha256=hashlib.sha256(data).hexdigest(), chunk_size=8 * 1024 * 1024))
        size = created.chunk_size  # 서버가 허용한 값으로 자른다
        for index in range(0, max(1, -(-len(data) // size))):
            await client.upload_chunk(created.upload_id, index,
                                      data[index * size:(index + 1) * size])
        await client.upload_complete(created.upload_id)
        status = await client.upload_status(created.upload_id)
        print("upload:", status.state.value)

        # 5. 명령. 롱폴로 받고, 처리 결과를 반드시 ack 한다
        listed = await client.poll_commands(wait_s=5)
        for command in listed.items:
            print("command:", command.type, command.args)
            await client.ack_command(command.id, CommandAck(status=AckStatus.OK,
                                                            result={"handled": True}))
        print("commands handled:", len(listed.items))


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except AuthError as exc:
        print("token rejected:", exc.status_code, exc.message)
    except OfflineError as exc:
        print("central unreachable:", exc)

출력(중앙에 ping 명령을 하나 등록한 상태):

registered: edge-bench-01 ACTIVE
heartbeat ok, policy_revision = 0
records: 1 accepted / 1 duplicate
upload: assembling
command: ping {}
commands handled: 1

참고 사항:

• 업로드 완료는 비동기. upload_complete 직후 상태는 assembling, 중앙이 조각을 합친 뒤 done 으로 전환. 결과가 필요하면 간격을 두고 upload_status 를 다시 호출.
• 청크 크기는 서버 값 사용. upload_init 에 요청한 값이 아니라 응답의 chunk_size 로 분할 필요. 서버가 앞단 프록시 한도에 맞춰 줄인 값을 반환.
• 이어서 업로드: 중단 후 다시 시작할 때는 upload_status(upload_id).received 에 서버가 이미 수신한 조각 번호 포함. 같은 번호를 다시 전송해도 문제없음.
• UploadInit(dataset_id=...) 를 지정하면 조립이 끝난 파일이 그 데이터셋 파일로 등록. 지정하지 않으면 수집 데이터에 보관.
• 토큰 교체는 연결을 새로 만들지 않고 client.set_token(new_token) 으로 수행.

오류 계층#

상태 코드는 반환값이 아니라 예외로 발생. 수신하지 못한 본문으로 계속 진행하는 실수 방지.

ApiError 에는 status_code, code, message, detail, error_id, method, url, retry_after 포함. 틀린 토큰과 접근할 수 없는 주소로 실행하면 다음과 같이 출력.

token rejected: 401 invalid or expired edge token
central unreachable: POST /api/v1/edge/register: All connection attempts failed

엣지에서는 둘의 구분 필요. OfflineError 는 데이터를 계속 저장했다가 나중에 전송해야 하는 상황, ApiError 는 원인을 수정해야 하는 상황.

재시도: RetryPolicy#

429, 5xx, 전송 실패(OfflineError)만 재시도. 4xx 는 재전송해도 똑같이 실패하므로 재시도하지 않음.

from geo_mlops_sdk.client import NO_RETRY, CentralClient, RetryPolicy

client = CentralClient(url, token, retry=RetryPolicy(attempts=5, jitter=0.5))
await client.send_heartbeat(body, retry=NO_RETRY)   # 이 호출만 재시도 비활성화

• 서버가 Retry-After 를 반환하면 그 값을 따르되 max_backoff_s 를 넘지 않음. 이때 jitter(대기 시간을 무작위로 조금씩 변경하는 것)는 대기를 늘리는 방향으로만 적용.
• jitter 는 함께 연결이 끊어졌던 장비들이 동시에 재접속해 서버에 부하가 집중되는 것을 방지.
• health() 와 poll_commands() 는 호출 측이 이미 루프로 반복하므로 재시도하지 않음.

그 밖의 옵션#

CentralClient(
    base_url,                 # 예: https://mlops.example.com (스킴과 호스트 필수)
    token="",
    timeout=10.0,             # 초
    retry=RetryPolicy(),
    verify=True,              # TLS 검증
    user_agent=None,          # 기본 geo-mlops-sdk/0.2.0
)