# -*- coding: UTF-8 -*- from xmlrpc.client import Boolean import serial import time import binascii from baseSerail import BaseSerial # CRC16矩阵表; CRC_TABLE = [ 0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50A5, 0x60C6, 0x70E7, 0x8108, 0x9129, 0xA14A, 0xB16B, 0xC18C, 0xD1AD, 0xE1CE, 0xF1EF ] # 计算CRC16值; def crc16(bys, bysLen): uwCRC, ucTemp = 0xFFFF, 0x00 index = 0 while index < bysLen: ucTemp = (uwCRC >> 0x0C) & 0xFF uwCRC <<= 4 uwCRC ^= CRC_TABLE[ucTemp ^ ((bys[index]) >> 0x04)] uwCRC &= 0xFFFF ucTemp = (uwCRC >> 0x0C) & 0xFF uwCRC <<= 4 uwCRC ^= CRC_TABLE[ucTemp ^ (bys[index] & 0x0F)] uwCRC &= 0xFFFF index += 1 return uwCRC # 包头码（包引导码） PHeader = { "TV_Debug": 0xAA, "TV_Return": 0xAB, "TV_Panel_Debug": 0xAC, "TV_Panel_Return": 0xAD, "TV_Debug_Other": 0xAE, "TV_Other_Return": 0xAF, } # 命令结果码; RCode = { "RC_OK": 0x0A, # 命令执行通过; "RC_ERR": 0x0E, # 命令错误或无法执行; "RC_LOSE": 0x0F, # 命令丢失（链路层出错）; } # 命令的封装与解析; class MokaComman(): def __init__(self): # 头引导码(默认1字节,0x86命令二字节); self.Header = [PHeader['TV_Debug']] # 包长(默认1字节,0x86长度二字节); self.Length = [0x00] # 命令码,1字节(int类型); self.Command = [0x00] # 子命令参数; self.SubCommand = [] # 数据,bytearray格式; self.Data = [] # crch,1字节; self.CRCH = 0xFF # crcl,1字节; self.CRCL = 0xFF # 是否是特殊命令; self.FEFlag = False # 是否是多参数命令;(同一个Command，有多个含义的参数) self.isMultipleParams = False # 正确执行后的结果值，bytearray格式的二维数组; self.successData = [] # 生成命令; ''' 描述：将参数根据协议格式转化为协议指令; 参数： head：协议包引导头; command:协议命令码; data:协议命令码数据; FEFlag:协议特殊标记，协议包引导头是否有2个; 注意：data必须是bytearray格式传递; 返回： ''' def parseCommand(self, head: int, command: list[int], subCommand: list[int], data: bytearray, isMultipleParams=False, FEFlag=False): self.Header[0] = head self.Command = command self.SubCommand = subCommand self.Data = data self.FEFlag = FEFlag self.isMultipleParams = isMultipleParams if FEFlag: self.Header.append(0xFE) # 注意：4 = crch + crcl + lenh + lenl length = 4 + self.Header.__len__() + self.Data.__len__() + self.Command.__len__() + self.SubCommand.__len__() # 高字节; self.Length[0] = length >> 8 # 低字节; self.Length.append(length & 0xFF) else: # 注意：3 = crch + crcl + len self.Length[0] = 3 + self.Header.__len__() + self.Data.__len__() + self.Command.__len__() + self.SubCommand.__len__() # 生成package; package = bytearray(self.Header + self.Length + self.Command + self.SubCommand) + self.Data # 计算crc crc = crc16(package, package.__len__()) # 高字节; self.CRCH = crc >> 8 # 低字节; self.CRCL = crc & 0xFF # 形成最终的命令; package = bytearray(self.Header + self.Length + self.Command + self.SubCommand) + self.Data + bytearray([self.CRCH, self.CRCL]) return package ''' 描述：解析读串口数据是否正确返回,若正确返回剥离出结果值; 参数：data 16进制的字符串; 返回： ''' def parseResult(self, data): # 将16进制字符串转成字节数组; data = bytearray(data) # 检测长度; if data.__len__() < 5: return False # 根据请求的引导码,识别对应的返回码; retCode = 0 if self.Header[0] == PHeader['TV_Debug']: retCode = PHeader['TV_Return'] elif self.Header[0] == PHeader['TV_Panel_Debug']: retCode = PHeader['TV_Panel_Return'] elif self.Header[0] == PHeader['TV_Debug_Other']: retCode = PHeader['TV_Other_Return'] # 检测包引导码; if retCode != data[0]: return False package = [] tooken_len = 0 while True: if tooken_len >= data.__len__(): break # 取出包长; if self.FEFlag: package_len = data[tooken_len + 1] << 8 + data[tooken_len + 2] else: package_len = data[tooken_len + 1] # 取出整包; package = data[tooken_len:tooken_len + package_len] # 再次验证包头是否有效; if package[0] != retCode: print('Incorrect package head!\n') return False # 检测crc是否正确; crc = crc16(package, package_len - 2) CRCH = crc >> 8 CRCL = crc & 0xFF if CRCH != package[-2] and CRCL != package[-1]: return False # 第一次,检测结果码(第三字节); if tooken_len == 0: if package[2] != RCode['RC_OK']: return False else: # 第二次，检测命令码 if self.Command[0] == 0xFC: pass else: if package[2] - 1 != self.Command[0]: return False if package[2] == 0xFE: #返回超过255 self.successData.append(package[5:-2]) else: self.successData.append(package[3:-2]) tooken_len += package_len print('successData:', binascii.b2a_hex(bytearray(self.Command)), self.successData) return True class MokaSerial(BaseSerial): def __init__(self): BaseSerial.__init__(self) def __del__(self): self.close() # 用于发命令前,检测串口是否打开; def checkport(self): if not self.ser.is_open: self.reOpen() return self.ser.is_open ''' 发送普通命令 ''' def sendcmd(self, cmd: list[int]): self.write(bytearray(cmd)) return self.read() def sendcmdEx(self, head: int, command: list[int], subCommand: list[int], data: bytearray = b'', FEFlag: Boolean = False, returnParam: Boolean = False): cmd = MokaComman() package = cmd.parseCommand(head, command, subCommand, data, returnParam, FEFlag) if self.write(package): package = self.read() return cmd.parseResult(package) return False '''发送pattern''' def send_parttern(self, rgb: list[int]): cmd = MokaComman() package = cmd.parseCommand(0xAA, [0x28], [], bytearray(rgb)) if self.write(package): package = self.read() return cmd.parseResult(package) return False def send_gamma(self, file_path: str): fp = open(file_path, 'rb') cmd = MokaComman() data = fp.read() package = cmd.parseCommand(0xAA, [0xE9], [0x02], data, False, True) if self.write(package): package = self.read() return cmd.parseResult(package) def send_gamma_active(self, ini_file: str): ini_fp = open(ini_file, 'rb') ind_data = ini_fp.read() ini_fp.close() # 计算crc crc = crc16(ind_data, ind_data.__len__()) CRC_GM = [crc >> 8, crc & 0xFF] cmd = MokaComman() package = cmd.parseCommand(0xAA, [0x99], [0x06], bytearray(CRC_GM)) if self.write(package): package = self.read() return cmd.parseResult(package) if __name__ == "__main__": # 打开TV串口; tv_port = MokaSerial() tv_port.open('COM7') print("进工厂模式") data = tv_port.sendcmd([0xAA, 0x06, 0x10, 0x01, 0xA7, 0xEF]) print("白平衡初始化") data = tv_port.sendcmd([0xAA, 0x06, 0x16, 0x01, 0x0D, 0x49]) print("关闭Localdimming") data = tv_port.sendcmd([0xAA, 0x07, 0x9F, 0x07, 0x00, 0xE1, 0x74]) print("打开内置pattern") data = tv_port.sendcmd([0xAA, 0x06, 0x27, 0x02, 0x0B, 0x8E]) print("切换标准色温") data = tv_port.sendcmd([0xAA, 0x06, 0x31, 0x01, 0x92, 0x38]) print("gamma初始化") data = tv_port.sendcmd([0xAA, 0x07, 0x9F, 0x09, 0x01, 0xD2, 0x5A]) print("读取 mac") data = tv_port.sendcmdEx(0xAA, [0xBE], [0x00]) print("读取 wifimac") data = tv_port.sendcmdEx(0xAA, [0xBE], [0x15]) print("读取 clientType") data = tv_port.sendcmdEx(0xAA, [0x8C], [0x00]) for item in [[0,0,0],[26,26,26],[51,51,51],[76,76,76],[102,102,102],[127,127,127],[153,153,153],[178,178,178],[204,204,204],[229,229,229],[255,255,255],[0,0,255],[255,0,0]]: if item.__len__() == 3: tv_port.send_parttern([int(item[0]), int(item[1]), int(item[2])]) time.sleep(0.5) print("关闭内置pattern") data = tv_port.sendcmd([0xAA, 0x06, 0x27, 0x00, 0x2B, 0xCC])