CyclicTask

Allow a task to be executed regulary

Table of content

• Installation
• ContinuousTask class usage
  • Minimum usage
  • Maximum execution time
• TimedTask class
  • Minimum usage
  • Maximum execution time

Installation

python3 -m pip install CyclicTask

ContinuousTask class usage

Minimum usage

The ContinuousTask class allow to execute a task as fast as possible, until the stop request.

from CyclicTask.ContinuousTask import ContinuousTask

# Sub-classing `ContinuousTask` class
class Task(ContinuousTask):
    # override `task` function
    def task(self) -> None:
        # --- Task definition ---
        pass

# Task creation
task = Task()

# Task start
task.start()

# Task stop
task.stop()

# Waiting for the end of the current cycle
task.join()

Maximum execution time

maximumTime parameter allow to monitor the execution duration of one cycle.

If this one exceed 80% of maximumTime the attribut cycleTimeWarning is set for one cycle, and a message in the logger is written.

If the duration exceed maximumTime the exception CycleTimeError is raised, and a message is written in the logger.

from CyclicTask.ContinuousTask import ContinuousTask

# Sub-classing `ContinuousTask` class
class Task(ContinuousTask):
    # override `task` function
    def task(self) -> None:
        # --- Task definition ---
        pass

# Task creation with `maximumTime` set to 1 second
task = Task(maximumTime = 1.0)

# Task start
task.start()

# Task stop
task.stop()

# Waiting for the end of the current cycle
task.join()

TimedTask class

Minimum usage

The TimedTask class allow to execute a task every x seconds, until the stop request.

from CyclicTask.TimedTask import TimedTask

# Sub-classing `TimedTask` class
class Task(TimedTask):
    # override `task` function
    def task(task) -> None:
        # --- Task definition ---
        pass

# Task creation with `cycleTime` set to 1 second
task = Task(cycleTime = 1.0)

# Task start
task.start()

# Task stop
task.stop()

# Waiting for the end of the current cycle
task.join()

Maximum execution time

maximumTime parameter allow to monitor the execution duration of one cycle. If maximumTime is left to None, the maximum execution time is set to 150% of the cycle time.

If this one exceed 80% of maximumTime the attribut cycleTimeWarning is set for one cycle, and a message in the logger is written.

If the duration exceed maximumTime the exception CycleTimeError is raised, and a message is written in the logger.

from CyclicTask.TimedTask import TimedTask

# Sub-classing `TimedTask` class
class Task(TimedTask):
    # override `task` function
    def task(task) -> None:
        # --- Task definition ---
        pass

# Task creation with `cycleTime` set to 1 second, and `maximumTime` set to 1.0 second
task = Task(cycleTime = 1.0, maximumTime = 1.0)

# Task start
task.start()

# Task stop
task.stop()

# Waiting for the end of the current cycle
task.join()