Tempo¶

Simple schedules via ReccurrentEvent class¶

RecurrentEvent models simple schedules like “from 10 to 19 hours of the day”, “from 1 to 4 months of the year”, etc. For example this will define schedule “from first to sixth day of the week”:

>>> from tempo.recurrentevent import RecurrentEvent
>>> recurrentevent = RecurrentEvent(1, 6, 'day', 'week')

Then, we can perform queries - containment of a single date:

>>> import datetime as dt
>>> d1 = dt.datetime(2000, 1, 10)
>>> d1.weekday()  # Monday - first day of the week
0
>>> d1 in recurrentevent
True
>>> d2 = dt.datetime(2000, 1, 14)
>>> d2.weekday()  # Friday - fifth day of the week
4
>>> d2 in recurrentevent
True
>>> d3 = dt.datetime(2000, 1, 15)
>>> d3.weekday()  # Saturday - sixth day of the week
5
>>> d3 in recurrentevent
False

We also can query for further occurences starting from certain point of time:

>>> from itertools import islice
>>> start = dt.datetime(2000, 1, 4)
>>> start.weekday()  # Tuesday
1
>>> list(islice(recurrentevent.forward(start), 3))
[(datetime.datetime(2000, 1, 4, 0, 0),
  datetime.datetime(2000, 1, 8, 0, 0)),
 (datetime.datetime(2000, 1, 10, 0, 0),
  datetime.datetime(2000, 1, 15, 0, 0)),
 (datetime.datetime(2000, 1, 17, 0, 0),
  datetime.datetime(2000, 1, 22, 0, 0))]

RecurrentEvent.forward() returns a generator, that yields as largest possible interval each time. In this case it’s a time span between a monday and a saturday (non-inclusive) of each week.

Notice - start defines Tuesday, but our schedule starts on Monday - and forward(), yielded the first time interval, that starts on Tuesday, the time, that equals our start argument. It shrinked the first time interval by the start, since otherwise the first time interval would be started from the time earlier, than start. We can change this behaviour, by passing additional argument trim:

>>> list(islice(recurrentevent.forward(start, trim=False), 3))
[(datetime.datetime(2000, 1, 3, 0, 0),
  datetime.datetime(2000, 1, 8, 0, 0)),
 (datetime.datetime(2000, 1, 10, 0, 0),
  datetime.datetime(2000, 1, 15, 0, 0)),
 (datetime.datetime(2000, 1, 17, 0, 0),
  datetime.datetime(2000, 1, 22, 0, 0))]

Now RecurrentEvent.forward() yielded largest possible interval not only in future direction for the start, but also in past direction.

Composite schedules via ReccurrentEvent class¶

Let’s now take a look at RecurrentEventSet. It makes possible to compose simple schedules to more complex ones, using operators of union (OR), intersection (AND) and negation (NOT).

For example:

>>> from tempo.recurrenteventset import RecurrentEventSet
>>> recurrenteventset = RecurrentEventSet.from_json(
...     ('OR',
...         ('AND',
...             ('NOT', [12, 13, 'hour', 'day']),
...             ('AND', [1, 4, 'day', 'week'], [10, 19, 'hour', 'day']),
...             ('AND', [5, 6, 'day', 'week'], [10, 16, 'hour', 'day'])))
... )

That defines “from Monday to Thursday from 10am to 7pm and in Friday from 10am to 4pm with the gap from 12am to 1pm”.

RecurrentEventSet has the same interface as RecurrentEvent: it provides RecurrentEventSet.forward() and RecurrentEventSet.__contains__() methods, which has exactly the same meaning as RecurrentEvent ones has.

tempo.recurrentevent¶

Provides RecurrentEvent class.

class tempo.recurrentevent.RecurrentEvent(start, stop, unit, recurrence=None)¶

An interval of time expressed in some ‘unit’ of time (second, week, year, etc), recurring with some ‘recurrence’, also expressed in some unit of time. For example minutes interval can recur hourly or yearly, but can’t recur secondly.

With None passed as ‘recurrence’, time interval will be defined without recurrence, just as a single non-recurring interval between two points in time and counted from “the beginning of time”. By convention “the beginning of time” is 1-1-1 00:00:00.

Parameters:	start (int) – Start of recurring interval. stop (int) – Non-inclusive end of recurring interval. unit (str) – Unit of time in which time interval is expressed. recurrence (str, optional) – Recurrence of time interval. Can be (and by default is) None, which means - “no recurrence”.

Examples

>>> from datetime import datetime
>>> recurrentevent = RecurrentEvent(0, 15, Unit.SECOND, Unit.MINUTE)
>>> datetime(2000, 1, 1, 5, 3, 10) in recurrentevent
... True
>>> datetime(2000, 1, 1, 5, 3, 16) in recurrentevent
... False

__contains__(item)¶

Test given datetime ‘item’ for containment in the recurrent event.

Parameters:	item (datetime.datetime) – A ‘datetime’ object to test.
Returns:	Result of containment test.
Return type:	bool

Notes

The algorithm here consists of following steps:

If recurrence is set:

1. Given datetime floored to unit of ‘recurrence’ and stored.
2. Then given datetime floored to unit of ‘unit’ and stored.
3. Delta between resulting datetime objects is calculated and expressed in units of ‘unit’. For example if delta is “2 days” and ‘unit’ is minutes, delta will be “2*24*60 minutes”.

If recurrence is not set:

1. Delta between date of “the beginning of time” and given date is calculated and expressed in units of ‘unit’.

4. Resulting delta tested for containment in the interval.

forward(start, trim=True)¶

Iterate time intervals starting from ‘start’. Intervals returned in form of (start, end) pair, where start is a datetime object representing the start of the interval and end is the non-inclusive end of the interval.

Parameters:	start (datetime.datetime) – A lower bound for the resulting sequence of intervals. trim (bool) – Whether a first interval should be trimmed by ‘start’ or it should be full, so it’s start point may potentially be earlier, that ‘start’.
Yields:	start (datetime.datetime) – Start of an interval. end (datetime.datetime) – End of an interval.

classmethod from_json(value)¶

Constructs RecurrentEvent instance from JSON serializable representation or from JSON string.

isgapless()¶

Tests if the RecurrentEvent instance defines infinite time interval.

to_json()¶

Exports RecurrentEvent instance to JSON serializable representation.

tempo.recurrenteventset¶

Provides RecurrentEventSet class.

class tempo.recurrenteventset.RecurrentEventSet(expression)¶

A set of time intervals, combined with a set logic operators: AND, OR and NOT.

Parameters:

expression (tuple) – A nested expression, composed of operators and arguments, which are RecurrentEvent instances or sub-expressions. Example of an expression: (AND, RecurrentEvent(Interval(10, 19), 'hour', 'day'), (NOT, RecurrentEvent(Interval(14, 15), 'hour', 'day')), (NOT, RecurrentEvent(Interval(6, 7), 'day', 'week')), ]) It means: ‘From 10:00 to 19:00 every day, except from 14:00 to 15:00, and weekends’.

__contains__(item)¶

Containment test. Accepts whatever RecurrentEvent can test for containment.

forward(start, trim=True)¶

Generates intervals according to the expression.

Intervals never overlap. Each next interval is largest possbile interval.

Parameters:	start (datetime.datetime) – Inclusive start date. trim (bool) – If True (which is default), the starting point of a first interval will always be equal to or greater than’start’. Otherwise it will be equal to the point, where the interval actually starts, which may be placed earlier in time, that ‘start’.
Yields:	tuple – Inclusive start and non-inclusive dates of an interval.

Notes

The alghorithm is simple:

1. It generates intervals from RecurrentEvent instances and applies set logic operators on them.
2. Checks if resulting interval has gap.
3. Checks if there is a possibility, that this gap will gone, by checking if some of the generators could possibly generate interval that will intersect with gap.
4. If checks succeed, yields interval previous to gap.
5. If not - iterates generators until check succeed.

This implementation if fairly ineffective and should be otimized.

classmethod from_json(value)¶

Constructs RecurrentEventSet instance from JSON serializable representation or from JSON string.

static from_json_callback(operator, *args)¶

Converts arguments that are time intervals to Python.

to_json()¶

Exports RecurrentEventSet instance to JSON serializable representation.

static to_json_callback(operator, *args)¶

Converts arguments that are time intervals to JSON.

static validate_json(expression)¶

Validates JSON expression.

class tempo.recurrenteventset.Result(value)¶

Callback result wrapper.

Intended to be used to avoid confusion between expressions and callback results, in case if they are expressions themselves.

tempo.django.fields¶

Provides Django model fields API for RecurrentEventSet.

class tempo.django.fields.Contains(lhs, rhs)¶

Provides contains lookup for RecurrentEventSetField.

Checks a single datetime object for containment in RecurrentEventSet.

class tempo.django.fields.Intersects(lhs, rhs)¶

Provides intersects lookup for RecurrentEventSetField.

Checks a given time interval in form of a pair-tuple of datetime objects, intersects with time defined by time interval set in given column.

class tempo.django.fields.OccursWithin(lhs, rhs)¶

Provides occurs_within lookup for RecurrentEventSetField.

Checks if some of continous events, defined in time interval set is enclosed by dates in given pair-tuple of datetime objects.

class tempo.django.fields.RecurrentEventSetField(verbose_name=None, name=None, primary_key=False, max_length=None, unique=False, blank=False, null=False, db_index=False, rel=None, default=<class django.db.models.fields.NOT_PROVIDED>, editable=True, serialize=True, unique_for_date=None, unique_for_month=None, unique_for_year=None, choices=None, help_text=u'', db_column=None, db_tablespace=None, auto_created=False, validators=[], error_messages=None)¶

DB representation of recurrenteventset. Requires PostgreSQL 9.4+.

tempo.rest_framework.serializers¶

Provides utilities for serialization/deserialization of Tempo data types.

class tempo.rest_framework.serializers.RecurrentEventSetField(read_only=False, write_only=False, required=None, default=<class rest_framework.fields.empty>, initial=<class rest_framework.fields.empty>, source=None, label=None, help_text=None, style=None, error_messages=None, validators=None, allow_null=False)¶

Representation of RecurrentEventSet.