chatgpt/letters/fifo-sample_with-commands.py

36 lines
882 B
Python
Raw Normal View History

#!/usr/bin/python3
import os
import sys
import queue
import threading
command_queue = queue.Queue()
fifo_file = "/tmp/my_fifo"
def listen_to_fifo(queue):
if not os.path.exists(fifo_file):
os.mkfifo(fifo_file)
with open(fifo_file, 'r') as f:
while True:
data = f.readline().strip()
if not data:
break
queue.put(data)
def read_queue(queue):
while True:
data = queue.get()
if data == "reboot":
fifo_file.close()
os.execv(sys.executable, [sys.executable] + sys.argv)
print(data)
sys.stdout.flush()
queue.task_done()
if __name__ == '__main__':
t1 = threading.Thread(target=listen_to_fifo, args=(command_queue,))
t2 = threading.Thread(target=read_queue, args=(command_queue,))
t1.start()
t2.start()
command_queue.join()