이런 상황 있으신가요?
Omron PLC와 통신해야 하는데 Python 라이브러리를 찾다 보면 이런 상황이 생깁니다.
pip search fins-python → 결과 없음
pip install fins → 설치는 되는데 문서가 없거나 오래됨
pip install omron-fins → 없음Mitsubishi SLMP는 자체 구현했고, Modbus는 pymodbus가 있었는데, Omron FINS는 제대로 된 공개 라이브러리가 없었습니다.
결국 Omron 공식 스펙 문서를 열고 직접 짰어요.
결론부터 말하면, 구현 자체는 생각보다 단순합니다.
TCP 연결 → 20바이트 핸드쉐이크 → 읽기 명령 전송 → 응답 파싱. 이 순서예요.
단, 한 군데서 1바이트를 잘못 짜서 전체 응답 파싱이 어긋나는 버그가 있었습니다.
그 포인트를 제일 자세하게 다룰게요.
용어 먼저 짚고 넘어갈게요
FINS (Factory Interface Network Service)란
Omron이 만든 PLC 통신 프로토콜입니다.
SLMP가 Mitsubishi/Keyence 전용인 것처럼, FINS는 Omron PLC 전용이에요.
TCP 위에서 동작하는 버전이 FINS/TCP입니다.
핸드쉐이크(Handshake)란
FINS/TCP는 TCP 연결 후 곧바로 명령을 보내는 게 아니라, 먼저 "나 여기 있어요, 번호가 뭐예요?"라는 확인 과정을 거칩니다.
이 첫 번 대화를 핸드쉐이크라고 해요.
악수를 먼저 하고 대화를 시작하는 것처럼.
struct.pack이란
Python에서 숫자 여러 개를 바이너리 형식으로 묶는 함수입니다.
PLC에 보낼 패킷을 만들 때 씁니다.
">BBBHBH" 같은 형식 문자열이 "몇 바이트짜리 숫자가 몇 개 있는가"를 정의해요.
이 형식 문자열 하나가 틀리면 파싱 전체가 어긋납니다.
Area Code란
FINS에서 "어느 메모리 영역을 읽을 것인가"를 구분하는 1바이트 코드입니다.
DM 영역은 0x82, CIO 워드는 0xB0, CIO 비트는 0x30 같은 식으로 구분합니다.
편의점 진열대의 구역 번호처럼, 같은 번호라도 어느 구역이냐에 따라 전혀 다른 물건이 나옵니다.
FINS/TCP 통신 구조 : 두 단계로 이뤄진다
FINS/TCP 통신은 크게 두 단계입니다.
1단계: FINS/TCP 핸드쉐이크 (TCP 연결 직후 1회)
클라이언트 → 서버: "안녕하세요, 제 노드 번호가 뭔가요?"
서버 → 클라이언트: "당신은 N번입니다, 저는 M번입니다"
2단계: FINS 명령 전송 (읽기/쓰기 요청)
클라이언트 → 서버: "DM0부터 5개 읽어주세요"
서버 → 클라이언트: [값 5개]핸드쉐이크에서 받은 노드 번호를 이후 FINS 명령 헤더에 포함해서 보내야 합니다.
이 노드 번호가 없으면 명령이 거부됩니다.
1단계 : 핸드쉐이크 (20바이트)
핸드쉐이크 패킷은 20바이트 고정입니다.
import struct
def build_fins_handshake():
return (
b"FINS" # 매직 헤더 4바이트
+ struct.pack(">IIII",
0x0000000C, # length: 이후 데이터 12바이트
0x00000000, # command: 0 = 노드 주소 확인
0x00000000, # error code: 0
0x00000000, # client node: 0 = 자동 할당 요청
)
)b"FINS"로 시작하는 게 특징입니다.
이걸 보내면 서버가 응답으로 클라이언트 노드 번호와 서버 노드 번호를 알려줍니다.
async def connect_and_handshake(host, port):
reader, writer = await asyncio.open_connection(host, port)
# 핸드쉐이크 전송
writer.write(build_fins_handshake())
await writer.drain()
# 응답 24바이트 수신
resp = await reader.readexactly(24)
# 노드 번호 파싱
client_node = resp[19] # 20번째 바이트
server_node = resp[23] # 24번째 바이트
return reader, writer, client_node, server_node2단계 : 메모리 읽기 명령
핸드쉐이크가 끝나면 실제 읽기 명령을 보냅니다.
FINS 패킷은 크게 세 부분으로 구성됩니다.
[FINS/TCP 헤더 16바이트] + [FINS 헤더 10바이트] + [명령 데이터]FINS 명령 데이터 (메모리 읽기):
def build_memory_read_command(area_code, start_addr, count):
return struct.pack(">BBBHBH",
0x01, 0x01, # MRC=0x01, SRC=0x01 (메모리 읽기)
area_code, # 0x82=DM, 0xB0=CIO 워드, 0x30=CIO 비트
start_addr, # 시작 주소 (2바이트)
0x00, # 비트 위치 (워드 접근 시 0x00)
count, # 읽을 개수 (2바이트)
)여기서 형식 문자열 ">BBBHBH" 를 주목해야 합니다.
이게 버그의 핵심입니다.
1바이트 차이가 전체 파싱을 망친 버그
처음에 형식 문자열을 이렇게 짰습니다.
# 잘못된 코드
struct.pack(">BBBBHBH",
0x01, 0x01, area_code, start_addr_high, start_addr_low, 0x00, count
)">BBBBHBH" 와 ">BBBHBH" 의 차이가 보이시나요?
B가 하나 더 있습니다. 이게 무슨 의미냐면:
올바른 ">BBBHBH":
B → 1바이트 (MRC = 0x01)
B → 1바이트 (SRC = 0x01)
B → 1바이트 (area_code)
H → 2바이트 (start_addr, big-endian 16비트)
B → 1바이트 (bit_pos = 0x00)
H → 2바이트 (count, big-endian 16비트)
합계: 8바이트
잘못된 ">BBBBHBH":
B → 1바이트 (MRC = 0x01)
B → 1바이트 (SRC = 0x01)
B → 1바이트 (area_code)
B → 1바이트 (start_addr의 상위 바이트만!)
H → 2바이트 (start_addr_low + bit_pos를 주소로 읽음)
B → 1바이트 (count의 상위 바이트만!)
H → 2바이트 (쓰레기 값)
합계: 9바이트파이프 공사 비유로 설명하면 이렇습니다.
10cm짜리 파이프 이음새에 11cm짜리를 끼우면 1cm가 삐져나옵니다.
그 뒤의 모든 이음새가 1cm씩 밀려서 전체 배관이 어긋나는 것과 같아요.
H(16비트 정수)가 받아야 할 2바이트 중 1바이트를 앞의 B가 가로챘습니다.
그래서 이후 모든 값이 1바이트씩 어긋났어요.
증상은 이랬습니다.
응답을 받는데 에러도 아니고, 이상한 주소의 데이터가 조금씩 나왔어요.
"왜 DM0을 요청했는데 엉뚱한 값이 나오지?"라고 한참 디버깅했습니다.
Area Code : 어느 메모리 영역을 읽을 것인가
FINS에서 메모리 영역 구분은 area code 한 바이트로 합니다.
AREA_CODES = {
"DM": 0x82, # DM 영역 워드 (CP1H, CJ2M, CS1D 공통)
"CIO_WORD": 0xB0, # CIO 영역 워드
"CIO_BIT": 0x30, # CIO 영역 비트
"WR_WORD": 0xB1, # WR (작업 릴레이) 워드
"WR_BIT": 0x31, # WR 비트
"HR_WORD": 0xB2, # HR (홀딩 릴레이) 워드
"HR_BIT": 0x32, # HR 비트
}주의할 점이 있습니다.
위 코드는 CP1H, CJ2M, CS1D 같은 현행 모델 기준이에요.
구형 CS1 일부 모델은 area code가 다를 수 있습니다.
실기로 확인하는 게 안전합니다.
전체 코드 구조
import asyncio
import struct
class FINSClient:
def __init__(self, host, port, timeout=3.0):
self.host = host
self.port = port
self.timeout = timeout
self._reader = None
self._writer = None
self._client_node = 0
self._server_node = 0
async def connect(self):
self._reader, self._writer = await asyncio.wait_for(
asyncio.open_connection(self.host, self.port),
timeout=self.timeout
)
await self._handshake()
async def _handshake(self):
# 20바이트 전송
self._writer.write(
b"FINS" + struct.pack(">IIII", 0xC, 0, 0, 0)
)
await self._writer.drain()
resp = await asyncio.wait_for(
self._reader.readexactly(24), timeout=self.timeout
)
self._client_node = resp[19]
self._server_node = resp[23]
async def read_dm(self, start_addr, count):
cmd = self._build_command(0x82, start_addr, count)
self._writer.write(cmd)
await self._writer.drain()
resp = await asyncio.wait_for(
self._reader.read(1024), timeout=self.timeout
)
return self._parse_response(resp, count)
def _build_command(self, area_code, start_addr, count):
# FINS 명령 데이터
fins_cmd = struct.pack(">BBBHBH",
0x01, 0x01, area_code, start_addr, 0x00, count
)
# FINS 헤더 10바이트 조립
fins_header = struct.pack(">BBBBBBBBBB",
0x80, # ICF
0x00, # RSV
0x02, # GCT
0x00, # DNA (목적지 네트워크)
self._server_node, # DA1 (목적지 노드)
0x00, # DA2
0x00, # SNA (출발지 네트워크)
self._client_node, # SA1 (출발지 노드)
0x00, # SA2
0x01, # SID (서비스 ID)
)
data = fins_header + fins_cmd
# FINS/TCP 헤더 16바이트
tcp_header = (
b"FINS"
+ struct.pack(">III",
len(data) + 4, # length
0x00000002, # command: FINS command
0x00000000, # error code
)
)
return tcp_header + data
def _parse_response(self, resp, count):
# 헤더 건너뛰기: TCP(16) + FINS헤더(10) + MRC/SRC(2) + 종료코드(2) = 30
payload = resp[30:]
values = []
for i in range(count):
word = struct.unpack(">H", payload[i*2:(i+1)*2])[0]
values.append(word)
return values한 줄 정리 : 바이너리 프로토콜 직접 구현할 때
처음 바이너리 프로토콜을 직접 구현할 때는 이 순서로 접근하면 됩니다.
1. 스펙 문서에서 패킷 구조 확인
2. struct 형식 문자열 작성
3. 형식 문자열의 총 바이트 수를 스펙과 대조
4. 소켓으로 보내고 Wireshark나 로그로 실제 패킷 확인
5. 응답 파싱 오프셋 계산 (헤더 몇 바이트 건너뛰어야 하나)struct 형식 문자열 작성 후, 총 바이트 수를 반드시 스펙과 대조하세요.
struct.calcsize(">BBBHBH") 로 바이트 수를 확인할 수 있어요.
스펙의 패킷 크기와 다르면 형식 문자열이 틀린 겁니다.
마치며
FINS/TCP 구현 자체는 복잡하지 않았습니다.
핸드쉐이크 20바이트, 명령 헤더 10바이트, 읽기 명령 8바이트.
각각 스펙 보면서 struct.pack으로 조립하면 돼요.
어려운 건 1바이트 오프셋 버그처럼 응답 자체는 오는데 값이 이상한 케이스입니다.
에러 코드도 없고 연결도 됐는데 값이 기묘할 때, struct 형식 문자열의 바이트 수를 먼저 확인해보세요.
Omron FINS/TCP를 Python으로 구현하려는 분들에게 이 글이 시작점이 됐으면 합니다.
