measure_ctrl/playgrounds/i2c_playground.py

import smbus2

def main():
    # Initialize configuration
    config = {
        'i2c': {
            'bus': 1,
            'debug': False
        }
    }
    
    # Create I2C service instance
    i2c_service = I2CService(config)
    
    # Test parameters
    i2c_address = 0x48
    num_slots = 1
    
    try:
        # Request status list from the device
        # status_list = i2c_service.request_status_list(i2c_address, num_slots)
        status_list = i2c_service.request_measure_values(i2c_address, 0)
        # status_list = i2c_service.send_cell_limits(i2c_address, 0, [2500, 4200, 500, 2800, 200])
        print(f"Status list received: {status_list}")
        
    except Exception as e:
        print(f"Error occurred: {str(e)}")


class I2CService:
    status_register = 0x01
    cell_data_register = 0x02
    battery_limit_register = 0x03

    def __init__(self, config: dict):
        self.config = config
        bus_number = config.get('i2c', {}).get('bus', 1)
        self.bus = smbus2.SMBus(bus_number)
    
    def request_status_list(self, i2c_adress: int, num_slots: int):
        """Request the status of a all slots."""
        print(f"Requesting status list (i2c_adress: {i2c_adress}, register: {self.status_register})")
        status_list = self.bus.read_i2c_block_data(i2c_adress, self.status_register, num_slots)
        print(f"Received status list: {status_list} (i2c_adress: {i2c_adress})")
        return status_list

    def request_measure_values(self, i2c_adress: int, slot_id: int) -> list[int]:
        """Request the cell values of a specific slot.
        """
        print(f"Requesting measure values (i2c_adress: {i2c_adress}, slot_id: {slot_id})")
        slot_request = self.cell_data_register << 4 | slot_id
        response = self.bus.read_i2c_block_data(i2c_adress, slot_request, 8)
        
        print(f"Received response: {response} (i2c_adress: {i2c_adress}, slot_id: {slot_id})")
        # slot = int.from_bytes(response[0], byteorder='little')
        slot = response[0]
        voltage = int.from_bytes(response[2:4], byteorder='little')
        current = int.from_bytes(response[4:6], byteorder='little')
        temperature = int.from_bytes(response[6:8], byteorder='little')
        print(f"Decoded values: slot={slot}, voltage={voltage}, current={current}, temperature={temperature}")
        return response
    
    def send_cell_limits(self, i2c_adress: int, slot_id: int, limits: list[int]) -> bool:
        """Send the battery limits to the device.
        
        Args:
            i2c_adress: Device address
            slot_id: Target slot (0-15)
            limits: List of 16-bit values [max_voltage, min_voltage, max_current, min_current, max_temp]
        """
        # Convert each 16-bit value to 2 bytes in little-endian order
        msg_bytes = bytearray([slot_id])  # Start with slot_id
        for value in limits:
            # Convert each value to 2 bytes (16-bit) in little-endian order
            value_bytes = value.to_bytes(2, byteorder='little')
            msg_bytes.extend(value_bytes)

        self.bus.write_i2c_block_data(i2c_adress, self.battery_limit_register, msg_bytes)
        print(f"Sent msg (bytes): {list(msg_bytes)} (i2c_adress: {i2c_adress}, slot_id: {slot_id})")
        return True

if __name__ == "__main__":
    main()