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 collections.abc import Callable
from typing import Any

import numpy as np

from materialize.feature_benchmark.measurement import Measurement


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

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

    def aggregate(self) -> Any:
        if len(self._data) == 0:
            return None
        else:
            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, dtype=float)
        mean: float = np.mean(self._data, dtype=float)
        val = mean - (stdev * self._num_stdevs)
        return val


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


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: Measurement) -> None:
        self._data.append(measurement.value)

    def aggregate(self) -> Any:
        if len(self._data) == 0:
            return None
        else:
            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:
    if len(self._data) == 0:
        return None
    else:
        return self.func()([*self._data])
def append(self, measurement: Measurement) ‑> None
Expand source code Browse git 
def append(self, measurement: Measurement) -> None:
    self._data.append(measurement.value)
def func(self) ‑> collections.abc.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) ‑> collections.abc.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) ‑> collections.abc.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, dtype=float), sigma=np.std(self._data, dtype=float)
        )

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, dtype=float), sigma=np.std(self._data, dtype=float)
    )
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, dtype=float)
        mean: float = np.mean(self._data, dtype=float)
        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, dtype=float)
    mean: float = np.mean(self._data, dtype=float)
    val = mean - (stdev * self._num_stdevs)
    return val