Skip to main content
 首页 » 编程设计

python之使用 Python 获取基于 Spark 塞驱动的内燃机状态

2025年04月02日41dflying

致力于为定制 buggy 创建数据采集系统。使用 Raspberry Pi 和自定义转速计(测试并确认工作)来测量 RPM。在以下代码中使用中断来获取 RPM 值。

def get_rpm():                                          
    GPIO.wait_for_edge(17, GPIO.FALLING) 
    start = time.time() 
    GPIO.wait_for_edge(17, GPIO.FALLING) 
    end = time.time() 
    duration = end - start 
    rpm = (1/duration)*60 
    return rpm 

此代码仅在发动机运行并产生 Spark 时有效。如果没有 Spark ,代码将等待该边缘并且不会继续。调用 get_rpm() 时,如果代码正在等待边缘,这会导致其他进程挂起。

我打算解决这个问题的方法是在另一个进程中获取引擎的状态。我认为它最好分为两部分。

第 1 部分,在单独的线程中运行(循环):

GPIO.wait_for_edge(17, GPIO.RISING) 
last = time.time 

第 2 部分,根据需要将调用作为函数运行:

def get_state(): 
    while time.time - last < .5: 
        engine_state = true 
    else: 
        engine_state = false 
    return engine_state 

第 1 部分将 last 保存到第 2 部分可访问的内存中,第 2 部分将根据 Spark 塞最后一次点火来确定汽车是否正在运行。使用engine_state作为比较器,只有当engine_state为真时,数据采集系统才会从get_rpm()获取并存储RPM值。

如何以可以在第 2 部分中使用 last 变量的方式实现第 1 部分? last 的变化非常非常快。我不想每次更新 last 时都将它存储到 Raspberry Pi 的 SD 卡上的文本文件中。我想将 last 存储在 RAM 中。

非常感谢!

请您参考如下方法:

这只是为了灵感。我手边没有我的 Pis,所以这是凭内存盲目写的。

import RPi.GPIO as GPIO 
import time 
 
GPIO.setmode(GPIO.BCM) 
GPIO.setup(17, GPIO.IN) 
# the line below is important. It'll fire on both rising and falling edges 
# and it's non-blocking 
GPIO.add_event_detect(17, GPIO.BOTH, callback=callback_func) 
 
length_of_last_high = 0 
length_of_last_low = 0 
 
last_rise = 0 
last_fall = 0 
last_callback = 0 
 
def callback_func(pin): 
    # all of these global variables aren't needed  
    # but I left them in for demo purposes 
    global last_rise 
    global last_fall 
    global last_callback 
    global length_of_last_high 
    global length_of_last_low 
    last_callback = time.time() 
    if GPIO.input(17): 
        print "Pin 17 is rising!" 
        length_of_last_high = last_callback - last_rise 
        last_rise = last_callback 
    else: 
        print "Pin 17 is falling!" 
        length_of_last_low = last_callback - last_fall  
        last_fall = last_callback 
 
 
# last passed as parameter (the preferred solution if it works). 
# You test it - I can't at the moment 
def get_state(last=last_rise): 
    engine_state = False 
    if time.time() - last < .5: 
        engine_state = True 
    return engine_state 
 
# last as global variable.  
# This works too but global is bad practice if it can be avoided      
# (namespace littering) 
def get_state2(): 
    global last_rise 
    engine_state = False 
    if time.time() - last_rise < .5: 
        engine_state = True 
    return engine_state 
 
 
def main(): 
    while True: 
        print "Not blocking! This loop could sleep forever. It wouldn't affect the readings" 
        time.sleep(1) 
        print get_state() 
 
""" 
GPIO.cleanup() 
"""