208 lines
7.2 KiB
Python
208 lines
7.2 KiB
Python
import serial
|
|
import time
|
|
|
|
# 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)
|
|
|
|
# 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 "Não Ativo"
|
|
alarme_baixa_tensao = "Ativo" if low_alarm_raw == 0xFFFF else "Não Ativo"
|
|
|
|
# Retorno formatado com todos os valores
|
|
resultado = {
|
|
"Tensão": f"{tensao} V",
|
|
"Corrente": f"{corrente} A",
|
|
"Potência": f"{potencia} W",
|
|
"Energia": f"{energia} Wh",
|
|
"Alarme de Alta Tensão": alarme_alta_tensao,
|
|
"Alarme de Baixa Tensão": alarme_baixa_tensao
|
|
}
|
|
print(resultado)
|
|
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:
|
|
print(comando.hex())
|
|
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()
|
|
|
|
if __name__ == "__main__":
|
|
main() |