Определите, включены ли аварийные огни, используя python для управления светодиодами SK6812.

В нашем туристическом прицепе у меня есть устройство ESP32, которое отслеживает входы от 7-контактного жгута прицепа. 7-контактный жгут проводов прицепа используется тягачом для включения/выключения различных огней, зарядки аккумулятора и управления тормозами прицепа. ESP32 отправляет темы сообщений MQTT в зависимости от состояния каждого вывода. Например, когда горит левый или правый сигнал поворота и стоп-сигнал, я получу:

Topic: harness/left_turn
Payload: True/False
Topic: harness/right_turn
Payload: True/False

В моем коде python3 у меня есть класс, который собирает эти состояния:

self.left_turn
self.right_turn

Когда включены и left_turn, и right_turn, предполагается, что тормоза включены. Я не подключал контакт управления тормозом к ESP32, так как вообще не хочу возиться с тормозами.

У меня уже есть метод, чтобы определить, горят ли стоп-сигналы, и устанавливает:

self.brake_lights_state # bool

Проблема, с которой я столкнулся сейчас, заключается в том, чтобы определить, включены ли аварийные огни, то есть если «self.brake_lights_state» изменяется с обычной частотой.

Предполагая, что аварийные огни включаются на 400 мс и выключаются на 400 мс, как я могу это обнаружить? Я хочу, чтобы ложные срабатывания не нажимали на тормоза. Например, при движении задним ходом прицепа обычно немного приглушают тормоза/газ. Итак, я бы хотел, чтобы опасности включались, скажем, после того, как он обнаружит, что огни мигают 3 или 4 раза в течение этого почти точного интервала времени 400 мс. Я говорю почти точно, потому что иногда по какой-то странной причине каждое 5-е мигание тягача имеет скорость 400 мс при включении и 325 мс при выключении.

Я совершенно не в себе. Я предполагаю, что я мог бы каким-то образом использовать коллекцию deque и time(), чтобы отслеживать, когда изменяетсяrake_lights_state.

У кого-нибудь есть идеи, с чего начать? Я запускаю все это в потоке и использую метод run(), когда трейлер буксируется.

def run(self):
 """
 Called when the thread is running.
 Monitors the 7 pin harness state and updates the SK6812 leds as needed.
 """
 while not self.die:
 if self.is_driving_state:
 if self.brake_lights_state:
 self.do_brake()
 elif self.right_blinker_state:
 self.do_right_blinker()
 elif self.left_blinker_state:
 self.do_right_blinker()
 if self.reverse_lights_state:
 self.do_reverse()
 if self.marker_lights_state:
 self.do_marker()
 sleep(0.05)
 else:
 sleep(1)
def do_brake(self):
 # check if hazard lights are on first... 
 # Then call self.do_hazard(), else do this:
 if self.brake_lights_state:
 self.set_lights(self.leds['rear_right_blinker'], BRAKE_COLOR)
 self.set_lights(self.leds['rear_left_blinker'], BRAKE_COLOR)
 else:
 self.set_lights(self.leds['rear_right_blinker'], OFF_COLOR)
 self.set_lights(self.leds['rear_left_blinker'], OFF_COLOR)

Решение проблемы

Я не разработчик Python, но именно так мне удалось заставить его работать. Это не изящно, но это работает.

from time import time, sleep
from collections import deque
BLINKER_ON_TIME = [350, 400] # Range of time the blinker can be on
BLINKER_OFF_TIME = [350, 400] # Range of time the blinker can be off
HAZARD_MIN_CYCLE_COUNT = 5
MAX_BRAKE_HISTORY_TIME = ((((BLINKER_ON_TIME[1]+5)*HAZARD_MIN_CYCLE_COUNT)+((BLINKER_OFF_TIME[1]+5)*HAZARD_MIN_CYCLE_COUNT)) - BLINKER_ON_TIME[0]) / 1000
MIN_BRAKE_HISTORY_TIME = ((((BLINKER_ON_TIME[0])*HAZARD_MIN_CYCLE_COUNT)+((BLINKER_OFF_TIME[0])*HAZARD_MIN_CYCLE_COUNT)) - BLINKER_ON_TIME[1])/ 1000
brake_history = deque([], maxlen=HAZARD_MIN_CYCLE_COUNT*2)
brake = False
previous_brake = False
hazard = False
timer = time()
change_timeout =.400
while True:
 # Simulate changing the brakes on/off after every 'change_timeout'
 if time() - timer > change_timeout:
 timer = time()
 brake = not brake
 # If the brake state has changed, lets check if hazards should be on/off
 if brake!= previous_brake:
 brake_history.append([time(), brake])
 previous_brake = brake
 if len(brake_history) == HAZARD_MIN_CYCLE_COUNT*2: # must have some data.
 if MIN_BRAKE_HISTORY_TIME < time() - brake_history[0][0] < MAX_BRAKE_HISTORY_TIME:
 if hazard is False:
  print(f"{time()}: hazard on....")
 hazard = True
 change_timeout =.300 # Lets simulate the timing being way off, to clear the hazard state.
 else:
 if hazard is True:
  print(f"{time()}: hazard off....")
 hazard = False
 brake_history.clear()
 change_timeout =.375
 sleep(0.05)