全网最适合入门的面向对象编程教程：20 类和对象的 Python 实现-组合关系的实现与 CSV 文件保存
全网最适合入门的面向对象编程教程：20 类和对象的 Python 实现-组合关系的实现与 CSV 文件保存
摘要：
本文主要介绍了在使用 Python 面向对象编程时，如何实现组合关系，同时对比了组合关系和继承关系的优缺点，并讲解了如何通过 csv 模块来保存 Python 接收/生成的数据。
原文链接：
往期推荐：
全网最适合入门的面向对象编程教程：00 面向对象设计方法导论
全网最适合入门的面向对象编程教程：01 面向对象编程的基本概念
全网最适合入门的面向对象编程教程：02 类和对象的 Python 实现-使用 Python 创建类
全网最适合入门的面向对象编程教程：03 类和对象的 Python 实现-为自定义类添加属性
全网最适合入门的面向对象编程教程：04 类和对象的Python实现-为自定义类添加方法
全网最适合入门的面向对象编程教程：05 类和对象的Python实现-PyCharm代码标签
全网最适合入门的面向对象编程教程：06 类和对象的Python实现-自定义类的数据封装
全网最适合入门的面向对象编程教程：07 类和对象的Python实现-类型注解
全网最适合入门的面向对象编程教程：08 类和对象的Python实现-@property装饰器
全网最适合入门的面向对象编程教程：09 类和对象的Python实现-类之间的关系
全网最适合入门的面向对象编程教程：10 类和对象的Python实现-类的继承和里氏替换原则
全网最适合入门的面向对象编程教程：11 类和对象的Python实现-子类调用父类方法
全网最适合入门的面向对象编程教程：12 类和对象的Python实现-Python使用logging模块输出程序运行日志
全网最适合入门的面向对象编程教程：13 类和对象的Python实现-可视化阅读代码神器Sourcetrail的安装使用
全网最适合入门的面向对象编程教程：全网最适合入门的面向对象编程教程：14 类和对象的Python实现-类的静态方法和类方法
全网最适合入门的面向对象编程教程：15 类和对象的 Python 实现-__slots__魔法方法
全网最适合入门的面向对象编程教程：16 类和对象的Python实现-多态、方法重写与开闭原则
全网最适合入门的面向对象编程教程：17 类和对象的Python实现-鸭子类型与“file-like object“
全网最适合入门的面向对象编程教程：18 类和对象的Python实现-多重继承与PyQtGraph串口数据绘制曲线图
全网最适合入门的面向对象编程教程：19 类和对象的 Python 实现-使用 PyCharm 自动生成文件注释和函数注释
更多精彩内容可看：
给你的 Python 加加速：一文速通 Python 并行计算
一个MicroPython的开源项目集锦：awesome-micropython，包含各个方面的Micropython工具库
文档和代码获取：
可访问如下链接进行对文档下载：
https://github.com/leezisheng/Doc
本文档主要介绍如何使用 Python 进行面向对象编程，需要读者对 Python 语法和单片机开发具有基本了解。相比其他讲解 Python 面向对象编程的博客或书籍而言，本文档更加详细、侧重于嵌入式上位机应用，以上位机和下位机的常见串口数据收发、数据处理、动态图绘制等为应用实例，同时使用 Sourcetrail 代码软件对代码进行可视化阅读便于读者理解。
相关示例代码获取链接如下：
https://github.com/leezisheng/Python-OOP-Demo
正文
前面讲了面向类与对象的继承，知道了继承是一种什么“是”什么的关系。然而类与类之间还有另一种关系，这就是组合。
组合是将几个对象收集在一起生成一个新对象的行为。当一个对象是另外一个对象的一部分时，组合通常是不错的选择。
例如，汽车是由发动机、传动装置、启动装置、车前灯、挡风玻璃以及其他部件组成的，发动机又是由活塞、曲柄轴和阀门等组合而成的。
汽车是发动机等多个元器件的抽象，而发动机是活塞等元器件的抽象，二者处于不同的层次而又有彼此交互的接口，组合是提供不同抽象层的好办法。
汽车对象可以提供司机所需要的接口，同时也能够获取内在组成部分，从而为机械师提供适合操作的深层抽象。当然，如果机械师需要更多信息来诊断问题或调整发动机，这些组成部分也可以进一步被细分。
总的来说，组合就是
让不同的类混合并且加入其他类中来增加功能和代码重用性
，这种适用于
由多个小类组成一个大类的情况
，并且不需要对小类进行太多修改。在前面示例中，我们实现了主机的串口收发和绘图功能，在实际应用中，我们往往需要将传感器数据存储到文件中，以便后续的查看和处理，很明显前面的传感器数据为一维的时间序列数据，
适合存储为表格类型（即列标题为索引和值），我们通常将该类数据保存为 csv 格式文件，csv 是一种字符串文件的格式，它组织数据的语法就是在字符串之间加分隔符（行与行之间是加换行符，同行字符之间是加逗号分隔）
，可以用任意的文本编辑器打开（如记事本），也可以用 Excel 打开，还可以通过 Excel 把文件另存为 csv 格式。
用 csv 格式存储数据，读写比较方便，易于实现，文件也会比 Excel 文件小。
但 csv 文件缺少 Excel 文件本身的很多功能，如不能嵌入图像和图表，不能生成公式等等。
操作 csv 文件我们需要借助 csv 模块，python 自带 csv 模块，不需要我们使用 pip 安装，我们可以点击如下链接查看 csv 模块使用方法：
https://docs.python.org/zh-cn/3.13/library/csv.html#csv.writer
这里，我们首先定义一个 FileIOClass 类，其中具有初始化方法、写入传感器数据到文件方法和关闭文件方法，示例代码如下：
import csv
_# 使用typing模块提供的复合注解功能_
from typing import List
class FileIOClass:
def __init__(self,path:str="G:\\Python面向对象编程\\Demo\\file.csv"):
'''
初始化csv文件和列标题
:param path: 文件路径和文件名
'''
self.path = path
_# path为输出路径和文件名，newline=''是为了不出现空行_
self.csvFile = open(path, "w+", newline='')
_# rowname为列名，index-索引，data-数据_
self.rowname = ['index', 'data']
_# 返回一个writer对象，将用户的数据在给定的文件型对象上转换为带分隔符的字符串_
self.writer = csv.writer(self.csvFile)
_# 写入csv文件的列标题_
self.writer.writerow( self.rowname)
def WriteFile(self,index:List[int],data:List[int])->None:
'''
:param index: 传感器索引列表
:param data: 传感器数据列表
:return:
'''
writedatalist = []
for i in range(len(data)):
writedatalist.append([index[i],data[i]])
_# 将列表中的每个元素将被写入CSV文件的一列中_
self.writer.writerow(writedatalist[i])
def CloseFile(self)->None:
'''
关闭文件
:return: None
'''
self.csvFile.close()
这里，在初始化方法中，我们需要传入文件保存路径。之后创建一个 writer 对象，将用户的数据在给定的文件型对象上转换为带分隔符的字符串，同时写入 csv 文件的列标题。在 WriteFile 方法中传入数据的索引列表用于表示数据的先后顺序，之后是数据列表（这里的类型注解需要使用 typing 模块提供的复合注解功能），并循环将每个元素将被写入 CSV 文件的一列中，最后定义了文件的关闭方法。
在主函数中，我们创建 FileIOClass 对象，写入模拟传感器数据后关闭文件，以下为示例代码和运行效果：
if __name__ == '__main__':
path = "G:\\Python面向对象编程\\Demo\\file.csv"
data = [11,42,307,46,55,61,78,80,19,11]
index = [count for count in range(len(data))]
file = FileIOClass(path)
file.WriteFile(index,data)
file.CloseFile()
这里，我们可以直接在 MasterClass 类的初始化中创建 FileIOClass 类的实例化对象来实现组合。代码如下：
_# 文件保存路径_
self.savepath = "G:\\Python面向对象编程\\Demo\\file.csv"
_# 创建FileIOClass类的实例化对象_
self.fileio = FileIOClass(self.savepath)
通过 sourcetrail，我们可以清晰看到类之间的组合与继承关系：
在主程序中，我们在主机接收 10 次数据后，将数据保存到 file.csv 中：
if __name__ == "__main__":
_ # 创建数据列表_
datalist = []
m = MasterClass(state = MasterClass.IDLE_STATE,
port = "COM17",
wintitle = "Basic plotting examples",
plottitle = "Updating plot",
width = 1000,
height = 600)
m.StartMaster()
m.SendSensorCMD(MasterClass.SENDID_CMD)
m.RecvSensorID()
# 循环10次接收数据
for i in range(10):
m.SendSensorCMD(MasterClass.SENDVALUE_CMD)
value = m.RecvSensorValue()
datalist.append(value)
indexlist = [count for count in range(len(datalist))]
# 写入数据
m.fileio.WriteFile(indexlist,datalist)
m.fileio.CloseFile()
如下为运行效果：
目前，整个文件的完整代码如下，可以看到单单是这么一个简单程序就有了三百多行，对于代码查找修改来讲，非常不便。同时我们注意到，几个不同类之间似乎功能并不相同，不应该放到一个文件中。下一节我们将会说如何利用 Python 中的模块和包来组织我们的代码。
完整代码如下：
_# 串口相关库_
import serial
import serial.tools.list_ports
_# 队列相关_
import queue
import random
_# 日志输出相关库_
import logging
_# 曲线作图相关库_
import pyqtgraph as pg
import numpy as np
from pyqtgraph.Qt import QtCore
_# 文件读写相关库_
import csv
_# 使用typing模块提供的复合注解功能_
from typing import List
import time
_# # 设置日志输出级别_
_# logging.basicConfig(level=logging.DEBUG)_
_# 在配置下日志输出目标文件和日志格式_
LOG_FORMAT="%(asctime)s-%(levelname)s-%(message)s"
logging.basicConfig(filename='my.log',level=logging.DEBUG,format=LOG_FORMAT)
class SerialClass:
_# 限定SerialClass对象只能绑定以下属性_
__slots__ = ('dev','_SerialClass__devstate')
_# 初始化_
_# 使用默认参数_
def __init__(self,
devport:str = "COM17",
devbaudrate:int = 115200,
devbytesize:int = serial.EIGHTBITS,
devparity :str = serial.PARITY_NONE,
devstopbits:int = serial.STOPBITS_ONE):
_# 直接传入serial.Serial()类_
self.dev = serial.Serial()
self.dev.port = devport
self.dev.baudrate = devbaudrate
self.dev.bytesize = devbytesize
self.dev.parity = devparity
self.dev.stopbits = devstopbits
_# 表示串口设备的状态-打开或者关闭_
_# 初始化时为关闭_
self.__devstate = False
print("SerialClass init")
logging.info("SerialClass init")
_# 取值方法_
@property
def devstate(self):
return self.__devstate
_# 打开串口_
def OpenSerial(self):
print("SerialClass-OpenSerial")
logging.info("SerialClass-OpenSerial")
self.dev.open()
self.__devstate = True
_# 关闭串口_
def CloseSerial(self):
print("SerialClass-CloseSerial")
logging.info("SerialClass-CloseSerial")
self.dev.close()
self.__devstate = False
_# 串口读取_
def ReadSerial(self):
print("SerialClass-ReadSerial")
logging.info("SerialClass-ReadSerial")
if self.__devstate:
_# 阻塞方式读取_
_# 按行读取_
data = self.dev.readline()
_# 收到为二进制数据_
_# 用utf-8编码将二进制数据解码为unicode字符串_
_# 字符串转为int类型_
data = int(data.decode('utf-8', 'replace'))
return data
_# 串口写入_
def WriteSerial(self,write_data):
print("SerialClass-WriteSerial")
logging.info("SerialClass-WriteSerial")
if self.__devstate:
_# 非阻塞方式写入_
self.dev.write(write_data.encode())
_# 输出换行符_
_# write的输入参数必须是bytes 格式_
_# 字符串数据需要encode()函数将其编码为二进制数据，然后才可以顺利发送_
_# \r\n表示换行回车_
self.dev.write('\r\n'.encode())
def RetSerialState(self):
if self.dev.isOpen():
self.__devstate = True
return True
else:
self.__devstate = False
return False
class PlotClass:
_# 绘图类初始化_
def __init__(self,wintitle:str="Basic plotting examples",plottitle:str="Updating plot",width:int=1000,height:int=600):
'''
用于初始化Plot类
:param wintitle: 窗口标题
:param plottitle: 图层标题
:param width: 窗口宽度
:param height: 窗口高度
'''
_# Qt应用实例对象_
self.app = None
_# 窗口对象_
self.win = None
_# 设置窗口标题_
self.title = wintitle
_# 设置窗口尺寸_
self.width = width
self.height = height
_# 传感器数据_
self.value = 0
_# 计数变量_
self.__count = 0
_# 传感器数据缓存列表_
self.valuelist = []
_# 绘图曲线_
self.curve = None
_# 图层对象_
self.plotob = None
_# 图层标题_
self.plottitle = plottitle
_# 定时器对象_
self.timer = QtCore.QTimer()
_# 定时时间_
self.time = 0
_# Qt应用和窗口初始化_
self.appinit()
print("PLOT INIT SUCCESS")
logging.info("PLOT INIT SUCCESS")
_# 应用程序初始化_
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')
_# 接收数据_
def GetValue(self,value):
'''
用于接收传感器数据，加入缓存列表
:param value: 传感器数据
:return: None
'''
self.value = value
_# 加入数据缓存列表_
self.valuelist.append(value)
print("PLOT RECV DATA : "+str(self.value))
logging.info("PLOT RECV DATA : "+str(self.value))
_# 更新曲线数据_
def DataUpdate(self):
'''
用于定时进行曲线更新,这里模拟绘制正弦曲线
:return: None
'''
_# 模拟绘制正弦曲线_
_# 计数变量更新_
self.__count = self.__count + 0.1
self.value = np.sin(self.__count)
self.GetValue(self.value)
_# 将数据转化为图形_
self.curve.setData(self.valuelist)
_# 设置定时更新_
def SetUpdate(self,time:int = 100):
'''
设置定时更新任务
:param time: 定时的时间
:return: None
'''
_# 定时器结束，触发DataUpdate方法_
self.timer.timeout.connect(self.DataUpdate)
_# 启动定时器_
self.timer.start(time)
_# 定时时间_
self.time = time
print("PLOT SET UPDATA")
logging.info("PLOT SET UPDATA")
_# 进入主事件循环并等待_
pg.exec()
class FileIOClass:
def __init__(self,path:str="G:\\Python面向对象编程\\Demo\\file.csv"):
'''
初始化csv文件和列标题
:param path: 文件路径和文件名
'''
self.path = path
_# path为输出路径和文件名，newline=''是为了不出现空行_
self.csvFile = open(path, "w+", newline='')
_# rowname为列名，index-索引，data-数据_
self.rowname = ['index', 'data']
_# 返回一个writer对象，将用户的数据在给定的文件型对象上转换为带分隔符的字符串_
self.writer = csv.writer(self.csvFile)
_# 写入csv文件的列标题_
self.writer.writerow(self.rowname)
def WriteFile(self,index:List[int],data:List[int])->None:
'''
:param index: 传感器索引列表
:param data: 传感器数据列表
:return:
'''
writedatalist = []
for i in range(len(data)):
writedatalist.append([index[i],data[i]])
_# 将列表中的每个元素将被写入CSV文件的一列中_
self.writer.writerow(writedatalist[i])
def CloseFile(self)->None:
'''
关闭文件
:return: None
'''
self.csvFile.close()
class SensorClass(SerialClass):
_# 类变量：_
_# RESPOND_MODE -响应模式-0_
_# LOOP_MODE -循环模式-1_
RESPOND_MODE,LOOP_MODE = (0,1)
_# 类变量：_
_# 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,port:str = "COM11",id:int = 0,state:int = RESPOND_MODE):
_# 调用父类的初始化方法，super() 函数将父类和子类连接_
super().__init__(port)
self.sensorvalue = 0
self.sensorid = id
self.sensorstate = state
print("Sensor Init")
logging.info("Sensor Init")
@staticmethod
_# 判断传感器ID号是否正确：这里判断ID号是否在0到99之间_
def IsTrueID(id:int = 0):
if id >= 0 and id <= 99:
print("Sensor ID True")
return True
else:
print("Sensor ID False")
return False
_# 传感器上电初始化_
def InitSensor(self):
_# 传感器上电初始化工作_
_# 同时输出ID号以及状态_
print("Sensor %d Init complete : %d"%(self.sensorid,self.sensorstate))
logging.info("Sensor %d Init complete : %d"%(self.sensorid,self.sensorstate))
_# 开启传感器_
def StartSensor(self):
super().OpenSerial()
print("Sensor %d start serial %s "%(self.sensorid,self.dev.port))
logging.info("Sensor %d start serial %s "%(self.sensorid,self.dev.port))
_# 停止传感器_
def StopSensor(self):
super().CloseSerial()
print("Sensor %d close serial %s " % (self.sensorid, self.dev.port))
logging.info("Sensor %d close serial %s " % (self.sensorid, self.dev.port))
_# 发送传感器ID号_
def SendSensorID(self):
super().WriteSerial(str(self.sensorid))
print("Sensor %d send id "%self.sensorid)
logging.info("Sensor %d send id "%self.sensorid)
_# 发送传感器数据_
def SendSensorValue(self):
_# 生成[1, 10]内的随机整数_
data = random.randint(1, 10)
super().WriteSerial(str(data))
print("Sensor %d send data %d" % (self.sensorid,data))
logging.info("Sensor %d send data %d" % (self.sensorid,data))
_# 接收主机指令_
def RecvMasterCMD(self):
cmd = super().ReadSerial()
print("Sensor %d recv cmd %d " % (self.sensorid,cmd))
logging.info("Sensor %d recv cmd %d " % (self.sensorid,cmd))
return cmd
class MasterClass(SerialClass,PlotClass):
_# 类变量：_
_# BUSY_STATE -忙碌状态-0_
_# IDLE_STATE -空闲状态-1_
BUSY_STATE, IDLE_STATE = (0, 1)
_# 类变量：_
_# 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,state:int = IDLE_STATE,port:str = "COM17",wintitle:str="Basic plotting examples",plottitle:str="Updating plot",width:int=1000,height:int=600):
_# 分别调用不同父类的__init__方法_
SerialClass.__init__(self,port)
PlotClass.__init__(self,wintitle,plottitle,width,height)
self.valuequeue = queue.Queue(10)
self.__masterstatue = state
_# 初始化完成的标志量_
self.INIT_FLAG = False
_# 文件保存路径_
self.savepath = "G:\\Python面向对象编程\\Demo\\file.csv"
_# 创建FileIOClass类的实例化对象_
self.fileio = FileIOClass(self.savepath)
print("MASTER INIT SUCCESSS")
logging.info("MASTER INIT SUCCESSS")
@classmethod
def MasterInfo(cls):
print("Info : "+str(cls))
_# 开启主机_
def StartMaster(self):
super().OpenSerial()
print("START MASTER :"+self.dev.port)
logging.info("START MASTER :"+self.dev.port)
_# 停止主机_
def StopMaster(self):
super().CloseSerial()
print("CLOSE MASTER :" + self.dev.port)
logging.info("CLOSE MASTER :" + self.dev.port)
_# 接收传感器ID号_
def RecvSensorID(self):
sensorid = super().ReadSerial()
print("MASTER RECIEVE ID : " + str(sensorid))
logging.info("MASTER RECIEVE ID : " + str(sensorid))
return sensorid
_# 接收传感器数据_
def RecvSensorValue(self):
data = super().ReadSerial()
print("MASTER RECIEVE DATA : " + str(data))
logging.info("MASTER RECIEVE DATA : " + str(data))
self.valuequeue.put(data)
return data
_# 主机发送命令_
def SendSensorCMD(self,cmd):
super().WriteSerial(str(cmd))
print("MASTER SEND CMD : " + str(cmd))
logging.info("MASTER SEND CMD : " + str(cmd))
_# 主机返回工作状态-_
def RetMasterStatue(self):
return self.__masterstatue
_# 重写父类的DataUpdate方法_
def DataUpdate(self):
self.SendSensorCMD(self.SENDVALUE_CMD)
self.value = self.RecvSensorValue()
self.WriteSerial("Recv:"+str(self.value))
self.GetValue(self.value)
self.curve.setData(self.valuelist)
print("PLOT UPDATA : " + str(self.value))
logging.info("PLOT UPDATA : " + str(self.value))
class DevClass(SerialClass):
def __init__(self,port:str = "COM1"):
super().__init__(port)
_# 开启设备_
def StartDev(self):
super().OpenSerial()
print("START Dev :" + self.dev.port)
def ReadSerial(self,byte_size):
if super().RetSerialState():
data = self.dev.read(byte_size)
data = int(data.decode('utf-8', 'replace'))
return data
_# 判断串口类对象的串口是否开启_
def IsSerialConnected(serialclass):
return serialclass.RetSerialState()
if __name__ == "__main__":
_# 创建数据列表_
datalist = []
m = MasterClass(state = MasterClass.IDLE_STATE,
port = "COM17",
wintitle = "Basic plotting examples",
plottitle = "Updating plot",
width = 1000,
height = 600)
m.StartMaster()
m.SendSensorCMD(MasterClass.SENDID_CMD)
m.RecvSensorID()
_# 循环10次接收数据_
for i in range(10):
m.SendSensorCMD(MasterClass.SENDVALUE_CMD)
value = m.RecvSensorValue()
datalist.append(value)
indexlist = [count for count in range(len(datalist))]
_# 写入数据_
m.fileio.WriteFile(indexlist,datalist)
m.fileio.CloseFile()
