LG DDC/CI control via python hidapi

lg_switch_inputs.py

import struct
import time
import hid
import sys
import argparse

#
# Constants
#

# Activate PBP
DDC_51_PBP = 0xD7
DDC_50_INPUT_SWITCH = 0xF4

# DDC_51_PBP
PBP_NONE = 0x1
PBP_LR_UNK2 = 0x2
PBP_LR_66_33 = 0x3
PBP_LR_50_50 = 0x5

# DDC_50_INPUT_SWITCH
MONITOR_AUTO = 0x0
MONITOR_DP1 = 0xd0
MONITOR_DP2 = 0xd1
MONITOR_DP3 = 0xd2
MONITOR_USB_C = 0xd2
MONITOR_HDMI1 = 0x90
MONITOR_HDMI2 = 0x91

#
# Helpers
#
def msg_checksum(msg):
    sum = 0x6E^0x50
    for i in range(0, len(msg)):
        sum ^= msg[i]
    return sum

def msg_add_checksum_2(msg):
    sum = 0x6E
    for i in range(0, len(msg)):
        sum ^= msg[i]
    msg += [sum]
    return msg

def hex_dump(b, prefix=""):
    p = prefix
    b = bytes(b)
    for i in range(0, len(b)):
        if i != 0 and i % 16 == 0:
            print (p)
            p = prefix
        p += ("%02x " % b[i])
    print (p)
    print ("")

class UsbHost:

    def __init__(self):
        # USB
        self.has_usb = False
        self.dev = None
        self.ep_in = None
        self.ep_out = None

    def init_usb(self):
        # find our device
        self.dev = hid.device()
        self.dev.open(0x043E, 0x9a39)

        self.has_usb = True

    def send_raw(self, pkt):
        if not self.has_usb:
            return

        try:
            self.dev.write(bytes(pkt + [0] * (0x40 - len(pkt))))
        except Exception as e:
            print ("Failed to write", e)

    def read_raw(self, amt=0x40):
        if not self.has_usb:
            return

        try:
            return bytes(self.dev.read(amt, 200))
        except Exception as e:
            print ("Failed to read", e)
        return []
    
    def wrap_send_ddc_51(self, data, expected_back=0xb):
        wrapped = [0x08, 0x01, 0x55, 0x03, 0x00, 0x00, 0x03, 0x37]
        
        data_len = len(data)
        data = [0x80 | data_len] + data
        data = [0x51] + data
        data = msg_add_checksum_2(data)
        wrapped[4] += len(data)
        wrapped += data
        
        #hex_dump(wrapped)

        self.send_raw(wrapped)
        wrapped[1] = 0x02
        wrapped[3] = 0x04
        wrapped[4] = expected_back
        wrapped[6] = 0x0b
        self.send_raw(wrapped)
        #hex_dump(wrapped)

        return wrapped
    
    def wrap_send_ddc_50(self, data, expected_back=0xb):
        wrapped = [0x08, 0x01, 0x55, 0x03, 0x00, 0x00, 0x03, 0x37]
        
        data_len = len(data)
        data = [0x80 | data_len] + data
        data = [0x50] + data
        data = msg_add_checksum_2(data)
        wrapped[4] += len(data)
        wrapped += data
        
        #hex_dump(wrapped)

        self.send_raw(wrapped)
        wrapped[1] = 0x02
        wrapped[3] = 0x04
        wrapped[4] = expected_back
        wrapped[6] = 0x0b
        self.send_raw(wrapped)
        #hex_dump(wrapped)

        return wrapped
    
    def get_vcp(self, idx):
        for i in range(0, 100):
        
            self.wrap_send_ddc_51([0x01, idx])
            
            data = device.read_raw()
            
            if (len(data) < 8):
                hex_dump(data)
                continue
            
            data_len = data[5] & 0x7F
            if data_len > len(data)-5-2:
                data_len =len(data)-5-2

            test = msg_checksum(data[5:5+data_len+2])
            #hex_dump(data[5:5+data_len+1])
            if (test == 0 and data[8] == idx):
                return data[13] | data[12] << 8
        return -1
    
    def lg_special(self, idx, val):
        for i in range(0, 1):

            self.wrap_send_ddc_50([0x03,idx,(val >> 8) & 0xFF, val & 0xFF], 0x26)
            
            data = device.read_raw()
            #hex_dump(data)
            if (len(data) < 8):
                hex_dump(data)
                continue
            
            data_len = data[5] & 0x7F
            if data_len > len(data)-5-2:
                data_len =len(data)-5-2

            test = msg_checksum(data[5:5+data_len+2])
            
            if (test == 0):
                return data[13]
        return -1
    
    def set_vcp(self, idx, val, val2=0):

        for i in range(0, 1):
        
            self.wrap_send_ddc_51([0x03, idx,(val >> 8) & 0xFF,(val >> 0) & 0xFF])
            
            data = device.read_raw()
            
            if (len(data) < 8):
                hex_dump(data)
                continue
            
            data_len = data[5] & 0x7F
            if data_len > len(data)-5-2:
                data_len =len(data)-5-2

            test = msg_checksum(data[5:5+data_len+2])
            
            if (test == 0 and data[8] == idx):
                return data[13]

        return -1

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
                    prog = 'switch_inputs.py',
                    description = 'Switches LG monitor inputs')
    parser.add_argument('--pbp', type=int, default=-1, help='Switch to a PBP mode (1 = none, 3 = 2/3 and 1/3 left-right, 5 = 1/2 and 1/2 left-right)')
    parser.add_argument('--input', default="", help='Switch to an input (valid values: hdmi1, hdmi2, dp1, dp2, dp3, usbc, usb-c)')
    parser.add_argument('--get_pbp', action='store_true', help='Get current PBP mode (0 = top-bottom, 1 = none, 3 = 2/3 and 1/3 left-right, 5 = 1/2 and 1/2 left-right)')
    args = parser.parse_args()
    
    device = UsbHost()
    device.init_usb()
    
    pbp_val = args.pbp
    input_val = -1
    
    # Switch monitor inputs
    valid_inputs = ["auto", "hdmi1", "hdmi2", "dp1", "dp2", "dp3", "usbc", "usb-c"]
    port = args.input.lower()
    if port in valid_inputs:
        if port == "auto":
            input_val = MONITOR_AUTO
        if port == "hdmi1":
            input_val = MONITOR_HDMI1
        elif port == "hdmi2":
            input_val = MONITOR_HDMI2
        elif port == "dp1":
            input_val = MONITOR_DP1
        elif port == "dp2":
            input_val = MONITOR_DP2
        elif port == "dp3":
            input_val = MONITOR_DP3
        elif port == "usbc" or port == "usb-c":
            input_val = MONITOR_USB_C
    elif port != "":
        print ("invalid input type `" + port + "`")
        exit(-1)

    # Switch PBP mode
    if pbp_val > 0:
        device.set_vcp(0xd7, pbp_val)
    
    if args.get_pbp:
        val = device.get_vcp(0xd7)
        print("pbp is", val)

    if input_val >= 0:
        device.lg_special(DDC_50_INPUT_SWITCH, input_val)
    
    if input_val < 0 and pbp_val < 0 and not args.get_pbp:
        parser.print_help()