repos_gamma/Python/mokaSerail.py

# -*- coding: UTF-8 -*-
import gzip
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 MokaParse():
    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 = b'', 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 = MokaParse()
        package = cmd.parseCommand(head, command, subCommand, data, returnParam, FEFlag)
        if self.write(package):
            package = self.read()
            return cmd.parseResult(package)

        return False


    '''生成gamma.gzip切割文件'''
    def gen_gzip(self, gm_file, save_file):
        # 压缩gamma.ini成zip;
        fp_ini = open(gm_file, 'rb')
        gf = gzip.open('gamma_org.gzip', 'wb')
        data = fp_ini.read()
        gf.write(data)
        fp_ini.close()
        gf.close()

        # 重新读取gzip，再切掉8个字节?后再保存
        fg = open('gamma_org.gzip', 'rb')
        g_data = fg.read()
        fg.close()

        # 生成新的gamma.zip.
        fg = open(save_file, 'wb+')
        start_index = 10 + 'gamma_org.gzip'.__len__() + 1
        data_save = g_data[start_index:-4]
        fg.write(g_data[8:10])
        fg.write(data_save)
        fg.close()


    '''发送pattern'''
    def send_parttern(self, rgb: list[int]):
        cmd = MokaParse()
        package = cmd.parseCommand(0xAA, [0x28], [], bytearray(rgb))
        if self.write(package):
            package = self.read()
            return cmd.parseResult(package)

        return False

    '''发送gamma文件'''
    def send_gamma(self, file_path: str):
        fp = open(file_path, 'rb')
        cmd = MokaParse()
        data = fp.read()
        package = cmd.parseCommand(0xAA, [0xE9], [0x02], data, False, True)
        if self.write(package):
            package = self.read()
            return cmd.parseResult(package)

    '''激活gamma文件'''
    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 = MokaParse()
        package = cmd.parseCommand(0xAA, [0x99], [0x06], bytearray(CRC_GM))
        if self.write(package):
            package = self.read()
            return cmd.parseResult(package)

    '''进工厂模式'''
    def enterFactory(self):
        return self.sendcmdEx(0xAA, [0x10], [0x01])

    '''白平衡初始化'''
    def initWhiteBalance(self):
        return self.sendcmdEx(0xAA, [0x16], [0x01])

    '''关闭localdimming'''
    def closeLocaldimming(self):
        return self.sendcmdEx(0xAA, [0x9F,0x07], [0x00])

    '''打开内置pattern'''
    def openBuiltInPattern(self):
        return self.sendcmdEx(0xAA, [0x27], [0x02])

    '''关闭内置pattern'''
    def closeBuiltInPattern(self):
        return self.sendcmdEx(0xAA, [0x27], [0x00])

    '''切换标准色温'''
    def switchStdColorTemperature(self):
        return self.sendcmdEx(0xAA, [0x31], [0x01])

    '''初始化gamma'''
    def initGamma(self):
        return self.sendcmdEx(0xAA, [0x9F,0x09], [0x01])

if __name__ == "__main__":
    # 打开TV串口;
    tv = MokaSerial()
    tv.open('COM7')

    print("进工厂模式")
    tv.enterFactory()

    print("白平衡初始化")
    tv.initWhiteBalance()

    print("关闭Localdimming")
    tv.closeLocaldimming()

    print("打开内置pattern")
    tv.openBuiltInPattern()

    print("切换标准色温")
    tv.switchStdColorTemperature()

    print("gamma初始化")
    tv.initGamma()

    print("读取 mac")
    data = tv.sendcmdEx(0xAA, [0xBE], [0x00])

    print("读取 wifimac")
    data = tv.sendcmdEx(0xAA, [0xBE], [0x15])

    print("读取 clientType")
    data = tv.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.send_parttern([int(item[0]), int(item[1]), int(item[2])])
            time.sleep(0.5)

    print("关闭内置pattern")
    tv.closeBuiltInPattern()