🔍 Code Extractor

A threading class that executes a callback function periodically at specified intervals, with optional count limits and timeout controls.

Purpose

The periodic class provides a mechanism to run a callback function repeatedly at regular intervals without blocking the main thread. It extends Python's Thread class to enable both blocking and non-blocking execution modes. The class is useful for scheduled tasks, polling operations, periodic data collection, or any scenario requiring repeated function execution at fixed time intervals. It supports limiting executions by count, timeout duration, or running indefinitely until manually stopped.

Source Code

class periodic(Thread):
    """
    Run a callback count times with a given period without blocking.

    If count is None, will run till timeout (which may be forever if None).
    """

    def __init__(self, period, count, callback, timeout=None, block=False):

        if isinstance(count, int):
            if count < 0: raise ValueError('Count value must be positive')
        elif not type(count) is type(None):
            raise ValueError('Count value must be a positive integer or None')

        if block is False and count is None and timeout is None:
            raise ValueError('When using a non-blocking thread, please specify '
                             'either a count or a timeout')

        super().__init__()
        self.period = period
        self.callback = callback
        self.count = count
        self.counter = 0
        self.block = block
        self.timeout = timeout
        self._completed = Event()
        self._start_time = None

    @property
    def completed(self):
        return self._completed.is_set()

    def start(self):
        self._start_time = time.time()
        if self.block is False:
            super().start()
        else:
            self.run()

    def stop(self):
        self.timeout = None
        self._completed.set()

    def __repr__(self):
        return 'periodic(%s, %s, %s)' % (self.period,
                                         self.count,
                                         callable_name(self.callback))
    def __str__(self):
        return repr(self)

    def run(self):
        while not self.completed:
            if self.block:
                time.sleep(self.period)
            else:
                self._completed.wait(self.period)
            self.counter += 1
            try:
                self.callback(self.counter)
            except Exception:
                self.stop()

            if self.timeout is not None:
                dt = (time.time() - self._start_time)
                if dt > self.timeout:
                    self.stop()
            if self.counter == self.count:
                self.stop()

Parameters

Parameter Details

period: The time interval (in seconds) between consecutive callback executions. This is a float value representing the sleep duration between calls.

count: The maximum number of times to execute the callback. Must be a positive integer or None. If None, the callback runs until timeout is reached or stop() is called. If an integer, execution stops after that many iterations.

callback: A callable function that will be executed periodically. The callback receives one argument: the current counter value (iteration number starting from 1).

timeout: Optional maximum duration (in seconds) for the periodic execution. If specified, the thread will stop after this time has elapsed since start(). Can be None for no timeout limit.

block: Boolean flag controlling execution mode. If False (default), runs in a separate thread (non-blocking). If True, runs in the current thread (blocking). When False with count=None and timeout=None, raises ValueError to prevent infinite non-blocking threads.

Return Value

Instantiation returns a periodic object (Thread subclass). The start() method returns None but initiates the periodic execution. The completed property returns a boolean indicating whether execution has finished. The stop() method returns None but terminates execution.

Class Interface

Methods

Attributes

Dependencies

• threading
• time

Required Imports

from threading import Thread
from threading import Event
import time

Usage Example

from threading import Thread, Event
import time

# Define a callback function
def my_callback(iteration):
    print(f'Iteration {iteration} at {time.time()}')

# Example 1: Run 5 times with 1-second intervals (non-blocking)
periodic_task = periodic(period=1.0, count=5, callback=my_callback)
periodic_task.start()
# Main thread continues...
periodic_task.join()  # Wait for completion

# Example 2: Run for 10 seconds with 0.5-second intervals
periodic_task2 = periodic(period=0.5, count=None, callback=my_callback, timeout=10)
periodic_task2.start()

# Example 3: Blocking execution
periodic_task3 = periodic(period=2.0, count=3, callback=my_callback, block=True)
periodic_task3.start()  # Blocks until complete

# Example 4: Manual stop
periodic_task4 = periodic(period=1.0, count=100, callback=my_callback)
periodic_task4.start()
time.sleep(5)
periodic_task4.stop()  # Stop early

# Check completion status
if periodic_task4.completed:
    print('Task completed')

Best Practices

• Always specify either count or timeout when using non-blocking mode (block=False) to prevent uncontrollable infinite threads
• Call join() on non-blocking periodic instances to wait for completion before program exit
• Use the completed property to check if execution has finished before accessing results
• Handle exceptions within your callback function or be aware that exceptions will trigger automatic stop()
• Call stop() to terminate execution early if needed, especially for long-running or infinite loops
• Be cautious with very short period values as they may cause high CPU usage
• The callback receives the iteration counter starting from 1, not 0
• In blocking mode (block=True), the start() method will not return until execution completes
• The timeout is checked after each callback execution, so actual runtime may slightly exceed the timeout value
• Thread safety: if your callback modifies shared state, ensure proper synchronization mechanisms are in place

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class periodic_v1 49.1% similar

  A utility class that manages periodic event updates for DynamicMap objects, allowing scheduled triggering of stream updates that can be started and stopped.

  From: /tf/active/vicechatdev/patches/spaces.py

• function run_with_timeout 47.2% similar

  Executes a function with a specified timeout using threading, raising a TimeoutException if the function exceeds the time limit.

  From: /tf/active/vicechatdev/docchat/document_indexer.py

• class ExecutionGuard 43.0% similar

  A guard mechanism that prevents recursive or repeated function calls by tracking active executions and enforcing cooldown periods between calls.

  From: /tf/active/vicechatdev/CDocs/__init__.py

• function guard_execution 42.8% similar

  A decorator factory that prevents rapid repeated execution of a function by enforcing a cooldown period between calls.

  From: /tf/active/vicechatdev/CDocs/__init__.py

• class StoppableThread 40.1% similar

  A custom thread class that extends threading.Thread with the ability to gracefully stop a Bokeh server running in the thread by stopping its IOLoop.

  From: /tf/active/vicechatdev/patches/server.py