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 = 0x20
    num_slots = 4
    
    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, 25, 500, 75, 0])
        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 = slot_id << 4 | self.cell_data_register
        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})")
        voltage = int.from_bytes(response[0:2], byteorder='little')
        current = int.from_bytes(response[2:4], byteorder='little')
        temperature = int.from_bytes(response[4:6], byteorder='little')
        cycle_num = int.from_bytes(response[6:7], byteorder='little')
        cycle_state = int.from_bytes(response[7:8], byteorder='little')
        print(f"Decoded response: voltage={voltage}, current={current}, temperature={temperature}, cycle_num={cycle_num}, cycle_state={cycle_state}")
        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: [min_volts (uint16), max_volts (int16), charge_fraction (uint8), capacity (uint16), cut_off_current (uint8), cycle_number (uint8)]
        """
        msg_bytes = bytearray()
        slot_request = slot_id << 4 | self.battery_limit_register
        # min_volts: uint16
        msg_bytes.extend(limits[0].to_bytes(2, byteorder='little', signed=False))
        # max_volts: int16
        msg_bytes.extend(limits[1].to_bytes(2, byteorder='little', signed=True))
        # charge_fraction: uint8
        msg_bytes.append(limits[2] & 0xFF) # Ensure it's within 0-255
        # capacity: uint16
        msg_bytes.extend(limits[3].to_bytes(2, byteorder='little', signed=False))
        # cut_off_current: uint8
        msg_bytes.append(limits[4] & 0xFF)
        # cycle_number: uint8
        msg_bytes.append(limits[5] & 0xFF)

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

if __name__ == "__main__":
    main()