killian
6 months ago
parent
d2496fa8a2
commit
fc97ef6a24
@ -0,0 +1,42 @@
|
||||
"""
|
||||
Mac only.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
|
||||
def beep(sound):
|
||||
if "." not in sound:
|
||||
sound = sound + ".aiff"
|
||||
try:
|
||||
subprocess.Popen(["afplay", f"/System/Library/Sounds/{sound}"])
|
||||
except:
|
||||
pass # No big deal
|
||||
|
||||
class RepeatedBeep:
|
||||
def __init__(self):
|
||||
self.sound = "Pop"
|
||||
self.running = False
|
||||
self.thread = threading.Thread(target=self._play_sound, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
def _play_sound(self):
|
||||
while True:
|
||||
if self.running:
|
||||
try:
|
||||
subprocess.call(["afplay", f"/System/Library/Sounds/{self.sound}.aiff"])
|
||||
except:
|
||||
pass # No big deal
|
||||
time.sleep(0.6)
|
||||
time.sleep(0.05)
|
||||
|
||||
def start(self):
|
||||
if not self.running:
|
||||
time.sleep(0.6*4)
|
||||
self.running = True
|
||||
|
||||
def stop(self):
|
||||
self.running = False
|
||||
|
||||
beeper = RepeatedBeep()
|
||||
@ -1,64 +0,0 @@
|
||||
import os
|
||||
from datetime import datetime
|
||||
from pytimeparse import parse
|
||||
from crontab import CronTab
|
||||
from uuid import uuid4
|
||||
from platformdirs import user_data_dir
|
||||
|
||||
|
||||
def schedule(message="", start=None, interval=None) -> None:
|
||||
"""
|
||||
Schedules a task at a particular time, or at a particular interval
|
||||
"""
|
||||
if start and interval:
|
||||
raise ValueError("Cannot specify both start time and interval.")
|
||||
|
||||
if not start and not interval:
|
||||
raise ValueError("Either start time or interval must be specified.")
|
||||
|
||||
# Read the temp file to see what the current session is
|
||||
session_file_path = os.path.join(user_data_dir("01"), "01-session.txt")
|
||||
|
||||
with open(session_file_path, "r") as session_file:
|
||||
file_session_value = session_file.read().strip()
|
||||
|
||||
prefixed_message = "AUTOMATED MESSAGE FROM SCHEDULER: " + message
|
||||
|
||||
# Escape the message and the json, cron is funky with quotes
|
||||
escaped_question = prefixed_message.replace('"', '\\"')
|
||||
json_data = f'{{\\"text\\": \\"{escaped_question}\\"}}'
|
||||
|
||||
command = f"""bash -c 'if [ "$(cat "{session_file_path}")" == "{file_session_value}" ]; then /usr/bin/curl -X POST -H "Content-Type: application/json" -d "{json_data}" http://localhost:10001/; fi' """
|
||||
|
||||
cron = CronTab(user=True)
|
||||
job = cron.new(command=command)
|
||||
# Prefix with 01 dev preview so we can delete them all in the future
|
||||
job_id = "01-dev-preview-" + str(uuid4())
|
||||
job.set_comment(job_id)
|
||||
if start:
|
||||
try:
|
||||
start_time = datetime.strptime(start, "%Y-%m-%d %H:%M:%S")
|
||||
except ValueError:
|
||||
raise ValueError(f"Invalid datetime format: {start}.")
|
||||
job.setall(start_time)
|
||||
print(f"Task scheduled for {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
|
||||
elif interval:
|
||||
seconds = parse(interval)
|
||||
if seconds <= 60:
|
||||
job.minute.every(1)
|
||||
print("Task scheduled every minute")
|
||||
elif seconds < 3600:
|
||||
minutes = max(int(seconds / 60), 1)
|
||||
job.minute.every(minutes)
|
||||
print(f"Task scheduled every {minutes} minutes")
|
||||
elif seconds < 86400:
|
||||
hours = max(int(seconds / 3600), 1)
|
||||
job.hour.every(hours)
|
||||
print(f"Task scheduled every {hours} hour(s)")
|
||||
else:
|
||||
days = max(int(seconds / 86400), 1)
|
||||
job.day.every(days)
|
||||
print(f"Task scheduled every {days} day(s)")
|
||||
|
||||
cron.write()
|
||||
Loading…
Reference in new issue