Source code for polyglotdb.query.base.func


[docs]class AggregateFunction(object): function = '' template = '{function}({property})' collection_templates = {'sum': 'reduce(count = 0, n in {property} | count + n)', 'avg': 'reduce(count = 0, n in {property} | count + n) / toFloat(size({property}))', 'count': 'size({property})'} def __init__(self, attribute=None): self.attribute = attribute self.output_label = None def __hash__(self): return hash((self.function, self.attribute)) @property def nodes(self): if self.attribute is None: return [] return self.attribute.nodes @property def acoustic(self): """ Returns ------- :class:`~polyglotdb.graph.attributes.Attribute` acoustic attribute """ return self.attribute.acoustic @property def base_annotation(self): """ Returns ------- :class:`~polyglotdb.graph.attributes.AnnotationAttribute` base annotation """ return self.attribute.base_annotation @property def with_alias(self): """ Returns ------- with_alias : str the alias of a `~polyglotdb.graph.attributes.AnnotationAttribute` object """ return self.attribute.with_alias @property def with_aliases(self): """ Returns ------- with_aliases : str the with strings of a `~polyglotdb.graph.attributes.AnnotationAttribute` object """ return self.attribute.with_aliases @property def output_alias(self): """ Returns ------- output_label : str the output label """ if self.output_label is None: if self.attribute is not None: name = self.attribute.label else: name = 'all' return '{}_{}'.format(self.__class__.__name__.lower(), name) else: return self.output_label def aliased_for_output(self): """ Returns ------- str output alias cypher string """ prop = self.for_cypher() output = self.output_alias return '{} AS {}'.format(prop, output) @property def collapsing(self): """ Returns ------- False if there is a PathAttribute, True otherwise """ if self.attribute is not None and self.attribute.collapsing: return False return True def column_name(self, label): """ sets output label Parameters ---------- label : str the label to set Returns ------- self """ self.output_label = label return self def for_cypher(self): """ Return a Cypher representation of the clause. """ if not self.collapsing: return self.collection_templates[self.function].format( property=self.attribute.for_cypher()) elif self.attribute is not None: element = self.attribute.for_cypher() else: element = '*' return self.template.format(function=self.function, property=element)
[docs]class Average(AggregateFunction): function = 'avg'
[docs]class Count(AggregateFunction): function = 'count'
[docs]class Sum(AggregateFunction): function = 'sum'
[docs]class Stdev(AggregateFunction): function = 'stdev'
[docs]class Max(AggregateFunction): function = 'max'
[docs]class Min(AggregateFunction): function = 'min'
class Quantile(AggregateFunction): function = 'percentileDisc' template = '{function}({property}, {percentile})' def __init__(self, attribute, percentile=0.5): self.attribute = attribute self.percentile = percentile self.output_label = None def for_cypher(self): """ Return a Cypher representation of the clause. """ if self.attribute is not None: element = self.attribute.for_cypher() else: raise (AttributeError) return self.template.format(function=self.function, percentile=self.percentile, property=element) class Median(Quantile): def __init__(self, attribute): Quantile.__init__(self, attribute, 0.5) class InterquartileRange(AggregateFunction): template = 'percentileDisc({property}, 0.75) - percentileDisc({property}, 0.25)' def for_cypher(self): """ Return a Cypher representation of the clause. """ if self.attribute is not None: element = self.attribute.for_cypher() else: raise (AttributeError) return self.template.format(property=element)