Python sched: Event Scheduling

import sched
import time
 
scheduler = sched.scheduler(time.time, time.sleep)
 
def say_hello(name):
    print(f"Hello, {name}! Time: {time.strftime('%H:%M:%S')}")
 
# Schedule for 3 seconds from now
scheduler.enter(3, 1, say_hello, ('Alice',))
 
print(f"Starting at {time.strftime('%H:%M:%S')}")
scheduler.run()  # Blocks until all events complete

import sched
import time
 
scheduler = sched.scheduler(time.time, time.sleep)
 
def task(name):
    print(f"{time.strftime('%H:%M:%S')} - {name}")
 
# Schedule multiple tasks
scheduler.enter(1, 1, task, ('First',))
scheduler.enter(2, 1, task, ('Second',))
scheduler.enter(3, 1, task, ('Third',))
 
print(f"Started at {time.strftime('%H:%M:%S')}")
scheduler.run()

import sched
import time
 
scheduler = sched.scheduler(time.time, time.sleep)
 
def task(name):
    print(f"{name}")
 
# Same time, different priorities
scheduler.enter(1, 2, task, ('Low priority',))
scheduler.enter(1, 1, task, ('High priority',))  # Runs first
 
scheduler.run()
# Output:
# High priority
# Low priority

import sched
import time
 
scheduler = sched.scheduler(time.time, time.sleep)
 
def alarm():
    print(f"ALARM! {time.strftime('%H:%M:%S')}")
 
# Schedule for absolute timestamp
target_time = time.time() + 5  # 5 seconds from now
scheduler.enterabs(target_time, 1, alarm)
 
scheduler.run()

import sched
import time
 
scheduler = sched.scheduler(time.time, time.sleep)
 
def task(name):
    print(name)
 
# Schedule events
event1 = scheduler.enter(1, 1, task, ('First',))
event2 = scheduler.enter(2, 1, task, ('Second',))
event3 = scheduler.enter(3, 1, task, ('Third',))
 
# Cancel the second event
scheduler.cancel(event2)
 
scheduler.run()
# Output: First, Third (Second was canceled)

import sched
import time
 
scheduler = sched.scheduler(time.time, time.sleep)
 
scheduler.enter(1, 1, lambda: print('A'))
scheduler.enter(2, 1, lambda: print('B'))
 
# Check if events are pending
print(f"Empty: {scheduler.empty()}")  # False
print(f"Queue: {scheduler.queue}")    # List of events

import sched
import time
 
scheduler = sched.scheduler(time.time, time.sleep)
 
scheduler.enter(0.1, 1, lambda: print('Quick'))
scheduler.enter(10, 1, lambda: print('Later'))
 
# Run with blocking=False
time.sleep(0.2)  # Let first event become ready
scheduler.run(blocking=False)  # Runs 'Quick', returns immediately
print("Continuing without waiting for 'Later'")

import sched
import time
 
scheduler = sched.scheduler(time.time, time.sleep)
 
def repeat(interval, func, *args):
    """Schedule a repeating task."""
    func(*args)
    scheduler.enter(interval, 1, repeat, (interval, func) + args)
 
def heartbeat():
    print(f"Beat: {time.strftime('%H:%M:%S')}")
 
# Start repeating every 2 seconds
scheduler.enter(0, 1, repeat, (2, heartbeat))
 
# Run for limited time
try:
    scheduler.run()
except KeyboardInterrupt:
    print("Stopped")

import sched
import time
import threading
 
def run_with_timeout(func, timeout, *args):
    """Run function with timeout."""
    result = [None]
    exception = [None]
    
    def wrapper():
        try:
            result[0] = func(*args)
        except Exception as e:
            exception[0] = e
    
    thread = threading.Thread(target=wrapper)
    thread.start()
    thread.join(timeout)
    
    if thread.is_alive():
        raise TimeoutError(f"Function timed out after {timeout}s")
    
    if exception[0]:
        raise exception[0]
    
    return result[0]

import sched
import time
 
class Simulation:
    def __init__(self):
        self.scheduler = sched.scheduler(time.time, time.sleep)
        self.events = []
    
    def log_event(self, message):
        timestamp = time.strftime('%H:%M:%S')
        self.events.append(f"{timestamp}: {message}")
        print(f"{timestamp}: {message}")
    
    def schedule(self, delay, message):
        self.scheduler.enter(delay, 1, self.log_event, (message,))
    
    def run(self):
        self.scheduler.run()
 
# Run simulation
sim = Simulation()
sim.schedule(1, "Server started")
sim.schedule(2, "Client connected")
sim.schedule(3, "Data received")
sim.schedule(4, "Response sent")
sim.run()

import sched
import time
 
def countdown(scheduler, seconds):
    """Display countdown timer."""
    if seconds > 0:
        print(f"{seconds}...")
        scheduler.enter(1, 1, countdown, (scheduler, seconds - 1))
    else:
        print("Time's up!")
 
scheduler = sched.scheduler(time.time, time.sleep)
scheduler.enter(0, 1, countdown, (scheduler, 5))
scheduler.run()

import sched
import time
from collections import deque
 
class RateLimiter:
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
        self.calls = deque()
        self.scheduler = sched.scheduler(time.time, time.sleep)
    
    def __call__(self, func):
        def wrapper(*args, **kwargs):
            now = time.time()
            
            # Remove old calls
            while self.calls and self.calls[0] < now - self.period:
                self.calls.popleft()
            
            if len(self.calls) >= self.max_calls:
                wait = self.calls[0] + self.period - now
                time.sleep(wait)
            
            self.calls.append(time.time())
            return func(*args, **kwargs)
        
        return wrapper
 
@RateLimiter(max_calls=3, period=1.0)
def api_call():
    print(f"API call at {time.strftime('%H:%M:%S.%f')}")

import sched
 
class VirtualClock:
    def __init__(self):
        self.time = 0
    
    def get_time(self):
        return self.time
    
    def sleep(self, duration):
        self.time += duration
 
clock = VirtualClock()
scheduler = sched.scheduler(clock.get_time, clock.sleep)
 
# Events run instantly with virtual time
scheduler.enter(100, 1, lambda: print("100 virtual seconds"))
scheduler.enter(200, 1, lambda: print("200 virtual seconds"))
scheduler.run()  # Completes immediately