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

Python凭借其简洁的语法和丰富的第三方库，已成为産業用IoT、Data Acquisition、自動化测试领域的人気语言。pymodbus是Python生态中最流行的Modbus通讯库，サポートModbus TCP、RTU、ASCII三种协议，可作为クライアント側は和サーバー側。。本記事详细介绍pymodbus的インストール設定、连接管理、コイル読み書き、レジスタ読み書き、浮動小数点数処理、异步プログラミング以及完全的项目サンプル。

一、pymodbus库介绍

pymodbus是一个功能完全的Python Modbusプロトコル実装，由Galen Collins发起维护，目前已更新到3.x版本。它具有以下特点：

• 协议包括的：サポートModbus TCP、RTU、ASCII、UDP，以及TLS加密
• 主从兼备：既可作为クライアント側は（master），也可作为サーバー側。（slave）
• 同步异步：同时サポート同步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，Recommended for use

本記事以pymodbus 3.x版本为例，2.x版本API略有差异。

三、基础连接与读書き込み操作。

3.1 建立TCPconnect

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 リードコイルを読む（function code01）

# 読み取り住所から。0始まりました。8个コイルの状態
result = client.read_coils(address=0, count=8, slave=SLAVE_ID)

if not result.isError():
    # result.bits是一个リスト，包含読み取り的コイルの状態
    print("Coil status：")
    for i, bit in enumerate(result.bits[:8]):
        print(f"  コイル{i}：{'吸合' if bit else '断开'}")
else:
    print(f"Read failed：{result}")

# 方法参数説明：
# address: starting address（0-based）
# count: Reading quantity（1-2000）
# slave: slave address（Unit ID）

3.3 read discrete inputs（function code02）

# 読み取り住所から。0始まりました。4個別入力です。
result = client.read_discrete_inputs(address=0, count=4, slave=SLAVE_ID)

if not result.isError():
    for i, bit in enumerate(result.bits[:4]):
        print(f"入力{i}：{'高电平' if bit else '低电平'}")

3.4 read holding registers（function code03）

# 読み取り住所から。0始まりました。10レジスタを保持する。
result = client.read_holding_registers(address=0, count=10, slave=SLAVE_ID)

if not result.isError():
    print("レジスタを保持する。：")
    for i, reg in enumerate(result.registers):
        print(f"  register{i}：{reg}")
else:
    print(f"Read failed：{result}")

# result.registers是intリスト，每个元素为16ビット符号なし整数（0-65535）
# 如需記号付き。整数：value = reg - 65536 if reg >= 32768 else reg

3.5 read input registers（function code04）

# 読み取りセンサーはデータ（temperature、湿度等）
result = client.read_input_registers(address=0, count=2, slave=SLAVE_ID)

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

3.6 write single coil（function code05）

# 控制アドレス0的コイル吸合
result = client.write_coil(address=0, value=True, slave=SLAVE_ID)

if not result.isError():
    print("コイル吸合成功")
else:
    print(f"写入失敗：{result}")

# 控制アドレス0的コイル断开
client.write_coil(address=0, value=False, slave=SLAVE_ID)

3.7 write single register（function code06）

# 向アドレス0写入数值1234
result = client.write_register(address=0, value=1234, slave=SLAVE_ID)

if not result.isError():
    print("レジスタ写入成功")

# 写入記号付き。数（-100）
signed_value = -100
unsigned_value = signed_value & 0xFFFF  # 转换为符号なし。
client.write_register(address=0, value=unsigned_value, slave=SLAVE_ID)

3.8 write multiple coils（function code15）

# 同时控制前41つのコイル
values = [True, False, True, False]
result = client.write_coils(address=0, values=values, slave=SLAVE_ID)

if not result.isError():
    print("バッチ書き込みコイル成功")

3.9 write multiple registers（function code16）

# 向アドレス0开始写入31つのレジスタ
values = [100, 200, 300]
result = client.write_registers(address=0, values=values, slave=SLAVE_ID)

if not result.isError():
    print("一括書き込みき込みレジスタ成功")

四、浮動小数点数与32位データの処理

Modbusプロトコル传输的是16位レジスタ，32ビット浮動小数点数需要占用2つ连续的レジスタ。pymodbus提供了BinaryPayloadDecoder和BinaryPayloadBuilder工具类来処理各种データ型。

4.1 read32ビット浮動小数点数

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

# read21つのレジスタ（32ビット浮動小数点数）
result = client.read_holding_registers(address=10, count=2, slave=SLAVE_ID)

if not result.isError():
    # 作成解码器，指定バイトオーダー
    # Endian.Big = 前の高さ。（ABCD）
    # Endian.Little = 低バイト数前に（CDAB）
    decoder = BinaryPayloadDecoder.fromRegisters(
        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"register value：{registers}")  # 出力类似 [16973, 52429]

# 写入デバイス
client.write_registers(address=10, values=registers, slave=SLAVE_ID)

4.3 読み取りストリング·ストリング

# 読み取りストリング·ストリング（すべてのレジスタ。存2个文字，read101つのレジスタ=20文字）
result = client.read_holding_registers(address=0, count=10, slave=SLAVE_ID)

if not result.isError():
    decoder = BinaryPayloadDecoder.fromRegisters(
        result.registers,
        byteorder=Endian.Big
    )
    # 解码20バイトストリング·ストリング
    str_value = decoder.decode_string(20).decode('ascii').strip('\x00')
    print(f"ストリング·ストリング：{str_v)

4. 4 바 이트 순 서 설명

			endianness
			ABCD（ビッグエンディアン）
			CDAB（Word exchange）
			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 , address , count):
        ' ' ' ' 읽기 유지 레 지 스터 (재 시 도 포함) ' ' ' '
        for attempt in range (3) :
            시도 하십시오 :
                self.client.read result = 
                    address = address , 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 , address):
        "" 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 , address , 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 , address , values):
        \ "
        for attempt in range (3) :
            시도 하십시오 :
                result = self . client . write _ register s (
                    address = address , 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 RuntimeError(f"写入レジスタ失敗，address：{address}")
    
    def read_coils(self, address, count):
        """Read coil status"""
        result = self.client.read_coils(
            address=address, count=count, slave=self.slave_id
        )
        if result.isError():
            raise RuntimeError(f"リードコイルを読む失敗：{result}")
        return result.bits[:count]
    
    def write_coil(self, address, value):
        """write single coil"""
        result = self.client.write_coil(
            address=address, value=value, slave=self.slave_id
        )
        if result.isError():
            raise RuntimeError(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"temperature：{temp:.2f}℃")
        
        # read101つのレジスタ
        regs = mb.read_registers(0, 10)
        print(f"register：{regs}")
        
        # 控制继电器
        mb.write_coil(0, True)  # 吸合
        print("Relay1已吸合")
        
        # 読み取り继电器ステータス
        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)
    
    # connect
    await client.connect()
    
    # 异步読み取り
    result = await client.read_holding_registers(0, 10, slave=1)
    if not result.isError():
        print(f"register：{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"device{i+1}：{regs}")
    
    client.close()

asyncio.run(main())

八、Modbusサーバー側。（slave）example

from pymodbus.server import StartTcpServer
from pymodbus.datastore import (
    ModbusServerContext,
    ModbusSlaveContext,
    ModbusSequentialDataBlock
)

# 作成データ块
# address0开始，初期値は全为0
coils = ModbusSequentialDataBlock(0, [False] * 100)
discrete_inputs = ModbusSequentialDataBlock(0, [False] * 100)
holding_registers = ModbusSequentialDataBlock(0, [0] * 100)
input_registers = ModbusSequentialDataBlock(0, [0] * 100)

# 設定一些初期値は
holding_registers.setValues(0, [100, 200, 300])  # register0-2
coils.setValues(0, [True, False, True])           # コイル0-2

# 作成スレーブContextより
slave_context = ModbusSlaveContext(
    di=discrete_inputs,
    co=coils,
    hr=holding_registers,
    ir=input_registers,
    unit=1  # slave address
)

# 作成サーバー·サーバーContextより
server_context = ModbusServerContext(slaves=slave_context, single=True)

# 启动TCPサーバー側。
print("Modbus TCPサーバー側。启动，监听ポート502...")
StartTcpServer(
    context=server_context,
    address=("0.0.0.0", 502)
)

九、实际应用：继电器モジュール控制

import time

class RelayController:
    """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

• 使用Contextより管理器
• バッチ読み取り
• 合理設定タイムアウト
• 追加重试机制
• 控制ポーリング频率
• 異常処理
• ジャーナル·レコード
• 多デバイス管理

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

			库名
			pymodbus
			modbus-tk
			pymodbusTCP
			uModbus
			minimalmodbus

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