多Python脚本协同整合咨询：树莓派项目串口数据传输方案

核心思路 #

避免单线程阻塞导致的数据丢失，采用多进程架构拆分任务：

• 进程1：独立运行BLE设备扫描与数据更新
• 进程2：独立运行HTML页面数据读取与更新
• 进程3：读取两个进程的最新数据，按HMI要求格式拼接后发送串口

这种方式既能保证两个采集任务互不干扰，又能实现数据实时共享，避免合并脚本带来的阻塞风险。

实现方案：使用multiprocessing共享内存 #

Python的multiprocessing模块提供了线程/进程安全的共享内存容器，可让不同进程直接读写共享变量，无需额外文件或网络开销。

步骤1：重构采集脚本为进程函数 #

将原脚本的循环逻辑封装成函数，接收共享内存变量作为参数，实时更新数据。

重构后的BLE采集脚本（ble_scanner.py）

#! /usr/bin/python3 -u
import struct
from time import sleep
from bluepy.btle import Scanner
from multiprocessing import Array, Value

SENSORS = {
    "80:ea:ca:12:23:0b": "Front Left Tyre:",
    "81:ea:ca:22:20:f7": "Front Right Tyre:",
    "82:ea:ca:32:24:87": "Rear Left Tyre:",
    "83:ea:ca:42:23:07": "Rear Right Tyre:"
}

# 初始化BLE数据共享内存
def init_ble_shared_data():
    # 胎压、温度、电池、状态：每个字段预留20字节空间
    pres = Array('c', b' ' * (4*20))
    temp = Array('c', b' ' * (4*20))
    batt = Array('c', b' ' * (4*20))
    status = Array('c', b' ' * (4*20))
    # 胎压异常标记：每个标记2字节
    flat = Array('c', b'00' * 4)
    return pres, temp, batt, status, flat

def ble_scanner_process(pres, temp, batt, status, flat):
    scanner = Scanner()
    front_min_pres = 30
    rear_min_pres = 30
    max_temp = 50
    min_batt = 30
    max_loss = 5

    fl_count = fr_count = rl_count = rr_count = 0

    def sort_data(chunks):
        id_byte = chunks[2:8]
        pres_byte = chunks[8:12]
        temp_byte = chunks[12:16]
        batt_byte = chunks[16]
        flat_byte = chunks[17]

        id_str = ''.join(id_byte)
        pres_str = ''.join(pres_byte)
        temp_str = ''.join(temp_byte)
        
        pres_hex = bytes.fromhex(pres_str)
        temp_hex = bytes.fromhex(temp_str)

        pres_int = int.from_bytes(pres_hex, 'little')
        temp_int = int.from_bytes(temp_hex, 'little', signed=True)
        batt_int = int(batt_byte, 16)

        pres = str(round(((pres_int/100000)*14.5), 1))
        temp = str(round((temp_int/100), 1))
        batt = str(batt_int)
        return id_str, pres, temp, batt, flat_byte
    
    def check(input_pres, input_temp, input_batt, device_location):
        input_pres = float(input_pres)
        input_temp = float(input_temp)
        input_batt = float(input_batt)
        
        if "Front" in device_location:
            if input_pres <= front_min_pres:
                return "LOW PRESSURE"
            elif input_temp >= max_temp:
                return "HIGH TEMP"
            elif input_batt <= min_batt:
                return "LOW BATTERY"
            else:
                return "OKAY"
        elif "Rear" in device_location:
            if input_pres <= rear_min_pres:
                return "LOW PRESSURE"
            elif input_temp >= max_temp:
                return "HIGH TEMP"
            elif input_batt <= min_batt:
                return "LOW BATTERY"
            else:
                return "OKAY"

    while True:
        scan_entries = scanner.scan(2.0)
        for scan_entry in scan_entries:
            if scan_entry.addr in SENSORS:
                device_location = SENSORS[scan_entry.addr]
                manufacturer_hex = next(value for _, desc, value in scan_entry.getScanData() if desc == 'Manufacturer')
                chunks = [manufacturer_hex[i:i+2] for i in range(0, len(manufacturer_hex), 2)]
                
                id_str, pres_val, temp_val, batt_val, flat_byte = sort_data(chunks)

                # 更新失联计数
                def update_counts(reset_idx):
                    nonlocal fl_count, fr_count, rl_count, rr_count
                    counts = [fl_count, fr_count, rl_count, rr_count]
                    counts[reset_idx] = 0
                    for i in range(4):
                        if i != reset_idx:
                            counts[i] += 1
                            if counts[i] >= max_loss:
                                status[i*20:(i+1)*20] = b"LOST".ljust(20)
                    fl_count, fr_count, rl_count, rr_count = counts

                if flat_byte == "00":
                    if id_str == "80eaca12230b":
                        pres[0*20:1*20] = pres_val.ljust(20).encode()
                        temp[0*20:1*20] = temp_val.ljust(20).encode()
                        batt[0*20:1*20] = batt_val.ljust(20).encode()
                        status[0*20:1*20] = check(pres_val, temp_val, batt_val, device_location).ljust(20).encode()
                        flat[0*2:1*2] = flat_byte.encode()
                        update_counts(0)
                    elif id_str == "81eaca2220f7":
                        pres[1*20:2*20] = pres_val.ljust(20).encode()
                        temp[1*20:2*20] = temp_val.ljust(20).encode()
                        batt[1*20:2*20] = batt_val.ljust(20).encode()
                        status[1*20:2*20] = check(pres_val, temp_val, batt_val, device_location).ljust(20).encode()
                        flat[1*2:2*2] = flat_byte.encode()
                        update_counts(1)
                    elif id_str == "82eaca322487":
                        pres[2*20:3*20] = pres_val.ljust(20).encode()
                        temp[2*20:3*20] = temp_val.ljust(20).encode()
                        batt[2*20:3*20] = batt_val.ljust(20).encode()
                        status[2*20:3*20] = check(pres_val, temp_val, batt_val, device_location).ljust(20).encode()
                        flat[2*2:3*2] = flat_byte.encode()
                        update_counts(2)
                    elif id_str == "83eaca422307":
                        pres[3*20:4*20] = pres_val.ljust(20).encode()
                        temp[3*20:4*20] = temp_val.ljust(20).encode()
                        batt[3*20:4*20] = batt_val.ljust(20).encode()
                        status[3*20:4*20] = check(pres_val, temp_val, batt_val, device_location).ljust(20).encode()
                        flat[3*2:4*2] = flat_byte.encode()
                        update_counts(3)
                else:
                    status_val = "PUNCTURE"
                    if id_str == "80eaca12230b":
                        pres[0*20:1*20] = pres_val.ljust(20).encode()
                        temp[0*20:1*20] = temp_val.ljust(20).encode()
                        batt[0*20:1*20] = batt_val.ljust(20).encode()
                        status[0*20:1*20] = status_val.ljust(20).encode()
                        flat[0*2:1*2] = flat_byte.encode()
                    elif id_str == "81eaca2220f7":
                        pres[1*20:2*20] = pres_val.ljust(20).encode()
                        temp[1*20:2*20] = temp_val.ljust(20).encode()
                        batt[1*20:2*20] = batt_val.ljust(20).encode()
                        status[1*20:2*20] = status_val.ljust(20).encode()
                        flat[1*2:2*2] = flat_byte.encode()
                    elif id_str == "82eaca322487":
                        pres[2*20:3*20] = pres_val.ljust(20).encode()
                        temp[2*20:3*20] = temp_val.ljust(20).encode()
                        batt[2*20:3*20] = batt_val.ljust(20).encode()
                        status[2*20:3*20] = status_val.ljust(20).encode()
                        flat[2*2:3*2] = flat_byte.encode()
                    elif id_str == "83eaca422307":
                        pres[3*20:4*20] = pres_val.ljust(20).encode()
                        temp[3*20:4*20] = temp_val.ljust(20).encode()
                        batt[3*20:4*20] = batt_val.ljust(20).encode()
                        status[3*20:4*20] = status_val.ljust(20).encode()
                        flat[3*2:4*2] = flat_byte.encode()
        sleep(0)

重构后的HTML采集脚本（html_reader.py）

#! /usr/bin/python3 -u
import urllib.request
from time import sleep
from multiprocessing import Array

# 初始化HTML数据共享内存
def init_html_shared_data():
    # 名称和值各预留50字节空间
    html_name = Array('c', b' ' * 50)
    html_value = Array('c', b' ' * 50)
    return html_name, html_value

def html_reader_process(html_name, html_value):
    bad_chars = ['{', '":', ' "', '}', '"']
    coding = "Windows-1252"

    def convert_economy(input2):
        input2 = float(input2)
        output2 = round((input2 * 282.481), 1)
        return str(output2)

    def convert_speed(input3):
        input3 = float(input3)
        output3 = round((input3 * 0.6214))
        return str(output3)

    def convert_pres(input5):
        input5 = float(input5)
        output5 = (input5 / 100)
        return str(output5)

    while True:
        try:
            data = urllib.request.urlopen("http://192.168.4.1/readVal").read()
            decode = data.decode()
            for i in bad_chars:
                decode = decode.replace(i, '')

            list_data = decode.split(",")
            name_raw = list_data[0]
            value_raw = list_data[1]
            unit_raw = list_data[2]

            name = name_raw.replace("n", "", 1)
            value = value_raw.replace("v", "",).replace('<br />', '  ')
            unit = unit_raw.replace("u", "",)

            if name == 'Turbo boost':
                name = 'Turbo Boost'
                unit = 'bar'
                value = convert_pres(value)

            if unit != '':
                if unit == '&deg;C':
                    subunit = '\u00b0' + "C"
                    name = name + ' (' + subunit + ')'
                elif unit == 'km/h':
                    subunit = 'mph'
                    name = name + ' (' + subunit + ')'
                    value = convert_speed(value)
                elif unit == 'l/100':
                    subunit = 'mpg'
                    name = name + ' (' + subunit + ')'
                    value = convert_economy(value)
                else:
                    name = name + ' (' + unit + ')'

            # 更新共享内存
            html_name[:] = name.ljust(50).encode(coding)
            html_value[:] = value.ljust(50).encode(coding)

        except Exception as e:
            print(f"HTML读取异常: {e}")
        sleep(1)

步骤2：编写整合发送脚本（main.py） #

负责读取共享数据，按HMI要求拼接后发送串口。

#! /usr/bin/python3 -u
import serial
from time import sleep
from multiprocessing import Process
from ble_scanner import init_ble_shared_data, ble_scanner_process
from html_reader import init_html_shared_data, html_reader_process

def main():
    # 初始化共享内存
    pres, temp, batt, status, flat = init_ble_shared_data()
    html_name, html_value = init_html_shared_data()

    # 启动采集进程
    ble_process = Process(target=ble_scanner_process, args=(pres, temp, batt, status, flat))
    html_process = Process(target=html_reader_process, args=(html_name, html_value))
    ble_process.daemon = True
    html_process.daemon = True
    ble_process.start()
    html_process.start()

    # 初始化串口
    ser = serial.Serial('/dev/ttyS0', 9600)
    coding = "Windows-1252"

    while True:
        # 读取共享数据
        name = bytes(html_name).decode(coding).strip()
        value = bytes(html_value).decode(coding).strip()

        pres_fl = bytes(pres[0*20:1*20]).decode(coding).strip()
        pres_fr = bytes(pres[1*20:2*20]).decode(coding).strip()
        pres_rl = bytes(pres[2*20:3*20]).decode(coding).strip()
        pres_rr = bytes(pres[3*20:4*20]).decode(coding).strip()
        
        temp_fl = bytes(temp[0*20:1*20]).decode(coding).strip()
        temp_fr = bytes(temp[1*20:2*20]).decode(coding).strip()
        temp_rl = bytes(temp[2*20:3*20]).decode(coding).strip()
        temp_rr = bytes(temp[3*20:4*20]).decode(coding).strip()
        
        batt_fl = bytes(batt[0*20:1*20]).decode(coding).strip()
        batt_fr = bytes(batt[1*20:2*20]).decode(coding).strip()
        batt_rl = bytes(batt[2*20:3*20]).decode(coding).strip()
        batt_rr = bytes(batt[3*20:4*20]).decode(coding).strip()
        
        status_fl = bytes(status[0*20:1*20]).decode(coding).strip()
        status_fr = bytes(status[1*20:2*20]).decode(coding).strip()
        status_rl = bytes(status[2*20:3*20]).decode(coding).strip()
        status_rr = bytes(status[3*20:4*20]).decode(coding).strip()
        
        flat_fl = bytes(flat[0*2:1*2]).decode(coding).strip()
        flat_fr = bytes(flat[1*2:2*2]).decode(coding).strip()
        flat_rl = bytes(flat[2*2:3*2]).decode(coding).strip()
        flat_rr = bytes(flat[3*2:4*2]).decode(coding).strip()

        # 按HMI要求拼接：前两部分来自HTML，剩余来自BLE
        parts = [
            name, value,
            pres_fl, pres_fr, pres_rl, pres_rr,
            temp_fl, temp_fr, temp_rl, temp_rr,
            batt_fl, batt_fr, batt_rl, batt_rr,
            flat_fl, flat_fr, flat_rl, flat_rr,
            status_fl, status_fr, status_rl, status_rr
        ]
        # 替换空值为默认值，避免HMI解析错误
        parts = [p if p else "N