Reputation: 317
For the first "if" statement, I want the GPIO to be low for 5 seconds and then high for 1 second. At the same time, I do not want to sleep to pause the second if statement. I.e. I want the script to be able to execute the ss.moisture if statement while the photo_volt() is pausing.
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()
Upvotes: 0
Views: 62
Reputation: 15349
One of the problems here is that we don't have a specification of the complete behaviour of the system (for example, after each high photo_volt()
or moisture_read()
reading, is there a minimum length of time the corresponding relay should remain HIGH
, like there is for LOW
...?). But here's an example of how it could work. I'll use independent queues of events for the two relays, without threads (I have no idea if the GPIO
library is going to be thread-safe).
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()
Upvotes: 0
Reputation: 2756
@dryliketoast 's answer is best if it works for this - certainly best if you can just do it with logic, without needing parallel execution.
anyhow, here is what a version with async task might look like - similar to logic of threads, but without using threads
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())
maybe not exactly right but in the direction, is on repl.it at: https://repl.it/repls/DoubleDeepskyblueCommas
that's relatively new python feats since 3.5, earlier there were generators with yield, https://docs.python.org/3/library/asyncio-task.html
Upvotes: 0
Reputation: 153
Instead of sprinkling sleep
within your if
blocks (which halts execution) you could use counters and a single sleep
command at the end of the loop which is only used to pause at 1 second intervals. Something like this:
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)
Upvotes: 2
Reputation: 531275
Start a new thread for the pair of GPIO calls.
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()
Upvotes: 2