python 常用知识点介绍2_python serial库

1. serial库

安装：pip3 install pyserial

import threading
import serial
import time
 
def logPrinter(mesg='Nothing to log.', log_obs=1):
    if log_obs:
        mesg = time.strftime('[%Y.%m.%d %H:%M:%S] ', time.localtime(time.time())) + str(mesg)
        print(mesg)
 
 
class SerialCtrl():
    '''
        串口控制类 ，执行串口命令，并且通过多线程记录串口log
    '''
 
    def __init__(self, port, logfife='', baudrate=115200, log_obs=1):
        self.ignore = False  # 是否忽略串口log的检测
        self.threadLock = threading.Lock()  # 多线程锁
        self.logfile = logfife
        super().__init__()
        self.log_obs = log_obs
        self.port = port
        self.console = serial.Serial(port=self.port, baudrate=baudrate, timeout=0.1)
        self.serRunning = True
        if self.logfile == '':
            self.logfile = 'E:/log/%s_%s.log' % (time.strftime('%Y%m%d_%H%M%S', time.localtime(time.time())), self.port)
 
        self.crashlogFile = self.logfile.replace('.log', '') + '_crash.log'  # 串口log异常文件名
        logPrinter(mesg='Create Log File "%s"' % self.logfile, log_obs=self.log_obs)
 
        # 守护线程，记录串口log
        self.T1 = threading.Thread(target=self.LogtoFile)
        self.T1.daemon = True
        self.T1.start()
 
    def Analysis_log(self, line):
        # 分析串口log是否有异常
        keywords = ['crash',
                    'oom']
        excludes = [
                    ': "crash_syslog"',
                    ]
 
        for k in keywords:
            if k in line:
                for exekey in excludes:
                    if exekey in line:
                        return ''
                line = time.strftime('[%Y.%m.%d %H:%M:%S]', time.localtime(time.time())) + line.replace('\n','').replace(
                    '\r', '') + '\n'
                with open(self.crashlogFile, 'a') as f:
                    line = '[warning from test]' + line
                    f.write(line)  # log异常
                return '[warning from test]'
 
        return ''
 
    def LogtoFile(self):
        # 调用子线程，记录串口log
        while self.serRunning:
            self.threadLock.acquire()
            line = self.console.readline()
            self.threadLock.release()
            try:
                line = line.decode('ascii')
                self.writeline(line)
            except Exception as e:
                print(e)
 
    def writeline(self, line):
        # 将串口log写进文件中， 同时进行log的分析
        if not self.ignore:
            # 检测串口log
            ii = self.Analysis_log(line)
        else:
            ii = '[Ignore]'
 
        with open(self.logfile, 'a+') as f:
            if line and line != '\r\n' and line != '\r':
                line = line.replace('\n', '').replace('\r', '') + '\n'
                line = time.strftime('[%Y.%m.%d %H:%M:%S]', time.localtime(time.time())) + ii + line
                f.writelines(line)
 
    def command(self, com):
        # 执行串口命令
 
        com = com + '\n'
        com = com.encode('utf-8')
 
        self.threadLock.acquire()
        self.console.write(com)
        time.sleep(1)
        output = self.console.readall()
        self.threadLock.release()
 
        # 返回输出结果
        try:
            output = output.decode('utf-8')
            logPrinter(log_obs=self.log_obs, mesg='[COM %s] %s' % (self.port, output))
            output_list = output.split('\n')
            for line in output_list:
                self.writeline(line)
            for k in range(len(output_list)):
                output_list[k] = output_list[k].replace('\r', '')
            return output_list
        except Exception as e:
            logPrinter(mesg=str(e))
            return ''
 
    def close(self):
        logPrinter(log_obs=self.log_obs,  mesg='[%s] Close SerialCtrl of DUT, COM %s' % (self.__class__.__name__, self.port))
        self.serRunning = False
        self.console.close()
 
 
if __name__ =='__main__':
    ser = SerialCtrl(port='com19')
    ser.command('reboot')
    time.sleep(100)
    ser.command("\004") # 模拟输入ctrl + D
    
    # https://blog.51cto.com/u_19261/6599846

2. python3中 str和bytes类型的相互转换

    s1='你好'
    print(type(s1)) # <class 'str'>
 
    # str --->bytes 通过encode()或者bytes()函数来进行
    s2=s1.encode(encoding='gbk')
    print(s2) #输出 b'\xc4\xe3\xba\xc3'
    s3=bytes(s1,encoding='utf-8')
    print(s3) # 输出 b'\xe4\xbd\xa0\xe5\xa5\xbd'
 
 
    # bytes--->str  通过decode()或者str()函数来进行
    print(str(s2,encoding='gbk'))   # 输出 你好
    print(s3.decode(encoding='utf-8')) # 输出 你好
 

3， python 操作mysql数据库

import pymysql
#打开数据库连接
conn = pymysql.connect(host='127.0.0.1', user ="root", passwd ="root", db ="test")
cursor=conn.cursor()
# 查询数据
cursor.execute("select * from wifi where include='yes'") # 执行mysql
res=cursor.fetchone() # 读取一行,返回的数据是元组
while res:
    print(res)
    res = cursor.fetchone()
 
# 插入一条记录
cursor.execute("insert into newrper_mirouters(id,mi_routers) Values (%s, '%s')" % (id, k))
conn.commit()
 
# 断开连接
conn.close()
 

4，json和字典区别

• python中没有json这种数据类型
• 在python，json中被识别为str类型，字典是dict类型
• json的格式和dict有点相似，都是通过键值对的方式构成的，但是也有不同的地方：
•     json的key只能是字符串，python的dict的key必须是不可变类型（str，数字，元组）
•     json的字符串强制双引号，dict字符串可以单引号、双引号。

json和dict之间可以转换：

json.dumps()  可以将字典转化为json格式

json.loads() 可以将json转化为字典格式

import json
 
dict1={'A':-1,'B':8,'c':4}
str1='{"ee":1,"rr":2}'
 
# d1=json.dumps(dict1)  # 将字典转化为json格式
# print type(d1)  # 输出str
# print d1     # 输出{"A": -1, "c": 4, "B": 8}
 
d2=json.loads(str1)
print type(d2) # 输出<type 'dict'>
print d2   # 输出{u'ee': 1, u'rr': 2}

5，通过python脚本执行cmd命令的方法

可以通过subprocess或者os模块中的方法来实现

subprocess的目的就是启动一个新的进程并且与之通信，可以用来来替代os.system, os.popen命令。

subprocess模块常用的函数如下：

1, subprocess.call()

# 执行命令，并且返回状态码，多个参数用列表进行传输，retcode = call(["ls", "-l"])

也可以用字符串作为参数运行命令(通过设置参数shell=True)

a=subprocess.call(['ping','qq.com'])
print(a)  # 输出0
print(type(a)) # int

2,subprocess.popen(args,shell=False,stdout=None)

args: 表示要执行的命令，可以是str类型，也可以是数组类型
shell=True ,表示在系统默认的shell环境中执行新的进程，此shell在windows表示为cmd.exe，在linux为/bin/sh，args是字符串类型的时候，shell需要为True
stdin, stdout and stderr: 用来指定标准输入、标准输出和错误输出 常用值 subprocess.PIPE，结合stdout.readline()可以得到输出结果

a=subprocess.Popen(args='ping qq.com',shell=True,stdout=subprocess.PIPE)
while a.poll() is None:  # poll()方法用来检查进程是否还在之执行中,None表示正在执行中
    line = a.stdout.readline().decode('gbk')
    if line!='':
        print(line)
print(a.poll()) # 输出0

def mysuprocess(com, timeout):
    p = subprocess.Popen(com, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
    try:
        out, err = p.communicate(timeout=timeout)
        logPrinter("子进程的标准输出：\n" + out.decode())
        return  out.decode()
    except TimeoutExpired:
        logPrinter('cmd命令超时')
        p.terminate()
        logPrinter("子进程超时退出！")
        return ''

3,subprocess.check_output(**kwargs)

执行命令并且将输出结果返回，如果命令执行出错，将会抛出一个CalledProcessError异常

a=subprocess.check_output('ipconfig')
print(a.decode('gbk')) #输出执行结果

4, communicate()方法

communicate()方法可以和子进程进行交互，发送数据到stdin，从stdout和stderr读取数据，

subp = subprocess.Popen('ping qq.com',shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
print(subp.communicate('-n 2'))  # 输出执行结果

注意：communicate 其实是循环读取管道中的数据（每次 32768 字节）并将其存在一个 list 里面，到最后将 list 中的所有数据连接起来（b''.join(list)） 返回给用户。

于是就出现了一个坑：如果子进程输出内容非常多甚至无限输出，则机器内存会被撑爆

解决办法：

将stdout重定向到文件中

f=open('aa.txt','a+',encoding='gbk')
subp = subprocess.Popen('ping qq.com -n 2',shell=True,stdout=f)
f.close()

6，通过python对比文件的异同

（1）通过文件的md5值来比较文件的不同

利用hashlib模块的md5方法，生成固定为16字节长度的字符串（用32位的16进制字符串来表示），通过对比两个文件的MD5值，来比较文件是否发生变化

import hashlib
def get_content_mtd(self,file_name):
        md5 = hashlib.md5(open(file_name,'rb').read()).hexdigest()
        return md5

（2）比较两个文件最后被修改的时间戳

os.stat() 方法用于在给定的路径上执行一个系统 stat 的调用，st_mtime返回最后一次修改的时间

    def get_modify_time(self,file_name):
        mtime= os.stat(file_name).st_mtime
        print(mtime)
        return mtime

7，利用python识别简单的验证码

环境搭建：

（1）安装tesseract-OCR，下载地址：Index of /tesseract

（2）安装完成后，需要配置环境变量，在path中添加tesseract的安装目录：E:\softinstall\Tesseract-OCR；在系统变量中添加TESSDATA_PREFIX；

（3）pip3 install pytesseract

from PIL import Image
import pytesseract
 
image = Image.open('aa.png')
text = pytesseract.image_to_string(image)
print(text)

注：这种方法只能是被简单，清晰地验证码；对于模糊的验证码，需要更复杂的处理（降噪、切割等）。

8， python时间戳相减

# ---------------- 两个时间戳相减 ---------------------
 
t1 = '2022-09-30 10:31:06'
print('t1 ', t1)
t_now = time.strftime('%Y-%m-%d %H:%M:%S')  # str
print('time now ', t_now)
# 两个时间字符串相减
td = datetime.datetime.strptime(t_now, '%Y-%m-%d %H:%M:%S') - datetime.datetime.strptime(t1, '%Y-%m-%d %H:%M:%S')
print('时间相减得到的时间差 ',td)  # datetime.timedelta
print('时间差中的天数 ', td.days) # int
print('时间差中的秒数', td.seconds) # int

输出：

t1  2022-09-30 10:31:06
time now  2022-09-30 15:34:19
时间相减得到的时间差  5:03:13
时间差中的天数  0
时间差中的秒数 18193

# ----------------------- 时间移动 -----------------
 
t1 = '2022-09-30 10:31:06'
print('t1: ', t1)
t_now = datetime.datetime.strptime(t1 ,'%Y-%m-%d %H:%M:%S') # <class 'datetime.datetime'>
print('当前时间：', t_now)
t2 = t_now + datetime.timedelta(days=-1, hours=1, seconds=1, minutes=2) # <class 'datetime.datetime'>
print('移动后的时间：', t2)
 

输出：

t1:  2022-09-30 10:31:06
当前时间： 2022-09-30 10:31:06
移动后的时间： 2022-09-29 11:33:07

9，python快速生成大文件

import time
 
t1 = time.time()
with open('2G', 'w') as f:
    f.seek(2*1024*1024*1024)   # 移动文件读取指针到指定位置
    f.write('123')
t2 = time.time()
print(t2-t1)  # 10s

10， BeautifulSoup分析html文件

import requests
import re
from bs4 import BeautifulSoup
 
res = requests.get('http://xx.xx.xx.xx:xx/file/test.html')
html = res.text
 
soup = BeautifulSoup(html, 'html.parser') #  html.parser是解析器类型
 
all_a = soup.find_all(name='a', string=re.compile("1"))  # find_all 获取所有的标签, re.compile("1")过滤包含1的标签
print(all_a)
ele = all_a[0]
 
print(ele.text)  # 获取标签的文本值
print(ele['href']) # 获取标签的href属性值

11. dir()

 print(dir('')) # 内置方法dir()用于列出对象的所有属性及方法

12. windows下获取网卡IPv4 /IPv6地址的方法

import psutil
 
def get_netcard_v6(ifname='office'):
    # 获取Ipv6地址
    info = psutil.net_if_addrs()
    for i in info[ifname]:
        print(i)
        if i[0] == 23 and 'fe80' not in i[1]:
            return i[1]
    raise Exception('获取Ipv6地址失败')
 
def get_netcard_v4(ifname='office'):
    # 获取Ipv4地址
    info = psutil.net_if_addrs()
    for i in info[ifname]:
        if i[0] == 2:
            return i[1]
    raise Exception('获取Ipv4地址失败')

13. python修改图片的尺寸

pip install pillow

from PIL import Image
 
 
root = '8e6558c0b6cbf00c.png'
pic = Image.open(root)
pic = pic.resize((500, 1000)) # 单位是像素，宽，高
pic.save('111.png)

14. python进行url编码和解码

from urllib import parse
 
 
option = {
    'A': '123',
    'B': '啦啦啦'
}
 
# urlencode进行编码，参数为dict类型，输出字符串，拼接键值对
o1 = parse.urlencode(option)
print(o1)
# 输出 A=123&B=%E5%95%A6%E5%95%A6%E5%95%A6
 
# quote进行编码， 参数为字符串， 输出为字符串，将&和=一起编码
o2 = parse.quote(o1)
print(o2)
# 输出 A%3D123%26B%3D%25E5%2595%25A6%25E5%2595%25A6%25E5%2595%25A6
 
# unquote进行url解码，输出字符串
o3 = parse.unquote(o2)
print(o3)
# 输出 A=123&B=%E5%95%A6%E5%95%A6%E5%95%A6
 
o4 = parse.quote(str(option))
print(o4)
# 输出%7B%27A%27%3A%20%27123%27%2C%20%27B%27%3A%20%27%E5%95%A6%E5%95%A6%E5%95%A6%27%7D
 
o5 = parse.unquote(o4)
print(o5)
# 输出{'A': '123', 'B': '啦啦啦'}

14、python logging模块

1、logging模块默认定义了日志的6个级别，分别为 CRIRICAL>ERROR>WARNING>INFO>DEBUG>NOTEST,

默认的日志级别是warning，低于warning的日志将不会被打印。

import logging
 
logging.warning(11)
logging.error(99)
logging.info('88')

输出：

WARNING:root:11
ERROR:root:99

2、设置日志打印级别、日志的格式

basicConfig方法可以设置日志的级别、日志的格式

import logging
 
logging.basicConfig(
    level=logging.NOTSET,
    format="%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s"
)
 
logging.info(123)

输出

2024-03-21 15:29:36,626 - debug.py[line:14] - INFO: 123

3、进阶使用

日志库采用模块化方法，记录器、处理程序、格式化程序

def setLogger():
    if data.logger == '':
        # 第一步，创建一个logger
        data.logger = logging.getLogger()
        data.logger.setLevel(logging.INFO)  # Log等级总开关
 
        # 第二步，创建一个handler，用于写入日志文件
        logfile = data.result_log
        fh = logging.FileHandler(logfile, mode='a')
        fh.setLevel(logging.INFO)  # 输出到file的log等级的开关
 
        # 第三步，定义handler的输出格式
        formatter = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")
        fh.setFormatter(formatter)
 
        # 创建一个handler，用于输出到控制台
        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)  # 输出到console的log等级的开关
        ch.setFormatter(formatter)
        data.logger.addHandler(ch)
 
        # 第四步，将logger添加到handler里面
        data.logger.addHandler(fh)

15、adb命令控制安卓手机录屏方法

（1）screenrecord 命令可以进行录屏，最长3分钟

adb -s 9fa50772 shell screenrecord /sdcard/20240416130828.mp4

（2）停止录屏

在python代码中停止adb录屏的方法有两种：

方法一：

import subprocess
 
# 启动录屏
process = subprocess.Popen('adb shell screenrecord /sdcard/123.mp4')
time.sleep(5)
 
# 结束录屏
process.terminate()
# 将文件传输到电脑
res = subprocess.check_output('adb pull /sdcard/123.mp4  123.mp4').decode()
print(res)

方法二：

import subprocess
 
# 启动录屏
subprocess.Popen('adb shell screenrecord /sdcard/123.mp4')
time.sleep(5)
 
# 杀掉进程，结束录屏
subprocess.check_output('adb shell pkill -2 screenrecord')

Kill -2 ：功能类似于Ctrl + C 是程序在结束之前，能够保存相关数据，然后再退出。

Kill -9 ：直接强制结束程序。