3、Python基础回顾与强化：面向对象编程在硬件项目中的应用

说实话，很多做硬件的朋友一听到「面向对象」就头疼。我刚开始也是这样，觉得写个脚本控制传感器，搞什么类不类的？直到我在一个项目里同时管理8个温湿度传感器、3个压力变送器和2个流量计……那代码写得，我自己都不想看第二遍。

后来我学乖了。面向对象编程（OOP）在硬件项目里，说白了就是给每个设备办一张「身份证」。你想想看，每个传感器都有自己的型号、地址、数据格式，用类来封装，再合适不过了。

3.1 用类来封装硬件设备

我习惯把每个设备抽象成一个类。比如一个温湿度传感器，它的属性就是温度值、湿度值、I2C地址，它的行为就是读取数据、校准、复位。

class SHT30Sensor:
    def __init__(self, i2c_addr=0x44):
        self.i2c_addr = i2c_addr
        self.temperature = 0.0
        self.humidity = 0.0
        self._connected = False

    def connect(self):
        # 模拟I2C连接
        self._connected = True
        print(f"传感器已连接，地址: 0x{self.i2c_addr:02X}")

    def read_data(self):
        if not self._connected:
            raise ConnectionError("设备未连接，请先调用connect()")
        # 模拟读取数据
        self.temperature = 25.3
        self.humidity = 60.1
        return self.temperature, self.humidity

    def calibrate(self, offset=0.0):
        self.temperature += offset
        print(f"校准完成，偏移量: {offset}")

你看，这样写的好处是什么？每个传感器实例都是独立的。我在项目中遇到过同时接5个同型号传感器的情况，每个实例互不干扰，代码可读性也高。

3.2 继承与多态：让代码少写一半

做监控系统，设备种类多了去了。温湿度传感器、压力传感器、气体传感器……它们都有共同点：都需要连接、读取、断开。但具体实现又不一样。

这时候就该继承上场了。我定义一个基类 BaseSensor，把公共逻辑放进去，子类只管实现自己的特殊部分。

class BaseSensor:
    def __init__(self, name, address):
        self.name = name
        self.address = address
        self.data = None

    def connect(self):
        raise NotImplementedError("子类必须实现connect方法")

    def read(self):
        raise NotImplementedError("子类必须实现read方法")

    def disconnect(self):
        print(f"{self.name} 已断开连接")


class PressureSensor(BaseSensor):
    def connect(self):
        print(f"压力传感器 {self.name} 通过模拟通道连接")

    def read(self):
        # 模拟读取压力值
        self.data = 101.3  # kPa
        return self.data


class GasSensor(BaseSensor):
    def connect(self):
        print(f"气体传感器 {self.name} 通过串口连接")

    def read(self):
        # 模拟读取气体浓度
        self.data = 0.05  # ppm
        return self.data

嗯，这里要注意：基类里的方法故意抛出 NotImplementedError，就是逼着子类去实现。我曾经偷懒没写这个，结果运行时才发现某个传感器没实现读取方法，排查了半天……

3.3 多线程处理并发数据流

监控系统最头疼的是什么？多个设备同时上报数据。如果你一个一个轮询，后面的设备得等到猴年马月。我早期一个项目就是吃了这个亏，10个传感器轮询一遍要3秒，实时性根本没法看。

后来我改用 threading，每个传感器一个线程，各自跑各自的读取逻辑。

import threading
import time

class SensorReader(threading.Thread):
    def __init__(self, sensor, interval=1.0):
        super().__init__()
        self.sensor = sensor
        self.interval = interval
        self.running = True

    def run(self):
        self.sensor.connect()
        while self.running:
            data = self.sensor.read()
            print(f"[{self.sensor.name}] 数据: {data}")
            time.sleep(self.interval)

    def stop(self):
        self.running = False
        self.sensor.disconnect()


# 使用示例
sensor1 = PressureSensor("压力计A", "CH1")
sensor2 = GasSensor("气体检测仪B", "COM3")

reader1 = SensorReader(sensor1, interval=0.5)
reader2 = SensorReader(sensor2, interval=1.0)

reader1.start()
reader2.start()

# 运行10秒后停止
time.sleep(10)
reader1.stop()
reader2.stop()

3.4 asyncio：另一种并发思路

多线程虽然好用，但线程切换有开销。如果你的设备都是IO密集型（比如网络请求、串口读写），asyncio 可能更合适。它用单线程实现并发，说白了就是「等数据的时候去干点别的」。

import asyncio

class AsyncSensor:
    def __init__(self, name):
        self.name = name

    async def read_data(self):
        # 模拟异步IO操作
        await asyncio.sleep(1)
        return f"{self.name} 数据: {25.0}°C"

async def main():
    sensors = [AsyncSensor(f"传感器{i}") for i in range(3)]
    tasks = [sensor.read_data() for sensor in sensors]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

# 运行
asyncio.run(main())

我个人习惯：如果设备数量少（10个以内），用 threading 简单直接。如果数量多或者有网络请求，我会选 asyncio。没有绝对的好坏，看场景。

3.5 异常处理：设备通信的保命符

做硬件通信，不出异常是不可能的。线松了、设备掉电了、数据格式错了……什么情况都可能发生。我见过最惨的案例：一个监控系统因为传感器偶尔返回空数据，直接崩溃了，整个产线停了半小时。

所以，try-except-finally 不是可选项，是必选项。

def safe_read_sensor(sensor, retries=3):
    for attempt in range(retries):
        try:
            sensor.connect()
            data = sensor.read()
            return data
        except ConnectionError as e:
            print(f"连接失败 (第{attempt+1}次): {e}")
            time.sleep(1)
        except ValueError as e:
            print(f"数据格式错误: {e}")
            # 记录日志，但继续尝试
            continue
        except Exception as e:
            print(f"未知错误: {e}")
            raise  # 未知错误直接抛出，别藏着
        finally:
            # 不管成功还是失败，都要断开连接
            try:
                sensor.disconnect()
            except:
                pass  # 断开失败也不能影响主流程
    else:
        # 所有重试都失败了
        raise RuntimeError(f"传感器 {sensor.name} 读取失败，已重试{retries}次")

为什么会这样？因为设备通信不像本地函数调用，它受外部环境影响太大。我曾在项目里遇到一个奇葩问题：传感器每隔100次就会返回一个乱码。如果不加异常处理，整个系统每100次就崩一次。加了重试和异常捕获后，系统稳定运行了两年。

3.6 综合实战：一个简单的监控节点

把上面这些知识点串起来，就是一个完整的监控节点了。

class MonitoringNode:
    def __init__(self):
        self.sensors = []
        self.readers = []

    def add_sensor(self, sensor):
        self.sensors.append(sensor)

    def start_monitoring(self):
        for sensor in self.sensors:
            reader = SensorReader(sensor, interval=1.0)
            self.readers.append(reader)
            reader.start()

    def stop_monitoring(self):
        for reader in self.readers:
            reader.stop()
        for reader in self.readers:
            reader.join()  # 等待线程结束

    def get_all_data(self):
        data = {}
        for sensor in self.sensors:
            try:
                data[sensor.name] = safe_read_sensor(sensor)
            except Exception as e:
                data[sensor.name] = f"错误: {e}"
        return data


# 使用
node = MonitoringNode()
node.add_sensor(PressureSensor("压力计1", "CH1"))
node.add_sensor(GasSensor("气体检测仪1", "COM3"))

node.start_monitoring()
time.sleep(5)
print(node.get_all_data())
node.stop_monitoring()

你看，面向对象、多线程、异常处理，三个东西合在一起，就是一个健壮的监控节点。每个部分各司其职，出了问题也好排查。

好了，这一章的内容就这些。面向对象帮你组织代码，多线程和asyncio帮你处理并发，异常处理帮你兜底。这三板斧用好了，远程监控系统的基础就稳了。

---