Module materialize.feature_benchmark.aggregation

Expand source code Browse git
# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.

import statistics
from typing import Any, Callable, List

import numpy as np


class Aggregation:
    def __init__(self) -> None:
        self._data: List[float] = []

    def append(self, measurement: float) -> None:
        self._data.append(measurement)

    def aggregate(self) -> Any:
        return self.func()([*self._data])

    def func(self) -> Callable:
        assert False


class MinAggregation(Aggregation):
    def func(self) -> Callable:
        return min


class MeanAggregation(Aggregation):
    def func(self) -> Callable:
        return np.mean


class StdDevAggregation(Aggregation):
    def __init__(self, num_stdevs: float) -> None:
        self._data = []
        self._num_stdevs = num_stdevs

    def aggregate(self) -> float:
        stdev: float = np.std(self._data)
        mean: float = np.mean(self._data)
        val = mean - (stdev * self._num_stdevs)
        return val


class NormalDistributionAggregation(Aggregation):
    def aggregate(self) -> statistics.NormalDist:
        return statistics.NormalDist(mu=np.mean(self._data), sigma=np.std(self._data))


class NoAggregation(Aggregation):
    def aggregate(self) -> Any:
        return self._data[0]

Classes

class Aggregation
Expand source code Browse git
class Aggregation:
    def __init__(self) -> None:
        self._data: List[float] = []

    def append(self, measurement: float) -> None:
        self._data.append(measurement)

    def aggregate(self) -> Any:
        return self.func()([*self._data])

    def func(self) -> Callable:
        assert False

Subclasses

Methods

def aggregate(self) ‑> Any
Expand source code Browse git
def aggregate(self) -> Any:
    return self.func()([*self._data])
def append(self, measurement: float) ‑> None
Expand source code Browse git
def append(self, measurement: float) -> None:
    self._data.append(measurement)
def func(self) ‑> Callable
Expand source code Browse git
def func(self) -> Callable:
    assert False
class MeanAggregation
Expand source code Browse git
class MeanAggregation(Aggregation):
    def func(self) -> Callable:
        return np.mean

Ancestors

Methods

def func(self) ‑> Callable
Expand source code Browse git
def func(self) -> Callable:
    return np.mean
class MinAggregation
Expand source code Browse git
class MinAggregation(Aggregation):
    def func(self) -> Callable:
        return min

Ancestors

Methods

def func(self) ‑> Callable
Expand source code Browse git
def func(self) -> Callable:
    return min
class NoAggregation
Expand source code Browse git
class NoAggregation(Aggregation):
    def aggregate(self) -> Any:
        return self._data[0]

Ancestors

Methods

def aggregate(self) ‑> Any
Expand source code Browse git
def aggregate(self) -> Any:
    return self._data[0]
class NormalDistributionAggregation
Expand source code Browse git
class NormalDistributionAggregation(Aggregation):
    def aggregate(self) -> statistics.NormalDist:
        return statistics.NormalDist(mu=np.mean(self._data), sigma=np.std(self._data))

Ancestors

Methods

def aggregate(self) ‑> statistics.NormalDist
Expand source code Browse git
def aggregate(self) -> statistics.NormalDist:
    return statistics.NormalDist(mu=np.mean(self._data), sigma=np.std(self._data))
class StdDevAggregation (num_stdevs: float)
Expand source code Browse git
class StdDevAggregation(Aggregation):
    def __init__(self, num_stdevs: float) -> None:
        self._data = []
        self._num_stdevs = num_stdevs

    def aggregate(self) -> float:
        stdev: float = np.std(self._data)
        mean: float = np.mean(self._data)
        val = mean - (stdev * self._num_stdevs)
        return val

Ancestors

Methods

def aggregate(self) ‑> float
Expand source code Browse git
def aggregate(self) -> float:
    stdev: float = np.std(self._data)
    mean: float = np.mean(self._data)
    val = mean - (stdev * self._num_stdevs)
    return val