Perfusion Event Scheduler

Event scheduler, monitor, and executer.

1. Key Features

  • Simplifies the creation of complex perfusion programs

  • Create ‘events’ - tasks to be carried out at a specific time

  • Recurring events - schedule events that repeat at arbitrary intervals

  • Monitors scheduled events and executes tasks automatically

  • Tasks can be any function - positional and keyword arguments are passed

2. Quick Start

Event execution times in scheduler are defined using datetime objects, therefore require the Python datetime module to create. timedelta is also used when defining relative times.

>>> from plateflo import scheduler
>>> from datetime import datetime, timedelta

>>> # instantiate the scheduler object
>>> sched = scheduler.Scheduler()

Create a one-time event, SingleEvent(), scheduled for 5s after the script starts, and add it to the Scheduler():

>>> # define the time
>>> event_time = datetime.now() + timedelta(seconds=5)

>>> # create the event
>>> dinger_event = scheduler.SingleEvent(event_time, print, args=['Ding!'])

>>> # add the event to the scheduler
>>> sched.add_event(dinger_event)

Call monitor() in the main loop of your script to execute the scheduled event at the defined time.

>>> while sched.events:
>>>     sched.monitor()
>>> # five seconds later...
Ding!

Tip

Repeatedly calling monitor() (or anything else, for that matter) in the main loop will needlessly consume CPU cycles. Depending on what else your main loop is doing and the granularity of your event scheduling, adding a simple time.sleep(0.05) 50ms delay can dramatically reduce CPU usage.

3. Usage

i. Functional Overview

Events

The functional core of the scheduler module is its Event objects. Events contain all of the information required to schedule, execute, and reschedule tasks - function calls.

There are two classes of Event object:

One-Time Events:

SingleEvent()

Recurring Events:

RecurringEvent(), DailyEvent()

As their names imply, a SingleEvent() is executed at a specific time, a datetime object, while a RecurringEvent() executes repeatedly at a given interval, a timedelta object.

Both events execute the supplied task when due - a callable object. Both positional and keyword arguments are passed to the task's function pointer on execution.

Scheduling

The Scheduler() keeps track of all Events in an events list, each with its own eventID, generated and returned when adding events to the scheduler.

Event objects are added Scheduler() with the add_event() method, and removed using the remove_event() method. Events are automatically sorted in the events list based on their scheduled execution time, therefore the events list can be considered a queue.

The monitor() method compares the current system time with that of the first Event in the queue and calls the task function when due. The Event is then popped from the queue and transferred to the event_history list.

ii. Creating Event Objects

One-Time Events

Tasks that need only be executed once are created using SingleEvent() objects.

The event is executed at the absolute time specified by a datetime object. See the section below on defining event tasks.

class plateflo.scheduler.SingleEvent(dateTime, task, _eventID=None, args=[], **kwargs)

Simple event object, executes once at the specified time.

Parameters:

  • dateTime (datetime) – Scheduled execution start time

  • task (function pointer) – Function to excecute at scheduled time

  • args (list) – positional arguments, passed to task

  • **kwargs – keyword arguments passed to task

Recurring Events

Events that take place at regular intervals are created using RecurringEvent() objects, which can take several parameters to fine tune define their behaviour:

class plateflo.scheduler.RecurringEvent(interval, task, start_time=None, stop_time=None, delay=None, _eventID=None, _resched=False, args=[], **kwargs)

Event triggered at the specified interval w/ optional start time OR delay.

Parameters:

  • interval (timedelta) – Recurrance interval

  • task (function pointer) – Function to excute at scheduled interval

  • start_time (datetime, optional) – First occurance at specified time. Excludes delay

  • stop_time (datetime, optional) – Terminate recurrance after this time

  • delay (timedelta, optional) – Delay first occurance by this amount. Excludes start_time

  • args (list) – Positional arguments, passed to task

  • **kwargs – Keyword arguments, passed to task

Raises:

  • ValueError – both start_time and delay parameters are provided. task function not provided

  • TypeError – interval not of type timedelta

To simplify the definition of daily events that take place at a specific clock time, the DailyEvent() class can be used.

class plateflo.scheduler.DailyEvent(task, hh=0, mm=0, s=0, args=[], **kwargs)

Wrapper around RecurringEvent for convenient daily task execution.

Parameters:

  • hh (int) – Time of day, hour [0-23]

  • mm (int, default = 0) – Time of day, minute [0-59]

  • s (int, default = 0) – Time of day, seconds [0-59]

  • task (function pointer) – Function executed at scheduled time

  • args (list) – Add positional arguments passed to task

  • kwargs – Additional keyword arguments passed to task

Returns:

Daily recurring event.

Return type:

RecurringEvent

Passing Functions to Event Objects

Tasks are passed into Events passed in as callable objects, basically a function’s name without the round brackets that usually follow. Positional arguments, args, are passed in as a list and keyword arguments passed after all other arguments.

For example, to toggle a valve using a FETbox using hit_hold_chan() after 3 seconds and then disable_chan() 3 seconds after that.

>>> # FETbox object
>>> fet = auto_connect_fetbox()[0]

>>> now = datetime.now()

>>> # Using positional arguments
>>> valve_on = scheduler.SingleEvent(dateTime = now + timedelta(seconds=3),
>>>                                  task = fet.hit_hold_chan,
>>>                                  args=[1, 0.8])

>>> # Using a keyword argument
>>> valve_off = scheduler.SingleEvent(dateTime = now + timedelta(seconds=6),
>>>                                   task = fet.disable_chan,
>>>                                   chan = 1)

A lambda function can be used to define a callable object whose arguments are undetermined until the time of execution.

>>> print_time = lambda : print(datetime.now())
>>> ticker = scheduler.RecurringEvent(interval = timedelta(seconds=1),
>>>                                   task = print_time)

iii. Adding/Removing Events

Events are added to the Scheduler() using the add_event() method, which returns the unique serialized eventID for that event.

>>> my_event # any Event type
>>> my_event_id = sched.add_event(my_event)
>>> print(my_event_id)
1

The eventID can be used to remove the event from the scheduler e.g. to cancel a RecurringEvent():

>>> print(sched.events)
[{'eventID': 1,
  'dateTime': datetime.datetime(2021, 2, 11, 16, 22, 37, 196192),
  'event': <plateflo.scheduler.RecurringEvent object at 0x000001F2946FBA90>}]
>>> sched.remove_event(my_event_id)
>>> print(sched.events)
[]

4. Working Examples

Toggle a FETbox-attached valve at regular intervals for a few cycles:

 1>>> from plateflo import scheduler, fetbox, serial_io
 2>>> from datetime import datetime, date, timedelta
 3>>> from time import sleep
 4>>> import logging
 5
 6>>> # enable logging to see info in the terminal
 7>>> logging.basicConfig(level=logging.INFO)
 8
 9>>> sched = scheduler.Scheduler()
10>>> # auto-connect to FETbox w/ ID 0
11>>> fet = fetbox.auto_connect_fetbox()[0]
12
13>>> # Create recurring events
14
15>>> INTERVAL = 1 # on/off interval
16>>> CYCLES = 8   # total on/off cycles
17
18>>> interval = timedelta(seconds = INTERVAL*2)
19>>> stop_time = datetime.now() + timedelta(seconds = INTERVAL*2*CYCLES)
20
21>>> valve_on_evt = scheduler.RecurringEvent(interval = interval,
22>>>                                         stop_time = stop_time,
23>>>                                         task = fet.enable_chan,
24>>>                                         chan = 1) # this is a task keyword argument
25>>> valve_off_evt = scheduler.RecurringEvent(interval = interval,
26>>>                                          stop_time = stop_time,
27>>>                                          task = fet.disable_chan,
28>>>                                          delay = timedelta(seconds=1),
29>>>                                          chan = 1) # this is a task keyword argument
30
31>>> sched.add_event(valve_on_evt)
32>>> sched.add_event(valve_off_evt)
33
34>>> def main():
35>>>     while sched.events:
36>>>         sched.monitor()
37>>>         sleep(1E-6)
38>>>     fet.kill()
39
40>>> if __name__ == "__main__":
41>>>     main()

>>> # program output:
INFO:FETbox:Scanning for connected FETbox(es)...
INFO:FETbox:Scanning COM6...
INFO:FETbox:            FETbox (ID 0) detected.
INFO:FETbox:COM6 FETbox (ID: 0) initialized
INFO:FETbox:COM6 Enabled chan. 1
INFO:FETbox:COM6 Disabled chan. 1
INFO:FETbox:COM6 Enabled chan. 1
INFO:FETbox:COM6 Disabled chan. 1
INFO:FETbox:COM6 Enabled chan. 1
INFO:FETbox:COM6 Disabled chan. 1
INFO:FETbox:COM6 FETbox CLOSED