ec718/ec_fullsdk/PLAT/testscript/script/opentest/testsys/PyEcLib/PyATLib.py

# -*- coding: utf-8 -*-
"""
Created on Fri August 30 12:04:02 2019

@author: eigencomm wqzhao
"""

#!/usr/bin/env python

#-*- coding:utf-8 -*-
import random
import time
import re
import winsound

LIB_VERSION = "PyLib 0.1"


import threading

class EcATCmdTester:

    __AT = b'AT\r\n'
    __CEREG_GET = b'AT+CEREG?\r\n'

    def __init__(self,SerialHandle,uart_timeout=0.1):
        self.__UartTimeoutVar = uart_timeout
        self.__ATWaitTimeout_Flag = 1
        self.__SerialHandle = SerialHandle
        self.__NeedInterLog = True
        self.__ATNeedHandShake = True
        self.__LogFileEnable = False

    def EC_Setting(self,need_handshake=True,need_interlog=True):
        self.__ATNeedHandShake = need_handshake
        self.__NeedInterLog = need_interlog

    def EC_LogtoFile(self, fileHandle , log_file_enable=False):
        self.__LogFileHandle = fileHandle
        self.__LogFileEnable = log_file_enable

    def EC_LOG(self,log_str):
        log_buf = str(time.asctime(time.localtime(time.time()))) + ":   " + log_str
        print(log_buf)
        if(self.__LogFileEnable == True):
            self.__LogFileHandle.write(log_buf + str('\n'))
            self.__LogFileHandle.flush()

    def __EC_TimeoutCb(self):
        self.__ATWaitTimeout_Flag = 0
        if self.__NeedInterLog:
            self.EC_LOG('At Timeout Trigger')
        self.ATimer_Handle.cancel()

    def __EC_StartATTimeout(self,at_timeout):
        self.__ATWaitTimeout_Flag = 1
        self.ATimer_Handle = threading.Timer(at_timeout,self.__EC_TimeoutCb)
        self.ATimer_Handle.start()

    def EC_PollString(self, find_str, timeout, err_str=''):
        self.__EC_StartATTimeout(timeout)
        while self.__ATWaitTimeout_Flag == 1:
            data = self.__SerialHandle.readline()
            if(data != b''):
                self.EC_LOG('PC<-UE: ' + str(data))
                result = re.search(find_str, str(data))
                if( result != None):
                    self.ATimer_Handle.cancel()
                    return True,data
                else:
                    if(err_str != ''):
                        result = re.search(err_str, str(data))
                        if(result != None):
                            self.ATimer_Handle.cancel()
                            return False, data
        if self.__NeedInterLog:
            self.EC_LOG('Polling String Timeout')
        return False,'';

    def __EC_PollOK(self):
        for i in range(1, 30):
            data = self.__SerialHandle.readline()
            if(data == b'OK\r\n'):
                self.EC_LOG('PC<-UE: ' + str(data))
                return True;
            elif(data != b''):
                self = self
                self.EC_LOG('PC<-UE: ' + str(data))
        return False;

    def EC_CheckNetReady(self,timeout):
        self.EC_UartSend(self.__CEREG_GET)
        if(self.EC_PollString('CEREG: \d,1',timeout)[0] == True):
            return True;
        else:
            return False;

    def EC_PollNetReady(self):
        self.EC_UartSend(self.__CEREG_GET)
        if(self.EC_PollString('CEREG: \d,1',2)[0] == True):
            return True;
        else:
            return False;

    def __EC_Handshake(self):
        if(self.__ATNeedHandShake == False):
            return True

        tmp = self.__SerialHandle.timeout
        self.__SerialHandle.timeout = 0.1
        self.__SerialHandle.flushInput()
        self.__SerialHandle.flushOutput()
        for i in range(1, 100):
            self.__SerialHandle.write(self.__AT)
            if(self.__EC_PollOK() == True):
                self.__SerialHandle.timeout = tmp
                return True
        self.__SerialHandle.timeout = tmp

        return False;

    def EC_UartSend(self,at_str):
        self.__SerialHandle.timeout = self.__UartTimeoutVar
        if(self.__EC_Handshake()):
            time.sleep(0.2)
            self.__SerialHandle.flushInput()
            self.__SerialHandle.flushOutput()
            self.__SerialHandle.write(at_str)
            if self.__NeedInterLog:
                self.EC_LOG('PC->UE: '+str(at_str))
        else:
            if self.__NeedInterLog:
                self.EC_LOG('Wakeup Handshake Timeout')

    def EC_Version(self):
        self.EC_LOG(str(LIB_VERSION))

    def EC_RandomSlp(self, a, b):
        sleep_time = random.randint(a, b);
        self.EC_LOG('python sleep: ' + str(sleep_time) + ' Second')
        time.sleep(sleep_time)

    def EC_Beep(self, time):
        winsound.Beep(1000, time)

if __name__ == '__main__':
      run()