Olá Pessoal,
Apenas compartilhando o que fiz aqui.
DST238-7 Ver:TY
Hoje tenho um DST238-7 Ver:TY , instalado em meu quadro principal , monitorando toda carga de energia da casa.
- Conectado na nuvem da TUYA e coletando dados via integração TuyaLocal.
- Na integração tuya , coleta apenas o consumo total das 3 fases , apesar de ele ter a informação por fase.
- Salvo o histórico dos dados em um Influx , que usei para fazer levantamentos de horas de consumo e definir a compra da fotovoltaica que vou instalar.
PROBLEMAS com este modelo
- Este modelo, oficialmente não é bi-direcional, para controlar com geração fotovoltaica, quando está injetando na rede.
Descobri que existe a versão IVAP (acho que não existia quando comprei o meu).
E que a diferença entre eles é o fato do IVAP ser oficialmente bidirecional (para geração fotovoltaica) e ter muito mais recursos de proteção. - A integração TUYA , não traz informação por fase, o que atrapalhou meu levantamento sobre a distribuição de carga aqui em casa (trifásico) e assim definição sobre a configuração da fotovoltaica (microinversores, 1 por fase , também para balanceamento)
RS485
Este modelo tem uma porta de comunicação RS485 , que ignorei quando instalei ele (não devia ter ignorado).
Com isso, é possivel coletar dados diretamente sem depender da nuvem tuya , não precisa nem parear o device com o app da tuya.
Comprei um modulo RS485-WIFI no aliexpress (Elfin EW11A mini) , conectei nele e com a ajuda da IA, criei um script python para fazer a coleta dos dados. Confesso que não foi fácil, devido à falta de documentação no site da TOMZN , mas para minha sorte achei a documentação dos comandos RS485 em outra variação do modelo que serviu perfeitamente.
Passei a coletar TODAS as informações exibida no display:
# ./modbus.dst238-7.ver-ty.py modbus.conf
ADDR | NOME | VALOR | UNIDADE
-----------------------------------------------------------------
0 | Active Energy Total | 14328.04 | kWh
4 | Reactive Energy Positive | 60.14 | kvarh
8 | Reversing Energy | 0.0 | kWh
10 | Forward Energy | 14328.04 | kWh
17 | Frequency | 59.97 | Hz
21 | Comms Info | Addr: 1, Baud: 9600 |
128 | Voltage Phase A | 124.9 | V
129 | Voltage Phase B | 124.0 | V
130 | Voltage Phase C | 124.5 | V
131 | Current Phase A | 1.66 | A
132 | Current Phase B | 1.19 | A
133 | Current Phase C | 0.95 | A
134 | Active Power Total | 0.323 | kW
136 | Active Power Phase A | 0.155 | kW
137 | Active Power Phase B | 0.059 | kW
138 | Active Power Phase C | 0.108 | kW
139 | Reactive Power Total | -0.237 | kvar
141 | Reactive Power Phase A | -0.104 | kvar
142 | Reactive Power Phase B | -0.105 | kvar
143 | Reactive Power Phase C | -0.027 | kvar
144 | Apparent Power Total | 0.475 | kVA
146 | Apparent Power Phase A | 0.207 | kVA
147 | Apparent Power Phase B | 0.149 | kVA
148 | Apparent Power Phase C | 0.118 | kVA
149 | Power Factor Total | 0.678 | PF
150 | Power Factor Phase A | 0.745 | PF
151 | Power Factor Phase B | 0.396 | PF
152 | Power Factor Phase C | 0.914 | PF
Pelas informações da documentação que encontrei, dá a entender que ele suporta o bi-direcional, pois tem registradores para isso. Mas não tenho como afirmar na prática e não terei no futuro, porque vou troca-lo pela versão IVAP.
Então ainda com a ajuda da IA, ajustei o script para já enviar os dados para o meu Home Assistant, via MQTT ou integração Modbus.
Na opção MQTT tem a vantagem de que ele cria o device , adiciona os sensores, tudo automático.
Porém o disparo de coleta é manual , ou você cria uma automação que roda o script ou cria uma chamada via crontab.
E a coleta é sempre 100% dos sensores, não é possivel separar qual sensor coletar (talvez ajustar o script para permitir receber como parametro esta informação).
No meu caso, como rodo meu HA em docker, tive que usar a opção de crontab do host para coletar a cada 20 segundos:
* * * * * { cd ~/bin ; for x in 1 2 3 ; do ./modbus.dst238-7.ver-ty.py --mqtt modbus.conf >/dev/null ; sleep 20 ; done ; }
Já a versão Modbus tem um ponto de vantagem e outras desvantagens:
- Vantagem unica: possivel definir o intervalo de coleta por sensor
- Desvantages
- Não cria um device , todos os sensores aparecem dentro da integração Modbus
- Faz a coleta item a item , o que deixa o processo muito lento, já que não é super rapido esta coleta. (coletar todos os itens em 1 comando leva ~2 segundos, item a item leva ~25seg).
Codigo
Segue abaixo a copia do script python e arquivo de configuração já com os registradores disponiveis para este modelo.
#!/usr/bin/python3
# Versão: 17.0
# Data: 2026-02-17
# Descrição: Leitor Modbus Tomzn DST238-7
# Funcionalidades:
# - Leitura otimizada por Batch (blocos)
# - Tratamento de tipos DWORD/INT32 e Signed/Unsigned
# - Integração MQTT com Auto-Discovery do Home Assistant
# - Geração de YAML para Cards (Lovelace)
# - Geração de YAML para integração Modbus nativa
import socket
import struct
import time
import sys
import json
import argparse
import os
# Tenta importar paho-mqtt para funcionalidade opcional de MQTT
try:
import paho.mqtt.client as mqtt
except ImportError:
mqtt = None
# --- CONSTANTES GLOBAIS ---
SLAVE_ID = 1
def load_config(config_file):
"""Carrega o arquivo JSON de configuração."""
if not os.path.exists(config_file):
print(f"ERRO: Arquivo '{config_file}' não encontrado.")
sys.exit(1)
try:
with open(config_file, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"ERRO JSON: {e}")
sys.exit(1)
def calculate_crc(data):
"""Calcula o CRC16 (Modbus) para validação de pacotes."""
crc = 0xFFFF
for char in data:
crc ^= char
for _ in range(8):
if (crc & 0x0001):
crc >>= 1
crc ^= 0xA001
else:
crc >>= 1
return struct.pack('<H', crc)
def read_batch(sock, start_addr, quantity):
"""
Realiza a leitura de um bloco contínuo de registradores.
Usa Modbus TCP encapsulando frames RTU (Modbus-TCP-to-RTU).
"""
# Monta frame RTU: ID + Func(3) + Endereço + Qtd + CRC
req = struct.pack('>BBHH', SLAVE_ID, 3, start_addr, quantity)
req += calculate_crc(req)
try:
sock.send(req)
# Resposta esperada: ID + Func + BytesCount + Dados + CRC
response = sock.recv(1024)
if len(response) < 5:
return None, f"Resposta Curta ({len(response)}b)"
# Valida integridade dos dados recebidos via CRC
if calculate_crc(response[:-2]) != response[-2:]:
return None, "Erro CRC"
# Retorna apenas a carga útil (dados), removendo cabeçalho e CRC
return response[3:-2], None
except Exception as e:
return None, str(e)
def parse_bytes_to_value(raw_chunk, reg_config):
"""
Converte os bytes brutos lidos do medidor para valores humanos.
Aplica escala, sinal e formatação baseada no 'type' definido no conf.
"""
dtype = reg_config.get('type', 'WORD')
scale = reg_config.get('scale', 1.0)
hex_s = raw_chunk.hex().upper()
try:
if dtype == "HI_LO":
# Tratamento especial para registrador 21 (0x15): High Byte = Address, Low Byte = Baudrate
val_full = struct.unpack('>H', raw_chunk)[0]
high = (val_full >> 8) & 0xFF
low = val_full & 0xFF
baud_map = {1: 9600, 2: 4800, 3: 2400, 4: 1200}
baud_str = baud_map.get(low, f"Unknown({low})")
val_fmt = f"Addr: {high}, Baud: {baud_str}"
return hex_s, val_full, val_fmt
# Definição de formatos para struct.unpack (Big Endian '>')
fmt_map = {
"DWORD": ">I", # Unsigned 32-bit (4 bytes)
"INT32": ">i", # Signed 32-bit (4 bytes)
"WORD": ">H", # Unsigned 16-bit (2 bytes)
"INT16": ">h" # Signed 16-bit (2 bytes)
}
fmt = fmt_map.get(dtype, ">H")
val_int = struct.unpack(fmt, raw_chunk)[0]
val_scaled = val_int * scale
# Ajuste fino para float point issues
if isinstance(val_scaled, float):
val_scaled = round(val_scaled, 4)
return hex_s, val_int, val_scaled
except Exception as e:
return hex_s, 0, 0
def get_mqtt_device_class(unit):
"""Helper para definir classes do Home Assistant baseadas na unidade."""
if unit in ["kWh", "Wh", "MWh"]: return "energy", "total_increasing"
elif unit in ["V"]: return "voltage", "measurement"
elif unit in ["A"]: return "current", "measurement"
elif unit in ["W", "kW", "VA", "kVA", "var", "kvar"]: return "power", "measurement"
elif unit in ["Hz"]: return "frequency", "measurement"
elif unit in ["PF"]: return "power_factor", "measurement"
return None, "measurement"
def send_mqtt(results, full_conf):
"""Publica os dados coletados num broker MQTT com Auto-Discovery."""
if not mqtt:
print("ERRO: Biblioteca paho-mqtt não encontrada.")
return
m_conf = full_conf.get('mqtt', {})
d_conf = full_conf.get('device', {})
r_map = full_conf.get('registers', {})
broker = m_conf.get('broker', 'localhost')
port = int(m_conf.get('port', 1883))
user = m_conf.get('username')
passwd = m_conf.get('password')
prefix = m_conf.get('topic_prefix', 'homeassistant')
client = mqtt.Client()
# Autenticação condicional: só usa se user E pass estiverem preenchidos
if user and passwd:
client.username_pw_set(user, passwd)
try:
client.connect(broker, port, 60)
except Exception as e:
print(f"Erro Conexão MQTT ({broker}:{port}): {e}")
return
dev_id = d_conf.get('unique_id', 'tomzn_meter')
dev_name = d_conf.get('name', 'Tomzn Meter')
# Payload do Dispositivo (para agrupar sensores)
device_info = {
"identifiers": [dev_id],
"name": dev_name,
"model": d_conf.get('model', 'DST238-7'),
"manufacturer": d_conf.get('manufacturer', 'Tomzn')
}
payload_values = {}
for addr, (hex_s, raw, val) in results.items():
addr_s = str(addr)
if addr_s not in r_map: continue
reg = r_map[addr_s]
if reg['type'] == 'HI_LO': continue
safe_name = reg['name'].lower().replace(" ", "_").replace(".", "").replace("ç","c").replace("ã","a").replace("ê","e")
unique_id = f"{dev_id}_{addr}"
object_id = f"{safe_name}_{addr}"
config_topic = f"{prefix}/sensor/{dev_id}/{object_id}/config"
state_topic = f"{prefix}/sensor/{dev_id}/state"
dc, sc = get_mqtt_device_class(reg['unit'])
cfg_payload = {
"name": f"{dev_name} {reg['name']}",
"unique_id": unique_id,
"state_topic": state_topic,
"value_template": f"{{{{ value_json.{object_id} }}}}",
"unit_of_measurement": reg['unit'],
"device": device_info,
"state_class": sc
}
if dc: cfg_payload["device_class"] = dc
# Publica configuração (Retain)
client.publish(config_topic, json.dumps(cfg_payload), retain=True)
payload_values[object_id] = val
# Publica valores (Estado)
if payload_values:
client.publish(f"{prefix}/sensor/{dev_id}/state", json.dumps(payload_values))
print(f"MQTT: Enviados {len(payload_values)} sensores para {broker}")
client.disconnect()
def generate_ha_card(full_conf):
"""Gera YAML para cartão 'Entities' do Dashboard do HA."""
d_conf = full_conf.get('device', {})
r_map = full_conf.get('registers', {})
dev_id = d_conf.get('unique_id', 'tomzn_meter')
print("\n# --- COPIE PARA SEU DASHBOARD (YAML) ---")
print("type: entities")
print(f"title: {d_conf.get('name', 'Medidor')}")
print("entities:")
sorted_addrs = sorted([int(k) for k in r_map.keys()])
for addr in sorted_addrs:
reg = r_map[str(addr)]
if reg['type'] == 'HI_LO': continue
safe_name = reg['name'].lower().replace(" ", "_").replace(".", "").replace("ç","c").replace("ã","a").replace("ê","e")
print(f" - entity: sensor.{dev_id}_{safe_name}_{addr}")
print(f" name: {reg['name']}")
def generate_modbus_integration_yaml(full_conf):
"""
Gera configuração YAML para a integração 'modbus:' nativa do HA.
Utiliza o 'scan_interval' definido por item no JSON.
"""
ew11_conf = full_conf.get('ew11', {})
d_conf = full_conf.get('device', {})
r_map = full_conf.get('registers', {})
# Valor padrão se não houver config específica
default_scan = d_conf.get('scan_interval', 30)
print("\n# --- COPIE PARA SEU CONFIGURATION.YAML (INTEGRAÇÃO MODBUS) ---")
print("modbus:")
print(" - name: tomzn_hub")
print(" type: tcp")
print(f" host: {ew11_conf.get('ip', '192.168.1.163')}")
print(f" port: {ew11_conf.get('port', 8899)}")
print(" sensors:")
# Mapeamento de tipos internos para tipos do HA Modbus
ha_type_map = {
"DWORD": "uint32",
"INT32": "int32",
"WORD": "uint16",
"INT16": "int16"
}
sorted_addrs = sorted([int(k) for k in r_map.keys()])
for addr in sorted_addrs:
reg = r_map[str(addr)]
# Ignora HI_LO (Endereço 15) pois requer template complexo não suportado nativamente de forma simples
if reg['type'] == 'HI_LO':
continue
ha_type = ha_type_map.get(reg['type'], "uint16")
# Usa scan_interval do item, ou o default do dispositivo se não existir
item_scan = reg.get('scan_interval', default_scan)
safe_name = reg['name'].lower().replace(" ", "_").replace(".", "").replace("ç","c").replace("ã","a").replace("ê","e")
print(f" - name: \"{reg['name']}\"")
# Unique ID ajuda a editar o sensor via UI depois
print(f" unique_id: {d_conf.get('unique_id', 'tomzn')}_{safe_name}_{addr}")
print(f" address: {addr}")
print(f" slave: {SLAVE_ID}")
print(f" input_type: holding")
print(f" data_type: {ha_type}")
print(f" scale: {reg['scale']}")
print(f" unit_of_measurement: \"{reg['unit']}\"")
print(f" scan_interval: {item_scan}")
# Define device_class e state_class para estatísticas de energia funcionarem
dc, sc = get_mqtt_device_class(reg['unit'])
if dc:
print(f" device_class: {dc}")
if sc == "total_increasing":
print(f" state_class: total_increasing")
elif sc == "measurement":
print(f" state_class: measurement")
def main():
parser = argparse.ArgumentParser(description="Leitor Tomzn DST238-7 v17.0")
parser.add_argument('config_file', help='Arquivo modbus.conf')
parser.add_argument('--mqtt', action='store_true', help='Envia leitura atual para MQTT')
parser.add_argument('--gen-card', action='store_true', help='Gera YAML para card Lovelace')
parser.add_argument('--gen-modbus', action='store_true', help='Gera YAML para integração Modbus nativa')
args = parser.parse_args()
full_conf = load_config(args.config_file)
ew11_conf = full_conf.get('ew11', {})
reg_map = full_conf.get('registers', {})
# Se for apenas para gerar configurações, não conecta no medidor
if args.gen_card:
generate_ha_card(full_conf)
return
if args.gen_modbus:
generate_modbus_integration_yaml(full_conf)
return
# Parâmetros de Conexão com o conversor EW11
ip = ew11_conf.get('ip', '192.168.1.163')
port = int(ew11_conf.get('port', 8899))
# Estabelece Conexão TCP
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3.0)
try:
sock.connect((ip, port))
except Exception as e:
print(f"ERRO: Não foi possível conectar em {ip}:{port} -> {e}")
return
target_addrs = sorted([int(k) for k in reg_map.keys()])
results = {}
# --- ALGORITMO DE LEITURA OTIMIZADA (BATCH) ---
# Agrupa endereços próximos para reduzir número de requisições
groups = []
if target_addrs:
current_group = [target_addrs[0]]
for addr in target_addrs[1:]:
last_addr = current_group[-1]
last_reg = reg_map[str(last_addr)]
# Determina tamanho do dado anterior (1 ou 2 words)
last_size = 2 if last_reg['type'] in ['DWORD', 'INT32'] else 1
expected_next = last_addr + last_size
gap = addr - expected_next
# Regras de agrupamento: gap pequeno (<10) e bloco total razoável (<60)
curr_size = 2 if reg_map[str(addr)]['type'] in ['DWORD', 'INT32'] else 1
block_len = (addr + curr_size) - current_group[0]
if gap < 10 and block_len < 60:
current_group.append(addr)
else:
groups.append(current_group)
current_group = [addr]
groups.append(current_group)
# --- EXECUÇÃO DAS LEITURAS ---
for grp in groups:
start = grp[0]
last = grp[-1]
last_type = reg_map[str(last)]['type']
last_size = 2 if last_type in ['DWORD', 'INT32'] else 1
count = (last - start) + last_size
raw_data, err = read_batch(sock, start, count)
if not err and raw_data:
for addr in grp:
offset = (addr - start) * 2
dtype = reg_map[str(addr)]['type']
dsize = 4 if dtype in ['DWORD', 'INT32'] else 2
# Garante que temos bytes suficientes no buffer para este registrador
if offset + dsize <= len(raw_data):
chunk = raw_data[offset:offset+dsize]
results[addr] = parse_bytes_to_value(chunk, reg_map[str(addr)])
else:
results[addr] = ("ERR", 0, "Buffer Incompleto")
else:
for addr in grp:
results[addr] = ("ERR", 0, f"Falha Leitura: {err}")
sock.close()
# --- EXIBIÇÃO NO TERMINAL ---
print(f"{'ADDR':<5} | {'NOME':<25} | {'VALOR':<15} | {'UNIDADE'}")
print("-" * 65)
for addr in target_addrs:
if addr in results:
hex_s, raw, val = results[addr]
reg = reg_map[str(addr)]
unit = reg.get('unit', '')
print(f"{addr:<5} | {reg['name']:<25} | {str(val):<15} | {unit}")
# Envio Opcional MQTT
if args.mqtt:
send_mqtt(results, full_conf)
if __name__ == "__main__":
main()
Arquivo de configuração:
{
"ew11": {
"ip": "192.168.#.#,
"port": 8899
},
"mqtt": {
"broker": "192.168.#.#",
"port": 1883,
"username": "",
"password": "",
"topic_prefix": "homeassistant"
},
"device": {
"name": "Disjuntor Geral",
"unique_id": "dst2387_rs485",
"model": "DST238-7 Ver:TY",
"manufacturer": "Tomzn"
},
"registers": {
"0": { "name": "Active Energy Total", "type": "DWORD", "scale": 0.01, "unit": "kWh", "sign": "Unsigned", "scan_interval": 60 },
"4": { "name": "Reactive Energy Positive", "type": "DWORD", "scale": 0.01, "unit": "kvarh", "sign": "Unsigned", "scan_interval": 60 },
"8": { "name": "Reversing Energy", "type": "DWORD", "scale": 0.01, "unit": "kWh", "sign": "Unsigned", "scan_interval": 60 },
"10": { "name": "Forward Energy", "type": "DWORD", "scale": 0.01, "unit": "kWh", "sign": "Unsigned", "scan_interval": 60 },
"17": { "name": "Frequency", "type": "WORD", "scale": 0.01, "unit": "Hz", "sign": "Unsigned", "scan_interval": 20 },
"21": { "name": "Comms Info", "type": "HI_LO", "scale": 1, "unit": "", "sign": "Unsigned", "scan_interval": 300 },
"128": { "name": "Voltage Phase A", "type": "WORD", "scale": 0.1, "unit": "V", "sign": "Unsigned", "scan_interval": 15 },
"129": { "name": "Voltage Phase B", "type": "WORD", "scale": 0.1, "unit": "V", "sign": "Unsigned", "scan_interval": 15 },
"130": { "name": "Voltage Phase C", "type": "WORD", "scale": 0.1, "unit": "V", "sign": "Unsigned", "scan_interval": 15 },
"131": { "name": "Current Phase A", "type": "WORD", "scale": 0.01, "unit": "A", "sign": "Unsigned", "scan_interval": 10 },
"132": { "name": "Current Phase B", "type": "WORD", "scale": 0.01, "unit": "A", "sign": "Unsigned", "scan_interval": 10 },
"133": { "name": "Current Phase C", "type": "WORD", "scale": 0.01, "unit": "A", "sign": "Unsigned", "scan_interval": 10 },
"134": { "name": "Active Power Total", "type": "INT32", "scale": 0.001, "unit": "kW", "sign": "Signed", "scan_interval": 10 },
"136": { "name": "Active Power Phase A", "type": "INT16", "scale": 0.001, "unit": "kW", "sign": "Signed", "scan_interval": 10 },
"137": { "name": "Active Power Phase B", "type": "INT16", "scale": 0.001, "unit": "kW", "sign": "Signed", "scan_interval": 10 },
"138": { "name": "Active Power Phase C", "type": "INT16", "scale": 0.001, "unit": "kW", "sign": "Signed", "scan_interval": 10 },
"139": { "name": "Reactive Power Total", "type": "INT32", "scale": 0.001, "unit": "kvar", "sign": "Signed", "scan_interval": 20 },
"141": { "name": "Reactive Power Phase A", "type": "INT16", "scale": 0.001, "unit": "kvar", "sign": "Signed", "scan_interval": 20 },
"142": { "name": "Reactive Power Phase B", "type": "INT16", "scale": 0.001, "unit": "kvar", "sign": "Signed", "scan_interval": 20 },
"143": { "name": "Reactive Power Phase C", "type": "INT16", "scale": 0.001, "unit": "kvar", "sign": "Signed", "scan_interval": 20 },
"144": { "name": "Apparent Power Total", "type": "DWORD", "scale": 0.001, "unit": "kVA", "sign": "Unsigned", "scan_interval": 20 },
"146": { "name": "Apparent Power Phase A", "type": "WORD", "scale": 0.001, "unit": "kVA", "sign": "Unsigned", "scan_interval": 20 },
"147": { "name": "Apparent Power Phase B", "type": "WORD", "scale": 0.001, "unit": "kVA", "sign": "Unsigned", "scan_interval": 20 },
"148": { "name": "Apparent Power Phase C", "type": "WORD", "scale": 0.001, "unit": "kVA", "sign": "Unsigned", "scan_interval": 20 },
"149": { "name": "Power Factor Total", "type": "WORD", "scale": 0.001, "unit": "PF", "sign": "Unsigned", "scan_interval": 30 },
"150": { "name": "Power Factor Phase A", "type": "WORD", "scale": 0.001, "unit": "PF", "sign": "Unsigned", "scan_interval": 30 },
"151": { "name": "Power Factor Phase B", "type": "WORD", "scale": 0.001, "unit": "PF", "sign": "Unsigned", "scan_interval": 30 },
"152": { "name": "Power Factor Phase C", "type": "WORD", "scale": 0.001, "unit": "PF", "sign": "Unsigned", "scan_interval": 30 }
}
}
Elfin EW11
No Elfin, a configuração serial dele :
- precisa ser 9600,8,1,NONE ( se assim estiver no medidor).
- Flow Control = none
- Protocol = None
Sobre a versão IVAP
Eu já comprei a versão IVAP e estou testando ela conectada na tomada (com apenas 2 fases , 127v/127v).
Achei bem legal o sistema de proteção existente , algo inexistente na versão TY.
Porém esta versão não possui a porta RS485, o que limita o uso com o TuyaLocal .
Mas, funciona! Então acho valido a troca.
Custos
E para calcular os meus custos de energia : Preço do kwh e bandeira tarifaria - #40 de ceinmart