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 datettime 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 datettime instance in the local time zone.
Return type: datetime

backpack.timepiece.panorama_timestamp_to_datetime(panorama_ts)

Converts panoramasdk.media.time_stamp (seconds, microsececonds) 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 occuring 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 managaer, 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

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]]) – Positional arguments of the callback
cbkwargs (Optional[Dict[str, Any]]) – Keyword arguments of the callback
executor (Optional[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: bool

class backpack.timepiece.AtSchedule(at, *args, **kwargs)

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) – When to execute the task
args (Any) – Positional arguments to be passed to superclass initializer
kwargs (Any) – Keyword arguments to be passed to superclass initializer

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: bool

class backpack.timepiece.IntervalSchedule(interval, *args, **kwargs)

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 (timedelta) – The time interval of the executions
args (Any) – Positional arguments to be passed to superclass initializer
kwargs (Any) – Keyword arguments to be passed to superclass initializer

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: bool

class backpack.timepiece.OrdinalSchedule(ordinal, *args, **kwargs)

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.
args (Any) – Positional arguments to be passed to superclass initializer
kwargs (Any) – Keyword arguments to be passed to superclass initializer

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)

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.Tachometer(stats_callback, stats_interval=datetime.timedelta(seconds=60), executor=None)

Bases: object

A Tachometer can be used to measure the frequency of recurring events, and periodically report statistics about it.

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

Tachometer will periodically call a callback function with the following signature:

def stats_callback(timestamp, ticker):
    pass

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

The tick() method will return a tuple with these elements:

bool: True, if the internal schedule was fired (you can ignore this value)
a tuple with these elements: - bool: True, if the stats_callback was called - the return value of stats_callback if it was called, else None

Parameters

stats_callback (Callable[[datetime, Ticker], None]) – A callable with the above signature that will be called when new statistics is available.
stat_interval – The interval of the statistics calculation
executor (Optional[Executor]) – If specified, callback will be called asynchronously using this executor
stats_interval (timedelta) –

tick()

Call this method when a recurring event happens.

Returns: True if the stat_callback was called in this tick.
Return type: bool