How to Collect and Visualize Energy Meter Data from Your Factory Floor

System Architecture

The stack is built around a Raspberry Pi acting as an edge gateway that bridges the RS-485 serial world of the energy meters to the IP world of the monitoring stack. All components run on the same Pi for compact deployments, or the data storage and visualisation components can be moved to a server for larger installations with higher data volumes.

Hardware Setup

The physical connection between the Raspberry Pi and the energy meters uses a USB to RS-485 adapter. This is a low-cost converter that plugs into the Pi's USB port and exposes a serial interface (/dev/ttyUSB0) that Python can communicate with using the Modbus RTU protocol.

RS-485 wiring

RS-485 uses a two-wire differential pair. Connect A+ on the adapter to A+ on the energy meter, and B- to B-. If you have multiple meters on the same RS-485 bus, daisy-chain the A+ and B- wires from meter to meter. Each meter must have a unique Modbus slave address configured on its DIP switches or front panel. Add a 120-ohm termination resistor at the far end of the bus if the total cable run exceeds 10 metres.

Connect the USB adapter to the Raspberry Pi and confirm it is recognised:

ls /dev/ttyUSB*
# Expected: /dev/ttyUSB0

Python Modbus RTU Reader

The Python reader script polls the energy meter registers on a timer and converts raw register values to engineering units. The register map varies by meter model. The example below uses the Eastron SDM120 register map, which is typical for single-phase DIN-rail meters widely used in European factories. Adjust the register addresses and scaling factors for your specific meter model by consulting its datasheet.

pip install pymodbus paho-mqtt --break-system-packages

#!/usr/bin/env python3
# energy_reader.py
# Polls Modbus RTU energy meter and publishes readings to MQTT

import time
import struct
import json
import logging
import paho.mqtt.client as mqtt
from pymodbus.client import ModbusSerialClient

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s'
)
logger = logging.getLogger(__name__)

# RS-485 serial port configuration
SERIAL_PORT  = '/dev/ttyUSB0'
BAUDRATE     = 9600
PARITY       = 'N'
STOPBITS     = 1
BYTESIZE     = 8
TIMEOUT      = 1

# Modbus slave address of the energy meter (set on the meter's DIP switches)
SLAVE_ADDRESS = 1

# MQTT broker configuration
MQTT_BROKER   = '127.0.0.1'
MQTT_PORT     = 1883
MQTT_USER     = 'factoryuser'
MQTT_PASSWORD = 'yourpassword'
MQTT_TOPIC    = 'factory/energy/panel_a/meter_01'

# Poll interval in seconds
POLL_INTERVAL = 10

# Eastron SDM120 register map (input registers, function code 04)
# Two consecutive 16-bit registers form one IEEE 754 float (big-endian)
REGISTERS = {
    'voltage_v':          0x0000,  # Volts
    'current_a':          0x0006,  # Amps
    'active_power_w':     0x000C,  # Watts
    'apparent_power_va':  0x0012,  # VA
    'reactive_power_var': 0x0018,  # VAr
    'power_factor':       0x001E,  # dimensionless
    'frequency_hz':       0x0046,  # Hz
    'import_energy_kwh':  0x0048,  # kWh (total imported)
    'export_energy_kwh':  0x004A,  # kWh (total exported)
    'total_energy_kwh':   0x0156,  # kWh (import + export)
}

def registers_to_float(high, low):
    """Convert two 16-bit Modbus registers to a 32-bit IEEE 754 float."""
    packed = struct.pack('>HH', high, low)
    return struct.unpack('>f', packed)[0]

def read_meter(client):
    """Read all configured registers from the energy meter."""
    readings = {}
    for name, address in REGISTERS.items():
        result = client.read_input_registers(address, count=2, slave=SLAVE_ADDRESS)
        if result.isError():
            logger.warning(f'Error reading register {name} at address {hex(address)}')
            readings[name] = None
        else:
            readings[name] = round(registers_to_float(
                result.registers[0], result.registers[1]
            ), 4)
    return readings

def main():
    client = ModbusSerialClient(
        method='rtu',
        port=SERIAL_PORT,
        baudrate=BAUDRATE,
        parity=PARITY,
        stopbits=STOPBITS,
        bytesize=BYTESIZE,
        timeout=TIMEOUT
    )

    mqttc = mqtt.Client()
    mqttc.username_pw_set(MQTT_USER, MQTT_PASSWORD)
    mqttc.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
    mqttc.loop_start()

    if not client.connect():
        logger.error('Could not connect to Modbus serial port')
        return

    logger.info(f'Connected to energy meter on {SERIAL_PORT}, slave address {SLAVE_ADDRESS}')

    while True:
        readings = read_meter(client)

        payload = {
            'timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
            'meter':     'meter_01',
            'panel':     'panel_a',
            **readings
        }

        mqttc.publish(MQTT_TOPIC, json.dumps(payload), qos=1)
        logger.info(f'Published: {payload}')

        time.sleep(POLL_INTERVAL)

if __name__ == '__main__':
    main()

Run the reader and confirm data is appearing in the MQTT broker:

python3 energy_reader.py &

# Verify MQTT messages are arriving
mosquitto_sub -h 127.0.0.1 -u factoryuser -P yourpassword -t factory/energy/#

You should see a JSON payload every 10 seconds containing all the energy parameters from the meter.

Install Mosquitto, InfluxDB, and Telegraf

# Mosquitto MQTT broker
sudo apt install -y mosquitto mosquitto-clients
sudo mosquitto_passwd -c /etc/mosquitto/passwd factoryuser
sudo tee /etc/mosquitto/conf.d/energy.conf <<'EOF'
listener 1883 127.0.0.1
allow_anonymous false
password_file /etc/mosquitto/passwd
EOF
sudo systemctl enable mosquitto && sudo systemctl restart mosquitto

# InfluxDB
curl https://repos.influxdata.com/influxdata-archive.key | \
  gpg --dearmor | sudo tee /usr/share/keyrings/influxdata-keyring.gpg > /dev/null

echo "deb [signed-by=/usr/share/keyrings/influxdata-keyring.gpg] \
  https://repos.influxdata.com/debian stable main" | \
  sudo tee /etc/apt/sources.list.d/influxdata.list

sudo apt update && sudo apt install -y influxdb2 telegraf
sudo systemctl enable influxdb && sudo systemctl start influxdb

influx setup \
  --username admin \
  --password yourpassword \
  --org factory \
  --bucket energy \
  --retention 365d \
  --force

sudo nano /etc/telegraf/telegraf.conf

[agent]
  interval = "10s"
  flush_interval = "10s"

[[inputs.mqtt_consumer]]
  servers          = ["tcp://127.0.0.1:1883"]
  topics           = ["factory/energy/#"]
  username         = "factoryuser"
  password         = "yourpassword"
  data_format      = "json"
  json_time_key    = "timestamp"
  json_time_format = "2006-01-02T15:04:05Z"
  tag_keys         = ["meter", "panel"]

[[outputs.influxdb_v2]]
  urls         = ["http://127.0.0.1:8086"]
  token        = "YOUR_INFLUXDB_API_TOKEN"
  organization = "factory"
  bucket       = "energy"

sudo systemctl enable telegraf && sudo systemctl start telegraf

Verify data is flowing into InfluxDB:

influx query 'from(bucket:"energy") |> range(start: -5m) |> limit(n:3)'

Install Grafana and Build the Energy Dashboard

sudo apt install -y apt-transport-https software-properties-common wget

sudo mkdir -p /etc/apt/keyrings/
wget -q -O - https://apt.grafana.com/gpg.key | gpg --dearmor | \
  sudo tee /etc/apt/keyrings/grafana.gpg > /dev/null

echo "deb [signed-by=/etc/apt/keyrings/grafana.gpg] \
  https://apt.grafana.com stable main" | \
  sudo tee /etc/apt/sources.list.d/grafana.list

sudo apt update && sudo apt install -y grafana
sudo systemctl enable grafana-server && sudo systemctl start grafana-server

Open http://localhost:3000, log in, and add InfluxDB as a data source under Connections → Data sources with Query Language set to Flux.

Useful Flux queries for the energy dashboard

Current active power (Stat panel, live value):

from(bucket: "energy")
  |> range(start: -1m)
  |> filter(fn: (r) => r["_measurement"] == "mqtt_consumer")
  |> filter(fn: (r) => r["meter"] == "meter_01")
  |> filter(fn: (r) => r["_field"] == "active_power_w")
  |> last()

Power trend over the last 8 hours (Time series panel):

from(bucket: "energy")
  |> range(start: -8h)
  |> filter(fn: (r) => r["_measurement"] == "mqtt_consumer")
  |> filter(fn: (r) => r["meter"] == "meter_01")
  |> filter(fn: (r) => r["_field"] == "active_power_w")
  |> aggregateWindow(every: 1m, fn: mean, createEmpty: false)

Energy consumed today in kWh (Stat panel with cost calculation):

from(bucket: "energy")
  |> range(start: today())
  |> filter(fn: (r) => r["_measurement"] == "mqtt_consumer")
  |> filter(fn: (r) => r["meter"] == "meter_01")
  |> filter(fn: (r) => r["_field"] == "import_energy_kwh")
  |> difference()
  |> sum()
  |> map(fn: (r) => ({r with _value: r._value * 0.18}))  // multiply by cost per kWh

Power factor trend (low values indicate reactive load issues):

from(bucket: "energy")
  |> range(start: -24h)
  |> filter(fn: (r) => r["_measurement"] == "mqtt_consumer")
  |> filter(fn: (r) => r["meter"] == "meter_01")
  |> filter(fn: (r) => r["_field"] == "power_factor")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)

Remote Dashboard Access with Localtonet

With Grafana running on port 3000, create a Localtonet HTTP tunnel to give energy managers and maintenance teams remote access from any browser without VPN or firewall changes.

curl -fsSL https://localtonet.com/install.sh | sh
localtonet --authtoken <YOUR_TOKEN>

Go to the HTTP tunnel page, set local IP to 127.0.0.1 and port to 3000. Click Create and start the tunnel.

Register all services as systemd services so they survive a reboot:

# Register energy reader as a systemd service
sudo tee /etc/systemd/system/energy-reader.service <<'EOF'
[Unit]
Description=Energy Meter Modbus Reader
After=network.target mosquitto.service

[Service]
ExecStart=/usr/bin/python3 /home/pi/energy_reader.py
Restart=always
User=pi

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl enable energy-reader
sudo systemctl start energy-reader

# Register Localtonet as a systemd service
sudo localtonet --install-service --authtoken <YOUR_TOKEN>
sudo localtonet --start-service --authtoken <YOUR_TOKEN>

Verify all services are active:

sudo systemctl status energy-reader mosquitto influxdb telegraf grafana-server localtonet

Security Recommendations

Frequently Asked Questions