Pasted

import serial
import struct
import time
 
# CRC-16 calculation function (Modbus flavor)
def crc16(data):
    crc = 0xFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            if crc & 0x0001:
                crc = (crc >> 1) ^ 0xA001
            else:
                crc >>= 1
    return crc
 
# Function to convert IEEE 754 floating point bytes to a float value
def ieee754_to_float(bytes):
    return struct.unpack('>f', struct.pack('>L', bytes))[0]
 
# Function to parse a Modbus response
def parse_modbus_response(response_bytes):
    # Slave ID
    slave_id = response_bytes[0]
   
    # Function code
    function_code = response_bytes[1]
   
    # Byte count
    byte_count = response_bytes[2]
   
    # Data bytes
    data_bytes = response_bytes[3:-2]
   
    # CRC bytes
    crc_bytes = response_bytes[-2:]
   
    # Calculate CRC and check
    calculated_crc = crc16(response_bytes[:-2])
    received_crc = int.from_bytes(crc_bytes, byteorder='little')
    if calculated_crc != received_crc:
        print("CRC error in the response")
        return
   
    # Check for error response
    if function_code == 0x83:
        exception_code = data_bytes[0]
        print(f"Error response: Exception code {exception_code}")
        return
   
    # Parse data based on expected data type
    if function_code == 0x03:  # Read holding registers
        if byte_count == 4:  # 32-bit float
            value = ieee754_to_float(int.from_bytes(data_bytes, byteorder='big'))
            return value
        elif byte_count == 2:  # 16-bit unsigned int
            value = int.from_bytes(data_bytes, byteorder='big')
            return value
        else:
            print("Unexpected byte count for read holding registers response")
            return None
    else:
        print(f"Unhandled function code: {function_code}")
        return None
 
# Function to construct a Modbus request
def construct_modbus_request(register_choice, slave_id=0x01):
    requests = {
        1: bytes.fromhex(f'01 03 00 00 00 02'),  # Total active energy (kWh)
        7: bytes.fromhex(f'01 03 00 90 00 02'),  # Grid frequency (Hz)
    }
   
    if register_choice in requests:
        request_bytes = requests[register_choice]
        crc = crc16(request_bytes)
        crc_bytes = crc.to_bytes(2, byteorder='little')
        return request_bytes + crc_bytes
    else:
        print("Invalid choice, please try again.")
        return None
 
# Main program
# Get the COM port from the user
com_port = input("Enter the COM port (e.g., COM1, /dev/ttyUSB0): ")
 
# Device ID is fixed to 26
device_id = 26
 
# Open the serial port
try:
    ser = serial.Serial(com_port, baudrate=9600, bytesize=8, parity=serial.PARITY_EVEN, stopbits=1)
except serial.SerialException as e:
    print(f"Error opening serial port: {e}")
    exit(1)
 
while True:
    print("\nSelect the register you want to request:")
    print("1. Total active energy (kWh)")
    print("7. Grid frequency (Hz)")
    print("0. Exit")
   
    choice = int(input("Enter your choice (0, 1, or 7): "))
   
    if choice == 0:
        break
   
    request_bytes = construct_modbus_request(choice, device_id)
   
    if request_bytes:
        print(f"Modbus request: {request_bytes.hex()}")
       
        # Send the request over the serial port
        ser.write(request_bytes)
       
        # Wait for the response
        response_bytes = ser.read(size=9)
       
        if len(response_bytes) == 9:
            value = parse_modbus_response(response_bytes)
            if value is not None:
                if choice == 1:
                    print(f"Total active energy (kWh): {value}")
                elif choice == 7:
                    print(f"Grid frequency (Hz): {value}")
        else:
            print("Invalid response length")
           
        # Add a delay to avoid flooding the meter with requests
        time.sleep(0.1)
 
print("Exiting program...")
ser.close()