예제: YOLO 영상 데모

영상에 YOLO 객체 검출을 실행하고 결과를 중앙에 보고하는 데모 앱 mlops-edge-demo. 에이전트(geo-mlops-edge)를 따로 실행하지 않고 애플리케이션 안에 SDK 런타임(EdgeRuntime)을 포함하는 구성의 예. 소스는 공개 저장소 https://github.com/osom8979/mlops-edge.

구성#

중앙과 주고받는 데이터는 다음과 같음.

준비#

• uv. Python 3.12 는 저장소의 .python-version 에 맞춰 uv 가 설치.
• 창을 표시할 화면. 화면이 없는 장비는 --no-show 로 실행.
• (선택) NVIDIA GPU. 없으면 CPU 로 추론.
• 장비에서 중앙 API 주소로 향하는 HTTPS 연결, 디바이스 등록용 테넌트 ADMIN 계정(🔑 설정ADMIN).

git clone https://github.com/osom8979/mlops-edge.git
cd mlops-edge
uv sync

Linux 에서는 CUDA 빌드 PyTorch 와 GPU 라이브러리(합계 약 3 GB)를 함께 설치. 회선이 느려 Failed to download distribution due to network timeout 이 발생하면 UV_HTTP_TIMEOUT=600 uv sync 로 재실행.

1. 디바이스와 데이터셋 생성#

2. 설정#

저장소의 edge.yaml 에서 세 곳 변경.

central:
  base_url: https://mlops.example.com   # 중앙 API 주소

device:
  id: edge-yolo-01                      # 1단계의 디바이스 ID

collectors:
  - type: frames
    name: demo-frames
    priority: 30                        # 추론 결과(50)보다 낮게. 회선이 좁으면 결과를 먼저 전송
    options:
      every_n_frames: 30                # 30 fps 영상이면 초당 1장
      dataset_id: "ds-0123456789ab"     # 1단계의 데이터셋 ID. 비우면 수집 데이터
      labels: true                      # LabelMe 사전 라벨 동반
      label_conf: 0.5                   # 이 신뢰도 이상인 검출만 사전 라벨로 사용

토큰은 설정 파일이 아니라 .env 에 저장.

cp .env.example .env
# .env 에 입력: GEO_EDGE_CENTRAL__TOKEN=<device-token>

• 설정 우선순위는 환경 변수 > .env > edge.yaml > 기본값(SDK 규칙과 동일). 전체 항목은 설정 레퍼런스 참고.
• 데모의 edge.yaml 은 화면에서 빨리 확인하도록 하트비트를 10초로 지정(기본 30초).
• 토큰이 없어도 실행 가능. 결과는 로컬 큐(data/edge/)에만 누적되고, 토큰 지정 후 재실행하면 전송.

3. 실행#

uv run mlops-edge-demo

인자를 생략하면 기본 영상(도심 주행 17초)을 cache/videos/ 에 한 번 내려받아 반복 재생. 가중치 yolo26n.pt(COCO 사전학습)도 cache/weights/ 에 내려받음. 창에서 q 또는 ESC 로 종료.

창 왼쪽 위 상태 표시줄의 의미는 다음과 같음.

다른 영상과 실행 방식:

uv run mlops-edge-demo ./line-camera.mp4            # 로컬 파일 (내려받지 않고 바로 열기)
uv run mlops-edge-demo rtsp://192.0.2.10/stream     # RTSP 스트림
uv run mlops-edge-demo --no-show                    # 화면 없는 장비. 5초마다 상태를 로그로 출력

데모 안의 런타임이 로컬 API(127.0.0.1:8600)를 열기 때문에 SDK CLI 로 상태 조회 가능(CLI).

uv run geo-mlops-edge status --config edge.yaml          # 연결, 대기 건수, 모델, 조치 필요 사항
uv run geo-mlops-edge queue --config edge.yaml --limit 5

같은 PC 에서 geo-mlops-edge 에이전트가 이미 8600 포트를 사용 중이면 데모는 로컬 API 만 비활성화(local API disabled 경고)하고 중앙 연동은 유지.

4. 중앙에서 확인#

• 파일 이름은 <영상 이름>_<프레임 번호 6자리>.jpg. URL 로 받은 영상은 캐시 파일 이름(URL 해시 12자 + 원래 이름)이 영상 이름. 예: c5144a2b5077-kitti-inference-vid_000420.jpg.
• 사전 라벨은 현재 모델의 예측. 그대로 학습하면 모델이 자기 오류를 다시 학습하므로 학습 전 사람의 검수 필요.
• 반복 재생하거나 재실행해도 이미 큐에 넣은 프레임은 다시 전송하지 않음. 기록은 data/edge/collectors/demo-frames.seen.

5. 학습한 모델로 교체#

수집한 프레임으로 학습한 모델을 이 장비로 다시 전송하는 절차.

• 화면에 반영하는 것은 YOLO 프레임워크 모델만. 다른 프레임워크 모델은 경고 로그만 남기고 기존 모델로 계속 재생.
• 캐시한 모델은 다음 실행 시 다시 활성화. 활성화 규칙은 모델과 추론 참고.
• 승인(Production 승격)이나 배포 화면은 장비로 모델을 전송하지 않음. 장비로 보내는 방법은 pull_model 명령, 장비 CLI, 로컬 API 의 세 가지.

재시작과 종료#

• q·ESC·Ctrl-C·SIGTERM 으로 종료하면 런타임이 큐 정리와 DB 닫기를 마친 뒤 종료.
• 디바이스 상세의 재시작 이나 restart 명령을 받으면 종료 코드 3 으로 종료. systemd 같은 감시 프로그램 아래에서 실행하면 재기동(배포, 명령·정책·보존).

코드 구성#

애플리케이션에 런타임을 포함할 때의 핵심은 스레드 분리. OpenCV 창은 메인 스레드에서 동작하고 EdgeRuntime 은 asyncio 이벤트 루프가 필요하므로, 런타임은 별도 스레드의 루프에서 실행하고 영상 루프는 asyncio.run_coroutine_threadsafe 로 작업만 넘긴 뒤 기다리지 않음.

# edge.py 의 요약. 런타임 스레드에서 실행
runtime = EdgeRuntime(settings)
runtime.events.add_listener(on_event)   # pull_model 로 모델이 활성화되면 알림
await runtime.start()                   # 등록·하트비트·명령·업로더·수집기 시작
await serve(runtime, settings, stop)    # 로컬 API (geo_mlops_sdk.edge.daemon.serve)

# 영상 루프(메인 스레드)에서: 수집 대상 프레임만 인코딩해 frames 수집기로 전달
if frames.wants(source, index, seekable=True):
    ok, jpeg = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 90])
    asyncio.run_coroutine_threadsafe(
        frames.collect(source, index, jpeg.tobytes(), output, model=model_ref), loop
    )

frames 수집기 API 는 데이터 수집기, 추론 결과 형식(InferenceOutput)은 모델과 추론 참고.