Huabang DDS228 Modbus Parser

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()