Python DLMS Library

Hello Gurux team and community,

I'm trying to establish a connection with a DLMS meter using the Python bindings for gurux_dlms and gurux_net libraries. I am using a standard HDLC-over-TCP configuration.

My Setup:

Connection Type: TCP/IP

Protocol: HDLC Wrapper (InterfaceType.HDLC)

Host IP: 192.168.1.1 -- ip is not actual ip

Port: 8080

Client Address: 16

Server Address: 1

Authentication: Authentication.NONE

Libraries: Name: gurux_dlms
Version: 1.0.191
Summary: Gurux DLMS library for Python.
Home-page: https://github.com/gurux/gurux.dlms.python
Author: Gurux Ltd
Author-email: gurux@gurux.fi
License: GPLv2
Location: C:\Users\User\PyCharmMiscProject\.venv\Lib\site-packages
Requires:
Required-by:

I`m not very proficient in programming, any help given to me can mean big to me.

The problem is:

Connected to 192.168.1.1:8080
Error during communication: 'bytearray' object has no attribute 'eop'

Here are my code:

from gurux_dlms import GXDLMSClient, GXReplyData, GXByteBuffer
from gurux_dlms.enums import Authentication, InterfaceType
from gurux_dlms.objects import GXDLMSRegister
from gurux_net import GXNet

HOST = "192.168.1.1"
PORT = 8080
BUFFER_SIZE = 1024 # buffer size for receive

def read_feeder_f21():
client = GXDLMSClient(
useLogicalNameReferencing=True,
clientAddress=16,
serverAddress=1,
forAuthentication=Authentication.NONE,
interfaceType=InterfaceType.HDLC
)

connection = GXNet("TCP", HOST, PORT)

try:
connection.open()
print(f"Connected to {HOST}:{PORT}")

# --------------------------
# Handshake
# --------------------------
data = client.snrmRequest()
if data:
connection.send(data)
buffer = bytearray(BUFFER_SIZE)
length = connection.receive(buffer)
reply = bytes(buffer[:length]) # Convert to bytes
reply_buffer = GXByteBuffer(reply)
client.parseUAResponse(reply_buffer)

data = client.aarqRequest()
connection.send(data)
buffer = bytearray(BUFFER_SIZE)
length = connection.receive(buffer)
reply = bytes(buffer[:length]) # Convert to bytes
reply_buffer = GXByteBuffer(reply)
client.parseAareResponse(reply_buffer)
print("Handshake completed")

# --------------------------
# OBIS list
# --------------------------
obis_list = {
"L1 Voltage": "1.1.32.7.0.255",
"L2 Voltage": "1.1.52.7.0.255",
"L3 Voltage": "1.1.72.7.0.255",
"L1 Current": "1.1.31.7.0.255",
"L2 Current": "1.1.51.7.0.255",
"L3 Current": "1.1.71.7.0.255",
"Neutral Current": "1.1.91.7.0.255",
"Frequency": "1.1.0.2.0.255",
"Active Power": "1.1.16.7.0.255",
"Power Factor": "1.1.13.7.0.255",
}

for name, obis in obis_list.items():
try:
obj = GXDLMSRegister(obis)
request_data = client.read(obj, 2)
connection.send(request_data)

# --------------------------
# Receive full response (multi-frame support)
# --------------------------
response = b""
buffer = bytearray(BUFFER_SIZE)
length = connection.receive(buffer)
response += bytes(buffer[:length])

while not client.isComplete(response):
length = connection.receive(buffer)
response += bytes(buffer[:length])

# --------------------------
# Parse response
# --------------------------
reply_buffer = GXByteBuffer(response)
reply_data = GXReplyData()
client.getData(reply_buffer, reply_data)

if reply_data.value is not None:
print(f"{name}: {reply_data.value}")
else:
print(f"{name}: No value")
except Exception as e:
print(f"Failed to read {name} ({obis}): {e}")

# --------------------------
# Release and disconnect
# --------------------------
release_data = client.releaseRequest()
if release_data:
connection.send(release_data)
connection.close()
print("Disconnected from F21")

except Exception as e:
print("Error during communication:", e)
connection.close()

if __name__ == "__main__":
read_feeder_f21()