timepiece

This module contains time-related utility methods for measuring code execution time and scheduling tasks in an external event loop.

backpack.timepiece.local_now()

Returns the current time in local time zone.

Returns:: A timezone aware datetime instance in the local time zone.
Return type:: datetime

backpack.timepiece.local_dt(dt)

Converts the supplied naive datetime to be time zone aware in the local time zone.

Parameters:: dt (datetime) – The naive datetime instance.
Returns:: A timezone aware datetime instance in the local time zone.
Return type:: datetime

backpack.timepiece.panorama_timestamp_to_datetime(panorama_ts)

Converts panoramasdk.media.time_stamp (seconds, microseconds) tuple to python datetime.

Parameters:: panorama_ts (Tuple[int, int]) – The Panorama timestamp
Returns:: A python datetime instance.
Return type:: datetime

class backpack.timepiece.BaseTimer(max_intervals=10)

Bases: ABC

Base class for code execution time measuring timers.

Parameters:: max_intervals (int) – Maximum number of intervals to remember.

min()

Returns the shortest time interval between the events (in seconds).

Return type:: float

max()

Returns the longest time interval between the events (in seconds).

Return type:: float

mean()

Returns the mean time interval between the events (in seconds).

Return type:: float

sum()

Returns the sum of the time interval between recorded events (in seconds).

Return type:: float

len()

Returns the number of recorded events.

Return type:: int

freq()

Returns the mean frequency of the events (in Hertz).

Return type:: float

reset()

Resets the timer.

Return type:: None

class backpack.timepiece.Ticker(max_intervals=10)

Bases: BaseTimer

A performance profiler that measures the time interval between repeatedly occurring events.

Ticker can also calculate basic statistics of the time intervals.

Example usage:

ticker = Ticker(max_intervals=5)
for i in range(10):
    ticker.tick()
    time.sleep(random.random() / 10)
print(ticker)

Results:

<Ticker intervals=[0.0899, 0.0632, 0.0543, 0.0713, 0.0681] min=0.0543 mean=0.0694 max=0.0899>

Variables:

max_intervals – Maximum number of time intervals to be recorded. Only the last max_intervals number of intervals will be kept.
intervals – The recorded intervals in seconds between the successive events.

Parameters:

max_intervals (int) –

tick()

Registers a tick in this Ticker.

Return type:: None

class backpack.timepiece.StopWatch(name, max_intervals=10)

Bases: BaseTimer

A simple performance profiler with context managers.

There are two ways to use StopWatch: as a context manager, or with the tick() method. You can use the same StopWatch object in both ways at the same time.

When used as a context manager, StopWatch can be used to measure the performance of python code using an elegant API based on context manager. You can measure nested and serial execution.

The second way is to measure the average execution of repeating tasks with the tick() function call. After enough data samples were collected with tick(), you can calculate the average execution of the task calling mean_tick().

Example usage:

import time
with StopWatch('root') as root:
    with root.child('task1', max_ticks=5) as task:
        time.sleep(0.01)
        with task.child('subtask1.1'):
            time.sleep(0.03)
        with task.child('subtask1.2'):
            time.sleep(0.07)
        with task.child('subtask1.3'):
            time.sleep(0.09)
    with root.child('task2') as task:
        time.sleep(0.17)
print(root)

Results:

<StopWatch name=root total_elapsed=0.9222 children=[
    <StopWatch
        name=task1
        total_elapsed=0.7520
        ticks=[0.0501, 0.0601, 0.0701, 0.0802, 0.0902]
        mean_tick=0.0701
        children=[
            <StopWatch name=subtask1.1 total_elapsed=0.0301>,
            <StopWatch name=subtask1.2 total_elapsed=0.0701>,
            <StopWatch name=subtask1.3 total_elapsed=0.0902>
        ]
    >,
    <StopWatch name=task2 total_elapsed=0.1702>
]>

Parameters:

name (str) – The name of this StopWatch
max_intervals (int) – Maximum number of intervals to be recorded.

name

The name of this StopWatch

Type:: str

parent

The parent StopWatch

Type:: Optional[‘StopWatch’]

children

The name-indexed dictionary of the children StopWatches

Type:: Dict[str, ‘StopWatch’]

child(name, max_intervals=None)

Creates a new or returns an existing child of this StopWatch.

Parameters:

name (str) – Name of the child StopWatch.
max_intervals (Optional[int]) – Maximum number of intervals to be recorded in the child. If None, max_intervals of the parent (this object) will be used.

Return type:

StopWatch

parents()

Returns a generator of all parents of this StopWatch.

Return type:: Iterator[StopWatch]

full_name()

Returns the fully qualified name of this StopWatch, including all parents’ name.

Return type:: str

measure(fun)

Use stopwatch as decorator.

Usage:

import time
from backpack.timepiece import StopWatch

stopwatch = StopWatch('stopwatch')

@stopwatch.measure
def long_running_func():
    time.sleep(10)

property level: int: Returns the number of parents.

class backpack.timepiece.Callback(cb, cbargs=None, cbkwargs=None, executor=None)

Bases: object

Encapsulates a callback function and its arguments.

The callback can be optionally called asynchronously.

Parameters:

cb (Callable) – The callback function to be called
cbargs (Optional[List[Any]]) – The callback function to be called
cbkwargs (Optional[Dict[str, Any]]) – Keyword arguments of the callback
executor (Optional[concurrent.futures.Executor]) – If specified, the callback function will be called asynchronously using this executor.

class backpack.timepiece.Schedule(repeating, callback)

Bases: ABC

Schedules a task to be called later with the help of an external scheduler.

The external scheduler is expected to call the tick() method periodically, most likely from an event-loop.

Parameters:

repeating (bool) – If this schedule fires repeatedly
callback (Callback) – The callback to be called when the scheduler fires

fire()

Fires the schedule calling the callback.

Returns:: If not using an executor (the callback is called synchronously), fire() returns the return value of the callback. Otherwise it returns None.
Return type:: None

abstract tick()

The heartbeat of the schedule to be called periodically.

Returns:: A tuple of (True, callback_return_value) if the schedule was fired, otherwise (False, None)
Return type:: Tuple[bool, Any]

class backpack.timepiece.AtSchedule(at, callback)

Bases: Schedule

Schedules a task to be executed only once at a specific time.

The task will be executed at the next tick after the specified datetime.

Parameters:

at (datetime.datetime) – When to execute the task
callback (Callback) – The callback to be called when the scheduler fires

property at: datetime: Property accessor for ‘at’.

tick()

The heartbeat of the schedule to be called periodically.

Returns:: A tuple of (True, callback_return_value) if the schedule was fired, otherwise (False, None)
Return type:: Tuple[bool, Any]

class backpack.timepiece.IntervalSchedule(interval, callback)

Bases: Schedule

Schedules a task to be executed at regular time intervals.

The task will be executed at the first tick and at each tick after the specified time interval has passed.

Parameters:

interval (datetime.timedelta) – The time interval of the executions
callback (Callback) – The callback to be called when the scheduler fires

tick()

The heartbeat of the schedule to be called periodically.

Returns:: A tuple of (True, callback_return_value) if the schedule was fired, otherwise (False, None)
Return type:: Tuple[bool, Any]

class backpack.timepiece.OrdinalSchedule(ordinal, callback)

Bases: Schedule

Schedules a task to be executed at each nth tick.

At the first tick the task will not be executed.

Parameters:

ordinal (int) – Execute the task once in every ordinal number of ticks. An OrdinalSchedule with zero ordinal will never fire.
callback (Callback) – The callback to be called when the scheduler fires

tick()

The heartbeat of the schedule to be called periodically.

Returns:: A tuple of (True, callback_return_value) if the schedule was fired, otherwise (False, None)
Return type:: Tuple[bool, Any]

class backpack.timepiece.AlarmClock(schedules=None)

Bases: object

An alarm clock can be used to bundle different schedules and send them the tick event at once.

Parameters:: schedules (List[Schedule]) – The list of the schedules.

tick()

The heartbeat of the alarm clock to be called periodically.

Will forward the tick to the registered schedules.

Return type:: None

class backpack.timepiece.BaseTachometer(timer, stats_callback, stats_interval=datetime.timedelta(seconds=60), executor=None)

Bases: ABC

Abstract base class for tachometers.

A Tachometer can be used to measure the frequency of recurring events, and periodically report statistics about it by calling a callback function with the following signature:

def stats_callback(timestamp: datetime.datetime, timer: BaseTimer):
    pass

passing the timestamp of the last event, as well as the BaseTimer instance that collected the events. You can access the min(), max(), sum() (total processing time) and the len() (number of measurements) methods of the timer.

This class is not intended to be instantiated. Use one of the subclasses instead.

Parameters:

timer (BaseTimer) – Instance of the BaseTimer subclass that will be used to report events.
stats_callback (Callable[[datetime.datetime, BaseTimer]) – A callable with the above signature that will be called when new statistics is available.
stats_interval (datetime.timedelta) – The interval of the statistics calculation. Defaults to one minute.
executor (Optional[concurrent.futures.Executor]) – If specified, callback will be called asynchronously using this executor

class backpack.timepiece.TickerTachometer(stats_callback, stats_interval=datetime.timedelta(seconds=60), executor=None)

Bases: BaseTachometer

TickerTachometer is a combination of a Ticker and a Tachometer.

It reports statistics about the call frequency of the tick() method.

Call the tick() method of the tachometer each time an atomic event happens. For example, if you are interested in the spastics of the frame processing time of your application, call tick method each time you process a new frame.

Parameters:

stats_callback (Callable[[datetime.datetime, Ticker]) – A callable that will be called when new statistics is available.
stats_interval (datetime.timedelta) – The interval of the statistics calculation. Defaults to one minute.
executor (Optional[concurrent.futures.Executor]) – If specified, callback will be called asynchronously using this executor

tick()

Call this method when a recurring event happens.

Returns:

A tuple with these elements:

bool: True, if stats_callback was called

the return value of stats_callback if it was called, else None

Return type:

Tuple[bool, Any]

class backpack.timepiece.StopWatchTachometer(stats_callback, stats_interval=datetime.timedelta(seconds=60), executor=None)

Bases: BaseTachometer

StopWatchTachometer is a combination of a StopWatch and a Tachometer.

It reports statistics about the elapsed time intervals.

Much like StopWatch, you can use StopWatchTachometer as a context manager or as a function decorator.

For example:

swt = StopWatchTachometer(
    stats_callback=lambda dt, timer: print('Mean interval:', timer.mean()),
    stats_interval=datetime.timedelta(seconds=10)
)

for i in range(11):
    with swt:
        print('tick')
        time.sleep(2)

@swt.measure
def long_running_func():
    print('tack')
    time.sleep(1.5)

for i in range(15):
    long_running_func()

Parameters:

stats_callback (Callable[[datetime.datetime, Ticker]) – A callable that will be called when new statistics is available.
stats_interval (datetime.timedelta) – The interval of the statistics calculation. Defaults to one minute.
executor (Optional[concurrent.futures.Executor]) – If specified, callback will be called asynchronously using this executor