全网最适合入门的面向对象编程教程：60 Python面向对象综合实例-传感器数据 ...

泉缘泉 · 2024-12-8 23:51:07

全网最适合入门的面向对象编程教程：60 Python 面向对象综合实例-传感器数据实时绘图器

本文档主要介绍如何使用 Python 举行面向对象编程，必要读者对 Python 语法和单片机开辟具有根本相识。相比其他讲解 Python 面向对象编程的博客或书籍而言，本文档更加详细、侧重于嵌入式上位机应用，以上位机和下位机的常见串口数据收发、数据处理、动态图绘制等为应用实例，同时使用 Sourcetrail 代码软件对代码举行可视化阅读便于读者理解。
相干示例代码获取链接如下：https://github.com/leezisheng/Python-OOP-Demo
正文

接下来，我们将运用之前学习过的内容实现如下任务：

模拟传感器线程天生数据，通过串口协议传输给模拟主机历程；
模拟主机历程通过串口发送给传感器线程指令以获取数据，举行数据滤波和处理，并将数据传输给绘图线程；
绘图线程完成数据曲线绘制和更新操纵。

可以注意到，我们之前为了论述面向对象编程的本领，所提及的某些类和类之间的关联设计，在现实应用中并非最佳实践。此外，原先的程序代码中，为了增强读者的直观感受，我们使用了大量的 print 语句举行输出。然而，这样的做法在程序举行并行运行时，不但会降低整体的执行效率，而且在观察终端输出时，对于多线程或多历程的运行序次也易造成混淆，使得理解变得困难。因此，我们有必要对这部分代码举行优化调解，以提高程序的性能和可读性。
在以下代码中，我们定义了一个 MasterProcess 主机多历程类，它继续了 Process 类，用于创建和管理主机多历程。可以看到 MasterProcess 主机多历程类直接在__ReadMasterSerial 和__WriteMasterSerial 方法中实现了串口的读写，而不像原先代码 MasterClass 类继续于 SerialClass 类。
示例代码如下：

class MasterProcess(Process):
'''
主机多进程类
'''
_# 类变量：_
_# START_CMD - 开启命令 -0_
_# STOP_CMD - 关闭命令 -1_
_# SENDID_CMD - 发送ID命令 -2_
_# SENDVALUE_CMD - 发送数据命令 -3_
START_CMD, STOP_CMD, SENDID_CMD, SENDVALUE_CMD = (0, 1, 2, 3)
def __init__(self,
lock,
Queue,
simplequeue,
port:str = "COM17",
baudrate:int = 115200,
bytesize:int = serial.EIGHTBITS,
parity :str = serial.PARITY_NONE,
stopbits:int = serial.STOPBITS_ONE):
'''
MasterProcess初始化函数
:param lock: 互斥锁
:param Queue: 队列
:param port: 端口号
:param baudrate: 波特率
:param bytesize: 数据位
:param parity: 校验位
:param stopbits: 停止位
'''
self.lock = lock
self.Queue = Queue
self.simplequeue = simplequeue
self.dev = serial.Serial()
self.dev.port = port
self.dev.baudrate = baudrate
self.dev.bytesize = bytesize
self.dev.parity = parity
self.dev.stopbits = stopbits
_# 设置读取timeout超时时间_
self.dev.timeout = 0.3
_# 设置写入timeout超时时间_
self.dev.write_timeout = 0.3
_# 数据缓存_
self.datalist = []
_# 滤波器长度_
self.filterlength = 3
_# 数据处理类实例_
self.dataprocessobj = DateProcessClass(self.datalist,self.filterlength)
_# Process初始化方法_
Process.__init__(self)
def StartMasterSerial(self):
'''
打开主机串口
:return: None
'''
self.dev.open()
def StopMasterSerial(self):
'''
停止主机串口
:return: None
'''
self.dev.close()
def __ReadMasterSerial(self):
'''
读取主机串口，私有方法
:return data[int] : 读取的数据
'''
_# 按行读取_
data = self.dev.readline()
_# 如果接收到字节的情况下，进行处理_
if data != b'':
_# 收到为二进制数据_
_# 用utf-8编码将二进制数据解码为unicode字符串_
_# 字符串转为int类型_
data = int(data.decode('utf-8', 'replace'))
_# 否则，设置data为-1_
else:
data = -1
return data
def __WriteMasterSerial(self,write_data):
'''
写入主机串口，私有方法
:param write_data: 写入的数据
:return:
'''
_# 非阻塞方式写入_
self.dev.write(write_data.encode())
_# 输出换行符_
_# write的输入参数必须是bytes格式_
_# 字符串数据需要encode()函数将其编码为二进制数据，然后才可以顺利发送_
_# \r\n表示换行回车_
self.dev.write('\r\n'.encode())
def RecvSensorID(self):
'''
读取传感器ID
:return sensorid[int] : 读取的传感器id号
'''
sensorid = self.__ReadMasterSerial()
return sensorid
def RecvSensorValue(self):
'''
读取传感器数据值
:return data[int] : 读取的传感器数据
'''
data = self.__ReadMasterSerial()
return data
def SendSensorCMD(self,cmd):
'''
主机发送命令
:param cmd : MasterProcess中的类变量
:return: None
'''
self.__WriteMasterSerial(str(cmd))

复制代码

同时，我们在其中定义了一个 run 方法，这个方法在多历程启动后被调用。在这个方法中，起首打开串口，然后发送获取 ID 的指令，接收传感器的 ID 号。然后进入一个无限循环，每 9 次循环，盘算一次最大值和最小值，并打印出来。然后发送获取数据的指令，接收传感器的数据，将数据放入队列和列表中，然后对数据举行滤波处理，将滤波后的数据放入另一个队列。末了，打印出接收到的传感器数据，并让当前历程休眠 0.5 秒。
示例代码如下：

def run(self):
'''
多进程start后运行的方法
:return: None
'''
_# 运行计数变量_
count = 0
_# 文件保存索引计数变量_
index = 0
_# 打开串口_
self.StartMasterSerial()
self.lock.acquire()
print(" Master Process Started ")
self.lock.release()
_# 发送获取ID指令_
self.SendSensorCMD(self.SENDID_CMD)
_# 获取传感器ID号_
id = self.RecvSensorID()
self.lock.acquire()
print(" Recv Sensor ID : ", id)
self.lock.release()
while True:
if count == 9:
maxvalue = self.dataprocessobj.DateCalMax()
minvalue = self.dataprocessobj.DateCalMin()
self.lock.acquire()
print("----------------------------------")
print("Max Value: ", maxvalue)
print("Min Value: ", minvalue)
print("----------------------------------")
self.lock.release()
count = 0
else:
count = count + 1
_# 发送获取数据指令_
self.SendSensorCMD(self.SENDVALUE_CMD)
self.lock.acquire()
print("Master Send SENDVALUE_CMD")
self.lock.release()
_# 接收传感器数据值_
data = self.RecvSensorValue()
self.Queue.put(data)
self.datalist.append(data)
filterdata,filterdatalist = self.dataprocessobj.DateFilter()
self.simplequeue.put(filterdata)
self.lock.acquire()
print(" Recv Sensor Data : ",data)
self.lock.release()
time.sleep(0.5)

复制代码

同时，我们必要注意 MasterProcess 主机历程类和 DateProcessClass 数据处理类为组合关系，使用如下语句实现：

_# 数据缓存_
self.datalist = []
_# 滤波器长度_
self.filterlength = 3
_# 数据处理类实例_
self.dataprocessobj = DateProcessClass(self.datalist,self.filterlength)

复制代码

同时我们重写了 DateProcessClass 数据处理类，为方便理解程序运行情况，我们去掉了非常类、日志和 print 输出语句，同时在 DateFilter 方法中将盘算出的平均值也举行返回。

class DateProcessClass():
def __init__(self, DateList: List[int], FilterLength: int):
self.DateList = DateList
self.FilterLength = FilterLength
self.TempList = [0] * (self.FilterLength)
def DateFilter(self) -> List:
_# 遍历DateList_
for index, value in enumerate(self.DateList):
_# 把每个值都当成传入的新的传感器的值_
NowValue = value
_# 表示列表求和的变量_
sum = 0
for i in range(self.FilterLength - 1):
_# 实现列表的移位操作_
self.TempList[i] = self.TempList[i + 1]
_# 实现列表求和_
sum += self.TempList[i]
self.TempList[self.FilterLength - 1] = NowValue
sum += self.TempList[self.FilterLength - 1]
_# 求平均值_
average = sum / self.FilterLength
_# 将计算得到的平均值替换原始值_
self.DateList[index] = average
_# 计算完成后将TempList中元素清零_
self.TempList = [0] * (self.FilterLength)
return average,self.DateList
def DateCalMax(self) -> int:
max_value = max(self.DateList)
return int(max_value)
def DateCalMin(self) -> int:
min_value = min(self.DateList)
return int(min_value)

复制代码

同时，我们重写了绘图类，该类包含了初始化、数据更新和定时更新等方法。在初始化方法中，程序创建了一个 Qt 应用实例对象、一个多面板图形窗口对象以及两个绘图曲线对象。数据更新方法用于接收传感器数据并将其添加到缓存列表中，然后将数据转化为图形。定时更新方法则用于定时举行曲线更新。
这里主要对 GetValue 方法和 DataUpdate 方法举行改写，同时我们可以看到在初始化方法和绘图曲线中增加了滤波后数据的相干属性，用于查看滤波结果。
示例代码如下：

class PlotThread:
def __init__(self,lock,queue,simplequeue,wintitle:str="Basic plotting examples",plottitle:str="Updating plot",width:int=1000,height:int=600):
'''
用于初始化PlotThread类
:param wintitle: 窗口标题
:param plottitle: 图层标题
:param width: 窗口宽度
:param height: 窗口高度
'''
self.lock = lock
self.queue = queue
self.simplequeue = simplequeue
_# Qt应用实例对象_
self.app = None
_# 窗口对象_
self.win = None
_# 设置窗口标题_
self.title = wintitle
_# 设置窗口尺寸_
self.width = width
self.height = height
_# 传感器数据_
self.value = 0
_# 存放滤波后数据_
self.filtervalue = 0
_# 计数变量_
self.__count = 0
_# 传感器数据缓存列表_
self.valuelist = []
_# 传感器滤波数据缓存列表_
self.filtervaluelist = []
_# 绘图曲线_
self.curve = None
_# 滤波后绘图曲线_
self.filtercurve = None
_# 图层对象_
self.plotob = None
_# 图层标题_
self.plottitle = plottitle
_# 定时器对象_
self.timer = QtCore.QTimer()
_# 定时时间_
self.time = 0
_# Qt应用和窗口初始化_
self.appinit()
self.lock.acquire()
print(" PlotClass Object Init Complete ")
self.lock.release()
def appinit(self):
'''
用于qt应用程序初始化，添加窗口、曲线和图层
:return: None
'''
_# 创建一个Qt应用，并返回该应用的实例对象_
self.app = pg.mkQApp("Plotting Example")
_# 生成多面板图形_
_# show：(bool) 如果为 True，则在创建小部件后立即显示小部件。_
_# title：(str 或 None)如果指定，则为此小部件设置窗口标题。_
self.win = pg.GraphicsLayoutWidget(show=True, title=self.title)
_# 设置窗口尺寸_
self.win.resize(self.width, self.height)
_# 进行窗口全局设置，setConfigOptions一次性配置多项参数_
_# antialias启用抗锯齿，useNumba对图像进行加速_
pg.setConfigOptions(antialias=True, useNumba=True)
_# 添加图层_
self.plotob = self.win.addPlot(title=self.plottitle)
_# 添加曲线_
_# 原始数据-黄色曲线_
self.curve = self.plotob.plot(pen='y')
_# 滤波后数据-红色曲线_
self.filtercurve = self.plotob.plot(pen='r')
def GetValue(self,value,filtervalue):
'''
用于接收传感器数据，加入缓存列表
:param value: 传感器数据
:param filtervalue: 传感器滤波后数据
:return: None
'''
self.value = value
self.valuelist.append(self.value)
self.filtervalue = filtervalue
self.filtervaluelist.append(self.filtervalue)
def DataUpdate(self):
'''
用于定时进行曲线更新,这里模拟绘制正弦曲线
:return: None
'''
self.value = self.queue.get()
self.filtervalue = self.simplequeue.get()
self.GetValue(self.value,self.filtervalue)
_# 将数据转化为图形_
self.curve.setData(self.valuelist)
self.filtercurve.setData(self.filtervaluelist)
def SetUpdate(self,time:int = 100):
'''
设置定时更新任务
:param time: 定时的时间
:return: None
'''
_# 定时器结束，触发DataUpdate方法_
self.timer.timeout.connect(self.DataUpdate)
_# 启动定时器_
self.timer.start(time)
_# 定时时间_
self.time = time
_# 进入主事件循环并等待_
pg.exec()

复制代码

GetValue 方法用于接收传感器数据并将其加入缓存列表，DataUpdate 方法定时运行，它使用队列接收 MasterProcess 历程中产生和传输的数据，并将更新后的数据通过 setData 方法实现绘图曲线的更新。可以看到，数据的绘图和更新都是在主线程中举行的，这是由于在使用 PyQtGraph 绘图时，如果在主线程之外举行绘图操纵，大概会出现绘图不生效或程序崩溃的题目。

在主线程中完成创建互斥锁、消息队列和历程实例，然后创建传感器线程和绘图线程实例，末了启动历程和线程。示例代码如下：

if __name__ == "__main__":
_# 创建互斥锁_
lock = Lock()
_# 创建消息队列_
queue = Queue(5)
_# 创建消息队列_
simplequeue = SimpleQueue()
_# 创建进程实例_
m_process = MasterProcess(lock,queue,simplequeue,port = "COM17")
_# 创建线程实例_
s_thread = SensorThread(lock,port="COM11", id=0, state=SensorThread.WORK_MODE["RESPOND_MODE"])
_# 创建绘图类实例_
p_thread = PlotThread(lock,queue,simplequeue)
_# 启动进程_
m_process.start()
_# 开启线程，start方法以并发方式执行_
s_thread.start()
_# 启动p_thread的定时任务_
p_thread.SetUpdate(600)

复制代码

接下来，我们运行程序：

可以看到整个程序可以并行运行，对于并行运行的 Python 程序来说，我们可以使用 Profile/CProfile 等工具举行分析调试，或者使用 viztracer/SnakeViz 等可视化工具天生线程/历程运行的火焰图，这些工具可以完成记录函数的入口/出口，函数参数/返回值、恣意变量的值以及线程/历程的运行序次等操纵。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

		自动登录	找回密码
密码			立即注册

全网最适合入门的面向对象编程教程：60 Python面向对象综合实例-传感器数据 ...

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云