Source code for trace

# coding=utf-8
import datetime
import itertools


def _agent_in_msg(agent, msg):
    # return msg.to == agent or msg.sender == agent
    return msg.app_id == agent


[docs]class TraceStore(object): """Stores and allows queries about events.""" def __init__(self, size): self.size = size self.store = []
[docs] def reset(self): """Resets the trace store""" self.store = []
[docs] def append(self, event, category=None): """ Adds a new event to the trace store. The event may hava a category Args: event (spade.message.Message): the event to be stored category (str, optional): a category to classify the event (Default value = None) """ date = datetime.datetime.now() self.store.insert(0, (date, event, category)) if len(self.store) > self.size: del self.store[-1]
[docs] def len(self): """ Length of the store Returns: int: the size of the trace store """ return len(self.store)
def latest(self): return self.store[0]
[docs] def all(self, limit=None): """ Returns all the events, until a limit if defined Args: limit (int, optional): the max length of the events to return (Default value = None) Returns: list: a list of events """ return self.store[:limit][::-1]
[docs] def received(self, limit=None): """ Returns all the events that have been received (excluding sent events), until a limit if defined Args: limit (int, optional): the max length of the events to return (Default value = None) Returns: list: a list of received events """ return list(itertools.islice((itertools.filterfalse(lambda x: x[1].sent, self.store)), limit))[::-1]
[docs] def filter(self, limit=None, app_id=None, category=None): """ Returns the events that match the filters Args: limit (int, optional): the max length of the events to return (Default value = None) app_id (str, optional): only events that have been sent or received from 'app_id' (Default value = None) category (str, optional): only events belonging to the category (Default value = None) Returns: list: a list of filtered events """ if category and not app_id: msg_slice = itertools.islice((x for x in self.store if x[2] == category), limit) elif app_id and not category: msg_slice = itertools.islice((x for x in self.store if _agent_in_msg(app_id, x[1])), limit) elif app_id and category: msg_slice = itertools.islice((x for x in self.store if _agent_in_msg(app_id, x[1]) and x[2] == category), limit) else: msg_slice = self.all(limit=limit) return msg_slice return list(msg_slice)[::-1]