EInkUSBController.py

"""
Copyright (c) 2025 Jon Cox (joncox123). All rights reserved.

WARNING: This software is provided "AS IS", without any warranty of any kind. It may contain bugs or other defects
that result in data loss, corruption, hardware damage or other issues. Use at your own risk.
It may temporarily or permanently render your hardware inoperable.
It may corrupt or damage the Embedded Controller or eInk T-CON controller in your laptop.
The author is not responsible for any damage, data loss or lost productivity caused by use of this software. 
By downloading and using this software you agree to these terms and acknowledge the risks involved.
"""

import usb.core
import usb.util
import struct
import random
import time


class EInkUSBController:
    """USB communication with E-Ink T-CON controller"""
    
    # USB Device IDs
    VID = 0x048d
    PID = 0x8957
    INTERFACE = 0
    ENDPOINT_OUT = 0x02
    ENDPOINT_IN = 0x81
    TIMEOUT = 5000  # milliseconds
    
    # Packet sizes
    CBW_LEN = 31
    CSW_LEN = 13
    PAYLOAD_LEN = 36
    
    # CBW template (Command Block Wrapper)
    CBW_TEMPLATE = bytes([
        0x55, 0x53, 0x42, 0x43, 0x00, 0x00, 0x00, 0x00,
        0x24, 0x00, 0x00, 0x00, 0x00, 0x00, 0x10,
        0xFE, 0x00, 0x00, 0x00, 0x00, 0x00, 0xA8, 0x00,
        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00
    ])
    
    # Payload sequences from enable-eink.c
    ENABLE_EINK = [
        bytes([0x92,0x00,0x5B,0x57,0x40,0x00,0x01,0x00,0x00,0x00,0x00,0x0A,0x00,0x00,0x40,0x06,0x40,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0xA8,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0x40,0x06,0x00,0x00]),
        bytes([0x60,0x00,0x5B,0x57,0xFD,0x7F,0x00,0x00,0x00,0x00,0x5B,0x57,0xFD,0x7F,0x00,0x00,0x03,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0xF0,0xF3,0xDF,0xD6]),
        bytes([0x4B,0x00,0x5B,0x57,0x00,0x00,0x06,0x00,0x06,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x70,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0xF0,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0xF0,0xF6,0xDF,0xD6]),
        bytes([0x61,0x00,0x5B,0x57,0x03,0x00,0x01,0x00,0x10,0x09,0x6C,0xE2,0xF6,0x7F,0x00,0x00,0x70,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0xF0,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0xF0,0xF6,0xDF,0xD6]),
        bytes([0x34,0x00,0x5B,0x57,0x00,0x00,0x00,0x00,0x00,0x0A,0x40,0x06,0x05,0x00,0xD0,0x75,0x76,0x02,0x00,0x00,0x00,0x00,0x00,0x00,0x40,0x06,0x00,0x00,0x00,0x00,0x00,0x00,0xF0,0xF6,0xDF,0xD6]),
        bytes([0x60,0x00,0x5B,0x57,0xFD,0x7F,0x00,0x00,0x00,0x00,0x5B,0x57,0xFD,0x7F,0x00,0x00,0x02,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0xF0,0xF3,0xDF,0xD6]),
        bytes([0x4B,0x00,0x5B,0x57,0x01,0x00,0x06,0x00,0x06,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x70,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0xF0,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0xF0,0xF6,0xDF,0xD6]),
        bytes([0x61,0x00,0x5B,0x57,0x02,0x00,0x01,0x00,0x10,0x09,0x6C,0xE2,0xF6,0x7F,0x00,0x00,0x70,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0xF0,0xF3,0xDF,0xD6,0x03,0x00,0x00,0x00,0xF0,0xF6,0xDF,0xD6]),
        bytes([0x9C,0x00,0x5B,0x57,0x01,0x00,0x57,0x89,0x03,0x00,0x0F,0x00,0x03,0x00,0x00,0x00,0x03,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x0F,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0xF0,0xF6,0xDF,0xD6]),
        bytes([0x60,0x00,0x5B,0x57,0xFD,0x7F,0x00,0x00,0x00,0x00,0x5B,0x57,0xFD,0x7F,0x00,0x00,0x02,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0xF0,0xF3,0xDF,0xD6]),
        bytes([0x9D,0x00,0x5B,0x57,0x01,0x00,0x01,0x00,0x68,0xF9,0xDF,0xD6,0x03,0x00,0x00,0x00,0x10,0x09,0x6C,0xE2,0xF6,0x7F,0x00,0x00,0x01,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0xF0,0xF6,0xDF,0xD6]),
        bytes([0x60,0x00,0x5B,0x57,0xFD,0x7F,0x00,0x00,0x00,0x00,0x5B,0x57,0xFD,0x7F,0x00,0x00,0x10,0x09,0x6C,0xE2,0xF6,0x7F,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x70,0xF3,0xDF,0xD6]),
    ]
    
    DISABLE_EINK = [
        bytes([0x9c,0x00,0x9b,0xbe,0x01,0x00,0x57,0x89,0x00,0x00,0x9f,0xf6,0x5f,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x0a,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x40,0xf7,0x9f,0xf6]),
        bytes([0x60,0x00,0x9b,0xbe,0xfa,0x7f,0x00,0x00,0x00,0x00,0x9b,0xbe,0xfa,0x7f,0x00,0x00,0x03,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x40,0xf4,0x9f,0xf6]),
        bytes([0x4b,0x00,0x9b,0xbe,0x00,0x00,0x06,0x00,0x06,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0xc0,0xf3,0x9f,0xf6,0x5f,0x00,0x00,0x00,0x40,0xf4,0x9f,0xf6,0x5f,0x00,0x00,0x00,0x40,0xf7,0x9f,0xf6]),
        # The disable video command:
        bytes([0x61,0x00,0x9b,0xbe,0x03,0x00,0x01,0x00,0x10,0x09,0xd9,0xd3,0xf6,0x7f,0x00,0x00,0xc0,0xf3,0x9f,0xf6,0x5f,0x00,0x00,0x00,0x40,0xf4,0x9f,0xf6,0x5f,0x00,0x00,0x00,0x40,0xf7,0x9f,0xf6]),
        bytes([0x92,0x00,0x9b,0xbe,0x70,0x00,0x01,0x00,0x00,0x00,0x00,0x0a,0x00,0x00,0x40,0x06,0x70,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0xf8,0xf3,0x9f,0xf6,0x5f,0x00,0x00,0x00,0x40,0x06,0x00,0x00]),
        # the show thinkbook logo command:
        # bytes([0x34,0x00,0x9b,0xbe,0x00,0x00,0x00,0x00,0x00,0x0a,0x40,0x06,0x05,0x00,0xd0,0xf5,0x37,0x02,0x00,0x00,0x00,0x00,0x00,0x00,0x40,0x06,0x00,0x00,0x00,0x00,0x00,0x00,0x40,0xf7,0x9f,0xf6]),
        bytes([0x9d,0x00,0x9b,0xbe,0x01,0x00,0x00,0x00,0xb8,0xf9,0x9f,0xf6,0x5f,0x00,0x00,0x00,0x10,0x09,0xd9,0xd3,0xf6,0x7f,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x40,0xf7,0x9f,0xf6]),
    ]
    
    REFRESH_FULL = [
        bytes([0x60,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x63,0x00,0x87,0x13,0x00,0x00,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x60,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x63,0x00,0x87,0x13,0x00,0x00,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x60,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x63,0x00,0x87,0x13,0x00,0x00,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x60,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x63,0x00,0x87,0x13,0x00,0x00,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x60,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
        bytes([0x63,0x00,0x87,0x13,0x00,0x00,0x00,0x00,0x00,0x00,0x87,0x13,0xFC,0x7F,0x00,0x00,0x98,0xF5,0x9F,0xE6,0x75,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00]),
    ]

    # Set Dynamic Mode (fast refresh mode)
    SET_DYNAMIC_MODE = [
        bytes([0x60,0x00,0x2C,0x8A,0xFC,0x7F,0x00,0x00,0x00,0x00,0x2C,0x8A,0xFC,0x7F,0x00,0x00,0x02,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x90,0xF1,0xAF,0x48]),
        bytes([0x4B,0x00,0x2C,0x8A,0x01,0x00,0x06,0x00,0x06,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x10,0xF1,0xAF,0x48,0x25,0x00,0x00,0x00,0x90,0xF1,0xAF,0x48,0x25,0x00,0x00,0x00,0x90,0xF4,0xAF,0x48]),
        bytes([0x61,0x00,0x2C,0x8A,0x02,0x00,0x01,0x00,0x10,0x09,0x68,0x09,0xF6,0x7F,0x00,0x00,0x10,0xF1,0xAF,0x48,0x25,0x00,0x00,0x00,0x90,0xF1,0xAF,0x48,0x25,0x00,0x00,0x00,0x90,0xF4,0xAF,0x48]),
    ]

    # Set Reading Mode (high-quality refresh mode)
    SET_READING_MODE = [
        bytes([0x60,0x00,0x2C,0x8A,0xFC,0x7F,0x00,0x00,0x00,0x00,0x2C,0x8A,0xFC,0x7F,0x00,0x00,0xF0,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x74,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x90,0xF1,0xAF,0x48]),
        bytes([0x4B,0x00,0x2C,0x8A,0x01,0x00,0x06,0x00,0x06,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x10,0xF1,0xAF,0x48,0x25,0x00,0x00,0x00,0x90,0xF1,0xAF,0x48,0x25,0x00,0x00,0x00,0x90,0xF4,0xAF,0x48]),
        bytes([0x61,0x00,0x2C,0x8A,0xF0,0x00,0x01,0x00,0x10,0x09,0x68,0x09,0xF6,0x7F,0x00,0x00,0x10,0xF1,0xAF,0x48,0x25,0x00,0x00,0x00,0x90,0xF1,0xAF,0x48,0x25,0x00,0x00,0x00,0x90,0xF4,0xAF,0x48]),
        bytes([0x9A,0x00,0x2C,0x8A,0x01,0x00,0x00,0x00,0x01,0x00,0x01,0x00,0x01,0x00,0x00,0x00,0x10,0x09,0x68,0x09,0xF6,0x7F,0x00,0x00,0x01,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x90,0xF4,0xAF,0x48]),
    ]
    
    def __init__(self, logger):
        self.logger = logger
        self.dev = None
        self.ctx = None
        
    def connect(self):
        """Initialize USB connection to E-Ink device"""
        try:
            # Find the device
            self.dev = usb.core.find(idVendor=self.VID, idProduct=self.PID)
            if self.dev is None:
                raise RuntimeError(f"E-Ink USB device not found (VID=0x{self.VID:04x}, PID=0x{self.PID:04x})")
            
            # Detach kernel driver if active
            if self.dev.is_kernel_driver_active(self.INTERFACE):
                self.logger.info("Detaching kernel driver from interface")
                self.dev.detach_kernel_driver(self.INTERFACE)
            
            # Claim the interface
            usb.util.claim_interface(self.dev, self.INTERFACE)
            self.logger.info(f"Connected to E-Ink device: VID=0x{self.VID:04x}, PID=0x{self.PID:04x}")
            
        except Exception as e:
            self.logger.error(f"Failed to connect to USB device: {e}")
            raise
    
    def disconnect(self):
        """Release USB connection"""
        if self.dev:
            try:
                usb.util.release_interface(self.dev, self.INTERFACE)
                # Reattach kernel driver
                self.dev.attach_kernel_driver(self.INTERFACE)
                self.logger.info("Disconnected from E-Ink device")
            except Exception as e:
                self.logger.warning(f"Error during USB disconnect: {e}")
    
    def _send_payload(self, payload):
        """Send a single payload command via USB bulk transfer"""
        if not self.dev:
            raise RuntimeError("USB device not connected")
        
        # Create CBW with random tag
        cbw = bytearray(self.CBW_TEMPLATE)
        tag = random.randint(0, 0xFFFFFFFF)
        struct.pack_into('<I', cbw, 4, tag)
        
        # Send CBW (Command Block Wrapper)
        try:
            transferred = self.dev.write(self.ENDPOINT_OUT, cbw, self.TIMEOUT)
            if transferred != self.CBW_LEN:
                raise IOError(f"CBW transfer incomplete: {transferred}/{self.CBW_LEN} bytes")
        except usb.core.USBError as e:
            raise IOError(f"CBW send failed: {e}")
        
        # Send payload
        try:
            transferred = self.dev.write(self.ENDPOINT_OUT, payload, self.TIMEOUT)
            if transferred != self.PAYLOAD_LEN:
                raise IOError(f"Payload transfer incomplete: {transferred}/{self.PAYLOAD_LEN} bytes")
        except usb.core.USBError as e:
            raise IOError(f"Payload send failed: {e}")
        
        # Read CSW (Command Status Wrapper)
        try:
            csw = self.dev.read(self.ENDPOINT_IN, self.CSW_LEN, self.TIMEOUT)
            if len(csw) != self.CSW_LEN:
                raise IOError(f"CSW read incomplete: {len(csw)}/{self.CSW_LEN} bytes")
            
            # Check status byte (byte 12)
            if csw[12] != 0:
                raise IOError(f"CSW status error: 0x{csw[12]:02x}")
                
        except usb.core.USBError as e:
            raise IOError(f"CSW receive failed: {e}")
        
        return True
    
    def _send_sequence(self, payloads, description):
        """Send a sequence of payloads"""
        self.logger.info(f"Sending {description} ({len(payloads)} commands)")
        
        for i, payload in enumerate(payloads):
            try:
                self._send_payload(payload)
            except Exception as e:
                raise IOError(f"Failed at command {i+1}/{len(payloads)}: {e}")
        
        self.logger.info(f"{description} completed successfully")
        return True
    
    def enable_eink(self):
        """Enable E-Ink display mode"""
        response = self._send_sequence(self.ENABLE_EINK, "Enable E-Ink")
        time.sleep(1.0)  # Wait for the eink to fully enable
        return response

    def disable_eink(self):
        """Disable E-Ink display mode (switch to OLED)"""
        response = self._send_sequence(self.DISABLE_EINK, "Disable E-Ink")
        time.sleep(1.0)  # Wait for the eink to fully disable
        return response
    
    def refresh_full(self):
        """Perform full E-Ink refresh (clears ghosting)"""
        return self._send_sequence(self.REFRESH_FULL, "Full refresh")

    def set_dynamic_mode(self):
        """Set E-Ink to Dynamic Mode (fast refresh mode)"""
        return self._send_sequence(self.SET_DYNAMIC_MODE, "Set Dynamic Mode")

    def set_reading_mode(self):
        """Set E-Ink to Reading Mode (high-quality refresh mode)"""
        return self._send_sequence(self.SET_READING_MODE, "Set Reading Mode")