Как приостановить «если», но одновременно запустить другую «если» рутину?

Question

Для первого оператора «if» я хочу, чтобы GPIO был низким в течение 5 секунд, а затем высоким в течение 1 секунды. В то же время, я не хочу спать, чтобы приостановить второе заявление if. Т.е. я хочу, чтобы скрипт мог выполнять инструкцию ss.moisture if во время приостановки photo_volt ().

try:
    while True:
        if photo_volt() < 1.6:
            GPIO.output(RELAY_1, GPIO.LOW)
            sleep(5)
            GPIO.output(RELAY_1, GPIO.HIGH)
            sleep(1)
        elif photo_volt() > 1.6:
            GPIO.output(RELAY_1, GPIO.HIGH)
        if ss.moisture_read() < 350
            GPIO.output(RELAY_2, GPIO.LOW)
            sleep(5)
         elif ss.moisture_read() > 350
            GPIO.output(RELAY_2, GPIO.HIGH)
except KeyboardInterrupt:
    GPIO.cleanup()

Answer 1

Вместо добавления sleep в ваши блоки if (что останавливает выполнение) вы можете использовать счетчики и одну команду sleep в конце l oop, которая используется только для паузы с интервалом в 1 секунду. , Как то так:

counter = 0
while True:
    counter += 1
    if counter == 1:
        if photo_volt() < 1.6:
            GPIO.output(RELAY_1, GPIO.LOW)
        elif photo_volt() > 1.6:
            GPIO.output(RELAY_1, GPIO.HIGH)
        if ss.moisture_read() < 350
            GPIO.output(RELAY_2, GPIO.LOW)
        elif ss.moisture_read() > 350
            GPIO.output(RELAY_2, GPIO.HIGH)
    if counter == 5:
        counter = 0
    sleep(1)

Answer 2

Создать новый поток для пары вызовов GPIO.

from threading import Thread


def relay1_low_high():
    GPIO.output(RELAY_1, GPIO.LOW)
    sleep(5)
    GPIO.output(RELAY_1, GPIO.HIGH)


try:
    while True:
        if photo_volt() < 1.6:
            Thread(target=low_high).start()
        elif photo_volt() > 1.6:
            GPIO.output(RELAY_1, GPIO.HIGH)

        ...
finally:  # You probably want to do this no matter why the `try` statement exits.
    GPIO.cleanup()

Answer 3

Одна из проблем здесь заключается в том, что у нас нет спецификации поведения системы complete (например, после каждого high photo_volt() или moisture_read() читая, есть ли минимальное время, в течение которого соответствующее реле должно оставаться HIGH, как для LOW ...?). Но вот пример того, как это может работать. Я буду использовать независимые очереди событий для двух реле без потоков (я понятия не имею, будет ли библиотека GPIO поточно-ориентированной).

import time
import functools # we'll use functools.partial to "pre-bake" our GPIO.output calls


def check_queue(queue, t):
    while True:
        if not queue:
            return True              # Return True to indicate "this queue is empty: feel free to take new readings and add more commands"
        first = queue[0]
        if isinstance(first, float): # A numeric entry in the queue means "do nothing until the clock reaches this number"
            if t < first:
                return False         # Return False to indicate "commands are still pending in this queue: don't take any more readings yet"
            else:
                queue.pop(0)         # Time's up: immediately proceed to the next item in the queue
        else:                        # Any non-numeric queue entry is assumed to be a command for immediate execution
            command = queue.pop(0)
            command()

try:

    relay1_commands = []
    relay2_commands = []

    while True:
        t = time.time()  # current time

        if check_queue(relay1_commands, t):
            if photo_volt() < 1.6:
                relay1_commands += [
                    functools.partial(GPIO.output, RELAY_1, GPIO.LOW),
                    t + 5.0,   # wait until 5 seconds from now (no more photo_volt readings or outputs on RELAY_1 till then)
                    functools.partial(GPIO.output, RELAY_1, GPIO.HIGH),
                    t + 6.0,   # this is literally what the question asks for, but is it necessary?

                    # so, after a low photo_volt() reading, RELAY_1 should be constantly LOW for
                    # at least 5 seconds, then automatically HIGH (regardless of photo_volt state)
                    # for at least 1 second
                ]
            else:
                relay1_commands += [
                    functools.partial(GPIO.output, RELAY_1, GPIO.HIGH),
                    t + 1.0,
                    # after a high photo_volt() reading, RELAY_1 should be constantly HIGH for
                    # at least.... what?  That's not specified in the question. Here I've arbitrarily
                    # said 1 second.
                ]

        if check_queue(relay2_commands, t):
            if ss.moisture_read() < 350:
                relay2_commands += [
                    functools.partial(GPIO.output, RELAY_2, GPIO.LOW),
                    t + 5.0,  # wait until then (no more photo_volt readings or outputs on RELAY_2)

                    # so, after a low ss.moisture() reading, RELAY_2 should be constantly LOW for
                    # at least 5 seconds, but unlike RELAY_1 there shouldn't be an automatic HIGH
                    # period after that (that's what the question appears to specify)
                ]
            else:
                relay2_commands += [
                    functools.partial(GPIO.output, RELAY_2, GPIO.HIGH),
                    t + 1.0,
                    # after a high ss.moisture() reading, RELAY_2 should be constantly HIGH for
                    # at least.... what?  That's not specified in the question. Here I've arbitrarily
                    # said 1 second.
                ]

        time.sleep(0.001)  # or however long is appropriate between repeated checks

finally:  # as chepner says:  You probably want to do this no matter why the `try` statement exits.
    GPIO.cleanup()

Answer 4

@ dryliketoast ответ лучше, если он работает для этого - конечно, лучше, если вы можете просто сделать это с помощью logi c, без необходимости параллельного выполнения.

в любом случае, вот какая версия с asyn c задача может выглядеть так - похоже на логи c потоков, но без использования потоков

import asyncio
import random

def photo_volt():
  return random.choice([1,2])

async def relay1_low_high():
  while True:
    print("Relay1: GPIO.output(RELAY_1, GPIO.LOW)")
    await asyncio.sleep(5)
    print("Relay1: GPIO.output(RELAY_1, GPIO.HIGH)")

async def main():
  task = asyncio.create_task(relay1_low_high())
  try:
      while True:
          if photo_volt() < 1.6:
              await task
          elif photo_volt() > 1.6:
              print("Main: GPIO.output(RELAY_1, GPIO.HIGH)")

  finally:  # You probably want to do this no matter why the `try` statement exits.
      print("GPIO.cleanup()")

asyncio.run(main())

может быть не совсем верно, но в направлении, на repl.it по адресу: https://repl.it/repls/DoubleDeepskyblueCommas

это относительно новые python умения с 3.5, ранее были генераторы с доходностью, https://docs.python.org/3/library/asyncio-task.html