pipeline/docs/FLEET_EDGE_INTEGRATION.md

# Fleet × Edge Data Collector 연동 가이드

로컬 Pipeline과 엣지(공장) Python Data Collector를 연동하는 방법입니다.

## 연동 방식

```
[Python Data Collector]           [Pipeline (로컬)]
  ▲                                 ▲
  │ 1. GET /api/fleet/v1/edges/     │
  │      {edgeId}/config            │
  │                                 │
  │                                 │
  │ 2. PLC 수집 수행                │
  │                                 │
  │ 3. vexplor/devices/{edgeId}/    │
  │      data 로 MQTT publish       │
  │      vexplor/devices/{edgeId}/  │
  └──── status (heartbeat) ────────▶│
                                    │
                                    ▼
                               fleet_edge_raw_data
                               fleet_heartbeats
```

## 엣지 설정 (.env)

기존 엣지 `/home/wace/data-collector/.env` 수정:

```bash
# Pipeline 서버 URL (Fleet API + MQTT)
EDGE_SERVER_URL=http://<pipeline-host>:8080
MQTT_BROKER_URL=mqtt://<pipeline-host>:1883

# 기존 유지
DEVICE_ID=spifox-001
COMPANY_CODE=spifox
EDGE_CONFIG_SOURCE=api   # 'aas' 대신 'api' 선택 시 Pipeline Fleet API 호출
LOG_LEVEL=INFO

# Kafka는 로컬에서 불필요 (Pipeline 내장 MQTT 사용)
# KAFKA_BROKERS= (비워두기)
```

## Pipeline API 엔드포인트

Python Data Collector가 호출하는 엔드포인트:

| 메서드 | 경로 | 용도 |
|---|---|---|
| `GET` | `/api/fleet/v1/edges/{edgeId}/config` | 수집 설정 조회 (ETag 캐싱) |
| `GET` | `/api/fleet/edge/{edgeId}/config` | 위와 동일 (alias) |

응답 형식 (Python `EdgeConfig` 모델과 호환):

```json
{
  "version": "2026-04-17T07:25:26.766Z",
  "edge_id": "edge-spifox-001",
  "edge_name": "스피폭스 엣지 #1",
  "devices": [
    {
      "id": "1",
      "name": "CASE프레스_PLC_01",
      "protocol": "plc_ethernet",
      "connection": {
        "host": "192.168.1.10",
        "port": 2004,
        "unit_id": 1
      },
      "interval_ms": 1000,
      "enabled": true,
      "tags": [
        {
          "name": "temperature",
          "address": "40001",
          "data_type": "UINT16",
          "byte_order": "BIG_ENDIAN",
          "scale": 0.1,
          "offset": 0,
          "unit": "°C"
        }
      ]
    }
  ],
  "aggregation_interval_sec": 60,
  "local_retention_days": 7
}
```

## MQTT 토픽 규칙

Python이 발행하는 토픽:

| 토픽 | 페이로드 | 주기 |
|---|---|---|
| `vexplor/devices/{edgeId}/status` | heartbeat (CPU/메모리/디스크) | 30초 |
| `vexplor/devices/{edgeId}/data` | 태그 값 (아래 참조) | interval_ms |
| `vexplor/devices/{edgeId}/responses` | 커맨드 응답 | 요청 시 |

### 데이터 페이로드 예시

```json
{
  "timestamp": "2026-04-17T08:00:00.123Z",
  "equipment_id": 4,
  "connection_id": 1,
  "tags": {
    "temperature": 25.4,
    "pressure": 11.2,
    "status": true,
    "mode": "AUTO"
  }
}
```

Pipeline은 이 데이터를 `fleet_edge_raw_data` 테이블에 자동 적재합니다.

## 로컬 테스트

Pipeline이 로컬에 떠있는 상태에서 테스트 엣지 시뮬레이터:

```bash
# MQTT heartbeat 발송 (자동 등록)
docker exec pipeline-backend node -e "
const mqtt = require('mqtt');
const c = mqtt.connect('mqtt://127.0.0.1:1883');
c.on('connect', () => {
  c.publish('vexplor/devices/edge-test-001/status', JSON.stringify({
    cpu_percent: 25, memory_percent: 45, disk_percent: 60,
    ip_address: '192.168.1.100', status: 'online'
  }), { qos: 1 }, () => c.end(true));
});
"

# 설정 조회
curl http://localhost:8080/api/fleet/edge/edge-test-001/config

# 태그 데이터 발송
docker exec pipeline-backend node -e "
const mqtt = require('mqtt');
const c = mqtt.connect('mqtt://127.0.0.1:1883');
c.on('connect', () => {
  c.publish('vexplor/devices/edge-test-001/data', JSON.stringify({
    timestamp: new Date().toISOString(),
    equipment_id: 4,
    tags: { temperature: 25.5, pressure: 11.2 }
  }), { qos: 1 }, () => c.end(true));
});
"
```

## 포트 정리

로컬 Pipeline이 노출하는 포트:

| 포트 | 용도 |
|---|---|
| `8080` | REST API (Fleet + Pipeline) |
| `1883` | MQTT TCP 브로커 (내장 aedes) |
| `8083` | MQTT WebSocket (브라우저 클라이언트) |
| `9771` | 프론트엔드 |

## 흐름 요약

1. **엣지 부팅**: Python이 Pipeline에 heartbeat 발행 → `fleet_devices`에 자동 등록
2. **설정 조회**: Python이 `/api/fleet/v1/edges/{id}/config` 호출 → 현재 장비/태그 설정 받음
3. **PLC 수집**: 설정된 대로 Modbus/OPC UA/S7 등으로 주기 수집
4. **MQTT 발행**: `vexplor/devices/{id}/data` 로 실시간 값 발행
5. **Pipeline 저장**: MQTT 구독 → `fleet_edge_raw_data` 적재
6. **대시보드 표시**: `/admin/fleet/data` 에서 실시간 차트 + 최신값 조회

## 설정 변경 시 반영

사용자가 **웹에서 태그 설정을 변경**하면:
- `pipeline_tag_mappings` UPDATE
- Python이 다음 config sync 주기(기본 30초) 시 변경 감지
- `version` (ETag) 기반이라 변경 없으면 304 응답 (트래픽 절약)
- Python이 자동으로 새 설정으로 수집 재시작

**Python 재시작 불필요** — 설정은 런타임에 동적 반영됩니다.