AirlockTools/utils/perstscheduler.py

import schedule
import time
import json
import os
from typing import Callable, Any, List, Dict

# File where all jobs are persisted
JOBS_FILE = "scheduling\\jobs.json"

# Registry of functions that can be scheduled
FUNCTION_MAP: Dict[str, Callable] = {}

# -------------------------------
# Function Registration
# -------------------------------

def register_function(name: str, func: Callable):
    """
    Register a function so it can be called by name later.

    Example:
        register_function("say_hello", say_hello)
    """
    FUNCTION_MAP[name] = func

# -------------------------------
# Persistence Helpers
# -------------------------------

def load_jobs() -> List[Dict[str, Any]]:
    """Load jobs from the JSON file, or return [] if none exist."""
    if not os.path.exists(JOBS_FILE):
        return []
    with open(JOBS_FILE, "r") as f:
        return json.load(f)

def save_jobs(jobs: List[Dict[str, Any]]):
    """Save jobs to the JSON file (overwrite)."""
    with open(JOBS_FILE, "w") as f:
        json.dump(jobs, f, indent=4)

# -------------------------------
# Run Once Jobs
# -------------------------------

def run_once_job(job_id: str, func_name: str, run_at_timestamp: float, args=None, kwargs=None):
    """
    Schedule a job to run once at a specific timestamp.
    """
    args = args or []
    kwargs = kwargs or {}

    def job_wrapper():
        """Executes the job once, then removes it."""
        if func_name not in FUNCTION_MAP:
            print(f"[ERROR] Function '{func_name}' is not registered.")
            return
        FUNCTION_MAP[func_name](*args, **kwargs)

        # Remove from persistence
        jobs = load_jobs()
        jobs = [j for j in jobs if j["id"] != job_id]
        save_jobs(jobs)

        # Clear from in-memory schedule
        schedule.clear(job_id)

    delay_seconds = run_at_timestamp - time.time()
    if delay_seconds <= 0:
        print(f"[WARN] Job {job_id} scheduled in the past. Skipping.")
        return

    # Schedule via schedule library
    schedule.every(delay_seconds).seconds.do(job_wrapper).tag(job_id)

    # Save to JSON
    jobs = load_jobs()
    jobs.append({
        "id": job_id,
        "type": "once",
        "run_at": run_at_timestamp,
        "function": func_name,
        "args": args,
        "kwargs": kwargs
    })
    save_jobs(jobs)

# -------------------------------
# Recurring Jobs
# -------------------------------

def recurring_job(job_id: str, func_name: str, interval: int, unit: str, args=None, kwargs=None):
    """
    Schedule a recurring job.
    
    Args:
        job_id: Unique job name
        func_name: Function to call (must be registered)
        interval: Number of units between runs
        unit: "seconds", "minutes", "hours", "days"
        args: Positional arguments for the function
        kwargs: Keyword arguments for the function
    """
    args = args or []
    kwargs = kwargs or {}

    def job_wrapper():
        if func_name not in FUNCTION_MAP:
            print(f"[ERROR] Function '{func_name}' is not registered.")
            return
        FUNCTION_MAP[func_name](*args, **kwargs)

    # Pick correct scheduling unit
    if unit == "seconds":
        schedule.every(interval).seconds.do(job_wrapper).tag(job_id)
    elif unit == "minutes":
        schedule.every(interval).minutes.do(job_wrapper).tag(job_id)
    elif unit == "hours":
        schedule.every(interval).hours.do(job_wrapper).tag(job_id)
    elif unit == "days":
        schedule.every(interval).days.do(job_wrapper).tag(job_id)
    else:
        raise ValueError(f"Unsupported unit: {unit}")

    # Save to JSON
    jobs = load_jobs()
    # Ensure no duplicate job id in file
    jobs = [j for j in jobs if j["id"] != job_id]
    jobs.append({
        "id": job_id,
        "type": "recurring",
        "interval": interval,
        "unit": unit,
        "function": func_name,
        "args": args,
        "kwargs": kwargs
    })
    save_jobs(jobs)

# -------------------------------
# Reload Saved Jobs
# -------------------------------

def reload_jobs():
    """Reload jobs from JSON and reschedule them."""
    jobs = load_jobs()
    for job in jobs:
        if job["type"] == "once":
            # Only reschedule if still in the future
            if job["run_at"] > time.time():
                run_once_job(job["id"], job["function"], job["run_at"], job.get("args"), job.get("kwargs"))
        elif job["type"] == "recurring":
            recurring_job(
                job["id"],
                job["function"],
                job["interval"],
                job["unit"],
                job.get("args"),
                job.get("kwargs")
            )

# -------------------------------
# Scheduler Loop
# -------------------------------

def start_scheduler():
    """
    Start the scheduler loop (blocking).
    Call this once in your main program to begin.
    """
    try:
        while True:
            schedule.run_pending()
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("[INFO] Scheduler stopped.")


"""
import time
from persistent_schedule import register_function, run_once_job, recurring_job, reload_jobs, start_scheduler

# Example functions
def greet(name, loud=False):
    if loud:
        print(f"HELLO, {name}!")
    else:
        print(f"Hello, {name}.")

def add(a, b):
    print(f"{a} + {b} = {a + b}")

# Register functions
register_function("greet", greet)
register_function("add", add)

# Reload saved jobs
reload_jobs()

# Schedule a run-once job in 5 seconds
run_once_job("job1", "greet", time.time() + 5, args=["Alice"], kwargs={"loud": True})

# Schedule a recurring job every 10 seconds
recurring_job("job2", "add", interval=10, unit="seconds", args=[2, 3])

# Start scheduler loop
start_scheduler()

"""