[Python] Issues with reading data from electric meter using esp32

I'm trying to read data from electric meter with custom dlms protocol (requires lls authorization with password as serial number: 00053346) through esp32 in pair with max485. I have an official program which can access to meter and read data.

I came up with a Python script which reads obis 1.0.1.8 from meter, and gets data successfully, however when I (deepseek) tried to remake this code in Micropython, I only can successfully wake up the meter. it fails on aarq/authentification. What am I doing wrong?

Here's working code on Python:

import time
import serial
from gurux_common import GXCommon

PORT = "/dev/ttyUSB0"
BAUDRATE = 9600
SERIAL_NUMBER = "88935860"
CLIENT_ADDRESS = "00023445"

# Пакеты
WAKE_UP_REQUEST = bytearray.fromhex("7E A0 0A FE FE FE FF 09 93 B9 7E 7E")
AARQ_REQUEST = bytearray.fromhex(
"7E A0 47 00 02 34 45 09 10 BB 2E E6 E6 00 60 36 A1 09 06 07 60 85 74 05 08 01 01 "
"8A 02 07 80 8B 07 60 85 74 05 08 02 01 AC 0A 80 08 38 38 39 33 35 38 36 30 "
"BE 10 04 0E 01 00 00 00 06 5F 1F 04 00 B9 58 02 04 B0 0E 44 7E"
)
OBIS_REQUEST = bytearray.fromhex("7E A0 1C 00 02 34 45 09 32 E6 28 E6 E6 00 C0 01 C1 00 03 01 00 01 08 00 FF 02 00 32 68 7E")
DISCONNECT_REQUEST = bytearray.fromhex("7E A0 0A FE FE FE FF 03 53 C5 45 7E")

def send_and_receive(port, data, timeout=3):
port.reset_input_buffer()
port.rts = True
time.sleep(0.1)
print(f"TX > {GXCommon.toHex(data)}")
port.write(data)
time.sleep(0.2)
port.rts = False

response = bytearray()
start_time = time.time()

while time.time() - start_time < timeout:
if port.in_waiting:
response.extend(port.read(port.in_waiting))
if len(response) >= 3 and response[0] == 0x7E and response[-1] == 0x7E:
print(f"RX < {GXCommon.toHex(response)}")
return response
time.sleep(0.05)

print("Нет ответа")
return None

def parse_energy_data(response):
"""Парсинг данных для 1.0.1.8.0.255 (Активная энергия)"""
try:
# Пример ответа: 7E A0 18 09 00 02 34 45 52 C7 25 E6 E7 00 C4 01 C1 00 06 00 00 0E 27 D0 D7 7E
if len(response) < 22:
return None

# Ищем маркер данных
data_marker = bytearray.fromhex("C4 01 C1 00 06")
marker_pos = response.find(data_marker)

if marker_pos == -1:
return None

# Извлекаем 4 байта значения (00 00 0E 27 в примере)
value_bytes = response[marker_pos+5:marker_pos+9]

# Преобразуем в целое (big-endian)
value = int.from_bytes(value_bytes, byteorder='big')

# Коэффициент масштабирования (1000 для энергии в кВт*ч)
return value / 1000.0

except Exception as e:
print(f"Ошибка парсинга: {e}")
return None

def main():
try:
ser = serial.Serial(PORT, BAUDRATE, timeout=2)
print(f"Порт {PORT} открыт\n")

# 1. Wake-up
print("[1/4] Wake-up...")
if not send_and_receive(ser, WAKE_UP_REQUEST):
print("Ошибка: Нет ответа на Wake-up")
return

# 2. Аутентификация
print("\n[2/4] Аутентификация...")
aarq = AARQ_REQUEST.replace(
bytearray.fromhex("00 02 34 45"),
bytearray.fromhex(CLIENT_ADDRESS)
).replace(
bytearray.fromhex("38 38 39 33 35 38 36 30"),
bytearray(SERIAL_NUMBER, 'ascii')
)

if not send_and_receive(ser, aarq, timeout=5):
print("Ошибка аутентификации")
return

# 3. Чтение данных
print("\n[3/4] Чтение активной энергии (1.0.1.8.0.255)...")
req = OBIS_REQUEST.replace(
bytearray.fromhex("00 02 34 45"),
bytearray.fromhex(CLIENT_ADDRESS)
)
resp = send_and_receive(ser, req, timeout=5)

if resp:
value = parse_energy_data(resp)
if value is not None:
print(f"\n=== Результат ===")
print(f"Активная энергия: {value:.3f} кВт·ч")
else:
print("Не удалось распарсить ответ")
print(f"Сырые данные: {resp.hex()}")
else:
print("Нет ответа от счетчика")

except Exception as e:
print(f"\nОшибка: {str(e)}")
finally:
if 'ser' in locals() and ser.is_open:
print("\n[4/4] Завершение сессии...")
send_and_receive(ser, DISCONNECT_REQUEST)
ser.close()
print("Порт закрыт")

if __name__ == "__main__":
main()


the code that im using for esp32:

from machine import UART, Pin
import time
import ubinascii

# Конфигурация UART для RS485
uart = UART(2, baudrate=9600, bits=8, parity=None, stop=1, tx=17, rx=16)
rs485_dir = Pin(4, Pin.OUT) # Пин управления направлением DE/RE

# Параметры счетчика
CLIENT_ADDRESS = b'\x00\x02\x34\x45' # Адрес клиента
SERVER_ADDRESS = b'\x00\x00\x00\x00' # Адрес сервера
PASSWORD = b'00053346' # Пароль LLS (серийный номер)

# Функция для отправки данных с управлением направлением RS485
def send_rs485(data):
rs485_dir.value(1) # Включаем передачу
time.sleep_ms(10)
uart.write(data)
time.sleep_ms(len(data)*2) # Ждем окончания передачи
rs485_dir.value(0) # Включаем прием

# Функция для чтения ответа с таймаутом
def receive_rs485(timeout=1000):
start = time.ticks_ms()
buffer = bytearray()

while time.ticks_diff(time.ticks_ms(), start) < timeout:
if uart.any():
chunk = uart.read(uart.any())
print("RX chunk:", ubinascii.hexlify(chunk)) # Debug received data
buffer.extend(chunk)
if len(buffer) >= 3 and buffer[0] == 0x7E and buffer[-1] == 0x7E:
return buffer
time.sleep_ms(10)

print("Timeout after", timeout, "ms") # Debug timeout
return None

# Wake-up запрос
def send_wakeup():
wakeup = b'\x7E\xA0\x0A\xFE\xFE\xFE\xFF\x09\x93\xB9\x7E\x7E'
print("TX Wake-up:", ubinascii.hexlify(wakeup))
send_rs485(wakeup)
return receive_rs485()

# AARQ запрос аутентификации
def send_aarq():
# Шаблон AARQ запроса
aarq_template = (
b'\x7E\xA0\x47\x00\x02\x34\x45'
b'\x09\x10\xBB\x2E\xE6\xE6\x00\x60\x36\xA1\x09\x06\x07\x60\x85\x74\x05\x08\x01\x01'
b'\x8A\x02\x07\x80\x8B\x07\x60\x85\x74\x05\x08\x02\x01\xAC\x0A\x80\x08'
b'00053346' # ASCII password
b'\xBE\x10\x04\x0E\x01\x00\x00\x00\x06\x5F\x1F\x04\x00\xB9\x58\x02\x04\xB0\x0E\x44\x7E'
)

# Заменяем только адрес клиента, пароль остается в формате ASCII
aarq = aarq_template.replace(b'\x00\x02\x34\x45', CLIENT_ADDRESS)

print("TX AARQ:", ubinascii.hexlify(aarq))
send_rs485(aarq)
return receive_rs485(2000) # Увеличенный таймаут для аутентификации



# Запрос чтения данных
def read_data(obis_code):
# Подготовка запроса в зависимости от OBIS кода
if obis_code == "1.0.1.8.0.255":
request = (
b'\x7E\xA0\x1C\x00\x02\x34\x45' # Адрес будет заменен
b'\x09\x32\xE6\x28\xE6\xE6\x00\xC0\x01\xC1\x00\x03\x01\x00\x01\x08\x00\xFF\x02\x00\x32\x68\x7E'
)
# Добавьте другие OBIS коды по аналогии

request = request.replace(b'\x00\x02\x34\x45', CLIENT_ADDRESS)
print(f"TX Read {obis_code}:", ubinascii.hexlify(request))
send_rs485(request)
return receive_rs485()

# Парсинг значения энергии из ответа
def parse_energy_value(response):
try:
# Ищем маркер начала данных
marker = b'\xC4\x01\xC1\x00\x06'
pos = response.find(marker)
if pos == -1:
return None

# Извлекаем 4 байта значения
value_bytes = response[pos+5os+9]
value = int.from_bytes(value_bytes, 'big')
return value / 1000.0 # Масштабируем до kWh

except Exception as e:
print("Parse error:", e)
return None

# Основной цикл работы
def main():
print("Starting communication with meter...")

# 1. Wake-up
print("\n1. Sending Wake-up...")
wake_response = send_wakeup()
if wake_response:
print("RX Wake-up:", ubinascii.hexlify(wake_response))
else:
print("No wake-up response!")
return

# 2. Аутентификация
print("\n2. Authenticating...")
aarq_response = send_aarq()
if aarq_response:
print("RX AARQ:", ubinascii.hexlify(aarq_response))
else:
print("Authentication failed!")
return

# 3. Чтение данных
print("\n3. Reading data...")
data_response = read_data("1.0.1.8.0.255")
if data_response:
print("RX Data:", ubinascii.hexlify(data_response))
energy = parse_energy_value(data_response)
if energy is not None:
print(f"Active energy: {energy:.3f} kWh")
else:
print("Failed to parse energy value!")
else:
print("No data received!")

# 4. Завершение сессии
print("\n4. Disconnecting...")
disconnect = b'\x7E\xA0\x0A\xFE\xFE\xFE\xFF\x03\x53\xC5\x45\x7E'
send_rs485(disconnect)
print("Session closed")

if __name__ == "__main__":
main()


the logs I get: Starting communication with meter...

Sending Wake-up... TX Wake-up: b'7ea00afefefeff0993b97e7e' RX chunk: b'7e' RX chunk: b'a02309000234457384' RX chunk: b'b5818014050201000602' RX chunk: b'010007040000000108' RX chunk: b'0400000001696d7e' RX Wake-up: b'7ea02309000234457384b58180140502010006020100070400000001080400000001696d7e'

Authenticating... TX AARQ: b'7ea047000234450910bb2ee6e6006036a1090607608574050801018a0207808b0760857405080201ac0a80083030303533333436be10040e01000000065f1f0400b9580204b00e447e' Timeout after 2000 ms Authentication failed!


I've tried work further with deepseek, but it can't give the answer.

Continue reading...