Source code for functional.lineage

from functional.execution import ExecutionEngine
from functional.transformations import CACHE_T

[docs]class Lineage(object): """ Class for tracking the lineage of transformations, and applying them to a given sequence. """
[docs] def __init__(self, prior_lineage=None, engine=None): """ Construct an empty lineage if prior_lineage is None or if its not use it as the list of current transformations :param prior_lineage: Lineage object to inherit :return: new Lineage object """ self.transformations = ( [] if prior_lineage is None else list(prior_lineage.transformations) ) self.engine = ( (engine or ExecutionEngine()) if prior_lineage is None else prior_lineage.engine )
[docs] def __repr__(self): """ Returns readable representation of Lineage :return: readable Lineage """ return "Lineage: " + " -> ".join( ["sequence"] + [ for transform in self.transformations] )
[docs] def __len__(self): """ Number of transformations in lineage :return: number of transformations """ return len(self.transformations)
[docs] def __getitem__(self, item): """ Return specific transformation in lineage. :param item: Transformation to retrieve :return: Requested transformation """ return self.transformations[item]
[docs] def apply(self, transform): """ Add the transformation to the lineage :param transform: Transformation to apply """ self.transformations.append(transform)
[docs] def evaluate(self, sequence): """ Compute the lineage on the sequence. :param sequence: Sequence to compute :return: Evaluated sequence """ last_cache_index = self.cache_scan() transformations = self.transformations[last_cache_index:] return self.engine.evaluate(sequence, transformations)
[docs] def cache_scan(self): """ Scan the lineage for the index of the most recent cache. :return: Index of most recent cache """ try: return len(self.transformations) - self.transformations[::-1].index(CACHE_T) except ValueError: return 0