Building Protocol Bridges: Modbus to MQTT

Legacy integration patterns that actually work in production.

The Reality of Industrial Networks

Despite the push for modern protocols, 70% of industrial devices still communicate via legacy protocols like Modbus, BACnet, or proprietary serial interfaces. Protocol bridges enable these devices to participate in modern IoT architectures.

Modbus Protocol Overview

Register Types

Common Function Codes

1MODBUS_FUNCTIONS = {
2    0x01: 'Read Coils',
3    0x02: 'Read Discrete Inputs',
4    0x03: 'Read Holding Registers',
5    0x04: 'Read Input Registers',
6    0x05: 'Write Single Coil',
7    0x06: 'Write Single Register',
8    0x0F: 'Write Multiple Coils',
9    0x10: 'Write Multiple Registers',
10}

Bridge Architecture

┌─────────────────────────────────────────────────────────────┐
│                    Protocol Bridge                          │
│  ┌───────────────┐  ┌─────────────┐  ┌───────────────────┐  │
│  │ Modbus Client │→ │ Data Mapper │→ │ MQTT Publisher    │  │
│  │ (RS-485/TCP)  │  │             │  │                   │  │
│  └───────────────┘  └─────────────┘  └───────────────────┘  │
│          ↑                                      ↓           │
│  ┌───────────────┐                   ┌───────────────────┐  │
│  │ Poll Scheduler│                   │ Command Handler   │  │
│  └───────────────┘                   └───────────────────┘  │
└─────────────────────────────────────────────────────────────┘
         ↓                                        ↑
┌─────────────────┐                    ┌─────────────────────┐
│ Modbus Devices  │                    │ MQTT Broker         │
│ (PLC, VFD, etc) │                    │ (Cloud/Edge)        │
└─────────────────┘                    └─────────────────────┘

Implementation

Configuration-Driven Mapping

1# bridge-config.yaml
2devices:
3  - name: "pump-station-1"
4    connection:
5      type: "modbus-tcp"
6      host: "192.168.1.100"
7      port: 502
8      unit_id: 1
9    
10    registers:
11      - name: "flow_rate"
12        address: 40001
13        type: "float32"
14        scale: 0.1
15        unit: "m3/h"
16        poll_interval: 1000
17        
18      - name: "pressure"
19        address: 40003
20        type: "int16"
21        scale: 0.01
22        unit: "bar"
23        poll_interval: 1000
24        
25      - name: "motor_speed"
26        address: 40005
27        type: "uint16"
28        unit: "rpm"
29        poll_interval: 5000
30        writable: true
31    
32    mqtt:
33      topic_prefix: "plant/pump-station-1"
34      qos: 1

Bridge Core

1from pymodbus.client import ModbusTcpClient
2import paho.mqtt.client as mqtt
3import asyncio
4import struct
5
6class ModbusMQTTBridge:
7    def __init__(self, config):
8        self.config = config
9        self.modbus = ModbusTcpClient(
10            config['connection']['host'],
11            port=config['connection']['port']
12        )
13        self.mqtt = mqtt.Client()
14        
15    async def start(self):
16        self.modbus.connect()
17        self.mqtt.connect(self.config['mqtt']['broker'])
18        self.mqtt.loop_start()
19        
20        # Subscribe to command topics
21        for reg in self.config['registers']:
22            if reg.get('writable'):
23                topic = f"{self.config['mqtt']['topic_prefix']}/{reg['name']}/set"
24                self.mqtt.subscribe(topic)
25        
26        self.mqtt.on_message = self.handle_command
27        
28        # Start polling
29        await self.poll_loop()
30    
31    async def poll_loop(self):
32        while True:
33            for register in self.config['registers']:
34                value = await self.read_register(register)
35                if value is not None:
36                    await self.publish_value(register, value)
37            
38            await asyncio.sleep(0.1)  # Minimum poll interval
39    
40    async def read_register(self, register):
41        address = register['address'] - 40001  # Convert to 0-based
42        
43        if register['type'] == 'float32':
44            result = self.modbus.read_holding_registers(address, 2)
45            if result.isError():
46                return None
47            raw = struct.pack('>HH', *result.registers)
48            value = struct.unpack('>f', raw)[0]
49        elif register['type'] in ['int16', 'uint16']:
50            result = self.modbus.read_holding_registers(address, 1)
51            if result.isError():
52                return None
53            value = result.registers[0]
54        
55        return value * register.get('scale', 1)
56    
57    async def publish_value(self, register, value):
58        topic = f"{self.config['mqtt']['topic_prefix']}/{register['name']}"
59        payload = {
60            'value': value,
61            'unit': register.get('unit', ''),
62            'timestamp': time.time()
63        }
64        self.mqtt.publish(topic, json.dumps(payload), qos=1)
65    
66    def handle_command(self, client, userdata, message):
67        # Parse topic to get register name
68        parts = message.topic.split('/')
69        register_name = parts[-2]
70        
71        register = next(
72            (r for r in self.config['registers'] if r['name'] == register_name),
73            None
74        )
75        
76        if register and register.get('writable'):
77            value = json.loads(message.payload)['value']
78            self.write_register(register, value)

Production Considerations

Error Handling

1class ResilientBridge:
2    def __init__(self):
3        self.retry_count = 0
4        self.max_retries = 5
5        self.backoff_base = 1.0
6        
7    async def read_with_retry(self, register):
8        while self.retry_count < self.max_retries:
9            try:
10                return await self.read_register(register)
11            except ModbusException as e:
12                self.retry_count += 1
13                wait_time = self.backoff_base * (2 ** self.retry_count)
14                logger.warning(f"Modbus error, retry {self.retry_count} in {wait_time}s")
15                await asyncio.sleep(wait_time)
16        
17        logger.error("Max retries exceeded, reconnecting...")
18        await self.reconnect()

Metrics & Monitoring

Track bridge health:

• Poll success rate
• Average response time
• Message publish rate
• Connection uptime

Protocol bridges unlock the value trapped in legacy industrial systems, enabling modern analytics and automation without costly equipment replacement.