Python 에서 Modbus TCP 통신 프로그램을 작성하는 예제 코드

Python凭借其简洁的语法和丰富的第三方库，已成为工业物联网、데이터采集、自动化测试领域的热门언어。pymodbus是Python生态中最流行的Modbus通讯库，支持Modbus TCP、RTU、ASCII三种프로토콜，可作为클라이언트和服务端。本文详细介绍pymodbus的安装설정、연결管理、코일读写、레지스터读写、浮点数处理、异步编程以及完整的项目예제。

一、pymodbus库介绍

pymodbus是一个功能完整的Python Modbus프로토콜实现，由Galen Collins发起维护，目前已更新到3.x버전。它具有以下特点：

• 프로토콜全面：支持Modbus TCP、RTU、ASCII、UDP，以及TLS加密
• 主从兼备：既可作为클라이언트（마스터），也可作为服务端（슬레이브）
• 同步异步：同时支持同步API和基于asyncio的异步API
• 功能完整：支持所有표준기능 코드（01-16、22、43等）
• 跨平台：Windows、Linux、Mac OS均可运行
• 易于扩展：支持自定义기능 코드、自定义데이터块
• 문서丰富：官方문서完善，社区活跃

二、安装与환경설정

2.1 pip安装

# 安装最신버전本
pip install pymodbus

# 安装指定버전
pip install pymodbus==3.6.9

# 安装시리얼 포트支持（RTU모드需要）
pip install pyserial

# 安装异步支持（3.x버전已内置asyncio）
pip install pymodbus[asynchronous]

# 验证安装
python -c "import pymodbus; print(pymodbus.__version__)"

2.2 버전差异설명

버전	导入방식	特点
pymodbus 2.x	from pymodbus.client.sync import ModbusTcpClient	经典버전，同步API
pymodbus 3.x	from pymodbus.client import ModbusTcpClient	重构버전，统一API，推荐使用

本文以pymodbus 3.x버전为例，2.x버전API略有差异。

三、基础연결与读写操作

3.1 建立TCP연결

from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ModbusException

# 创建클라이언트实例
client = ModbusTcpClient(
    host='192.168.1.100',
    port=502,
    timeout=3.0,        # 超时时间（秒）
    retries=3,          # 重试次数
    retry_on_empty=True # 空响应时重试
)

# 建立연결
if client.connect():
    print("연결成功!")
else:
    print("연결失败!")
    exit(1)

# 设置슬레이브 주소（Unit ID）
SLAVE_ID = 1

try:
    # 业务逻辑...
    pass
except ModbusException as e:
    print(f"Modbus异常：{e}")
finally:
    # 닫기연결
    client.close()
    print("연결已닫기")

3.2 读코일（기능 코드01）

# 읽기从주소0开始的8个코일 상태
result = client.read_coils(주소=0, count=8, slave=SLAVE_ID)

if not result.is오답():
    # result.bits是一个列表，包含읽기的코일 상태
    print("코일 상태：")
    for i, bit in enumerate(result.bits[:8]):
        print(f"  코일{i}：{'吸合' if bit else '断开'}")
else:
    print(f"읽기失败：{result}")

# 方法매개변수설명：
# 주소: 시작주소（0-based）
# count: 읽을 수량（1-2000）
# slave: 슬레이브 주소（Unit ID）

3.3 读이산 입력（기능 코드02）

# 읽기从주소0开始的4个이산 입력
result = client.read_discrete_inputs(주소=0, count=4, slave=SLAVE_ID)

if not result.is오답():
    for i, bit in enumerate(result.bits[:4]):
        print(f"입력{i}：{'高电平' if bit else '低电平'}")

3.4 读유지 레지스터（기능 코드03）

# 읽기从주소0开始的10个유지 레지스터
result = client.read_holding_registers(주소=0, count=10, slave=SLAVE_ID)

if not result.is오답():
    print("유지 레지스터：")
    for i, reg in enumerate(result.registers):
        print(f"  레지스터{i}：{reg}")
else:
    print(f"읽기失败：{result}")

# result.registers是int列表，每个元素为16位无符号整数（0-65535）
# 如需有符号整数：value = reg - 65536 if reg >= 32768 else reg

3.5 读입력 레지스터（기능 코드04）

# 읽기센서데이터（온도、湿度等）
result = client.read_input_registers(주소=0, count=2, slave=SLAVE_ID)

if not result.is오답():
    temperature = result.registers[0] / 10.0  # 假设온도放大10倍
    humidity = result.registers[1] / 10.0
    print(f"온도：{temperature}℃")
    print(f"湿度：{humidity}%RH")

3.6 写单个코일（기능 코드05）

# 控制주소0的코일吸合
result = client.write_coil(주소=0, value=True, slave=SLAVE_ID)

if not result.is오답():
    print("코일吸合成功")
else:
    print(f"쓰기失败：{result}")

# 控制주소0的코일断开
client.write_coil(주소=0, value=False, slave=SLAVE_ID)

3.7 写单个레지스터（기능 코드06）

# 向주소0쓰기数值1234
result = client.write_register(주소=0, value=1234, slave=SLAVE_ID)

if not result.is오답():
    print("레지스터쓰기成功")

# 쓰기有符号数（-100）
signed_value = -100
unsigned_value = signed_value & 0xFFFF  # 转换为无符号
client.write_register(주소=0, value=unsigned_value, slave=SLAVE_ID)

3.8 写多个코일（기능 코드15）

# 同时控制前4个코일
values = [True, False, True, False]
result = client.write_coils(주소=0, values=values, slave=SLAVE_ID)

if not result.is오답():
    print("批量写코일成功")

3.9 写多个레지스터（기능 코드16）

# 向주소0开始쓰기3个레지스터
values = [100, 200, 300]
result = client.write_registers(주소=0, values=values, slave=SLAVE_ID)

if not result.is오답():
    print("批量写레지스터成功")

四、浮点数与32位데이터处理

Modbus프로토콜传输的是16位레지스터，32位浮点数需要占用两个连续的레지스터。pymodbus提供了BinaryPayloadDecoder和BinaryPayloadBuilder도구类来处理各种데이터유형。

4.1 읽기32位浮点数

from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian

# 읽기2个레지스터（32位浮点数）
result = client.read_holding_registers(주소=10, count=2, slave=SLAVE_ID)

if not result.is오답():
    # 创建解码器，指定바이트序
    # Endian.Big = 高바이트在前（ABCD）
    # Endian.Little = 低바이트在前（CDAB）
    decoder = BinaryPayloadDecoder.from회원가입s(
        result.registers,
        byteorder=Endian.Big,
        wordorder=Endian.Big
    )
    
    # 解码为32位浮点数
    float_value = decoder.decode_32bit_float()
    print(f"浮点数值：{float_value}")
    
    # 也可以解码为32位整数
    # int_value = decoder.decode_32bit_int()
    # long_value = decoder.decode_64bit_int()

4.2 쓰기32位浮点数

from pymodbus.payload import BinaryPayloadBuilder

# 创建构建器
builder = BinaryPayloadBuilder(
    byteorder=Endian.Big,
    wordorder=Endian.Big
)

# 추가32位浮点数
builder.add_32bit_float(25.5)

# 转换为레지스터列表
registers = builder.to_registers()
print(f"레지스터 값：{registers}")  # 출력类似 [16973, 52429]

# 쓰기设备
client.write_registers(주소=10, values=registers, slave=SLAVE_ID)

4.3 읽기문자串

# 읽기문자串（每个레지스터存2个문자，읽기10个레지스터=20문자）
result = client.read_holding_registers(주소=0, count=10, slave=SLAVE_ID)

if not result.is오답():
    decoder = BinaryPayloadDecoder.from회원가입s(
        result.registers,
        byteorder=Endian.Big
    )
    # 解码20바이트문자串
    str_value = decoder.decode_string(20).decode('ascii').strip('\x00')
    print(f"문자串：{str_v)

4. 4 바 이트 순 서 설명

			바이트序
			ABCD（大端）
			CDAB（字交换）
			BADC（바이트交换）
			DCBA（小端）

Mod bus R TU 직 렬 통신

p ym od bus . client 에서 Mod bus S erial C li ent 가져 오 기

# R TU 클 라이언 트 만들기
클 라이언 트 = Mod bus S erial C li ent (
    port = '/ dev / tt y US B 0 ',    # 직 렬 포 트 장치 (Windows 에서 ' COM 3 ')
    보 트 레이 트 = 96 00          # 보 트 속도
    비트 크기 = 8,             # 데이터 비트
    par ity = ' N ',             # 체크 비 트 : N = 없음 , E = 偶 , O = 홀 수
    stop bit = 1,             # 정 지 비트
    시간 초 과 = 3. 0            # 시간 초 과
    재 시 도 = 3
)

if client . connect ():
    print (" R TU 연결 이 성공 했습니다 ")
    
    # 읽기 및 쓰 기 작업 은 TCP 와 동일 합니다 .
    client.read result = 
    if not result . is Er ror ():
        print (res ult . register s)
    
    client . close ()

6. 완전한 도구 패키 지

p ym od bus . client 에서 가져 오 기 Mod bus T c p C li ent
p ym od bus . pay load 가져 오 기 Bin ary Pay load Dec oder , Bin ary Pay load Bu ilder 에서
p ym od bus . const ants import 엔 디 안 에서
from p ym od bus . excep tions import Mod bus Ex ception
로 깅 가져 오 기

log ging . basic Con fig (level = log ging . IN FO)
log ger = log ging . get L og ger (__ name __)

클래 스 Mod bus T c p U til :
    "" Mod bus TCP 도구 클래 스 , 일반 작업 캡 슐 화 ""
    
    __ def init __ (self , host , port = 50 2, slave _ id = 1, time out = 3. 0):
        self.host
        self . port = port
        self . slave _ id = slave _ id
        self . client = Mod bus T c p C li ent (host = host , port = port , time out = time out)
        self . connected = Fal se
    
    def connect (self) :
        "" "" " 연결 설정 (재 시 도 포함) "" "
        for attempt in range (3) :
            if self . client . connect () :
                self . connected = True
                logger.info
                return True True
            log ger . warning (f " 연 결 실패 , 재 시 도 {att emp t + 1} / 3 ")
        log ger . error (" 연 결 에 실패 함 ")
        return false false
    
    def dis conne ct (self):
        ""
        if self . connected :
            self . client . close ()
            self . connected = Fal se
            logger.info
    
    def read _ register s (self , 주소 , count):
        ' ' ' ' 읽기 유지 레 지 스터 (재 시 도 포함) ' ' ' '
        for attempt in range (3) :
            시도 하십시오 :
                self.client.read result = 
                    주소 = 주소 , count = count , slave = self . slave _ id
                )
                if not result . is Er ror ():
                    result . register s 를 반환
                log ger . warning (f " 읽 기 오류 : {res ult} ")
            except Mod bus Ex ception 으로 e :
                log ger . warning (f " Mod bus 예외 : {e} ")
            self . _ recon ne ct ()
        raise 런 타 임 오류 (f " 레 지 스터 읽기 실패 , 주소 : {add ress } ")
    
    def read _ flo at (self , 주소):
        "" 32 비트 부동 소 수 점 숫자 읽기 ""
        self.read regs = 
        디 코 더 = Bin ary Pay load Dec oder . from Reg ister s (
            reg s , by te order = En dian . Big , word order = En dian . Big
        )
        return dec oder . dec ode _ 32 bit _ flo at ()
    
    def write _ flo at (self , 주소 , value):
        "" 32 비트 부동 소 수 점 숫자 쓰 기 ""
        바이 너 리 Pay load Bu ilder (
            by te order = En dian . Big , word order = En dian . Big
        )
        bu ilder . add _ 32 bit _ flo at (val ue)
        builder.to regs = 
        return self . write _ register s (add ress , reg s)
    
    def write _ register s (self , 주소 , values):
        \ "
        for attempt in range (3) :
            시도 하십시오 :
                result = self . client . write _ register s (
                    주소 = 주소 , values = values , slave = self . slave _ id
                )
                if not result . is Er ror():
                    return True
            except ModbusException as e:
                logger.warning(f"쓰기异常：{e}")
            self._reconnect()
        raise Runtime오답(f"쓰기레지스터失败，주소：{주소}")
    
    def read_coils(self, 주소, count):
        """코일 읽기 (FC01)"""
        result = self.client.read_coils(
            주소=주소, count=count, slave=self.slave_id
        )
        if result.is오답():
            raise Runtime오답(f"读코일失败：{result}")
        return result.bits[:count]
    
    def write_coil(self, 주소, value):
        """写单个코일"""
        result = self.client.write_coil(
            주소=주소, value=value, slave=self.slave_id
        )
        if result.is오답():
            raise Runtime오답(f"写코일失败：{result}")
        return True
    
    def _reconnect(self):
        """重新연결"""
        self.disconnect()
        self.connect()
    
    def __enter__(self):
        self.connect()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.disconnect()


# 使用예제
if __name__ == '__main__':
    with ModbusTcpUtil('192.168.1.100', slave_id=1) as mb:
        # 읽기온도（浮点数）
        temp = mb.read_float(10)
        print(f"온도：{temp:.2f}℃")
        
        # 읽기10个레지스터
        regs = mb.read_registers(0, 10)
        print(f"레지스터：{regs}")
        
        # 控制릴레이
        mb.write_coil(0, True)  # 吸合
        print("릴레이1已吸合")
        
        # 읽기릴레이状态
        coils = mb.read_coils(0, 8)
        print(f"릴레이状态：{coils}")

七、异步编程（asyncio）

import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def main():
    # 创建异步클라이언트
    client = AsyncModbusTcpClient('192.168.1.100', port=502)
    
    # 연결
    await client.connect()
    
    # 异步읽기
    result = await client.read_holding_registers(0, 10, slave=1)
    if not result.is오답():
        print(f"레지스터：{result.registers}")
    
    # 异步쓰기
    await client.write_coil(0, True, slave=1)
    
    # 并发읽기多个设备
    async def read_device(host):
        c = AsyncModbusTcpClient(host)
        await c.connect()
        r = await c.read_holding_registers(0, 5, slave=1)
        c.close()
        return r.registers
    
    # 同时읽기3个设备
    tasks = [read_device(f'192.168.1.{i}') for i in range(100, 103)]
    results = await asyncio.gather(*tasks)
    for i, regs in enumerate(results):
        print(f"设备{i+1}：{regs}")
    
    client.close()

asyncio.run(main())

八、Modbus服务端（슬레이브）예제

from pymodbus.server import StartTcpServer
from pymodbus.데이터store import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequential데이터Block
)

# 创建데이터块
# 주소0开始，初始值全为0
coils = ModbusSequential데이터Block(0, [False] * 100)
discrete_inputs = ModbusSequential데이터Block(0, [False] * 100)
holding_registers = ModbusSequential데이터Block(0, [0] * 100)
input_registers = ModbusSequential데이터Block(0, [0] * 100)

# 设置一些初始值
holding_registers.setValues(0, [100, 200, 300])  # 레지스터0-2
coils.setValues(0, [True, False, True])           # 코일0-2

# 创建슬레이브上下文
slave_context = ModbusSlaveContext(
    di=discrete_inputs,
    co=coils,
    hr=holding_registers,
    ir=input_registers,
    unit=1  # 슬레이브 주소
)

# 创建服务器上下文
server_context = ModbusServerContext(slaves=slave_context, single=True)

# 启动TCP服务端
print("Modbus TCP服务端启动，监听포트502...")
StartTcpServer(
    context=server_context,
    주소=("0.0.0.0", 502)
)

九、实际적용：릴레이模块控制

import time

class Relay컨트롤러:
    """8路네트워크릴레이제어기"""
    
    def __init__(self, host, slave_id=1):
        self.mb = ModbusTcpUtil(host, slave_id=slave_id)
        self.mb.connect()
    
    def get_all_status(self):
        """읽기所有继电장치 상태 "" "
        self.mb.read coils = 
        리 턴 코 일 (ret urn co il)
    
    def turn _ on (self , channel):
        ' ' 지정 된 채널 열 기 (1 - 8) ' ' '
        self . mb . write _ co il (ch ann el - 1, True)
    
    def turn _ off (self , channel):
        ' ' 지정 된 채널 닫 기 ' '
        self . mb . write _ co il (ch ann el - 1, Fal se)
    
    def tog gle (self , channel):
        ' ' 지정 된 채널 상태 전환 ' ' '
        self.mb.read coils = 
        self . mb . write _ co il (ch ann el - 1, co ils [ 0 ])
    
    def all _ on (self):
        " " 모 두 열 기 " "
        self . mb . client . write _ co ils (0, [ T rue ] * 8, slave = self . mb . slave _ id)
    
    def all _ off (self):
        " " 모두 닫 기 " "
        self . mb . client . write _ co ils (0, [ F alse ] * 8, slave = self . mb . slave _ id)
    
    def mar que e (self , delay = 0. 2):
        ' ' '
        for i in range (8) :
            self . turn _ on (i + 1)
            sleep (de lay)
            self . turn _ off (i + 1)
    
    def close (self) :
        self . mb . dis conne ct ()


# 사용 예 제
if __ name __ == ' __ main __ ':
    re lay = Rela y Con trol ler (' 19 2. 16 8. 1. 100 ')
    
    # 현재 상태 읽기
    status = re lay . get _ all _ status ()
    print ( " 현 재 상태 : ", [" ON " if s else " O FF " for s in status ")
    
    # 조 명 테스트
    print (" 램 프 테스트 ... ")
    re lay . mar que e (de lay = 0. 3)
    
    # 순 차 적으로 제어
    re lay . turn _ on (1)
    시간 . 잠 들 기 (0. 5)
    re lay . turn _ on (2)
    시간 . sleep (0. 5)
    re lay . turn _ off (1)
    시간 . sleep (0. 5)
    re lay . turn _ off (2)
    
    re lay . close ()

일반적인 질문 과 모 범 사례

10. 1 자주 묻는 질문

• 연결超时
• 返回异常响应
• 浮点数분석错误
• RTU시리얼 포트打不开
• 데이터읽기为0
• 频繁断连

10. 2 Best Pra cti ces

• 使用上下文管理器
• 批量읽기
• 合理设置超时
• 추가重试机制
• 控制轮询频率
• 异常处理
• 로그记录
• 多设备管理

다른 Python Mod bus 라이 브 러 리 비교

			库名
			pymodbus
			modbus-tk
			pymodbusTCP
			uModbus
			minimalmodbus

Python Mod bus 개발 을위한 선호 라이 브 러 리 인 p ym od bus 는 완벽한 기능 과 활 발 한 커뮤니티 를 통해 산업 사 물 인터넷 , 데이터 수집 및 자동 화 테스트 에서 널리 사용됩니다 .이 문서 에서 제공 되는 도구 클래 스는 프로젝트 개발 에 직접 사용할 수 있으며 , Mod bus Poll 과 같은 도구를 사용하여 장치 통신 을 확인 한 다음 Python 코드를 실제 프로젝트에 통합 하는 것이 좋습니다 .