import serial import time import threading import sys import json import uuid import paho.mqtt.client as mqtt mqtt_client = mqtt.Client(f"client_pzem_{uuid.uuid4()}") mqtt_client.connect("localhost", port=1883) mqtt_topic = sys.argv[1] # Configuração da porta serial porta_serial = 'COM14' # Substitua '/dev/ttyUSB0' pelo nome da porta serial correta no seu sistema baud_rate = 9600 timeout = 2 # Tempo de espera para a resposta CodigosFuncoes = { "LeituraParametro": 0x3, "AquisicaoDados": 0x04, "GravacaoParametro": 0x06, "ResetEnergia": 0x42, } Parametros = { "AlarmeTensaoMaxima": 0x0000, "AlarmeTensaoMinima": 0x0001, "EnderecoRTU": 0x0002, "EscalaCorrenteAtual": 0x0003, } slave_address = 0x01 # Inicialização da porta serial ser = serial.Serial(porta_serial, baud_rate, timeout=timeout) # Função para calcular o CRC16 de um comando def crc16(data: bytes): crc = 0xFFFF for pos in data: crc ^= pos for i in range(8): if (crc & 1) != 0: crc >>= 1 crc ^= 0xA001 else: crc >>= 1 return crc.to_bytes(2, byteorder='little') def verificar_crc(resposta_hex): resposta_bytes = bytes.fromhex(resposta_hex) crc_recebido = resposta_bytes[-2:] crc_calculado = crc16(resposta_bytes[:-2]) return crc_recebido == crc_calculado # Monta o comando para ler os registros def comando_aquisicao_dados(): function_code = CodigosFuncoes["AquisicaoDados"] start_address = 0x0000 num_of_registers = 0x0008 command = bytearray([slave_address, function_code]) + start_address.to_bytes(2, byteorder='big') + num_of_registers.to_bytes(2, byteorder='big') command += crc16(command) return command def decifrar_resposta_aquisicao(resposta): resposta = resposta.hex() # Certifique-se de que a resposta é do tipo bytes if isinstance(resposta, str): resposta = bytes.fromhex(resposta) print(resposta.hex()) print(len(resposta)) print(resposta[1]) print(CodigosFuncoes["AquisicaoDados"]) # Verifica o comprimento da resposta e o código da função if len(resposta) >= 5 and resposta[1] == CodigosFuncoes["AquisicaoDados"]: # Extrai o número de bytes de dados num_bytes = resposta[2] if num_bytes == 16: # Se o número de bytes de dados for 16 # Decodificação de tensão, corrente, potência e energia tensao_raw = int.from_bytes(resposta[3:5], byteorder='big') corrente_raw = int.from_bytes(resposta[5:7], byteorder='big') potencia_raw = int.from_bytes(resposta[7:11], byteorder='big') # 4 bytes para potência energia_raw = int.from_bytes(resposta[11:15], byteorder='big') # 4 bytes para energia high_alarm_raw = resposta[15] low_alarm_raw = resposta[16] # Conversões (ajuste conforme necessário) tensao = tensao_raw * 0.01 corrente = corrente_raw * 0.01 potencia = potencia_raw * 0.1 energia = energia_raw * 1 # Decodificação dos estados de alarme alarme_alta_tensao = "Ativo" if high_alarm_raw == 0xFFFF else "Nao Ativo" alarme_baixa_tensao = "Ativo" if low_alarm_raw == 0xFFFF else "Nao Ativo" # Retorno formatado com todos os valores resultado = { "Tensao": f"{tensao} V", "Corrente": f"{corrente} A", "Potencia": f"{potencia} W", "Energia": f"{energia} Wh", "Alarme de Alta Tensao": alarme_alta_tensao, "Alarme de Baixa Tensao": alarme_baixa_tensao } mqtt_client.publish(mqtt_topic, json.dumps(resultado).encode('utf-8')) else: print("Número de bytes de dados não esperado.") else: print("Resposta inválida ou código de função incorreto.") return None def comando_set_parametro(Parametro, valor): # Convertendo o valor de tensão para o formato correto (LSB = 0.01V) valor_int = int(valor * 100) # Convertendo para inteiro conforme a resolução function_code = CodigosFuncoes["GravacaoParametro"] register_address = Parametro # Construindo o comando command = bytearray([slave_address, function_code]) + register_address.to_bytes(2, byteorder='big') + valor_int.to_bytes(2, byteorder='big') # Adicionando o CRC ao comando command += crc16(command) return command def comando_get_parametro(Parametro): function_code = CodigosFuncoes["LeituraParametro"] register_address = Parametro number_of_registers = 0x0001 # Lendo apenas um registro # Construindo o comando command = bytearray([slave_address, function_code]) + register_address.to_bytes(2, byteorder='big') + number_of_registers.to_bytes(2, byteorder='big') # Adicionando o CRC ao comando command += crc16(command) return command def decifrar_resposta_get_parametro(resposta_hex): # Primeiro, verifique se a resposta é válida verificando o CRC if not verificar_crc(resposta_hex): return "CRC inválido. A resposta pode estar corrompida." # Converte a parte dos dados da resposta de hex para decimal valor_hex = resposta_hex[6:10] valor_decimal = int(valor_hex, 16) # Converte o valor para a unidade correta (LSB = 0.01V) valor_unidade = valor_decimal * 0.01 return valor_unidade def comando_reset_energia(): # Código de função específico para reset de energia function_code = CodigosFuncoes["ResetEnergia"] # Construindo o comando command = bytearray([slave_address, function_code]) # Adicionando o CRC ao comando command += crc16(command) return command def comando_calibragem(): # Bytes fixos para o comando de calibragem bytes_comando = bytearray([0xF8, 0x41, 0x37, 0x21]) # Calculando o CRC para os bytes do comando crc = crc16(bytes_comando) # Adicionando o CRC ao comando comando_calibragem = bytes_comando + crc return comando_calibragem def enviar_comando(comando): ser.write(comando) time.sleep(0.1) # Pequena pausa para garantir o envio resposta = ser.read(ser.inWaiting()) # Lê todos os dados disponíveis na porta return resposta def main(): comando = comando_aquisicao_dados() enviar_comando(comando_calibragem()) time.sleep(5) #com_set = comando_set_parametro(Parametros["EscalaCorrenteAtual"], 0) #enviar_comando(com_set) #time.sleep(1) vmax = decifrar_resposta_get_parametro(enviar_comando(comando_get_parametro(Parametros["AlarmeTensaoMaxima"])).hex()) vmin = decifrar_resposta_get_parametro(enviar_comando(comando_get_parametro(Parametros["AlarmeTensaoMinima"])).hex()) addr = decifrar_resposta_get_parametro(enviar_comando(comando_get_parametro(Parametros["EnderecoRTU"])).hex()) rang = decifrar_resposta_get_parametro(enviar_comando(comando_get_parametro(Parametros["EscalaCorrenteAtual"])).hex()) print(f"Alarme tensao maxima: {vmax}") print(f"Alarme tensao minima: {vmin}") print(f"Endereco RTU: {addr}") print(f"Escala de corrente: {rang}") try: while True: resposta = enviar_comando(comando) try: decifrar_resposta_aquisicao(resposta) except: print("Erro ao decodificar mensagem: " + resposta.hex()) #time.sleep(1) # Espera 1 segundo antes da próxima leitura except KeyboardInterrupt: print("Programa encerrado pelo usuário") finally: ser.close() # Esta função é para confirmar que script está pronto para execução def send_script_ready(): mqtt_client.publish(mqtt_topic, "OK") if __name__ == "__main__": # Cria trhead separada para informar que o script iniciou com sucesso mqtt_thread = threading.Thread(target=send_script_ready) mqtt_thread.start() main()