Innovenergy_trunk/csharp/Lib/Devices/BatteryDeligreen/Doc/retrieve Telemetry.py

import serial

def check_starting_byte_and_extract_details(response):
    # Ensure the response is a valid hex string
    if not response or len(response) < 38 + (16 * 8) + 4 + (8 * 14) + 4:  # Update minimum length check
        print("Response is too short to contain valid data.")
        return

    # Extract the first byte and check if it's '7E'
    starting_byte = response[:2]
    if starting_byte.upper() == "7E":
        print(f"Starting byte: {starting_byte} (Hex)")
    else:
        print(f"Incorrect starting byte: {starting_byte}")
        return

    # Extract the next two bytes for the firmware version
    version_bytes = response[2:6]
    try:
        version_ascii = bytes.fromhex(version_bytes).decode('ascii')
        print(f"Firmware version: {version_bytes} (Hex), ASCII: {version_ascii}")
    except ValueError:
        print(f"Failed to decode firmware version from bytes: {version_bytes}")
        return

    # Extract the next two bytes for the address
    address_bytes = response[6:10]
    try:
        address_ascii = bytes.fromhex(address_bytes).decode('ascii')
        address_decimal = int(address_ascii, 16)
        print(f"Device Address: {address_bytes} (Hex), ASCII: {address_ascii}, Decimal: {address_decimal}")
    except ValueError:
        print(f"Failed to decode device address from bytes: {address_bytes}")
        return

    # Extract the next two bytes for CID1 (Device Code)
    cid1_bytes = response[10:14]
    try:
        cid1_ascii = bytes.fromhex(cid1_bytes).decode('ascii')
        cid1_decimal = int(cid1_ascii, 16)
        print(f"Device Code (CID1): {cid1_bytes} (Hex), ASCII: {cid1_ascii}, Decimal: {cid1_decimal}")
    except ValueError:
        print(f"Failed to decode device code from bytes: {cid1_bytes}")

    # Extract the next two bytes for the Function Code
    function_code_bytes = response[14:18]
    try:
        function_code_ascii = bytes.fromhex(function_code_bytes).decode('ascii')
        function_code_decimal = int(function_code_ascii, 16)
        print(f"Function Code: {function_code_bytes} (Hex), ASCII: {function_code_ascii}, Decimal: {function_code_decimal}")
    except ValueError:
        print(f"Failed to decode function code from bytes: {function_code_bytes}")

    # Extract the next 4 bytes for the Length Code
    length_code_bytes = response[18:26]
    try:
        length_ascii = bytes.fromhex(length_code_bytes).decode('ascii')
        length_decimal = int(length_ascii, 16)
        print(f"Length Code: {length_code_bytes} (Hex), ASCII: {length_ascii}, Decimal: {length_decimal}")
    except ValueError:
        print(f"Failed to decode length code from bytes: {length_code_bytes}")

    # Extract the next 2 bytes for the Data Flag
    data_flag_bytes = response[26:30]
    try:
        data_flag_ascii = bytes.fromhex(data_flag_bytes).decode('ascii')
        data_flag_decimal = int(data_flag_ascii, 16)
        print(f"Data Flag: {data_flag_bytes} (Hex), ASCII: {data_flag_ascii}, Decimal: {data_flag_decimal}")
    except ValueError:
        print(f"Failed to decode data flag from bytes: {data_flag_bytes}")

    # Extract the next 2 bytes for the Command Group
    command_group_bytes = response[30:34]
    try:
        command_group_ascii = bytes.fromhex(command_group_bytes).decode('ascii')
        command_group_decimal = int(command_group_ascii, 16)
        print(f"Command Group: {command_group_bytes} (Hex), ASCII: {command_group_ascii}, Decimal: {command_group_decimal}")
    except ValueError:
        print(f"Failed to decode command group from bytes: {command_group_bytes}")

    # Extract the next 2 bytes for the Number of Cells
    num_cells_bytes = response[34:38]
    try:
        num_cells_ascii = bytes.fromhex(num_cells_bytes).decode('ascii')
        num_cells_decimal = int(num_cells_ascii, 16)
        print(f"Number of Cells: {num_cells_bytes} (Hex), ASCII: {num_cells_ascii}, Decimal: {num_cells_decimal}")
    except ValueError:
        print(f"Failed to decode number of cells from bytes: {num_cells_bytes}")

    # Extract and process the voltages for all 16 cells
    for cell_index in range(16):
        start = 38 + (cell_index * 8)
        end = start + 8
        cell_voltage_bytes = response[start:end]
        try:
            cell_voltage_ascii = bytes.fromhex(cell_voltage_bytes).decode('ascii')
            cell_voltage_decimal = int(cell_voltage_ascii, 16) / 1000.0  # Convert to volts
            print(f"Voltage of Cell {cell_index + 1}: {cell_voltage_bytes} (Hex), ASCII: {cell_voltage_ascii}, Voltage: {cell_voltage_decimal:.3f} V")
        except ValueError:
            print(f"Failed to decode Voltage of Cell {cell_index + 1} from bytes: {cell_voltage_bytes}")

    # Extract the number of temperature sensors (4 hex bytes)
    num_temp_start = 38 + (16 * 8)
    num_temp_end = num_temp_start + 4
    num_temp_bytes = response[num_temp_start:num_temp_end]
    try:
        num_temp_ascii = bytes.fromhex(num_temp_bytes).decode('ascii')
        num_temp_decimal = int(num_temp_ascii, 16)
        print(f"Number of Temperature Sensors: {num_temp_bytes} (Hex), ASCII: {num_temp_ascii}, Decimal: {num_temp_decimal}")
    except ValueError:
        print(f"Failed to decode number of temperature sensors from bytes: {num_temp_bytes}")

    # Extract and process additional temperature and battery information
    current_index = num_temp_end

    # Cell Temperature 1
    for temp_index in range(1, 5):
        temp_bytes = response[current_index:current_index + 8]
        try:
            temp_ascii = bytes.fromhex(temp_bytes).decode('ascii')
            temp_decimal = (int(temp_ascii, 16)- 2731 )/ 10.0  # Convert to Celsius
            print(f"Cell Temperature {temp_index}: {temp_bytes} (Hex), ASCII: {temp_ascii}, Temperature: {temp_decimal:.2f} °C")
        except ValueError:
            print(f"Failed to decode Cell Temperature {temp_index} from bytes: {temp_bytes}")
        current_index += 8

    # Environment Temperature
    env_temp_bytes = response[current_index:current_index + 8]
    try:
        env_temp_ascii = bytes.fromhex(env_temp_bytes).decode('ascii')
        env_temp_decimal = (int(env_temp_ascii, 16) - 2731 )/ 10.0
        print(f"Environment Temperature: {env_temp_bytes} (Hex), ASCII: {env_temp_ascii}, Temperature: {env_temp_decimal:.2f} °C")
    except ValueError:
        print(f"Failed to decode Environment Temperature from bytes: {env_temp_bytes}")
    current_index += 8

    # Power Temperature
    power_temp_bytes = response[current_index:current_index + 8]
    try:
        power_temp_ascii = bytes.fromhex(power_temp_bytes).decode('ascii')
        power_temp_decimal = (int(power_temp_ascii, 16)- 2731 )/ 10.0
        print(f"Power Temperature: {power_temp_bytes} (Hex), ASCII: {power_temp_ascii}, Temperature: {power_temp_decimal:.2f} °C")
    except ValueError:
        print(f"Failed to decode Power Temperature from bytes: {power_temp_bytes}")
    current_index += 8

    # Charge/Discharge Current
    current_bytes = response[current_index:current_index + 8]
    try:
        current_ascii = bytes.fromhex(current_bytes).decode('ascii')
        current_decimal = int(current_ascii, 16) / 100.0
        print(f"Charge/Discharge Current: {current_bytes} (Hex), ASCII: {current_ascii}, Current: {current_decimal:.3f} A")
    except ValueError:
        print(f"Failed to decode Charge/Discharge Current from bytes: {current_bytes}")
    current_index += 8

    # Total Battery Voltage
    total_voltage_bytes = response[current_index:current_index + 8]
    try:
        total_voltage_ascii = bytes.fromhex(total_voltage_bytes).decode('ascii')
        total_voltage_decimal = int(total_voltage_ascii, 16) / 100.0
        print(f"Total Battery Voltage: {total_voltage_bytes} (Hex), ASCII: {total_voltage_ascii}, Voltage: {total_voltage_decimal:.3f} V")
    except ValueError:
        print(f"Failed to decode Total Battery Voltage from bytes: {total_voltage_bytes}")
    current_index += 8

    # Residual Capacity
    residual_capacity_bytes = response[current_index:current_index + 8]
    try:
        residual_capacity_ascii = bytes.fromhex(residual_capacity_bytes).decode('ascii')
        residual_capacity_decimal = int(residual_capacity_ascii, 16) / 100.0
        print(f"Residual Capacity: {residual_capacity_bytes} (Hex), ASCII: {residual_capacity_ascii}, Capacity: {residual_capacity_decimal:.3f} Ah")
    except ValueError:
        print(f"Failed to decode Residual Capacity from bytes: {residual_capacity_bytes}")
    current_index += 8

    # Custom Number
    custom_number_bytes = response[current_index:current_index + 4]
    try:
        custom_number_ascii = bytes.fromhex(custom_number_bytes).decode('ascii')
        custom_number_decimal = int(custom_number_ascii, 16)
        print(f"Custom Number: {custom_number_bytes} (Hex), ASCII: {custom_number_ascii}, Decimal: {custom_number_decimal}")
    except ValueError:
        print(f"Failed to decode Custom Number from bytes: {custom_number_bytes}")
    current_index += 4

    # Battery Capacity
    battery_capacity_bytes = response[current_index:current_index + 8]
    try:
        battery_capacity_ascii = bytes.fromhex(battery_capacity_bytes).decode('ascii')
        battery_capacity_decimal = int(battery_capacity_ascii, 16) / 100.0
        print(f"Battery Capacity: {battery_capacity_bytes} (Hex), ASCII: {battery_capacity_ascii}, Capacity: {battery_capacity_decimal:.3f} Ah")
    except ValueError:
        print(f"Failed to decode Battery Capacity from bytes: {battery_capacity_bytes}")
    current_index += 8

    # SOC
    soc_bytes = response[current_index:current_index + 8]
    try:
        soc_ascii = bytes.fromhex(soc_bytes).decode('ascii')
        soc_decimal = int(soc_ascii, 16) / 10.0
        print(f"SOC: {soc_bytes} (Hex), ASCII: {soc_ascii}, SOC: {soc_decimal:.2f}%")
    except ValueError:
        print(f"Failed to decode SOC from bytes: {soc_bytes}")
    current_index += 8

    # Rated Capacity
    rated_capacity_bytes = response[current_index:current_index + 8]
    try:
        rated_capacity_ascii = bytes.fromhex(rated_capacity_bytes).decode('ascii')
        rated_capacity_decimal = int(rated_capacity_ascii, 16) / 100.0
        print(f"Rated Capacity: {rated_capacity_bytes} (Hex), ASCII: {rated_capacity_ascii}, Capacity: {rated_capacity_decimal:.3f} Ah")
    except ValueError:
        print(f"Failed to decode Rated Capacity from bytes: {rated_capacity_bytes}")
    current_index += 8

    # Number of Cycles
    num_cycles_bytes = response[current_index:current_index + 8]
    try:
        num_cycles_ascii = bytes.fromhex(num_cycles_bytes).decode('ascii')
        num_cycles_decimal = int(num_cycles_ascii, 16)
        print(f"Number of Cycles: {num_cycles_bytes} (Hex), ASCII: {num_cycles_ascii}, Cycles: {num_cycles_decimal}")
    except ValueError:
        print(f"Failed to decode Number of Cycles from bytes: {num_cycles_bytes}")
    current_index += 8

    # SOH
    soh_bytes = response[current_index:current_index + 8]
    try:
        soh_ascii = bytes.fromhex(soh_bytes).decode('ascii')
        soh_decimal = int(soh_ascii, 16) / 10.0
        print(f"SOH: {soh_bytes} (Hex), ASCII: {soh_ascii}, SOH: {soh_decimal:.2f}%")
    except ValueError:
        print(f"Failed to decode SOH from bytes: {soh_bytes}")
    current_index += 8

    # bus voltage
    bus_bytes = response[current_index:current_index + 8]
    try:
        bus_ascii = bytes.fromhex(bus_bytes).decode('ascii')
        bus_decimal = int(bus_ascii, 16) / 100.0
        print(f"bus voltage: {bus_bytes} (Hex), ASCII: {bus_ascii}, bus voltage: {bus_decimal:.2f}V")
    except ValueError:
        print(f"Failed to decode bus voltage from bytes: {bus_bytes}")


def send_command():
    # Define the serial port and baud rate
    port = '/dev/ttyUSB0'  # Ensure the full path is correct
    baudrate = 19200  # Replace with the correct baud rate for your BMS

    # Create the serial connection
    try:
        with serial.Serial(port, baudrate, timeout=1) as ser:
            # Convert the hex string to bytes
            command = bytes.fromhex("7E3230303134363432453030323031464433350D")

            # Send the command
            ser.write(command)
            print("Command sent successfully.")

            # Wait for and read the response
            response = ser.read(200)  # Adjust the number of bytes to read as needed
            if response:
                hex_response = response.hex()
                print("Response received:", hex_response)
                # Process the response to check details
                check_starting_byte_and_extract_details(hex_response)
            else:
                print("No response received.")
    except serial.SerialException as e:
        print(f"Error opening serial port: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

if __name__ == "__main__":
    send_command()